Chore: share the authorizing owner rule between the id sets and the trash filter

The trash list needs to show a version to whoever owns its root document, but
it deliberately shows only owned and unowned documents and ignores explicit
shares, so it cannot use permitted_object_ids. #14384 gave it its own
filter_queryset with a hand-built condition through the root, which describes
the same rule as the annotations in permitted_object_ids a second time.

Move those annotations into annotate_authorizing_fields. PermittedObjectsFilter
gains a parent_field attribute that its owner-only and granted paths both use,
and the trash filter shrinks to setting include_granted and parent_field.

src/documents/filters.py

@@ -50,6 +50,7 @@ from documents.models import ShareLink
 from documents.models import ShareLinkBundle
 from documents.models import StoragePath
 from documents.models import Tag
+from documents.permissions import annotate_authorizing_fields
 from documents.permissions import permitted_document_ids
 from documents.permissions import permitted_object_ids
 from documents.versioning import annotate_effective_content
@@ -1075,6 +1076,9 @@ class PermittedObjectsFilter(BaseFilterBackend):
 
     include_granted: bool = True
     perm_codename: str | None = None
+    # A self-referencing foreign key whose target authorizes a row, so a row is
+    # judged by its parent's owner and grants (``Document.root_document``).
+    parent_field: str | None = None
 
     def filter_queryset(self, request, queryset, view):
         # Before the superuser and owner-only paths, neither of which consults
@@ -1086,11 +1090,23 @@ class PermittedObjectsFilter(BaseFilterBackend):
         if request.user.is_superuser:
             return queryset
         if not self.include_granted:
-            return queryset.filter(Q(owner=request.user) | Q(owner__isnull=True))
+            owner_field = "owner"
+            if self.parent_field is not None:
+                queryset = annotate_authorizing_fields(queryset, self.parent_field)
+                owner_field = "authorizing_owner"
+            return queryset.filter(
+                Q(**{owner_field: request.user.pk})
+                | Q(**{f"{owner_field}__isnull": True}),
+            )
         model = queryset.model
         perm = self.perm_codename or f"view_{model._meta.model_name}"
         return queryset.filter(
-            id__in=permitted_object_ids(request.user, model, perm),
+            id__in=permitted_object_ids(
+                request.user,
+                model,
+                perm,
+                parent_field=self.parent_field,
+            ),
         )
 
 

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/permissions.py

@@ -30,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):
@@ -386,14 +387,7 @@ def permitted_object_ids(
     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(),
-            ),
-        )
+        base_qs = annotate_authorizing_fields(base_qs, parent_field)
     unowned = Q(**{f"{owner_field}__isnull": True})
 
     if user is None or not getattr(user, "is_authenticated", False):
@@ -445,6 +439,26 @@ def permitted_object_ids(
 ModelT = TypeVar("ModelT", bound=Model)
 
 
+def annotate_authorizing_fields(
+    queryset: QuerySet[ModelT],
+    parent_field: str,
+) -> QuerySet[ModelT]:
+    """
+    Annotate each row with ``authorizing_id`` and ``authorizing_owner``: the id
+    and owner of the row that authorizes it. A row with a parent (the
+    self-referencing foreign key ``parent_field``) is authorized by its parent,
+    any other row by itself.
+    """
+    return queryset.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(),
+        ),
+    )
+
+
 def user_is_unrestricted(user: User | None) -> bool:
     """
     True when ``user`` means "no restriction at all" (an absent user, or an
@@ -676,7 +690,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

@@ -284,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)
@@ -311,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",

src/documents/serialisers.py

@@ -90,7 +90,6 @@ from documents.templating.utils import convert_format_str_to_template_format
 from documents.templating.workflows import validate_workflow_template
 from documents.validators import uri_validator
 from documents.validators import url_validator
-from documents.versioning import get_root_document
 from documents.versioning import has_prefetched_effective_content
 from documents.versioning import sort_versions_newest_first
 
@@ -2895,7 +2894,7 @@ class ShareLinkSerializer(OwnedObjectSerializer):
             and has_perms_owner_aware(
                 self.user,
                 "view_document",
-                get_root_document(document),
+                document,
             )
         ):
             return document

src/documents/tasks.py

@@ -490,23 +490,20 @@ def update_document_content_maybe_archive_file(
                     shutil.move(thumbnail, document.thumbnail_path)
 
             document.refresh_from_db()
-            root_document = (
-                document.root_document if document.root_document_id else document
-            )
             logger.info(
-                f"Updating index for document {root_document.pk} ({document.archive_checksum})",
+                f"Updating index for document {document_id} ({document.archive_checksum})",
             )
             from documents.search import get_backend
 
-            get_backend().add_or_update(root_document)
+            get_backend().add_or_update(document)
 
             ai_config = AIConfig()
             if ai_config.llm_index_enabled:
-                llm_index_add_or_update_document(root_document)
+                llm_index_add_or_update_document(document)
 
             clear_document_caches(document.pk)
-            if root_document.pk != document.pk:
-                clear_document_caches(root_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

@@ -293,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."""
 

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_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_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
@@ -470,6 +471,92 @@ class TestPermittedDocumentIdsVersions:
         )
 
 
