Fix: index document versions as their root document

The search index and the LLM index hold root documents only, a root being
indexed with its newest version's content. Every write path therefore had to
remember to hand them the root. Several did not: adding or deleting a note on
a version, restoring a trashed document together with its versions, and
reprocessing a version all wrote the version into the index under its own id,
where it could be returned as a separate search result. The paths that walk the
whole library did the same: the document_index reindex command put every
version in the search index, a full LLM index rebuild embedded every version,
and an incremental LLM update scoped to a version id indexed it under its own
id. Asking for documents like a version looked the version's own id up in the
index and silently found nothing.

WriteBatch.add_or_update, WriteBatch.add_or_update_ids and
llm_index_add_or_update_document now resolve a version to its root themselves.
The reindex command and update_llm_index only walk root documents, and an
incremental LLM update is scoped to the roots of the given ids through
versioning.root_document_ids, which add_or_update_ids shares. more_like_id
returns the root's id for a version. The reprocess task no longer picks the root
for the indexes and only still clears the caches of both documents.

.mypy-baseline.txt

@@ -38,6 +38,7 @@ src/documents/bulk_edit.py:0: error: Incompatible types in assignment (expressio
 src/documents/bulk_edit.py:0: error: Invalid index type "str" for "dict[FieldDataType, str]"; expected type "FieldDataType"  [index]
 src/documents/bulk_edit.py:0: error: List comprehension has incompatible type List[tuple[int, Any]]; expected List[int]  [misc]
 src/documents/bulk_edit.py:0: error: List comprehension has incompatible type List[tuple[int, None]]; expected List[int]  [misc]
+src/documents/bulk_edit.py:0: error: Missing named argument "p" for "remove" of "PageList"  [call-arg]
 src/documents/bulk_edit.py:0: error: Missing type arguments for generic type "dict"  [type-arg]
 src/documents/bulk_edit.py:0: error: Missing type arguments for generic type "dict"  [type-arg]
 src/documents/bulk_edit.py:0: error: Need type annotation for "to_create" (hint: "to_create: list[<type>] = ...")  [var-annotated]

.pyrefly-baseline.json

@@ -91,6 +91,13 @@
       "concise_description": "Argument `list[int]` is not assignable to parameter `args` with type `tuple[Any, ...] | None` in function `celery.app.task.Task.apply_async`",
       "severity": "error"
     },
