test(search): move schema fingerprint/version tests from PR2

Both files exercise documents.search._schema exclusively (build_schema,
schema_fingerprint, needs_rebuild, open_or_rebuild_index) with no
dependency on whoosh-compat query routing, so they belong with the
schema/field-registry work rather than PR2's query rewrite.

pyproject.toml

@@ -77,6 +77,7 @@ dependencies = [
   "torch~=2.13.0",
   "watchfiles>=1.2",
   "whitenoise~=6.11",
+  "whoosh-compat[tantivy]==0.1",
   "zxing-cpp~=3.1.0",
 ]
 [project.optional-dependencies]

src/documents/bulk_edit.py

@@ -28,7 +28,7 @@ from documents.models import DocumentType
 from documents.models import PaperlessTask
 from documents.models import StoragePath
 from documents.models import Tag
-from documents.permissions import set_permissions_for_objects
+from documents.permissions import set_permissions_for_object
 from documents.plugins.helpers import DocumentsStatusManager
 from documents.tasks import bulk_update_documents
 from documents.tasks import consume_file
@@ -433,13 +433,10 @@ def set_permissions(
     else:
         qs.update(owner=owner)
 
+    for doc in qs:
+        set_permissions_for_object(permissions=set_permissions, object=doc, merge=merge)
+
     affected_docs = list(qs.values_list("pk", flat=True))
-    set_permissions_for_objects(
-        permissions=set_permissions,
-        model=Document,
-        pks=affected_docs,
-        merge=merge,
-    )
 
     bulk_update_documents.apply_async(
         kwargs={"document_ids": affected_docs},

src/documents/models.py

@@ -375,7 +375,6 @@ class Document(SoftDeleteModel, ModelWithOwner):  # type: ignore[django-manager-
         If the queryset already annotated ``effective_content``, that value is used.
         """
         # Here to avoid circular import
-        from documents.versioning import LATEST_VERSION_CONTENT_PREFETCH_ATTR
         from documents.versioning import sort_versions_newest_first
         from documents.versioning import versions_newest_first
 
@@ -385,19 +384,6 @@ class Document(SoftDeleteModel, ModelWithOwner):  # type: ignore[django-manager-
         if self.root_document_id is not None or self.pk is None:
             return self.content
 
-        latest_version_prefetch = getattr(
-            self,
-            LATEST_VERSION_CONTENT_PREFETCH_ATTR,
-            None,
-        )
-        if latest_version_prefetch is not None:
-            # Empty list means prefetch ran and found no versions — use own content.
-            return (
-                latest_version_prefetch[0].content
-                if latest_version_prefetch
-                else self.content
-            )
-
         prefetched_cache = getattr(self, "_prefetched_objects_cache", None)
         prefetched_versions = (
             prefetched_cache.get("versions")

src/documents/permissions.py

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

src/documents/search/_fields.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+from whoosh_compat import FieldKind
+from whoosh_compat import FieldSpec
+from whoosh_compat import SubpathSpec
+
+# Internal-only schema fields with no query-syntax meaning of their own
+# (sort shadow fields, bigram CJK fields, simple_title/simple_content,
+# autocomplete_word, notes_text) are NOT represented here, they are
+# declared in _schema.py's field_descriptors().
+#
+# analyzer/pattern_normalizer are deliberately left at FieldSpec's default
+# (None): they're language-specific and only meaningful to whoosh-compat's
+# parser, so _registry.py attaches them per-language via dataclasses.replace()
+# rather than PUBLIC_FIELDS declaring them itself. _schema.py only reads
+# name/kind/fast and never sees the analyzer at all.
+PUBLIC_FIELDS: tuple[FieldSpec, ...] = (
+    FieldSpec("title", FieldKind.TEXT),
+    FieldSpec("content", FieldKind.TEXT),
+    FieldSpec("correspondent", FieldKind.TEXT),
+    FieldSpec("document_type", FieldKind.TEXT, aliases=("type",)),
+    FieldSpec("storage_path", FieldKind.TEXT, aliases=("path",)),
+    FieldSpec("original_filename", FieldKind.TEXT),
+    FieldSpec("tag", FieldKind.TEXT, comma_values=True),
+    FieldSpec("checksum", FieldKind.KEYWORD),
+    FieldSpec("asn", FieldKind.U64, fast=True),
+    FieldSpec("page_count", FieldKind.U64, fast=True),
+    FieldSpec("num_notes", FieldKind.U64, fast=True),
+    FieldSpec("created", FieldKind.DATE, date_only=True, fast=True),
+    FieldSpec("modified", FieldKind.DATETIME, fast=True),
+    FieldSpec("added", FieldKind.DATETIME, fast=True),
+    FieldSpec(
+        "notes",
+        FieldKind.JSON,
+        subpaths={"user": SubpathSpec(), "note": SubpathSpec(default=True)},
+    ),
+    FieldSpec(
+        "custom_fields",
+        FieldKind.JSON,
+        subpaths={"name": SubpathSpec(), "value": SubpathSpec(default=True)},
+    ),
+)

src/documents/search/_registry.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+import dataclasses
+from typing import TYPE_CHECKING
+
+from whoosh_compat import FieldKind
+from whoosh_compat import FieldRegistry
+
+from documents.search._fields import PUBLIC_FIELDS
+from documents.search._tokenizer import ascii_fold
+from documents.search._tokenizer import paperless_text_analyzer
+from documents.search._tokenizer import stem_pattern_text
+
+if TYPE_CHECKING:
+    from whoosh_compat import PatternNormalizer
+
+_registry_cache: dict[str | None, FieldRegistry] = {}
+
+
+def _identity_analyzer(text: str) -> list[str]:
+    """Analyzer for KEYWORD fields indexed with the raw tokenizer (no splitting)."""
+    return [text]
+
+
+def _fold_normalizer(text: str) -> str:
+    """Wildcard/regex literal-run normalizer for fields indexed without stemming."""
+    return ascii_fold(text.lower())
+
+
+def _make_pattern_normalizer(language: str | None) -> PatternNormalizer:
+    """Build the wildcard/regex literal-run normalizer for a search language."""
+
+    def _pattern_normalizer(text: str) -> tuple[str, ...]:
+        """Normalize a literal run into the forms a term may match.
+
+        TEXT index terms go through lowercase -> ascii_fold -> stem, so a
+        pattern that skips stemming can never match one: "invoice*" would look
+        for a term starting with "invoice" while the index holds "invoic". The
+        run is therefore offered stemmed as well. KEYWORD fields are indexed
+        raw and get _fold_normalizer instead, so their patterns stay literal.
+
+        Both forms are returned, as alternatives, because neither is a prefix
+        of the other in general: English stemming substitutes as well as
+        truncates ("copy" -> "copi"), so the stem alone loses the compounds
+        the typed run reaches ("copyright") while the typed run alone loses
+        the inflections the stem reaches ("copies"). whoosh-compat ORs the
+        alternatives per literal run and deduplicates them, so a run the
+        stemmer leaves alone costs exactly the one branch it did before.
+
+        Inside a bracket class the emitter calls this once per character and
+        uses the answer only if it is a single one-character form; two forms
+        there leave the character as typed. A stemmer does not change a lone
+        character, so the two forms deduplicate to one and the class body is
+        folded as before.
+        """
+        folded = ascii_fold(text.lower())
+        stemmed = stem_pattern_text(folded, language)
+        return (folded, stemmed)
+
+    return _pattern_normalizer
+
+
+def get_field_registry(language: str | None) -> FieldRegistry:
+    """Build (or return the cached) FieldRegistry for the given search language.
+
+    Cached keyed by language, rebuilt on the same trigger register_tokenizers()
+    uses (settings.SEARCH_LANGUAGE change). A fresh call with a new language
+    builds and caches a new registry rather than mutating the old one.
+    """
+    if language in _registry_cache:
+        return _registry_cache[language]
+
+    text_analyzer = paperless_text_analyzer(language).analyze
+    pattern_normalizer = _make_pattern_normalizer(language)
+
+    specs = [
+        dataclasses.replace(
+            field,
+            analyzer=_identity_analyzer
+            if field.kind is FieldKind.KEYWORD
+            else text_analyzer,
+            pattern_normalizer=_fold_normalizer
+            if field.kind is FieldKind.KEYWORD
+            else pattern_normalizer,
+        )
+        for field in PUBLIC_FIELDS
+    ]
+
+    registry = FieldRegistry(specs)
+    _registry_cache[language] = registry
+    return registry

src/documents/search/_schema.py

@@ -1,14 +1,19 @@
 from __future__ import annotations
 
+import hashlib
 import json
 import logging
 import shutil
 from typing import TYPE_CHECKING
 from typing import Final
+from typing import NamedTuple
 from typing import cast
 
 import tantivy
 from django.conf import settings
+from whoosh_compat import FieldKind
+
+from documents.search._fields import PUBLIC_FIELDS
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -16,7 +21,185 @@ if TYPE_CHECKING:
 logger = logging.getLogger("paperless.search")
 
 # v1 - Initial tantivy schema format
-SCHEMA_VERSION: Final[int] = 1
+# v2 - build_schema() derived from PUBLIC_FIELDS, changing the field declaration
+#      order, and the write-only correspondent/document_type/storage_path/tag id
+#      columns dropped. tantivy compares schemas by ordered field list, so an
+#      index built by v1 rejects every write against the v2 schema.
+SCHEMA_VERSION: Final[int] = 2
+
+
+class FieldDescriptor(NamedTuple):
+    """One tantivy field, in declaration order.
+
+    The descriptor vocabulary is paperless', not tantivy-py's: it is both the
+    input to the SchemaBuilder and the input to schema_fingerprint(), so the
+    persisted fingerprint cannot move under a tantivy-py upgrade.
+    """
+
+    name: str
+    kind: str
+    stored: bool
+    indexed: bool
+    fast: bool
+    tokenizer: str | None
+
+
+# (schema kind, tokenizer) for the FieldKind -> FieldDescriptor mapping that
+# doesn't need special-casing. JSON is handled separately below since it can
+# emit a second, synthetic descriptor.
+_KIND_TABLE: Final[dict[FieldKind, tuple[str, str | None]]] = {
+    FieldKind.TEXT: ("text", "paperless_text"),
+    FieldKind.KEYWORD: ("text", "raw"),
+    FieldKind.U64: ("u64", None),
+    FieldKind.DATE: ("date", None),
+    FieldKind.DATETIME: ("date", None),
+}
+# Kinds whose fast-field flag follows FieldSpec.fast rather than always False.
+_FAST_FROM_FIELD: Final[frozenset[FieldKind]] = frozenset(
+    {FieldKind.U64, FieldKind.DATE, FieldKind.DATETIME},
+)
+
+
+def _public_field_descriptors() -> list[FieldDescriptor]:
+    """Descriptors for the query-visible fields declared in PUBLIC_FIELDS."""
+    descriptors: list[FieldDescriptor] = []
+    for field in PUBLIC_FIELDS:
+        if field.kind is FieldKind.JSON:
+            descriptors.append(
+                FieldDescriptor(
+                    field.name,
+                    "json",
+                    stored=True,
+                    indexed=True,
+                    fast=False,
+                    tokenizer="paperless_text",
+                ),
+            )
+            if field.name == "notes":
+                # Plain-text companion for snippet generation: tantivy's
+                # SnippetGenerator does not support JSON fields. Schema-only,
+                # no query-syntax meaning, not in PUBLIC_FIELDS.
+                descriptors.append(
+                    FieldDescriptor(
+                        "notes_text",
+                        "text",
+                        stored=True,
+                        indexed=True,
+                        fast=False,
+                        tokenizer="paperless_text",
+                    ),
+                )
+            continue
+        schema_kind, tokenizer = _KIND_TABLE[field.kind]
+        descriptors.append(
+            FieldDescriptor(
+                field.name,
+                schema_kind,
+                stored=True,
+                indexed=True,
+                fast=field.fast if field.kind in _FAST_FROM_FIELD else False,
+                tokenizer=tokenizer,
+            ),
+        )
+    return descriptors
+
+
+def field_descriptors() -> list[FieldDescriptor]:
+    """Every field of the document index, in the order tantivy declares them.
+
+    tantivy compares schemas by *ordered* field list, so the order here is
+    part of the on-disk contract: schema_fingerprint() hashes it and
+    needs_rebuild() acts on the result.
+    """
+    return [
+        FieldDescriptor(
+            "id",
+            "u64",
+            stored=True,
+            indexed=True,
+            fast=True,
+            tokenizer=None,
+        ),
+        *_public_field_descriptors(),
+        # Shadow sort fields - fast, not stored
+        *(
+            FieldDescriptor(
+                name,
+                "text",
+                stored=False,
+                indexed=True,
+                fast=True,
+                tokenizer="simple_analyzer",
+            )
+            for name in ("title_sort", "correspondent_sort", "type_sort")
+        ),
+        # CJK support - not stored, indexed only
+        *(
+            FieldDescriptor(
+                name,
+                "text",
+                stored=False,
+                indexed=True,
+                fast=False,
+                tokenizer="bigram_analyzer",
+            )
+            for name in (
+                "bigram_content",
+                "bigram_title",
+                "bigram_correspondent",
+                "bigram_document_type",
+                "bigram_tag",
+            )
+        ),
+        # Simple substring search support for title/content - not stored,
+        # indexed only
+        *(
+            FieldDescriptor(
+                name,
+                "text",
+                stored=False,
+                indexed=True,
+                fast=False,
+                tokenizer="simple_search_analyzer",
+            )
+            for name in ("simple_title", "simple_content")
+        ),
+        # Autocomplete prefix scan via terms_with_prefix, which walks the
+        # field's term dictionary - so the field must be indexed (term dict),
+        # not stored. The stored value is never read back, so storing it only
+        # wastes space.
+        FieldDescriptor(
+            "autocomplete_word",
+            "text",
+            stored=False,
+            indexed=True,
+            fast=False,
+            tokenizer="raw",
+        ),
+        # Permission filter columns, read by build_permission_filter.
+        *(
+            FieldDescriptor(
+                name,
+                "u64",
+                stored=False,
+                indexed=True,
+                fast=True,
+                tokenizer=None,
+            )
+            for name in ("owner_id", "viewer_id", "viewer_group_id")
+        ),
+    ]
+
+
+def schema_fingerprint() -> str:
+    """Hash of the field descriptors, stamped into .index_settings.json.
+
+    Changes whenever a field is added, removed, retyped, re-optioned or
+    reordered, so an index built from a different schema shape is detected
+    even when SCHEMA_VERSION was not bumped.
+    """
+    payload = json.dumps([list(descriptor) for descriptor in field_descriptors()])
+    return hashlib.blake2b(payload.encode()).hexdigest()
 
 
 def build_schema() -> tantivy.Schema:
@@ -32,85 +215,37 @@ def build_schema() -> tantivy.Schema:
     """
     sb = tantivy.SchemaBuilder()
 
-    sb.add_unsigned_field("id", stored=True, indexed=True, fast=True)
-    sb.add_text_field("checksum", stored=True, tokenizer_name="raw")
-
-    for field in (
-        "title",
-        "correspondent",
-        "document_type",
-        "storage_path",
-        "original_filename",
-        "content",
-    ):
-        sb.add_text_field(field, stored=True, tokenizer_name="paperless_text")
-
-    # Shadow sort fields - fast, not stored/indexed
-    for field in ("title_sort", "correspondent_sort", "type_sort"):
-        sb.add_text_field(
-            field,
-            stored=False,
-            tokenizer_name="simple_analyzer",
-            fast=True,
-        )
-
-    # CJK support - not stored, indexed only
-    sb.add_text_field("bigram_content", stored=False, tokenizer_name="bigram_analyzer")
-    sb.add_text_field("bigram_title", stored=False, tokenizer_name="bigram_analyzer")
-    sb.add_text_field(
-        "bigram_correspondent",
-        stored=False,
-        tokenizer_name="bigram_analyzer",
-    )
-    sb.add_text_field(
-        "bigram_document_type",
-        stored=False,
-        tokenizer_name="bigram_analyzer",
-    )
-    sb.add_text_field("bigram_tag", stored=False, tokenizer_name="bigram_analyzer")
-
-    # Simple substring search support for title/content - not stored, indexed only
-    sb.add_text_field(
-        "simple_title",
-        stored=False,
-        tokenizer_name="simple_search_analyzer",
-    )
-    sb.add_text_field(
-        "simple_content",
-        stored=False,
-        tokenizer_name="simple_search_analyzer",
-    )
-
-    # Autocomplete prefix scan via terms_with_prefix, which walks the field's
-    # term dictionary - so the field must be indexed (term dict), not stored.
-    # The stored value is never read back, so storing it only wastes space.
-    sb.add_text_field("autocomplete_word", stored=False, tokenizer_name="raw")
-
-    sb.add_text_field("tag", stored=True, tokenizer_name="paperless_text")
-
-    # JSON fields — structured queries: notes.user:alice, custom_fields.name:invoice
-    sb.add_json_field("notes", stored=True, tokenizer_name="paperless_text")
-    # Plain-text companion for notes — tantivy's SnippetGenerator does not support
-    # JSON fields, so highlights require a text field with the same content.
-    sb.add_text_field("notes_text", stored=True, tokenizer_name="paperless_text")
-    sb.add_json_field("custom_fields", stored=True, tokenizer_name="paperless_text")
-
-    for field in (
-        "correspondent_id",
-        "document_type_id",
-        "storage_path_id",
-        "tag_id",
-        "owner_id",
-        "viewer_id",
-        "viewer_group_id",
-    ):
-        sb.add_unsigned_field(field, stored=False, indexed=True, fast=True)
-
-    for field in ("created", "modified", "added"):
-        sb.add_date_field(field, stored=True, indexed=True, fast=True)
-
-    for field in ("asn", "page_count", "num_notes"):
-        sb.add_unsigned_field(field, stored=True, indexed=True, fast=True)
+    for descriptor in field_descriptors():
+        if descriptor.kind == "text":
+            sb.add_text_field(
+                descriptor.name,
+                stored=descriptor.stored,
+                fast=descriptor.fast,
+                tokenizer_name=cast("str", descriptor.tokenizer),
+            )
+        elif descriptor.kind == "json":
+            sb.add_json_field(
+                descriptor.name,
+                stored=descriptor.stored,
+                fast=descriptor.fast,
+                tokenizer_name=cast("str", descriptor.tokenizer),
+            )
+        elif descriptor.kind == "u64":
+            sb.add_unsigned_field(
+                descriptor.name,
+                stored=descriptor.stored,
+                indexed=descriptor.indexed,
+                fast=descriptor.fast,
+            )
+        elif descriptor.kind == "date":
+            sb.add_date_field(
+                descriptor.name,
+                stored=descriptor.stored,
+                indexed=descriptor.indexed,
+                fast=descriptor.fast,
+            )
+        else:
+            raise ValueError(f"Unknown schema field kind: {descriptor.kind}")
 
     return sb.build()
 
@@ -119,9 +254,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 and
-    search language against the current configuration. Returns True if the
-    file is missing, unparsable, or either value mismatches.
+    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.
 
     Args:
         index_dir: Path to the search index directory
@@ -140,6 +275,9 @@ def needs_rebuild(index_dir: Path) -> bool:
         if "language" not in data or data["language"] != settings.SEARCH_LANGUAGE:
             logger.info("Search index language changed - rebuilding.")
             return True
+        if data.get("schema_fingerprint") != schema_fingerprint():
+            logger.info("Search index schema fingerprint mismatch - rebuilding.")
+            return True
     except ValueError:
         return True
     return False
@@ -170,6 +308,7 @@ def _write_sentinels(index_dir: Path) -> None:
             {
                 "schema_version": SCHEMA_VERSION,
                 "language": settings.SEARCH_LANGUAGE,
+                "schema_fingerprint": schema_fingerprint(),
             },
         ),
     )