+@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_tasks.py

@@ -310,7 +310,7 @@ class TestUpdateContent(DirectoriesMixin, TestCase):
 
     @mock.patch("documents.tasks.clear_document_caches")
     @mock.patch("documents.search.get_backend")
-    def test_update_content_version_indexes_root(
+    def test_update_content_version_clears_caches_for_root(
         self,
         mock_get_backend: mock.Mock,
         mock_clear_caches: mock.Mock,
@@ -321,8 +321,8 @@ class TestUpdateContent(DirectoriesMixin, TestCase):
         WHEN:
             - Update content task is called for the version
         THEN:
-            - The version's content is updated
-            - The root document is indexed rather than the version
+            - 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()
@@ -334,8 +334,7 @@ class TestUpdateContent(DirectoriesMixin, TestCase):
             "my document",
         )
         self.assertEqual(Document.objects.get(pk=root.pk).content, "root content")
-        indexed = mock_get_backend.return_value.add_or_update.call_args.args[0]
-        self.assertEqual(indexed.pk, root.pk)
+        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)],
         )
@@ -343,7 +342,7 @@ class TestUpdateContent(DirectoriesMixin, TestCase):
     @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_for_root(
+    def test_update_content_version_updates_llm_index(
         self,
         mock_get_backend: mock.Mock,
         mock_llm_index: mock.Mock,
@@ -355,14 +354,13 @@ class TestUpdateContent(DirectoriesMixin, TestCase):
         WHEN:
             - Update content task is called for the version
         THEN:
-            - The LLM index is updated for the root document, not the version
+            - The LLM index is updated
         """
-        root, version = self._create_root_with_version()
+        _, version = self._create_root_with_version()
 
         tasks.update_document_content_maybe_archive_file(version.pk)
 
         mock_llm_index.assert_called_once()
-        self.assertEqual(mock_llm_index.call_args.args[0].pk, root.pk)
 
 
 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):
@@ -1559,7 +1561,7 @@ class DocumentViewSet(
         if request.user is not None and not has_perms_owner_aware(
             request.user,
             "change_document",
-            get_root_document(doc),
+            doc,
         ):
             return HttpResponseForbidden("Insufficient permissions")
 
@@ -1622,7 +1624,7 @@ class DocumentViewSet(
         if request.user is not None and not has_perms_owner_aware(
             request.user,
             "change_document",
-            get_root_document(doc),
+            doc,
         ):
             return HttpResponseForbidden("Insufficient permissions")
 
@@ -1875,7 +1877,7 @@ class DocumentViewSet(
             if currentUser is not None and not has_perms_owner_aware(
                 currentUser,
                 "view_document",
-                get_root_document(doc),
+                doc,
             ):
                 return HttpResponseForbidden("Insufficient permissions to view notes")
         except Document.DoesNotExist:
@@ -1897,7 +1899,7 @@ class DocumentViewSet(
                 if currentUser is not None and not has_perms_owner_aware(
                     currentUser,
                     "change_document",
-                    get_root_document(doc),
+                    doc,
                 ):
                     return HttpResponseForbidden(
                         "Insufficient permissions to create notes",
@@ -1940,7 +1942,7 @@ class DocumentViewSet(
             if currentUser is not None and not has_perms_owner_aware(
                 currentUser,
                 "change_document",
-                get_root_document(doc),
+                doc,
             ):
                 return HttpResponseForbidden("Insufficient permissions to delete notes")
 
@@ -1990,7 +1992,7 @@ class DocumentViewSet(
             if currentUser is not None and not has_perms_owner_aware(
                 currentUser,
                 "change_document",
-                get_root_document(doc),
+                doc,
             ):
                 return HttpResponseForbidden(
                     "Insufficient permissions to add share link",
@@ -2451,7 +2453,7 @@ class ChatStreamingView(GenericAPIView[Any]):
             if not has_perms_owner_aware(
                 request.user,
                 "view_document",
-                get_root_document(document),
+                document,
             ):
                 return HttpResponseForbidden("Insufficient permissions")
 
@@ -5600,23 +5602,7 @@ 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__")
-                ),
-            )
+        parent_field = "root_document"
 
     filter_backends = (_TrashPermittedObjectsFilter,)
     pagination_class = StandardPagination

src/paperless_ai/ai_classifier.py

@@ -7,6 +7,7 @@ from documents.models import Document
 from documents.permissions import permitted_object_ids
 from documents.permissions import restrict_queryset_to_visible
 from documents.permissions import user_is_unrestricted
+from documents.versioning import annotate_effective_content
 from paperless.config import AIConfig
 from paperless_ai.base_model import ClassificationSuggestions
 from paperless_ai.base_model import TaxonomyChoiceDict
@@ -113,7 +114,7 @@ def build_prompt_without_rag(
 ) -> str:
     filename = document.filename or ""
     content = truncate_content(
-        document.content[:4000] or "",
+        (document.get_effective_content() or "")[:4000],
         chunk_size=config.llm_embedding_chunk_size,
         context_size=config.llm_context_size,
     )
@@ -227,7 +228,9 @@ def get_taxonomy_context(
 
         # similar_documents is already ordered by descending weight; don't lose it.
         similar_document_ids = [s["document_id"] for s in similar_documents]
-        similar_documents_by_id = Document.objects.in_bulk(similar_document_ids)
+        similar_documents_by_id = annotate_effective_content(
+            Document.objects.all(),
+        ).in_bulk(similar_document_ids)
         similar_docs = [
             similar_documents_by_id[document_id]
             for document_id in similar_document_ids
@@ -235,7 +238,7 @@ def get_taxonomy_context(
         ][:max_docs]
         context_blocks = []
         for similar in similar_docs:
-            text = similar.content[:1000] or ""
+            text = (similar.get_effective_content() or "")[:1000]
             title = similar.title or similar.filename or "Untitled"
             context_blocks.append(f"TITLE: {title}\n{text}")
     except Exception:

src/paperless_ai/embedding.py

@@ -135,6 +135,6 @@ def build_llm_index_text(doc: Document) -> str:
         lines.append(f"Custom Field - {instance.field.name}: {instance}")
 
     lines.append("\nContent:\n")
-    lines.append(doc.content or "")
+    lines.append(doc.get_effective_content() or "")
 
     return _normalize_llm_index_text("\n".join(lines))

src/paperless_ai/indexing.py

@@ -17,6 +17,8 @@ from documents.models import PaperlessTask
 from documents.utils import IterWrapper
 from documents.utils import QuerySetStream
 from documents.utils import identity
+from documents.versioning import annotate_effective_content
+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 +445,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 = annotate_effective_content(
+        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 +485,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 +512,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,
@@ -688,7 +695,7 @@ def retrieve_similar_nodes(
     )
 
     query_text = truncate_embedding_query(
-        (document.title or "") + "\n" + (document.content or ""),
+        (document.title or "") + "\n" + (document.get_effective_content() or ""),
         chunk_size=config.llm_embedding_chunk_size,
     )
     # Hold the shared read lock for the whole retrieval so the connection is

src/paperless_ai/tests/test_ai_classifier.py

@@ -55,6 +55,7 @@ def mock_document():
     doc.storage_path = None
     doc.archive_serial_number = "12345"
     doc.content = "This is the document content."
+    doc.get_effective_content.return_value = "This is the document content."
 
     cf1 = MagicMock(__str__=lambda x: "Value1")
     cf1.field = MagicMock()
@@ -434,6 +435,55 @@ def test_get_taxonomy_context_preserves_similarity_order_and_distinct_documents(
     )
 
 
+@pytest.mark.django_db
+class TestClassifierEffectiveContent:
+    """A root document's text for the LLM is its newest version's content."""
+
+    @staticmethod
+    def _root_with_version() -> Document:
+        root = DocumentFactory(title="Statement", content="stale text")
+        DocumentFactory(root_document=root, version_index=1, content="latest text")
+        return root
+
+    def test_prompt_uses_the_newest_versions_content(self) -> None:
+        """
+        GIVEN:
+            - A root document with a version
+        WHEN:
+            - The classification prompt is built for the root
+        THEN:
+            - It contains the newest version's content
+        """
+        prompt = build_prompt_without_rag(self._root_with_version(), AIConfig())
+
+        assert "latest text" in prompt
+        assert "stale text" not in prompt
+
+    @override_settings(LLM_EMBEDDING_BACKEND="huggingface")
+    def test_similar_document_context_uses_the_newest_versions_content(self) -> None:
+        """
+        GIVEN:
+            - A similar root document with a version
+        WHEN:
+            - The similar-document context is built
+        THEN:
+            - It contains the newest version's content
+        """
+        similar = self._root_with_version()
+        document = DocumentFactory(content="Some content")
+        fake_nodes = [
+            SimpleNamespace(metadata={"document_id": str(similar.pk)}, score=0.9),
+        ]
+
+        with patch(
+            "paperless_ai.ai_classifier.retrieve_similar_nodes",
+            return_value=fake_nodes,
+        ):
+            _candidates, context = get_taxonomy_context(document, user=None)
+
+        assert context == "TITLE: Statement\nlatest text"
+
+
 @pytest.mark.django_db
 @override_settings(LLM_EMBEDDING_BACKEND="huggingface")
 def test_get_taxonomy_context_no_similar_docs():

src/paperless_ai/tests/test_ai_indexing.py

@@ -388,6 +388,108 @@ 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_rebuild_indexes_the_newest_versions_content(
+        self,
+        temp_llm_index_dir: Path,
+        mock_embed_model: FakeEmbedding,
+        mocker: pytest_mock.MockerFixture,
+    ) -> None:
+        """
+        GIVEN:
+            - A root document with a version
+        WHEN:
+            - The LLM index is rebuilt
+        THEN:
+            - The root's text for the index is the version's content, answered
+              from the query rather than a query per document
+        """
+        root = DocumentFactory(content="stale text")
+        DocumentFactory(root_document=root, version_index=1, content="latest text")
+        spy = mocker.spy(indexing, "build_document_node")
+
+        indexing.update_llm_index(rebuild=True)
+
+        indexed = spy.call_args.args[0]
+        assert indexed.pk == root.pk
+        assert indexed.effective_content == "latest text"
+
+    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 +739,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)
 

src/paperless_ai/tests/test_embedding.py

@@ -12,6 +12,7 @@ from paperless_ai.embedding import _normalize_llm_index_text
 from paperless_ai.embedding import build_llm_index_text
 from paperless_ai.embedding import get_configured_model_name
 from paperless_ai.embedding import get_embedding_model
+from paperless_testing.factories import DocumentFactory
 
 
 @pytest.fixture
@@ -46,6 +47,7 @@ def mock_document():
     doc.correspondent.name = "Test Correspondent"
     doc.archive_serial_number = "12345"
     doc.content = "This is the document content."
+    doc.get_effective_content.return_value = "This is the document content."
 
     cf1 = MagicMock(__str__=lambda x: "Value1")
     cf1.field = MagicMock()
@@ -280,7 +282,7 @@ def test_build_llm_index_text(mock_document):
 
 
 def test_build_llm_index_text_normalizes_ocr_punctuation_runs(mock_document):
-    mock_document.content = (
+    mock_document.get_effective_content.return_value = (
         "Introduction ................................................ 7\n"
         "Hardware Limitation ________________________________________ 9\n"
         "Keep short punctuation like INV-100 and ellipses..."
@@ -294,6 +296,43 @@ def test_build_llm_index_text_normalizes_ocr_punctuation_runs(mock_document):
     assert "ellipses..." in result
 
 
+@pytest.mark.django_db
+class TestBuildLlmIndexTextVersions:
+    """A root document is indexed with its effective content, like in the search index."""
+
+    def test_root_uses_the_newest_versions_content(self) -> None:
+        """
+        GIVEN:
+            - A root document with two versions
+        WHEN:
+            - The LLM index text is built for the root
+        THEN:
+            - It contains the newest version's content and not the others'
+        """
+        root = DocumentFactory(content="stale text")
+        DocumentFactory(root_document=root, version_index=1, content="older text")
+        DocumentFactory(root_document=root, version_index=2, content="latest text")
+
+        text = build_llm_index_text(root)
+
+        assert "latest text" in text
+        assert "stale text" not in text
+        assert "older text" not in text
+
+    def test_root_without_versions_uses_its_own_content(self) -> None:
+        """
+        GIVEN:
+            - A root document without versions
+        WHEN:
+            - The LLM index text is built for it
+        THEN:
+            - It contains the document's own content
+        """
+        root = DocumentFactory(content="own text")
+
+        assert "own text" in build_llm_index_text(root)
+
+
 def test_normalize_llm_index_text_collapses_ocr_leaders_without_joining_lines():
     assert _normalize_llm_index_text("A........B\nC____D----E") == "A B\nC D E"