+    {
+      "column": 33,
+      "path": "src/documents/bulk_edit.py",
+      "name": "missing-argument",
+      "concise_description": "Missing argument `p` in function `pikepdf._core.PageList.remove`",
+      "severity": "error"
+    },
     {
       "column": 25,
       "path": "src/documents/caching.py",

src/documents/bulk_edit.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import logging
 import tempfile
 import uuid
-from functools import partial
 from pathlib import Path
 from typing import TYPE_CHECKING
 from typing import Literal
@@ -14,11 +13,16 @@ from celery import group
 from celery import shared_task
 from django.conf import settings
 from django.db import transaction
+from django.db.models import Case
+from django.db.models import F
 from django.db.models import Max
+from django.db.models import OuterRef
 from django.db.models import Q
+from django.db.models import Subquery
+from django.db.models import When
+from django.db.models.functions import Coalesce
 from django.utils import timezone
 
-from documents import pdf_ops
 from documents.data_models import ConsumableDocument
 from documents.data_models import DocumentMetadataOverrides
 from documents.data_models import DocumentSource
@@ -38,6 +42,7 @@ from documents.tasks import remove_document_from_index
 from documents.tasks import update_document_content_maybe_archive_file
 from documents.versioning import get_latest_version_for_root
 from documents.versioning import get_root_document
+from documents.versioning import versions_newest_first
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
@@ -118,11 +123,6 @@ def _resolve_root_and_source_doc(
     )
 
 
-def _scratch_path(name: str) -> Path:
-    """A path inside a fresh directory under SCRATCH_DIR."""
-    return Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR)) / name
-
-
 def set_correspondent(
     doc_ids: list[int],
     correspondent: Correspondent,
@@ -415,10 +415,28 @@ def reprocess(doc_ids: list[int], *, remote_ocr: bool = False) -> Literal["OK"]:
 
     Consumption workflows do not run here, so ``remote_ocr`` is how the user
     asks for the remote engine when it is not configured to handle everything.
+
+    A root document with versions reprocesses its latest version, which is the
+    file whose content, archive and thumbnail are shown for it.
     """
-    for document_id in doc_ids:
+    latest_version = versions_newest_first(
+        Document.objects.filter(root_document=OuterRef("pk")),
+    ).values("id")[:1]
+    source_ids = (
+        Document.objects.filter(id__in=doc_ids)
+        .annotate(
+            source_id=Case(
+                When(root_document__isnull=False, then=F("id")),
+                default=Coalesce(Subquery(latest_version), F("id")),
+            ),
+        )
+        .order_by()
+        .values_list("source_id", flat=True)
+        .distinct()
+    )
+    for source_id in source_ids:
         update_document_content_maybe_archive_file.apply_async(
-            kwargs={"document_id": document_id, "remote_ocr": remote_ocr},
+            kwargs={"document_id": source_id, "remote_ocr": remote_ocr},
             headers={"trigger_source": PaperlessTask.TriggerSource.MANUAL},
         )
 
@@ -481,6 +499,8 @@ def rotate(
         pair = _resolve_root_and_source_doc(doc, source_mode=source_mode)
         docs_by_root_id.setdefault(pair.root_doc.id, pair)
 
+    import pikepdf
+
     for pair in docs_by_root_id.values():
         if pair.source_doc.mime_type != "application/pdf":
             logger.warning(
@@ -493,7 +513,11 @@ def rotate(
                 Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR))
                 / f"{pair.root_doc.id}_rotated.pdf"
             )
-            pdf_ops.rotate_pdf(pair.source_doc.source_path, filepath, degrees)
+            with pikepdf.open(pair.source_doc.source_path) as pdf:
+                for page in pdf.pages:
+                    page.rotate(degrees, relative=True)
+                pdf.remove_unreferenced_resources()
+                pdf.save(filepath)
 
             # Preserve metadata/permissions via overrides; mark as new version
             overrides = DocumentMetadataOverrides().from_document(pair.root_doc)
@@ -536,45 +560,48 @@ def merge(
     qs = Document.objects.select_related("root_document").filter(id__in=doc_ids)
     docs_by_id = {doc.id: doc for doc in qs}
     affected_docs: list[int] = []
-    handoff_asn: int | None = None
-    with pdf_ops.PdfMerger() as merger:
-        # use doc_ids to preserve order
-        for doc_id in doc_ids:
-            doc = docs_by_id.get(doc_id)
-            if doc is None:
-                continue
-            pair = _resolve_root_and_source_doc(doc, source_mode=source_mode)
-            try:
-                # archive_path is None when there is no archive version
-                archive_path = (
-                    pair.source_doc.archive_path
-                    if archive_fallback
-                    and pair.source_doc.mime_type != "application/pdf"
-                    else None
-                )
-                merger.add(
-                    archive_path
-                    if archive_path is not None
-                    else pair.source_doc.source_path,
-                )
-                affected_docs.append(doc.id)
-                if handoff_asn is None and doc.archive_serial_number is not None:
-                    handoff_asn = doc.archive_serial_number
-            except Exception as e:
-                logger.exception(
-                    f"Error merging document {doc.id}, it will not be included in the merge: {e}",
-                )
-        if len(affected_docs) == 0:
-            logger.warning("No documents were merged")
-            return "OK"
+    import pikepdf
 
-        filepath = (
-            Path(
-                tempfile.mkdtemp(dir=settings.SCRATCH_DIR),
+    merged_pdf = pikepdf.new()
+    version: str = merged_pdf.pdf_version
+    handoff_asn: int | None = None
+    # use doc_ids to preserve order
+    for doc_id in doc_ids:
+        doc = docs_by_id.get(doc_id)
+        if doc is None:
+            continue
+        pair = _resolve_root_and_source_doc(doc, source_mode=source_mode)
+        try:
+            doc_path = (
+                pair.source_doc.archive_path
+                if archive_fallback
+                and pair.source_doc.mime_type != "application/pdf"
+                and pair.source_doc.has_archive_version
+                else pair.source_doc.source_path
             )
-            / f"{'_'.join([str(doc_id) for doc_id in affected_docs])[:100]}_merged.pdf"
+            with pikepdf.open(str(doc_path)) as pdf:
+                version = max(version, pdf.pdf_version)
+                merged_pdf.pages.extend(pdf.pages)
+            affected_docs.append(doc.id)
+            if handoff_asn is None and doc.archive_serial_number is not None:
+                handoff_asn = doc.archive_serial_number
+        except Exception as e:
+            logger.exception(
+                f"Error merging document {doc.id}, it will not be included in the merge: {e}",
+            )
+    if len(affected_docs) == 0:
+        logger.warning("No documents were merged")
+        return "OK"
+
+    filepath = (
+        Path(
+            tempfile.mkdtemp(dir=settings.SCRATCH_DIR),
         )
-        merger.save(filepath)
+        / f"{'_'.join([str(doc_id) for doc_id in affected_docs])[:100]}_merged.pdf"
+    )
+    merged_pdf.remove_unreferenced_resources()
+    merged_pdf.save(filepath, min_version=version)
+    merged_pdf.close()
 
     if metadata_document_id:
         metadata_document = qs.get(id=metadata_document_id)
@@ -750,60 +777,64 @@ def split(
     )
     doc = Document.objects.select_related("root_document").get(id=doc_ids[0])
     pair = _resolve_root_and_source_doc(doc, source_mode=source_mode)
+    import pikepdf
 
     consume_tasks = []
 
     try:
-        outputs = [
-            (
-                [pdf_ops.PageSpec(page) for page in split_doc],
-                partial(_scratch_path, f"{doc.id}_{split_doc[0]}-{split_doc[-1]}.pdf"),
-            )
-            for split_doc in pages
-        ]
-        filepaths = pdf_ops.build_pdfs(pair.source_doc.source_path, outputs)
+        with pikepdf.open(pair.source_doc.source_path) as pdf:
+            for idx, split_doc in enumerate(pages):
+                dst: pikepdf.Pdf = pikepdf.new()
+                for page in split_doc:
+                    dst.pages.append(pdf.pages[page - 1])
+                filepath: Path = (
+                    Path(
+                        tempfile.mkdtemp(dir=settings.SCRATCH_DIR),
+                    )
+                    / f"{doc.id}_{split_doc[0]}-{split_doc[-1]}.pdf"
+                )
+                dst.remove_unreferenced_resources()
+                dst.save(filepath)
+                dst.close()
 
-        for idx, (split_doc, filepath) in enumerate(
-            zip(pages, filepaths, strict=True),
-        ):
-            overrides: DocumentMetadataOverrides = (
-                DocumentMetadataOverrides().from_document(doc)
-            )
-            overrides.title = f"{doc.title} (split {idx + 1})"
-            if user is not None:
-                overrides.owner_id = user.id
-            if not delete_originals:
-                overrides.skip_asn_if_exists = True
-            logger.info(
-                f"Adding split document with pages {split_doc} to the task queue.",
-            )
-            consume_tasks.append(
-                consume_file.s(
-                    input_doc=ConsumableDocument(
-                        source=DocumentSource.ConsumeFolder,
-                        original_file=filepath,
-                    ),
-                    overrides=overrides,
-                ).set(headers={"trigger_source": trigger_source}),
-            )
+                overrides: DocumentMetadataOverrides = (
+                    DocumentMetadataOverrides().from_document(doc)
+                )
+                overrides.title = f"{doc.title} (split {idx + 1})"
+                if user is not None:
+                    overrides.owner_id = user.id
+                if not delete_originals:
+                    overrides.skip_asn_if_exists = True
+                logger.info(
+                    f"Adding split document with pages {split_doc} to the task queue.",
+                )
+                consume_tasks.append(
+                    consume_file.s(
+                        input_doc=ConsumableDocument(
+                            source=DocumentSource.ConsumeFolder,
+                            original_file=filepath,
+                        ),
+                        overrides=overrides,
+                    ).set(headers={"trigger_source": trigger_source}),
+                )
 
-        if delete_originals:
-            backup = release_archive_serial_numbers([doc.id])
-            logger.info(
-                "Queueing removal of original document after consumption of the split documents",
-            )
-            try:
-                chord(
-                    header=consume_tasks,
-                    body=delete.si([doc.id]),
-                ).on_error(
-                    restore_archive_serial_numbers_task.s(backup),
-                ).apply_async()
-            except Exception:
-                restore_archive_serial_numbers(backup)
-                raise
-        else:
-            group(consume_tasks).delay()
+            if delete_originals:
+                backup = release_archive_serial_numbers([doc.id])
+                logger.info(
+                    "Queueing removal of original document after consumption of the split documents",
+                )
+                try:
+                    chord(
+                        header=consume_tasks,
+                        body=delete.si([doc.id]),
+                    ).on_error(
+                        restore_archive_serial_numbers_task.s(backup),
+                    ).apply_async()
+                except Exception:
+                    restore_archive_serial_numbers(backup)
+                    raise
+            else:
+                group(consume_tasks).delay()
 
     except Exception as e:
         logger.exception(f"Error splitting document {doc.id}: {e}")
@@ -824,6 +855,8 @@ def delete_pages(
     )
     doc = Document.objects.select_related("root_document").get(id=doc_ids[0])
     pair = _resolve_root_and_source_doc(doc, source_mode=source_mode)
+    pages = sorted(pages)  # sort pages to avoid index issues
+    import pikepdf
 
     try:
         # Produce edited PDF to a temp file and create a new version
@@ -831,7 +864,13 @@ def delete_pages(
             Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR))
             / f"{pair.root_doc.id}_pages_deleted.pdf"
         )
-        pdf_ops.remove_pages(pair.source_doc.source_path, filepath, pages)
+        with pikepdf.open(pair.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(filepath)
 
         overrides = DocumentMetadataOverrides().from_document(pair.root_doc)
         if user is not None:
@@ -880,28 +919,47 @@ def edit_pdf(
     )
     doc = Document.objects.select_related("root_document").get(id=doc_ids[0])
     pair = _resolve_root_and_source_doc(doc, source_mode=source_mode)
+    import pikepdf
+
+    pdf_docs: list[pikepdf.Pdf] = []
+
     try:
-        output_count = pdf_ops.validate_page_operations(
-            operations,
-            single_output=update_document,
-        )
-        page_specs: list[list[pdf_ops.PageSpec]] = [[] for _ in range(output_count)]
-        for op in operations:
-            page_specs[op.get("doc", 0)].append(
-                pdf_ops.PageSpec(op["page"], op.get("rotate", 0)),
+        if not operations:
+            raise ValueError("Output document index is out of bounds")
+
+        max_idx = max(op.get("doc", 0) for op in operations)
+        if update_document and max_idx > 0:
+            logger.error(
+                "Update requested but multiple output documents specified",
             )
+            raise ValueError("Multiple output documents specified")
+
+        if any(
+            op.get("doc", 0) < 0 or op.get("doc", 0) >= len(operations)
+            for op in operations
+        ):
+            raise ValueError("Output document index is out of bounds")
+
+        with pikepdf.open(pair.source_doc.source_path) as src:
+            # prepare output documents
+            pdf_docs = [pikepdf.new() for _ in range(max_idx + 1)]
+
+            for op in operations:
+                dst = pdf_docs[op.get("doc", 0)]
+                page = src.pages[op["page"] - 1]
+                dst.pages.append(page)
+                if op.get("rotate"):
+                    dst.pages[-1].rotate(op["rotate"], relative=True)
 
         if update_document:
             # Create a new version from the edited PDF rather than replacing in-place
-            (filepath,) = pdf_ops.build_pdfs(
-                pair.source_doc.source_path,
-                [
-                    (
-                        page_specs[0],
-                        partial(_scratch_path, f"{pair.root_doc.id}_edited.pdf"),
-                    ),
-                ],
+            pdf = pdf_docs[0]
+            pdf.remove_unreferenced_resources()
+            filepath: Path = (
+                Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR))
+                / f"{pair.root_doc.id}_edited.pdf"
             )
+            pdf.save(filepath)
             overrides = (
                 DocumentMetadataOverrides().from_document(pair.root_doc)
                 if include_metadata
@@ -922,19 +980,6 @@ def edit_pdf(
                 headers={"trigger_source": trigger_source},
             )
         else:
-            version_filepaths = pdf_ops.build_pdfs(
-                pair.source_doc.source_path,
-                [
-                    (
-                        specs,
-                        partial(
-                            _scratch_path,
-                            f"{pair.root_doc.id}_edit_{idx}.pdf",
-                        ),
-                    )
-                    for idx, specs in enumerate(page_specs, start=1)
-                ],
-            )
             consume_tasks = []
             overrides = (
                 DocumentMetadataOverrides().from_document(pair.root_doc)
@@ -946,9 +991,15 @@ def edit_pdf(
                 overrides.actor_id = user.id
             if not delete_original:
                 overrides.skip_asn_if_exists = True
-            if delete_original and output_count == 1:
+            if delete_original and len(pdf_docs) == 1:
                 overrides.asn = pair.root_doc.archive_serial_number
-            for version_filepath in version_filepaths:
+            for idx, pdf in enumerate(pdf_docs, start=1):
+                version_filepath: Path = (
+                    Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR))
+                    / f"{pair.root_doc.id}_edit_{idx}.pdf"
+                )
+                pdf.remove_unreferenced_resources()
+                pdf.save(version_filepath)
                 consume_tasks.append(
                     consume_file.s(
                         input_doc=ConsumableDocument(
@@ -998,6 +1049,8 @@ def remove_password(
     """
     Remove password protection from PDF documents.
     """
+    import pikepdf
+
     for doc_id in doc_ids:
         doc = Document.objects.select_related("root_document").get(id=doc_id)
         pair = _resolve_root_and_source_doc(doc, source_mode=source_mode)
@@ -1011,69 +1064,76 @@ def remove_password(
                 doc.id,
                 pair.source_doc.source_path,
             )
-            if not pdf_ops.needs_decrypt(source_path):
-                logger.info(
-                    "Skipping password removal for document %s because the "
-                    "source PDF is not encrypted",
-                    pair.root_doc.id,
-                )
-                continue
+            try:
+                with pikepdf.open(source_path) as pdf:
+                    if not pdf.is_encrypted:
+                        logger.info(
+                            "Skipping password removal for document %s because the "
+                            "source PDF is not encrypted",
+                            pair.root_doc.id,
+                        )
+                        continue
+            except pikepdf.PasswordError:
+                # Password-protected PDFs need the supplied password below.
+                pass
 
-            filepath = pdf_ops.decrypt_pdf(
-                source_path,
-                partial(_scratch_path, f"{pair.root_doc.id}_unprotected.pdf"),
-                password,
-            )
+            with pikepdf.open(source_path, password=password) as pdf:
+                filepath: Path = (
+                    Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR))
+                    / f"{pair.root_doc.id}_unprotected.pdf"
+                )
+                pdf.remove_unreferenced_resources()
+                pdf.save(filepath)
 
-            if update_document:
-                # Create a new version rather than modifying the root/original in place.
-                overrides = (
-                    DocumentMetadataOverrides().from_document(pair.root_doc)
-                    if include_metadata
-                    else DocumentMetadataOverrides()
-                )
-                if user is not None:
-                    overrides.owner_id = user.id
-                    overrides.actor_id = user.id
-                consume_file.apply_async(
-                    kwargs={
-                        "input_doc": ConsumableDocument(
-                            source=DocumentSource.ConsumeFolder,
-                            original_file=filepath,
-                            root_document_id=pair.root_doc.id,
-                        ),
-                        "overrides": overrides,
-                    },
-                    headers={"trigger_source": trigger_source},
-                )
-            else:
-                consume_tasks = []
-                overrides = (
-                    DocumentMetadataOverrides().from_document(pair.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(
-                        input_doc=ConsumableDocument(
-                            source=DocumentSource.ConsumeFolder,
-                            original_file=filepath,
-                        ),
-                        overrides=overrides,
-                    ).set(headers={"trigger_source": trigger_source}),
-                )
-
-                if delete_original:
-                    chord(
-                        header=consume_tasks,
-                        body=delete.si([doc.id]),
-                    ).delay()
+                if update_document:
+                    # Create a new version rather than modifying the root/original in place.
+                    overrides = (
+                        DocumentMetadataOverrides().from_document(pair.root_doc)
+                        if include_metadata
+                        else DocumentMetadataOverrides()
+                    )
+                    if user is not None:
+                        overrides.owner_id = user.id
+                        overrides.actor_id = user.id
+                    consume_file.apply_async(
+                        kwargs={
+                            "input_doc": ConsumableDocument(
+                                source=DocumentSource.ConsumeFolder,
+                                original_file=filepath,
+                                root_document_id=pair.root_doc.id,
+                            ),
+                            "overrides": overrides,
+                        },
+                        headers={"trigger_source": trigger_source},
+                    )
                 else:
-                    group(consume_tasks).delay()
+                    consume_tasks = []
+                    overrides = (
+                        DocumentMetadataOverrides().from_document(pair.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(
+                            input_doc=ConsumableDocument(
+                                source=DocumentSource.ConsumeFolder,
+                                original_file=filepath,
+                            ),
+                            overrides=overrides,
+                        ).set(headers={"trigger_source": trigger_source}),
+                    )
+
+                    if delete_original:
+                        chord(
+                            header=consume_tasks,
+                            body=delete.si([doc.id]),
+                        ).delay()
+                    else:
+                        group(consume_tasks).delay()
 
         except Exception as e:
             logger.exception(

src/documents/management/commands/document_index.py

@@ -67,18 +67,22 @@ class Command(PaperlessCommand):
                 if options.get("recreate"):
                     wipe_index(settings.INDEX_DIR)
 
-                documents = Document.objects.select_related(
-                    "correspondent",
-                    "document_type",
-                    "storage_path",
-                    "owner",
-                ).prefetch_related(
-                    "tags",
-                    "notes__user",
-                    "custom_fields__field",
-                    "versions",
-                    "barcodes",
-                    "versions__barcodes",
+                documents = (
+                    Document.objects.filter(root_document__isnull=True)
+                    .select_related(
+                        "correspondent",
+                        "document_type",
+                        "storage_path",
+                        "owner",
+                    )
+                    .prefetch_related(
+                        "tags",
+                        "notes__user",
+                        "custom_fields__field",
+                        "versions",
+                        "barcodes",
+                        "versions__barcodes",
+                    )
                 )
                 total = documents.count()
                 rebuild_kwargs = {}

src/documents/pdf_ops.py

@@ -1,194 +0,0 @@
-"""
-Pure PDF page operations used by documents.bulk_edit.
-
-This module deliberately knows nothing about Django, Celery or the documents
-app: callers resolve documents, choose output paths and queue work. Every
-function that writes a PDF removes unreferenced resources before saving.
-
-pikepdf is always called as ``pikepdf.open(...)`` / ``pikepdf.new()`` (never
-``from pikepdf import open``) so tests can patch those module attributes.
-"""
-
-from __future__ import annotations
-
-from typing import TYPE_CHECKING
-from typing import NamedTuple
-
-import pikepdf
-
-if TYPE_CHECKING:
-    from collections.abc import Callable
-    from collections.abc import Iterable
-    from collections.abc import Mapping
-    from collections.abc import Sequence
-    from pathlib import Path
-    from types import TracebackType
-
-
-class PageSpec(NamedTuple):
-    """One page of an output PDF: a 1-indexed source page, optionally rotated."""
-
-    page: int
-    rotate: int = 0  # relative degrees, 0 leaves the page alone
-
-
-def _require_positive(pages: Iterable[int]) -> None:
-    for page in pages:
-        if page < 1:
-            raise ValueError(f"Page numbers start at 1, got {page}")
-
-
-def rotate_pdf(src: Path, dst: Path, degrees: int) -> None:
-    """
-    Rotate every page relatively on the opened document, not a rebuild, so Info,
-    XMP and outlines are kept. ``src`` is not modified.
-    """
-    with pikepdf.open(src) as pdf:
-        for page in pdf.pages:
-            page.rotate(degrees, relative=True)
-        pdf.remove_unreferenced_resources()
-        pdf.save(dst)
-
-
-def remove_pages(src: Path, dst: Path, pages: Iterable[int]) -> None:
-    """
-    Remove 1-indexed pages from the opened document, not a rebuild, so Info, XMP
-    and outlines are kept. ``src`` is not modified.
-
-    Duplicates are ignored. Pages are removed highest first so earlier removals
-    never shift the index of later ones.
-    """
-    unique = sorted(set(pages))
-    _require_positive(unique)
-    with pikepdf.open(src) as pdf:
-        for page_num in reversed(unique):
-            del pdf.pages[page_num - 1]
-        pdf.remove_unreferenced_resources()
-        pdf.save(dst)
-
-
-def build_pdfs(
-    src: Path,
-    outputs: Sequence[tuple[Sequence[PageSpec], Callable[[], Path]]],
-) -> list[Path]:
-    """
-    Build one new PDF per output from pages of ``src``, opening ``src`` once.
-
-    Each output is ``(page_specs, make_dst)``. Every page number is checked against
-    ``src`` before any output is built, and ``make_dst`` is called after that
-    output's pages are copied and immediately before it is saved, so a bad page
-    number in any output never leaves a destination behind. Document-level data
-    (Info, XMP, outlines) is not carried over. Returns the written paths in output
-    order.
-    """
-    for specs, _ in outputs:
-        _require_positive(spec.page for spec in specs)
-
-    written: list[Path] = []
-    with pikepdf.open(src) as source:
-        page_count = len(source.pages)
-        for specs, _ in outputs:
-            for spec in specs:
-                if spec.page > page_count:
-                    raise IndexError(
-                        f"Page {spec.page} is out of range, the PDF has "
-                        f"{page_count} pages",
-                    )
-        for specs, make_dst in outputs:
-            dst = pikepdf.new()
-            for spec in specs:
-                dst.pages.append(source.pages[spec.page - 1])
-                if spec.rotate:
-                    dst.pages[-1].rotate(spec.rotate, relative=True)
-            dst.remove_unreferenced_resources()
-            path = make_dst()
-            dst.save(path)
-            dst.close()
-            written.append(path)
-    return written
-
-
-def validate_page_operations(
-    operations: Sequence[Mapping[str, int]],
-    *,
-    single_output: bool,
-) -> int:
-    """
-    Validate ``edit_pdf`` style operations and return the output document count.
-
-    Each operation has ``page`` and optionally ``rotate`` and ``doc`` (the output
-    document index, default 0). The bounds rule is kept as it was: a ``doc`` index
-    must be below the number of operations.
-    """
-    if not operations:
-        raise ValueError("Output document index is out of bounds")
-
-    max_idx = max(op.get("doc", 0) for op in operations)
-    if single_output and max_idx > 0:
-        raise ValueError("Multiple output documents specified")
-
-    if any(
-        op.get("doc", 0) < 0 or op.get("doc", 0) >= len(operations) for op in operations
-    ):
-        raise ValueError("Output document index is out of bounds")
-
-    return max_idx + 1
-
-
-def needs_decrypt(src: Path) -> bool:
-    """
-    True if ``src`` is encrypted. A PDF that needs a password to open at all
-    counts as encrypted.
-    """
-    try:
-        with pikepdf.open(src) as pdf:
-            return bool(pdf.is_encrypted)
-    except pikepdf.PasswordError:
-        return True
-
-
-def decrypt_pdf(src: Path, make_dst: Callable[[], Path], password: str) -> Path:
-    """
-    Write an unencrypted copy of ``src`` and return its path.
-
-    ``make_dst`` is only called once the password has been accepted, so a wrong
-    password never leaves a destination behind.
-    """
-    with pikepdf.open(src, password=password) as pdf:
-        pdf.remove_unreferenced_resources()
-        dst = make_dst()
-        pdf.save(dst)
-    return dst
-
-
-class PdfMerger:
-    """
-    Accumulates the pages of several PDFs into one new PDF.
-
-    ``add`` raises if a source cannot be read; deciding whether to skip it is the
-    caller's policy. Use as a context manager so the merged PDF is closed.
-    """
-
-    def __init__(self) -> None:
-        self._pdf = pikepdf.new()
-        self._version: str = self._pdf.pdf_version
-
-    def __enter__(self) -> PdfMerger:
-        return self
-
-    def __exit__(
-        self,
-        exc_type: type[BaseException] | None,
-        exc: BaseException | None,
-        tb: TracebackType | None,
-    ) -> None:
-        self._pdf.close()
-
-    def add(self, path: Path) -> None:
-        with pikepdf.open(str(path)) as pdf:
-            self._version = max(self._version, pdf.pdf_version)
-            self._pdf.pages.extend(pdf.pages)
-
-    def save(self, dst: Path) -> None:
-        self._pdf.remove_unreferenced_resources()
-        self._pdf.save(dst, min_version=self._version)

src/documents/permissions.py

@@ -6,14 +6,19 @@ from django.contrib.auth.models import Permission
 from django.contrib.auth.models import User
 from django.contrib.contenttypes.models import ContentType
 from django.db.models import Case
+from django.db.models import CharField
 from django.db.models import Count
+from django.db.models import Exists
+from django.db.models import F
 from django.db.models import IntegerField
 from django.db.models import Model
+from django.db.models import OuterRef
 from django.db.models import Q
 from django.db.models import QuerySet
 from django.db.models import Value
 from django.db.models import When
 from django.db.models.functions import Cast
+from django.db.models.functions import Coalesce
 from guardian.core import ObjectPermissionChecker
 from guardian.models import GroupObjectPermission
 from guardian.models import UserObjectPermission
@@ -25,6 +30,7 @@ from rest_framework.permissions import BasePermission
 from rest_framework.permissions import DjangoObjectPermissions
 
 from documents.models import Document
+from documents.versioning import get_root_document
 
 
 class PaperlessObjectPermissions(DjangoObjectPermissions):
@@ -348,6 +354,7 @@ def permitted_object_ids(
     perm: str,
     *,
     include_deleted: bool = False,
+    parent_field: str | None = None,
 ) -> QuerySet[int]:
     """
     Generic version of ``permitted_document_ids`` for any model with an
@@ -356,6 +363,20 @@ def permitted_object_ids(
     soft-delete pattern (currently only ``Document``); for every other model
     it is accepted but has no effect, since those models have no soft-delete
     concept.
+
+    ``parent_field`` names a self-referencing foreign key whose target
+    authorizes the row (``Document.root_document``). A row with a parent is
+    visible exactly when its parent is, judged by the parent's owner and
+    grants, so the row's own owner and grants are ignored.
+
+    Guardian stores ``object_pk`` as a string, so each grant is an ``EXISTS``
+    keyed on the row's id cast to a string, which can use guardian's
+    (user, permission, object_pk) unique index. Casting every ``object_pk`` to
+    an integer for an ``id IN (...)`` is not indexable, and MariaDB cannot
+    materialize it inside the owner ``OR``, so it re-scans the user's grants
+    for every row. The user's groups are matched with an ``IN`` subquery
+    rather than a join through the membership table, which SQLite plans badly
+    once the grant tables grow.
     """
     has_soft_delete = hasattr(model, "global_objects")
     manager = (
@@ -363,8 +384,21 @@ def permitted_object_ids(
     )
     base_qs = manager.all().only("id", "owner")
 
+    owner_field, key_field = "owner", "pk"
+    if parent_field is not None:
+        owner_field, key_field = "authorizing_owner", "authorizing_id"
+        base_qs = base_qs.annotate(
+            authorizing_id=Coalesce(f"{parent_field}_id", "id"),
+            authorizing_owner=Case(
+                When(**{f"{parent_field}_id__isnull": True}, then=F("owner_id")),
+                default=F(f"{parent_field}__owner_id"),
+                output_field=IntegerField(),
+            ),
+        )
+    unowned = Q(**{f"{owner_field}__isnull": True})
+
     if user is None or not getattr(user, "is_authenticated", False):
-        return base_qs.filter(owner__isnull=True).values_list("id", flat=True)
+        return base_qs.filter(unowned).values_list("id", flat=True)
 
     # Deactivated users get nothing, deactivated superusers included, so this
     # has to come before the superuser shortcut. guardian's
@@ -388,20 +422,24 @@ def permitted_object_ids(
         "permission__content_type": content_type,
     }
 
-    user_perm_ids = (
-        UserObjectPermission.objects.filter(user=user, **perm_filter)
-        .annotate(object_pk_int=Cast("object_pk", IntegerField()))
-        .values_list("object_pk_int", flat=True)
+    key_as_text = Cast(OuterRef(key_field), CharField(max_length=64))
+    granted_to_user = Exists(
+        UserObjectPermission.objects.filter(
+            user=user,
+            object_pk=key_as_text,
+            **perm_filter,
+        ),
     )
-    group_perm_ids = (
-        GroupObjectPermission.objects.filter(group__user=user, **perm_filter)
-        .annotate(object_pk_int=Cast("object_pk", IntegerField()))
-        .values_list("object_pk_int", flat=True)
+    granted_to_group = Exists(
+        GroupObjectPermission.objects.filter(
+            group_id__in=user.groups.values("id"),
+            object_pk=key_as_text,
+            **perm_filter,
+        ),
     )
-    permitted_ids = user_perm_ids.union(group_perm_ids)
 
     return base_qs.filter(
-        Q(owner=user) | Q(owner__isnull=True) | Q(id__in=permitted_ids),
+        Q(**{owner_field: user.pk}) | unowned | granted_to_user | granted_to_group,
     ).values_list("id", flat=True)
 
 
@@ -470,8 +508,17 @@ def permitted_document_ids(
     ``include_deleted=True`` for callers that need to check permission on
     soft-deleted documents (e.g. trash restore). This intentionally avoids
     ``get_objects_for_user`` to keep the subquery small and index-friendly.
+
+    A version is authorized by its root document, so a version's own owner and
+    grants never matter.
     """
-    return permitted_object_ids(user, Document, perm, include_deleted=include_deleted)
+    return permitted_object_ids(
+        user,
+        Document,
+        perm,
+        include_deleted=include_deleted,
+        parent_field="root_document",
+    )
 
 
 def get_document_count_filter_for_user(user, related_name: str = "documents"):
@@ -630,7 +677,14 @@ def has_perms_owner_aware(user, perms, obj):
     single-object check still has many production callers. Several callers
     remain across ``documents/``, ``paperless_mail/``, and ``paperless_ai/``
     -- grep for this function name before removing it.
+
+    A document version is authorized by its root document, like in
+    ``permitted_document_ids``, so a version's own owner and grants never
+    matter. Fetch the root with ``select_related("root_document__owner")`` to
+    avoid extra queries.
     """
+    if isinstance(obj, Document):
+        obj = get_root_document(obj)
     checker = ObjectPermissionChecker(user)
     return obj.owner is None or obj.owner == user or checker.has_perm(perms, obj)
 

src/documents/search/_backend.py

@@ -7,6 +7,7 @@ import threading
 import time
 from datetime import UTC
 from datetime import datetime
+from datetime import timedelta
 from enum import StrEnum
 from itertools import islice
 from typing import TYPE_CHECKING
@@ -31,6 +32,7 @@ from documents.search._query import parse_user_query
 from documents.search._schema import _write_sentinels
 from documents.search._schema import build_schema
 from documents.search._schema import open_or_rebuild_index
+from documents.search._schema import rebuild_in_progress
 from documents.search._schema import wipe_index
 from documents.search._tokenizer import ascii_fold
 from documents.search._tokenizer import autocomplete_tokens
@@ -54,6 +56,19 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger("paperless.search")
 
+# tantivy stores dates as signed 64-bit nanoseconds since the Unix epoch, which
+# covers 1677-09-21T00:12:43 to 2262-04-11T23:47:16 UTC
+_INDEX_DATE_NANOS_MIN: Final[int] = -(2**63)
+_INDEX_DATE_NANOS_MAX: Final[int] = 2**63 - 1
+_UNIX_EPOCH: Final[datetime] = datetime(1970, 1, 1, tzinfo=UTC)
+
+
+def _is_indexable_date(value: datetime) -> bool:
+    """Whether value, at whole-second precision, fits tantivy's date range."""
+    nanos = ((value - _UNIX_EPOCH) // timedelta(seconds=1)) * 1_000_000_000
+    return _INDEX_DATE_NANOS_MIN <= nanos <= _INDEX_DATE_NANOS_MAX
+
+
 _LOCK_TIMEOUT_SECONDS: Final[float] = 10.0  # per-attempt acquire timeout
 _LOCK_RETRY_ATTEMPTS: Final[int] = 4  # total attempts (1 initial + 3 retries)
 _LOCK_BACKOFF_BASE: Final[float] = 1.0  # seconds
@@ -269,9 +284,14 @@ class WriteBatch:
         and adding the new version. This ensures stale document data (e.g., after
         permission changes) doesn't persist in the index.
 
+        Only root documents are indexed, with their effective content, so a
+        version is indexed as its root document.
+
         Args:
             document: Django Document instance to index
         """
+        if document.root_document_id is not None:
+            document = document.root_document
         self.remove(document.pk)
         doc = self._backend._build_tantivy_doc(document)
         self._writer.add_document(doc)
@@ -296,20 +316,22 @@ class WriteBatch:
         An id with no matching document (e.g. deleted between the caller
         collecting ids and the batch running) is silently skipped, matching
         ``add_or_update()``'s existing single-document deferred-task behavior
-        rather than erroring or leaving a stale index entry.
+        rather than erroring or leaving a stale index entry. The id of a
+        version stands for its root document.
 
         Args:
             ids: Primary keys of Document instances to index
         """
         from documents.models import Document
         from documents.versioning import annotate_effective_content
+        from documents.versioning import root_document_ids
 
         ids = list(ids)
         if not ids:
             return
 
         queryset = annotate_effective_content(
-            Document.objects.filter(pk__in=ids)
+            Document.objects.filter(pk__in=root_document_ids(ids))
             .select_related("correspondent", "document_type", "storage_path", "owner")
             .prefetch_related(
                 "tags",
@@ -627,7 +649,15 @@ class TantivyBackend:
             document.created.day,
             tzinfo=UTC,
         )
-        doc.add_date("created", created_date)
+        if _is_indexable_date(created_date):
+            doc.add_date("created", created_date)
+        else:
+            logger.warning(
+                "Document %s has a created date (%s) outside the range the search "
+                "index can store; it will be indexed without a created date",
+                document.pk,
+                document.created,
+            )
         doc.add_date("modified", document.modified)
         doc.add_date("added", document.added)
 
@@ -1110,39 +1140,44 @@ class TantivyBackend:
                 flushing a segment, deferring merge work; they do not avoid it.
         """
         wipe_index(self._path)
-        new_index = tantivy.Index(build_schema(), path=str(self._path))
-        _write_sentinels(self._path)
-        register_tokenizers(new_index, settings.SEARCH_LANGUAGE)
+        # The marker covers the window where the empty index is already stamped
+        # as current but not yet populated, so an interrupted rebuild is retried.
+        with rebuild_in_progress(self._path):
+            new_index = tantivy.Index(build_schema(), path=str(self._path))
+            _write_sentinels(self._path)
+            register_tokenizers(new_index, settings.SEARCH_LANGUAGE)
 
-        # Point instance at the new index so _build_tantivy_doc uses it
-        old_index, old_schema = self._raw_index, self._raw_schema
-        self._raw_index = new_index
-        self._raw_schema = new_index.schema
-        # Stream documents one-by-one (so the progress bar advances per
-        # document) while fetching viewer permissions one SQL query per chunk.
-        # The stream is Sized, so iter_wrapper can still discover the total.
-        documents_stream = _DocumentViewerStream(documents, chunk_size=1000)
-        try:
-            writer = new_index.writer(heap_size=writer_heap_bytes)
-            for document, (viewer_ids, viewer_group_ids) in iter_wrapper(
-                documents_stream,
-            ):
-                doc = self._build_tantivy_doc(
-                    document,
-                    viewer_ids=viewer_ids,
-                    viewer_group_ids=viewer_group_ids,
-                )
-                writer.add_document(doc)
-            writer.commit()
-            # Wait for background merge threads to finish so all segments are
-            # fully merged and persisted before the index is considered rebuilt.
-            writer.wait_merging_threads()
-            new_index.reload()
-        except BaseException:  # pragma: no cover
-            # Restore old index on failure so the backend remains usable
-            self._raw_index = old_index
-            self._raw_schema = old_schema
-            raise
+            # Point instance at the new index so _build_tantivy_doc uses it
+            old_index, old_schema = self._raw_index, self._raw_schema
+            self._raw_index = new_index
+            self._raw_schema = new_index.schema
+            # Stream documents one-by-one (so the progress bar advances per
+            # document) while fetching viewer permissions one SQL query per
+            # chunk. The stream is Sized, so iter_wrapper can still discover
+            # the total.
+            documents_stream = _DocumentViewerStream(documents, chunk_size=1000)
+            try:
+                writer = new_index.writer(heap_size=writer_heap_bytes)
+                for document, (viewer_ids, viewer_group_ids) in iter_wrapper(
+                    documents_stream,
+                ):
+                    doc = self._build_tantivy_doc(
+                        document,
+                        viewer_ids=viewer_ids,
+                        viewer_group_ids=viewer_group_ids,
+                    )
+                    writer.add_document(doc)
+                writer.commit()
+                # Wait for background merge threads to finish so all segments
+                # are fully merged and persisted before the index is considered
+                # rebuilt.
+                writer.wait_merging_threads()
+                new_index.reload()
+            except BaseException:  # pragma: no cover
+                # Restore old index on failure so the backend remains usable
+                self._raw_index = old_index
+                self._raw_schema = old_schema
+                raise
 
 
 def chunked(iterable, size):

src/documents/search/_schema.py

@@ -4,6 +4,7 @@ import hashlib
 import json
 import logging
 import shutil
+from contextlib import contextmanager
 from typing import TYPE_CHECKING
 from typing import Final
 from typing import NamedTuple
@@ -16,6 +17,7 @@ from whoosh_compat import FieldKind
 from documents.search._fields import PUBLIC_FIELDS
 
 if TYPE_CHECKING:
+    from collections.abc import Iterator
     from pathlib import Path
 
 logger = logging.getLogger("paperless.search")
@@ -28,6 +30,11 @@ logger = logging.getLogger("paperless.search")
 # v3 - barcodes JSON field for stored barcode contents
 SCHEMA_VERSION: Final[int] = 3
 
+# Present in the index directory from the moment a full rebuild starts until it
+# finishes. If a rebuild is interrupted it is left behind, so the half-built
+# index is not mistaken for a complete one.
+REBUILD_MARKER: Final[str] = ".rebuilding"
+
 
 class FieldDescriptor(NamedTuple):
     """One tantivy field, in declaration order.
@@ -255,9 +262,9 @@ def needs_rebuild(index_dir: Path) -> bool:
     """
     Check if the search index needs rebuilding.
 
-    Reads .index_settings.json to compare the stored schema version, search
-    language and schema fingerprint against the current configuration. Returns
-    True if the file is missing, unparsable, or any value mismatches.
+    True if a previous full rebuild never finished (the rebuild marker is still
+    present), or if the index's stamped settings no longer match the current
+    configuration. See _settings_mismatch().
 
     Args:
         index_dir: Path to the search index directory
@@ -265,6 +272,40 @@ def needs_rebuild(index_dir: Path) -> bool:
     Returns:
         True if the index needs rebuilding, False if it's up to date
     """
+    if (index_dir / REBUILD_MARKER).exists():
+        logger.warning("Previous search index rebuild did not finish - rebuilding.")
+        return True
+    return _settings_mismatch(index_dir)
+
+
+@contextmanager
+def rebuild_in_progress(index_dir: Path) -> Iterator[None]:
+    """
+    Flag the index as incomplete for the duration of a full rebuild.
+
+    The marker is cleared only if the block exits cleanly. There is deliberately
+    no try/finally: an exception must leave the marker behind so the next
+    needs_rebuild() check retries the rebuild.
+    """
+    marker = index_dir / REBUILD_MARKER
+    marker.touch()
+    yield
+    marker.unlink(missing_ok=True)
+
+
+def _settings_mismatch(index_dir: Path) -> bool:
+    """
+    Check the stamped settings against the current configuration.
+
+    Reads .index_settings.json to compare the stored schema version, search
+    language and schema fingerprint. Returns True if the file is missing,
+    unparsable, or any value mismatches.
+
+    This deliberately ignores the rebuild marker: open_or_rebuild_index() uses it
+    so that a process opening the index while another process is mid-rebuild
+    (or after one died) does not wipe the partial index out from under it.
+    Repopulating is the job of ``document_index reindex``.
+    """
     settings_file = index_dir / ".index_settings.json"
     if not settings_file.exists():
         return True
@@ -333,7 +374,7 @@ def open_or_rebuild_index(index_dir: Path | None = None) -> tantivy.Index:
         index_dir = cast("Path", settings.INDEX_DIR)
     if not index_dir.exists():
         return tantivy.Index(build_schema())
-    if needs_rebuild(index_dir):
+    if _settings_mismatch(index_dir):
         wipe_index(index_dir)
         idx = tantivy.Index(build_schema(), path=str(index_dir))
         _write_sentinels(index_dir)

src/documents/serialisers.py

@@ -2144,8 +2144,6 @@ class BulkEditSerializer(
             raise serializers.ValidationError("pages must be a list")
         if not all(isinstance(i, int) for i in parameters["pages"]):
             raise serializers.ValidationError("pages must be a list of integers")
-        if any(i < 1 for i in parameters["pages"]):
-            raise serializers.ValidationError("pages must be positive integers")
 
     def _validate_parameters_merge(self, parameters) -> None:
         if "delete_originals" in parameters:

src/documents/tasks.py

@@ -502,6 +502,8 @@ def update_document_content_maybe_archive_file(
                 llm_index_add_or_update_document(document)
 
             clear_document_caches(document.pk)
+            if document.root_document_id is not None:
+                clear_document_caches(document.root_document_id)
 
         except Exception:
             logger.exception(

src/documents/tests/search/test_backend.py

@@ -1,4 +1,6 @@
 import json
+import logging
+from datetime import date
 from pathlib import Path
 
 import pytest
@@ -16,7 +18,10 @@ from documents.search._backend import TantivyBackend
 from documents.search._backend import WriteBatch
 from documents.search._backend import get_backend
 from documents.search._backend import reset_backend
+from documents.search._schema import REBUILD_MARKER
+from documents.search._schema import needs_rebuild
 from documents.signals.handlers import add_to_index
+from paperless_testing.dirs import PaperlessDirs
 from paperless_testing.factories import CorrespondentFactory
 from paperless_testing.factories import DocumentFactory
 from paperless_testing.factories import DocumentTypeFactory
@@ -288,6 +293,80 @@ class TestAddOrUpdateIds:
         assert backend.search_ids("updated", user=None) == [doc.pk]
 
 
+class TestVersionsAreIndexedAsTheirRoot:
+    """Only root documents are indexed, with their effective content, so
+    every write path that is handed a version indexes its root instead."""
+
+    @staticmethod
+    def _root_with_version() -> tuple[Document, Document]:
+        root = DocumentFactory(title="Statement", content="stale text")
+        version = DocumentFactory(
+            title="Statement",
+            content="latest text",
+            root_document=root,
+            version_index=1,
+        )
+        return root, version
+
+    def test_add_or_update_indexes_the_root_of_a_version(
+        self,
+        backend: TantivyBackend,
+    ) -> None:
+        """
+        GIVEN:
+            - A root document with a version
+        WHEN:
+            - The version is passed to add_or_update
+        THEN:
+            - The root is indexed with the version's text, and the version is not
+        """
+        root, version = self._root_with_version()
+
+        backend.add_or_update(version)
+
+        assert backend.search_ids("latest", user=None) == [root.pk]
+        assert backend.search_ids("stale", user=None) == []
+
+    def test_add_or_update_ids_indexes_each_root_once(
+        self,
+        backend: TantivyBackend,
+    ) -> None:
+        """
+        GIVEN:
+            - A root document with a version
+        WHEN:
+            - Both ids are passed to add_or_update_ids
+        THEN:
+            - The root is indexed once and the version is not indexed
+        """
+        root, version = self._root_with_version()
+
+        with backend.batch_update() as batch:
+            batch.add_or_update_ids([version.pk, root.pk])
+
+        assert backend.search_ids("Statement", user=None) == [root.pk]
+        assert backend.search_ids("latest", user=None) == [root.pk]
+
+    def test_add_or_update_ids_resolves_a_lone_version(
+        self,
+        backend: TantivyBackend,
+    ) -> None:
+        """
+        GIVEN:
+            - A root document with a version
+        WHEN:
+            - Only the version's id is passed to add_or_update_ids
+        THEN:
+            - The root is indexed
+        """
+        root, version = self._root_with_version()
+
+        with backend.batch_update() as batch:
+            batch.add_or_update_ids([version.pk])
+
+        assert backend.search_ids("latest", user=None) == [root.pk]
+
+
 class TestSearch:
     """Test search query parsing and matching via search_ids."""
 
@@ -823,6 +902,53 @@ class TestRebuild:
         backend.rebuild(Document.objects.all(), iter_wrapper=wrapper)
         assert 30 in seen
 
+    def test_successful_rebuild_leaves_index_up_to_date(
+        self,
+        backend: TantivyBackend,
+        paperless_dirs: PaperlessDirs,
+    ) -> None:
+        """
+        GIVEN:
+            - A backend and one document
+        WHEN:
+            - rebuild() completes
+        THEN:
+            - needs_rebuild() is False and no rebuild marker remains
+        """
+        DocumentFactory.create()
+
+        backend.rebuild(Document.objects.all())
+
+        assert needs_rebuild(paperless_dirs.index_dir) is False
+        assert not (paperless_dirs.index_dir / REBUILD_MARKER).exists()
+
+    def test_interrupted_rebuild_is_retried(
+        self,
+        backend: TantivyBackend,
+        paperless_dirs: PaperlessDirs,
+    ) -> None:
+        """
+        GIVEN:
+            - A rebuild that dies while indexing documents (e.g. the database
+              connection is lost)
+        WHEN:
+            - needs_rebuild() is checked afterwards
+        THEN:
+            - It is True, even though the empty index was already stamped with
+              current settings, so the next start rebuilds instead of reporting
+              the index as up to date
+        """
+        DocumentFactory.create()
+
+        def die(pairs):
+            raise RuntimeError("terminating connection due to administrator command")
+            yield  # pragma: no cover
+
+        with pytest.raises(RuntimeError):
+            backend.rebuild(Document.objects.all(), iter_wrapper=die)
+
+        assert needs_rebuild(paperless_dirs.index_dir) is True
+
     def test_includes_group_granted_viewers(self, backend: TantivyBackend) -> None:
         """Rebuild must index viewer ids for group-only grants, not just direct ones.
 
@@ -854,6 +980,79 @@ class TestRebuild:
         assert ids == [doc.pk]
 
 
+class TestCreatedDateOutOfRange:
+    """The index stores dates as nanosecond i64 values (1677-09-22 to 2262-04-11).
+
+    A document whose created date falls outside that window must not abort
+    indexing: it is indexed without a created value and a warning names it.
+    """
+
+    @pytest.mark.parametrize(
+        ("created", "expected_warnings"),
+        [
+            pytest.param(date(1677, 9, 22), 0, id="first-representable-day"),
+            pytest.param(date(2262, 4, 11), 0, id="last-representable-day"),
+            pytest.param(date(1677, 9, 21), 1, id="day-before-first"),
+            pytest.param(date(2262, 4, 12), 1, id="day-after-last"),
+            pytest.param(date(16, 8, 30), 1, id="two-digit-year-read-as-year-16"),
+            pytest.param(date(9999, 12, 31), 1, id="max-python-date"),
+        ],
+    )
+    def test_add_or_update_indexes_document_and_warns_when_out_of_range(
+        self,
+        backend: TantivyBackend,
+        caplog: pytest.LogCaptureFixture,
+        created: date,
+        expected_warnings: int,
+    ) -> None:
+        """
+        GIVEN:
+            - A document with a created date at or beyond the index date limits
+        WHEN:
+            - The document is added to the index
+        THEN:
+            - The document is indexed and searchable either way
+            - A warning naming the document is logged only for out-of-range dates
+        """
+        doc = DocumentFactory(created=created, content="boundarycontent")
+
+        with caplog.at_level(logging.WARNING, logger="paperless.search"):
+            backend.add_or_update(doc)
+
+        assert backend.search_ids("boundarycontent", user=None) == [doc.pk]
+        warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
+        assert len(warnings) == expected_warnings
+        if expected_warnings:
+            assert f"Document {doc.pk}" in warnings[0].getMessage()
+
+    def test_rebuild_continues_past_out_of_range_document(
+        self,
+        backend: TantivyBackend,
+        caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        """
+        GIVEN:
+            - A document with an unrepresentable created date among valid ones
+        WHEN:
+            - The index is rebuilt
+        THEN:
+            - Rebuild completes and every document is searchable
+            - A warning names the offending document
+        """
+        good = DocumentFactory(created=date(2016, 8, 30), content="rebuildcontent")
+        bad = DocumentFactory(created=date(16, 8, 30), content="rebuildcontent")
+
+        with caplog.at_level(logging.WARNING, logger="paperless.search"):
+            backend.rebuild(Document.objects.all())
+
+        assert sorted(backend.search_ids("rebuildcontent", user=None)) == sorted(
+            [good.pk, bad.pk],
+        )
+        warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
+        assert len(warnings) == 1
+        assert f"Document {bad.pk}" in warnings[0].getMessage()
+
+
 class TestAutocomplete:
     """Test autocomplete functionality."""
 

src/documents/tests/test_api_bulk_edit.py

@@ -1843,36 +1843,6 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         m.assert_called_once()
         self.assertEqual(m.call_args.kwargs["pages"], [[1], [2, 3, 4], [5]])
 
-    @mock.patch("documents.serialisers.bulk_edit.delete_pages")
-    def test_bulk_edit_delete_pages_rejects_pages_below_one(self, m) -> None:
-        """
-        GIVEN:
-            - A legacy delete_pages bulk edit
-        WHEN:
-            - API to bulk edit is called with a page number below 1
-        THEN:
-            - API returns HTTP 400
-            - delete_pages is not called
-        """
-        self.setup_mock(m, "delete_pages")
-
-        for pages in ([0], [-1], [1, 0]):
-            with self.subTest(pages=pages):
-                response = self.client.post(
-                    "/api/documents/bulk_edit/",
-                    json.dumps(
-                        {
-                            "documents": [self.doc2.id],
-                            "method": "delete_pages",
-                            "parameters": {"pages": pages},
-                        },
-                    ),
-                    content_type="application/json",
-                )
-                self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
-                self.assertIn(b"pages must be positive integers", response.content)
-        m.assert_not_called()
-
     @mock.patch("documents.views.bulk_edit.rotate")
     def test_rotate_insufficient_permissions(self, m) -> None:
         self.doc1.owner = User.objects.get(username="temp_admin")

src/documents/tests/test_api_document_versions.py

@@ -8,6 +8,7 @@ from auditlog.models import LogEntry  # type: ignore[import-untyped]
 from django.contrib.contenttypes.models import ContentType
 from django.core.files.uploadedfile import SimpleUploadedFile
 from django.test import TestCase as DjangoTestCase
+from django.test import override_settings
 from django.utils import timezone
 from rest_framework import status
 from rest_framework.test import APITestCase
@@ -16,13 +17,17 @@ from documents.data_models import DocumentSource
 from documents.filters import EffectiveContentFilter
 from documents.filters import TitleContentFilter
 from documents.models import Document
+from documents.models import Note
+from documents.models import ShareLink
 from documents.versioning import annotate_effective_content
 from documents.views import DocumentSelectionMixin
 from paperless_testing.dirs import DirectoriesMixin
 from paperless_testing.factories import DocumentFactory
 from paperless_testing.factories import UserFactory
 from paperless_testing.http import read_streaming_response
+from paperless_testing.permissions import grant_all_global
 from paperless_testing.permissions import grant_global
+from paperless_testing.permissions import grant_object
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -1043,3 +1048,152 @@ class TestBulkSelectionExcludesVersions(DjangoTestCase):
         )
 
         self.assertEqual(selected, [root.id])
+
+
+class TestVersionActionPermissions(DirectoriesMixin, APITestCase):
+    def setUp(self):
+        super().setUp()
+        self.user = UserFactory()
+        grant_all_global(self.user)
+        self.client.force_authenticate(self.user)
+        self.root = DocumentFactory(owner=UserFactory())
+        self.version = DocumentFactory(root_document=self.root, owner=None)
+
+    @override_settings(AUDIT_LOG_ENABLED=True)
+    def test_actions_reject_stale_version_ownership(self):
+        note = Note.objects.create(document=self.version, note="Version note")
+        for owner in (None, self.user):
+            self.version.owner = owner
+            self.version.save(update_fields=["owner"])
+            for action in (
+                "notes",
+                "suggestions",
+                "ai_suggestions",
+                "history",
+                "share_links",
+            ):
+                with self.subTest(owner=owner, action=action):
+                    response = self.client.get(
+                        f"/api/documents/{self.version.pk}/{action}/",
+                    )
+                    self.assertEqual(response.status_code, 403)
+            response = self.client.post(
+                f"/api/documents/{self.version.pk}/notes/",
+                {"note": "New note"},
+            )
+            self.assertEqual(response.status_code, 403)
+            response = self.client.delete(
+                f"/api/documents/{self.version.pk}/notes/?id={note.pk}",
+            )
+            self.assertEqual(response.status_code, 403)
+            response = self.client.post(
+                "/api/share_links/",
+                {"document": self.version.pk, "file_version": "original"},
+            )
+            self.assertEqual(response.status_code, 403)
+            response = self.client.post(
+                "/api/share_link_bundles/",
+                {"document_ids": [self.version.pk], "file_version": "original"},
+                format="json",
+            )
+            self.assertEqual(response.status_code, 400)
+            response = self.client.post(
+                "/api/documents/email/",
+                {
+                    "documents": [self.version.pk],
+                    "addresses": "recipient@example.com",
+                    "subject": "Version",
+                    "message": "Version",
+                },
+                format="json",
+            )
+            self.assertEqual(response.status_code, 403)
+            with (
+                mock.patch("documents.views.AIConfig") as ai_config,
+                mock.patch("documents.views.stream_chat_with_documents") as chat,
+            ):
+                ai_config.return_value.ai_enabled = True
+                response = self.client.post(
+                    "/api/documents/chat/",
+                    {"q": "Version?", "document_id": self.version.pk},
+                    format="json",
+                )
+            self.assertEqual(response.status_code, 403)
+            chat.assert_not_called()
+        self.assertTrue(Note.objects.filter(pk=note.pk).exists())
+        self.assertFalse(ShareLink.objects.exists())
+
+    @mock.patch("documents.views.build_share_link_bundle.apply_async")
+    def test_root_permissions_allow_sharing_a_private_version(self, build_mock):
+        self.version.owner = UserFactory()
+        self.version.save(update_fields=["owner"])
+        grant_object(self.user, self.root, "view_document", "change_document")
+        note = Note.objects.create(document=self.version, note="Version note")
+        response = self.client.get(f"/api/documents/{self.version.pk}/notes/")
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.data[0]["id"], note.pk)
+        response = self.client.post(
+            "/api/share_links/",
+            {"document": self.version.pk, "file_version": "original"},
+        )
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(ShareLink.objects.get().document_id, self.version.pk)
+        response = self.client.get(f"/api/documents/{self.version.pk}/share_links/")
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data), 1)
+        response = self.client.post(
+            "/api/share_link_bundles/",
+            {"document_ids": [self.version.pk], "file_version": "original"},
+            format="json",
+        )
+        self.assertEqual(response.status_code, 201)
+        build_mock.assert_called_once()
+
+    def test_root_view_permission_does_not_allow_note_changes(self):
+        grant_object(self.user, self.root, "view_document")
+        note = Note.objects.create(document=self.version, note="Version note")
+        response = self.client.get(f"/api/documents/{self.version.pk}/notes/")
+        self.assertEqual(response.status_code, 200)
+        response = self.client.post(
+            f"/api/documents/{self.version.pk}/notes/",
+            {"note": "New note"},
+        )
+        self.assertEqual(response.status_code, 403)
+        response = self.client.delete(
+            f"/api/documents/{self.version.pk}/notes/?id={note.pk}",
+        )
+        self.assertEqual(response.status_code, 403)
+        self.assertTrue(Note.objects.filter(pk=note.pk).exists())
+
+    @override_settings(AUDIT_LOG_ENABLED=True)
+    def test_history_uses_root_ownership(self):
+        self.root.owner = self.user
+        self.root.save(update_fields=["owner"])
+        self.version.owner = UserFactory()
+        self.version.save(update_fields=["owner"])
+        response = self.client.get(f"/api/documents/{self.version.pk}/history/")
+        self.assertEqual(response.status_code, 200)
+
+    def test_selection_data_rejects_stale_version_ownership(self):
+        for owner in (None, self.user):
+            self.version.owner = owner
+            self.version.save(update_fields=["owner"])
+            with self.subTest(owner=owner):
+                response = self.client.post(
+                    "/api/documents/selection_data/",
+                    {"documents": [self.version.pk]},
+                    format="json",
+                )
+                self.assertEqual(response.status_code, 403)
+
+    def test_selection_data_allows_private_version_of_permitted_root(self):
+        self.version.owner = UserFactory()
+        self.version.save(update_fields=["owner"])
+        grant_object(self.user, self.root, "view_document")
+        other = DocumentFactory(owner=self.user)
+        response = self.client.post(
+            "/api/documents/selection_data/",
+            {"documents": [self.version.pk, other.pk]},
+            format="json",
+        )
+        self.assertEqual(response.status_code, 200)

src/documents/tests/test_api_search.py

@@ -1196,6 +1196,59 @@ class TestDocumentSearchApi(DirectoriesMixin, APITestCase):
         self.assertIn(d3.id, result_ids)
         self.assertNotIn(d4.id, result_ids)
 
+    def test_search_more_like_version_uses_its_root(self) -> None:
+        """
+        GIVEN:
+            - A document similar in content to a root document, and one that is not
+            - A version of the root document, which is never indexed
+        WHEN:
+            - API request for more like the version
+        THEN:
+            - The documents similar to the root are returned, not the version
+        """
+        indexed = {}
+        for name, title, content, day in (
+            ("root", "bank statement 1", "things i paid for in august", (2019, 3, 4)),
+            (
+                "similar",
+                "bank statement 3",
+                "things i paid for in september",
+                (2020, 7, 9),
+            ),
+            (
+                "other",
+                "Quarterly Report",
+                "quarterly revenue profit margin",
+                (2021, 11, 30),
+            ),
+        ):
+            with time_machine.travel(
+                timezone.make_aware(datetime.datetime(*day)),
+                tick=False,
+            ):
+                indexed[name] = DocumentFactory(
+                    title=title,
+                    content=content,
+                    created=datetime.date(*day),
+                    added=timezone.make_aware(datetime.datetime(*day)),
+                )
+        version = DocumentFactory(
+            root_document=indexed["root"],
+            version_index=1,
+            content="things i paid for in august",
+        )
+        backend = get_backend()
+        for document in indexed.values():
+            backend.add_or_update(document)
+
+        response = self.client.get(f"/api/documents/?more_like_id={version.id}")
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        result_ids = [r["id"] for r in response.data["results"]]
+        self.assertIn(indexed["similar"].id, result_ids)
+        self.assertNotIn(indexed["other"].id, result_ids)
+        self.assertNotIn(version.id, result_ids)
+
     def test_more_like_requires_id_of_existing_document(self) -> None:
         """
         GIVEN:

src/documents/tests/test_api_trash.py

@@ -6,6 +6,7 @@ from rest_framework.test import APITestCase
 
 from documents.models import Document
 from paperless_testing.dirs import DirectoriesMixin
+from paperless_testing.factories import DocumentFactory
 from paperless_testing.factories import UserFactory
 from paperless_testing.permissions import grant_all_global
 
@@ -279,3 +280,53 @@ class TestTrashAPI(DirectoriesMixin, APITestCase):
             Document.objects.filter(root_document=root).values_list("id", flat=True),
             [version.pk for version in versions],
         )
+
+    def test_api_trash_version_follows_root_owner(self) -> None:
+        """
+        GIVEN:
+            - A deleted version of user2's document, owned by nobody
+            - A deleted version of the user's document, owned by user2
+        WHEN:
+            - The user lists the trash and tries to restore or empty the versions
+        THEN:
+            - Only the version of the user's own document is listed
+            - The other version can't be restored or emptied
+            - The version of the user's own document can be restored
+        """
+        user2 = UserFactory(username="user2")
+        other_version = DocumentFactory(
+            root_document=DocumentFactory(owner=user2),
+            version_index=1,
+        )
+        other_version.delete()
+        own_version = DocumentFactory(
+            owner=user2,
+            root_document=DocumentFactory(owner=self.user),
+            version_index=1,
+        )
+        own_version.delete()
+
+        resp = self.client.get("/api/trash/")
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertEqual(
+            [doc["id"] for doc in resp.data["results"]],
+            [own_version.pk],
+        )
+
+        for action in ("restore", "empty"):
+            with self.subTest(action=action):
+                resp = self.client.post(
+                    "/api/trash/",
+                    {"action": action, "documents": [other_version.pk]},
+                )
+                self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+                self.assertTrue(
+                    Document.deleted_objects.filter(pk=other_version.pk).exists(),
+                )
+
+        resp = self.client.post(
+            "/api/trash/",
+            {"action": "restore", "documents": [own_version.pk]},
+        )
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertTrue(Document.objects.filter(pk=own_version.pk).exists())

src/documents/tests/test_bulk_edit.py

@@ -1,9 +1,10 @@
 import shutil
-from collections.abc import Callable
 from datetime import date
 from pathlib import Path
 from unittest import mock
 
+import pikepdf
+import pytest
 from django.contrib.auth.models import Group
 from django.contrib.auth.models import Permission
 from django.contrib.auth.models import User
@@ -12,6 +13,7 @@ from django.test import TestCase
 from django.test.utils import CaptureQueriesContext
 from guardian.shortcuts import get_groups_with_perms
 from guardian.shortcuts import get_users_with_perms
+from pytest_mock import MockerFixture
 
 from documents import bulk_edit
 from documents.models import Correspondent
@@ -21,9 +23,9 @@ from documents.models import Document
 from documents.models import DocumentType
 from documents.models import StoragePath
 from documents.models import Tag
-from documents.pdf_ops import PageSpec
 from documents.permissions import set_permissions_for_objects
 from paperless_testing.dirs import DirectoriesMixin
+from paperless_testing.factories import DocumentFactory
 from paperless_testing.permissions import grant_object
 
 
@@ -794,14 +796,16 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         self.img_doc.save()
 
     @staticmethod
-    def fake_decrypt(
-        src: Path,
-        make_dst: Callable[[], Path],
-        password: str,
-    ) -> Path:
-        dst = make_dst()
-        dst.write_bytes(b"password removed")
-        return dst
+    def mock_password_required_pdf(
+        mock_open: mock.Mock,
+        fake_pdf: mock.Mock,
+    ) -> None:
+        password_context = mock.MagicMock()
+        password_context.__enter__.return_value = fake_pdf
+        mock_open.side_effect = [
+            pikepdf.PasswordError("password required"),
+            password_context,
+        ]
 
     @mock.patch("documents.tasks.consume_file.s")
     def test_merge(self, mock_consume_file) -> None:
@@ -846,12 +850,12 @@ class TestPDFActions(DirectoriesMixin, TestCase):
 
         self.assertEqual(result, "OK")
 
-    @mock.patch("documents.pdf_ops.PdfMerger")
+    @mock.patch("pikepdf.open")
     @mock.patch("documents.tasks.consume_file.s")
     def test_merge_uses_latest_version_source_for_root_selection(
         self,
         mock_consume_file,
-        mock_merger,
+        mock_open_pdf,
     ) -> None:
         version_file = self.dirs.scratch_dir / "sample2_version_merge.pdf"
         shutil.copy(self.doc2.source_path, version_file)
@@ -862,14 +866,16 @@ class TestPDFActions(DirectoriesMixin, TestCase):
             filename=version_file,
             mime_type="application/pdf",
         )
-        merger = mock_merger.return_value.__enter__.return_value
-        merger.save.side_effect = lambda dst: shutil.copy(version.source_path, dst)
+        fake_pdf = mock.MagicMock()
+        fake_pdf.pdf_version = "1.7"
+        fake_pdf.pages = [mock.Mock()]
+        mock_open_pdf.return_value.__enter__.return_value = fake_pdf
 
         result = bulk_edit.merge([self.doc2.id])
 
         self.assertEqual(result, "OK")
-        merger.add.assert_called_once_with(version.source_path)
-        mock_consume_file.assert_called_once()
+        mock_open_pdf.assert_called_once_with(str(version.source_path))
+        mock_consume_file.assert_not_called()
 
     @mock.patch("documents.bulk_edit.delete.si")
     @mock.patch("documents.tasks.consume_file.s")
@@ -1031,18 +1037,18 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         )
 
     @mock.patch("documents.tasks.consume_file.delay")
-    @mock.patch("documents.pdf_ops.PdfMerger.add")
-    def test_merge_with_errors(self, mock_add, mock_consume_file) -> None:
+    @mock.patch("pikepdf.open")
+    def test_merge_with_errors(self, mock_open_pdf, mock_consume_file) -> None:
         """
         GIVEN:
             - Existing documents
         WHEN:
             - Merge action is called with 2 documents
-            - Error occurs when adding both files
+            - Error occurs when opening both files
         THEN:
             - Consume file should not be called
         """
-        mock_add.side_effect = Exception("Error opening PDF")
+        mock_open_pdf.side_effect = Exception("Error opening PDF")
         doc_ids = [self.doc2.id, self.doc3.id]
 
         with self.assertLogs("paperless.bulk_edit", level="ERROR") as cm:
@@ -1079,12 +1085,12 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         self.assertEqual(result, "OK")
 
     @mock.patch("documents.bulk_edit.group")
-    @mock.patch("documents.pdf_ops.build_pdfs")
+    @mock.patch("pikepdf.open")
     @mock.patch("documents.tasks.consume_file.s")
     def test_split_uses_latest_version_source_for_root_selection(
         self,
         mock_consume_file,
-        mock_build_pdfs,
+        mock_open_pdf,
         mock_group,
     ) -> None:
         version_file = self.dirs.scratch_dir / "sample2_version_split.pdf"
@@ -1096,15 +1102,17 @@ class TestPDFActions(DirectoriesMixin, TestCase):
             filename=version_file,
             mime_type="application/pdf",
         )
-        mock_build_pdfs.return_value = [version.source_path, version.source_path]
+        fake_pdf = mock.MagicMock()
+        fake_pdf.pages = [mock.Mock(), mock.Mock()]
+        mock_open_pdf.return_value.__enter__.return_value = fake_pdf
         mock_group.return_value.delay.return_value = None
 
         result = bulk_edit.split([self.doc2.id], [[1], [2]])
 
         self.assertEqual(result, "OK")
-        self.assertEqual(mock_build_pdfs.call_args.args[0], version.source_path)
-        self.assertEqual(mock_consume_file.call_count, 2)
-        mock_group.return_value.delay.assert_called_once()
+        mock_open_pdf.assert_called_once_with(version.source_path)
+        mock_consume_file.assert_not_called()
+        mock_group.return_value.delay.assert_not_called()
 
     @mock.patch("documents.bulk_edit.delete.si")
     @mock.patch("documents.tasks.consume_file.s")
@@ -1192,18 +1200,18 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         self.assertEqual(self.doc2.archive_serial_number, 222)
 
     @mock.patch("documents.tasks.consume_file.apply_async")
-    @mock.patch("documents.pdf_ops.build_pdfs")
-    def test_split_with_errors(self, mock_build_pdfs, mock_consume_file) -> None:
+    @mock.patch("pikepdf.Pdf.save")
+    def test_split_with_errors(self, mock_save_pdf, mock_consume_file) -> None:
         """
         GIVEN:
             - Existing documents
         WHEN:
             - Split action is called with 1 document and 2 page groups
-            - Error occurs when building the files
+            - Error occurs when saving the files
         THEN:
             - Consume file should not be called
         """
-        mock_build_pdfs.side_effect = Exception("Error building PDFs")
+        mock_save_pdf.side_effect = Exception("Error saving PDF")
         doc_ids = [self.doc2.id]
         pages = [[1, 2], [3]]
 
@@ -1238,10 +1246,10 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         self.assertEqual(result, "OK")
 
     @mock.patch("documents.tasks.consume_file.apply_async")
-    @mock.patch("documents.pdf_ops.rotate_pdf")
+    @mock.patch("pikepdf.Pdf.save")
     def test_rotate_with_error(
         self,
-        mock_rotate_pdf,
+        mock_pdf_save,
         mock_consume_delay,
     ) -> None:
         """
@@ -1249,11 +1257,11 @@ class TestPDFActions(DirectoriesMixin, TestCase):
             - Existing documents
         WHEN:
             - Rotate action is called with 2 documents
-            - Rotating the PDF raises an error
+            - PikePDF raises an error
         THEN:
             - Rotate action should be called 0 times
         """
-        mock_rotate_pdf.side_effect = Exception("Error rotating PDF")
+        mock_pdf_save.side_effect = Exception("Error saving PDF")
         doc_ids = [self.doc2.id, self.doc3.id]
 
         with self.assertLogs("paperless.bulk_edit", level="ERROR") as cm:
@@ -1288,10 +1296,10 @@ class TestPDFActions(DirectoriesMixin, TestCase):
 
     @mock.patch("documents.data_models.magic.from_file", return_value="application/pdf")
     @mock.patch("documents.tasks.consume_file.apply_async")
-    @mock.patch("documents.pdf_ops.rotate_pdf")
+    @mock.patch("pikepdf.open")
     def test_rotate_explicit_selection_uses_root_source_when_root_selected(
         self,
-        mock_rotate_pdf,
+        mock_open,
         mock_consume_delay,
         mock_magic,
     ) -> None:
@@ -1300,6 +1308,9 @@ class TestPDFActions(DirectoriesMixin, TestCase):
             title="B version 1",
             root_document=self.doc2,
         )
+        fake_pdf = mock.MagicMock()
+        fake_pdf.pages = [mock.Mock()]
+        mock_open.return_value.__enter__.return_value = fake_pdf
 
         result = bulk_edit.rotate(
             [self.doc2.id],
@@ -1308,35 +1319,26 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         )
 
         self.assertEqual(result, "OK")
-        self.assertEqual(mock_rotate_pdf.call_args.args[0], self.doc2.source_path)
+        mock_open.assert_called_once_with(self.doc2.source_path)
         mock_consume_delay.assert_called_once()
 
     @mock.patch("documents.tasks.consume_file.apply_async")
-    @mock.patch("documents.pdf_ops.remove_pages")
+    @mock.patch("pikepdf.Pdf.save")
     @mock.patch("documents.data_models.magic.from_file", return_value="application/pdf")
-    def test_delete_pages(
-        self,
-        mock_magic,
-        mock_remove_pages,
-        mock_consume_delay,
-    ) -> None:
+    def test_delete_pages(self, mock_magic, mock_pdf_save, mock_consume_delay) -> None:
         """
         GIVEN:
             - Existing documents
         WHEN:
             - Delete pages action is called with 1 document and 2 pages
         THEN:
-            - The pages are removed from the document's source PDF
+            - Save should be called once
             - A new version should be enqueued via consume_file
         """
         doc_ids = [self.doc2.id]
         pages = [1, 3]
         result = bulk_edit.delete_pages(doc_ids, pages)
-        mock_remove_pages.assert_called_once_with(
-            self.doc2.source_path,
-            mock.ANY,
-            [1, 3],
-        )
+        mock_pdf_save.assert_called_once()
         mock_consume_delay.assert_called_once()
         task_kwargs = mock_consume_delay.call_args.kwargs["kwargs"]
         self.assertEqual(task_kwargs["input_doc"].root_document_id, self.doc2.id)
@@ -1348,10 +1350,10 @@ class TestPDFActions(DirectoriesMixin, TestCase):
 
     @mock.patch("documents.data_models.magic.from_file", return_value="application/pdf")
     @mock.patch("documents.tasks.consume_file.apply_async")
-    @mock.patch("documents.pdf_ops.remove_pages")
+    @mock.patch("pikepdf.open")
     def test_delete_pages_explicit_selection_uses_root_source_when_root_selected(
         self,
-        mock_remove_pages,
+        mock_open,
         mock_consume_delay,
         mock_magic,
     ) -> None:
@@ -1360,6 +1362,9 @@ class TestPDFActions(DirectoriesMixin, TestCase):
             title="B version 1",
             root_document=self.doc2,
         )
+        fake_pdf = mock.MagicMock()
+        fake_pdf.pages = [mock.Mock(), mock.Mock()]
+        mock_open.return_value.__enter__.return_value = fake_pdf
 
         result = bulk_edit.delete_pages(
             [self.doc2.id],
@@ -1368,26 +1373,23 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         )
 
         self.assertEqual(result, "OK")
-        self.assertEqual(mock_remove_pages.call_args.args[0], self.doc2.source_path)
+        mock_open.assert_called_once_with(self.doc2.source_path)
         mock_consume_delay.assert_called_once()
 
     @mock.patch("documents.tasks.consume_file.apply_async")
-    @mock.patch("documents.pdf_ops.remove_pages")
-    def test_delete_pages_with_error(
-        self,
-        mock_remove_pages,
-        mock_consume_delay,
-    ) -> None:
+    @mock.patch("pikepdf.Pdf.save")
+    def test_delete_pages_with_error(self, mock_pdf_save, mock_consume_delay) -> None:
         """
         GIVEN:
             - Existing documents
         WHEN:
             - Delete pages action is called with 1 document and 2 pages
-            - Removing the pages raises an error
+            - PikePDF raises an error
         THEN:
+            - Save should be called once
             - No new version should be enqueued
         """
-        mock_remove_pages.side_effect = Exception("Error removing pages")
+        mock_pdf_save.side_effect = Exception("Error saving PDF")
         doc_ids = [self.doc2.id]
         pages = [1, 3]
 
@@ -1417,41 +1419,6 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         self.assertEqual(result, "OK")
         mock_group.return_value.delay.assert_called_once()
 
-    @mock.patch("documents.bulk_edit.group")
-    @mock.patch("documents.pdf_ops.build_pdfs")
-    @mock.patch("documents.tasks.consume_file.s")
-    def test_edit_pdf_maps_operations_to_outputs(
-        self,
-        mock_consume_file: mock.Mock,
-        mock_build_pdfs: mock.Mock,
-        mock_group: mock.Mock,
-    ) -> None:
-        """
-        GIVEN:
-            - Existing document
-        WHEN:
-            - edit_pdf is called with operations interleaved across two outputs,
-              some of them rotated
-        THEN:
-            - Each output is built from its own operations, in operation order,
-              with the requested rotation
-        """
-        mock_build_pdfs.return_value = [self.doc2.source_path, self.doc2.source_path]
-        mock_group.return_value.delay.return_value = None
-        operations = [
-            {"page": 3, "doc": 1},
-            {"page": 1, "doc": 0, "rotate": 90},
-            {"page": 2, "doc": 1, "rotate": 180},
-        ]
-
-        bulk_edit.edit_pdf([self.doc2.id], operations)
-
-        outputs = mock_build_pdfs.call_args.args[1]
-        self.assertEqual(
-            [specs for specs, _ in outputs],
-            [[PageSpec(1, 90)], [PageSpec(3), PageSpec(2, 180)]],
-        )
-
     @mock.patch("documents.bulk_edit.group")
     @mock.patch("documents.tasks.consume_file.s")
     def test_edit_pdf_with_user_override(self, mock_consume_file, mock_group) -> None:
@@ -1567,10 +1534,12 @@ class TestPDFActions(DirectoriesMixin, TestCase):
 
     @mock.patch("documents.data_models.magic.from_file", return_value="application/pdf")
     @mock.patch("documents.tasks.consume_file.apply_async")
-    @mock.patch("documents.pdf_ops.build_pdfs")
+    @mock.patch("pikepdf.new")
+    @mock.patch("pikepdf.open")
     def test_edit_pdf_explicit_selection_uses_root_source_when_root_selected(
         self,
-        mock_build_pdfs,
+        mock_open,
+        mock_new,
         mock_consume_delay,
         mock_magic,
     ) -> None:
@@ -1579,7 +1548,12 @@ class TestPDFActions(DirectoriesMixin, TestCase):
             title="B version 1",
             root_document=self.doc2,
         )
