Merge branch 'dev' into feature-document-versions-1218

- store content per version
- root doc retrieval returns latest content
- updating content affects the latest version
- load metadata per version

.mypy-baseline.txt

@@ -96,9 +96,7 @@ src/documents/conditionals.py:0: error: Function is missing a type annotation fo
 src/documents/conditionals.py:0: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]
 src/documents/conditionals.py:0: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]
 src/documents/conditionals.py:0: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]
-src/documents/conditionals.py:0: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]
 src/documents/consumer.py:0: error: "ConsumerPluginMixin" has no attribute "input_doc"  [attr-defined]
-src/documents/consumer.py:0: error: "ConsumerPluginMixin" has no attribute "log"  [attr-defined]
 src/documents/consumer.py:0: error: "ConsumerPluginMixin" has no attribute "metadata"  [attr-defined]
 src/documents/consumer.py:0: error: "ConsumerPluginMixin" has no attribute "metadata"  [attr-defined]
 src/documents/consumer.py:0: error: "ConsumerPluginMixin" has no attribute "metadata"  [attr-defined]
@@ -173,7 +171,6 @@ src/documents/filters.py:0: error: Function is missing a type annotation  [no-un
 src/documents/filters.py:0: error: Function is missing a type annotation  [no-untyped-def]
 src/documents/filters.py:0: error: Function is missing a type annotation  [no-untyped-def]
 src/documents/filters.py:0: error: Function is missing a type annotation  [no-untyped-def]
-src/documents/filters.py:0: error: Function is missing a type annotation  [no-untyped-def]
 src/documents/filters.py:0: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]
 src/documents/filters.py:0: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]
 src/documents/filters.py:0: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]
@@ -345,18 +342,11 @@ src/documents/migrations/0001_initial.py:0: error: Skipping analyzing "multisele
 src/documents/migrations/0008_sharelinkbundle.py:0: error: Function is missing a type annotation  [no-untyped-def]
 src/documents/migrations/0008_sharelinkbundle.py:0: error: Function is missing a type annotation  [no-untyped-def]
 src/documents/models.py:0: error: Argument 1 to "Path" has incompatible type "Path | None"; expected "str | PathLike[str]"  [arg-type]
-src/documents/models.py:0: error: Could not resolve manager type for "documents.models.Document.deleted_objects"  [django-manager-missing]
-src/documents/models.py:0: error: Could not resolve manager type for "documents.models.Document.global_objects"  [django-manager-missing]
-src/documents/models.py:0: error: Could not resolve manager type for "documents.models.Document.objects"  [django-manager-missing]
-src/documents/models.py:0: error: Couldn't resolve related manager 'custom_fields' for relation 'documents.models.CustomFieldInstance.document'.  [django-manager-missing]
 src/documents/models.py:0: error: Couldn't resolve related manager 'documents' for relation 'documents.models.Document.correspondent'.  [django-manager-missing]
 src/documents/models.py:0: error: Couldn't resolve related manager 'documents' for relation 'documents.models.Document.document_type'.  [django-manager-missing]
 src/documents/models.py:0: error: Couldn't resolve related manager 'documents' for relation 'documents.models.Document.storage_path'.  [django-manager-missing]
 src/documents/models.py:0: error: Couldn't resolve related manager 'fields' for relation 'documents.models.CustomFieldInstance.field'.  [django-manager-missing]
-src/documents/models.py:0: error: Couldn't resolve related manager 'notes' for relation 'documents.models.Note.document'.  [django-manager-missing]
 src/documents/models.py:0: error: Couldn't resolve related manager 'runs' for relation 'documents.models.WorkflowRun.workflow'.  [django-manager-missing]
-src/documents/models.py:0: error: Couldn't resolve related manager 'share_links' for relation 'documents.models.ShareLink.document'.  [django-manager-missing]
-src/documents/models.py:0: error: Couldn't resolve related manager 'workflow_runs' for relation 'documents.models.WorkflowRun.document'.  [django-manager-missing]
 src/documents/models.py:0: error: Function is missing a return type annotation  [no-untyped-def]
 src/documents/models.py:0: error: Function is missing a return type annotation  [no-untyped-def]
 src/documents/models.py:0: error: Function is missing a return type annotation  [no-untyped-def]
@@ -982,10 +972,6 @@ src/documents/tests/test_bulk_edit.py:0: error: Function is missing a type annot
 src/documents/tests/test_bulk_edit.py:0: error: Function is missing a type annotation  [no-untyped-def]
 src/documents/tests/test_bulk_edit.py:0: error: Function is missing a type annotation  [no-untyped-def]
 src/documents/tests/test_bulk_edit.py:0: error: Function is missing a type annotation  [no-untyped-def]
-src/documents/tests/test_bulk_edit.py:0: error: Function is missing a type annotation  [no-untyped-def]
-src/documents/tests/test_bulk_edit.py:0: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]
-src/documents/tests/test_bulk_edit.py:0: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]
-src/documents/tests/test_bulk_edit.py:0: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]
 src/documents/tests/test_bulk_edit.py:0: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]
 src/documents/tests/test_bulk_edit.py:0: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]
 src/documents/tests/test_bulk_edit.py:0: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]
@@ -1011,7 +997,6 @@ src/documents/tests/test_bulk_edit.py:0: error: Item "dict[Any, Any]" of "Group
 src/documents/tests/test_bulk_edit.py:0: error: Item "dict[Any, Any]" of "Group | dict[Any, Any]" has no attribute "count"  [union-attr]
 src/documents/tests/test_bulk_edit.py:0: error: Too few arguments for "count" of "list"  [call-arg]
 src/documents/tests/test_bulk_edit.py:0: error: Too few arguments for "count" of "list"  [call-arg]
-src/documents/tests/test_bulk_edit.py:0: error: Unsupported operand types for - ("None" and "int")  [operator]
 src/documents/tests/test_caching.py:0: error: Function is missing a type annotation for one or more arguments  [no-untyped-def]
 src/documents/tests/test_classifier.py:0: error: "None" has no attribute "classes_"  [attr-defined]
 src/documents/tests/test_classifier.py:0: error: "None" has no attribute "classes_"  [attr-defined]

docs/api.md

@@ -211,6 +211,21 @@ However, querying the tasks endpoint with the returned UUID e.g.
 `/api/tasks/?task_id={uuid}` will provide information on the state of the
 consumption including the ID of a created document if consumption succeeded.
 
+## Document Versions
+
+Document versions are file-level versions linked to one root document.
+
+-   Root document metadata (title, tags, correspondent, document type, storage path, custom fields, permissions) remains shared.
+-   Version-specific file data (file, mime type, checksums, archive info, extracted text content) belongs to the selected/latest version.
+
+Version-aware endpoints:
+
+-   `GET /api/documents/{id}/`: returns root document data; `content` resolves to latest version content by default. Use `?version={version_id}` to resolve content for a specific version.
+-   `PATCH /api/documents/{id}/`: content updates target the selected version (`?version={version_id}`) or latest version by default; non-content metadata updates target the root document.
+-   `GET /api/documents/{id}/download/`, `GET /api/documents/{id}/preview/`, `GET /api/documents/{id}/thumb/`, `GET /api/documents/{id}/metadata/`: accept `?version={version_id}`.
+-   `POST /api/documents/{id}/update_version/`: uploads a new version using multipart form field `document` and optional `version_label`.
+-   `DELETE /api/documents/{root_id}/versions/{version_id}/`: deletes a non-root version.
+
 ## Permissions
 
 All objects (documents, tags, etc.) allow setting object-level permissions
@@ -300,13 +315,13 @@ The following methods are supported:
             -   `"doc": OUTPUT_DOCUMENT_INDEX` Optional index of the output document for split operations.
     -   Optional `parameters`:
         -   `"delete_original": true` to delete the original documents after editing.
-        -   `"update_document": true` to update the existing document with the edited PDF.
+        -   `"update_document": true` to add the edited PDF as a new version of the root document.
         -   `"include_metadata": true` to copy metadata from the original document to the edited document.
 -   `remove_password`
     -   Requires `parameters`:
         -   `"password": "PASSWORD_STRING"` The password to remove from the PDF documents.
     -   Optional `parameters`:
-        -   `"update_document": true` to replace the existing document with the password-less PDF.
+        -   `"update_document": true` to add the password-less PDF as a new version of the root document.
         -   `"delete_original": true` to delete the original document after editing.
         -   `"include_metadata": true` to copy metadata from the original document to the new password-less document.
 -   `merge`

docs/usage.md

@@ -89,6 +89,16 @@ You can view the document, edit its metadata, assign tags, correspondents,
 document types, and custom fields. You can also view the document history,
 download the document or share it via a share link.
 
+### Document File Versions
+
+Think of versions as **file history** for a document.
+
+-   Versions track the underlying file and extracted text content (OCR/text).
+-   Metadata such as tags, correspondent, document type, storage path and custom fields stay on the "root" document.
+-   By default, search and document content use the latest version.
+-   In document detail, selecting a version switches the preview, file metadata and content (and download etc buttons) to that version.
+-   Deleting a non-root version keeps metadata and falls back to the latest remaining version.
+
 ### Management Lists
 
 Paperless-ngx includes management lists for tags, correspondents, document types

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