src/documents/search/_tokenizer.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+from functools import cache
 from typing import Final
 
 import tantivy
@@ -71,7 +72,7 @@ def register_tokenizers(index: tantivy.Index, language: str | None) -> None:
         use fast=True and Tantivy requires fast-field tokenizers to exist
         even for documents that omit those fields.
     """
-    index.register_tokenizer("paperless_text", _paperless_text(language))
+    index.register_tokenizer("paperless_text", paperless_text_analyzer(language))
     index.register_tokenizer("simple_analyzer", _simple_analyzer())
     index.register_tokenizer("bigram_analyzer", _bigram_analyzer())
     index.register_tokenizer("simple_search_analyzer", _simple_search_analyzer())
@@ -79,7 +80,7 @@ def register_tokenizers(index: tantivy.Index, language: str | None) -> None:
     index.register_fast_field_tokenizer("simple_analyzer", _simple_analyzer())
 
 
-def _paperless_text(language: str | None) -> tantivy.TextAnalyzer:
+def paperless_text_analyzer(language: str | None) -> tantivy.TextAnalyzer:
     """Main full-text tokenizer for content, title, etc: simple -> remove_long(129) -> lowercase -> ascii_fold [-> stemmer]"""
     builder = (
         tantivy.TextAnalyzerBuilder(tantivy.Tokenizer.simple())
@@ -100,6 +101,54 @@ def _paperless_text(language: str | None) -> tantivy.TextAnalyzer:
     return builder.build()
 
 
+@cache
+def _pattern_stemmer(language: str | None) -> tantivy.TextAnalyzer | None:
+    """The stemming tail of paperless_text_analyzer, over a whole literal run.
+
+    Same language gate and same Snowball stemmer paperless_text_analyzer
+    applies at index time, so query patterns follow SEARCH_LANGUAGE. Returns
+    None when that gate disables stemming; paperless_text_analyzer already
+    warns about an unsupported language, so this stays quiet.
+
+    The raw tokenizer keeps the run whole (a wildcard literal is a fragment,
+    not necessarily a word), and remove_long is kept so an over-long run is
+    treated the same way the index treats it.
+    """
+    if not language:
+        return None
+    tantivy_lang = _LANGUAGE_MAP.get(language.lower())
+    if tantivy_lang is None:
+        return None
+    return (
+        tantivy.TextAnalyzerBuilder(tantivy.Tokenizer.raw())
+        .filter(tantivy.Filter.remove_long(_TOKEN_REMOVE_LONG_LIMIT))
+        .filter(tantivy.Filter.stemmer(tantivy_lang))
+        .build()
+    )
+
+
+def stem_pattern_text(text: str, language: str | None) -> str:
+    """Stem an already lowercased/ascii-folded run the way index terms are.
+
+    Returns text unchanged when stemming is disabled for language, and also
+    when the stem step does not yield exactly one token: remove_long drops a run
+    past the length limit, leaving no stem to substitute. Falling back to the
+    text as typed is the safe direction for a pattern prefix, since it can only
+    be as narrow as it was before stemming was considered.
+
+    The raw tokenizer emits one token whatever the input and the stemmer is
+    1-to-1, so only the zero-token case can fire today; the guard covers both
+    counts so a tokenizer change cannot turn this into an IndexError.
+    """
+    analyzer = _pattern_stemmer(language)
+    if analyzer is None:
+        return text
+    tokens = analyzer.analyze(text)
+    if len(tokens) != 1:
+        return text
+    return tokens[0]
+
+
 def _simple_analyzer() -> tantivy.TextAnalyzer:
     """Tokenizer for shadow sort fields (title_sort, correspondent_sort, type_sort): simple -> lowercase -> ascii_fold."""
     return (

src/documents/serialisers.py

@@ -89,7 +89,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 has_prefetched_effective_content
 from documents.versioning import sort_versions_newest_first
 
 if TYPE_CHECKING:
@@ -1153,14 +1152,8 @@ class DocumentSerializer(
 
     def to_representation(self, instance):
         doc = super().to_representation(instance)
-        if "content" in self.fields and has_prefetched_effective_content(instance):
-            # Only resolve version-aware content when it's cheap: an SQL
-            # annotation or a versions prefetch is already on the instance.
-            # A caller that set up neither (e.g. TrashView, GlobalSearchView,
-            # which build their own querysets) gets the document's own,
-            # unresolved content instead of paying for an extra per-instance
-            # query -- same as before effective_content resolution existed.
-            doc["content"] = instance.get_effective_content() or ""
+        if "content" in self.fields and hasattr(instance, "effective_content"):
+            doc["content"] = getattr(instance, "effective_content") or ""
         if self.truncate_content and "content" in self.fields:
             doc["content"] = doc.get("content")[0:550]
         return doc

src/documents/tests/search/test_fields.py

@@ -0,0 +1,10 @@
+from whoosh_compat import FieldKind
+
+from documents.search._fields import PUBLIC_FIELDS
+
+
+class TestPublicFields:
+    def test_json_fields_have_subpaths(self) -> None:
+        for field in PUBLIC_FIELDS:
+            if field.kind is FieldKind.JSON:
+                assert field.subpaths, f"{field.name} is JSON but has no subpaths"

src/documents/tests/search/test_json_subpath_completeness.py

@@ -0,0 +1,83 @@
+"""Every declared JSON subpath must actually be written to the index.
+
+PUBLIC_FIELDS declares each JSON field's subpaths (e.g. ``notes`` ->
+{"user", "note"}), but nothing coupled that declaration to what
+``_backend.py``'s document builder actually writes into the JSON blob at
+index time. A subpath declared but never written would be
+queryable-but-always-empty -- syntactically valid, silently matching
+nothing -- with no test failure anywhere.
+
+This indexes one real document carrying values for every JSON field
+(a Note, a CustomFieldInstance) and inspects the document's own stored
+JSON payload, rather than running field-specific queries: that way a
+future JSON field's subpaths are covered automatically, without a new
+per-subpath query having to be added by hand each time.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+import tantivy
+from django.contrib.auth.models import User
+from whoosh_compat import FieldKind
+
+from documents.models import CustomField
+from documents.models import CustomFieldInstance
+from documents.models import Document
+from documents.models import Note
+from documents.search._fields import PUBLIC_FIELDS
+
+if TYPE_CHECKING:
+    from documents.search._backend import TantivyBackend
+
+pytestmark = [pytest.mark.search, pytest.mark.django_db]
+
+
+class TestJsonSubpathsAreWrittenAtIndexTime:
+    def test_every_declared_json_subpath_appears_in_the_stored_document(
+        self,
+        backend: TantivyBackend,
+    ) -> None:
+        user = User.objects.create_user(username="completeness-user")
+        field = CustomField.objects.create(
+            name="Completeness Field",
+            data_type=CustomField.FieldDataType.STRING,
+        )
+        doc = Document.objects.create(
+            title="Completeness doc",
+            content="x",
+            checksum="json-subpath-completeness",
+        )
+        Note.objects.create(document=doc, user=user, note="a note")
+        CustomFieldInstance.objects.create(
+            document=doc,
+            field=field,
+            value_text="a value",
+        )
+        backend.add_or_update(doc)
+
+        index = backend._index
+        searcher = index.searcher()
+        hits = searcher.search(
+            tantivy.Query.term_query(index.schema, "id", doc.pk),
+            limit=1,
+        ).hits
+        assert hits, "the document was not indexed"
+        stored = searcher.doc(hits[0][1]).to_dict()
+
+        json_fields = [f for f in PUBLIC_FIELDS if f.kind is FieldKind.JSON]
+        assert json_fields, "no JSON fields declared - fixture is stale"
+        for field_spec in json_fields:
+            stored_values = stored.get(field_spec.name)
+            assert stored_values, (
+                f"{field_spec.name} was not written to the index at all"
+            )
+            written_keys = stored_values[0].keys()
+            for subpath in field_spec.subpaths:
+                assert subpath in written_keys, (
+                    f"{field_spec.name}.{subpath} is declared in PUBLIC_FIELDS "
+                    "but _backend.py's document builder never writes it - it "
+                    "would be queryable but always empty"
+                )