-        mock_build_pdfs.return_value = [Path("edited.pdf")]
+        fake_pdf = mock.MagicMock()
+        fake_pdf.pages = [mock.Mock()]
+        mock_open.return_value.__enter__.return_value = fake_pdf
+        output_pdf = mock.MagicMock()
+        output_pdf.pages = []
+        mock_new.return_value = output_pdf
 
         result = bulk_edit.edit_pdf(
             [self.doc2.id],
@@ -1589,7 +1563,7 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         )
 
         self.assertEqual(result, "OK")
-        self.assertEqual(mock_build_pdfs.call_args.args[0], self.doc2.source_path)
+        mock_open.assert_called_once_with(self.doc2.source_path)
         mock_consume_delay.assert_called_once()
 
     @mock.patch("documents.bulk_edit.group")
@@ -1615,6 +1589,31 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         self.assertEqual(result, "OK")
         mock_group.return_value.delay.assert_called_once()
 
+    @mock.patch("documents.bulk_edit.group")
+    @mock.patch("documents.tasks.consume_file.s")
+    def test_edit_pdf_open_failure(
+        self,
+        mock_consume_file: mock.Mock,
+        mock_group: mock.Mock,
+    ) -> None:
+        """
+        GIVEN:
+            - Existing document
+        WHEN:
+            - edit_pdf fails to open PDF
+        THEN:
+            - Task group is not called
+        """
+        doc_ids = [self.doc2.id]
+        operations = [
+            {"page": 9999},  # invalid page, forces error during PDF load
+        ]
+        with self.assertLogs("paperless.bulk_edit", level="ERROR"):
+            with self.assertRaises(Exception):
+                bulk_edit.edit_pdf(doc_ids, operations)
+        mock_group.assert_not_called()
+        mock_consume_file.assert_not_called()
+
     @mock.patch("documents.bulk_edit.group")
     @mock.patch("documents.tasks.consume_file.s")
     def test_edit_pdf_multiple_outputs_with_update_flag_errors(
@@ -1641,15 +1640,23 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         mock_group.assert_not_called()
         mock_consume_file.assert_not_called()
 
+    @mock.patch("pikepdf.open")
+    def test_edit_pdf_rejects_invalid_operations(self, mock_open) -> None:
+        for operations in ([], [{"page": 1, "doc": 2**32}]):
+            with self.subTest(operations=operations):
+                with self.assertLogs("paperless.bulk_edit", level="ERROR"):
+                    with self.assertRaisesRegex(ValueError, "index is out of bounds"):
+                        bulk_edit.edit_pdf([self.doc2.id], operations)
+
+        mock_open.assert_not_called()
+
     @mock.patch("documents.bulk_edit.update_document_content_maybe_archive_file.delay")
     @mock.patch("documents.tasks.consume_file.apply_async")
     @mock.patch("documents.bulk_edit.tempfile.mkdtemp")
-    @mock.patch("documents.pdf_ops.decrypt_pdf")
-    @mock.patch("documents.pdf_ops.needs_decrypt", return_value=True)
+    @mock.patch("pikepdf.open")
     def test_remove_password_update_document(
         self,
-        mock_needs_decrypt,
-        mock_decrypt,
+        mock_open,
         mock_mkdtemp,
         mock_consume_delay,
         mock_update_document,
@@ -1658,7 +1665,16 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         temp_dir = self.dirs.scratch_dir / "remove-password-update"
         temp_dir.mkdir(parents=True, exist_ok=True)
         mock_mkdtemp.return_value = str(temp_dir)
-        mock_decrypt.side_effect = self.fake_decrypt
+
+        fake_pdf = mock.MagicMock()
+        fake_pdf.pages = [mock.Mock(), mock.Mock(), mock.Mock()]
+        fake_pdf.is_encrypted = True
+
+        def save_side_effect(target_path):
+            Path(target_path).write_bytes(b"new pdf content")
+
+        fake_pdf.save.side_effect = save_side_effect
+        mock_open.return_value.__enter__.return_value = fake_pdf
 
         result = bulk_edit.remove_password(
             [doc.id],
@@ -1667,8 +1683,14 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         )
 
         self.assertEqual(result, "OK")
-        mock_needs_decrypt.assert_called_once_with(doc.source_path)
-        mock_decrypt.assert_called_once_with(doc.source_path, mock.ANY, "secret")
+        self.assertEqual(
+            mock_open.call_args_list,
+            [
+                mock.call(doc.source_path),
+                mock.call(doc.source_path, password="secret"),
+            ],
+        )
+        fake_pdf.remove_unreferenced_resources.assert_called_once()
         mock_update_document.assert_not_called()
         mock_consume_delay.assert_called_once()
         task_kwargs = mock_consume_delay.call_args.kwargs["kwargs"]
@@ -1681,15 +1703,40 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         self.assertEqual(task_kwargs["input_doc"].root_document_id, doc.id)
         self.assertIsNotNone(task_kwargs["overrides"])
 
+    @mock.patch("documents.tasks.consume_file.apply_async")
+    @mock.patch("documents.bulk_edit.tempfile.mkdtemp")
+    @mock.patch("pikepdf.open")
+    def test_remove_password_update_document_skips_unencrypted_pdf(
+        self,
+        mock_open,
+        mock_mkdtemp,
+        mock_consume_delay,
+    ) -> None:
+        doc = self.doc1
+        fake_pdf = mock.MagicMock()
+        fake_pdf.is_encrypted = False
+        mock_open.return_value.__enter__.return_value = fake_pdf
+
+        result = bulk_edit.remove_password(
+            [doc.id],
+            password="secret",
+            update_document=True,
+        )
+
+        self.assertEqual(result, "OK")
+        mock_open.assert_called_once_with(doc.source_path)
+        fake_pdf.remove_unreferenced_resources.assert_not_called()
+        fake_pdf.save.assert_not_called()
+        mock_mkdtemp.assert_not_called()
+        mock_consume_delay.assert_not_called()
+
     @mock.patch("documents.bulk_edit.update_document_content_maybe_archive_file.delay")
     @mock.patch("documents.tasks.consume_file.apply_async")
     @mock.patch("documents.bulk_edit.tempfile.mkdtemp")
-    @mock.patch("documents.pdf_ops.decrypt_pdf")
-    @mock.patch("documents.pdf_ops.needs_decrypt", return_value=True)
+    @mock.patch("pikepdf.open")
     def test_remove_password_update_document_uses_source_paths(
         self,
-        mock_needs_decrypt,
-        mock_decrypt,
+        mock_open,
         mock_mkdtemp,
         mock_consume_delay,
         mock_update_document,
@@ -1700,7 +1747,14 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         temp_dir = self.dirs.scratch_dir / "remove-password-source-file"
         temp_dir.mkdir(parents=True, exist_ok=True)
         mock_mkdtemp.return_value = str(temp_dir)
-        mock_decrypt.side_effect = self.fake_decrypt
+
+        fake_pdf = mock.MagicMock()
+        self.mock_password_required_pdf(mock_open, fake_pdf)
+
+        def save_side_effect(target_path):
+            Path(target_path).write_bytes(b"new pdf content")
+
+        fake_pdf.save.side_effect = save_side_effect
 
         result = bulk_edit.remove_password(
             [doc.id],
@@ -1710,19 +1764,22 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         )
 
         self.assertEqual(result, "OK")
-        mock_needs_decrypt.assert_called_once_with(source_file)
-        mock_decrypt.assert_called_once_with(source_file, mock.ANY, "secret")
+        self.assertEqual(
+            mock_open.call_args_list,
+            [
+                mock.call(source_file),
+                mock.call(source_file, password="secret"),
+            ],
+        )
         mock_update_document.assert_not_called()
         mock_consume_delay.assert_called_once()
 
     @mock.patch("documents.data_models.magic.from_file", return_value="application/pdf")
     @mock.patch("documents.tasks.consume_file.apply_async")
-    @mock.patch("documents.pdf_ops.decrypt_pdf")
-    @mock.patch("documents.pdf_ops.needs_decrypt", return_value=True)
+    @mock.patch("pikepdf.open")
     def test_remove_password_explicit_selection_uses_root_source_when_root_selected(
         self,
-        mock_needs_decrypt,
-        mock_decrypt,
+        mock_open,
         mock_consume_delay,
         mock_magic,
     ) -> None:
@@ -1731,7 +1788,8 @@ class TestPDFActions(DirectoriesMixin, TestCase):
             title="A version 1",
             root_document=self.doc1,
         )
-        mock_decrypt.return_value = Path("unprotected.pdf")
+        fake_pdf = mock.MagicMock()
+        self.mock_password_required_pdf(mock_open, fake_pdf)
 
         result = bulk_edit.remove_password(
             [self.doc1.id],
@@ -1741,11 +1799,12 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         )
 
         self.assertEqual(result, "OK")
-        mock_needs_decrypt.assert_called_once_with(self.doc1.source_path)
-        mock_decrypt.assert_called_once_with(
-            self.doc1.source_path,
-            mock.ANY,
-            "secret",
+        self.assertEqual(
+            mock_open.call_args_list,
+            [
+                mock.call(self.doc1.source_path),
+                mock.call(self.doc1.source_path, password="secret"),
+            ],
         )
         mock_consume_delay.assert_called_once()
 
@@ -1753,12 +1812,10 @@ class TestPDFActions(DirectoriesMixin, TestCase):
     @mock.patch("documents.bulk_edit.group")
     @mock.patch("documents.tasks.consume_file.s")
     @mock.patch("documents.bulk_edit.tempfile.mkdtemp")
-    @mock.patch("documents.pdf_ops.decrypt_pdf")
-    @mock.patch("documents.pdf_ops.needs_decrypt", return_value=True)
+    @mock.patch("pikepdf.open")
     def test_remove_password_creates_consumable_document(
         self,
-        mock_needs_decrypt: mock.Mock,
-        mock_decrypt: mock.Mock,
+        mock_open: mock.Mock,
         mock_mkdtemp: mock.Mock,
         mock_consume_file: mock.Mock,
         mock_group: mock.Mock,
@@ -1768,7 +1825,15 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         temp_dir = self.dirs.scratch_dir / "remove-password"
         temp_dir.mkdir(parents=True, exist_ok=True)
         mock_mkdtemp.return_value = str(temp_dir)
-        mock_decrypt.side_effect = self.fake_decrypt
+
+        fake_pdf = mock.MagicMock()
+        fake_pdf.pages = [mock.Mock(), mock.Mock()]
+        self.mock_password_required_pdf(mock_open, fake_pdf)
+
+        def save_side_effect(target_path: Path) -> None:
+            target_path.write_bytes(b"password removed")
+
+        fake_pdf.save.side_effect = save_side_effect
         mock_group.return_value.delay.return_value = None
 
         user = User.objects.create(username="owner")
@@ -1783,7 +1848,13 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         )
 
         self.assertEqual(result, "OK")
-        mock_decrypt.assert_called_once_with(doc.source_path, mock.ANY, "secret")
+        self.assertEqual(
+            mock_open.call_args_list,
+            [
+                mock.call(doc.source_path),
+                mock.call(doc.source_path, password="secret"),
+            ],
+        )
         mock_consume_file.assert_called_once()
         call_kwargs = mock_consume_file.call_args.kwargs
         consumable_document = call_kwargs["input_doc"]
@@ -1805,18 +1876,21 @@ class TestPDFActions(DirectoriesMixin, TestCase):
     @mock.patch("documents.bulk_edit.chord")
     @mock.patch("documents.bulk_edit.group")
     @mock.patch("documents.tasks.consume_file.s")
-    @mock.patch("documents.pdf_ops.decrypt_pdf")
-    @mock.patch("documents.pdf_ops.needs_decrypt", return_value=False)
+    @mock.patch("documents.bulk_edit.tempfile.mkdtemp")
+    @mock.patch("pikepdf.open")
     def test_remove_password_skips_unencrypted_pdf_without_queueing(
         self,
-        mock_needs_decrypt: mock.Mock,
-        mock_decrypt: mock.Mock,
+        mock_open: mock.Mock,
+        mock_mkdtemp: mock.Mock,
         mock_consume_file: mock.Mock,
         mock_group: mock.Mock,
         mock_chord: mock.Mock,
         mock_delete: mock.Mock,
     ) -> None:
         doc = self.doc2
+        fake_pdf = mock.MagicMock()
+        fake_pdf.is_encrypted = False
+        mock_open.return_value.__enter__.return_value = fake_pdf
 
         result = bulk_edit.remove_password(
             [doc.id],
@@ -1826,8 +1900,10 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         )
 
         self.assertEqual(result, "OK")
-        mock_needs_decrypt.assert_called_once_with(doc.source_path)
-        mock_decrypt.assert_not_called()
+        mock_open.assert_called_once_with(doc.source_path)
+        fake_pdf.remove_unreferenced_resources.assert_not_called()
+        fake_pdf.save.assert_not_called()
+        mock_mkdtemp.assert_not_called()
         mock_consume_file.assert_not_called()
         mock_group.assert_not_called()
         mock_chord.assert_not_called()
@@ -1838,12 +1914,10 @@ class TestPDFActions(DirectoriesMixin, TestCase):
     @mock.patch("documents.bulk_edit.group")
     @mock.patch("documents.tasks.consume_file.s")
     @mock.patch("documents.bulk_edit.tempfile.mkdtemp")
-    @mock.patch("documents.pdf_ops.decrypt_pdf")
-    @mock.patch("documents.pdf_ops.needs_decrypt", return_value=True)
+    @mock.patch("pikepdf.open")
     def test_remove_password_deletes_original(
         self,
-        mock_needs_decrypt: mock.Mock,
-        mock_decrypt: mock.Mock,
+        mock_open: mock.Mock,
         mock_mkdtemp: mock.Mock,
         mock_consume_file: mock.Mock,
         mock_group: mock.Mock,
@@ -1854,7 +1928,15 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         temp_dir = self.dirs.scratch_dir / "remove-password-delete"
         temp_dir.mkdir(parents=True, exist_ok=True)
         mock_mkdtemp.return_value = str(temp_dir)
-        mock_decrypt.side_effect = self.fake_decrypt
+
+        fake_pdf = mock.MagicMock()
+        fake_pdf.pages = [mock.Mock(), mock.Mock()]
+        self.mock_password_required_pdf(mock_open, fake_pdf)
+
+        def save_side_effect(target_path: Path) -> None:
+            target_path.write_bytes(b"password removed")
+
+        fake_pdf.save.side_effect = save_side_effect
         mock_chord.return_value.delay.return_value = None
 
         result = bulk_edit.remove_password(
@@ -1866,23 +1948,23 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         )
 
         self.assertEqual(result, "OK")
-        mock_decrypt.assert_called_once_with(doc.source_path, mock.ANY, "secret")
+        self.assertEqual(
+            mock_open.call_args_list,
+            [
+                mock.call(doc.source_path),
+                mock.call(doc.source_path, password="secret"),
+            ],
+        )
         mock_consume_file.assert_called_once()
         mock_group.assert_not_called()
         mock_chord.assert_called_once()
         mock_chord.return_value.delay.assert_called_once()
         mock_delete.si.assert_called_once_with([doc.id])
 
-    @mock.patch(
-        "documents.pdf_ops.decrypt_pdf",
-        side_effect=RuntimeError("wrong password"),
-    )
-    @mock.patch("documents.pdf_ops.needs_decrypt", return_value=True)
-    def test_remove_password_failure_raises_value_error(
-        self,
-        mock_needs_decrypt: mock.Mock,
-        mock_decrypt: mock.Mock,
-    ) -> None:
+    @mock.patch("pikepdf.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:
             with self.assertRaises(ValueError) as exc:
                 bulk_edit.remove_password([self.doc1.id], password="secret")
@@ -1891,18 +1973,22 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         self.assertIn("Error removing password from document", cm.output[0])
 
 
-class TestBulkEditReprocess(DirectoriesMixin, TestCase):
-    def setUp(self) -> None:
-        super().setUp()
-
-        self.doc = Document.objects.create(
-            title="test",
-            checksum="A",
-            mime_type="application/pdf",
+@pytest.mark.django_db
+class TestBulkEditReprocess:
+    @pytest.fixture
+    def mock_task(self, mocker: MockerFixture) -> mock.MagicMock:
+        return mocker.patch(
+            "documents.bulk_edit.update_document_content_maybe_archive_file",
         )
 
-    @mock.patch("documents.bulk_edit.update_document_content_maybe_archive_file")
-    def test_reprocess_defaults_to_local(self, mock_task: mock.Mock) -> None:
+    @staticmethod
+    def _queued_ids(mock_task: mock.MagicMock) -> list[int]:
+        return [
+            call.kwargs["kwargs"]["document_id"]
+            for call in mock_task.apply_async.call_args_list
+        ]
+
+    def test_reprocess_defaults_to_local(self, mock_task: mock.MagicMock) -> None:
         """
         GIVEN:
             - A reprocess request that says nothing about remote OCR
@@ -1911,18 +1997,17 @@ class TestBulkEditReprocess(DirectoriesMixin, TestCase):
         THEN:
             - The task is queued without asking for the remote engine
         """
-        result = bulk_edit.reprocess([self.doc.id])
+        doc = DocumentFactory()
+
+        assert bulk_edit.reprocess([doc.id]) == "OK"
 
-        self.assertEqual(result, "OK")
         mock_task.apply_async.assert_called_once()
-        _, kwargs = mock_task.apply_async.call_args
-        self.assertEqual(
-            kwargs["kwargs"],
-            {"document_id": self.doc.id, "remote_ocr": False},
-        )
+        assert mock_task.apply_async.call_args.kwargs["kwargs"] == {
+            "document_id": doc.id,
+            "remote_ocr": False,
+        }
 
-    @mock.patch("documents.bulk_edit.update_document_content_maybe_archive_file")
-    def test_reprocess_passes_remote_ocr(self, mock_task: mock.Mock) -> None:
+    def test_reprocess_passes_remote_ocr(self, mock_task: mock.MagicMock) -> None:
         """
         GIVEN:
             - A reprocess request that explicitly asks for remote OCR
@@ -1931,14 +2016,72 @@ class TestBulkEditReprocess(DirectoriesMixin, TestCase):
         THEN:
             - The request is forwarded to the task for every document
         """
-        other = Document.objects.create(
-            title="test2",
-            checksum="B",
-            mime_type="application/pdf",
+        docs = DocumentFactory.create_batch(2)
+
+        bulk_edit.reprocess([doc.id for doc in docs], remote_ocr=True)
+
+        assert mock_task.apply_async.call_count == 2
+        for call in mock_task.apply_async.call_args_list:
+            assert call.kwargs["kwargs"]["remote_ocr"]
+
+    def test_reprocess_root_uses_latest_version(
+        self,
+        mock_task: mock.MagicMock,
+    ) -> None:
+        """
+        GIVEN:
+            - A root document with two versions
+        WHEN:
+            - reprocess is called with the root document
+        THEN:
+            - The latest version is reprocessed, not the root's original file
+        """
+        root = DocumentFactory()
+        DocumentFactory(root_document=root, version_index=1)
+        latest = DocumentFactory(root_document=root, version_index=2)
+
+        bulk_edit.reprocess([root.id])
+
+        assert self._queued_ids(mock_task) == [latest.id]
+
+    def test_reprocess_explicit_version(self, mock_task: mock.MagicMock) -> None:
+        """
+        GIVEN:
+            - A root document with two versions
+        WHEN:
+            - reprocess is called with the older version
+        THEN:
+            - That version is reprocessed
+        """
+        root = DocumentFactory()
+        older = DocumentFactory(root_document=root, version_index=1)
+        DocumentFactory(root_document=root, version_index=2)
+
+        bulk_edit.reprocess([older.id])
+
+        assert self._queued_ids(mock_task) == [older.id]
+
+    def test_reprocess_root_and_latest_version_dispatches_once(
+        self,
+        mock_task: mock.MagicMock,
+    ) -> None:
+        """
+        GIVEN:
+            - A root document with two versions, the latest created on a
+              different date than the root
+        WHEN:
+            - reprocess is called with both the root and its latest version
+        THEN:
+            - The latest version is reprocessed only once
+        """
+        root = DocumentFactory(created=date(2024, 1, 1))
+        DocumentFactory(root_document=root, version_index=1)
+        latest = DocumentFactory(
+            root_document=root,
+            version_index=2,
+            created=date(2025, 1, 1),
         )
 
-        bulk_edit.reprocess([self.doc.id, other.id], remote_ocr=True)
+        bulk_edit.reprocess([root.id, latest.id])
 
-        self.assertEqual(mock_task.apply_async.call_count, 2)
-        for call in mock_task.apply_async.call_args_list:
-            self.assertTrue(call.kwargs["kwargs"]["remote_ocr"])
+        assert self._queued_ids(mock_task) == [latest.id]

src/documents/tests/test_management.py

@@ -22,6 +22,7 @@ from documents.models import Document
 from documents.tasks import update_document_content_maybe_archive_file
 from paperless_testing.assertions import FileSystemAssertsMixin
 from paperless_testing.dirs import DirectoriesMixin
+from paperless_testing.factories import DocumentFactory
 
 sample_file: Path = Path(__file__).parent / "samples" / "simple.pdf"
 
@@ -116,6 +117,27 @@ class TestMakeIndex:
         call_command("document_index", "reindex", skip_checks=True)
         mock_get_backend.return_value.rebuild.assert_called_once()
 
+    def test_reindex_skips_versions(self, mocker: MockerFixture) -> None:
+        """
+        GIVEN:
+            - A root document with a version
+        WHEN:
+            - The reindex command runs
+        THEN:
+            - Only the root document is handed to the rebuild, since a version
+              is indexed as its root
+        """
+        root = DocumentFactory()
+        DocumentFactory(root_document=root, version_index=1)
+        mock_get_backend = mocker.patch(
+            "documents.management.commands.document_index.get_backend",
+        )
+
+        call_command("document_index", "reindex", skip_checks=True)
+
+        documents = mock_get_backend.return_value.rebuild.call_args.args[0]
+        assert list(documents.values_list("pk", flat=True)) == [root.pk]
+
     def test_optimize(self) -> None:
         """Optimize command must execute without error (Tantivy handles optimization automatically)."""
         call_command("document_index", "optimize", skip_checks=True)

src/documents/tests/test_pdf_ops.py

@@ -1,520 +0,0 @@
-"""
-Tests for documents.pdf_ops.
-
-These use real PDFs from the sample directories. No database, Celery or mocks.
-Pages are compared by a hash of their content stream, so page identity and order
-are easy to assert.
-"""
-
-import hashlib
-from collections.abc import Callable
-from pathlib import Path
-
-import pikepdf
-import pytest
-
-from documents import pdf_ops
-from documents.pdf_ops import PageSpec
-
-SRC_ROOT = Path(__file__).parents[2]
-SAMPLES = Path(__file__).parent / "samples"
-THREE_PAGES = SAMPLES / "documents" / "originals" / "0000002.pdf"
-TWELVE_PAGES = SAMPLES / "barcodes" / "split-by-asn-2.pdf"
-ENCRYPTED = SAMPLES / "password-is-test.pdf"
-SIGNED = SRC_ROOT / "paperless" / "tests" / "samples" / "tesseract" / "signed.pdf"
-
-
-def _page_fingerprint(page: pikepdf.Page) -> str:
-    contents = page.obj.get("/Contents")
-    assert contents is not None, "sample page has no /Contents"
-    streams = list(contents) if isinstance(contents, pikepdf.Array) else [contents]
-    return hashlib.sha256(b"".join(s.read_bytes() for s in streams)).hexdigest()
-
-
-def fingerprints(path: Path) -> list[str]:
-    with pikepdf.open(path) as pdf:
-        return [_page_fingerprint(page) for page in pdf.pages]
-
-
-def rotations(path: Path) -> list[int]:
-    with pikepdf.open(path) as pdf:
-        return [int(page.obj.get("/Rotate", 0)) for page in pdf.pages]
-
-
-def docinfo_keys(path: Path) -> set[str]:
-    with pikepdf.open(path) as pdf:
-        return set(pdf.docinfo.keys())
-
-
-def constant(path: Path) -> Callable[[], Path]:
-    return lambda: path
-
-
-@pytest.fixture
-def source_fingerprints() -> list[str]:
-    fps = fingerprints(THREE_PAGES)
-    assert len(set(fps)) == 3, "sample must have three distinct pages"
-    return fps
-
-
-class TestRotatePdf:
-    def test_rotation_is_relative_and_applies_to_every_page(
-        self,
-        tmp_path: Path,
-    ) -> None:
-        """
-        GIVEN:
-            - A three page PDF
-        WHEN:
-            - It is rotated by 90 degrees, then the result is rotated by 90 again
-        THEN:
-            - Every page is rotated relative to its current rotation
-            - The page content itself is unchanged
-        """
-        once = tmp_path / "once.pdf"
-        twice = tmp_path / "twice.pdf"
-
-        pdf_ops.rotate_pdf(THREE_PAGES, once, 90)
-        pdf_ops.rotate_pdf(once, twice, 90)
-
-        assert rotations(once) == [90, 90, 90]
-        assert rotations(twice) == [180, 180, 180]
-        assert fingerprints(twice) == fingerprints(THREE_PAGES)
-
-    def test_keeps_document_info(self, tmp_path: Path) -> None:
-        """
-        GIVEN:
-            - A PDF with document info
-        WHEN:
-            - It is rotated
-        THEN:
-            - The document info is still present in the output
-        """
-        dst = tmp_path / "out.pdf"
-
-        pdf_ops.rotate_pdf(THREE_PAGES, dst, 90)
-
-        assert "/Creator" in docinfo_keys(dst)
-
-
-class TestRemovePages:
-    @pytest.mark.parametrize(
-        ("pages", "kept"),
-        [
-            pytest.param([2], [0, 2], id="single"),
-            pytest.param([3, 1], [1], id="unordered"),
-            pytest.param([2, 2], [0, 2], id="duplicates-remove-once"),
-            pytest.param([], [0, 1, 2], id="empty-keeps-everything"),
-            pytest.param([1, 2, 3], [], id="every-page"),
-        ],
-    )
-    def test_removes_only_the_requested_pages(
-        self,
-        tmp_path: Path,
-        source_fingerprints: list[str],
-        pages: list[int],
-        kept: list[int],
-    ) -> None:
-        """
-        GIVEN:
-            - A three page PDF
-        WHEN:
-            - Pages are removed, in any order and possibly repeated
-        THEN:
-            - Exactly the other pages remain, in their original order
-        """
-        dst = tmp_path / "out.pdf"
-
-        pdf_ops.remove_pages(THREE_PAGES, dst, pages)
-
-        assert fingerprints(dst) == [source_fingerprints[i] for i in kept]
-
-    def test_keeps_document_info(self, tmp_path: Path) -> None:
-        """
-        GIVEN:
-            - A PDF with document info
-        WHEN:
-            - A page is removed
-        THEN:
-            - The document info is still present in the output
-        """
-        dst = tmp_path / "out.pdf"
-
-        pdf_ops.remove_pages(THREE_PAGES, dst, [1])
-
-        assert "/Creator" in docinfo_keys(dst)
-
-    @pytest.mark.parametrize(
-        "bad_page",
-        [
-            pytest.param(0, id="zero"),
-            pytest.param(-1, id="negative"),
-        ],
-    )
-    def test_rejects_pages_below_one(self, tmp_path: Path, bad_page: int) -> None:
-        """
-        GIVEN:
-            - A three page PDF
-        WHEN:
-            - Pages are removed and one of them is below 1
-        THEN:
-            - ValueError is raised
-            - No output file is written
-        """
-        dst = tmp_path / "out.pdf"
-
-        with pytest.raises(ValueError, match="start at 1"):
-            pdf_ops.remove_pages(THREE_PAGES, dst, [1, bad_page])
-
-        assert not dst.exists()
-
-
-class TestBuildPdfs:
-    def test_selects_and_orders_pages(
-        self,
-        tmp_path: Path,
-        source_fingerprints: list[str],
-    ) -> None:
-        """
-        GIVEN:
-            - A three page PDF
-        WHEN:
-            - One output is built from pages 3 then 1
-        THEN:
-            - The output has those pages in that order
-            - Its path is returned
-        """
-        dst = tmp_path / "out.pdf"
-
-        written = pdf_ops.build_pdfs(
-            THREE_PAGES,
-            [([PageSpec(3), PageSpec(1)], constant(dst))],
-        )
-
-        assert written == [dst]
-        assert fingerprints(dst) == [source_fingerprints[2], source_fingerprints[0]]
-
-    def test_rotates_only_the_requested_pages(self, tmp_path: Path) -> None:
-        """
-        GIVEN:
-            - A three page PDF
-        WHEN:
-            - One output is built with a different rotation on each page
-        THEN:
-            - Each output page has exactly the rotation requested for it
-        """
-        dst = tmp_path / "out.pdf"
-
-        pdf_ops.build_pdfs(
-            THREE_PAGES,
-            [([PageSpec(1), PageSpec(2, 90), PageSpec(3, 180)], constant(dst))],
-        )
-
-        assert rotations(dst) == [0, 90, 180]
-
-    def test_writes_one_file_per_output_in_order(self, tmp_path: Path) -> None:
-        """
-        GIVEN:
-            - A twelve page PDF
-        WHEN:
-            - Two outputs are built from different page ranges
-        THEN:
-            - Two files are written and returned in output order
-            - Each holds exactly its own pages
-        """
-        first = tmp_path / "first.pdf"
-        second = tmp_path / "second.pdf"
-        source = fingerprints(TWELVE_PAGES)
-
-        written = pdf_ops.build_pdfs(
-            TWELVE_PAGES,
-            [
-                ([PageSpec(p) for p in (1, 2, 3)], constant(first)),
-                ([PageSpec(p) for p in range(4, 13)], constant(second)),
-            ],
-        )
-
-        assert written == [first, second]
-        assert fingerprints(first) == source[:3]
-        assert fingerprints(second) == source[3:]
-
-    def test_empty_page_list_writes_a_zero_page_file(self, tmp_path: Path) -> None:
-        """
-        GIVEN:
-            - A three page PDF
-        WHEN:
-            - An output with no pages is built
-        THEN:
-            - A PDF with zero pages is written
-        """
-        dst = tmp_path / "out.pdf"
-
-        pdf_ops.build_pdfs(THREE_PAGES, [([], constant(dst))])
-
-        assert fingerprints(dst) == []
-
-    def test_destination_is_not_requested_when_a_page_is_out_of_range(
-        self,
-        tmp_path: Path,
-    ) -> None:
-        """
-        GIVEN:
-            - A three page PDF
-        WHEN:
-            - Several outputs are built
-            - A later output refers to a page past the end
-        THEN:
-            - IndexError is raised
-            - No destination was requested for any output, including earlier valid ones
-        """
-        requested: list[Path] = []
-
-        def make_dst() -> Path:
-            requested.append(tmp_path / "out.pdf")
-            return requested[-1]
-
-        with pytest.raises(IndexError):
-            pdf_ops.build_pdfs(
-                THREE_PAGES,
-                [([PageSpec(1)], make_dst), ([PageSpec(99)], make_dst)],
-            )
-
-        assert requested == []
-
-    @pytest.mark.parametrize(
-        "bad_page",
-        [
-            pytest.param(0, id="zero"),
-            pytest.param(-1, id="negative"),
-        ],
-    )
-    def test_rejects_pages_below_one_before_opening_anything(
-        self,
-        tmp_path: Path,
-        bad_page: int,
-    ) -> None:
-        """
-        GIVEN:
-            - A three page PDF
-        WHEN:
-            - Several outputs are built and a later one has a page below 1
-        THEN:
-            - ValueError is raised
-            - No destination was requested for any output
-        """
-        requested: list[Path] = []
-
-        def make_dst() -> Path:
-            requested.append(tmp_path / "out.pdf")
-            return requested[-1]
-
-        with pytest.raises(ValueError, match="start at 1"):
-            pdf_ops.build_pdfs(
-                THREE_PAGES,
-                [([PageSpec(1)], make_dst), ([PageSpec(bad_page)], make_dst)],
-            )
-
-        assert requested == []
-
-
-class TestValidatePageOperations:
-    def test_returns_the_output_count(self) -> None:
-        """
-        GIVEN:
-            - Operations that all target the default output
-        WHEN:
-            - They are validated
-        THEN:
-            - One output document is reported
-        """
-        operations = [{"page": 1}, {"page": 2}, {"page": 3}]
-
-        assert pdf_ops.validate_page_operations(operations, single_output=True) == 1
-
-    def test_gap_in_output_indices_counts_up_to_the_highest(self) -> None:
-        """
-        GIVEN:
-            - Operations that target outputs 0 and 2 but never 1
-        WHEN:
-            - They are validated
-        THEN:
-            - Three output documents are reported
-        """
-        operations = [
-            {"page": 1, "doc": 0},
-            {"page": 2, "doc": 2},
-            {"page": 3, "doc": 0},
-        ]
-
-        count = pdf_ops.validate_page_operations(operations, single_output=False)
-
-        assert count == 3
-
-    def test_empty_operations_are_rejected(self) -> None:
-        """
-        GIVEN:
-            - No operations
-        WHEN:
-            - They are validated
-        THEN:
-            - ValueError is raised
-        """
-        with pytest.raises(ValueError, match="index is out of bounds"):
-            pdf_ops.validate_page_operations([], single_output=False)
-
-    def test_multiple_outputs_rejected_when_single_output_required(self) -> None:
-        """
-        GIVEN:
-            - Operations that target two outputs
-        WHEN:
-            - They are validated with a single output required
-        THEN:
-            - ValueError is raised
-        """
-        operations = [{"page": 1, "doc": 0}, {"page": 2, "doc": 1}]
-
-        with pytest.raises(ValueError, match="Multiple output documents"):
-            pdf_ops.validate_page_operations(operations, single_output=True)
-
-    @pytest.mark.parametrize(
-        "doc",
-        [
-            pytest.param(-1, id="negative"),
-            pytest.param(2, id="equal-to-operation-count"),
-            pytest.param(2**32, id="huge"),
-        ],
-    )
-    def test_output_index_out_of_bounds(self, doc: int) -> None:
-        """
-        GIVEN:
-            - Two operations, one with an output index that is out of bounds
-        WHEN:
-            - They are validated
-        THEN:
-            - ValueError is raised
-        """
-        operations = [{"page": 1, "doc": 0}, {"page": 2, "doc": doc}]
-
-        with pytest.raises(ValueError, match="index is out of bounds"):
-            pdf_ops.validate_page_operations(operations, single_output=False)
-
-
-class TestDecrypt:
-    @pytest.mark.parametrize(
-        ("path", "expected"),
-        [
-            pytest.param(ENCRYPTED, True, id="password-required"),
-            pytest.param(SIGNED, True, id="opens-without-password-but-encrypted"),
-            pytest.param(THREE_PAGES, False, id="not-encrypted"),
-        ],
-    )
-    def test_needs_decrypt(self, path: Path, *, expected: bool) -> None:
-        """
-        GIVEN:
-            - A PDF that is encrypted, or encrypted but openable, or plain
-        WHEN:
-            - needs_decrypt is asked about it
-        THEN:
-            - Only the unencrypted PDF reports False
-        """
-        assert pdf_ops.needs_decrypt(path) is expected
-
-    def test_decrypt_writes_an_unencrypted_copy(self, tmp_path: Path) -> None:
-        """
-        GIVEN:
-            - A password protected PDF
-        WHEN:
-            - It is decrypted with the correct password
-        THEN:
-            - The path from make_dst is returned
-            - The written copy no longer needs decrypting
-        """
-        dst = tmp_path / "out.pdf"
-
-        result = pdf_ops.decrypt_pdf(ENCRYPTED, constant(dst), "test")
-
-        assert result == dst
-        assert pdf_ops.needs_decrypt(dst) is False
-
-    def test_wrong_password_raises_and_never_requests_a_destination(
-        self,
-        tmp_path: Path,
-    ) -> None:
-        """
-        GIVEN:
-            - A password protected PDF
-        WHEN:
-            - It is decrypted with the wrong password
-        THEN:
-            - PasswordError is raised
-            - No destination was requested
-        """
-        requested: list[Path] = []
-
-        def make_dst() -> Path:
-            requested.append(tmp_path / "out.pdf")
-            return requested[-1]
-
-        with pytest.raises(pikepdf.PasswordError):
-            pdf_ops.decrypt_pdf(ENCRYPTED, make_dst, "wrong")
-
-        assert requested == []
-
-
-class TestPdfMerger:
-    def test_pages_are_appended_in_the_order_added(
-        self,
-        tmp_path: Path,
-        source_fingerprints: list[str],
-    ) -> None:
-        """
-        GIVEN:
-            - A reordered PDF and the original three page PDF
-        WHEN:
-            - Both are added to a merger in that order and saved
-        THEN:
-            - The output holds all pages in the order they were added
-        """
-        reordered = tmp_path / "reordered.pdf"
-        merged = tmp_path / "merged.pdf"
-        pdf_ops.build_pdfs(
-            THREE_PAGES,
-            [([PageSpec(3), PageSpec(1)], constant(reordered))],
-        )
-
-        with pdf_ops.PdfMerger() as merger:
-            merger.add(reordered)
-            merger.add(THREE_PAGES)
-            merger.save(merged)
-
-        assert fingerprints(merged) == [
-            source_fingerprints[2],
-            source_fingerprints[0],
-            *source_fingerprints,
-        ]
-
-    def test_output_version_is_at_least_the_highest_source_version(
-        self,
-        tmp_path: Path,
-    ) -> None:
-        """
-        GIVEN:
-            - Two PDFs with different PDF versions
-        WHEN:
-            - Both are added to a merger and saved
-        THEN:
-            - The output version is at least the highest source version
-        """
-        merged = tmp_path / "merged.pdf"
-        with pikepdf.open(TWELVE_PAGES) as pdf:
-            source_versions = [pdf.pdf_version]
-        with pikepdf.open(THREE_PAGES) as pdf:
-            source_versions.append(pdf.pdf_version)
-
-        with pdf_ops.PdfMerger() as merger:
-            merger.add(TWELVE_PAGES)
-            merger.add(THREE_PAGES)
-            merger.save(merged)
-
-        with pikepdf.open(merged) as pdf:
-            assert pdf.pdf_version >= max(source_versions)

src/documents/tests/test_permission_filtering_security.py

@@ -18,6 +18,7 @@ from documents.models import Correspondent
 from documents.models import DocumentType
 from documents.models import StoragePath
 from documents.models import Tag
+from documents.permissions import has_perms_owner_aware
 from documents.permissions import permitted_document_ids
 from documents.permissions import permitted_object_ids
 from documents.permissions import restrict_queryset_to_visible
@@ -32,6 +33,8 @@ from paperless_testing.permissions import grant_global
 from paperless_testing.permissions import grant_object
 
 if TYPE_CHECKING:
+    from django.contrib.auth.models import User
+
     from paperless_testing.dirs import PaperlessDirs
 
 
@@ -178,6 +181,382 @@ class TestPermittedDocumentIdsIncludeDeleted:
         )
 
 
+@pytest.mark.django_db
+class TestPermittedDocumentIdsVersions:
+    """
+    A version is authorized by its root document: the version's own owner and
+    grants never matter.
+    """
+
+    @pytest.mark.parametrize(
+        ("root_owner", "version_owner", "expected_visible"),
+        [
+            pytest.param(
+                "other",
+                "nobody",
+                False,
+                id="unowned-version-of-private-root",
+            ),
+            pytest.param("other", "user", False, id="own-version-of-private-root"),
+            pytest.param("user", "other", True, id="foreign-version-of-own-root"),
+            pytest.param("user", "nobody", True, id="unowned-version-of-own-root"),
+            pytest.param("nobody", "other", True, id="private-version-of-unowned-root"),
+        ],
+    )
+    def test_version_follows_root_owner(
+        self,
+        root_owner: str,
+        version_owner: str,
+        *,
+        expected_visible: bool,
+    ) -> None:
+        """
+        GIVEN:
+            - A root document and a version with differing owners
+        WHEN:
+            - The permitted document ids are resolved for the user
+        THEN:
+            - The version is visible exactly when its root is
+        """
+        user = UserFactory()
+        owners = {"user": user, "other": UserFactory(), "nobody": None}
+        root = DocumentFactory(owner=owners[root_owner])
+        version = DocumentFactory(root_document=root, owner=owners[version_owner])
+
+        visible = set(permitted_document_ids(user))
+
+        assert (version.pk in visible) is expected_visible
+        assert (root.pk in visible) is expected_visible
+
+    @staticmethod
+    def grantee(user: User, kind: str) -> User | Group:
+        """The user itself, or a new group the user belongs to."""
+        if kind == "user":
+            return user
+        group = Group.objects.create(name="shared")
+        user.groups.add(group)
+        return group
+
+    @pytest.mark.parametrize(
+        "grantee_kind",
+        [pytest.param("user", id="user"), pytest.param("group", id="group")],
+    )
+    def test_grant_on_root_applies_to_version(self, grantee_kind: str) -> None:
+        """
+        GIVEN:
+            - A private root document shared with a user or one of their groups
+            - A version of it owned by someone else
+        WHEN:
+            - The permitted document ids are resolved for the user
+        THEN:
+            - Both the root and the version are visible
+            - A user without the grant sees neither
+        """
+        user = UserFactory()
+        stranger = UserFactory()
+        root = DocumentFactory(owner=UserFactory())
+        version = DocumentFactory(root_document=root, owner=UserFactory())
+        grant_object(self.grantee(user, grantee_kind), root, "view_document")
+
+        assert_visible_document_ids(
+            permitted_document_ids(user),
+            expected_visible=[root.pk, version.pk],
+            expected_hidden=[],
+        )
+        assert_visible_document_ids(
+            permitted_document_ids(stranger),
+            expected_visible=[],
+            expected_hidden=[root.pk, version.pk],
+        )
+
+    @pytest.mark.parametrize(
+        "grantee_kind",
+        [pytest.param("user", id="user"), pytest.param("group", id="group")],
+    )
+    def test_grant_on_version_is_ignored(self, grantee_kind: str) -> None:
+        """
+        GIVEN:
+            - A private root document
+            - A version with an explicit grant for the user or one of their groups
+        WHEN:
+            - The permitted document ids are resolved for the user
+        THEN:
+            - Neither the root nor the version is visible
+        """
+        user = UserFactory()
+        root = DocumentFactory(owner=UserFactory())
+        version = DocumentFactory(root_document=root, owner=UserFactory())
+        grant_object(self.grantee(user, grantee_kind), version, "view_document")
+
+        assert_visible_document_ids(
+            permitted_document_ids(user),
+            expected_visible=[],
+            expected_hidden=[root.pk, version.pk],
+        )
+
+    def test_grant_on_one_root_does_not_reach_another_roots_version(self) -> None:
+        """
+        GIVEN:
+            - Two private roots, each with a version
+            - The user may view only the first root
+        WHEN:
+            - The permitted document ids are resolved for the user
+        THEN:
+            - Only the first root and its version are visible
+        """
+        user = UserFactory()
+        first = DocumentFactory(owner=UserFactory())
+        first_version = DocumentFactory(root_document=first, owner=UserFactory())
+        second = DocumentFactory(owner=UserFactory())
+        second_version = DocumentFactory(root_document=second, owner=user)
+        grant_object(user, first, "view_document")
+
+        assert_visible_document_ids(
+            permitted_document_ids(user),
+            expected_visible=[first.pk, first_version.pk],
+            expected_hidden=[second.pk, second_version.pk],
+        )
+
+    def test_user_in_several_groups(self) -> None:
+        """
+        GIVEN:
+            - A user in two groups
+            - Two private roots shared with one group each, and a third shared with nobody
+            - A version of each root
+        WHEN:
+            - The permitted document ids are resolved for the user
+        THEN:
+            - The two shared roots and their versions are visible
+            - The third root and its version are not
+        """
+        user = UserFactory()
+        groups = [Group.objects.create(name=f"group{i}") for i in range(2)]
+        user.groups.add(*groups)
+        shared = [DocumentFactory(owner=UserFactory()) for _ in groups]
+        for root, group in zip(shared, groups, strict=True):
+            grant_object(group, root, "view_document")
+        unshared = DocumentFactory(owner=UserFactory())
+        shared_versions = [
+            DocumentFactory(root_document=root, owner=UserFactory()) for root in shared
+        ]
+        unshared_version = DocumentFactory(root_document=unshared, owner=None)
+
+        assert_visible_document_ids(
+            permitted_document_ids(user),
+            expected_visible=[
+                *(root.pk for root in shared),
+                *(version.pk for version in shared_versions),
+            ],
+            expected_hidden=[unshared.pk, unshared_version.pk],
+        )
+
+    def test_permission_is_resolved_through_the_root(self) -> None:
+        """
+        GIVEN:
+            - A private root document where the user may view and change
+        WHEN:
+            - The permitted ids are resolved for view, change and delete
+        THEN:
+            - The version is visible for view and change only
+        """
+        user = UserFactory()
+        root = DocumentFactory(owner=UserFactory())
+        version = DocumentFactory(root_document=root, owner=UserFactory())
+        grant_object(user, root, "view_document", "change_document")
+
+        assert version.pk in set(permitted_document_ids(user))
+        assert version.pk in set(permitted_document_ids(user, perm="change_document"))
+        assert version.pk in set(
+            permitted_document_ids(user, perm="documents.change_document"),
+        )
+        assert version.pk not in set(
+            permitted_document_ids(user, perm="delete_document"),
+        )
+
+    def test_anonymous_sees_versions_of_unowned_roots_only(self) -> None:
+        """
+        GIVEN:
+            - A version owned by nobody under a private root
+            - A version owned by someone under an unowned root
+        WHEN:
+            - The permitted document ids are resolved for an anonymous user
+        THEN:
+            - Only the version of the unowned root is visible
+        """
+        private_root = DocumentFactory(owner=UserFactory())
+        private_version = DocumentFactory(root_document=private_root, owner=None)
+        open_root = DocumentFactory(owner=None)
+        open_version = DocumentFactory(root_document=open_root, owner=UserFactory())
+
+        assert_visible_document_ids(
+            permitted_document_ids(AnonymousUser()),
+            expected_visible=[open_root.pk, open_version.pk],
+            expected_hidden=[private_root.pk, private_version.pk],
+        )
+
+    def test_deleted_versions_follow_their_deleted_root(self) -> None:
+        """
+        GIVEN:
+            - A soft-deleted root document and its version, which deleting the
+              root soft-deletes too; the version is owned by someone else
+        WHEN:
+            - The permitted document ids are resolved with and without deleted
+              documents
+        THEN:
+            - Nothing is visible by default
+            - With deleted documents included, the version is visible to the
+              root's owner and not to the version's own owner
+        """
+        owner = UserFactory()
+        version_owner = UserFactory()
+        root = DocumentFactory(owner=owner)
+        version = DocumentFactory(root_document=root, owner=version_owner)
+        root.delete()
+
+        assert not {root.pk, version.pk} & set(permitted_document_ids(owner))
+        assert_visible_document_ids(
+            permitted_document_ids(owner, include_deleted=True),
+            expected_visible=[root.pk, version.pk],
+            expected_hidden=[],
+        )
+        assert_visible_document_ids(
+            permitted_document_ids(version_owner, include_deleted=True),
+            expected_visible=[],
+            expected_hidden=[root.pk, version.pk],
+        )
+
+    @pytest.mark.parametrize(
+        "is_superuser",
+        [
+            pytest.param(False, id="regular-user"),
+            pytest.param(True, id="superuser"),
+        ],
+    )
+    def test_inactive_user_sees_no_versions(self, *, is_superuser: bool) -> None:
+        """
+        GIVEN:
+            - An inactive user, possibly a superuser, who owns a root and its version
+        WHEN:
+            - The permitted document ids are resolved for them
+        THEN:
+            - Nothing is visible
+        """
+        user = UserFactory(is_active=False, is_superuser=is_superuser)
+        root = DocumentFactory(owner=user)
+        version = DocumentFactory(root_document=root, owner=user)
+
+        assert_visible_document_ids(
+            permitted_document_ids(user),
+            expected_visible=[],
+            expected_hidden=[root.pk, version.pk],
+        )
+
+    def test_superuser_sees_all_versions(self) -> None:
+        """
+        GIVEN:
+            - A private root owned by someone else, with a version
+        WHEN:
+            - The permitted document ids are resolved for a superuser
+        THEN:
+            - Both the root and the version are visible
+        """
+        superuser = UserFactory(superuser=True)
+        root = DocumentFactory(owner=UserFactory())
+        version = DocumentFactory(root_document=root, owner=UserFactory())
+
+        assert_visible_document_ids(
+            permitted_document_ids(superuser),
+            expected_visible=[root.pk, version.pk],
+            expected_hidden=[],
+        )
+
+
+@pytest.mark.django_db
+class TestHasPermsOwnerAwareVersions:
+    """
+    The single-object check agrees with permitted_document_ids: a version is
+    authorized by its root document.
+    """
+
+    @pytest.mark.parametrize(
+        ("root_owner", "version_owner", "expected"),
+        [
+            pytest.param(
+                "other",
+                "nobody",
+                False,
+                id="unowned-version-of-private-root",
+            ),
+            pytest.param("other", "user", False, id="own-version-of-private-root"),
+            pytest.param("user", "other", True, id="foreign-version-of-own-root"),
+            pytest.param("nobody", "other", True, id="private-version-of-unowned-root"),
+        ],
+    )
+    def test_version_follows_root_owner(
+        self,
+        root_owner: str,
+        version_owner: str,
+        *,
+        expected: bool,
+    ) -> None:
+        """
+        GIVEN:
+            - A root document and a version with differing owners
+        WHEN:
+            - The single-object check runs for the version
+        THEN:
+            - The version is allowed exactly when its root is
+        """
+        user = UserFactory()
+        owners = {"user": user, "other": UserFactory(), "nobody": None}
+        root = DocumentFactory(owner=owners[root_owner])
+        version = DocumentFactory(root_document=root, owner=owners[version_owner])
+
+        assert has_perms_owner_aware(user, "view_document", version) is expected
+        assert has_perms_owner_aware(user, "view_document", root) is expected
+
+    def test_grant_on_root_applies_and_grant_on_version_does_not(self) -> None:
+        """
+        GIVEN:
+            - A private root with a version, and a second private root with a version
+            - The user may change only the first root, and was granted the second
+              root's version directly
+        WHEN:
+            - The single-object check runs for each version
+        THEN:
+            - Only the first root's version is allowed
+        """
+        user = UserFactory()
+        shared_root = DocumentFactory(owner=UserFactory())
+        shared_version = DocumentFactory(root_document=shared_root, owner=UserFactory())
+        private_root = DocumentFactory(owner=UserFactory())
+        private_version = DocumentFactory(
+            root_document=private_root,
+            owner=UserFactory(),
+        )
+        grant_object(user, shared_root, "change_document")
+        grant_object(user, private_version, "change_document")
+
+        assert has_perms_owner_aware(user, "change_document", shared_version)
+        assert not has_perms_owner_aware(user, "change_document", private_version)
+
+    def test_other_models_use_their_own_owner(self) -> None:
+        """
+        GIVEN:
+            - A tag owned by someone else, and one owned by the user
+        WHEN:
+            - The single-object check runs for each
+        THEN:
+            - Only the user's own tag is allowed without a grant
+        """
+        user = UserFactory()
+        mine = TagFactory(owner=user)
+        theirs = TagFactory(owner=UserFactory())
+
+        assert has_perms_owner_aware(user, "view_tag", mine)
+        assert not has_perms_owner_aware(user, "view_tag", theirs)
+
+
 @pytest.mark.django_db
 class TestAiChatAllDocumentsPermissionBoundary:
     """

src/documents/tests/test_share_link_bundles.py

@@ -90,10 +90,13 @@ class ShareLinkBundleAPITests(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
         self.assertIn("document_ids", response.data)
 
-    @mock.patch("documents.views.permitted_document_ids", return_value=set())
-    def test_create_bundle_rejects_insufficient_permissions(self, perms_mock) -> None:
+    def test_create_bundle_rejects_insufficient_permissions(self) -> None:
+        requester = UserFactory(username="bundle_creator")
+        grant_global(requester, "add_sharelinkbundle", "view_document")
+        self.client.force_authenticate(requester)
+        document = DocumentFactory(owner=UserFactory(username="document_owner"))
         payload = {
-            "document_ids": [self.document.pk],
+            "document_ids": [self.document.pk, document.pk],
             "file_version": ShareLink.FileVersion.ARCHIVE,
             "expiration_days": 7,
         }
@@ -101,8 +104,8 @@ class ShareLinkBundleAPITests(DirectoriesMixin, APITestCase):
         response = self.client.post(self.ENDPOINT, payload, format="json")
 
         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
-        self.assertIn("document_ids", response.data)
-        perms_mock.assert_called()
+        self.assertIn(str(document.pk), str(response.data["document_ids"]))
+        self.assertFalse(ShareLinkBundle.objects.exists())
 
     @mock.patch("documents.views.build_share_link_bundle.apply_async")
     def test_rebuild_bundle_resets_state(self, delay_mock) -> None:

src/documents/tests/test_tasks.py

@@ -20,6 +20,7 @@ from documents.sanity_checker import SanityCheckMessages
 from documents.tests.helpers import dummy_preprocess
 from paperless_testing.assertions import FileSystemAssertsMixin
 from paperless_testing.dirs import DirectoriesMixin
+from paperless_testing.factories import DocumentFactory
 
 
 @pytest.mark.django_db
@@ -287,6 +288,80 @@ class TestUpdateContent(DirectoriesMixin, TestCase):
         tasks.update_document_content_maybe_archive_file(doc.pk)
         self.assertNotEqual(Document.objects.get(pk=doc.pk).content, "test")
 
+    def _create_root_with_version(self) -> tuple[Document, Document]:
+        sample1 = self.dirs.scratch_dir / "sample.pdf"
+        shutil.copy(
+            Path(__file__).parent
+            / "samples"
+            / "documents"
+            / "originals"
+            / "0000001.pdf",
+            sample1,
+        )
+        root = DocumentFactory(content="root content", mime_type="application/pdf")
+        version = DocumentFactory(
+            content="my document",
+            filename=sample1,
+            mime_type="application/pdf",
+            root_document=root,
+            version_index=1,
+        )
+        return root, version
+
+    @mock.patch("documents.tasks.clear_document_caches")
+    @mock.patch("documents.search.get_backend")
+    def test_update_content_version_clears_caches_for_root(
+        self,
+        mock_get_backend: mock.Mock,
+        mock_clear_caches: mock.Mock,
+    ) -> None:
+        """
+        GIVEN:
+            - A root document with a version
+        WHEN:
+            - Update content task is called for the version
+        THEN:
+            - The version's content is updated, not the root's
+            - The document is indexed
+            - Caches are cleared for both
+        """
+        root, version = self._create_root_with_version()
+
+        tasks.update_document_content_maybe_archive_file(version.pk)
+
+        self.assertNotEqual(
+            Document.objects.get(pk=version.pk).content,
+            "my document",
+        )
+        self.assertEqual(Document.objects.get(pk=root.pk).content, "root content")
+        mock_get_backend.return_value.add_or_update.assert_called_once()
+        mock_clear_caches.assert_has_calls(
+            [mock.call(version.pk), mock.call(root.pk)],
+        )
+
+    @override_settings(AI_ENABLED=True, LLM_EMBEDDING_BACKEND="huggingface")
+    @mock.patch("documents.tasks.llm_index_add_or_update_document")
+    @mock.patch("documents.search.get_backend")
+    def test_update_content_version_updates_llm_index(
+        self,
+        mock_get_backend: mock.Mock,
+        mock_llm_index: mock.Mock,
+    ) -> None:
+        """
+        GIVEN:
+            - A root document with a version
+            - The LLM index is enabled
+        WHEN:
+            - Update content task is called for the version
+        THEN:
+            - The LLM index is updated
+        """
+        _, version = self._create_root_with_version()
+
+        tasks.update_document_content_maybe_archive_file(version.pk)
+
+        mock_llm_index.assert_called_once()
+
 
 class TestUpdateContentRemoteOCR(DirectoriesMixin, TestCase):
     """

src/documents/versioning.py

@@ -17,9 +17,26 @@ from django.db.models.functions import RowNumber
 from documents.models import Document
 
 if TYPE_CHECKING:
+    from collections.abc import Iterable
+
     from rest_framework.request import Request
 
 
+def root_document_ids(ids: Iterable[int]) -> QuerySet[int]:
+    """
+    The ids of the root documents of the given documents: a root stands for
+    itself and a version for its root. Only the indexes' bookkeeping needs
+    this, since they hold root documents only.
+    """
+    return (
+        Document.objects.filter(pk__in=ids)
+        .annotate(root_id=Coalesce("root_document_id", "id"))
+        .order_by()
+        .values_list("root_id", flat=True)
+        .distinct()
+    )
+
+
 def versions_newest_first(documents: QuerySet[Document]) -> QuerySet[Document]:
     """
     Sorts versions so the newest one comes first using version_index and not on id,

src/documents/views.py

@@ -329,9 +329,10 @@ def _get_tantivy_query_and_mode(params):
 def _get_more_like_id(query_params: dict[str, Any], user: User | None) -> int:
     try:
         more_like_doc_id = int(query_params["more_like_id"])
-        more_like_doc = Document.objects.select_related("owner").get(
-            pk=more_like_doc_id,
-        )
+        more_like_doc = Document.objects.select_related(
+            "owner",
+            "root_document__owner",
+        ).get(pk=more_like_doc_id)
     except (TypeError, ValueError, Document.DoesNotExist):
         raise PermissionDenied(_("Invalid more_like_id"))
 
@@ -342,7 +343,8 @@ def _get_more_like_id(query_params: dict[str, Any], user: User | None) -> int:
     ):
         raise PermissionDenied(_("Insufficient permissions."))
 
-    return more_like_doc_id
+    # Only root documents are indexed, a version stands for its root
+    return more_like_doc.root_document_id or more_like_doc.pk
 
 
 class SearchParams(NamedTuple):
@@ -1550,7 +1552,10 @@ class DocumentViewSet(
     )
     def suggestions(self, request, pk=None):
         doc = get_object_or_404(
-            Document.objects.select_related("owner").prefetch_related("versions"),
+            Document.objects.select_related(
+                "owner",
+                "root_document__owner",
+            ).prefetch_related("versions"),
             pk=pk,
         )
         if request.user is not None and not has_perms_owner_aware(
@@ -1610,7 +1615,10 @@ class DocumentViewSet(
     @method_decorator(cache_control(no_cache=True))
     def ai_suggestions(self, request, pk=None):
         doc = get_object_or_404(
-            Document.objects.select_related("owner").prefetch_related("versions"),
+            Document.objects.select_related(
+                "owner",
+                "root_document__owner",
+            ).prefetch_related("versions"),
             pk=pk,
         )
         if request.user is not None and not has_perms_owner_aware(
@@ -1856,9 +1864,14 @@ class DocumentViewSet(
         currentUser = request.user
         try:
             doc = (
-                Document.objects.select_related("owner")
+                Document.objects.select_related("owner", "root_document__owner")
                 .prefetch_related("notes")
-                .only("pk", "owner__id")
+                .only(
+                    "pk",
+                    "owner__id",
+                    "root_document__id",
+                    "root_document__owner__id",
+                )
                 .get(pk=pk)
             )
             if currentUser is not None and not has_perms_owner_aware(
@@ -1973,7 +1986,9 @@ class DocumentViewSet(
     def share_links(self, request, pk=None):
         currentUser = request.user
         try:
-            doc = Document.objects.select_related("owner").get(pk=pk)
+            doc = Document.objects.select_related("owner", "root_document__owner").get(
+                pk=pk,
+            )
             if currentUser is not None and not has_perms_owner_aware(
                 currentUser,
                 "change_document",
@@ -2008,10 +2023,11 @@ class DocumentViewSet(
         if not settings.AUDIT_LOG_ENABLED:
             return HttpResponseBadRequest("Audit log is disabled")
         try:
-            doc = Document.objects.get(pk=pk)
+            doc = Document.objects.select_related("root_document__owner").get(pk=pk)
+            root_doc = get_root_document(doc)
             if not request.user.has_perm("auditlog.view_logentry") or (
-                doc.owner is not None
-                and doc.owner != request.user
+                root_doc.owner is not None
+                and root_doc.owner != request.user
                 and not request.user.is_superuser
             ):
                 return HttpResponseForbidden(
@@ -2102,9 +2118,7 @@ class DocumentViewSet(
         documents = Document.objects.filter(pk__in=document_ids)
         if (
             request.user is not None
-            and documents.exclude(
-                pk__in=permitted_document_ids(request.user),
-            ).exists()
+            and documents.exclude(id__in=permitted_document_ids(request.user)).exists()
         ):
             return HttpResponseForbidden("Insufficient permissions")
 
@@ -2430,11 +2444,17 @@ class ChatStreamingView(GenericAPIView[Any]):
 
         if doc_id:
             try:
-                document = Document.objects.get(id=doc_id)
+                document = Document.objects.select_related(
+                    "root_document__owner",
+                ).get(id=doc_id)
             except Document.DoesNotExist:
                 return HttpResponseBadRequest("Document not found")
 
-            if not has_perms_owner_aware(request.user, "view_document", document):
+            if not has_perms_owner_aware(
+                request.user,
+                "view_document",
+                document,
+            ):
                 return HttpResponseForbidden("Insufficient permissions")
 
             documents = Document.objects.filter(pk=document.pk)
@@ -2990,12 +3010,8 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
             user.has_perm(
                 "documents.change_document",
             )
-            and not Document.global_objects.filter(
-                pk__in=[doc.pk for doc in root_docs],
-            )
-            .exclude(
-                pk__in=permitted_document_ids(user, perm="change_document"),
-            )
+            and not Document.global_objects.filter(pk__in=documents)
+            .exclude(pk__in=permitted_document_ids(user, perm="change_document"))
             .exists()
         )
 
@@ -3605,10 +3621,11 @@ class SelectionDataView(DocumentSelectionMixin, GenericAPIView[Any]):
             user=request.user,
             validated_data=serializer.validated_data,
         )
-        permitted_documents = Document.objects.filter(
-            id__in=permitted_document_ids(request.user),
-        )
-        if permitted_documents.filter(pk__in=ids).count() != len(ids):
+        documents = Document.objects.filter(pk__in=ids)
+        if (
+            documents.count() != len(ids)
+            or documents.exclude(id__in=permitted_document_ids(request.user)).exists()
+        ):
             return HttpResponseForbidden("Insufficient permissions")
 
         correspondents = Correspondent.objects.annotate(
@@ -4104,21 +4121,16 @@ class BulkDownloadView(DocumentSelectionMixin, GenericAPIView[Any]):
             validated_data=serializer.validated_data,
         )
         documents = Document.objects.filter(pk__in=ids)
-        versioned_documents = []
         compression = serializer.validated_data.get("compression")
         content = serializer.validated_data.get("content")
         follow_filename_format = serializer.validated_data.get("follow_formatting")
 
-        permitted_ids = set(permitted_document_ids(request.user))
-        for document in documents:
-            root_doc = get_root_document(document)
-            if root_doc.pk not in permitted_ids:
-                return HttpResponseForbidden("Insufficient permissions")
-            versioned_documents.append(
-                get_latest_version_for_root(
-                    root_doc,
-                ),
-            )
+        if documents.exclude(id__in=permitted_document_ids(request.user)).exists():
+            return HttpResponseForbidden("Insufficient permissions")
+        versioned_documents = [
+            get_latest_version_for_root(get_root_document(document))
+            for document in documents
+        ]
 
         if content == "both":
             strategy_class = OriginalAndArchiveStrategy
@@ -4790,19 +4802,23 @@ class ShareLinkBundleViewSet(PassUserMixin, ModelViewSet[ShareLinkBundle]):
                 },
             )
 
-        documents = list(documents_qs)
-        permitted_ids = set(permitted_document_ids(request.user))
-        for document in documents:
-            if document.pk not in permitted_ids:
-                raise ValidationError(
-                    {
-                        "document_ids": _(
-                            "Insufficient permissions to share document %(id)s.",
-                        )
-                        % {"id": document.pk},
-                    },
-                )
+        denied_id = (
+            documents_qs.exclude(id__in=permitted_document_ids(request.user))
+            .order_by("pk")
+            .values_list("pk", flat=True)
+            .first()
+        )
+        if denied_id is not None:
+            raise ValidationError(
+                {
+                    "document_ids": _(
+                        "Insufficient permissions to share document %(id)s.",
+                    )
+                    % {"id": denied_id},
+                },
+            )
 
+        documents = list(documents_qs)
         document_map = {document.pk: document for document in documents}
         ordered_documents = [document_map[doc_id] for doc_id in document_ids]
 
@@ -5587,6 +5603,23 @@ class TrashView(ListModelMixin, PassUserMixin):
     class _TrashPermittedObjectsFilter(PermittedObjectsFilter):
         include_granted = False
 
+        def filter_queryset(self, request, queryset, view):
+            if request.user.is_superuser or not request.user.is_active:
+                return super().filter_queryset(request, queryset, view)
+
+            # A version belongs to whoever owns its root
+            def owned_or_unowned(prefix: str) -> Q:
+                return Q(**{f"{prefix}owner": request.user}) | Q(
+                    **{f"{prefix}owner__isnull": True},
+                )
+
+            return queryset.filter(
+                (Q(root_document__isnull=True) & owned_or_unowned(""))
+                | (
+                    Q(root_document__isnull=False) & owned_or_unowned("root_document__")
+                ),
+            )
+
     filter_backends = (_TrashPermittedObjectsFilter,)
     pagination_class = StandardPagination
 
@@ -5617,7 +5650,7 @@ class TrashView(ListModelMixin, PassUserMixin):
             else self.filter_queryset(self.get_queryset()).all()
         )
         if docs.exclude(
-            pk__in=permitted_document_ids(
+            id__in=permitted_document_ids(
                 request.user,
                 perm="delete_document",
                 include_deleted=True,

src/locale/en_US/LC_MESSAGES/django.po

@@ -2,7 +2,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: paperless-ngx\n"
 "Report-Msgid-Bugs-To: \n"
-"POT-Creation-Date: 2026-10-06 15:12+0000\n"
+"POT-Creation-Date: 2026-10-08 18:39+0000\n"
 "PO-Revision-Date: 2022-02-17 04:17\n"
 "Last-Translator: \n"
 "Language-Team: English\n"
@@ -1652,49 +1652,49 @@ msgstr ""
 msgid "workflow runs"
 msgstr ""
 
-#: documents/serialisers.py:515 documents/serialisers.py:872
-#: documents/serialisers.py:2902 documents/views.py:343 documents/views.py:2732
+#: documents/serialisers.py:516 documents/serialisers.py:873
+#: documents/serialisers.py:2903 documents/views.py:344 documents/views.py:2751
 #: paperless_mail/serialisers.py:156
 msgid "Insufficient permissions."
 msgstr ""
 
-#: documents/serialisers.py:708
+#: documents/serialisers.py:709
 msgid "Invalid color."
 msgstr ""
 
-#: documents/serialisers.py:2369
+#: documents/serialisers.py:2370
 #, python-format
 msgid "File type %(type)s not supported"
 msgstr ""
 
-#: documents/serialisers.py:2413
+#: documents/serialisers.py:2414
 #, python-format
 msgid "Custom field id must be an integer: %(id)s"
 msgstr ""
 
-#: documents/serialisers.py:2420
+#: documents/serialisers.py:2421
 #, python-format
 msgid "Custom field with id %(id)s does not exist"
 msgstr ""
 
-#: documents/serialisers.py:2437 documents/serialisers.py:2447
+#: documents/serialisers.py:2438 documents/serialisers.py:2448
 msgid ""
 "Custom fields must be a list of integers or an object mapping ids to values."
 msgstr ""
 
-#: documents/serialisers.py:2442
+#: documents/serialisers.py:2443
 msgid "Some custom fields don't exist or were specified twice."
 msgstr ""
 
-#: documents/serialisers.py:2589
+#: documents/serialisers.py:2590
 msgid "Invalid variable detected."
 msgstr ""
 
-#: documents/serialisers.py:2958
+#: documents/serialisers.py:2959
 msgid "Duplicate document identifiers are not allowed."
 msgstr ""
 
-#: documents/serialisers.py:2988 documents/views.py:4787
+#: documents/serialisers.py:2989 documents/views.py:4807
 #, python-format
 msgid "Documents not found: %(ids)s"
 msgstr ""
@@ -1945,40 +1945,40 @@ msgstr ""
 msgid ", "
 msgstr ""
 
-#: documents/views.py:336 documents/views.py:2729
+#: documents/views.py:337 documents/views.py:2748
 msgid "Invalid more_like_id"
 msgstr ""
 
-#: documents/views.py:1676
+#: documents/views.py:1683
 msgid "Invalid AI configuration."
 msgstr ""
 
-#: documents/views.py:1687
+#: documents/views.py:1694
 msgid "AI backend request timed out."
 msgstr ""
 
-#: documents/views.py:1699
+#: documents/views.py:1706
 msgid "AI backend rejected the request. Check logs for details."
 msgstr ""
 
-#: documents/views.py:2554 documents/views.py:2870
+#: documents/views.py:2573 documents/views.py:2889
 msgid "Specify only one of text, title_search, query, or more_like_id."
 msgstr ""
 
-#: documents/views.py:4800
+#: documents/views.py:4823
 #, python-format
 msgid "Insufficient permissions to share document %(id)s."
 msgstr ""
 
-#: documents/views.py:4846
+#: documents/views.py:4870
 msgid "Bundle is already being processed."
 msgstr ""
 
-#: documents/views.py:4910
+#: documents/views.py:4934
 msgid "The share link bundle is still being prepared. Please try again later."
 msgstr ""
 
-#: documents/views.py:4924
+#: documents/views.py:4948
 msgid "The share link bundle is unavailable."
 msgstr ""
 

src/paperless_ai/indexing.py

@@ -17,6 +17,7 @@ from documents.models import PaperlessTask
 from documents.utils import IterWrapper
 from documents.utils import QuerySetStream
 from documents.utils import identity
+from documents.versioning import root_document_ids
 from paperless.config import AIConfig
 from paperless_ai.db import db_connection_released
 from paperless_ai.embedding import build_llm_index_text
@@ -443,11 +444,11 @@ def update_llm_index(
                 "Skipping LLM index update: migration check deferred; "
                 "will retry next run."
             )
-    documents = Document.objects.select_related(
-        "correspondent",
-        "document_type",
-        "storage_path",
-    ).prefetch_related("tags", "notes", "custom_fields__field")
+    documents = (
+        Document.objects.filter(root_document__isnull=True)
+        .select_related("correspondent", "document_type", "storage_path")
+        .prefetch_related("tags", "notes", "custom_fields__field")
+    )
     no_documents = not documents.exists()
 
     # Fast exit before touching config: nothing to index and no existing index.
@@ -483,7 +484,7 @@ def update_llm_index(
             msg = "LLM index rebuilt successfully."
         else:
             scoped_documents = (
-                documents.filter(id__in=document_ids)
+                documents.filter(id__in=root_document_ids(document_ids))
                 if document_ids is not None
                 else documents
             )
@@ -510,7 +511,12 @@ def update_llm_index(
 
 
 def llm_index_add_or_update_document(document: Document):
-    """Add or atomically replace a document's chunks in the index."""
+    """
+    Add or atomically replace a document's chunks in the index. Only root
+    documents are indexed, so a version is indexed as its root document.
+    """
+    if document.root_document_id is not None:
+        document = document.root_document
     config = AIConfig()
     new_nodes = build_document_node(
         document,

src/paperless_ai/tests/test_ai_indexing.py

@@ -388,6 +388,83 @@ def test_update_llm_index_partial_update(
     assert after[str(doc2.pk)] == before[str(doc2.pk)]
 
 
+@pytest.mark.django_db
+class TestLlmIndexVersions:
+    """The LLM index holds root documents only: a version is indexed as its root."""
+
+    def test_add_or_update_document_indexes_a_version_as_its_root(
+        self,
+        temp_llm_index_dir: Path,
+        mock_embed_model: FakeEmbedding,
+    ) -> None:
+        """
+        GIVEN:
+            - A root document with a version
+        WHEN:
+            - The version is passed to llm_index_add_or_update_document
+        THEN:
+            - Only the root document is in the index
+        """
+        root = DocumentFactory(content="root content")
+        version = DocumentFactory(root_document=root, version_index=1)
+
+        indexing.llm_index_add_or_update_document(version)
+
+        with indexing.get_vector_store() as store:
+            indexed = store.get_modified_times()
+
+        assert set(indexed) == {str(root.pk)}
+
+    def test_rebuild_skips_versions(
+        self,
+        temp_llm_index_dir: Path,
+        mock_embed_model: FakeEmbedding,
+    ) -> None:
+        """
+        GIVEN:
+            - A root document with a version
+        WHEN:
+            - The LLM index is rebuilt
+        THEN:
+            - Only the root document is in the index
+        """
+        root = DocumentFactory()
+        DocumentFactory(root_document=root, version_index=1)
+
+        indexing.update_llm_index(rebuild=True)
+
+        with indexing.get_vector_store() as store:
+            indexed = store.get_modified_times()
+
+        assert set(indexed) == {str(root.pk)}
+
+    def test_incremental_update_by_version_id_refreshes_the_root(
+        self,
+        temp_llm_index_dir: Path,
+        mock_embed_model: FakeEmbedding,
+    ) -> None:
+        """
+        GIVEN:
+            - An indexed root document with a version whose root was modified since
+        WHEN:
+            - An incremental update is scoped to the version's id
+        THEN:
+            - The root's entry is refreshed and no entry exists for the version
+        """
+        root = DocumentFactory()
+        version = DocumentFactory(root_document=root, version_index=1)
+        indexing.update_llm_index(rebuild=True)
+        Document.objects.filter(pk=root.pk).update(modified=timezone.now())
+        root.refresh_from_db()
+
+        indexing.update_llm_index(document_ids=[version.pk])
+
+        with indexing.get_vector_store() as store:
+            indexed = store.get_modified_times()
+
+        assert indexed == {str(root.pk): root.modified.isoformat()}
+
+
 @pytest.mark.django_db
 def test_add_or_update_document_updates_existing_entry(
     temp_llm_index_dir: Path,
@@ -637,6 +714,7 @@ class TestLlmIndexAddOrUpdateDocumentEmptyContent:
 
         doc = MagicMock(spec=Document)
         doc.id = 42
+        doc.root_document_id = None
         # Must not raise
         indexing.llm_index_add_or_update_document(doc)