@@ -1,7 +1,7 @@
 <pngx-page-header [(title)]="title" [id]="documentId">
   @if (archiveContentRenderType === ContentRenderType.PDF && !useNativePdfViewer) {
     @if (previewNumPages) {
-      <div class="input-group input-group-sm d-none d-md-flex">
+      <div class="input-group input-group-sm ms-2 d-none d-md-flex">
         <div class="input-group-text" i18n>Page</div>
           <input class="form-control flex-grow-0 w-auto" type="number" min="1" [max]="previewNumPages" [(ngModel)]="previewCurrentPage" />
         <div class="input-group-text" i18n>of {{previewNumPages}}</div>
@@ -24,6 +24,16 @@
     <i-bs width="1.2em" height="1.2em" name="trash"></i-bs><span class="d-none d-lg-inline ps-1" i18n>Delete</span>
   </button>
 
+  <pngx-document-version-dropdown
+    [documentId]="documentId"
+    [versions]="document?.versions ?? []"
+    [selectedVersionId]="selectedVersionId"
+    [userIsOwner]="userIsOwner"
+    [userCanEdit]="userCanEdit"
+    (versionSelected)="onVersionSelected($event)"
+    (versionsUpdated)="onVersionsUpdated($event)"
+  />
+
   <div class="btn-group">
     <button (click)="download()" class="btn btn-sm btn-outline-primary" [disabled]="downloading">
       @if (downloading) {

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

@@ -294,6 +294,27 @@ describe('DocumentDetailComponent', () => {
     component = fixture.componentInstance
   })
 
+  function initNormally() {
+    jest
+      .spyOn(activatedRoute, 'paramMap', 'get')
+      .mockReturnValue(of(convertToParamMap({ id: 3, section: 'details' })))
+    jest
+      .spyOn(documentService, 'get')
+      .mockReturnValueOnce(of(Object.assign({}, doc)))
+    jest.spyOn(openDocumentsService, 'getOpenDocument').mockReturnValue(null)
+    jest
+      .spyOn(openDocumentsService, 'openDocument')
+      .mockReturnValueOnce(of(true))
+    jest.spyOn(customFieldsService, 'listAll').mockReturnValue(
+      of({
+        count: customFields.length,
+        all: customFields.map((f) => f.id),
+        results: customFields,
+      })
+    )
+    fixture.detectChanges()
+  }
+
   it('should load four tabs via url params', () => {
     jest
       .spyOn(activatedRoute, 'paramMap', 'get')
@@ -354,6 +375,117 @@ describe('DocumentDetailComponent', () => {
     expect(component.document).toEqual(doc)
   })
 
+  it('should redirect to root when opening a version document id', () => {
+    const navigateSpy = jest.spyOn(router, 'navigate')
+    jest
+      .spyOn(activatedRoute, 'paramMap', 'get')
+      .mockReturnValue(of(convertToParamMap({ id: 10, section: 'details' })))
+    jest
+      .spyOn(documentService, 'get')
+      .mockReturnValueOnce(throwError(() => ({ status: 404 }) as any))
+    const getRootSpy = jest
+      .spyOn(documentService, 'getRootId')
+      .mockReturnValue(of({ root_id: 3 }))
+    jest.spyOn(openDocumentsService, 'getOpenDocument').mockReturnValue(null)
+    jest
+      .spyOn(openDocumentsService, 'openDocument')
+      .mockReturnValueOnce(of(true))
+    jest.spyOn(customFieldsService, 'listAll').mockReturnValue(
+      of({
+        count: customFields.length,
+        all: customFields.map((f) => f.id),
+        results: customFields,
+      })
+    )
+
+    fixture.detectChanges()
+    httpTestingController.expectOne(component.previewUrl).flush('preview')
+
+    expect(getRootSpy).toHaveBeenCalledWith(10)
+    expect(navigateSpy).toHaveBeenCalledWith(['documents', 3, 'details'], {
+      replaceUrl: true,
+    })
+  })
+
+  it('should navigate to 404 when root lookup fails', () => {
+    const navigateSpy = jest.spyOn(router, 'navigate')
+    jest
+      .spyOn(activatedRoute, 'paramMap', 'get')
+      .mockReturnValue(of(convertToParamMap({ id: 10, section: 'details' })))
+    jest
+      .spyOn(documentService, 'get')
+      .mockReturnValueOnce(throwError(() => ({ status: 404 }) as any))
+    jest
+      .spyOn(documentService, 'getRootId')
+      .mockReturnValue(throwError(() => new Error('boom')))
+    jest.spyOn(openDocumentsService, 'getOpenDocument').mockReturnValue(null)
+    jest
+      .spyOn(openDocumentsService, 'openDocument')
+      .mockReturnValueOnce(of(true))
+    jest.spyOn(customFieldsService, 'listAll').mockReturnValue(
+      of({
+        count: customFields.length,
+        all: customFields.map((f) => f.id),
+        results: customFields,
+      })
+    )
+
+    fixture.detectChanges()
+    httpTestingController.expectOne(component.previewUrl).flush('preview')
+
+    expect(navigateSpy).toHaveBeenCalledWith(['404'], { replaceUrl: true })
+  })
+
+  it('should not render a delete button for the root/original version', () => {
+    const docWithVersions = {
+      ...doc,
+      versions: [
+        {
+          id: doc.id,
+          added: new Date('2024-01-01T00:00:00Z'),
+          version_label: 'Original',
+          checksum: 'aaaa',
+          is_root: true,
+        },
+        {
+          id: 10,
+          added: new Date('2024-01-02T00:00:00Z'),
+          version_label: 'Edited',
+          checksum: 'bbbb',
+          is_root: false,
+        },
+      ],
+    } as Document
+
+    jest
+      .spyOn(activatedRoute, 'paramMap', 'get')
+      .mockReturnValue(of(convertToParamMap({ id: 3, section: 'details' })))
+    jest.spyOn(documentService, 'get').mockReturnValueOnce(of(docWithVersions))
+    jest
+      .spyOn(documentService, 'getMetadata')
+      .mockReturnValue(of({ has_archive_version: true } as any))
+    jest.spyOn(openDocumentsService, 'getOpenDocument').mockReturnValue(null)
+    jest
+      .spyOn(openDocumentsService, 'openDocument')
+      .mockReturnValueOnce(of(true))
+    jest.spyOn(customFieldsService, 'listAll').mockReturnValue(
+      of({
+        count: customFields.length,
+        all: customFields.map((f) => f.id),
+        results: customFields,
+      })
+    )
+
+    fixture.detectChanges()
+    httpTestingController.expectOne(component.previewUrl).flush('preview')
+    fixture.detectChanges()
+
+    const deleteButtons = fixture.debugElement.queryAll(
+      By.css('pngx-confirm-button')
+    )
+    expect(deleteButtons.length).toEqual(1)
+  })
+
   it('should fall back to details tab when duplicates tab is active but no duplicates', () => {
     initNormally()
     component.activeNavID = component.DocumentDetailNavIDs.Duplicates
@@ -532,6 +664,18 @@ describe('DocumentDetailComponent', () => {
     expect(navigateSpy).toHaveBeenCalledWith(['404'], { replaceUrl: true })
   })
 
+  it('discard should request the currently selected version', () => {
+    initNormally()
+    const getSpy = jest.spyOn(documentService, 'get')
+    getSpy.mockClear()
+    getSpy.mockReturnValueOnce(of(doc))
+
+    component.selectedVersionId = 10
+    component.discard()
+
+    expect(getSpy).toHaveBeenCalledWith(component.documentId, 10)
+  })
+
   it('should 404 on invalid id', () => {
     const navigateSpy = jest.spyOn(router, 'navigate')
     jest
@@ -584,6 +728,18 @@ describe('DocumentDetailComponent', () => {
     )
   })
 
+  it('save should target currently selected version', () => {
+    initNormally()
+    component.selectedVersionId = 10
+    const patchSpy = jest.spyOn(documentService, 'patch')
+    patchSpy.mockReturnValue(of(doc))
+
+    component.save()
+
+    expect(patchSpy).toHaveBeenCalled()
+    expect(patchSpy.mock.calls[0][1]).toEqual(10)
+  })
+
   it('should show toast error on save if error occurs', () => {
     currentUserHasObjectPermissions = true
     initNormally()
@@ -1036,7 +1192,32 @@ describe('DocumentDetailComponent', () => {
     const metadataSpy = jest.spyOn(documentService, 'getMetadata')
     metadataSpy.mockReturnValue(of({ has_archive_version: true }))
     initNormally()
-    expect(metadataSpy).toHaveBeenCalled()
+    expect(metadataSpy).toHaveBeenCalledWith(doc.id, null)
+  })
+
+  it('should pass metadata version only for non-latest selected versions', () => {
+    const metadataSpy = jest.spyOn(documentService, 'getMetadata')
+    metadataSpy.mockReturnValue(of({ has_archive_version: true }))
+    initNormally()
+    httpTestingController.expectOne(component.previewUrl).flush('preview')
+
+    expect(metadataSpy).toHaveBeenCalledWith(doc.id, null)
+
+    metadataSpy.mockClear()
+    component.document.versions = [
+      { id: doc.id, is_root: true },
+      { id: 10, is_root: false },
+    ] as any
+    jest.spyOn(documentService, 'getPreviewUrl').mockReturnValue('preview-root')
+    jest.spyOn(documentService, 'getThumbUrl').mockReturnValue('thumb-root')
+    jest
+      .spyOn(documentService, 'get')
+      .mockReturnValue(of({ content: 'root' } as Document))
+
+    component.selectVersion(doc.id)
+    httpTestingController.expectOne('preview-root').flush('root')
+
+    expect(metadataSpy).toHaveBeenCalledWith(doc.id, doc.id)
   })
 
   it('should show an error if failed metadata retrieval', () => {
@@ -1441,26 +1622,88 @@ describe('DocumentDetailComponent', () => {
     expect(closeSpy).toHaveBeenCalled()
   })
 
-  function initNormally() {
+  it('selectVersion should update preview and handle preview failures', () => {
-    jest
+    const previewSpy = jest.spyOn(documentService, 'getPreviewUrl')
-      .spyOn(activatedRoute, 'paramMap', 'get')
+    initNormally()
-      .mockReturnValue(of(convertToParamMap({ id: 3, section: 'details' })))
+    httpTestingController.expectOne(component.previewUrl).flush('preview')
+
+    previewSpy.mockReturnValueOnce('preview-version')
+    jest.spyOn(documentService, 'getThumbUrl').mockReturnValue('thumb-version')
     jest
       .spyOn(documentService, 'get')
-      .mockReturnValueOnce(of(Object.assign({}, doc)))
+      .mockReturnValue(of({ content: 'version-content' } as Document))
-    jest.spyOn(openDocumentsService, 'getOpenDocument').mockReturnValue(null)
+
+    component.selectVersion(10)
+    httpTestingController.expectOne('preview-version').flush('version text')
+
+    expect(component.previewUrl).toBe('preview-version')
+    expect(component.thumbUrl).toBe('thumb-version')
+    expect(component.previewText).toBe('version text')
+    expect(component.documentForm.get('content').value).toBe('version-content')
+    const pdfSource = component.pdfSource as { url: string; password?: string }
+    expect(pdfSource.url).toBe('preview-version')
+    expect(pdfSource.password).toBeUndefined()
+
+    previewSpy.mockReturnValueOnce('preview-error')
+    component.selectVersion(11)
+    httpTestingController
+      .expectOne('preview-error')
+      .error(new ErrorEvent('fail'))
+
+    expect(component.previewText).toContain('An error occurred loading content')
+  })
+
+  it('selectVersion should show toast if version content retrieval fails', () => {
+    initNormally()
+    httpTestingController.expectOne(component.previewUrl).flush('preview')
+
+    jest.spyOn(documentService, 'getPreviewUrl').mockReturnValue('preview-ok')
+    jest.spyOn(documentService, 'getThumbUrl').mockReturnValue('thumb-ok')
     jest
-      .spyOn(openDocumentsService, 'openDocument')
+      .spyOn(documentService, 'getMetadata')
-      .mockReturnValueOnce(of(true))
+      .mockReturnValue(of({ has_archive_version: true } as any))
-    jest.spyOn(customFieldsService, 'listAll').mockReturnValue(
+    const contentError = new Error('content failed')
-      of({
+    jest
-        count: customFields.length,
+      .spyOn(documentService, 'get')
-        all: customFields.map((f) => f.id),
+      .mockReturnValue(throwError(() => contentError))
-        results: customFields,
+    const toastSpy = jest.spyOn(toastService, 'showError')
-      })
+
+    component.selectVersion(10)
+    httpTestingController.expectOne('preview-ok').flush('preview text')
+
+    expect(toastSpy).toHaveBeenCalledWith(
+      'Error retrieving version content',
+      contentError
     )
-    fixture.detectChanges()
+  })
-  }
+
+  it('onVersionSelected should delegate to selectVersion', () => {
+    const selectVersionSpy = jest
+      .spyOn(component, 'selectVersion')
+      .mockImplementation(() => {})
+
+    component.onVersionSelected(42)
+
+    expect(selectVersionSpy).toHaveBeenCalledWith(42)
+  })
+
+  it('onVersionsUpdated should sync open document versions and save', () => {
+    component.documentId = doc.id
+    component.document = { ...doc, versions: [] } as Document
+    const updatedVersions = [
+      { id: doc.id, is_root: true },
+      { id: 10, is_root: false },
+    ] as any
+    const openDoc = { ...doc, versions: [] } as Document
+    jest.spyOn(openDocumentsService, 'getOpenDocument').mockReturnValue(openDoc)
+    const saveSpy = jest.spyOn(openDocumentsService, 'save')
+
+    component.onVersionsUpdated(updatedVersions)
+
+    expect(component.document.versions).toEqual(updatedVersions)
+    expect(openDoc.versions).toEqual(updatedVersions)
+    expect(saveSpy).toHaveBeenCalled()
+  })
 
   it('createDisabled should return true if the user does not have permission to add the specified data type', () => {
     currentUserCan = false
@@ -1554,6 +1797,70 @@ describe('DocumentDetailComponent', () => {
     expect(urlRevokeSpy).toHaveBeenCalled()
   })
 
+  it('should include version in download and print only for non-latest selected version', () => {
+    initNormally()
+    component.document.versions = [
+      { id: doc.id, is_root: true },
+      { id: 10, is_root: false },
+    ] as any
+
+    const getDownloadUrlSpy = jest
+      .spyOn(documentService, 'getDownloadUrl')
+      .mockReturnValueOnce('download-latest')
+      .mockReturnValueOnce('print-latest')
+      .mockReturnValueOnce('download-non-latest')
+      .mockReturnValueOnce('print-non-latest')
+
+    component.selectedVersionId = 10
+    component.download()
+    expect(getDownloadUrlSpy).toHaveBeenNthCalledWith(1, doc.id, false, null)
+    httpTestingController
+      .expectOne('download-latest')
+      .error(new ProgressEvent('failed'))
+
+    component.printDocument()
+    expect(getDownloadUrlSpy).toHaveBeenNthCalledWith(2, doc.id, false, null)
+    httpTestingController
+      .expectOne('print-latest')
+      .error(new ProgressEvent('failed'))
+
+    component.selectedVersionId = doc.id
+    component.download()
+    expect(getDownloadUrlSpy).toHaveBeenNthCalledWith(3, doc.id, false, doc.id)
+    httpTestingController
+      .expectOne('download-non-latest')
+      .error(new ProgressEvent('failed'))
+
+    component.printDocument()
+    expect(getDownloadUrlSpy).toHaveBeenNthCalledWith(4, doc.id, false, doc.id)
+    httpTestingController
+      .expectOne('print-non-latest')
+      .error(new ProgressEvent('failed'))
+  })
+
+  it('should omit version in download and print when no version is selected', () => {
+    initNormally()
+    component.document.versions = [] as any
+    ;(component as any).selectedVersionId = undefined
+
+    const getDownloadUrlSpy = jest
+      .spyOn(documentService, 'getDownloadUrl')
+      .mockReturnValueOnce('download-no-version')
+      .mockReturnValueOnce('print-no-version')
+
+    component.download()
+    expect(getDownloadUrlSpy).toHaveBeenNthCalledWith(1, doc.id, false, null)
+    httpTestingController
+      .expectOne('download-no-version')
+      .error(new ProgressEvent('failed'))
+
+    component.printDocument()
+    expect(getDownloadUrlSpy).toHaveBeenNthCalledWith(2, doc.id, false, null)
+    httpTestingController
+      .expectOne('print-no-version')
+      .error(new ProgressEvent('failed'))
+  })
+
   it('should download a file with the correct filename', () => {
     const mockBlob = new Blob(['test content'], { type: 'text/plain' })
     const mockResponse = new HttpResponse({

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

@@ -36,7 +36,7 @@ import { Correspondent } from 'src/app/data/correspondent'
 import { CustomField, CustomFieldDataType } from 'src/app/data/custom-field'
 import { CustomFieldInstance } from 'src/app/data/custom-field-instance'
 import { DataType } from 'src/app/data/datatype'
-import { Document } from 'src/app/data/document'
+import { Document, DocumentVersionInfo } from 'src/app/data/document'
 import { DocumentMetadata } from 'src/app/data/document-metadata'
 import { DocumentNote } from 'src/app/data/document-note'
 import { DocumentSuggestions } from 'src/app/data/document-suggestions'
@@ -120,6 +120,7 @@ import { SuggestionsDropdownComponent } from '../common/suggestions-dropdown/sug
 import { DocumentNotesComponent } from '../document-notes/document-notes.component'
 import { ComponentWithPermissions } from '../with-permissions/with-permissions.component'
 import { DocumentHistoryComponent } from './document-history/document-history.component'
+import { DocumentVersionDropdownComponent } from './document-version-dropdown/document-version-dropdown.component'
 import { MetadataCollapseComponent } from './metadata-collapse/metadata-collapse.component'
 
 enum DocumentDetailNavIDs {
@@ -177,6 +178,7 @@ enum ContentRenderType {
     TextAreaComponent,
     RouterModule,
     PngxPdfViewerComponent,
+    DocumentVersionDropdownComponent,
   ],
 })
 export class DocumentDetailComponent
@@ -184,6 +186,7 @@ export class DocumentDetailComponent
   implements OnInit, OnDestroy, DirtyComponent
 {
   PdfRenderMode = PdfRenderMode
+
   documentsService = inject(DocumentService)
   private route = inject(ActivatedRoute)
   private tagService = inject(TagService)
@@ -235,6 +238,9 @@ export class DocumentDetailComponent
   tiffURL: string
   tiffError: string
 
+  // Versioning
+  selectedVersionId: number
+
   correspondents: Correspondent[]
   documentTypes: DocumentType[]
   storagePaths: StoragePath[]
@@ -312,13 +318,19 @@ export class DocumentDetailComponent
   }
 
   get archiveContentRenderType(): ContentRenderType {
-    return this.document?.archived_file_name
+    const hasArchiveVersion =
+      this.metadata?.has_archive_version ?? !!this.document?.archived_file_name
+    return hasArchiveVersion
       ? this.getRenderType('application/pdf')
-      : this.getRenderType(this.document?.mime_type)
+      : this.getRenderType(
+          this.metadata?.original_mime_type || this.document?.mime_type
+        )
   }
 
   get originalContentRenderType(): ContentRenderType {
-    return this.getRenderType(this.document?.mime_type)
+    return this.getRenderType(
+      this.metadata?.original_mime_type || this.document?.mime_type
+    )
   }
 
   get showThumbnailOverlay(): boolean {
@@ -348,16 +360,46 @@ export class DocumentDetailComponent
   }
 
   private updatePdfSource() {
-    if (!this.previewUrl) {
-      this.pdfSource = undefined
-      return
-    }
     this.pdfSource = {
       url: this.previewUrl,
-      password: this.password || undefined,
+      password: this.password,
     }
   }
 
+  private loadMetadataForSelectedVersion() {
+    const selectedVersionId = this.getSelectedNonLatestVersionId()
+    this.documentsService
+      .getMetadata(this.documentId, selectedVersionId)
+      .pipe(
+        first(),
+        takeUntil(this.unsubscribeNotifier),
+        takeUntil(this.docChangeNotifier)
+      )
+      .subscribe({
+        next: (result) => {
+          this.metadata = result
+          this.tiffURL = null
+          this.tiffError = null
+          if (this.archiveContentRenderType === ContentRenderType.TIFF) {
+            this.tryRenderTiff()
+          }
+          if (
+            this.archiveContentRenderType !== ContentRenderType.PDF ||
+            this.useNativePdfViewer
+          ) {
+            this.previewLoaded = true
+          }
+        },
+        error: (error) => {
+          this.metadata = {} // allow display to fallback to <object> tag
+          this.toastService.showError(
+            $localize`Error retrieving metadata`,
+            error
+          )
+        },
+      })
+  }
+
   get isRTL() {
     if (!this.metadata || !this.metadata.lang) return false
     else {
@@ -433,7 +475,11 @@ export class DocumentDetailComponent
   }
 
   private loadDocument(documentId: number): void {
-    this.previewUrl = this.documentsService.getPreviewUrl(documentId)
+    let redirectedToRoot = false
+    this.selectedVersionId = documentId
+    this.previewUrl = this.documentsService.getPreviewUrl(
+      this.selectedVersionId
+    )
     this.updatePdfSource()
     this.http
       .get(this.previewUrl, { responseType: 'text' })
@@ -449,11 +495,29 @@ export class DocumentDetailComponent
             err.message ?? err.toString()
           }`),
       })
-    this.thumbUrl = this.documentsService.getThumbUrl(documentId)
+    this.thumbUrl = this.documentsService.getThumbUrl(this.selectedVersionId)
     this.documentsService
       .get(documentId)
       .pipe(
-        catchError(() => {
+        catchError((error) => {
+          if (error?.status === 404) {
+            // if not found, check if there's root document that exists and redirect if so
+            return this.documentsService.getRootId(documentId).pipe(
+              map((result) => {
+                const rootId = result?.root_id
+                if (rootId && rootId !== documentId) {
+                  const section =
+                    this.route.snapshot.paramMap.get('section') || 'details'
+                  redirectedToRoot = true
+                  this.router.navigate(['documents', rootId, section], {
+                    replaceUrl: true,
+                  })
+                }
+                return null
+              }),
+              catchError(() => of(null))
+            )
+          }
           // 404 is handled in the subscribe below
           return of(null)
         }),
@@ -464,6 +528,9 @@ export class DocumentDetailComponent
       .subscribe({
         next: (doc) => {
           if (!doc) {
+            if (redirectedToRoot) {
+              return
+            }
             this.router.navigate(['404'], { replaceUrl: true })
             return
           }
@@ -680,36 +747,15 @@ export class DocumentDetailComponent
 
   updateComponent(doc: Document) {
     this.document = doc
+    // Default selected version is the newest version
+    const versions = doc.versions ?? []
+    this.selectedVersionId = versions.length
+      ? Math.max(...versions.map((version) => version.id))
+      : doc.id
+    this.previewLoaded = false
     this.requiresPassword = false
     this.updateFormForCustomFields()
-    if (this.archiveContentRenderType === ContentRenderType.TIFF) {
+    this.loadMetadataForSelectedVersion()
-      this.tryRenderTiff()
-    }
-    this.documentsService
-      .getMetadata(doc.id)
-      .pipe(
-        first(),
-        takeUntil(this.unsubscribeNotifier),
-        takeUntil(this.docChangeNotifier)
-      )
-      .subscribe({
-        next: (result) => {
-          this.metadata = result
-          if (
-            this.archiveContentRenderType !== ContentRenderType.PDF ||
-            this.useNativePdfViewer
-          ) {
-            this.previewLoaded = true
-          }
-        },
-        error: (error) => {
-          this.metadata = {} // allow display to fallback to <object> tag
-          this.toastService.showError(
-            $localize`Error retrieving metadata`,
-            error
-          )
-        },
-      })
     if (
       this.permissionsService.currentUserHasObjectPermissions(
         PermissionAction.Change,
@@ -738,6 +784,78 @@ export class DocumentDetailComponent
     }
   }
 
+  // Update file preview and download target to a specific version (by document id)
+  selectVersion(versionId: number) {
+    this.selectedVersionId = versionId
+    this.previewLoaded = false
+    this.previewUrl = this.documentsService.getPreviewUrl(
+      this.documentId,
+      false,
+      this.selectedVersionId
+    )
+    this.updatePdfSource()
+    this.thumbUrl = this.documentsService.getThumbUrl(
+      this.documentId,
+      this.selectedVersionId
+    )
+    this.loadMetadataForSelectedVersion()
+    this.documentsService
+      .get(this.documentId, this.selectedVersionId, 'content')
+      .pipe(
+        first(),
+        takeUntil(this.unsubscribeNotifier),
+        takeUntil(this.docChangeNotifier)
+      )
+      .subscribe({
+        next: (doc) => {
+          const content = doc?.content ?? ''
+          this.document.content = content
+          this.documentForm.patchValue(
+            {
+              content,
+            },
+            {
+              emitEvent: false,
+            }
+          )
+        },
+        error: (error) => {
+          this.toastService.showError(
+            $localize`Error retrieving version content`,
+            error
+          )
+        },
+      })
+    // For text previews, refresh content
+    this.http
+      .get(this.previewUrl, { responseType: 'text' })
+      .pipe(
+        first(),
+        takeUntil(this.unsubscribeNotifier),
+        takeUntil(this.docChangeNotifier)
+      )
+      .subscribe({
+        next: (res) => (this.previewText = res.toString()),
+        error: (err) =>
+          (this.previewText = $localize`An error occurred loading content: ${
+            err.message ?? err.toString()
+          }`),
+      })
+  }
+
+  onVersionSelected(versionId: number) {
+    this.selectVersion(versionId)
+  }
+
+  onVersionsUpdated(versions: DocumentVersionInfo[]) {
+    this.document.versions = versions
+    const openDoc = this.openDocumentService.getOpenDocument(this.documentId)
+    if (openDoc) {
+      openDoc.versions = versions
+      this.openDocumentService.save()
+    }
+  }
+
   get customFieldFormFields(): FormArray {
     return this.documentForm.get('custom_fields') as FormArray
   }
@@ -906,7 +1024,7 @@ export class DocumentDetailComponent
 
   discard() {
     this.documentsService
-      .get(this.documentId)
+      .get(this.documentId, this.selectedVersionId)
       .pipe(
         first(),
         takeUntil(this.unsubscribeNotifier),
@@ -956,7 +1074,7 @@ export class DocumentDetailComponent
     this.networkActive = true
     ;(document.activeElement as HTMLElement)?.dispatchEvent(new Event('change'))
     this.documentsService
-      .patch(this.getChangedFields())
+      .patch(this.getChangedFields(), this.selectedVersionId)
       .pipe(first())
       .subscribe({
         next: (docValues) => {
@@ -1011,7 +1129,7 @@ export class DocumentDetailComponent
     this.networkActive = true
     this.store.next(this.documentForm.value)
     this.documentsService
-      .patch(this.getChangedFields())
+      .patch(this.getChangedFields(), this.selectedVersionId)
       .pipe(
         switchMap((updateResult) => {
           this.savedViewService.maybeRefreshDocumentCounts()
@@ -1154,11 +1272,24 @@ export class DocumentDetailComponent
     })
   }
 
+  private getSelectedNonLatestVersionId(): number | null {
+    const versions = this.document?.versions ?? []
+    if (!versions.length || !this.selectedVersionId) {
+      return null
+    }
+    const latestVersionId = Math.max(...versions.map((version) => version.id))
+    return this.selectedVersionId === latestVersionId
+      ? null
+      : this.selectedVersionId
+  }
+
   download(original: boolean = false) {
     this.downloading = true
+    const selectedVersionId = this.getSelectedNonLatestVersionId()
     const downloadUrl = this.documentsService.getDownloadUrl(
       this.documentId,
-      original
+      original,
+      selectedVersionId
     )
     this.http
       .get(downloadUrl, { observe: 'response', responseType: 'blob' })
@@ -1590,9 +1721,11 @@ export class DocumentDetailComponent
   }
 
   printDocument() {
+    const selectedVersionId = this.getSelectedNonLatestVersionId()
     const printUrl = this.documentsService.getDownloadUrl(
       this.document.id,
-      false
+      false,
+      selectedVersionId
     )
     this.http
       .get(printUrl, { responseType: 'blob' })
@@ -1640,7 +1773,7 @@ export class DocumentDetailComponent
     const modal = this.modalService.open(ShareLinksDialogComponent)
     modal.componentInstance.documentId = this.document.id
     modal.componentInstance.hasArchiveVersion =
-      !!this.document?.archived_file_name
+      this.metadata?.has_archive_version ?? !!this.document?.archived_file_name
   }
 
   get emailEnabled(): boolean {
@@ -1653,7 +1786,7 @@ export class DocumentDetailComponent
     })
     modal.componentInstance.documentIds = [this.document.id]
     modal.componentInstance.hasArchiveVersion =
-      !!this.document?.archived_file_name
+      this.metadata?.has_archive_version ?? !!this.document?.archived_file_name
   }
 
   private tryRenderTiff() {

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

@@ -0,0 +1,97 @@
+<div class="btn-group" ngbDropdown autoClose="outside">
+  <button class="btn btn-sm btn-outline-secondary dropdown-toggle" ngbDropdownToggle>
+    <i-bs name="file-earmark-diff"></i-bs>
+    <span class="d-none d-lg-inline ps-1" i18n>Versions</span>
+  </button>
+  <div class="dropdown-menu shadow" ngbDropdownMenu>
+    <div class="px-3 py-2">
+      @if (versionUploadState === UploadState.Idle) {
+        <div class="input-group input-group-sm mb-2">
+          <span class="input-group-text" i18n>Label</span>
+          <input
+            class="form-control"
+            type="text"
+            [(ngModel)]="newVersionLabel"
+            i18n-placeholder
+            placeholder="Optional"
+            [disabled]="!userIsOwner || !userCanEdit"
+          />
+        </div>
+        <input
+          #versionFileInput
+          type="file"
+          class="visually-hidden"
+          (change)="onVersionFileSelected($event)"
+        />
+        <button
+          class="btn btn-sm btn-outline-secondary w-100"
+          (click)="versionFileInput.click()"
+          [disabled]="!userIsOwner || !userCanEdit"
+        >
+          <i-bs name="file-earmark-plus"></i-bs><span class="ps-1" i18n>Add new version</span>
+        </button>
+      } @else {
+        @switch (versionUploadState) {
+          @case (UploadState.Uploading) {
+            <div class="small text-muted mt-1 d-flex align-items-center">
+              <output class="spinner-border spinner-border-sm me-2" aria-hidden="true"></output>
+              <span i18n>Uploading version...</span>
+            </div>
+          }
+          @case (UploadState.Processing) {
+            <div class="small text-muted mt-1 d-flex align-items-center">
+              <output class="spinner-border spinner-border-sm me-2" aria-hidden="true"></output>
+              <span i18n>Processing version...</span>
+            </div>
+          }
+          @case (UploadState.Failed) {
+            <div class="small text-danger mt-1 d-flex align-items-center justify-content-between">
+              <span i18n>Version upload failed.</span>
+              <button type="button" class="btn btn-link btn-sm p-0 ms-2" (click)="clearVersionUploadStatus()" i18n>Dismiss</button>
+            </div>
+            @if (versionUploadError) {
+              <div class="small text-muted mt-1">{{ versionUploadError }}</div>
+            }
+          }
+        }
+      }
+    </div>
+    <div class="dropdown-divider"></div>
+    @for (version of versions; track version.id) {
+      <div class="dropdown-item">
+        <div class="d-flex align-items-center w-100 version-item">
+          <button type="button" class="btn btn-link link-underline link-underline-opacity-0 d-flex align-items-center flex-grow-1 small ps-0 text-start" (click)="selectVersion(version.id)">
+            <div class="badge bg-light text-lowercase text-muted">
+              {{ version.checksum | slice:0:8 }}
+            </div>
+            <div class="d-flex flex-column small ms-2">
+              <div>
+                @if (version.version_label) {
+                  {{ version.version_label }}
+                } @else {
+                  <span i18n>ID</span> #{{version.id}}
+                }
+              </div>
+              <div class="text-muted">
+                {{ version.added | customDate:'short' }}
+              </div>
+            </div>
+          </button>
+          @if (selectedVersionId === version.id) { <span class="ms-2">✓</span> }
+          @if (!version.is_root) {
+            <pngx-confirm-button
+              buttonClasses="btn-link btn-sm text-danger ms-2"
+              iconName="trash"
+              confirmMessage="Delete this version?"
+              i18n-confirmMessage
+              [disabled]="!userIsOwner || !userCanEdit"
+              (confirm)="deleteVersion(version.id)"
+            >
+              <span class="visually-hidden" i18n>Delete version</span>
+            </pngx-confirm-button>
+          }
+        </div>
+      </div>
+    }
+  </div>
+</div>

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

@@ -0,0 +1,226 @@
+import { DatePipe } from '@angular/common'
+import { SimpleChange } from '@angular/core'
+import { ComponentFixture, TestBed } from '@angular/core/testing'
+import { NgxBootstrapIconsModule, allIcons } from 'ngx-bootstrap-icons'
+import { Subject, of, throwError } from 'rxjs'
+import { DocumentVersionInfo } from 'src/app/data/document'
+import { DocumentService } from 'src/app/services/rest/document.service'
+import { SettingsService } from 'src/app/services/settings.service'
+import { ToastService } from 'src/app/services/toast.service'
+import {
+  UploadState,
+  WebsocketStatusService,
+} from 'src/app/services/websocket-status.service'
+import { DocumentVersionDropdownComponent } from './document-version-dropdown.component'
+
+describe('DocumentVersionDropdownComponent', () => {
+  let component: DocumentVersionDropdownComponent
+  let fixture: ComponentFixture<DocumentVersionDropdownComponent>
+  let documentService: jest.Mocked<
+    Pick<DocumentService, 'deleteVersion' | 'getVersions' | 'uploadVersion'>
+  >
+  let toastService: jest.Mocked<Pick<ToastService, 'showError' | 'showInfo'>>
+  let finished$: Subject<{ taskId: string }>
+  let failed$: Subject<{ taskId: string; message?: string }>
+
+  beforeEach(async () => {
+    finished$ = new Subject<{ taskId: string }>()
+    failed$ = new Subject<{ taskId: string; message?: string }>()
+    documentService = {
+      deleteVersion: jest.fn(),
+      getVersions: jest.fn(),
+      uploadVersion: jest.fn(),
+    }
+    toastService = {
+      showError: jest.fn(),
+      showInfo: jest.fn(),
+    }
+
+    await TestBed.configureTestingModule({
+      imports: [
+        DocumentVersionDropdownComponent,
+        NgxBootstrapIconsModule.pick(allIcons),
+      ],
+      providers: [
+        DatePipe,
+        {
+          provide: DocumentService,
+          useValue: documentService,
+        },
+        {
+          provide: SettingsService,
+          useValue: {
+            get: () => null,
+          },
+        },
+        {
+          provide: ToastService,
+          useValue: toastService,
+        },
+        {
+          provide: WebsocketStatusService,
+          useValue: {
+            onDocumentConsumptionFinished: () => finished$,
+            onDocumentConsumptionFailed: () => failed$,
+          },
+        },
+      ],
+    }).compileComponents()
+
+    fixture = TestBed.createComponent(DocumentVersionDropdownComponent)
+    component = fixture.componentInstance
+    component.documentId = 3
+    component.selectedVersionId = 3
+    component.versions = [
+      {
+        id: 3,
+        is_root: true,
+        checksum: 'aaaa',
+      },
+      {
+        id: 10,
+        is_root: false,
+        checksum: 'bbbb',
+      },
+    ]
+    fixture.detectChanges()
+  })
+
+  it('selectVersion should emit the selected id', () => {
+    const emitSpy = jest.spyOn(component.versionSelected, 'emit')
+    component.selectVersion(10)
+    expect(emitSpy).toHaveBeenCalledWith(10)
+  })
+
+  it('deleteVersion should refresh versions and select fallback when deleting current selection', () => {
+    const updatedVersions: DocumentVersionInfo[] = [
+      { id: 3, is_root: true, checksum: 'aaaa' },
+      { id: 20, is_root: false, checksum: 'cccc' },
+    ]
+    component.selectedVersionId = 10
+    documentService.deleteVersion.mockReturnValue(
+      of({ result: 'deleted', current_version_id: 3 })
+    )
+    documentService.getVersions.mockReturnValue(
+      of({ id: 3, versions: updatedVersions } as any)
+    )
+    const versionsEmitSpy = jest.spyOn(component.versionsUpdated, 'emit')
+    const selectedEmitSpy = jest.spyOn(component.versionSelected, 'emit')
+
+    component.deleteVersion(10)
+
+    expect(documentService.deleteVersion).toHaveBeenCalledWith(3, 10)
+    expect(documentService.getVersions).toHaveBeenCalledWith(3)
+    expect(versionsEmitSpy).toHaveBeenCalledWith(updatedVersions)
+    expect(selectedEmitSpy).toHaveBeenCalledWith(3)
+  })
+
+  it('deleteVersion should show an error toast on failure', () => {
+    const error = new Error('delete failed')
+    documentService.deleteVersion.mockReturnValue(throwError(() => error))
+
+    component.deleteVersion(10)
+
+    expect(toastService.showError).toHaveBeenCalledWith(
+      'Error deleting version',
+      error
+    )
+  })
+
+  it('onVersionFileSelected should upload and update versions after websocket success', () => {
+    const versions: DocumentVersionInfo[] = [
+      { id: 3, is_root: true, checksum: 'aaaa' },
+      { id: 20, is_root: false, checksum: 'cccc' },
+    ]
+    const file = new File(['test'], 'new-version.pdf', {
+      type: 'application/pdf',
+    })
+    const input = document.createElement('input')
+    Object.defineProperty(input, 'files', { value: [file] })
+    component.newVersionLabel = '  Updated scan  '
+    documentService.uploadVersion.mockReturnValue(
+      of({ task_id: 'task-1' } as any)
+    )
+    documentService.getVersions.mockReturnValue(of({ id: 3, versions } as any))
+    const versionsEmitSpy = jest.spyOn(component.versionsUpdated, 'emit')
+    const selectedEmitSpy = jest.spyOn(component.versionSelected, 'emit')
+
+    component.onVersionFileSelected({ target: input } as Event)
+    finished$.next({ taskId: 'task-1' })
+
+    expect(documentService.uploadVersion).toHaveBeenCalledWith(
+      3,
+      file,
+      'Updated scan'
+    )
+    expect(toastService.showInfo).toHaveBeenCalled()
+    expect(documentService.getVersions).toHaveBeenCalledWith(3)
+    expect(versionsEmitSpy).toHaveBeenCalledWith(versions)
+    expect(selectedEmitSpy).toHaveBeenCalledWith(20)
+    expect(component.newVersionLabel).toEqual('')
+    expect(component.versionUploadState).toEqual(UploadState.Idle)
+    expect(component.versionUploadError).toBeNull()
+  })
+
+  it('onVersionFileSelected should set failed state after websocket failure', () => {
+    const file = new File(['test'], 'new-version.pdf', {
+      type: 'application/pdf',
+    })
+    const input = document.createElement('input')
+    Object.defineProperty(input, 'files', { value: [file] })
+    documentService.uploadVersion.mockReturnValue(of('task-1'))
+
+    component.onVersionFileSelected({ target: input } as Event)
+    failed$.next({ taskId: 'task-1', message: 'processing failed' })
+
+    expect(component.versionUploadState).toEqual(UploadState.Failed)
+    expect(component.versionUploadError).toEqual('processing failed')
+    expect(documentService.getVersions).not.toHaveBeenCalled()
+  })
+
+  it('onVersionFileSelected should fail when backend response has no task id', () => {
+    const file = new File(['test'], 'new-version.pdf', {
+      type: 'application/pdf',
+    })
+    const input = document.createElement('input')
+    Object.defineProperty(input, 'files', { value: [file] })
+    documentService.uploadVersion.mockReturnValue(of({} as any))
+
+    component.onVersionFileSelected({ target: input } as Event)
+
+    expect(component.versionUploadState).toEqual(UploadState.Failed)
+    expect(component.versionUploadError).toEqual('Missing task ID.')
+    expect(documentService.getVersions).not.toHaveBeenCalled()
+  })
+
+  it('onVersionFileSelected should show error when upload request fails', () => {
+    const file = new File(['test'], 'new-version.pdf', {
+      type: 'application/pdf',
+    })
+    const input = document.createElement('input')
+    Object.defineProperty(input, 'files', { value: [file] })
+    const error = new Error('upload failed')
+    documentService.uploadVersion.mockReturnValue(throwError(() => error))
+
+    component.onVersionFileSelected({ target: input } as Event)
+
+    expect(component.versionUploadState).toEqual(UploadState.Failed)
+    expect(component.versionUploadError).toEqual('upload failed')
+    expect(toastService.showError).toHaveBeenCalledWith(
+      'Error uploading new version',
+      error
+    )
+  })
+
+  it('ngOnChanges should clear upload status on document switch', () => {
+    component.versionUploadState = UploadState.Failed
+    component.versionUploadError = 'something failed'
+
+    component.ngOnChanges({
+      documentId: new SimpleChange(3, 4, false),
+    })
+
+    expect(component.versionUploadState).toEqual(UploadState.Idle)
+    expect(component.versionUploadError).toBeNull()
+  })
+})

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

@@ -0,0 +1,203 @@
+import { SlicePipe } from '@angular/common'
+import {
+  Component,
+  EventEmitter,
+  inject,
+  Input,
+  OnChanges,
+  OnDestroy,
+  Output,
+  SimpleChanges,
+} from '@angular/core'
+import { FormsModule } from '@angular/forms'
+import { NgbDropdownModule } from '@ng-bootstrap/ng-bootstrap'
+import { NgxBootstrapIconsModule } from 'ngx-bootstrap-icons'
+import { merge, of, Subject } from 'rxjs'
+import {
+  filter,
+  first,
+  map,
+  switchMap,
+  take,
+  takeUntil,
+  tap,
+} from 'rxjs/operators'
+import { DocumentVersionInfo } from 'src/app/data/document'
+import { CustomDatePipe } from 'src/app/pipes/custom-date.pipe'
+import { DocumentService } from 'src/app/services/rest/document.service'
+import { ToastService } from 'src/app/services/toast.service'
+import {
+  UploadState,
+  WebsocketStatusService,
+} from 'src/app/services/websocket-status.service'
+import { ConfirmButtonComponent } from '../../common/confirm-button/confirm-button.component'
+
+@Component({
+  selector: 'pngx-document-version-dropdown',
+  templateUrl: './document-version-dropdown.component.html',
+  imports: [
+    FormsModule,
+    NgbDropdownModule,
+    NgxBootstrapIconsModule,
+    ConfirmButtonComponent,
+    SlicePipe,
+    CustomDatePipe,
+  ],
+})
+export class DocumentVersionDropdownComponent implements OnChanges, OnDestroy {
+  UploadState = UploadState
+
+  @Input() documentId: number
+  @Input() versions: DocumentVersionInfo[] = []
+  @Input() selectedVersionId: number
+  @Input() userCanEdit: boolean = false
+  @Input() userIsOwner: boolean = false
+
+  @Output() versionSelected = new EventEmitter<number>()
+  @Output() versionsUpdated = new EventEmitter<DocumentVersionInfo[]>()
+
+  newVersionLabel: string = ''
+  versionUploadState: UploadState = UploadState.Idle
+  versionUploadError: string | null = null
+
+  private readonly documentsService = inject(DocumentService)
+  private readonly toastService = inject(ToastService)
+  private readonly websocketStatusService = inject(WebsocketStatusService)
+  private readonly destroy$ = new Subject<void>()
+  private readonly documentChange$ = new Subject<void>()
+
+  ngOnChanges(changes: SimpleChanges): void {
+    if (changes.documentId && !changes.documentId.firstChange) {
+      this.documentChange$.next()
+      this.clearVersionUploadStatus()
+    }
+  }
+
+  ngOnDestroy(): void {
+    this.documentChange$.next()
+    this.documentChange$.complete()
+    this.destroy$.next()
+    this.destroy$.complete()
+  }
+
+  selectVersion(versionId: number): void {
+    this.versionSelected.emit(versionId)
+  }
+
+  deleteVersion(versionId: number): void {
+    const wasSelected = this.selectedVersionId === versionId
+    this.documentsService
+      .deleteVersion(this.documentId, versionId)
+      .pipe(
+        switchMap((result) =>
+          this.documentsService
+            .getVersions(this.documentId)
+            .pipe(map((doc) => ({ doc, result })))
+        ),
+        first(),
+        takeUntil(this.destroy$)
+      )
+      .subscribe({
+        next: ({ doc, result }) => {
+          if (doc?.versions) {
+            this.versionsUpdated.emit(doc.versions)
+          }
+
+          if (wasSelected || this.selectedVersionId === versionId) {
+            const fallbackId = result?.current_version_id ?? this.documentId
+            this.versionSelected.emit(fallbackId)
+          }
+        },
+        error: (error) => {
+          this.toastService.showError($localize`Error deleting version`, error)
+        },
+      })
+  }
+
+  onVersionFileSelected(event: Event): void {
+    const input = event.target as HTMLInputElement
+    if (!input?.files || input.files.length === 0) return
+    const uploadDocumentId = this.documentId
+    const file = input.files[0]
+    input.value = ''
+    const label = this.newVersionLabel?.trim()
+    this.versionUploadState = UploadState.Uploading
+    this.versionUploadError = null
+    this.documentsService
+      .uploadVersion(uploadDocumentId, file, label)
+      .pipe(
+        first(),
+        tap(() => {
+          this.toastService.showInfo(
+            $localize`Uploading new version. Processing will happen in the background.`
+          )
+          this.newVersionLabel = ''
+          this.versionUploadState = UploadState.Processing
+        }),
+        map((taskId) =>
+          typeof taskId === 'string'
+            ? taskId
+            : (taskId as { task_id?: string })?.task_id
+        ),
+        switchMap((taskId) => {
+          if (!taskId) {
+            this.versionUploadState = UploadState.Failed
+            this.versionUploadError = $localize`Missing task ID.`
+            return of(null)
+          }
+          return merge(
+            this.websocketStatusService.onDocumentConsumptionFinished().pipe(
+              filter((status) => status.taskId === taskId),
+              map(() => ({ state: 'success' as const }))
+            ),
+            this.websocketStatusService.onDocumentConsumptionFailed().pipe(
+              filter((status) => status.taskId === taskId),
+              map((status) => ({
+                state: 'failed' as const,
+                message: status.message,
+              }))
+            )
+          ).pipe(takeUntil(merge(this.destroy$, this.documentChange$)), take(1))
+        }),
+        switchMap((result) => {
+          if (result?.state !== 'success') {
+            if (result?.state === 'failed') {
+              this.versionUploadState = UploadState.Failed
+              this.versionUploadError =
+                result.message || $localize`Upload failed.`
+            }
+            return of(null)
+          }
+          return this.documentsService.getVersions(uploadDocumentId)
+        }),
+        takeUntil(this.destroy$),
+        takeUntil(this.documentChange$)
+      )
+      .subscribe({
+        next: (doc) => {
+          if (uploadDocumentId !== this.documentId) return
+          if (doc?.versions) {
+            this.versionsUpdated.emit(doc.versions)
+            this.versionSelected.emit(
+              Math.max(...doc.versions.map((version) => version.id))
+            )
+            this.clearVersionUploadStatus()
+          }
+        },
+        error: (error) => {
+          if (uploadDocumentId !== this.documentId) return
+          this.versionUploadState = UploadState.Failed
+          this.versionUploadError = error?.message || $localize`Upload failed.`
+          this.toastService.showError(
+            $localize`Error uploading new version`,
+            error
+          )
+        },
+      })
+  }
+
+  clearVersionUploadStatus(): void {
+    this.versionUploadState = UploadState.Idle
+    this.versionUploadError = null
+  }
+}

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

@@ -161,6 +161,18 @@ export interface Document extends ObjectWithPermissions {
 
   duplicate_documents?: Document[]
 
+  // Versioning
+  root_document?: number
+  versions?: DocumentVersionInfo[]
+
   // Frontend only
   __changedFields?: string[]
 }
+
+export interface DocumentVersionInfo {
+  id: number
+  added?: Date
+  version_label?: string
+  checksum?: string
+  is_root: boolean
+}

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

@@ -165,6 +165,14 @@ describe(`DocumentService`, () => {
     expect(req.request.method).toEqual('GET')
   })
 
+  it('should call appropriate api endpoint for versioned metadata', () => {
+    subscription = service.getMetadata(documents[0].id, 123).subscribe()
+    const req = httpTestingController.expectOne(
+      `${environment.apiBaseUrl}${endpoint}/${documents[0].id}/metadata/?version=123`
+    )
+    expect(req.request.method).toEqual('GET')
+  })
+
   it('should call appropriate api endpoint for getting selection data', () => {
     const ids = [documents[0].id]
     subscription = service.getSelectionData(ids).subscribe()
@@ -233,11 +241,22 @@ describe(`DocumentService`, () => {
     )
   })
 
+  it('should return the correct preview URL for a specific version', () => {
+    const url = service.getPreviewUrl(documents[0].id, false, 123)
+    expect(url).toEqual(
+      `${environment.apiBaseUrl}${endpoint}/${documents[0].id}/preview/?version=123`
+    )
+  })
+
   it('should return the correct thumb URL for a single document', () => {
     let url = service.getThumbUrl(documents[0].id)
     expect(url).toEqual(
       `${environment.apiBaseUrl}${endpoint}/${documents[0].id}/thumb/`
     )
+    url = service.getThumbUrl(documents[0].id, 123)
+    expect(url).toEqual(
+      `${environment.apiBaseUrl}${endpoint}/${documents[0].id}/thumb/?version=123`
+    )
   })
 
   it('should return the correct download URL for a single document', () => {
@@ -249,6 +268,22 @@ describe(`DocumentService`, () => {
     expect(url).toEqual(
       `${environment.apiBaseUrl}${endpoint}/${documents[0].id}/download/?original=true`
     )
+    url = service.getDownloadUrl(documents[0].id, false, 123)
+    expect(url).toEqual(
+      `${environment.apiBaseUrl}${endpoint}/${documents[0].id}/download/?version=123`
+    )
+    url = service.getDownloadUrl(documents[0].id, true, 123)
+    expect(url).toEqual(
+      `${environment.apiBaseUrl}${endpoint}/${documents[0].id}/download/?original=true&version=123`
+    )
+  })
+
+  it('should pass optional get params for version and fields', () => {
+    subscription = service.get(documents[0].id, 123, 'content').subscribe()
+    const req = httpTestingController.expectOne(
+      `${environment.apiBaseUrl}${endpoint}/${documents[0].id}/?full_perms=true&version=123&fields=content`
+    )
+    expect(req.request.method).toEqual('GET')
   })
 
   it('should set search query', () => {
@@ -283,12 +318,65 @@ describe(`DocumentService`, () => {
     expect(req.request.body.remove_inbox_tags).toEqual(true)
   })
 
+  it('should pass selected version to patch when provided', () => {
+    subscription = service.patch(documents[0], 123).subscribe()
+    const req = httpTestingController.expectOne(
+      `${environment.apiBaseUrl}${endpoint}/${documents[0].id}/?version=123`
+    )
+    expect(req.request.method).toEqual('PATCH')
+  })
+
   it('should call appropriate api endpoint for getting audit log', () => {
     subscription = service.getHistory(documents[0].id).subscribe()
     const req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}${endpoint}/${documents[0].id}/history/`
     )
   })
+
+  it('should call appropriate api endpoint for getting root document id', () => {
+    subscription = service.getRootId(documents[0].id).subscribe()
+    const req = httpTestingController.expectOne(
+      `${environment.apiBaseUrl}${endpoint}/${documents[0].id}/root/`
+    )
+    expect(req.request.method).toEqual('GET')
+    req.flush({ root_id: documents[0].id })
+  })
+
+  it('should call appropriate api endpoint for getting document versions', () => {
+    subscription = service.getVersions(documents[0].id).subscribe()
+    const req = httpTestingController.expectOne(
+      `${environment.apiBaseUrl}${endpoint}/${documents[0].id}/?fields=id,versions`
+    )
+    expect(req.request.method).toEqual('GET')
+  })
+
+  it('should call appropriate api endpoint for deleting a document version', () => {
+    subscription = service.deleteVersion(documents[0].id, 10).subscribe()
+    const req = httpTestingController.expectOne(
+      `${environment.apiBaseUrl}${endpoint}/${documents[0].id}/versions/10/`
+    )
+    expect(req.request.method).toEqual('DELETE')
+    req.flush({ result: 'OK', current_version_id: documents[0].id })
+  })
+
+  it('should call appropriate api endpoint for uploading a new version', () => {
+    const file = new File(['hello'], 'test.pdf', { type: 'application/pdf' })
+
+    subscription = service
+      .uploadVersion(documents[0].id, file, 'Label')
+      .subscribe()
+    const req = httpTestingController.expectOne(
+      `${environment.apiBaseUrl}${endpoint}/${documents[0].id}/update_version/`
+    )
+    expect(req.request.method).toEqual('POST')
+    expect(req.request.body).toBeInstanceOf(FormData)
+
+    const body = req.request.body as FormData
+    expect(body.get('version_label')).toEqual('Label')
+    expect(body.get('document')).toBeInstanceOf(File)
+
+    req.flush('task-id')
+  })
 })
 
 it('should construct sort fields respecting permissions', () => {

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

@@ -155,44 +155,108 @@ export class DocumentService extends AbstractPaperlessService<Document> {
     }).pipe(map((response) => response.results.map((doc) => doc.id)))
   }
 
-  get(id: number): Observable<Document> {
+  get(
+    id: number,
+    versionID: number = null,
+    fields: string = null
+  ): Observable<Document> {
+    const params: { full_perms: boolean; version?: string; fields?: string } = {
+      full_perms: true,
+    }
+    if (versionID) {
+      params.version = versionID.toString()
+    }
+    if (fields) {
+      params.fields = fields
+    }
     return this.http.get<Document>(this.getResourceUrl(id), {
-      params: {
+      params,
-        full_perms: true,
-      },
     })
   }
 
-  getPreviewUrl(id: number, original: boolean = false): string {
+  getPreviewUrl(
+    id: number,
+    original: boolean = false,
+    versionID: number = null
+  ): string {
     let url = new URL(this.getResourceUrl(id, 'preview'))
     if (this._searchQuery) url.hash = `#search="${this.searchQuery}"`
     if (original) {
       url.searchParams.append('original', 'true')
     }
+    if (versionID) {
+      url.searchParams.append('version', versionID.toString())
+    }
     return url.toString()
   }
 
-  getThumbUrl(id: number): string {
+  getThumbUrl(id: number, versionID: number = null): string {
-    return this.getResourceUrl(id, 'thumb')
+    let url = new URL(this.getResourceUrl(id, 'thumb'))
+    if (versionID) {
+      url.searchParams.append('version', versionID.toString())
+    }
+    return url.toString()
   }
 
-  getDownloadUrl(id: number, original: boolean = false): string {
+  getDownloadUrl(
-    let url = this.getResourceUrl(id, 'download')
+    id: number,
+    original: boolean = false,
+    versionID: number = null
+  ): string {
+    let url = new URL(this.getResourceUrl(id, 'download'))
     if (original) {
-      url += '?original=true'
+      url.searchParams.append('original', 'true')
     }
-    return url
+    if (versionID) {
+      url.searchParams.append('version', versionID.toString())
+    }
+    return url.toString()
+  }
+
+  uploadVersion(documentId: number, file: File, versionLabel?: string) {
+    const formData = new FormData()
+    formData.append('document', file, file.name)
+    if (versionLabel) {
+      formData.append('version_label', versionLabel)
+    }
+    return this.http.post<string>(
+      this.getResourceUrl(documentId, 'update_version'),
+      formData
+    )
+  }
+
+  getVersions(documentId: number): Observable<Document> {
+    return this.http.get<Document>(this.getResourceUrl(documentId), {
+      params: {
+        fields: 'id,versions',
+      },
+    })
+  }
+
+  getRootId(documentId: number) {
+    return this.http.get<{ root_id: number }>(
+      this.getResourceUrl(documentId, 'root')
+    )
+  }
+
+  deleteVersion(rootDocumentId: number, versionId: number) {
+    return this.http.delete<{ result: string; current_version_id: number }>(
+      this.getResourceUrl(rootDocumentId, `versions/${versionId}`)
+    )
   }
 
   getNextAsn(): Observable<number> {
     return this.http.get<number>(this.getResourceUrl(null, 'next_asn'))
   }
 
-  patch(o: Document): Observable<Document> {
+  patch(o: Document, versionID: number = null): Observable<Document> {
     o.remove_inbox_tags = !!this.settingsService.get(
       SETTINGS_KEYS.DOCUMENT_EDITING_REMOVE_INBOX_TAGS
     )
-    return super.patch(o)
+    this.clearCache()
+    return this.http.patch<Document>(this.getResourceUrl(o.id), o, {
+      params: versionID ? { version: versionID.toString() } : {},
+    })
   }
 
   uploadDocument(formData) {
@@ -203,8 +267,15 @@ export class DocumentService extends AbstractPaperlessService<Document> {
     )
   }
 
-  getMetadata(id: number): Observable<DocumentMetadata> {
+  getMetadata(
-    return this.http.get<DocumentMetadata>(this.getResourceUrl(id, 'metadata'))
+    id: number,
+    versionID: number = null
+  ): Observable<DocumentMetadata> {
+    let url = new URL(this.getResourceUrl(id, 'metadata'))
+    if (versionID) {
+      url.searchParams.append('version', versionID.toString())
+    }
+    return this.http.get<DocumentMetadata>(url.toString())
   }
 
   bulkEdit(ids: number[], method: string, args: any) {

src-ui/src/app/services/websocket-status.service.ts

@@ -89,6 +89,13 @@ export class FileStatus {
   }
 }
 
+export enum UploadState {
+  Idle = 'idle',
+  Uploading = 'uploading',
+  Processing = 'processing',
+  Failed = 'failed',
+}
+
 @Injectable({
   providedIn: 'root',
 })

src-ui/src/main.ts

@@ -79,9 +79,11 @@ import {
   eye,
   fileEarmark,
   fileEarmarkCheck,
+  fileEarmarkDiff,
   fileEarmarkFill,
   fileEarmarkLock,
   fileEarmarkMinus,
+  fileEarmarkPlus,
   fileEarmarkRichtext,
   fileText,
   files,
@@ -298,9 +300,11 @@ const icons = {
   eye,
   fileEarmark,
   fileEarmarkCheck,
+  fileEarmarkDiff,
   fileEarmarkFill,
   fileEarmarkLock,
   fileEarmarkMinus,
+  fileEarmarkPlus,
   fileEarmarkRichtext,
   files,
   fileText,

src/documents/bulk_edit.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import hashlib
 import logging
 import tempfile
 from pathlib import Path
@@ -73,6 +72,48 @@ def restore_archive_serial_numbers(backup: dict[int, int | None]) -> None:
     logger.info(f"Restored archive serial numbers for documents {list(backup.keys())}")
 
 
+def _get_root_ids_by_doc_id(doc_ids: list[int]) -> dict[int, int]:
+    """
+    Resolve each provided document id to its root document id.
+
+    - If the id is already a root document: root id is itself.
+    - If the id is a version document: root id is its `root_document_id`.
+    """
+    qs = Document.objects.filter(id__in=doc_ids).only("id", "root_document_id")
+    return {doc.id: doc.root_document_id or doc.id for doc in qs}
+
+
+def _get_root_and_current_docs_by_root_id(
+    root_ids: set[int],
+) -> tuple[dict[int, Document], dict[int, Document]]:
+    """
+    Returns:
+      - root_docs: root_id -> root Document
+      - current_docs: root_id -> newest version Document (or root if none)
+    """
+    root_docs = {
+        doc.id: doc
+        for doc in Document.objects.filter(id__in=root_ids).select_related(
+            "owner",
+        )
+    }
+    latest_versions_by_root_id: dict[int, Document] = {}
+    for version_doc in Document.objects.filter(root_document_id__in=root_ids).order_by(
+        "root_document_id",
+        "-id",
+    ):
+        root_id = version_doc.root_document_id
+        if root_id is None:
+            continue
+        latest_versions_by_root_id.setdefault(root_id, version_doc)
+
+    current_docs: dict[int, Document] = {
+        root_id: latest_versions_by_root_id.get(root_id, root_docs[root_id])
+        for root_id in root_docs
+    }
+    return root_docs, current_docs
+
+
 def set_correspondent(
     doc_ids: list[int],
     correspondent: Correspondent,
@@ -309,16 +350,29 @@ def modify_custom_fields(
 @shared_task
 def delete(doc_ids: list[int]) -> Literal["OK"]:
     try:
-        Document.objects.filter(id__in=doc_ids).delete()
+        root_ids = (
+            Document.objects.filter(id__in=doc_ids, root_document__isnull=True)
+            .values_list("id", flat=True)
+            .distinct()
+        )
+        version_ids = (
+            Document.objects.filter(root_document_id__in=root_ids)
+            .exclude(id__in=doc_ids)
+            .values_list("id", flat=True)
+            .distinct()
+        )
+        delete_ids = list({*doc_ids, *version_ids})
+
+        Document.objects.filter(id__in=delete_ids).delete()
 
         from documents import index
 
         with index.open_index_writer() as writer:
-            for id in doc_ids:
+            for id in delete_ids:
                 index.remove_document_by_id(writer, id)
 
         status_mgr = DocumentsStatusManager()
-        status_mgr.send_documents_deleted(doc_ids)
+        status_mgr.send_documents_deleted(delete_ids)
     except Exception as e:
         if "Data too long for column" in str(e):
             logger.warning(
@@ -363,43 +417,60 @@ def set_permissions(
     return "OK"
 
 
-def rotate(doc_ids: list[int], degrees: int) -> Literal["OK"]:
+def rotate(
+    doc_ids: list[int],
+    degrees: int,
+    *,
+    user: User | None = None,
+) -> Literal["OK"]:
     logger.info(
         f"Attempting to rotate {len(doc_ids)} documents by {degrees} degrees.",
     )
-    qs = Document.objects.filter(id__in=doc_ids)
+    doc_to_root_id = _get_root_ids_by_doc_id(doc_ids)
-    affected_docs: list[int] = []
+    root_ids = set(doc_to_root_id.values())
+    root_docs_by_id, current_docs_by_root_id = _get_root_and_current_docs_by_root_id(
+        root_ids,
+    )
     import pikepdf
 
-    rotate_tasks = []
+    for root_id in root_ids:
-    for doc in qs:
+        root_doc = root_docs_by_id[root_id]
-        if doc.mime_type != "application/pdf":
+        source_doc = current_docs_by_root_id[root_id]
+        if source_doc.mime_type != "application/pdf":
             logger.warning(
-                f"Document {doc.id} is not a PDF, skipping rotation.",
+                f"Document {root_doc.id} is not a PDF, skipping rotation.",
             )
             continue
         try:
-            with pikepdf.open(doc.source_path, allow_overwriting_input=True) as pdf:
+            # Write rotated output to a temp file and create a new version via consume pipeline
+            filepath: Path = (
+                Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR))
+                / f"{root_doc.id}_rotated.pdf"
+            )
+            with pikepdf.open(source_doc.source_path) as pdf:
                 for page in pdf.pages:
                     page.rotate(degrees, relative=True)
-                pdf.save()
+                pdf.remove_unreferenced_resources()
-                doc.checksum = hashlib.md5(doc.source_path.read_bytes()).hexdigest()
+                pdf.save(filepath)
-                doc.save()
-                rotate_tasks.append(
-                    update_document_content_maybe_archive_file.s(
-                        document_id=doc.id,
-                    ),
-                )
-                logger.info(
-                    f"Rotated document {doc.id} by {degrees} degrees",
-                )
-                affected_docs.append(doc.id)
-        except Exception as e:
-            logger.exception(f"Error rotating document {doc.id}: {e}")
 
-    if len(affected_docs) > 0:
+            # Preserve metadata/permissions via overrides; mark as new version
-        bulk_update_task = bulk_update_documents.si(document_ids=affected_docs)
+            overrides = DocumentMetadataOverrides().from_document(root_doc)
-        chord(header=rotate_tasks, body=bulk_update_task).delay()
+            if user is not None:
+                overrides.actor_id = user.id
+
+            consume_file.delay(
+                ConsumableDocument(
+                    source=DocumentSource.ConsumeFolder,
+                    original_file=filepath,
+                    root_document_id=root_doc.id,
+                ),
+                overrides,
+            )
+            logger.info(
+                f"Queued new rotated version for document {root_doc.id} by {degrees} degrees",
+            )
+        except Exception as e:
+            logger.exception(f"Error rotating document {root_doc.id}: {e}")
 
     return "OK"
 
@@ -584,30 +655,62 @@ def split(
     return "OK"
 
 
-def delete_pages(doc_ids: list[int], pages: list[int]) -> Literal["OK"]:
+def delete_pages(
+    doc_ids: list[int],
+    pages: list[int],
+    *,
+    user: User | None = None,
+) -> Literal["OK"]:
     logger.info(
         f"Attempting to delete pages {pages} from {len(doc_ids)} documents",
     )
-    doc = Document.objects.get(id=doc_ids[0])
+    doc = Document.objects.select_related("root_document").get(id=doc_ids[0])
+    root_doc: Document
+    if doc.root_document_id is None or doc.root_document is None:
+        root_doc = doc
+    else:
+        root_doc = doc.root_document
+
+    source_doc = (
+        Document.objects.filter(Q(id=root_doc.id) | Q(root_document=root_doc))
+        .order_by("-id")
+        .first()
+    )
+    if source_doc is None:
+        source_doc = root_doc
     pages = sorted(pages)  # sort pages to avoid index issues
     import pikepdf
 
     try:
-        with pikepdf.open(doc.source_path, allow_overwriting_input=True) as pdf:
+        # Produce edited PDF to a temp file and create a new version
+        filepath: Path = (
+            Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR))
+            / f"{root_doc.id}_pages_deleted.pdf"
+        )
+        with pikepdf.open(source_doc.source_path) as pdf:
             offset = 1  # pages are 1-indexed
             for page_num in pages:
                 pdf.pages.remove(pdf.pages[page_num - offset])
                 offset += 1  # remove() changes the index of the pages
             pdf.remove_unreferenced_resources()
-            pdf.save()
+            pdf.save(filepath)
-            doc.checksum = hashlib.md5(doc.source_path.read_bytes()).hexdigest()
+
-            if doc.page_count is not None:
+        overrides = DocumentMetadataOverrides().from_document(root_doc)
-                doc.page_count = doc.page_count - len(pages)
+        if user is not None:
-            doc.save()
+            overrides.actor_id = user.id
-            update_document_content_maybe_archive_file.delay(document_id=doc.id)
+        consume_file.delay(
-            logger.info(f"Deleted pages {pages} from document {doc.id}")
+            ConsumableDocument(
+                source=DocumentSource.ConsumeFolder,
+                original_file=filepath,
+                root_document_id=root_doc.id,
+            ),
+            overrides,
+        )
+        logger.info(
+            f"Queued new version for document {root_doc.id} after deleting pages {pages}",
+        )
     except Exception as e:
-        logger.exception(f"Error deleting pages from document {doc.id}: {e}")
+        logger.exception(f"Error deleting pages from document {root_doc.id}: {e}")
 
     return "OK"
 
@@ -632,13 +735,26 @@ def edit_pdf(
     logger.info(
         f"Editing PDF of document {doc_ids[0]} with {len(operations)} operations",
     )
-    doc = Document.objects.get(id=doc_ids[0])
+    doc = Document.objects.select_related("root_document").get(id=doc_ids[0])
+    root_doc: Document
+    if doc.root_document_id is None or doc.root_document is None:
+        root_doc = doc
+    else:
+        root_doc = doc.root_document
+
+    source_doc = (
+        Document.objects.filter(Q(id=root_doc.id) | Q(root_document=root_doc))
+        .order_by("-id")
+        .first()
+    )
+    if source_doc is None:
+        source_doc = root_doc
     import pikepdf
 
     pdf_docs: list[pikepdf.Pdf] = []
 
     try:
-        with pikepdf.open(doc.source_path) as src:
+        with pikepdf.open(source_doc.source_path) as src:
             # prepare output documents
             max_idx = max(op.get("doc", 0) for op in operations)
             pdf_docs = [pikepdf.new() for _ in range(max_idx + 1)]
@@ -657,42 +773,56 @@ def edit_pdf(
                     dst.pages[-1].rotate(op["rotate"], relative=True)
 
         if update_document:
-            temp_path = doc.source_path.with_suffix(".tmp.pdf")
+            # Create a new version from the edited PDF rather than replacing in-place
             pdf = pdf_docs[0]
             pdf.remove_unreferenced_resources()
-            # save the edited PDF to a temporary file in case of errors
+            filepath: Path = (
-            pdf.save(temp_path)
+                Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR))
-            # replace the original document with the edited one
+                / f"{root_doc.id}_edited.pdf"
-            temp_path.replace(doc.source_path)
+            )
-            doc.checksum = hashlib.md5(doc.source_path.read_bytes()).hexdigest()
+            pdf.save(filepath)
-            doc.page_count = len(pdf.pages)
-            doc.save()
-            update_document_content_maybe_archive_file.delay(document_id=doc.id)
-        else:
-            consume_tasks = []
             overrides = (
-                DocumentMetadataOverrides().from_document(doc)
+                DocumentMetadataOverrides().from_document(root_doc)
                 if include_metadata
                 else DocumentMetadataOverrides()
             )
             if user is not None:
                 overrides.owner_id = user.id
+                overrides.actor_id = user.id
+            consume_file.delay(
+                ConsumableDocument(
+                    source=DocumentSource.ConsumeFolder,
+                    original_file=filepath,
+                    root_document_id=root_doc.id,
+                ),
+                overrides,
+            )
+        else:
+            consume_tasks = []
+            overrides = (
+                DocumentMetadataOverrides().from_document(root_doc)
+                if include_metadata
+                else DocumentMetadataOverrides()
+            )
+            if user is not None:
+                overrides.owner_id = user.id
+                overrides.actor_id = user.id
             if not delete_original:
                 overrides.skip_asn_if_exists = True
             if delete_original and len(pdf_docs) == 1:
-                overrides.asn = doc.archive_serial_number
+                overrides.asn = root_doc.archive_serial_number
             for idx, pdf in enumerate(pdf_docs, start=1):
-                filepath: Path = (
+                version_filepath: Path = (
                     Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR))
-                    / f"{doc.id}_edit_{idx}.pdf"
+                    / f"{root_doc.id}_edit_{idx}.pdf"
                 )
                 pdf.remove_unreferenced_resources()
-                pdf.save(filepath)
+                pdf.save(version_filepath)
                 consume_tasks.append(
                     consume_file.s(
                         ConsumableDocument(
                             source=DocumentSource.ConsumeFolder,
-                            original_file=filepath,
+                            original_file=version_filepath,
                         ),
                         overrides,
                     ),
@@ -714,7 +844,7 @@ def edit_pdf(
                 group(consume_tasks).delay()
 
     except Exception as e:
-        logger.exception(f"Error editing document {doc.id}: {e}")
+        logger.exception(f"Error editing document {root_doc.id}: {e}")
         raise ValueError(
             f"An error occurred while editing the document: {e}",
         ) from e
@@ -737,38 +867,61 @@ def remove_password(
     import pikepdf
 
     for doc_id in doc_ids:
-        doc = Document.objects.get(id=doc_id)
+        doc = Document.objects.select_related("root_document").get(id=doc_id)
+        root_doc: Document
+        if doc.root_document_id is None or doc.root_document is None:
+            root_doc = doc
+        else:
+            root_doc = doc.root_document
+
+        source_doc = (
+            Document.objects.filter(Q(id=root_doc.id) | Q(root_document=root_doc))
+            .order_by("-id")
+            .first()
+        )
+        if source_doc is None:
+            source_doc = root_doc
         try:
             logger.info(
                 f"Attempting password removal from document {doc_ids[0]}",
             )
-            with pikepdf.open(doc.source_path, password=password) as pdf:
+            with pikepdf.open(source_doc.source_path, password=password) as pdf:
-                temp_path = doc.source_path.with_suffix(".tmp.pdf")
+                filepath: Path = (
+                    Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR))
+                    / f"{root_doc.id}_unprotected.pdf"
+                )
                 pdf.remove_unreferenced_resources()
-                pdf.save(temp_path)
+                pdf.save(filepath)
 
                 if update_document:
-                    # replace the original document with the unprotected one
+                    # Create a new version rather than modifying the root/original in place.
-                    temp_path.replace(doc.source_path)
-                    doc.checksum = hashlib.md5(doc.source_path.read_bytes()).hexdigest()
-                    doc.page_count = len(pdf.pages)
-                    doc.save()
-                    update_document_content_maybe_archive_file.delay(document_id=doc.id)
-                else:
-                    consume_tasks = []
                     overrides = (
-                        DocumentMetadataOverrides().from_document(doc)
+                        DocumentMetadataOverrides().from_document(root_doc)
                         if include_metadata
                         else DocumentMetadataOverrides()
                     )
                     if user is not None:
                         overrides.owner_id = user.id
-
+                        overrides.actor_id = user.id
-                    filepath: Path = (
+                    consume_file.delay(
-                        Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR))
+                        ConsumableDocument(
-                        / f"{doc.id}_unprotected.pdf"
+                            source=DocumentSource.ConsumeFolder,
+                            original_file=filepath,
+                            root_document_id=root_doc.id,
+                        ),
+                        overrides,
                     )
-                    temp_path.replace(filepath)
+                else:
+                    consume_tasks = []
+                    overrides = (
+                        DocumentMetadataOverrides().from_document(root_doc)
+                        if include_metadata
+                        else DocumentMetadataOverrides()
+                    )
+                    if user is not None:
+                        overrides.owner_id = user.id
+                        overrides.actor_id = user.id
+
                     consume_tasks.append(
                         consume_file.s(
                             ConsumableDocument(
@@ -780,12 +933,17 @@ def remove_password(
                     )
 
                     if delete_original:
-                        chord(header=consume_tasks, body=delete.si([doc.id])).delay()
+                        chord(
+                            header=consume_tasks,
+                            body=delete.si([doc.id]),
+                        ).delay()
                     else:
                         group(consume_tasks).delay()
 
         except Exception as e:
-            logger.exception(f"Error removing password from document {doc.id}: {e}")
+            logger.exception(
+                f"Error removing password from document {root_doc.id}: {e}",
+            )
             raise ValueError(
                 f"An error occurred while removing the password: {e}",
             ) from e

src/documents/conditionals.py

@@ -1,5 +1,6 @@
 from datetime import datetime
 from datetime import timezone
+from typing import Any
 
 from django.conf import settings
 from django.core.cache import cache
@@ -12,6 +13,7 @@ from documents.caching import CLASSIFIER_VERSION_KEY
 from documents.caching import get_thumbnail_modified_key
 from documents.classifier import DocumentClassifier
 from documents.models import Document
+from documents.versioning import resolve_effective_document_by_pk
 
 
 def suggestions_etag(request, pk: int) -> str | None:
@@ -71,12 +73,10 @@ def metadata_etag(request, pk: int) -> str | None:
     Metadata is extracted from the original file, so use its checksum as the
     ETag
     """
-    try:
+    doc = resolve_effective_document_by_pk(pk, request).document
-        doc = Document.objects.only("checksum").get(pk=pk)
+    if doc is None:
-        return doc.checksum
-    except Document.DoesNotExist:  # pragma: no cover
         return None
-    return None
+    return doc.checksum
 
 
 def metadata_last_modified(request, pk: int) -> datetime | None:
@@ -85,28 +85,25 @@ def metadata_last_modified(request, pk: int) -> datetime | None:
     not the modification of the original file, but of the database object, but might as well
     error on the side of more cautious
     """
-    try:
+    doc = resolve_effective_document_by_pk(pk, request).document
-        doc = Document.objects.only("modified").get(pk=pk)
+    if doc is None:
-        return doc.modified
-    except Document.DoesNotExist:  # pragma: no cover
         return None
-    return None
+    return doc.modified
 
 
 def preview_etag(request, pk: int) -> str | None:
     """
     ETag for the document preview, using the original or archive checksum, depending on the request
     """
-    try:
+    doc = resolve_effective_document_by_pk(pk, request).document
-        doc = Document.objects.only("checksum", "archive_checksum").get(pk=pk)
+    if doc is None:
-        use_original = (
-            "original" in request.query_params
-            and request.query_params["original"] == "true"
-        )
-        return doc.checksum if use_original else doc.archive_checksum
-    except Document.DoesNotExist:  # pragma: no cover
         return None
-    return None
+    use_original = (
+        hasattr(request, "query_params")
+        and "original" in request.query_params
+        and request.query_params["original"] == "true"
+    )
+    return doc.checksum if use_original else doc.archive_checksum
 
 
 def preview_last_modified(request, pk: int) -> datetime | None:
@@ -114,24 +111,25 @@ def preview_last_modified(request, pk: int) -> datetime | None:
     Uses the documents modified time to set the Last-Modified header.  Not strictly
     speaking correct, but close enough and quick
     """
-    try:
+    doc = resolve_effective_document_by_pk(pk, request).document
-        doc = Document.objects.only("modified").get(pk=pk)
+    if doc is None:
-        return doc.modified
-    except Document.DoesNotExist:  # pragma: no cover
         return None
-    return None
+    return doc.modified
 
 
-def thumbnail_last_modified(request, pk: int) -> datetime | None:
+def thumbnail_last_modified(request: Any, pk: int) -> datetime | None:
     """
     Returns the filesystem last modified either from cache or from filesystem.
     Cache should be (slightly?) faster than filesystem
     """
     try:
-        doc = Document.objects.only("pk").get(pk=pk)
+        doc = resolve_effective_document_by_pk(pk, request).document
+        if doc is None:
+            return None
         if not doc.thumbnail_path.exists():
             return None
-        doc_key = get_thumbnail_modified_key(pk)
+        # Use the effective document id for cache key
+        doc_key = get_thumbnail_modified_key(doc.id)
 
         cache_hit = cache.get(doc_key)
         if cache_hit is not None:
@@ -145,5 +143,5 @@ def thumbnail_last_modified(request, pk: int) -> datetime | None:
         )
         cache.set(doc_key, last_modified, CACHE_50_MINUTES)
         return last_modified
-    except Document.DoesNotExist:  # pragma: no cover
+    except (Document.DoesNotExist, OSError):  # pragma: no cover
         return None

src/documents/consumer.py

@@ -102,6 +102,12 @@ class ConsumerStatusShortMessage(str, Enum):
 
 
 class ConsumerPluginMixin:
+    if TYPE_CHECKING:
+        from logging import Logger
+        from logging import LoggerAdapter
+
+        log: "LoggerAdapter"  # type: ignore[type-arg]
+
     def __init__(
         self,
         input_doc: ConsumableDocument,
@@ -116,6 +122,22 @@ class ConsumerPluginMixin:
 
         self.filename = self.metadata.filename or self.input_doc.original_file.name
 
+        if input_doc.root_document_id:
+            self.log.debug(
+                f"Document root document id: {input_doc.root_document_id}",
+            )
+            root_document = Document.objects.get(pk=input_doc.root_document_id)
+            version_index = Document.objects.filter(root_document=root_document).count()
+            filename_path = Path(self.filename)
+            if filename_path.suffix:
+                self.filename = str(
+                    filename_path.with_name(
+                        f"{filename_path.stem}_v{version_index}{filename_path.suffix}",
+                    ),
+                )
+            else:
+                self.filename = f"{self.filename}_v{version_index}"
+
     def _send_progress(
         self,
         current_progress: int,
@@ -161,6 +183,41 @@ class ConsumerPlugin(
 ):
     logging_name = LOGGING_NAME
 
+    def _clone_root_into_version(
+        self,
+        root_doc: Document,
+        *,
+        text: str | None,
+        page_count: int | None,
+        mime_type: str,
+    ) -> Document:
+        self.log.debug("Saving record for updated version to database")
+        version_doc = Document.objects.get(pk=root_doc.pk)
+        setattr(version_doc, "pk", None)
+        version_doc.root_document = root_doc
+        file_for_checksum = (
+            self.unmodified_original
+            if self.unmodified_original is not None
+            else self.working_copy
+        )
+        version_doc.checksum = hashlib.md5(
+            file_for_checksum.read_bytes(),
+        ).hexdigest()
+        version_doc.content = text or ""
+        version_doc.page_count = page_count
+        version_doc.mime_type = mime_type
+        version_doc.original_filename = self.filename
+        version_doc.storage_path = root_doc.storage_path
+        # Clear unique file path fields so they can be generated uniquely later
+        version_doc.filename = None
+        version_doc.archive_filename = None
+        version_doc.archive_checksum = None
+        if self.metadata.version_label is not None:
+            version_doc.version_label = self.metadata.version_label
+        version_doc.added = timezone.now()
+        version_doc.modified = timezone.now()
+        return version_doc
+
     def run_pre_consume_script(self) -> None:
         """
         If one is configured and exists, run the pre-consume script and
@@ -477,12 +534,65 @@ class ConsumerPlugin(
         try:
             with transaction.atomic():
                 # store the document.
-                document = self._store(
+                if self.input_doc.root_document_id:
-                    text=text,
+                    # If this is a new version of an existing document, we need
-                    date=date,
+                    # to make sure we're not creating a new document, but updating
-                    page_count=page_count,
+                    # the existing one.
-                    mime_type=mime_type,
+                    root_doc = Document.objects.get(
-                )
+                        pk=self.input_doc.root_document_id,
+                    )
+                    original_document = self._clone_root_into_version(
+                        root_doc,
+                        text=text,
+                        page_count=page_count,
+                        mime_type=mime_type,
+                    )
+                    actor = None
+
+                    # Save the new version, potentially creating an audit log entry for the version addition if enabled.
+                    if (
+                        settings.AUDIT_LOG_ENABLED
+                        and self.metadata.actor_id is not None
+                    ):
+                        actor = User.objects.filter(pk=self.metadata.actor_id).first()
+                        if actor is not None:
+                            from auditlog.context import (  # type: ignore[import-untyped]
+                                set_actor,
+                            )
+
+                            with set_actor(actor):
+                                original_document.save()
+                        else:
+                            original_document.save()
+                    else:
+                        original_document.save()
+
+                    # Create a log entry for the version addition, if enabled
+                    if settings.AUDIT_LOG_ENABLED:
+                        from auditlog.models import (  # type: ignore[import-untyped]
+                            LogEntry,
+                        )
+
+                        LogEntry.objects.log_create(
+                            instance=root_doc,
+                            changes={
+                                "Version Added": ["None", original_document.id],
+                            },
+                            action=LogEntry.Action.UPDATE,
+                            actor=actor,
+                            additional_data={
+                                "reason": "Version added",
+                                "version_id": original_document.id,
+                            },
+                        )
+                    document = original_document
+                else:
+                    document = self._store(
+                        text=text,
+                        date=date,
+                        page_count=page_count,
+                        mime_type=mime_type,
+                    )
 
                 # If we get here, it was successful. Proceed with post-consume
                 # hooks. If they fail, nothing will get changed.
@@ -700,6 +810,9 @@ class ConsumerPlugin(
         if self.metadata.asn is not None:
             document.archive_serial_number = self.metadata.asn
 
+        if self.metadata.version_label is not None:
+            document.version_label = self.metadata.version_label
+
         if self.metadata.owner_id:
             document.owner = User.objects.get(
                 pk=self.metadata.owner_id,

src/documents/data_models.py

@@ -31,6 +31,8 @@ class DocumentMetadataOverrides:
     change_groups: list[int] | None = None
     custom_fields: dict | None = None
     skip_asn_if_exists: bool = False
+    version_label: str | None = None
+    actor_id: int | None = None
 
     def update(self, other: "DocumentMetadataOverrides") -> "DocumentMetadataOverrides":
         """
@@ -50,8 +52,12 @@ class DocumentMetadataOverrides:
             self.storage_path_id = other.storage_path_id
         if other.owner_id is not None:
             self.owner_id = other.owner_id
+        if other.actor_id is not None:
+            self.actor_id = other.actor_id
         if other.skip_asn_if_exists:
             self.skip_asn_if_exists = True
+        if other.version_label is not None:
+            self.version_label = other.version_label
 
         # merge
         if self.tag_ids is None:
@@ -160,6 +166,7 @@ class ConsumableDocument:
 
     source: DocumentSource
     original_file: Path
+    root_document_id: int | None = None
     original_path: Path | None = None
     mailrule_id: int | None = None
     mime_type: str = dataclasses.field(init=False, default=None)

src/documents/filters.py

@@ -6,8 +6,10 @@ import json
 import operator
 from contextlib import contextmanager
 from typing import TYPE_CHECKING
+from typing import Any
 
 from django.contrib.contenttypes.models import ContentType
+from django.core.exceptions import FieldError
 from django.db.models import Case
 from django.db.models import CharField
 from django.db.models import Count
@@ -160,14 +162,37 @@ class InboxFilter(Filter):
 
 @extend_schema_field(serializers.CharField)
 class TitleContentFilter(Filter):
-    def filter(self, qs, value):
+    def filter(self, qs: Any, value: Any) -> Any:
         value = value.strip() if isinstance(value, str) else value
         if value:
-            return qs.filter(Q(title__icontains=value) | Q(content__icontains=value))
+            try:
+                return qs.filter(
+                    Q(title__icontains=value) | Q(effective_content__icontains=value),
+                )
+            except FieldError:
+                return qs.filter(
+                    Q(title__icontains=value) | Q(content__icontains=value),
+                )
         else:
             return qs
 
 
+@extend_schema_field(serializers.CharField)
+class EffectiveContentFilter(Filter):
+    def filter(self, qs: Any, value: Any) -> Any:
+        value = value.strip() if isinstance(value, str) else value
+        if not value:
+            return qs
+        try:
+            return qs.filter(
+                **{f"effective_content__{self.lookup_expr}": value},
+            )
+        except FieldError:
+            return qs.filter(
+                **{f"content__{self.lookup_expr}": value},
+            )
+
+
 @extend_schema_field(serializers.BooleanField)
 class SharedByUser(Filter):
     def filter(self, qs, value):
@@ -724,6 +749,11 @@ class DocumentFilterSet(FilterSet):
 
     title_content = TitleContentFilter()
 
+    content__istartswith = EffectiveContentFilter(lookup_expr="istartswith")
+    content__iendswith = EffectiveContentFilter(lookup_expr="iendswith")
+    content__icontains = EffectiveContentFilter(lookup_expr="icontains")
+    content__iexact = EffectiveContentFilter(lookup_expr="iexact")
+
     owner__id__none = ObjectFilter(field_name="owner", exclude=True)
 
     custom_fields__icontains = CustomFieldsFilter()
@@ -764,7 +794,6 @@ class DocumentFilterSet(FilterSet):
         fields = {
             "id": ID_KWARGS,
             "title": CHAR_KWARGS,
-            "content": CHAR_KWARGS,
             "archive_serial_number": INT_KWARGS,
             "created": DATE_KWARGS,
             "added": DATETIME_KWARGS,

src/documents/index.py

@@ -158,7 +158,11 @@ def open_index_searcher() -> Searcher:
         searcher.close()
 
 
-def update_document(writer: AsyncWriter, doc: Document) -> None:
+def update_document(
+    writer: AsyncWriter,
+    doc: Document,
+    effective_content: str | None = None,
+) -> None:
     tags = ",".join([t.name for t in doc.tags.all()])
     tags_ids = ",".join([str(t.id) for t in doc.tags.all()])
     notes = ",".join([str(c.note) for c in Note.objects.filter(document=doc)])
@@ -188,7 +192,7 @@ def update_document(writer: AsyncWriter, doc: Document) -> None:
     writer.update_document(
         id=doc.pk,
         title=doc.title,
-        content=doc.content,
+        content=effective_content or doc.content,
         correspondent=doc.correspondent.name if doc.correspondent else None,
         correspondent_id=doc.correspondent.id if doc.correspondent else None,
         has_correspondent=doc.correspondent is not None,
@@ -231,9 +235,12 @@ def remove_document_by_id(writer: AsyncWriter, doc_id) -> None:
     writer.delete_by_term("id", doc_id)
 
 
-def add_or_update_document(document: Document) -> None:
+def add_or_update_document(
+    document: Document,
+    effective_content: str | None = None,
+) -> None:
     with open_index_writer() as writer:
-        update_document(writer, document)
+        update_document(writer, document, effective_content=effective_content)
 
 
 def remove_document_from_index(document: Document) -> None:

src/documents/migrations/0012_document_root_document.py

@@ -0,0 +1,37 @@
+# Generated by Django 5.1.6 on 2025-02-26 17:08
+
+import django.db.models.deletion
+from django.db import migrations
+from django.db import models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("documents", "0011_optimize_integer_field_sizes"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="document",
+            name="root_document",
+            field=models.ForeignKey(
+                blank=True,
+                null=True,
+                on_delete=django.db.models.deletion.CASCADE,
+                related_name="versions",
+                to="documents.document",
+                verbose_name="root document for this version",
+            ),
+        ),
+        migrations.AddField(
+            model_name="document",
+            name="version_label",
+            field=models.CharField(
+                blank=True,
+                help_text="Optional short label for a document version.",
+                max_length=64,
+                null=True,
+                verbose_name="version label",
+            ),
+        ),
+    ]

src/documents/models.py

@@ -155,7 +155,7 @@ class StoragePath(MatchingModel):
         verbose_name_plural = _("storage paths")
 
 
-class Document(SoftDeleteModel, ModelWithOwner):
+class Document(SoftDeleteModel, ModelWithOwner):  # type: ignore[django-manager-missing]
     correspondent = models.ForeignKey(
         Correspondent,
         blank=True,
@@ -308,6 +308,23 @@ class Document(SoftDeleteModel, ModelWithOwner):
         ),
     )
 
+    root_document = models.ForeignKey(
+        "self",
+        blank=True,
+        null=True,
+        related_name="versions",
+        on_delete=models.CASCADE,
+        verbose_name=_("root document for this version"),
+    )
+
+    version_label = models.CharField(
+        _("version label"),
+        max_length=64,
+        blank=True,
+        null=True,
+        help_text=_("Optional short label for a document version."),
+    )
+
     class Meta:
         ordering = ("-created",)
         verbose_name = _("document")
@@ -419,6 +436,19 @@ class Document(SoftDeleteModel, ModelWithOwner):
         tags_to_add = self.tags.model.objects.filter(id__in=tag_ids)
         self.tags.add(*tags_to_add)
 
+    def delete(
+        self,
+        *args,
+        **kwargs,
+    ):
+        # If deleting a root document, move all its versions to trash as well.
+        if self.root_document_id is None:
+            Document.objects.filter(root_document=self).delete()
+        return super().delete(
+            *args,
+            **kwargs,
+        )
+
 
 class SavedView(ModelWithOwner):
     class DisplayMode(models.TextChoices):
@@ -1712,5 +1742,5 @@ class WorkflowRun(SoftDeleteModel):
         verbose_name = _("workflow run")
         verbose_name_plural = _("workflow runs")
 
-    def __str__(self):
+    def __str__(self) -> str:
         return f"WorkflowRun of {self.workflow} at {self.run_at} on {self.document}"

src/documents/serialisers.py

@@ -7,7 +7,9 @@ from datetime import datetime
 from datetime import timedelta
 from decimal import Decimal
 from typing import TYPE_CHECKING
+from typing import Any
 from typing import Literal
+from typing import TypedDict
 
 import magic
 from celery import states
@@ -89,6 +91,8 @@ if TYPE_CHECKING:
     from collections.abc import Iterable
 
     from django.db.models.query import QuerySet
+    from rest_framework.relations import ManyRelatedField
+    from rest_framework.relations import RelatedField
 
 
 logger = logging.getLogger("paperless.serializers")
@@ -1046,6 +1050,7 @@ def _get_viewable_duplicates(
     duplicates = Document.global_objects.filter(
         Q(checksum__in=checksums) | Q(archive_checksum__in=checksums),
     ).exclude(pk=document.pk)
+    duplicates = duplicates.filter(root_document__isnull=True)
     duplicates = duplicates.order_by("-created")
     allowed = get_objects_for_user_owner_aware(
         user,
@@ -1062,6 +1067,22 @@ class DuplicateDocumentSummarySerializer(serializers.Serializer):
     deleted_at = serializers.DateTimeField(allow_null=True)
 
 
+class DocumentVersionInfoSerializer(serializers.Serializer):
+    id = serializers.IntegerField()
+    added = serializers.DateTimeField()
+    version_label = serializers.CharField(required=False, allow_null=True)
+    checksum = serializers.CharField(required=False, allow_null=True)
+    is_root = serializers.BooleanField()
+
+
+class _DocumentVersionInfo(TypedDict):
+    id: int
+    added: datetime
+    version_label: str | None
+    checksum: str | None
+    is_root: bool
+
+
 @extend_schema_serializer(
     deprecate_fields=["created_date"],
 )
@@ -1082,6 +1103,10 @@ class DocumentSerializer(
     duplicate_documents = SerializerMethodField()
 
     notes = NotesSerializer(many=True, required=False, read_only=True)
+    root_document: RelatedField[Document, Document, Any] | ManyRelatedField = (
+        serializers.PrimaryKeyRelatedField(read_only=True)
+    )
+    versions = SerializerMethodField()
 
     custom_fields = CustomFieldInstanceSerializer(
         many=True,
@@ -1115,6 +1140,44 @@ class DocumentSerializer(
         duplicates = _get_viewable_duplicates(obj, user)
         return list(duplicates.values("id", "title", "deleted_at"))
 
+    @extend_schema_field(DocumentVersionInfoSerializer(many=True))
+    def get_versions(self, obj):
+        root_doc = obj if obj.root_document_id is None else obj.root_document
+        if root_doc is None:
+            return []
+
+        prefetched_cache = getattr(obj, "_prefetched_objects_cache", None)
+        prefetched_versions = (
+            prefetched_cache.get("versions")
+            if isinstance(prefetched_cache, dict)
+            else None
+        )
+
+        versions: list[Document]
+        if prefetched_versions is not None:
+            versions = [*prefetched_versions, root_doc]
+        else:
+            versions_qs = Document.objects.filter(root_document=root_doc).only(
+                "id",
+                "added",
+                "checksum",
+                "version_label",
+            )
+            versions = [*versions_qs, root_doc]
+
+        def build_info(doc: Document) -> _DocumentVersionInfo:
+            return {
+                "id": doc.id,
+                "added": doc.added,
+                "version_label": doc.version_label,
+                "checksum": doc.checksum,
+                "is_root": doc.id == root_doc.id,
+            }
+
+        info = [build_info(doc) for doc in versions]
+        info.sort(key=lambda item: item["id"], reverse=True)
+        return info
+
     def get_original_file_name(self, obj) -> str | None:
         return obj.original_filename
 
@@ -1126,6 +1189,8 @@ 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 self.truncate_content and "content" in self.fields:
             doc["content"] = doc.get("content")[0:550]
 
@@ -1303,6 +1368,8 @@ class DocumentSerializer(
             "remove_inbox_tags",
             "page_count",
             "mime_type",
+            "root_document",
+            "versions",
         )
         list_serializer_class = OwnedObjectListSerializer
 
@@ -1997,6 +2064,22 @@ class PostDocumentSerializer(serializers.Serializer):
             return created.date()
 
 
+class DocumentVersionSerializer(serializers.Serializer):
+    document = serializers.FileField(
+        label="Document",
+        write_only=True,
+    )
+    version_label = serializers.CharField(
+        label="Version label",
+        required=False,
+        allow_blank=True,
+        allow_null=True,
+        max_length=64,
+    )
+
+    validate_document = PostDocumentSerializer().validate_document
+
+
 class BulkDownloadSerializer(DocumentListSerializer):
     content = serializers.ChoiceField(
         choices=["archive", "originals", "both"],
@@ -2196,7 +2279,7 @@ class TasksViewSerializer(OwnedObjectSerializer):
         return list(duplicates.values("id", "title", "deleted_at"))
 
 
-class RunTaskViewSerializer(serializers.Serializer):
+class RunTaskViewSerializer(serializers.Serializer[dict[str, Any]]):
     task_name = serializers.ChoiceField(
         choices=PaperlessTask.TaskName.choices,
         label="Task Name",
@@ -2204,7 +2287,7 @@ class RunTaskViewSerializer(serializers.Serializer):
     )
 
 
-class AcknowledgeTasksViewSerializer(serializers.Serializer):
+class AcknowledgeTasksViewSerializer(serializers.Serializer[dict[str, Any]]):
     tasks = serializers.ListField(
         required=True,
         label="Tasks",
@@ -2951,7 +3034,7 @@ class TrashSerializer(SerializerWithPerms):
         write_only=True,
     )
 
-    def validate_documents(self, documents):
+    def validate_documents(self, documents: list[int]) -> list[int]:
         count = Document.deleted_objects.filter(id__in=documents).count()
         if not count == len(documents):
             raise serializers.ValidationError(

src/documents/signals/handlers.py

@@ -722,6 +722,12 @@ def add_to_index(sender, document, **kwargs) -> None:
     from documents import index
 
     index.add_or_update_document(document)
+    if document.root_document_id is not None and document.root_document is not None:
+        # keep in sync when a new version is consumed.
+        index.add_or_update_document(
+            document.root_document,
+            effective_content=document.content,
+        )
 
 
 def run_workflows_added(

src/documents/tasks.py

@@ -156,15 +156,22 @@ def consume_file(
     if overrides is None:
         overrides = DocumentMetadataOverrides()
 
-    plugins: list[type[ConsumeTaskPlugin]] = [
+    plugins: list[type[ConsumeTaskPlugin]] = (
-        ConsumerPreflightPlugin,
+        [
-        AsnCheckPlugin,
+            ConsumerPreflightPlugin,
-        CollatePlugin,
+            ConsumerPlugin,
-        BarcodePlugin,
+        ]
-        AsnCheckPlugin,  # Re-run ASN check after barcode reading
+        if input_doc.root_document_id is not None
-        WorkflowTriggerPlugin,
+        else [
-        ConsumerPlugin,
+            ConsumerPreflightPlugin,
-    ]
+            AsnCheckPlugin,
+            CollatePlugin,
+            BarcodePlugin,
+            AsnCheckPlugin,  # Re-run ASN check after barcode reading
+            WorkflowTriggerPlugin,
+            ConsumerPlugin,
+        ]
+    )
 
     with (
         ProgressManager(

src/documents/tests/test_api_document_versions.py

@@ -0,0 +1,710 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from unittest import TestCase
+from unittest import mock
+
+from auditlog.models import LogEntry  # type: ignore[import-untyped]
+from django.contrib.auth.models import Permission
+from django.contrib.auth.models import User
+from django.contrib.contenttypes.models import ContentType
+from django.core.exceptions import FieldError
+from django.core.files.uploadedfile import SimpleUploadedFile
+from rest_framework import status
+from rest_framework.test import APITestCase
+
+from documents.data_models import DocumentSource
+from documents.filters import EffectiveContentFilter
+from documents.filters import TitleContentFilter
+from documents.models import Document
+from documents.tests.utils import DirectoriesMixin
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+class TestDocumentVersioningApi(DirectoriesMixin, APITestCase):
+    def setUp(self) -> None:
+        super().setUp()
+
+        self.user = User.objects.create_superuser(username="temp_admin")
+        self.client.force_authenticate(user=self.user)
+
+    def _make_pdf_upload(self, name: str = "version.pdf") -> SimpleUploadedFile:
+        return SimpleUploadedFile(
+            name,
+            b"%PDF-1.4\n1 0 obj\n<<>>\nendobj\n%%EOF",
+            content_type="application/pdf",
+        )
+
+    def _write_file(self, path: Path, content: bytes = b"data") -> None:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(content)
+
+    def _create_pdf(
+        self,
+        *,
+        title: str,
+        checksum: str,
+        root_document: Document | None = None,
+    ) -> Document:
+        doc = Document.objects.create(
+            title=title,
+            checksum=checksum,
+            mime_type="application/pdf",
+            root_document=root_document,
+        )
+        self._write_file(doc.source_path, b"pdf")
+        self._write_file(doc.thumbnail_path, b"thumb")
+        return doc
+
+    def test_root_endpoint_returns_root_for_version_and_root(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+        version = Document.objects.create(
+            title="v1",
+            checksum="v1",
+            mime_type="application/pdf",
+            root_document=root,
+        )
+
+        resp_root = self.client.get(f"/api/documents/{root.id}/root/")
+        self.assertEqual(resp_root.status_code, status.HTTP_200_OK)
+        self.assertEqual(resp_root.data["root_id"], root.id)
+
+        resp_version = self.client.get(f"/api/documents/{version.id}/root/")
+        self.assertEqual(resp_version.status_code, status.HTTP_200_OK)
+        self.assertEqual(resp_version.data["root_id"], root.id)
+
+    def test_root_endpoint_returns_404_for_missing_document(self) -> None:
+        resp = self.client.get("/api/documents/9999/root/")
+
+        self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_root_endpoint_returns_403_when_user_lacks_permission(self) -> None:
+        owner = User.objects.create_user(username="owner")
+        viewer = User.objects.create_user(username="viewer")
+        viewer.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+            owner=owner,
+        )
+        self.client.force_authenticate(user=viewer)
+
+        resp = self.client.get(f"/api/documents/{root.id}/root/")
+
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_delete_version_disallows_deleting_root(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+
+        with mock.patch("documents.index.remove_document_from_index"):
+            resp = self.client.delete(f"/api/documents/{root.id}/versions/{root.id}/")
+
+        self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertTrue(Document.objects.filter(id=root.id).exists())
+
+    def test_delete_version_deletes_version_and_returns_current_version(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+            content="root-content",
+        )
+        v1 = Document.objects.create(
+            title="v1",
+            checksum="v1",
+            mime_type="application/pdf",
+            root_document=root,
+            content="v1-content",
+        )
+        v2 = Document.objects.create(
+            title="v2",
+            checksum="v2",
+            mime_type="application/pdf",
+            root_document=root,
+            content="v2-content",
+        )
+
+        with (
+            mock.patch("documents.index.remove_document_from_index"),
+            mock.patch("documents.index.add_or_update_document"),
+        ):
+            resp = self.client.delete(f"/api/documents/{root.id}/versions/{v2.id}/")
+
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertFalse(Document.objects.filter(id=v2.id).exists())
+        self.assertEqual(resp.data["current_version_id"], v1.id)
+        root.refresh_from_db()
+        self.assertEqual(root.content, "root-content")
+
+        with (
+            mock.patch("documents.index.remove_document_from_index"),
+            mock.patch("documents.index.add_or_update_document"),
+        ):
+            resp = self.client.delete(f"/api/documents/{root.id}/versions/{v1.id}/")
+
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertFalse(Document.objects.filter(id=v1.id).exists())
+        self.assertEqual(resp.data["current_version_id"], root.id)
+        root.refresh_from_db()
+        self.assertEqual(root.content, "root-content")
+
+    def test_delete_version_writes_audit_log_entry(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+        version = Document.objects.create(
+            title="v1",
+            checksum="v1",
+            mime_type="application/pdf",
+            root_document=root,
+        )
+        version_id = version.id
+
+        with (
+            mock.patch("documents.index.remove_document_from_index"),
+            mock.patch("documents.index.add_or_update_document"),
+        ):
+            resp = self.client.delete(
+                f"/api/documents/{root.id}/versions/{version_id}/",
+            )
+
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+        # Audit log entry is created against the root document.
+        entry = (
+            LogEntry.objects.filter(
+                content_type=ContentType.objects.get_for_model(Document),
+                object_id=root.id,
+            )
+            .order_by("-timestamp")
+            .first()
+        )
+        self.assertIsNotNone(entry)
+        assert entry is not None
+        self.assertIsNotNone(entry.actor)
+        assert entry.actor is not None
+        self.assertEqual(entry.actor.id, self.user.id)
+        self.assertEqual(entry.action, LogEntry.Action.UPDATE)
+        self.assertEqual(
+            entry.changes,
+            {"Version Deleted": ["None", version_id]},
+        )
+        additional_data = entry.additional_data or {}
+        self.assertEqual(additional_data.get("version_id"), version_id)
+
+    def test_delete_version_returns_404_when_version_not_related(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+        other_root = Document.objects.create(
+            title="other",
+            checksum="other",
+            mime_type="application/pdf",
+        )
+        other_version = Document.objects.create(
+            title="other-v1",
+            checksum="other-v1",
+            mime_type="application/pdf",
+            root_document=other_root,
+        )
+
+        with mock.patch("documents.index.remove_document_from_index"):
+            resp = self.client.delete(
+                f"/api/documents/{root.id}/versions/{other_version.id}/",
+            )
+
+        self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_delete_version_accepts_version_id_as_root_parameter(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+        version = Document.objects.create(
+            title="v1",
+            checksum="v1",
+            mime_type="application/pdf",
+            root_document=root,
+        )
+
+        with (
+            mock.patch("documents.index.remove_document_from_index"),
+            mock.patch("documents.index.add_or_update_document"),
+        ):
+            resp = self.client.delete(
+                f"/api/documents/{version.id}/versions/{version.id}/",
+            )
+
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertFalse(Document.objects.filter(id=version.id).exists())
+        self.assertEqual(resp.data["current_version_id"], root.id)
+
+    def test_delete_version_returns_404_when_root_missing(self) -> None:
+        resp = self.client.delete("/api/documents/9999/versions/123/")
+
+        self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_delete_version_reindexes_root_document(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+        version = Document.objects.create(
+            title="v1",
+            checksum="v1",
+            mime_type="application/pdf",
+            root_document=root,
+        )
+
+        with (
+            mock.patch("documents.index.remove_document_from_index") as remove_index,
+            mock.patch("documents.index.add_or_update_document") as add_or_update,
+        ):
+            resp = self.client.delete(
+                f"/api/documents/{root.id}/versions/{version.id}/",
+            )
+
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        remove_index.assert_called_once_with(version)
+        add_or_update.assert_called_once()
+        self.assertEqual(add_or_update.call_args[0][0].id, root.id)
+
+    def test_delete_version_returns_403_without_permission(self) -> None:
+        owner = User.objects.create_user(username="owner")
+        other = User.objects.create_user(username="other")
+        other.user_permissions.add(
+            Permission.objects.get(codename="delete_document"),
+        )
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+            owner=owner,
+        )
+        version = Document.objects.create(
+            title="v1",
+            checksum="v1",
+            mime_type="application/pdf",
+            root_document=root,
+        )
+        self.client.force_authenticate(user=other)
+
+        resp = self.client.delete(
+            f"/api/documents/{root.id}/versions/{version.id}/",
+        )
+
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_delete_version_returns_404_when_version_missing(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+
+        resp = self.client.delete(f"/api/documents/{root.id}/versions/9999/")
+
+        self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_download_version_param_errors(self) -> None:
+        root = self._create_pdf(title="root", checksum="root")
+
+        resp = self.client.get(
+            f"/api/documents/{root.id}/download/?version=not-a-number",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
+
+        resp = self.client.get(f"/api/documents/{root.id}/download/?version=9999")
+        self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
+
+        other_root = self._create_pdf(title="other", checksum="other")
+        other_version = self._create_pdf(
+            title="other-v1",
+            checksum="other-v1",
+            root_document=other_root,
+        )
+        resp = self.client.get(
+            f"/api/documents/{root.id}/download/?version={other_version.id}",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_download_preview_thumb_with_version_param(self) -> None:
+        root = self._create_pdf(title="root", checksum="root")
+        version = self._create_pdf(
+            title="v1",
+            checksum="v1",
+            root_document=root,
+        )
+        self._write_file(version.source_path, b"version")
+        self._write_file(version.thumbnail_path, b"thumb")
+
+        resp = self.client.get(
+            f"/api/documents/{root.id}/download/?version={version.id}",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertEqual(resp.content, b"version")
+
+        resp = self.client.get(
+            f"/api/documents/{root.id}/preview/?version={version.id}",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertEqual(resp.content, b"version")
+
+        resp = self.client.get(
+            f"/api/documents/{root.id}/thumb/?version={version.id}",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertEqual(resp.content, b"thumb")
+
+    def test_metadata_version_param_uses_version(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+        version = Document.objects.create(
+            title="v1",
+            checksum="v1",
+            mime_type="application/pdf",
+            root_document=root,
+        )
+
+        with mock.patch("documents.views.DocumentViewSet.get_metadata") as metadata:
+            metadata.return_value = []
+            resp = self.client.get(
+                f"/api/documents/{root.id}/metadata/?version={version.id}",
+            )
+
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertTrue(metadata.called)
+
+    def test_metadata_version_param_errors(self) -> None:
+        root = self._create_pdf(title="root", checksum="root")
+
+        resp = self.client.get(
+            f"/api/documents/{root.id}/metadata/?version=not-a-number",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
+
+        resp = self.client.get(f"/api/documents/{root.id}/metadata/?version=9999")
+        self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
+
+        other_root = self._create_pdf(title="other", checksum="other")
+        other_version = self._create_pdf(
+            title="other-v1",
+            checksum="other-v1",
+            root_document=other_root,
+        )
+        resp = self.client.get(
+            f"/api/documents/{root.id}/metadata/?version={other_version.id}",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_metadata_returns_403_when_user_lacks_permission(self) -> None:
+        owner = User.objects.create_user(username="owner")
+        other = User.objects.create_user(username="other")
+        other.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
+        doc = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+            owner=owner,
+        )
+        self.client.force_authenticate(user=other)
+
+        resp = self.client.get(f"/api/documents/{doc.id}/metadata/")
+
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_update_version_enqueues_consume_with_overrides(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+        upload = self._make_pdf_upload()
+
+        async_task = mock.Mock()
+        async_task.id = "task-123"
+
+        with mock.patch("documents.views.consume_file") as consume_mock:
+            consume_mock.delay.return_value = async_task
+            resp = self.client.post(
+                f"/api/documents/{root.id}/update_version/",
+                {"document": upload, "version_label": "  New Version  "},
+                format="multipart",
+            )
+
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertEqual(resp.data, "task-123")
+        consume_mock.delay.assert_called_once()
+        input_doc, overrides = consume_mock.delay.call_args[0]
+        self.assertEqual(input_doc.root_document_id, root.id)
+        self.assertEqual(input_doc.source, DocumentSource.ApiUpload)
+        self.assertEqual(overrides.version_label, "New Version")
+        self.assertEqual(overrides.actor_id, self.user.id)
+
+    def test_update_version_with_version_pk_normalizes_to_root(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+        version = Document.objects.create(
+            title="v1",
+            checksum="v1",
+            mime_type="application/pdf",
+            root_document=root,
+        )
+        upload = self._make_pdf_upload()
+
+        async_task = mock.Mock()
+        async_task.id = "task-123"
+
+        with mock.patch("documents.views.consume_file") as consume_mock:
+            consume_mock.delay.return_value = async_task
+            resp = self.client.post(
+                f"/api/documents/{version.id}/update_version/",
+                {"document": upload, "version_label": "  New Version  "},
+                format="multipart",
+            )
+
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertEqual(resp.data, "task-123")
+        consume_mock.delay.assert_called_once()
+        input_doc, overrides = consume_mock.delay.call_args[0]
+        self.assertEqual(input_doc.root_document_id, root.id)
+        self.assertEqual(overrides.version_label, "New Version")
+        self.assertEqual(overrides.actor_id, self.user.id)
+
+    def test_update_version_returns_500_on_consume_failure(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+        upload = self._make_pdf_upload()
+
+        with mock.patch("documents.views.consume_file") as consume_mock:
+            consume_mock.delay.side_effect = Exception("boom")
+            resp = self.client.post(
+                f"/api/documents/{root.id}/update_version/",
+                {"document": upload},
+                format="multipart",
+            )
+
+        self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+    def test_update_version_returns_403_without_permission(self) -> None:
+        owner = User.objects.create_user(username="owner")
+        other = User.objects.create_user(username="other")
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+            owner=owner,
+        )
+        self.client.force_authenticate(user=other)
+
+        resp = self.client.post(
+            f"/api/documents/{root.id}/update_version/",
+            {"document": self._make_pdf_upload()},
+            format="multipart",
+        )
+
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_update_version_returns_404_for_missing_document(self) -> None:
+        resp = self.client.post(
+            "/api/documents/9999/update_version/",
+            {"document": self._make_pdf_upload()},
+            format="multipart",
+        )
+
+        self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_update_version_requires_document(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+
+        resp = self.client.post(
+            f"/api/documents/{root.id}/update_version/",
+            {"version_label": "label"},
+            format="multipart",
+        )
+
+        self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_patch_content_updates_latest_version_content(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+            content="root-content",
+        )
+        v1 = Document.objects.create(
+            title="v1",
+            checksum="v1",
+            mime_type="application/pdf",
+            root_document=root,
+            content="v1-content",
+        )
+        v2 = Document.objects.create(
+            title="v2",
+            checksum="v2",
+            mime_type="application/pdf",
+            root_document=root,
+            content="v2-content",
+        )
+
+        resp = self.client.patch(
+            f"/api/documents/{root.id}/",
+            {"content": "edited-content"},
+            format="json",
+        )
+
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertEqual(resp.data["content"], "edited-content")
+        root.refresh_from_db()
+        v1.refresh_from_db()
+        v2.refresh_from_db()
+        self.assertEqual(v2.content, "edited-content")
+        self.assertEqual(root.content, "root-content")
+        self.assertEqual(v1.content, "v1-content")
+
+    def test_patch_content_updates_selected_version_content(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+            content="root-content",
+        )
+        v1 = Document.objects.create(
+            title="v1",
+            checksum="v1",
+            mime_type="application/pdf",
+            root_document=root,
+            content="v1-content",
+        )
+        v2 = Document.objects.create(
+            title="v2",
+            checksum="v2",
+            mime_type="application/pdf",
+            root_document=root,
+            content="v2-content",
+        )
+
+        resp = self.client.patch(
+            f"/api/documents/{root.id}/?version={v1.id}",
+            {"content": "edited-v1"},
+            format="json",
+        )
+
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertEqual(resp.data["content"], "edited-v1")
+        root.refresh_from_db()
+        v1.refresh_from_db()
+        v2.refresh_from_db()
+        self.assertEqual(v1.content, "edited-v1")
+        self.assertEqual(v2.content, "v2-content")
+        self.assertEqual(root.content, "root-content")
+
+    def test_retrieve_returns_latest_version_content(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+            content="root-content",
+        )
+        Document.objects.create(
+            title="v1",
+            checksum="v1",
+            mime_type="application/pdf",
+            root_document=root,
+            content="v1-content",
+        )
+
+        resp = self.client.get(f"/api/documents/{root.id}/")
+
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertEqual(resp.data["content"], "v1-content")
+
+    def test_retrieve_with_version_param_returns_selected_version_content(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+            content="root-content",
+        )
+        v1 = Document.objects.create(
+            title="v1",
+            checksum="v1",
+            mime_type="application/pdf",
+            root_document=root,
+            content="v1-content",
+        )
+
+        resp = self.client.get(f"/api/documents/{root.id}/?version={v1.id}")
+
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertEqual(resp.data["content"], "v1-content")
+
+
+class TestVersionAwareFilters(TestCase):
+    def test_title_content_filter_falls_back_to_content(self) -> None:
+        queryset = mock.Mock()
+        fallback_queryset = mock.Mock()
+        queryset.filter.side_effect = [FieldError("missing field"), fallback_queryset]
+
+        result = TitleContentFilter().filter(queryset, " latest ")
+
+        self.assertIs(result, fallback_queryset)
+        self.assertEqual(queryset.filter.call_count, 2)
+
+    def test_effective_content_filter_falls_back_to_content_lookup(self) -> None:
+        queryset = mock.Mock()
+        fallback_queryset = mock.Mock()
+        queryset.filter.side_effect = [FieldError("missing field"), fallback_queryset]
+
+        result = EffectiveContentFilter(lookup_expr="icontains").filter(
+            queryset,
+            " latest ",
+        )
+
+        self.assertIs(result, fallback_queryset)
+        first_kwargs = queryset.filter.call_args_list[0].kwargs
+        second_kwargs = queryset.filter.call_args_list[1].kwargs
+        self.assertEqual(first_kwargs, {"effective_content__icontains": "latest"})
+        self.assertEqual(second_kwargs, {"content__icontains": "latest"})
+
+    def test_effective_content_filter_returns_input_for_empty_values(self) -> None:
+        queryset = mock.Mock()
+
+        result = EffectiveContentFilter(lookup_expr="icontains").filter(queryset, "   ")
+
+        self.assertIs(result, queryset)
+        queryset.filter.assert_not_called()

src/documents/tests/test_api_documents.py

@@ -554,6 +554,36 @@ class TestDocumentApi(DirectoriesMixin, DocumentConsumeDelayMixin, APITestCase):
         self.assertIsNone(response.data[1]["actor"])
         self.assertEqual(response.data[1]["action"], "create")
 
+    def test_document_history_logs_version_deletion(self) -> None:
+        root_doc = Document.objects.create(
+            title="Root",
+            checksum="123",
+            mime_type="application/pdf",
+            owner=self.user,
+        )
+        version_doc = Document.objects.create(
+            title="Version",
+            checksum="456",
+            mime_type="application/pdf",
+            root_document=root_doc,
+            owner=self.user,
+        )
+
+        response = self.client.delete(
+            f"/api/documents/{root_doc.pk}/versions/{version_doc.pk}/",
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        response = self.client.get(f"/api/documents/{root_doc.pk}/history/")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(response.data), 2)
+        self.assertEqual(response.data[0]["actor"]["id"], self.user.id)
+        self.assertEqual(response.data[0]["action"], "update")
+        self.assertEqual(
+            response.data[0]["changes"],
+            {"Version Deleted": ["None", version_doc.pk]},
+        )
+
     @override_settings(AUDIT_LOG_ENABLED=False)
     def test_document_history_action_disabled(self) -> None:
         """
@@ -1212,6 +1242,38 @@ class TestDocumentApi(DirectoriesMixin, DocumentConsumeDelayMixin, APITestCase):
         self.assertIsNone(overrides.document_type_id)
         self.assertIsNone(overrides.tag_ids)
 
+    def test_document_filters_use_latest_version_content(self) -> None:
+        root = Document.objects.create(
+            title="versioned root",
+            checksum="root",
+            mime_type="application/pdf",
+            content="root-content",
+        )
+        version = Document.objects.create(
+            title="versioned root",
+            checksum="v1",
+            mime_type="application/pdf",
+            root_document=root,
+            content="latest-version-content",
+        )
+
+        response = self.client.get(
+            "/api/documents/?content__icontains=latest-version-content",
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        results = response.data["results"]
+        self.assertEqual(len(results), 1)
+        self.assertEqual(results[0]["id"], root.id)
+        self.assertEqual(results[0]["content"], version.content)
+
+        response = self.client.get(
+            "/api/documents/?title_content=latest-version-content",
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        results = response.data["results"]
+        self.assertEqual(len(results), 1)
+        self.assertEqual(results[0]["id"], root.id)
+
     def test_create_wrong_endpoint(self) -> None:
         response = self.client.post(
             "/api/documents/",

src/documents/tests/test_bulk_edit.py

@@ -1,4 +1,3 @@
-import hashlib
 import shutil
 from datetime import date
 from pathlib import Path
@@ -382,6 +381,55 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
             [self.doc3.id, self.doc4.id, self.doc5.id],
         )
 
+    def test_delete_root_document_deletes_all_versions(self) -> None:
+        version = Document.objects.create(
+            checksum="A-v1",
+            title="A version",
+            root_document=self.doc1,
+        )
+
+        bulk_edit.delete([self.doc1.id])
+
+        self.assertFalse(Document.objects.filter(id=self.doc1.id).exists())
+        self.assertFalse(Document.objects.filter(id=version.id).exists())
+
+    def test_delete_version_document_keeps_root(self) -> None:
+        version = Document.objects.create(
+            checksum="A-v1",
+            title="A version",
+            root_document=self.doc1,
+        )
+
+        bulk_edit.delete([version.id])
+
+        self.assertTrue(Document.objects.filter(id=self.doc1.id).exists())
+        self.assertFalse(Document.objects.filter(id=version.id).exists())
+
+    def test_get_root_and_current_doc_mapping(self) -> None:
+        version1 = Document.objects.create(
+            checksum="B-v1",
+            title="B version 1",
+            root_document=self.doc2,
+        )
+        version2 = Document.objects.create(
+            checksum="B-v2",
+            title="B version 2",
+            root_document=self.doc2,
+        )
+
+        root_ids_by_doc_id = bulk_edit._get_root_ids_by_doc_id(
+            [self.doc2.id, version1.id, version2.id],
+        )
+        self.assertEqual(root_ids_by_doc_id[self.doc2.id], self.doc2.id)
+        self.assertEqual(root_ids_by_doc_id[version1.id], self.doc2.id)
+        self.assertEqual(root_ids_by_doc_id[version2.id], self.doc2.id)
+
+        root_docs, current_docs = bulk_edit._get_root_and_current_docs_by_root_id(
+            {self.doc2.id},
+        )
+        self.assertEqual(root_docs[self.doc2.id].id, self.doc2.id)
+        self.assertEqual(current_docs[self.doc2.id].id, version2.id)
+
     @mock.patch("documents.tasks.bulk_update_documents.delay")
     def test_set_permissions(self, m) -> None:
         doc_ids = [self.doc1.id, self.doc2.id, self.doc3.id]
@@ -922,15 +970,8 @@ class TestPDFActions(DirectoriesMixin, TestCase):
 
         mock_consume_file.assert_not_called()
 
-    @mock.patch("documents.tasks.bulk_update_documents.si")
+    @mock.patch("documents.tasks.consume_file.delay")
-    @mock.patch("documents.tasks.update_document_content_maybe_archive_file.s")
+    def test_rotate(self, mock_consume_delay):
-    @mock.patch("celery.chord.delay")
-    def test_rotate(
-        self,
-        mock_chord,
-        mock_update_document,
-        mock_update_documents,
-    ) -> None:
         """
         GIVEN:
             - Existing documents
@@ -941,19 +982,22 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         """
         doc_ids = [self.doc1.id, self.doc2.id]
         result = bulk_edit.rotate(doc_ids, 90)
-        self.assertEqual(mock_update_document.call_count, 2)
+        self.assertEqual(mock_consume_delay.call_count, 2)
-        mock_update_documents.assert_called_once()
+        for call, expected_id in zip(
-        mock_chord.assert_called_once()
+            mock_consume_delay.call_args_list,
+            doc_ids,
+        ):
+            consumable, overrides = call.args
+            self.assertEqual(consumable.root_document_id, expected_id)
+            self.assertIsNotNone(overrides)
         self.assertEqual(result, "OK")
 
-    @mock.patch("documents.tasks.bulk_update_documents.si")
+    @mock.patch("documents.tasks.consume_file.delay")
-    @mock.patch("documents.tasks.update_document_content_maybe_archive_file.s")
     @mock.patch("pikepdf.Pdf.save")
     def test_rotate_with_error(
         self,
         mock_pdf_save,
-        mock_update_archive_file,
+        mock_consume_delay,
-        mock_update_documents,
     ):
         """
         GIVEN:
@@ -972,16 +1016,12 @@ class TestPDFActions(DirectoriesMixin, TestCase):
             error_str = cm.output[0]
             expected_str = "Error rotating document"
             self.assertIn(expected_str, error_str)
-            mock_update_archive_file.assert_not_called()
+            mock_consume_delay.assert_not_called()
 
-    @mock.patch("documents.tasks.bulk_update_documents.si")
+    @mock.patch("documents.tasks.consume_file.delay")
-    @mock.patch("documents.tasks.update_document_content_maybe_archive_file.s")
-    @mock.patch("celery.chord.delay")
     def test_rotate_non_pdf(
         self,
-        mock_chord,
+        mock_consume_delay,
-        mock_update_document,
-        mock_update_documents,
     ):
         """
         GIVEN:
@@ -993,17 +1033,18 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         """
         with self.assertLogs("paperless.bulk_edit", level="INFO") as cm:
             result = bulk_edit.rotate([self.doc2.id, self.img_doc.id], 90)
-            output_str = cm.output[1]
+            expected_str = f"Document {self.img_doc.id} is not a PDF, skipping rotation"
-            expected_str = "Document 4 is not a PDF, skipping rotation"
+            self.assertTrue(any(expected_str in line for line in cm.output))
-            self.assertIn(expected_str, output_str)
+            self.assertEqual(mock_consume_delay.call_count, 1)
-            self.assertEqual(mock_update_document.call_count, 1)
+            consumable, overrides = mock_consume_delay.call_args[0]
-            mock_update_documents.assert_called_once()
+            self.assertEqual(consumable.root_document_id, self.doc2.id)
-            mock_chord.assert_called_once()
+            self.assertIsNotNone(overrides)
             self.assertEqual(result, "OK")
 
-    @mock.patch("documents.tasks.update_document_content_maybe_archive_file.delay")
+    @mock.patch("documents.tasks.consume_file.delay")
     @mock.patch("pikepdf.Pdf.save")
-    def test_delete_pages(self, mock_pdf_save, mock_update_archive_file) -> None:
+    @mock.patch("documents.data_models.magic.from_file", return_value="application/pdf")
+    def test_delete_pages(self, mock_magic, mock_pdf_save, mock_consume_delay):
         """
         GIVEN:
             - Existing documents
@@ -1011,28 +1052,22 @@ class TestPDFActions(DirectoriesMixin, TestCase):
             - Delete pages action is called with 1 document and 2 pages
         THEN:
             - Save should be called once
-            - Archive file should be updated once
+            - A new version should be enqueued via consume_file
-            - The document's page_count should be reduced by the number of deleted pages
         """
         doc_ids = [self.doc2.id]
-        initial_page_count = self.doc2.page_count
         pages = [1, 3]
         result = bulk_edit.delete_pages(doc_ids, pages)
         mock_pdf_save.assert_called_once()
-        mock_update_archive_file.assert_called_once()
+        mock_consume_delay.assert_called_once()
+        consumable, overrides = mock_consume_delay.call_args[0]
+        self.assertEqual(consumable.root_document_id, self.doc2.id)
+        self.assertTrue(str(consumable.original_file).endswith("_pages_deleted.pdf"))
+        self.assertIsNotNone(overrides)
         self.assertEqual(result, "OK")
 
-        expected_page_count = initial_page_count - len(pages)
+    @mock.patch("documents.tasks.consume_file.delay")
-        self.doc2.refresh_from_db()
-        self.assertEqual(self.doc2.page_count, expected_page_count)
-
-    @mock.patch("documents.tasks.update_document_content_maybe_archive_file.delay")
     @mock.patch("pikepdf.Pdf.save")
-    def test_delete_pages_with_error(
+    def test_delete_pages_with_error(self, mock_pdf_save, mock_consume_delay):
-        self,
-        mock_pdf_save,
-        mock_update_archive_file,
-    ) -> None:
         """
         GIVEN:
             - Existing documents
@@ -1041,7 +1076,7 @@ class TestPDFActions(DirectoriesMixin, TestCase):
             - PikePDF raises an error
         THEN:
             - Save should be called once
-            - Archive file should not be updated
+            - No new version should be enqueued
         """
         mock_pdf_save.side_effect = Exception("Error saving PDF")
         doc_ids = [self.doc2.id]
@@ -1052,7 +1087,7 @@ class TestPDFActions(DirectoriesMixin, TestCase):
             error_str = cm.output[0]
             expected_str = "Error deleting pages from document"
             self.assertIn(expected_str, error_str)
-            mock_update_archive_file.assert_not_called()
+            mock_consume_delay.assert_not_called()
 
     @mock.patch("documents.bulk_edit.group")
     @mock.patch("documents.tasks.consume_file.s")
@@ -1151,24 +1186,18 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         self.doc2.refresh_from_db()
         self.assertEqual(self.doc2.archive_serial_number, 333)
 
-    @mock.patch("documents.tasks.update_document_content_maybe_archive_file.delay")
+    @mock.patch("documents.tasks.consume_file.delay")
-    def test_edit_pdf_with_update_document(
+    def test_edit_pdf_with_update_document(self, mock_consume_delay):
-        self,
-        mock_update_document: mock.Mock,
-    ) -> None:
         """
         GIVEN:
             - A single existing PDF document
         WHEN:
             - edit_pdf is called with update_document=True and a single output
         THEN:
-            - The original document is updated in-place
+            - A version update is enqueued targeting the existing document
-            - The update_document_content_maybe_archive_file task is triggered
         """
         doc_ids = [self.doc2.id]
         operations = [{"page": 1}, {"page": 2}]
-        original_checksum = self.doc2.checksum
-        original_page_count = self.doc2.page_count
 
         result = bulk_edit.edit_pdf(
             doc_ids,
@@ -1178,10 +1207,11 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         )
 
         self.assertEqual(result, "OK")
-        self.doc2.refresh_from_db()
+        mock_consume_delay.assert_called_once()
-        self.assertNotEqual(self.doc2.checksum, original_checksum)
+        consumable, overrides = mock_consume_delay.call_args[0]
-        self.assertNotEqual(self.doc2.page_count, original_page_count)
+        self.assertEqual(consumable.root_document_id, self.doc2.id)
-        mock_update_document.assert_called_once_with(document_id=self.doc2.id)
+        self.assertTrue(str(consumable.original_file).endswith("_edited.pdf"))
+        self.assertIsNotNone(overrides)
 
     @mock.patch("documents.bulk_edit.group")
     @mock.patch("documents.tasks.consume_file.s")
@@ -1258,10 +1288,20 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         mock_consume_file.assert_not_called()
 
     @mock.patch("documents.bulk_edit.update_document_content_maybe_archive_file.delay")
+    @mock.patch("documents.tasks.consume_file.delay")
+    @mock.patch("documents.bulk_edit.tempfile.mkdtemp")
     @mock.patch("pikepdf.open")
-    def test_remove_password_update_document(self, mock_open, mock_update_document):
+    def test_remove_password_update_document(
+        self,
+        mock_open,
+        mock_mkdtemp,
+        mock_consume_delay,
+        mock_update_document,
+    ):
         doc = self.doc1
-        original_checksum = doc.checksum
+        temp_dir = self.dirs.scratch_dir / "remove-password-update"
+        temp_dir.mkdir(parents=True, exist_ok=True)
+        mock_mkdtemp.return_value = str(temp_dir)
 
         fake_pdf = mock.MagicMock()
         fake_pdf.pages = [mock.Mock(), mock.Mock(), mock.Mock()]
@@ -1281,12 +1321,17 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         self.assertEqual(result, "OK")
         mock_open.assert_called_once_with(doc.source_path, password="secret")
         fake_pdf.remove_unreferenced_resources.assert_called_once()
-        doc.refresh_from_db()
+        mock_update_document.assert_not_called()
-        self.assertNotEqual(doc.checksum, original_checksum)
+        mock_consume_delay.assert_called_once()
-        expected_checksum = hashlib.md5(doc.source_path.read_bytes()).hexdigest()
+        consumable, overrides = mock_consume_delay.call_args[0]
-        self.assertEqual(doc.checksum, expected_checksum)
+        expected_path = temp_dir / f"{doc.id}_unprotected.pdf"
-        self.assertEqual(doc.page_count, len(fake_pdf.pages))
+        self.assertTrue(expected_path.exists())
-        mock_update_document.assert_called_once_with(document_id=doc.id)
+        self.assertEqual(
+            Path(consumable.original_file).resolve(),
+            expected_path.resolve(),
+        )
+        self.assertEqual(consumable.root_document_id, doc.id)
+        self.assertIsNotNone(overrides)
 
     @mock.patch("documents.bulk_edit.chord")
     @mock.patch("documents.bulk_edit.group")
@@ -1295,12 +1340,12 @@ class TestPDFActions(DirectoriesMixin, TestCase):
     @mock.patch("pikepdf.open")
     def test_remove_password_creates_consumable_document(
         self,
-        mock_open,
+        mock_open: mock.Mock,
-        mock_mkdtemp,
+        mock_mkdtemp: mock.Mock,
-        mock_consume_file,
+        mock_consume_file: mock.Mock,
-        mock_group,
+        mock_group: mock.Mock,
-        mock_chord,
+        mock_chord: mock.Mock,
-    ):
+    ) -> None:
         doc = self.doc2
         temp_dir = self.dirs.scratch_dir / "remove-password"
         temp_dir.mkdir(parents=True, exist_ok=True)
@@ -1309,8 +1354,8 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         fake_pdf = mock.MagicMock()
         fake_pdf.pages = [mock.Mock(), mock.Mock()]
 
-        def save_side_effect(target_path):
+        def save_side_effect(target_path: Path) -> None:
-            Path(target_path).write_bytes(b"password removed")
+            target_path.write_bytes(b"password removed")
 
         fake_pdf.save.side_effect = save_side_effect
         mock_open.return_value.__enter__.return_value = fake_pdf
@@ -1352,13 +1397,13 @@ class TestPDFActions(DirectoriesMixin, TestCase):
     @mock.patch("pikepdf.open")
     def test_remove_password_deletes_original(
         self,
-        mock_open,
+        mock_open: mock.Mock,
-        mock_mkdtemp,
+        mock_mkdtemp: mock.Mock,
-        mock_consume_file,
+        mock_consume_file: mock.Mock,
-        mock_group,
+        mock_group: mock.Mock,
-        mock_chord,
+        mock_chord: mock.Mock,
-        mock_delete,
+        mock_delete: mock.Mock,
-    ):
+    ) -> None:
         doc = self.doc2
         temp_dir = self.dirs.scratch_dir / "remove-password-delete"
         temp_dir.mkdir(parents=True, exist_ok=True)
@@ -1367,8 +1412,8 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         fake_pdf = mock.MagicMock()
         fake_pdf.pages = [mock.Mock(), mock.Mock()]
 
-        def save_side_effect(target_path):
+        def save_side_effect(target_path: Path) -> None:
-            Path(target_path).write_bytes(b"password removed")
+            target_path.write_bytes(b"password removed")
 
         fake_pdf.save.side_effect = save_side_effect
         mock_open.return_value.__enter__.return_value = fake_pdf
@@ -1391,7 +1436,7 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         mock_delete.si.assert_called_once_with([doc.id])
 
     @mock.patch("pikepdf.open")
-    def test_remove_password_open_failure(self, mock_open):
+    def test_remove_password_open_failure(self, mock_open: mock.Mock) -> None:
         mock_open.side_effect = RuntimeError("wrong password")
 
         with self.assertLogs("paperless.bulk_edit", level="ERROR") as cm:

src/documents/tests/test_consumer.py

@@ -16,6 +16,9 @@ from guardian.core import ObjectPermissionChecker
 
 from documents.barcodes import BarcodePlugin
 from documents.consumer import ConsumerError
+from documents.consumer import ConsumerPlugin
+from documents.consumer import ConsumerPreflightPlugin
+from documents.data_models import ConsumableDocument
 from documents.data_models import DocumentMetadataOverrides
 from documents.data_models import DocumentSource
 from documents.models import Correspondent
@@ -29,6 +32,7 @@ from documents.parsers import ParseError
 from documents.plugins.helpers import ProgressStatusOptions
 from documents.tasks import sanity_check
 from documents.tests.utils import DirectoriesMixin
+from documents.tests.utils import DummyProgressManager
 from documents.tests.utils import FileSystemAssertsMixin
 from documents.tests.utils import GetConsumerMixin
 from paperless_mail.models import MailRule
@@ -94,11 +98,13 @@ class FaultyGenericExceptionParser(_BaseTestParser):
         raise Exception("Generic exception.")
 
 
-def fake_magic_from_file(file, *, mime=False):
+def fake_magic_from_file(file, *, mime=False):  # NOSONAR
     if mime:
         filepath = Path(file)
         if filepath.name.startswith("invalid_pdf"):
             return "application/octet-stream"
+        if filepath.name.startswith("valid_pdf"):
+            return "application/pdf"
         if filepath.suffix == ".pdf":
             return "application/pdf"
         elif filepath.suffix == ".png":
@@ -664,6 +670,144 @@ class TestConsumer(
 
         self._assert_first_last_send_progress()
 
+    @mock.patch("documents.consumer.load_classifier")
+    def test_version_label_override_applies(self, m) -> None:
+        m.return_value = MagicMock()
+
+        with self.get_consumer(
+            self.get_test_file(),
+            DocumentMetadataOverrides(version_label="v1"),
+        ) as consumer:
+            consumer.run()
+
+        document = Document.objects.first()
+        assert document is not None
+
+        self.assertEqual(document.version_label, "v1")
+
+        self._assert_first_last_send_progress()
+
+    @override_settings(AUDIT_LOG_ENABLED=True)
+    @mock.patch("documents.consumer.load_classifier")
+    def test_consume_version_creates_new_version(self, m) -> None:
+        m.return_value = MagicMock()
+
+        with self.get_consumer(self.get_test_file()) as consumer:
+            consumer.run()
+
+        root_doc = Document.objects.first()
+        self.assertIsNotNone(root_doc)
+        assert root_doc is not None
+
+        actor = User.objects.create_user(
+            username="actor",
+            email="actor@example.com",
+            password="password",
+        )
+
+        version_file = self.get_test_file2()
+        status = DummyProgressManager(version_file.name, None)
+        overrides = DocumentMetadataOverrides(
+            version_label="v2",
+            actor_id=actor.pk,
+        )
+        doc = ConsumableDocument(
+            DocumentSource.ApiUpload,
+            original_file=version_file,
+            root_document_id=root_doc.pk,
+        )
+        preflight = ConsumerPreflightPlugin(
+            doc,
+            overrides,
+            status,  # type: ignore[arg-type]
+            self.dirs.scratch_dir,
+            "task-id",
+        )
+        preflight.setup()
+        preflight.run()
+
+        consumer = ConsumerPlugin(
+            doc,
+            overrides,
+            status,  # type: ignore[arg-type]
+            self.dirs.scratch_dir,
+            "task-id",
+        )
+        consumer.setup()
+        try:
+            self.assertTrue(consumer.filename.endswith("_v0.pdf"))
+            consumer.run()
+        finally:
+            consumer.cleanup()
+
+        versions = Document.objects.filter(root_document=root_doc)
+        self.assertEqual(versions.count(), 1)
+        version = versions.first()
+        assert version is not None
+        assert version.original_filename is not None
+        self.assertEqual(version.version_label, "v2")
+        self.assertTrue(version.original_filename.endswith("_v0.pdf"))
+        self.assertTrue(bool(version.content))
+
+    @override_settings(AUDIT_LOG_ENABLED=True)
+    @mock.patch("documents.consumer.load_classifier")
+    def test_consume_version_with_missing_actor_and_filename_without_suffix(
+        self,
+        m: mock.Mock,
+    ) -> None:
+        m.return_value = MagicMock()
+
+        with self.get_consumer(self.get_test_file()) as consumer:
+            consumer.run()
+
+        root_doc = Document.objects.first()
+        self.assertIsNotNone(root_doc)
+        assert root_doc is not None
+
+        version_file = self.get_test_file2()
+        status = DummyProgressManager(version_file.name, None)
+        overrides = DocumentMetadataOverrides(
+            filename="valid_pdf_version-upload",
+            actor_id=999999,
+        )
+        doc = ConsumableDocument(
+            DocumentSource.ApiUpload,
+            original_file=version_file,
+            root_document_id=root_doc.pk,
+        )
+
+        preflight = ConsumerPreflightPlugin(
+            doc,
+            overrides,
+            status,  # type: ignore[arg-type]
+            self.dirs.scratch_dir,
+            "task-id",
+        )
+        preflight.setup()
+        preflight.run()
+
+        consumer = ConsumerPlugin(
+            doc,
+            overrides,
+            status,  # type: ignore[arg-type]
+            self.dirs.scratch_dir,
+            "task-id",
+        )
+        consumer.setup()
+        try:
+            self.assertEqual(consumer.filename, "valid_pdf_version-upload_v0")
+            consumer.run()
+        finally:
+            consumer.cleanup()
+
+        version = (
+            Document.objects.filter(root_document=root_doc).order_by("-id").first()
+        )
+        self.assertIsNotNone(version)
+        assert version is not None
+        self.assertEqual(version.original_filename, "valid_pdf_version-upload_v0")
+        self.assertTrue(bool(version.content))
+
     @mock.patch("documents.consumer.load_classifier")
     def testClassifyDocument(self, m) -> None:
         correspondent = Correspondent.objects.create(
@@ -1179,7 +1323,7 @@ class PostConsumeTestCase(DirectoriesMixin, GetConsumerMixin, TestCase):
                 consumer.run_post_consume_script(doc)
 
     @mock.patch("documents.consumer.run_subprocess")
-    def test_post_consume_script_simple(self, m) -> None:
+    def test_post_consume_script_simple(self, m: mock.MagicMock) -> None:
         with tempfile.NamedTemporaryFile() as script:
             with override_settings(POST_CONSUME_SCRIPT=script.name):
                 doc = Document.objects.create(title="Test", mime_type="application/pdf")
@@ -1190,7 +1334,10 @@ class PostConsumeTestCase(DirectoriesMixin, GetConsumerMixin, TestCase):
                 m.assert_called_once()
 
     @mock.patch("documents.consumer.run_subprocess")
-    def test_post_consume_script_with_correspondent_and_type(self, m) -> None:
+    def test_post_consume_script_with_correspondent_and_type(
+        self,
+        m: mock.MagicMock,
+    ) -> None:
         with tempfile.NamedTemporaryFile() as script:
             with override_settings(POST_CONSUME_SCRIPT=script.name):
                 c = Correspondent.objects.create(name="my_bank")
@@ -1273,6 +1420,19 @@ class TestMetadataOverrides(TestCase):
         base.update(incoming)
         self.assertTrue(base.skip_asn_if_exists)
 
+    def test_update_actor_and_version_label(self) -> None:
+        base = DocumentMetadataOverrides(
+            actor_id=1,
+            version_label="root",
+        )
+        incoming = DocumentMetadataOverrides(
+            actor_id=2,
+            version_label="v2",
+        )
+        base.update(incoming)
+        self.assertEqual(base.actor_id, 2)
+        self.assertEqual(base.version_label, "v2")
+
 
 class TestBarcodeApplyDetectedASN(TestCase):
     """

src/documents/tests/test_document_model.py

@@ -78,6 +78,28 @@ class TestDocument(TestCase):
             empty_trash([document.pk])
             self.assertEqual(mock_unlink.call_count, 2)
 
+    def test_delete_root_deletes_versions(self) -> None:
+        root = Document.objects.create(
+            correspondent=Correspondent.objects.create(name="Test0"),
+            title="Head",
+            content="content",
+            checksum="checksum",
+            mime_type="application/pdf",
+        )
+        Document.objects.create(
+            root_document=root,
+            correspondent=root.correspondent,
+            title="Version",
+            content="content",
+            checksum="checksum2",
+            mime_type="application/pdf",
+        )
+
+        root.delete()
+
+        self.assertEqual(Document.objects.count(), 0)
+        self.assertEqual(Document.deleted_objects.count(), 2)
+
     def test_file_name(self) -> None:
         doc = Document(
             mime_type="application/pdf",

src/documents/tests/test_task_signals.py

@@ -7,7 +7,9 @@ from django.test import TestCase
 from documents.data_models import ConsumableDocument
 from documents.data_models import DocumentMetadataOverrides
 from documents.data_models import DocumentSource
+from documents.models import Document
 from documents.models import PaperlessTask
+from documents.signals.handlers import add_to_index
 from documents.signals.handlers import before_task_publish_handler
 from documents.signals.handlers import task_failure_handler
 from documents.signals.handlers import task_postrun_handler
@@ -198,3 +200,39 @@ class TestTaskSignalHandler(DirectoriesMixin, TestCase):
         task = PaperlessTask.objects.get()
 
         self.assertEqual(celery.states.FAILURE, task.status)
+
+    def test_add_to_index_indexes_root_once_for_root_documents(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+
+        with mock.patch("documents.index.add_or_update_document") as add:
+            add_to_index(sender=None, document=root)
+
+        add.assert_called_once_with(root)
+
+    def test_add_to_index_reindexes_root_for_version_documents(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+        version = Document.objects.create(
+            title="version",
+            checksum="version",
+            mime_type="application/pdf",
+            root_document=root,
+        )
+
+        with mock.patch("documents.index.add_or_update_document") as add:
+            add_to_index(sender=None, document=version)
+
+        self.assertEqual(add.call_count, 2)
+        self.assertEqual(add.call_args_list[0].args[0].id, version.id)
+        self.assertEqual(add.call_args_list[1].args[0].id, root.id)
+        self.assertEqual(
+            add.call_args_list[1].kwargs,
+            {"effective_content": version.content},
+        )

src/documents/tests/test_version_conditionals.py

@@ -0,0 +1,91 @@
+from types import SimpleNamespace
+from unittest import mock
+
+from django.test import TestCase
+
+from documents.conditionals import metadata_etag
+from documents.conditionals import preview_etag
+from documents.conditionals import thumbnail_last_modified
+from documents.models import Document
+from documents.tests.utils import DirectoriesMixin
+from documents.versioning import resolve_effective_document_by_pk
+
+
+class TestConditionals(DirectoriesMixin, TestCase):
+    def test_metadata_etag_uses_latest_version_for_root_request(self) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root-checksum",
+            archive_checksum="root-archive",
+            mime_type="application/pdf",
+        )
+        latest = Document.objects.create(
+            title="v1",
+            checksum="version-checksum",
+            archive_checksum="version-archive",
+            mime_type="application/pdf",
+            root_document=root,
+        )
+        request = SimpleNamespace(query_params={})
+
+        self.assertEqual(metadata_etag(request, root.id), latest.checksum)
+        self.assertEqual(preview_etag(request, root.id), latest.archive_checksum)
+
+    def test_resolve_effective_doc_returns_none_for_invalid_or_unrelated_version(
+        self,
+    ) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+        other_root = Document.objects.create(
+            title="other",
+            checksum="other",
+            mime_type="application/pdf",
+        )
+        other_version = Document.objects.create(
+            title="other-v1",
+            checksum="other-v1",
+            mime_type="application/pdf",
+            root_document=other_root,
+        )
+
+        invalid_request = SimpleNamespace(query_params={"version": "not-a-number"})
+        unrelated_request = SimpleNamespace(
+            query_params={"version": str(other_version.id)},
+        )
+
+        self.assertIsNone(
+            resolve_effective_document_by_pk(root.id, invalid_request).document,
+        )
+        self.assertIsNone(
+            resolve_effective_document_by_pk(root.id, unrelated_request).document,
+        )
+
+    def test_thumbnail_last_modified_uses_effective_document_for_cache_key(
+        self,
+    ) -> None:
+        root = Document.objects.create(
+            title="root",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+        latest = Document.objects.create(
+            title="v2",
+            checksum="v2",
+            mime_type="application/pdf",
+            root_document=root,
+        )
+        latest.thumbnail_path.parent.mkdir(parents=True, exist_ok=True)
+        latest.thumbnail_path.write_bytes(b"thumb")
+
+        request = SimpleNamespace(query_params={})
+        with mock.patch(
+            "documents.conditionals.get_thumbnail_modified_key",
+            return_value="thumb-modified-key",
+        ) as get_thumb_key:
+            result = thumbnail_last_modified(request, root.id)
+
+        self.assertIsNotNone(result)
+        get_thumb_key.assert_called_once_with(latest.id)

src/documents/versioning.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+from documents.models import Document
+
+
+class VersionResolutionError(str, Enum):
+    INVALID = "invalid"
+    NOT_FOUND = "not_found"
+
+
+@dataclass(frozen=True)
+class VersionResolution:
+    document: Document | None
+    error: VersionResolutionError | None = None
+
+
+def _document_manager(*, include_deleted: bool) -> Any:
+    return Document.global_objects if include_deleted else Document.objects
+
+
+def get_request_version_param(request: Any) -> str | None:
+    if hasattr(request, "query_params"):
+        return request.query_params.get("version")
+    return None
+
+
+def get_root_document(doc: Document, *, include_deleted: bool = False) -> Document:
+    # Use root_document_id to avoid a query when this is already a root.
+    # If root_document isn't available, fall back to the document itself.
+    if doc.root_document_id is None:
+        return doc
+    if doc.root_document is not None:
+        return doc.root_document
+
+    manager = _document_manager(include_deleted=include_deleted)
+    root_doc = manager.only("id").filter(id=doc.root_document_id).first()
+    return root_doc or doc
+
+
+def get_latest_version_for_root(
+    root_doc: Document,
+    *,
+    include_deleted: bool = False,
+) -> Document:
+    manager = _document_manager(include_deleted=include_deleted)
+    latest = manager.filter(root_document=root_doc).order_by("-id").first()
+    return latest or root_doc
+
+
+def resolve_requested_version_for_root(
+    root_doc: Document,
+    request: Any,
+    *,
+    include_deleted: bool = False,
+) -> VersionResolution:
+    version_param = get_request_version_param(request)
+    if not version_param:
+        return VersionResolution(
+            document=get_latest_version_for_root(
+                root_doc,
+                include_deleted=include_deleted,
+            ),
+        )
+
+    try:
+        version_id = int(version_param)
+    except (TypeError, ValueError):
+        return VersionResolution(document=None, error=VersionResolutionError.INVALID)
+
+    manager = _document_manager(include_deleted=include_deleted)
+    candidate = manager.only("id", "root_document_id").filter(id=version_id).first()
+    if candidate is None:
+        return VersionResolution(document=None, error=VersionResolutionError.NOT_FOUND)
+    if candidate.id != root_doc.id and candidate.root_document_id != root_doc.id:
+        return VersionResolution(document=None, error=VersionResolutionError.NOT_FOUND)
+    return VersionResolution(document=candidate)
+
+
+def resolve_effective_document(
+    request_doc: Document,
+    request: Any,
+    *,
+    include_deleted: bool = False,
+) -> VersionResolution:
+    root_doc = get_root_document(request_doc, include_deleted=include_deleted)
+    if get_request_version_param(request) is not None:
+        return resolve_requested_version_for_root(
+            root_doc,
+            request,
+            include_deleted=include_deleted,
+        )
+    if request_doc.root_document_id is None:
+        return VersionResolution(
+            document=get_latest_version_for_root(
+                root_doc,
+                include_deleted=include_deleted,
+            ),
+        )
+    return VersionResolution(document=request_doc)
+
+
+def resolve_effective_document_by_pk(
+    pk: int,
+    request: Any,
+    *,
+    include_deleted: bool = False,
+) -> VersionResolution:
+    manager = _document_manager(include_deleted=include_deleted)
+    request_doc = manager.only("id", "root_document_id").filter(pk=pk).first()
+    if request_doc is None:
+        return VersionResolution(document=None, error=VersionResolutionError.NOT_FOUND)
+    return resolve_effective_document(
+        request_doc,
+        request,
+        include_deleted=include_deleted,
+    )

src/documents/views.py

@@ -10,6 +10,7 @@ from collections import deque
 from datetime import datetime
 from pathlib import Path
 from time import mktime
+from typing import Any
 from typing import Literal
 from unicodedata import normalize
 from urllib.parse import quote
@@ -29,16 +30,23 @@ from django.db.migrations.loader import MigrationLoader
 from django.db.migrations.recorder import MigrationRecorder
 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
+from django.db.models import OuterRef
+from django.db.models import Prefetch
 from django.db.models import Q
+from django.db.models import Subquery
 from django.db.models import Sum
 from django.db.models import When
+from django.db.models.functions import Coalesce
 from django.db.models.functions import Lower
 from django.db.models.manager import Manager
+from django.db.models.query import QuerySet
 from django.http import FileResponse
 from django.http import Http404
+from django.http import HttpRequest
 from django.http import HttpResponse
 from django.http import HttpResponseBadRequest
 from django.http import HttpResponseForbidden
@@ -83,6 +91,7 @@ from rest_framework.mixins import ListModelMixin
 from rest_framework.mixins import RetrieveModelMixin
 from rest_framework.mixins import UpdateModelMixin
 from rest_framework.permissions import IsAuthenticated
+from rest_framework.request import Request
 from rest_framework.response import Response
 from rest_framework.viewsets import GenericViewSet
 from rest_framework.viewsets import ModelViewSet
@@ -168,6 +177,7 @@ from documents.serialisers import CustomFieldSerializer
 from documents.serialisers import DocumentListSerializer
 from documents.serialisers import DocumentSerializer
 from documents.serialisers import DocumentTypeSerializer
+from documents.serialisers import DocumentVersionSerializer
 from documents.serialisers import EmailSerializer
 from documents.serialisers import NotesSerializer
 from documents.serialisers import PostDocumentSerializer
@@ -196,6 +206,11 @@ from documents.tasks import sanity_check
 from documents.tasks import train_classifier
 from documents.tasks import update_document_parent_tags
 from documents.utils import get_boolean
+from documents.versioning import VersionResolutionError
+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 resolve_requested_version_for_root
 from paperless import version
 from paperless.celery import app as celery_app
 from paperless.config import AIConfig
@@ -747,7 +762,7 @@ class DocumentViewSet(
     GenericViewSet,
 ):
     model = Document
-    queryset = Document.objects.annotate(num_notes=Count("notes"))
+    queryset = Document.objects.all()
     serializer_class = DocumentSerializer
     pagination_class = StandardPagination
     permission_classes = (IsAuthenticated, PaperlessObjectPermissions)
@@ -758,7 +773,7 @@ class DocumentViewSet(
         ObjectOwnedOrGrantedPermissionsFilter,
     )
     filterset_class = DocumentFilterSet
-    search_fields = ("title", "correspondent__name", "content")
+    search_fields = ("title", "correspondent__name", "effective_content")
     ordering_fields = (
         "id",
         "title",
@@ -776,12 +791,33 @@ class DocumentViewSet(
     )
 
     def get_queryset(self):
+        latest_version_content = Subquery(
+            Document.objects.filter(root_document=OuterRef("pk"))
+            .order_by("-id")
+            .values("content")[:1],
+        )
         return (
-            Document.objects.distinct()
+            Document.objects.filter(root_document__isnull=True)
+            .distinct()
             .order_by("-created")
+            .annotate(effective_content=Coalesce(latest_version_content, F("content")))
             .annotate(num_notes=Count("notes"))
             .select_related("correspondent", "storage_path", "document_type", "owner")
-            .prefetch_related("tags", "custom_fields", "notes")
+            .prefetch_related(
+                Prefetch(
+                    "versions",
+                    queryset=Document.objects.only(
+                        "id",
+                        "added",
+                        "checksum",
+                        "version_label",
+                        "root_document_id",
+                    ),
+                ),
+                "tags",
+                "custom_fields",
+                "notes",
+            )
         )
 
     def get_serializer(self, *args, **kwargs):
@@ -803,15 +839,100 @@ class DocumentViewSet(
         )
         return super().get_serializer(*args, **kwargs)
 
+    @extend_schema(
+        operation_id="documents_root",
+        responses=inline_serializer(
+            name="DocumentRootResponse",
+            fields={
+                "root_id": serializers.IntegerField(),
+            },
+        ),
+    )
+    @action(methods=["get"], detail=True, url_path="root")
+    def root(self, request, pk=None):
+        try:
+            doc = Document.global_objects.select_related(
+                "owner",
+                "root_document",
+            ).get(pk=pk)
+        except Document.DoesNotExist:
+            raise Http404
+
+        root_doc = get_root_document(doc)
+        if request.user is not None and not has_perms_owner_aware(
+            request.user,
+            "view_document",
+            root_doc,
+        ):
+            return HttpResponseForbidden("Insufficient permissions")
+
+        return Response({"root_id": root_doc.id})
+
+    def retrieve(
+        self,
+        request: Request,
+        *args: Any,
+        **kwargs: Any,
+    ) -> Response:
+        response = super().retrieve(request, *args, **kwargs)
+        if (
+            "version" not in request.query_params
+            or not isinstance(response.data, dict)
+            or "content" not in response.data
+        ):
+            return response
+
+        root_doc = self.get_object()
+        content_doc = self._resolve_file_doc(root_doc, request)
+        response.data["content"] = content_doc.content or ""
+        return response
+
     def update(self, request, *args, **kwargs):
-        response = super().update(request, *args, **kwargs)
+        partial = kwargs.pop("partial", False)
+        root_doc = self.get_object()
+        content_doc = (
+            self._resolve_file_doc(root_doc, request)
+            if "version" in request.query_params
+            else get_latest_version_for_root(root_doc)
+        )
+        content_updated = "content" in request.data
+        updated_content = request.data.get("content") if content_updated else None
+
+        data = request.data.copy()
+        serializer_partial = partial
+        if content_updated and content_doc.id != root_doc.id:
+            if updated_content is None:
+                raise ValidationError({"content": ["This field may not be null."]})
+            data.pop("content", None)
+            serializer_partial = True
+
+        serializer = self.get_serializer(
+            root_doc,
+            data=data,
+            partial=serializer_partial,
+        )
+        serializer.is_valid(raise_exception=True)
+        self.perform_update(serializer)
+
+        if content_updated and content_doc.id != root_doc.id:
+            content_doc.content = (
+                str(updated_content) if updated_content is not None else ""
+            )
+            content_doc.save(update_fields=["content", "modified"])
+
+        refreshed_doc = self.get_queryset().get(pk=root_doc.pk)
+        response_data = self.get_serializer(refreshed_doc).data
+        if "version" in request.query_params and "content" in response_data:
+            response_data["content"] = content_doc.content
+        response = Response(response_data)
+
         from documents import index
 
-        index.add_or_update_document(self.get_object())
+        index.add_or_update_document(refreshed_doc)
 
         document_updated.send(
             sender=self.__class__,
-            document=self.get_object(),
+            document=refreshed_doc,
         )
 
         return response
@@ -839,18 +960,74 @@ class DocumentViewSet(
             and request.query_params["original"] == "true"
         )
 
-    def file_response(self, pk, request, disposition):
+    def _resolve_file_doc(self, root_doc: Document, request):
-        doc = Document.global_objects.select_related("owner").get(id=pk)
+        version_requested = get_request_version_param(request) is not None
+        resolution = resolve_requested_version_for_root(
+            root_doc,
+            request,
+            include_deleted=version_requested,
+        )
+        if resolution.error == VersionResolutionError.INVALID:
+            raise NotFound("Invalid version parameter")
+        if resolution.document is None:
+            raise Http404
+        return resolution.document
+
+    def _get_effective_file_doc(
+        self,
+        request_doc: Document,
+        root_doc: Document,
+        request: Request,
+    ) -> Document:
+        if (
+            request_doc.root_document_id is not None
+            and get_request_version_param(request) is None
+        ):
+            return request_doc
+        return self._resolve_file_doc(root_doc, request)
+
+    def _resolve_request_and_root_doc(
+        self,
+        pk,
+        request: Request,
+        *,
+        include_deleted: bool = False,
+    ) -> tuple[Document, Document] | HttpResponseForbidden:
+        manager = Document.global_objects if include_deleted else Document.objects
+        try:
+            request_doc = manager.select_related(
+                "owner",
+                "root_document",
+            ).get(id=pk)
+        except Document.DoesNotExist:
+            raise Http404
+
+        root_doc = get_root_document(
+            request_doc,
+            include_deleted=include_deleted,
+        )
         if request.user is not None and not has_perms_owner_aware(
             request.user,
             "view_document",
-            doc,
+            root_doc,
         ):
             return HttpResponseForbidden("Insufficient permissions")
+        return request_doc, root_doc
+
+    def file_response(self, pk, request, disposition):
+        resolved = self._resolve_request_and_root_doc(
+            pk,
+            request,
+            include_deleted=True,
+        )
+        if isinstance(resolved, HttpResponseForbidden):
+            return resolved
+        request_doc, root_doc = resolved
+        file_doc = self._get_effective_file_doc(request_doc, root_doc, request)
         return serve_file(
-            doc=doc,
+            doc=file_doc,
             use_archive=not self.original_requested(request)
-            and doc.has_archive_version,
+            and file_doc.has_archive_version,
             disposition=disposition,
         )
 
@@ -884,16 +1061,14 @@ class DocumentViewSet(
         condition(etag_func=metadata_etag, last_modified_func=metadata_last_modified),
     )
     def metadata(self, request, pk=None):
-        try:
+        resolved = self._resolve_request_and_root_doc(pk, request)
-            doc = Document.objects.select_related("owner").get(pk=pk)
+        if isinstance(resolved, HttpResponseForbidden):
-            if request.user is not None and not has_perms_owner_aware(
+            return resolved
-                request.user,
+        request_doc, root_doc = resolved
-                "view_document",
+
-                doc,
+        # Choose the effective document (newest version by default,
-            ):
+        # or explicit via ?version=).
-                return HttpResponseForbidden("Insufficient permissions")
+        doc = self._get_effective_file_doc(request_doc, root_doc, request)
-        except Document.DoesNotExist:
-            raise Http404
 
         document_cached_metadata = get_metadata_cache(doc.pk)
 
@@ -1062,29 +1237,38 @@ class DocumentViewSet(
         condition(etag_func=preview_etag, last_modified_func=preview_last_modified),
     )
     def preview(self, request, pk=None):
+        resolved = self._resolve_request_and_root_doc(pk, request)
+        if isinstance(resolved, HttpResponseForbidden):
+            return resolved
+        request_doc, root_doc = resolved
+
         try:
-            response = self.file_response(pk, request, "inline")
+            file_doc = self._get_effective_file_doc(request_doc, root_doc, request)
-            return response
+
-        except (FileNotFoundError, Document.DoesNotExist):
+            return serve_file(
+                doc=file_doc,
+                use_archive=not self.original_requested(request)
+                and file_doc.has_archive_version,
+                disposition="inline",
+            )
+        except FileNotFoundError:
             raise Http404
 
     @action(methods=["get"], detail=True, filter_backends=[])
     @method_decorator(cache_control(no_cache=True))
     @method_decorator(last_modified(thumbnail_last_modified))
     def thumb(self, request, pk=None):
-        try:
+        resolved = self._resolve_request_and_root_doc(pk, request)
-            doc = Document.objects.select_related("owner").get(id=pk)
+        if isinstance(resolved, HttpResponseForbidden):
-            if request.user is not None and not has_perms_owner_aware(
+            return resolved
-                request.user,
+        request_doc, root_doc = resolved
-                "view_document",
-                doc,
-            ):
-                return HttpResponseForbidden("Insufficient permissions")
 
-            handle = doc.thumbnail_file
+        try:
+            file_doc = self._get_effective_file_doc(request_doc, root_doc, request)
+            handle = file_doc.thumbnail_file
 
             return HttpResponse(handle, content_type="image/webp")
-        except (FileNotFoundError, Document.DoesNotExist):
+        except FileNotFoundError:
             raise Http404
 
     @action(methods=["get"], detail=True)
@@ -1373,6 +1557,164 @@ class DocumentViewSet(
                 "Error emailing documents, check logs for more detail.",
             )
 
+    @extend_schema(
+        operation_id="documents_update_version",
+        request=DocumentVersionSerializer,
+        responses={
+            200: OpenApiTypes.STR,
+        },
+    )
+    @action(methods=["post"], detail=True, parser_classes=[parsers.MultiPartParser])
+    def update_version(self, request, pk=None):
+        serializer = DocumentVersionSerializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
+        try:
+            request_doc = Document.objects.select_related(
+                "owner",
+                "root_document",
+            ).get(pk=pk)
+            root_doc = get_root_document(request_doc)
+            if request.user is not None and not has_perms_owner_aware(
+                request.user,
+                "change_document",
+                root_doc,
+            ):
+                return HttpResponseForbidden("Insufficient permissions")
+        except Document.DoesNotExist:
+            raise Http404
+
+        try:
+            doc_name, doc_data = serializer.validated_data.get("document")
+            version_label = serializer.validated_data.get("version_label")
+
+            t = int(mktime(datetime.now().timetuple()))
+
+            settings.SCRATCH_DIR.mkdir(parents=True, exist_ok=True)
+
+            temp_file_path = Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR)) / Path(
+                pathvalidate.sanitize_filename(doc_name),
+            )
+
+            temp_file_path.write_bytes(doc_data)
+
+            os.utime(temp_file_path, times=(t, t))
+
+            input_doc = ConsumableDocument(
+                source=DocumentSource.ApiUpload,
+                original_file=temp_file_path,
+                root_document_id=root_doc.pk,
+            )
+
+            overrides = DocumentMetadataOverrides()
+            if version_label:
+                overrides.version_label = version_label.strip()
+            if request.user is not None:
+                overrides.actor_id = request.user.id
+
+            async_task = consume_file.delay(
+                input_doc,
+                overrides,
+            )
+            logger.debug(
+                f"Updated document {root_doc.id} with new version",
+            )
+            return Response(async_task.id)
+        except Exception as e:
+            logger.warning(f"An error occurred updating document: {e!s}")
+            return HttpResponseServerError(
+                "Error updating document, check logs for more detail.",
+            )
+
+    @extend_schema(
+        operation_id="documents_delete_version",
+        parameters=[
+            OpenApiParameter(
+                name="version_id",
+                type=OpenApiTypes.INT,
+                location=OpenApiParameter.PATH,
+            ),
+        ],
+        responses=inline_serializer(
+            name="DeleteDocumentVersionResult",
+            fields={
+                "result": serializers.CharField(),
+                "current_version_id": serializers.IntegerField(),
+            },
+        ),
+    )
+    @action(
+        methods=["delete"],
+        detail=True,
+        url_path=r"versions/(?P<version_id>\d+)",
+    )
+    def delete_version(self, request, pk=None, version_id=None):
+        try:
+            root_doc = Document.objects.select_related(
+                "owner",
+                "root_document",
+            ).get(pk=pk)
+            root_doc = get_root_document(root_doc)
+        except Document.DoesNotExist:
+            raise Http404
+
+        if request.user is not None and not has_perms_owner_aware(
+            request.user,
+            "delete_document",
+            root_doc,
+        ):
+            return HttpResponseForbidden("Insufficient permissions")
+
+        try:
+            version_doc = Document.objects.select_related("owner").get(
+                pk=version_id,
+            )
+        except Document.DoesNotExist:
+            raise Http404
+
+        if version_doc.id == root_doc.id:
+            return HttpResponseBadRequest(
+                "Cannot delete the root/original version. Delete the document instead.",
+            )
+
+        if version_doc.root_document_id != root_doc.id:
+            raise Http404
+
+        from documents import index
+
+        index.remove_document_from_index(version_doc)
+        version_doc_id = version_doc.id
+        version_doc.delete()
+        index.add_or_update_document(root_doc)
+        if settings.AUDIT_LOG_ENABLED:
+            actor = (
+                request.user if request.user and request.user.is_authenticated else None
+            )
+            LogEntry.objects.log_create(
+                instance=root_doc,
+                changes={
+                    "Version Deleted": ["None", version_doc_id],
+                },
+                action=LogEntry.Action.UPDATE,
+                actor=actor,
+                additional_data={
+                    "reason": "Version deleted",
+                    "version_id": version_doc_id,
+                },
+            )
+
+        current = (
+            Document.objects.filter(Q(id=root_doc.id) | Q(root_document=root_doc))
+            .order_by("-id")
+            .first()
+        )
+        return Response(
+            {
+                "result": "OK",
+                "current_version_id": current.id if current else root_doc.id,
+            },
+        )
+
 
 class ChatStreamingSerializer(serializers.Serializer):
     q = serializers.CharField(required=True)
@@ -1461,7 +1803,7 @@ class ChatStreamingView(GenericAPIView):
     ),
 )
 class UnifiedSearchViewSet(DocumentViewSet):
-    def __init__(self, *args, **kwargs) -> None:
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
         self.searcher = None
 
@@ -1639,7 +1981,7 @@ class SavedViewViewSet(ModelViewSet, PassUserMixin):
             .prefetch_related("filter_rules")
         )
 
-    def perform_create(self, serializer) -> None:
+    def perform_create(self, serializer: serializers.BaseSerializer[Any]) -> None:
         serializer.save(owner=self.request.user)
 
 
@@ -1672,13 +2014,15 @@ class BulkEditView(PassUserMixin):
         "modify_custom_fields": "custom_fields",
         "set_permissions": None,
         "delete": "deleted_at",
-        "rotate": "checksum",
+        # These operations create new documents/versions no longer altering
-        "delete_pages": "checksum",
+        # fields on the selected document in place
+        "rotate": None,
+        "delete_pages": None,
         "split": None,
         "merge": None,
-        "edit_pdf": "checksum",
+        "edit_pdf": None,
         "reprocess": "checksum",
-        "remove_password": "checksum",
+        "remove_password": None,
     }
 
     permission_classes = (IsAuthenticated,)
@@ -1696,6 +2040,8 @@ class BulkEditView(PassUserMixin):
         if method in [
             bulk_edit.split,
             bulk_edit.merge,
+            bulk_edit.rotate,
+            bulk_edit.delete_pages,
             bulk_edit.edit_pdf,
             bulk_edit.remove_password,
         ]:
@@ -3181,7 +3527,7 @@ class CustomFieldViewSet(ModelViewSet):
 
     queryset = CustomField.objects.all().order_by("-created")
 
-    def get_queryset(self):
+    def get_queryset(self) -> QuerySet[CustomField]:
         filter = (
             Q(fields__document__deleted_at__isnull=True)
             if self.request.user is None or self.request.user.is_superuser
@@ -3494,11 +3840,16 @@ class TrashView(ListModelMixin, PassUserMixin):
 
     queryset = Document.deleted_objects.all()
 
-    def get(self, request, format=None):
+    def get(self, request: Request, format: str | None = None) -> Response:
         self.serializer_class = DocumentSerializer
         return self.list(request, format)
 
-    def post(self, request, *args, **kwargs):
+    def post(
+        self,
+        request: Request,
+        *args: Any,
+        **kwargs: Any,
+    ) -> Response | HttpResponse:
         serializer = self.get_serializer(data=request.data)
         serializer.is_valid(raise_exception=True)
 
@@ -3522,7 +3873,7 @@ class TrashView(ListModelMixin, PassUserMixin):
         return Response({"result": "OK", "doc_ids": doc_ids})
 
 
-def serve_logo(request, filename=None):
+def serve_logo(request: HttpRequest, filename: str | None = None) -> FileResponse:
     """
     Serves the configured logo file with Content-Disposition: attachment.
     Prevents inline execution of SVGs. See GHSA-6p53-hqqw-8j62