src/documents/tests/search/test_keyword_pattern_literal.py

@@ -0,0 +1,60 @@
+"""Wildcard patterns on KEYWORD fields must stay literal.
+
+``checksum`` is the only KEYWORD field: it is indexed with the raw tokenizer,
+so its terms are never lowercased, folded or stemmed. Running its wildcard
+patterns through the stemming normalizer rewrote hex prefixes ("ceded" ->
+"cede") and returned documents whose checksum did not start with what the user
+typed, which for an identity field is a wrong answer.
+
+This covers only the registry-level normalizer, which is all that exists to
+prove at this point in the stack: user queries are not yet routed through
+whoosh-compat (that lands with the query-layer PR), so the same fact proven
+end to end against real indexed documents lives in
+``test_checksum_prefix_queries.py``.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from documents.search._registry import get_field_registry
+
+if TYPE_CHECKING:
+    from whoosh_compat import FieldRegistry
+    from whoosh_compat import PatternNormalizer
+
+pytestmark = [pytest.mark.search, pytest.mark.django_db]
+
+
+def _normalizer(registry: FieldRegistry, name: str) -> PatternNormalizer:
+    ref = registry.make_ref(name)
+    assert ref is not None
+    resolved = registry.resolve(ref)
+    assert resolved is not None
+    assert resolved.spec.pattern_normalizer is not None
+    return resolved.spec.pattern_normalizer
+
+
+class TestKeywordPatternNormalizer:
+    @pytest.mark.parametrize(
+        "run",
+        [
+            pytest.param("ceded", id="stems_to_cede"),
+            pytest.param("added", id="stems_to_ad"),
+            pytest.param("cafed", id="stems_to_cafe"),
+        ],
+    )
+    def test_keyword_runs_are_folded_not_stemmed(self, run: str) -> None:
+        """One form, the run as typed: a KEYWORD pattern must never be widened
+        to a stem, which would return checksums that do not start with what
+        the user typed."""
+        normalize = _normalizer(get_field_registry("en"), "checksum")
+        assert normalize(run) == run
+
+    def test_text_runs_still_offer_their_stem(self) -> None:
+        """A TEXT field offers the stem alongside the typed run, so a term
+        matching either one is reachable."""
+        normalize = _normalizer(get_field_registry("en"), "title")
+        assert tuple(normalize("Running")) == ("running", "run")

src/documents/tests/search/test_registry.py

@@ -0,0 +1,149 @@
+from collections.abc import Sequence
+
+import pytest
+from whoosh_compat import FieldKind
+from whoosh_compat import FieldRegistry
+from whoosh_compat.fields import ResolvedField
+
+from documents.search._fields import PUBLIC_FIELDS
+from documents.search._registry import get_field_registry
+
+
+@pytest.fixture
+def registry() -> FieldRegistry:
+    return get_field_registry(None)
+
+
+def _resolve(registry: FieldRegistry, name: str) -> ResolvedField:
+    ref = registry.make_ref(name)
+    assert ref is not None, f"{name} is not a valid field ref"
+    resolved = registry.resolve(ref)
+    assert resolved is not None, f"{name} did not resolve"
+    return resolved
+
+
+def _distinct_forms(result: str | Sequence[str]) -> tuple[str, ...]:
+    """The forms a term may match, in order, the way whoosh-compat's emitter
+    reads a pattern_normalizer's answer: a bare str is one form, a sequence is
+    several, deduplicated."""
+    if isinstance(result, str):
+        return (result,)
+    return tuple(dict.fromkeys(result))
+
+
+class TestFieldRegistry:
+    def test_internal_id_fields_are_not_registered(
+        self,
+        registry: FieldRegistry,
+    ) -> None:
+        for name in (
+            "tag_id",
+            "owner_id",
+            "viewer_id",
+            "correspondent_id",
+            "document_type_id",
+            "storage_path_id",
+            "viewer_group_id",
+        ):
+            assert name not in registry
+
+    def test_no_queryable_field_name_ends_in_id(self) -> None:
+        # The list above names the seven that were dropped; this catches the
+        # eighth. Internal *_id columns are written for permission filtering
+        # and joins, and whoosh only exposed them as query fields by accident,
+        # so a new one reaching the query surface is a leak rather than a
+        # feature. Checked against PUBLIC_FIELDS rather than the registry so
+        # an internal field is caught where it is declared.
+        leaked = [f.name for f in PUBLIC_FIELDS if f.name.endswith("_id")]
+        assert not leaked, f"internal id fields reached the query surface: {leaked}"
+
+    def test_type_alias_resolves_to_document_type(
+        self,
+        registry: FieldRegistry,
+    ) -> None:
+        assert _resolve(registry, "type").spec.name == "document_type"
+
+    def test_path_alias_resolves_to_storage_path(self, registry: FieldRegistry) -> None:
+        assert _resolve(registry, "path").spec.name == "storage_path"
+
+    def test_notes_json_subpaths_resolve(self, registry: FieldRegistry) -> None:
+        resolved = _resolve(registry, "notes.user")
+        assert resolved.spec.name == "notes"
+        assert resolved.json_path == "user"
+        assert resolved.is_subpath is True
+
+    def test_custom_fields_json_subpaths_resolve(self, registry: FieldRegistry) -> None:
+        for raw in ("custom_fields.name", "custom_fields.value"):
+            _resolve(registry, raw)
+
+    def test_unregistered_json_subpath_does_not_resolve(
+        self,
+        registry: FieldRegistry,
+    ) -> None:
+        # An unregistered subpath is not even a valid FieldRef: make_ref
+        # returns None for a dotted name whose subpath isn't registered
+        # (it doesn't produce a ref for resolve() to then reject).
+        assert registry.make_ref("notes.bogus") is None
+
+    def test_tag_is_comma_values(self, registry: FieldRegistry) -> None:
+        assert _resolve(registry, "tag").spec.comma_values is True
+
+    def test_correspondent_is_not_comma_values(self, registry: FieldRegistry) -> None:
+        # "tag" is the only field that opts in. This is only observable here:
+        # end to end the two readings of "correspondent:foo,bar" agree,
+        # because the analyzer splits the literal value on the comma anyway,
+        # so a result-level test cannot tell a value list from literal text.
+        assert _resolve(registry, "correspondent").spec.comma_values is False
+
+    def test_created_is_date_kind(self, registry: FieldRegistry) -> None:
+        resolved = _resolve(registry, "created")
+        assert resolved.spec.kind is FieldKind.DATE
+        assert resolved.spec.date_only is True
+
+    def test_analyzer_lowercases_and_ascii_folds(self, registry: FieldRegistry) -> None:
+        # title uses the paperless_text analyzer: simple -> remove_long ->
+        # lowercase -> ascii_fold [-> stemmer]. With no language configured
+        # (None), no stemmer runs, so "Café" folds to the single token "cafe".
+        resolved = _resolve(registry, "title")
+        assert resolved.spec.analyzer is not None
+        assert resolved.spec.analyzer("Café") == ["cafe"]
+
+    def test_checksum_analyzer_is_identity_single_token(
+        self,
+        registry: FieldRegistry,
+    ) -> None:
+        # checksum uses the raw tokenizer at index time (no splitting).
+        resolved = _resolve(registry, "checksum")
+        assert resolved.spec.analyzer is not None
+        assert resolved.spec.analyzer("ABC-123") == ["ABC-123"]
+
+    def test_pattern_normalizer_follows_the_registry_language(
+        self,
+        registry: FieldRegistry,
+    ) -> None:
+        # Index terms are stemmed, so patterns offer their stem too, using the
+        # registry's own language: "Running" has to reach the indexed "run".
+        # Without a language the index holds surface forms, so there is no
+        # second form and the run is only case/accent-folded.
+        resolved = _resolve(registry, "title")
+        assert resolved.spec.pattern_normalizer is not None
+        assert _distinct_forms(resolved.spec.pattern_normalizer("Running")) == (
+            "running",
+        )
+
+        resolved_en = _resolve(get_field_registry("en"), "title")
+        assert resolved_en.spec.pattern_normalizer is not None
+        assert _distinct_forms(resolved_en.spec.pattern_normalizer("Running")) == (
+            "running",
+            "run",
+        )
+
+    def test_registry_is_cached_per_language(self) -> None:
+        a = get_field_registry("en")
+        b = get_field_registry("en")
+        assert a is b
+
+    def test_registry_rebuilds_on_language_change(self) -> None:
+        a = get_field_registry("en")
+        b = get_field_registry("de")
+        assert a is not b

src/documents/tests/search/test_schema.py

@@ -1,12 +1,20 @@
 from __future__ import annotations
 
 import json
+from datetime import UTC
+from datetime import datetime
 from typing import TYPE_CHECKING
 
 import pytest
+import tantivy
 
+from documents.search._fields import PUBLIC_FIELDS
 from documents.search._schema import SCHEMA_VERSION
+from documents.search._schema import build_schema
+from documents.search._schema import field_descriptors
 from documents.search._schema import needs_rebuild
+from documents.search._schema import schema_fingerprint
+from documents.search._tokenizer import register_tokenizers
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -30,7 +38,13 @@ class TestNeedsRebuild:
     ) -> None:
         settings.SEARCH_LANGUAGE = "en"
         (index_dir / ".index_settings.json").write_text(
-            json.dumps({"schema_version": SCHEMA_VERSION, "language": "en"}),
+            json.dumps(
+                {
+                    "schema_version": SCHEMA_VERSION,
+                    "language": "en",
+                    "schema_fingerprint": schema_fingerprint(),
+                },
+            ),
         )
         assert needs_rebuild(index_dir) is False
 
@@ -77,3 +91,72 @@ class TestNeedsRebuild:
             json.dumps({"schema_version": SCHEMA_VERSION, "language": "en"}),
         )
         assert needs_rebuild(index_dir) is True
+
+
+def _schema_fields(schema: tantivy.Schema) -> dict[str, dict]:
+    """{name: field-state} for every field declared on a tantivy Schema.
+
+    tantivy-py 0.26 exposes no public introspection API on Schema (no
+    __iter__, get_field, to_json, etc.) -- __reduce__() (used internally for
+    pickling) is the only way to recover the field list, so we lean on it
+    here for test assertions only.
+    """
+    state = schema.__reduce__()[1][0]
+    return {field["name"]: field for field in state["inner"]}
+
+
+class TestSchemaMatchesPublicFields:
+    def test_every_public_field_is_in_the_schema(self) -> None:
+        schema = build_schema()
+        schema_field_names = set(_schema_fields(schema))
+        for field in PUBLIC_FIELDS:
+            assert field.name in schema_field_names, (
+                f"{field.name} is in PUBLIC_FIELDS but missing from build_schema()"
+            )
+
+    def test_asn_page_count_num_notes_are_fast_unsigned_fields(self) -> None:
+        # Spot-check kind-derived construction for the U64 fields.
+        schema = build_schema()
+        doc = tantivy.Document()
+        doc.add_unsigned("id", 1)
+        doc.add_text("checksum", "x")
+        doc.add_unsigned("asn", 42)
+        doc.add_unsigned("page_count", 3)
+        doc.add_unsigned("num_notes", 0)
+        doc.add_date("created", datetime(2020, 1, 1, tzinfo=UTC))
+        doc.add_date("modified", datetime(2020, 1, 1, tzinfo=UTC))
+        doc.add_date("added", datetime(2020, 1, 1, tzinfo=UTC))
+        index = tantivy.Index(schema)
+        register_tokenizers(index, None)
+        writer = index.writer()
+        writer.add_document(doc)
+        writer.commit()
+        index.reload()
+        searcher = index.searcher()
+        results = searcher.search(tantivy.Query.term_query(schema, "asn", 42), limit=1)
+        assert len(results.hits) == 1
+
+
+class TestFastFlagAgreement:
+    def test_every_public_field_fast_flag_matches_the_built_schema(self) -> None:
+        # whoosh-compat's registry trusts PUBLIC_FIELDS' fast flag when resolving
+        # field:* existence checks (its FAST_FIELD strategy); a fast=True
+        # entry whose actual tantivy column is not fast would make those
+        # searches silently match nothing at search time. Only the U64 and
+        # DATE descriptors can carry the flag today, so this
+        # pins the agreement for EVERY kind: a future fast=True
+        # TEXT/KEYWORD/JSON entry the builder silently ignores fails here
+        # instead of at a user's query.
+        #
+        # field_descriptors() (not tantivy-py's __reduce__() pickling
+        # internals) is used as the probe here: it is exactly the input
+        # build_schema()'s SchemaBuilder consumes for the `fast` kwarg on
+        # every field kind, so it pins the same agreement without depending
+        # on a private pickled representation surviving a tantivy-py
+        # upgrade.
+        descriptor_fast = {d.name: d.fast for d in field_descriptors()}
+        for public_field in PUBLIC_FIELDS:
+            assert descriptor_fast[public_field.name] == public_field.fast, (
+                f"{public_field.name}: PUBLIC_FIELDS says fast={public_field.fast} but"
+                f" field_descriptors() says fast={descriptor_fast[public_field.name]}"
+            )

src/documents/tests/search/test_schema_fingerprint.py

@@ -0,0 +1,492 @@
+"""The schema fingerprint stamped into .index_settings.json.
+
+tantivy compares schemas by *ordered* field list, and `tantivy.Index(schema,
+path=...)` (what every write path does) raises on any difference. SCHEMA_VERSION
+is the manual guard against that, but build_schema() is edited for *parser*
+reasons - adding an alias, flipping fast=True, adding a subpath - by people not
+thinking about the on-disk index, and forgetting the bump is exactly how this
+branch's bug happened.
+
+The fingerprint is the automatic guard: it hashes the field descriptor list that
+build_schema() itself iterates, so any change to a field's name, kind, options
+or *position* forces a rebuild on its own.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from typing import TYPE_CHECKING
+
+import pytest
+import tantivy
+
+from documents.search import _schema
+from documents.search._schema import SCHEMA_VERSION
+from documents.search._schema import FieldDescriptor
+from documents.search._schema import _write_sentinels
+from documents.search._schema import build_schema
+from documents.search._schema import field_descriptors
+from documents.search._schema import needs_rebuild
+from documents.search._schema import schema_fingerprint
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from pytest_django.fixtures import SettingsWrapper
+
+pytestmark = pytest.mark.search
+
+# The on-disk field layout of a v2 index, pinned as data. Any edit here is an
+# index-format change: it must come with a rebuild, which the fingerprint now
+# forces automatically. Reproduced from build_schema()'s output as it stood
+# before the descriptor refactor, so it also pins that the refactor changed
+# nothing.
+PINNED_DESCRIPTORS: tuple[FieldDescriptor, ...] = (
+    FieldDescriptor("id", "u64", stored=True, indexed=True, fast=True, tokenizer=None),
+    FieldDescriptor(
+        "title",
+        "text",
+        stored=True,
+        indexed=True,
+        fast=False,
+        tokenizer="paperless_text",
+    ),
+    FieldDescriptor(
+        "content",
+        "text",
+        stored=True,
+        indexed=True,
+        fast=False,
+        tokenizer="paperless_text",
+    ),
+    FieldDescriptor(
+        "correspondent",
+        "text",
+        stored=True,
+        indexed=True,
+        fast=False,
+        tokenizer="paperless_text",
+    ),
+    FieldDescriptor(
+        "document_type",
+        "text",
+        stored=True,
+        indexed=True,
+        fast=False,
+        tokenizer="paperless_text",
+    ),
+    FieldDescriptor(
+        "storage_path",
+        "text",
+        stored=True,
+        indexed=True,
+        fast=False,
+        tokenizer="paperless_text",
+    ),
+    FieldDescriptor(
+        "original_filename",
+        "text",
+        stored=True,
+        indexed=True,
+        fast=False,
+        tokenizer="paperless_text",
+    ),
+    FieldDescriptor(
+        "tag",
+        "text",
+        stored=True,
+        indexed=True,
+        fast=False,
+        tokenizer="paperless_text",
+    ),
+    FieldDescriptor(
+        "checksum",
+        "text",
+        stored=True,
+        indexed=True,
+        fast=False,
+        tokenizer="raw",
+    ),
+    FieldDescriptor("asn", "u64", stored=True, indexed=True, fast=True, tokenizer=None),
+    FieldDescriptor(
+        "page_count",
+        "u64",
+        stored=True,
+        indexed=True,
+        fast=True,
+        tokenizer=None,
+    ),
+    FieldDescriptor(
+        "num_notes",
+        "u64",
+        stored=True,
+        indexed=True,
+        fast=True,
+        tokenizer=None,
+    ),
+    FieldDescriptor(
+        "created",
+        "date",
+        stored=True,
+        indexed=True,
+        fast=True,
+        tokenizer=None,
+    ),
+    FieldDescriptor(
+        "modified",
+        "date",
+        stored=True,
+        indexed=True,
+        fast=True,
+        tokenizer=None,
+    ),
+    FieldDescriptor(
+        "added",
+        "date",
+        stored=True,
+        indexed=True,
+        fast=True,
+        tokenizer=None,
+    ),
+    FieldDescriptor(
+        "notes",
+        "json",
+        stored=True,
+        indexed=True,
+        fast=False,
+        tokenizer="paperless_text",
+    ),
+    FieldDescriptor(
+        "notes_text",
+        "text",
+        stored=True,
+        indexed=True,
+        fast=False,
+        tokenizer="paperless_text",
+    ),
+    FieldDescriptor(
+        "custom_fields",
+        "json",
+        stored=True,
+        indexed=True,
+        fast=False,
+        tokenizer="paperless_text",
+    ),
+    FieldDescriptor(
+        "title_sort",
+        "text",
+        stored=False,
+        indexed=True,
+        fast=True,
+        tokenizer="simple_analyzer",
+    ),
+    FieldDescriptor(
+        "correspondent_sort",
+        "text",
+        stored=False,
+        indexed=True,
+        fast=True,
+        tokenizer="simple_analyzer",
+    ),
+    FieldDescriptor(
+        "type_sort",
+        "text",
+        stored=False,
+        indexed=True,
+        fast=True,
+        tokenizer="simple_analyzer",
+    ),
+    FieldDescriptor(
+        "bigram_content",
+        "text",
+        stored=False,
+        indexed=True,
+        fast=False,
+        tokenizer="bigram_analyzer",
+    ),
+    FieldDescriptor(
+        "bigram_title",
+        "text",
+        stored=False,
+        indexed=True,
+        fast=False,
+        tokenizer="bigram_analyzer",
+    ),
+    FieldDescriptor(
+        "bigram_correspondent",
+        "text",
+        stored=False,
+        indexed=True,
+        fast=False,
+        tokenizer="bigram_analyzer",
+    ),
+    FieldDescriptor(
+        "bigram_document_type",
+        "text",
+        stored=False,
+        indexed=True,
+        fast=False,
+        tokenizer="bigram_analyzer",
+    ),
+    FieldDescriptor(
+        "bigram_tag",
+        "text",
+        stored=False,
+        indexed=True,
+        fast=False,
+        tokenizer="bigram_analyzer",
+    ),
+    FieldDescriptor(
+        "simple_title",
+        "text",
+        stored=False,
+        indexed=True,
+        fast=False,
+        tokenizer="simple_search_analyzer",
+    ),
+    FieldDescriptor(
+        "simple_content",
+        "text",
+        stored=False,
+        indexed=True,
+        fast=False,
+        tokenizer="simple_search_analyzer",
+    ),
+    FieldDescriptor(
+        "autocomplete_word",
+        "text",
+        stored=False,
+        indexed=True,
+        fast=False,
+        tokenizer="raw",
+    ),
+    FieldDescriptor(
+        "owner_id",
+        "u64",
+        stored=False,
+        indexed=True,
+        fast=True,
+        tokenizer=None,
+    ),
+    FieldDescriptor(
+        "viewer_id",
+        "u64",
+        stored=False,
+        indexed=True,
+        fast=True,
+        tokenizer=None,
+    ),
+    FieldDescriptor(
+        "viewer_group_id",
+        "u64",
+        stored=False,
+        indexed=True,
+        fast=True,
+        tokenizer=None,
+    ),
+)
+
+
+def _schema_fields(schema: tantivy.Schema) -> list[dict]:
+    """The tantivy-level field list, in declaration order.
+
+    tantivy-py 0.26 exposes no public introspection API on Schema, so
+    __reduce__() (its pickling hook) is the only way to recover the field list.
+    It is used here, in a test, precisely because it is the representation the
+    persisted fingerprint must NOT depend on.
+    """
+    return schema.__reduce__()[1][0]["inner"]
+
+
+def _sentinels(index_dir: Path, **overrides: object) -> None:
+    data = {
+        "schema_version": SCHEMA_VERSION,
+        "language": None,
+        "schema_fingerprint": schema_fingerprint(),
+    }
+    data.update(overrides)
+    (index_dir / ".index_settings.json").write_text(json.dumps(data))
+
+
+class TestDescriptorsDescribeTheBuiltSchema:
+    def test_descriptors_match_the_pinned_field_layout(self) -> None:
+        assert tuple(field_descriptors()) == PINNED_DESCRIPTORS
+
+    def test_built_schema_matches_the_descriptors(self) -> None:
+        """The descriptors are not a parallel description - they are the input.
+
+        Reading the built schema back proves the loop honours every option, so
+        a descriptor edit cannot claim a shape the SchemaBuilder did not build.
+        """
+        kinds = {"text": "text", "json": "json_object", "u64": "u64", "date": "date"}
+        built = [
+            (
+                field["name"],
+                field["type"],
+                field["options"]["stored"],
+                bool(field["options"].get("fast")),
+                (field["options"].get("indexing") or {}).get("tokenizer"),
+            )
+            for field in _schema_fields(build_schema())
+        ]
+        expected = [
+            (
+                descriptor.name,
+                kinds[descriptor.kind],
+                descriptor.stored,
+                descriptor.fast,
+                descriptor.tokenizer,
+            )
+            for descriptor in field_descriptors()
+        ]
+        assert built == expected
+
+
+class TestFingerprintSensitivity:
+    def test_a_field_option_change_moves_the_fingerprint(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        before = schema_fingerprint()
+        changed = field_descriptors()
+        changed[1] = changed[1]._replace(fast=True)
+        monkeypatch.setattr(_schema, "field_descriptors", lambda: changed)
+
+        assert schema_fingerprint() != before
+
+    def test_reordering_alone_moves_the_fingerprint(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """The original bug: same fields, different declaration order.
+
+        A set- or dict-based fingerprint would be blind to this, and tantivy
+        would reject every write against the existing index.
+        """
+        before = schema_fingerprint()
+        swapped = field_descriptors()
+        swapped[1], swapped[2] = swapped[2], swapped[1]
+        monkeypatch.setattr(_schema, "field_descriptors", lambda: swapped)
+
+        assert schema_fingerprint() != before
+
+    def test_repeated_calls_agree(self) -> None:
+        assert schema_fingerprint() == schema_fingerprint()
+
+
+class TestFingerprintIsIndependentOfTantivy:
+    def test_a_tantivy_option_key_addition_would_not_move_it(self) -> None:
+        """A tantivy-py upgrade must not force a global reindex.
+
+        Hashing schema.__reduce__() would do exactly that: the simulated new
+        option key below changes that payload for every user with no schema
+        change at all.
+        """
+        fields = _schema_fields(build_schema())
+        upgraded = [
+            {**field, "options": {**field["options"], "coerce": True}}
+            for field in fields
+        ]
+        assert _hash(upgraded) != _hash(fields)
+        assert schema_fingerprint() == _fingerprint_of(field_descriptors())
+
+    def test_fingerprint_never_touches_the_schema_builder(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        before = schema_fingerprint()
+
+        class _RemovedSchemaBuilder:
+            def __init__(self) -> None:
+                raise AssertionError("tantivy.SchemaBuilder was consulted")
+
+        monkeypatch.setattr(tantivy, "SchemaBuilder", _RemovedSchemaBuilder)
+        with pytest.raises(AssertionError):
+            build_schema()
+
+        assert schema_fingerprint() == before
+
+
+def _hash(payload: object) -> str:
+    return hashlib.blake2b(json.dumps(payload).encode()).hexdigest()
+
+
+def _fingerprint_of(descriptors: list[FieldDescriptor]) -> str:
+    return _hash([list(descriptor) for descriptor in descriptors])
+
+
+class TestNeedsRebuildOnFingerprint:
+    def test_matching_fingerprint_does_not_rebuild(
+        self,
+        index_dir: Path,
+        settings: SettingsWrapper,
+    ) -> None:
+        settings.SEARCH_LANGUAGE = None
+        _sentinels(index_dir)
+
+        assert needs_rebuild(index_dir) is False
+
+    def test_stale_fingerprint_rebuilds_despite_a_matching_version(
+        self,
+        index_dir: Path,
+        settings: SettingsWrapper,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """The failure this task exists to prevent: schema edited, version not
+        bumped. Without the fingerprint check, `reindex --if-needed` reports the
+        index up to date and every write then raises."""
+        settings.SEARCH_LANGUAGE = None
+        _sentinels(index_dir)
+        extended = [
+            *field_descriptors(),
+            FieldDescriptor(
+                "new_field",
+                "u64",
+                stored=False,
+                indexed=True,
+                fast=True,
+                tokenizer=None,
+            ),
+        ]
+        monkeypatch.setattr(_schema, "field_descriptors", lambda: extended)
+
+        assert needs_rebuild(index_dir) is True
+
+    def test_reordered_schema_rebuilds(
+        self,
+        index_dir: Path,
+        settings: SettingsWrapper,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        settings.SEARCH_LANGUAGE = None
+        _sentinels(index_dir)
+        reordered = field_descriptors()
+        reordered[1], reordered[2] = reordered[2], reordered[1]
+        monkeypatch.setattr(_schema, "field_descriptors", lambda: reordered)
+
+        assert needs_rebuild(index_dir) is True
+
+    def test_missing_fingerprint_rebuilds(
+        self,
+        index_dir: Path,
+        settings: SettingsWrapper,
+    ) -> None:
+        """No seeding: an index whose schema shape nobody recorded is rebuilt
+        rather than trusted."""
+        settings.SEARCH_LANGUAGE = None
+        (index_dir / ".index_settings.json").write_text(
+            json.dumps({"schema_version": SCHEMA_VERSION, "language": None}),
+        )
+
+        assert needs_rebuild(index_dir) is True
+
+    def test_written_sentinels_satisfy_the_check(
+        self,
+        index_dir: Path,
+        settings: SettingsWrapper,
+    ) -> None:
+        settings.SEARCH_LANGUAGE = "en"
+        _write_sentinels(index_dir)
+
+        assert needs_rebuild(index_dir) is False

src/documents/tests/search/test_schema_version.py

@@ -0,0 +1,164 @@
+"""SCHEMA_VERSION must change whenever build_schema()'s field list or order does.
+
+tantivy compares schemas by *ordered* field list. ``Index.open()`` loads the
+schema from the index's own ``meta.json``, so reads against an index built by an
+older release keep working after a field reorder. Writes do not:
+``WriteBatch.__enter__`` calls ``tantivy.Index(build_schema(), path=...)``, an
+open-or-create that raises ``ValueError`` on any schema difference. Nothing
+catches that ValueError, so consumption, index_document and bulk edit all
+hard-fail while ``/api/status/`` still reports the index healthy.
+
+The only thing that saves such an install is ``needs_rebuild()`` noticing the
+version stamped in ``.index_settings.json`` is stale.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+
+import pytest
+import tantivy
+from django.conf import settings as django_settings
+
+from documents.search._schema import build_schema
+from documents.search._schema import needs_rebuild
+from documents.search._schema import open_or_rebuild_index
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+pytestmark = [pytest.mark.search]
+
+RELEASED_V1_SCHEMA_VERSION = 1
+
+
+def _build_released_v1_schema() -> tantivy.Schema:
+    """Frozen copy of build_schema() as shipped in v3.0.x (schema version 1).
+
+    Deliberately duplicated rather than imported: it must keep describing the
+    on-disk layout of already-deployed indexes even as build_schema() evolves.
+    """
+    sb = tantivy.SchemaBuilder()
+
+    sb.add_unsigned_field("id", stored=True, indexed=True, fast=True)
+    sb.add_text_field("checksum", stored=True, tokenizer_name="raw")
+
+    for field in (
+        "title",
+        "correspondent",
+        "document_type",
+        "storage_path",
+        "original_filename",
+        "content",
+    ):
+        sb.add_text_field(field, stored=True, tokenizer_name="paperless_text")
+
+    for field in ("title_sort", "correspondent_sort", "type_sort"):
+        sb.add_text_field(
+            field,
+            stored=False,
+            tokenizer_name="simple_analyzer",
+            fast=True,
+        )
+
+    for field in (
+        "bigram_content",
+        "bigram_title",
+        "bigram_correspondent",
+        "bigram_document_type",
+        "bigram_tag",
+    ):
+        sb.add_text_field(field, stored=False, tokenizer_name="bigram_analyzer")
+
+    for field in ("simple_title", "simple_content"):
+        sb.add_text_field(field, stored=False, tokenizer_name="simple_search_analyzer")
+
+    sb.add_text_field("autocomplete_word", stored=False, tokenizer_name="raw")
+    sb.add_text_field("tag", stored=True, tokenizer_name="paperless_text")
+
+    sb.add_json_field("notes", stored=True, tokenizer_name="paperless_text")
+    sb.add_text_field("notes_text", stored=True, tokenizer_name="paperless_text")
+    sb.add_json_field("custom_fields", stored=True, tokenizer_name="paperless_text")
+
+    for field in (
+        "correspondent_id",
+        "document_type_id",
+        "storage_path_id",
+        "tag_id",
+        "owner_id",
+        "viewer_id",
+        "viewer_group_id",
+    ):
+        sb.add_unsigned_field(field, stored=False, indexed=True, fast=True)
+
+    for field in ("created", "modified", "added"):
+        sb.add_date_field(field, stored=True, indexed=True, fast=True)
+
+    for field in ("asn", "page_count", "num_notes"):
+        sb.add_unsigned_field(field, stored=True, indexed=True, fast=True)
+
+    return sb.build()
+
+
+@pytest.fixture
+def released_v1_index(tmp_path: Path) -> Path:
+    """An index directory as a v3.0.x install would leave it on disk."""
+    index_dir = tmp_path / "index"
+    index_dir.mkdir()
+    tantivy.Index(_build_released_v1_schema(), path=str(index_dir))
+    (index_dir / ".index_settings.json").write_text(
+        json.dumps(
+            {
+                "schema_version": RELEASED_V1_SCHEMA_VERSION,
+                "language": django_settings.SEARCH_LANGUAGE,
+            },
+        ),
+    )
+    return index_dir
+
+
+class TestUpgradeFromReleasedV1Index:
+    def test_released_v1_index_is_flagged_for_rebuild(
+        self,
+        released_v1_index: Path,
+    ) -> None:
+        """The current schema differs from v1's, so the sentinel must be stale.
+
+        If this fails, `document_index reindex --if-needed` prints "Search index
+        is up to date" and skips, leaving the mismatched index in place.
+        """
+        assert needs_rebuild(released_v1_index) is True
+
+    def test_v1_index_rejects_writes_against_the_current_schema(
+        self,
+        released_v1_index: Path,
+    ) -> None:
+        """The failure mode the version bump exists to prevent.
+
+        This is exactly what WriteBatch.__enter__ does on every index write.
+        """
+        with pytest.raises(ValueError, match="schema does not match"):
+            tantivy.Index(build_schema(), path=str(released_v1_index))
+
+    def test_opening_a_v1_index_leaves_it_writable(
+        self,
+        released_v1_index: Path,
+    ) -> None:
+        """End to end: open_or_rebuild_index must hand back an index that the
+        write path can reopen. Before the version bump, needs_rebuild() returned
+        False here, the stale directory survived untouched, and every subsequent
+        write raised the ValueError above."""
+        open_or_rebuild_index(released_v1_index)
+
+        tantivy.Index(build_schema(), path=str(released_v1_index))
+
+    def test_rebuilt_index_is_not_rebuilt_again(
+        self,
+        released_v1_index: Path,
+    ) -> None:
+        """The rebuild must stamp the version it actually wrote, otherwise every
+        startup wipes and reindexes the whole corpus."""
+        open_or_rebuild_index(released_v1_index)
+
+        assert needs_rebuild(released_v1_index) is False

src/documents/tests/search/test_tokenizer.py

@@ -7,8 +7,8 @@ import pytest
 import tantivy
 
 from documents.search._tokenizer import _bigram_analyzer
-from documents.search._tokenizer import _paperless_text
 from documents.search._tokenizer import _simple_search_analyzer
+from documents.search._tokenizer import paperless_text_analyzer
 from documents.search._tokenizer import register_tokenizers
 
 if TYPE_CHECKING:
@@ -25,7 +25,7 @@ class TestTokenizers:
         sb.add_text_field("content", stored=True, tokenizer_name="paperless_text")
         schema = sb.build()
         idx = tantivy.Index(schema, path=None)
-        idx.register_tokenizer("paperless_text", _paperless_text(""))
+        idx.register_tokenizer("paperless_text", paperless_text_analyzer(""))
         return idx
 
     @pytest.fixture

src/documents/tests/test_api_chat.py

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

src/documents/tests/test_api_objects.py

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

src/documents/tests/test_bulk_edit.py

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

src/documents/tests/test_document_list_effective_content_annotation.py

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

src/documents/versioning.py

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

src/documents/views.py

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

src/paperless/middleware.py

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

src/paperless/settings/__init__.py

@@ -10,6 +10,7 @@ from pathlib import Path
 from typing import Final
 from urllib.parse import urlparse
 
+from compression_middleware.middleware import CompressionMiddleware
 from django.core.exceptions import ImproperlyConfigured
 from django.utils.translation import gettext_lazy as _
 from dotenv import load_dotenv
@@ -200,10 +201,22 @@ MIDDLEWARE = [
     "allauth.account.middleware.AccountMiddleware",
 ]
 
-# Optional to enable compression. The subclass leaves server-sent events
-# uncompressed; see paperless.middleware.StreamAwareCompressionMiddleware.
+# Optional to enable compression
 if get_bool_from_env("PAPERLESS_ENABLE_COMPRESSION", "yes"):  # pragma: no cover
-    MIDDLEWARE.insert(0, "paperless.middleware.StreamAwareCompressionMiddleware")
+    MIDDLEWARE.insert(0, "compression_middleware.middleware.CompressionMiddleware")
+
+# Workaround to not compress streaming responses (e.g. chat).
+# See https://github.com/friedelwolff/django-compression-middleware/pull/7
+original_process_response = CompressionMiddleware.process_response
+
+
+def patched_process_response(self, request, response):
+    if getattr(request, "compress_exempt", False):
+        return response
+    return original_process_response(self, request, response)
+
+
+CompressionMiddleware.process_response = patched_process_response
 
 ROOT_URLCONF = "paperless.urls"
 

src/paperless/tests/test_compression_middleware.py

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

src/paperless_ai/client.py

@@ -40,6 +40,7 @@ LLM_SYSTEM_PROMPT = (
 
 # openai-python rejects empty keys since 2.34.0, "fake" is the stand-in from
 # llama-index's own OpenAILike docs https://docs.llamaindex.ai/en/stable/api_reference/llms/openai_like/
+# TODO: remove pending resolution of https://github.com/openai/openai-python/issues/3224
 PLACEHOLDER_API_KEY: Final = "fake"
 
 

uv.lock

@@ -2932,6 +2932,7 @@ dependencies = [
     { name = "torch", version = "2.13.0+cpu", source = { registry = "https://download.pytorch.org/whl/cpu" }, marker = "sys_platform == 'linux'" },
     { name = "watchfiles" },
     { name = "whitenoise" },
+    { name = "whoosh-compat", extra = ["tantivy"] },
     { name = "zxing-cpp" },
 ]
 
@@ -3090,6 +3091,7 @@ requires-dist = [
     { name = "torch", specifier = "~=2.13.0", index = "https://download.pytorch.org/whl/cpu" },
     { name = "watchfiles", specifier = ">=1.2" },
     { name = "whitenoise", specifier = "~=6.11" },
+    { name = "whoosh-compat", extras = ["tantivy"], specifier = "==0.1.0" },
     { name = "zxing-cpp", specifier = "~=3.1.0" },
 ]
 provides-extras = ["mariadb", "postgres", "webserver"]
@@ -5638,6 +5640,23 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/db/eb/d5583a11486211f3ebd4b385545ae787f32363d453c19fffd81106c9c138/whitenoise-6.12.0-py3-none-any.whl", hash = "sha256:fc5e8c572e33ebf24795b47b6a7da8da3c00cff2349f5b04c02f28d0cc5a3cc2", size = 20302, upload-time = "2026-02-27T00:05:40.086Z" },
 ]
 
+[[package]]
+name = "whoosh-compat"
+version = "0.1.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "python-dateutil" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/5d/f7/3e45f4a484afa174cd42e424ce2b8c514ae54f564830122656ad17b765e2/whoosh_compat-0.1.0.tar.gz", hash = "sha256:86935bdc159ed9b0a06a4661d17f1251d8280340b84e218cc73d915a2edaddf7", size = 577543, upload-time = "2026-08-25T15:22:42.316Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/3f/51/9a8399d0f472814e136a2884bf910c6c59843a5a2c66afc4b5133eb531ea/whoosh_compat-0.1.0-py3-none-any.whl", hash = "sha256:3e7c5f519b4d397dbf4f8d7bbe4ca1eb6004da24c8892bc701844ed393bc97e7", size = 153875, upload-time = "2026-08-25T15:22:40.787Z" },
+]
+
+[package.optional-dependencies]
+tantivy = [
+    { name = "tantivy" },
+]
+
 [[package]]
 name = "wrapt"
 version = "2.0.1"