Test: consolidate TestStreamedDocuments into one test

update_llm_index() built its documents queryset with
prefetch_related("tags", "notes", "custom_fields__field") and iterated
it directly -- no .iterator(), so a full rebuild pulled every Document
(including content, which can be megabytes each) plus every prefetch
cache into memory simultaneously.

Add _StreamedDocuments, a thin QuerySet wrapper matching the shape of
documents/search/_backend.py's _DocumentViewerStream: iterates via
.iterator(chunk_size=1000) so Django streams from a server-side cursor
(and, since 4.1, still runs prefetch_related per batch instead of all at
once), while __len__ still supports a real progress-bar total. Used for
both loop sites (the full rebuild and the modified-time-scoped update).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

docker/rootfs/etc/s6-overlay/s6-rc.d/init-llmindex-migrate/run

@@ -1,12 +0,0 @@
-#!/command/with-contenv /usr/bin/bash
-# shellcheck shell=bash
-
-declare -r log_prefix="[init-llmindex-migrate]"
-
-echo "${log_prefix} Checking for pending LLM index migrations..."
-cd "${PAPERLESS_SRC_DIR}"
-if [[ -n "${USER_IS_NON_ROOT}" ]]; then
-	python3 manage.py document_llmindex migrate
-else
-	s6-setuidgid paperless python3 manage.py document_llmindex migrate
-fi

docker/rootfs/etc/s6-overlay/s6-rc.d/init-llmindex-migrate/type

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

docker/rootfs/etc/s6-overlay/s6-rc.d/init-llmindex-migrate/up

@@ -1 +0,0 @@
-/etc/s6-overlay/s6-rc.d/init-llmindex-migrate/run

docs/administration.md

@@ -212,16 +212,6 @@ following:
     This is a no-op if the index is already up to date, so it is safe to
     run on every upgrade.
 
-5.  Migrate the LLM index if needed.
-
-    ```shell-session
-    cd src
-    python3 manage.py document_llmindex migrate
-    ```
-
-    This is a no-op if the index schema is already current, so it is safe
-    to run on every upgrade.
-
 ### Database Upgrades
 
 Paperless-ngx is compatible with Django-supported versions of PostgreSQL and MariaDB and it is generally
@@ -542,7 +532,7 @@ index is updated automatically on the schedule set by
 can manage it manually:
 
 ```
-document_llmindex {rebuild,update,compact,migrate}
+document_llmindex {rebuild,update,compact}
 ```
 
 Specify `rebuild` to build the index from scratch from all documents in the database. Use

src-ui/messages.xlf

@@ -1849,7 +1849,7 @@
         </context-group>
         <context-group purpose="location">
           <context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
-          <context context-type="linenumber">154</context>
+          <context context-type="linenumber">159</context>
         </context-group>
         <context-group purpose="location">
           <context context-type="sourcefile">src/app/components/common/filterable-dropdown/filterable-dropdown.component.html</context>
@@ -3972,11 +3972,11 @@
         </context-group>
         <context-group purpose="location">
           <context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
-          <context context-type="linenumber">96</context>
+          <context context-type="linenumber">101</context>
         </context-group>
         <context-group purpose="location">
           <context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
-          <context context-type="linenumber">102</context>
+          <context context-type="linenumber">107</context>
         </context-group>
       </trans-unit>
       <trans-unit id="3800326155195149498" datatype="html">
@@ -3987,11 +3987,11 @@
         </context-group>
         <context-group purpose="location">
           <context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
-          <context context-type="linenumber">97</context>
+          <context context-type="linenumber">102</context>
         </context-group>
         <context-group purpose="location">
           <context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
-          <context context-type="linenumber">103</context>
+          <context context-type="linenumber">108</context>
         </context-group>
       </trans-unit>
       <trans-unit id="7551700625201096185" datatype="html">
@@ -4002,14 +4002,14 @@
         </context-group>
         <context-group purpose="location">
           <context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
-          <context context-type="linenumber">119</context>
+          <context context-type="linenumber">124</context>
         </context-group>
       </trans-unit>
       <trans-unit id="3184700926171002527" datatype="html">
         <source>Any</source>
         <context-group purpose="location">
           <context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
-          <context context-type="linenumber">152</context>
+          <context context-type="linenumber">157</context>
         </context-group>
         <context-group purpose="location">
           <context context-type="sourcefile">src/app/components/common/filterable-dropdown/filterable-dropdown.component.html</context>
@@ -4020,21 +4020,21 @@
         <source>Not</source>
         <context-group purpose="location">
           <context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
-          <context context-type="linenumber">157</context>
+          <context context-type="linenumber">162</context>
         </context-group>
       </trans-unit>
       <trans-unit id="6548676277933116532" datatype="html">
         <source>Add query</source>
         <context-group purpose="location">
           <context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
-          <context context-type="linenumber">176</context>
+          <context context-type="linenumber">181</context>
         </context-group>
       </trans-unit>
       <trans-unit id="5599577087865387184" datatype="html">
         <source>Add expression</source>
         <context-group purpose="location">
           <context context-type="sourcefile">src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html</context>
-          <context context-type="linenumber">179</context>
+          <context context-type="linenumber">184</context>
         </context-group>
       </trans-unit>
       <trans-unit id="6312759212949884929" datatype="html">

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

@@ -182,7 +182,7 @@
                     container="body" triggers="mouseenter:mouseleave" popoverClass="popover-slim">
                     <i-bs class="me-2" name="stack"></i-bs><span><ng-container i18n>Attributes</ng-container></span>
                   </a>
-                  @if (!slimSidebarEnabled) {
+                  @if (!slimSidebarEnabled && canSaveSettings) {
                     <button
                       type="button"
                       class="btn btn-link btn-sm text-muted p-0 me-3 attributes-expand-btn"

src-ui/src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.html

@@ -68,6 +68,11 @@
     ></ng-select>
   } @else if (getCustomFieldByID(atom.field)?.data_type === CustomFieldDataType.DocumentLink) {
     <pngx-input-document-link [(ngModel)]="atom.value" class="w-25 form-select doc-link-select p-0" placeholder="Search docs..." i18n-placeholder [minimal]="true"></pngx-input-document-link>
+  } @else if (getCustomFieldByID(atom.field)?.data_type === CustomFieldDataType.Monetary) {
+    <input class="w-25 form-control rounded-end" type="text" inputmode="decimal"
+      [ngModel]="atom.value"
+      (ngModelChange)="setMonetaryValue(atom, $event)"
+      [disabled]="disabled">
   } @else {
     <input class="w-25 form-control rounded-end" type="text" [(ngModel)]="atom.value" [disabled]="disabled">
   }

src-ui/src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.spec.ts

@@ -1,5 +1,6 @@
 import { provideHttpClient, withInterceptorsFromDi } from '@angular/common/http'
 import { provideHttpClientTesting } from '@angular/common/http/testing'
+import { LOCALE_ID } from '@angular/core'
 import { ComponentFixture, TestBed } from '@angular/core/testing'
 import { FormsModule, ReactiveFormsModule } from '@angular/forms'
 import { NgbDropdownModule } from '@ng-bootstrap/ng-bootstrap'
@@ -41,6 +42,12 @@ const customFields = [
       ],
     },
   },
+  {
+    id: 3,
+    name: 'Test Monetary Field',
+    data_type: CustomFieldDataType.Monetary,
+    extra_data: { default_currency: 'EUR' },
+  },
 ]
 
 describe('CustomFieldsQueryDropdownComponent', () => {
@@ -61,6 +68,7 @@ describe('CustomFieldsQueryDropdownComponent', () => {
       providers: [
         provideHttpClient(withInterceptorsFromDi()),
         provideHttpClientTesting(),
+        { provide: LOCALE_ID, useValue: 'de' },
       ],
     }).compileComponents()
 
@@ -150,6 +158,22 @@ describe('CustomFieldsQueryDropdownComponent', () => {
     expect(options2).toEqual([])
   })
 
+  it('should normalize localized monetary comparison values', () => {
+    const atom = new CustomFieldQueryAtom([3, 'exact', null])
+
+    component.setMonetaryValue(atom, '1.234,56')
+
+    expect(atom.value).toEqual('1234.56')
+  })
+
+  it('should preserve API-formatted monetary comparison values', () => {
+    const atom = new CustomFieldQueryAtom([3, 'exact', null])
+
+    component.setMonetaryValue(atom, '1234.56')
+
+    expect(atom.value).toEqual('1234.56')
+  })
+
   it('should remove an element from the selection model', () => {
     const expression = new CustomFieldQueryExpression()
     const atom = new CustomFieldQueryAtom()

src-ui/src/app/components/common/custom-fields-query-dropdown/custom-fields-query-dropdown.component.ts

@@ -1,9 +1,14 @@
-import { NgTemplateOutlet } from '@angular/common'
+import {
+  getLocaleNumberSymbol,
+  NgTemplateOutlet,
+  NumberSymbol,
+} from '@angular/common'
 import {
   Component,
   EventEmitter,
   inject,
   Input,
+  LOCALE_ID,
   Output,
   QueryList,
   signal,
@@ -212,6 +217,7 @@ export class CustomFieldQueriesModel {
 })
 export class CustomFieldsQueryDropdownComponent extends LoadingComponentWithPermissions {
   protected customFieldsService = inject(CustomFieldsService)
+  private readonly locale = inject(LOCALE_ID)
 
   public CustomFieldQueryComponentType = CustomFieldQueryElementType
   public CustomFieldQueryOperator = CustomFieldQueryOperator
@@ -376,4 +382,18 @@ export class CustomFieldsQueryDropdownComponent extends LoadingComponentWithPerm
     }
     return []
   }
+
+  setMonetaryValue(atom: CustomFieldQueryAtom, value: string) {
+    // Normalize the decimal symbol e.g. . vs , by locale
+    const decimalSymbol = getLocaleNumberSymbol(
+      this.locale,
+      NumberSymbol.Decimal
+    )
+    if (decimalSymbol !== '.' && value.includes(decimalSymbol)) {
+      const groupSymbol = getLocaleNumberSymbol(this.locale, NumberSymbol.Group)
+      value = value.split(groupSymbol).join('').split(decimalSymbol).join('.')
+    }
+
+    atom.value = value
+  }
 }

src-ui/src/app/components/manage/document-attributes/document-attributes.component.html

@@ -51,8 +51,8 @@
       *pngxIfPermissions="{ action: PermissionAction.Add, type: activeManagementList.permissionType }">
       <i-bs name="plus-circle" class="me-1"></i-bs><ng-container i18n>Create</ng-container>
     </button>
-  } @else if (activeCustomFields) {
-    <button type="button" class="btn btn-sm btn-outline-primary" (click)="activeCustomFields.editField()"
+  } @else if (customFieldsActive) {
+    <button type="button" class="btn btn-sm btn-outline-primary" (click)="addCustomField()"
       *pngxIfPermissions="{ action: PermissionAction.Add, type: PermissionType.CustomField }">
       <i-bs name="plus-circle" class="me-1"></i-bs><ng-container i18n>Add Field</ng-container>
     </button>

src-ui/src/app/components/manage/document-attributes/document-attributes.component.spec.ts

@@ -18,6 +18,7 @@ import {
   DocumentAttributesComponent,
   DocumentAttributesSectionKind,
 } from './document-attributes.component'
+import { CustomFieldsComponent } from './custom-fields/custom-fields.component'
 import { ManagementListComponent } from './management-list/management-list.component'
 
 @Component({
@@ -207,6 +208,29 @@ describe('DocumentAttributesComponent', () => {
     expect(component.activeSection.kind).toBe(
       DocumentAttributesSectionKind.CustomFields
     )
-    expect(component.activeCustomFields).toBeDefined()
+    const customFields = Object.create(CustomFieldsComponent.prototype)
+    customFields.editField = jest.fn()
+    component.activeOutlet = {
+      componentInstance: customFields,
+    } as any
+    expect(component.activeCustomFields).toBe(customFields)
+
+    component.addCustomField()
+    expect(customFields.editField).toHaveBeenCalled()
+  })
+
+  it('should show the add field button before the custom fields instance is available', async () => {
+    jest.spyOn(permissionsService, 'currentUserCan').mockReturnValue(true)
+
+    fixture.detectChanges()
+    component.activeNavID.set(2)
+    await fixture.whenStable()
+
+    expect(component.activeCustomFields).toBeNull()
+    expect(
+      fixture.nativeElement.querySelector(
+        'pngx-page-header .btn-outline-primary'
+      )?.textContent
+    ).toContain('Add Field')
   })
 })

src-ui/src/app/components/manage/document-attributes/document-attributes.component.ts

@@ -163,12 +163,17 @@ export class DocumentAttributesComponent implements OnInit, OnDestroy {
   }
 
   get activeCustomFields(): CustomFieldsComponent | null {
-    if (this.activeSection?.kind !== DocumentAttributesSectionKind.CustomFields)
-      return null
+    if (!this.customFieldsActive) return null
     const instance = this.activeOutlet?.componentInstance
     return instance instanceof CustomFieldsComponent ? instance : null
   }
 
+  get customFieldsActive(): boolean {
+    return (
+      this.activeSection?.kind === DocumentAttributesSectionKind.CustomFields
+    )
+  }
+
   get activeTabLabel(): string {
     return this.activeSection?.label ?? ''
   }
@@ -224,6 +229,10 @@ export class DocumentAttributesComponent implements OnInit, OnDestroy {
     this.router.navigate(['attributes', nextSection])
   }
 
+  addCustomField(): void {
+    this.activeCustomFields?.editField(null)
+  }
+
   private getDefaultNavID(): DocumentAttributesNavIDs | null {
     return this.visibleSections[0]?.id ?? null
   }

src/documents/consumer.py

@@ -57,9 +57,7 @@ from paperless.models import ArchiveFileGenerationChoices
 from paperless.parsers import ParserContext
 from paperless.parsers import ParserProtocol
 from paperless.parsers.registry import get_parser_registry
-from paperless.parsers.utils import PDF_TEXT_MIN_LENGTH
-from paperless.parsers.utils import extract_pdf_text
-from paperless.parsers.utils import is_tagged_pdf
+from paperless.parsers.utils import pdf_born_digital_text
 
 LOGGING_NAME: Final[str] = "paperless.consumer"
 
@@ -138,53 +136,45 @@ def should_produce_archive(
 
     # Must produce a PDF so the frontend can display the original format at all.
     if parser.requires_pdf_rendition:
-        _log.debug("Archive: yes — parser requires PDF rendition for frontend display")
+        _log.debug("Archive: yes - parser requires PDF rendition for frontend display")
         return True
 
     # Parser cannot produce an archive (e.g. TextDocumentParser).
     if not parser.can_produce_archive:
-        _log.debug("Archive: no — parser cannot produce archives")
+        _log.debug("Archive: no - parser cannot produce archives")
         return False
 
     generation = OcrConfig().archive_file_generation
 
     if generation == ArchiveFileGenerationChoices.ALWAYS:
-        _log.debug("Archive: yes — ARCHIVE_FILE_GENERATION=always")
+        _log.debug("Archive: yes - ARCHIVE_FILE_GENERATION=always")
         return True
     if generation == ArchiveFileGenerationChoices.NEVER:
-        _log.debug("Archive: no — ARCHIVE_FILE_GENERATION=never")
+        _log.debug("Archive: no - ARCHIVE_FILE_GENERATION=never")
         return False
 
     # auto: produce archives for scanned/image documents; skip for born-digital PDFs.
     if mime_type.startswith("image/"):
-        _log.debug("Archive: yes — image document, ARCHIVE_FILE_GENERATION=auto")
+        _log.debug("Archive: yes - image document, ARCHIVE_FILE_GENERATION=auto")
         return True
     if mime_type == "application/pdf":
-        text = extract_pdf_text(document_path)
-        has_text = text is not None and len(text) > 0
-        if has_text and is_tagged_pdf(document_path):
+        text, born_digital = pdf_born_digital_text(document_path, log=_log)
+        text_length = len(text) if text else 0
+        if born_digital:
             _log.debug(
-                "Archive: no — born-digital PDF (structure tags detected),"
+                "Archive: no - born-digital PDF (text_length=%d),"
                 " ARCHIVE_FILE_GENERATION=auto",
+                text_length,
             )
             return False
-        if text is None or len(text) <= PDF_TEXT_MIN_LENGTH:
-            _log.debug(
-                "Archive: yes — scanned PDF (text_length=%d ≤ %d),"
-                " ARCHIVE_FILE_GENERATION=auto",
-                len(text) if text else 0,
-                PDF_TEXT_MIN_LENGTH,
-            )
-            return True
         _log.debug(
-            "Archive: no — born-digital PDF (text_length=%d > %d),"
+            "Archive: yes - scanned/textless PDF (text_length=%d),"
             " ARCHIVE_FILE_GENERATION=auto",
-            len(text),
-            PDF_TEXT_MIN_LENGTH,
+            text_length,
         )
-        return False
+        return True
     _log.debug(
-        "Archive: no — MIME type %r not eligible for auto archive generation",
+        "Archive: no - MIME type %r not eligible for auto archive generation",
         mime_type,
     )
     return False

src/documents/management/commands/document_llmindex.py

@@ -3,7 +3,6 @@ from typing import Any
 from documents.management.commands.base import PaperlessCommand
 from documents.tasks import llmindex_index
 from paperless_ai.indexing import llm_index_compact
-from paperless_ai.indexing import llm_index_migrate
 
 
 class Command(PaperlessCommand):
@@ -14,18 +13,12 @@ class Command(PaperlessCommand):
 
     def add_arguments(self, parser: Any) -> None:
         super().add_arguments(parser)
-        parser.add_argument(
-            "command",
-            choices=["rebuild", "update", "compact", "migrate"],
-        )
+        parser.add_argument("command", choices=["rebuild", "update", "compact"])
 
     def handle(self, *args: Any, **options: Any) -> None:
         if options["command"] == "compact":
             llm_index_compact()
             return
-        if options["command"] == "migrate":
-            llm_index_migrate()
-            return
         llmindex_index(
             rebuild=options["command"] == "rebuild",
             iter_wrapper=lambda docs: self.track(

src/documents/management/commands/document_sanity_checker.py

@@ -61,7 +61,7 @@ class Command(PaperlessCommand):
         )
         table.add_column("Level", width=7, no_wrap=True)
         table.add_column("Document", min_width=20)
-        table.add_column("Issue", ratio=1)
+        table.add_column("Issue", ratio=1, overflow="fold")
 
         for doc_pk, doc_messages in messages.iter_messages():
             if doc_pk is not None:

src/documents/tests/management/test_management_document_llmindex.py

@@ -9,7 +9,6 @@ if TYPE_CHECKING:
 
 _COMPACT = "documents.management.commands.document_llmindex.llm_index_compact"
 _INDEX = "documents.management.commands.document_llmindex.llmindex_index"
-_MIGRATE = "documents.management.commands.document_llmindex.llm_index_migrate"
 
 
 class TestDocumentLlmindexCommand:
@@ -18,11 +17,6 @@ class TestDocumentLlmindexCommand:
         call_command("document_llmindex", "compact")
         mock_compact.assert_called_once_with()
 
-    def test_migrate_calls_llm_index_migrate(self, mocker: MockerFixture) -> None:
-        mock_migrate = mocker.patch(_MIGRATE)
-        call_command("document_llmindex", "migrate")
-        mock_migrate.assert_called_once_with()
-
     def test_rebuild_calls_llmindex_index_with_rebuild_true(
         self,
         mocker: MockerFixture,

src/documents/tests/test_consumer.py

@@ -1329,7 +1329,7 @@ class PreConsumeTestCase(DirectoriesMixin, GetConsumerMixin, TestCase):
         with self.get_consumer(self.test_file) as c:
             c.run()
             # Verify no pre-consume script subprocess was invoked
-            # (run_subprocess may still be called by _extract_text_for_archive_check)
+            # (run_subprocess may still be called by pdf_born_digital_text via pdftotext)
             script_calls = [
                 call
                 for call in m.call_args_list
@@ -1354,7 +1354,7 @@ class PreConsumeTestCase(DirectoriesMixin, GetConsumerMixin, TestCase):
                     self.assertTrue(m.called)
 
                     # Find the call that invoked the pre-consume script
-                    # (run_subprocess may also be called by _extract_text_for_archive_check)
+                    # (run_subprocess may also be called by pdf_born_digital_text via pdftotext)
                     script_call = next(
                         call
                         for call in m.call_args_list

src/documents/tests/test_consumer_archive.py

@@ -134,60 +134,32 @@ class TestShouldProduceArchive:
         assert should_produce_archive(parser, mime, Path("/tmp/doc")) is expected
 
     @pytest.mark.parametrize(
-        ("extracted_text", "expected"),
+        ("born_digital", "expected"),
         [
-            pytest.param(
-                "This is a born-digital PDF with lots of text content. " * 10,
-                False,
-                id="born-digital-long-text-skips-archive",
-            ),
-            pytest.param(None, True, id="no-text-scanned-produces-archive"),
-            pytest.param("tiny", True, id="short-text-treated-as-scanned"),
+            pytest.param(True, False, id="born-digital-skips-archive"),
+            pytest.param(False, True, id="not-born-digital-produces-archive"),
         ],
     )
     def test_auto_pdf_archive_decision(
         self,
         mocker: MockerFixture,
         settings,
-        extracted_text: str | None,
+        born_digital: bool,  # noqa: FBT001
         expected: bool,  # noqa: FBT001
     ) -> None:
+        """Archive decision tracks pdf_born_digital_text()'s verdict exactly.
+
+        should_produce_archive() defers entirely to pdf_born_digital_text()
+        for the has-real-text decision, so both callers of that predicate
+        (this function and RasterisedDocumentParser.parse()) always agree.
+        """
         settings.ARCHIVE_FILE_GENERATION = "auto"
-        mocker.patch("documents.consumer.is_tagged_pdf", return_value=False)
-        mocker.patch("documents.consumer.extract_pdf_text", return_value=extracted_text)
+        mocker.patch(
+            "documents.consumer.pdf_born_digital_text",
+            return_value=("some text", born_digital),
+        )
         parser = _parser_instance(can_produce=True, requires_rendition=False)
         assert (
             should_produce_archive(parser, "application/pdf", Path("/tmp/doc.pdf"))
             is expected
         )
-
-    def test_tagged_pdf_skips_archive_in_auto_mode(
-        self,
-        mocker: MockerFixture,
-        settings,
-    ) -> None:
-        """Tagged PDFs (e.g. Word exports) with real text are treated as born-digital, even below PDF_TEXT_MIN_LENGTH."""
-        settings.ARCHIVE_FILE_GENERATION = "auto"
-        mocker.patch("documents.consumer.is_tagged_pdf", return_value=True)
-        mocker.patch("documents.consumer.extract_pdf_text", return_value="tiny")
-        parser = _parser_instance(can_produce=True, requires_rendition=False)
-        assert (
-            should_produce_archive(parser, "application/pdf", Path("/tmp/doc.pdf"))
-            is False
-        )
-
-    def test_tagged_pdf_without_text_produces_archive(
-        self,
-        mocker: MockerFixture,
-        settings,
-    ) -> None:
-        """A tagged PDF with no actual extractable text (e.g. some scanner firmware) is not
-        trusted as born-digital — the tag alone must not bypass OCR."""
-        settings.ARCHIVE_FILE_GENERATION = "auto"
-        mocker.patch("documents.consumer.is_tagged_pdf", return_value=True)
-        mocker.patch("documents.consumer.extract_pdf_text", return_value=None)
-        parser = _parser_instance(can_produce=True, requires_rendition=False)
-        assert (
-            should_produce_archive(parser, "application/pdf", Path("/tmp/doc.pdf"))
-            is True
-        )

src/paperless/parsers/tesseract.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import importlib.resources
 import logging
 import os
-import re
 import shutil
 import tempfile
 from pathlib import Path
@@ -25,9 +24,9 @@ from paperless.config import OcrConfig
 from paperless.models import CleanChoices
 from paperless.models import ModeChoices
 from paperless.models import OutputTypeChoices
-from paperless.parsers.utils import PDF_TEXT_MIN_LENGTH
 from paperless.parsers.utils import extract_pdf_text
-from paperless.parsers.utils import is_tagged_pdf
+from paperless.parsers.utils import is_born_digital_text
+from paperless.parsers.utils import post_process_text
 from paperless.parsers.utils import read_file_handle_unicode_errors
 from paperless.version import __full_version_str__
 
@@ -510,10 +509,10 @@ class RasterisedDocumentParser:
 
         if mime_type == "application/pdf":
             text_original = self.extract_text(None, document_path)
-            has_text = text_original is not None and len(text_original) > 0
-            original_has_text = has_text and (
-                is_tagged_pdf(document_path, log=self.log)
-                or len(text_original) > PDF_TEXT_MIN_LENGTH
+            original_has_text = is_born_digital_text(
+                text_original,
+                document_path,
+                log=self.log,
             )
         else:
             text_original = None
@@ -658,17 +657,3 @@ class RasterisedDocumentParser:
                     f"No text was found in {document_path}, the content will be empty.",
                 )
                 self.text = ""
-
-
-def post_process_text(text: str | None) -> str | None:
-    if not text:
-        return None
-
-    collapsed_spaces = re.sub(r"([^\S\r\n]+)", " ", text)
-    no_leading_whitespace = re.sub(r"([\n\r]+)([^\S\n\r]+)", "\\1", collapsed_spaces)
-    no_trailing_whitespace = re.sub(r"([^\S\n\r]+)$", "", no_leading_whitespace)
-
-    # TODO: this needs a rework
-    # replace \0 prevents issues with saving to postgres.
-    # text may contain \0 when this character is present in PDF files.
-    return no_trailing_whitespace.strip().replace("\0", " ")

src/paperless/parsers/utils.py

@@ -111,6 +111,88 @@ def extract_pdf_text(
         return None
 
 
+def post_process_text(text: str | None) -> str | None:
+    """Normalize extracted PDF/OCR text: collapse whitespace, strip padding.
+
+    Returns ``None`` for ``None`` or whitespace-only input, so callers can
+    treat "no text" and "only layout padding" the same way.
+    """
+    if not text:
+        return None
+
+    collapsed_spaces = re.sub(r"([^\S\r\n]+)", " ", text)
+    no_leading_whitespace = re.sub(r"([\n\r]+)([^\S\n\r]+)", "\\1", collapsed_spaces)
+    no_trailing_whitespace = re.sub(r"([^\S\n\r]+)$", "", no_leading_whitespace)
+
+    # replace \0 prevents issues with saving to postgres.
+    # text may contain \0 when this character is present in PDF files.
+    result = no_trailing_whitespace.strip().replace("\0", " ")
+    return result or None
+
+
+def is_born_digital_text(
+    text: str | None,
+    path: Path,
+    log: logging.Logger | None = None,
+) -> bool:
+    """Decide whether already-extracted, normalized PDF text counts as born-digital.
+
+    This is the single source of truth for "does this PDF already have real
+    text", used both to decide whether to produce an archive file and to
+    decide whether OCR can be skipped. Both decisions must agree, or a
+    tagged-but-textless PDF can end up with no archive AND a forced OCR pass
+    (see GH #13387): raw ``pdftotext -layout`` output can be non-empty
+    (whitespace/form-feed padding) even when there is no real content, so
+    *text* must already be normalized via :func:`post_process_text`, not the
+    raw extraction.
+
+    Parameters
+    ----------
+    text:
+        The normalized extracted text (or ``None``) to evaluate.
+    path:
+        Absolute path to the PDF file, used for the tagged-PDF check.
+    log:
+        Logger for warnings.  Falls back to the module-level logger when omitted.
+
+    Returns
+    -------
+    bool
+        Whether the PDF counts as born-digital (has real text, and is either
+        tagged or exceeds ``PDF_TEXT_MIN_LENGTH``).
+    """
+    if not text:
+        return False
+    return is_tagged_pdf(path, log=log) or len(text) > PDF_TEXT_MIN_LENGTH
+
+
+def pdf_born_digital_text(
+    path: Path,
+    log: logging.Logger | None = None,
+) -> tuple[str | None, bool]:
+    """Extract a PDF's text and decide whether it should be treated as born-digital.
+
+    Convenience wrapper around :func:`is_born_digital_text` for callers that
+    don't already have the PDF's text extracted (e.g. the archive-generation
+    decision, which runs before any parser has touched the file).
+
+    Parameters
+    ----------
+    path:
+        Absolute path to the PDF file.
+    log:
+        Logger for warnings.  Falls back to the module-level logger when omitted.
+
+    Returns
+    -------
+    tuple[str | None, bool]
+        The normalized extracted text (or ``None``), and whether the PDF
+        counts as born-digital.
+    """
+    text = post_process_text(extract_pdf_text(path, log=log))
+    return text, is_born_digital_text(text, path, log=log)
+
+
 def read_file_handle_unicode_errors(
     filepath: Path,
     log: logging.Logger | None = None,

src/paperless/tests/conftest.py

@@ -36,6 +36,23 @@ def samples_dir() -> Path:
     return (Path(__file__).parent / "samples").resolve()
 
 
+@pytest.fixture(scope="session")
+def tagged_no_text_pdf_file(samples_dir: Path) -> Path:
+    """Path to a tagged PDF whose only "text" is pdftotext layout padding.
+
+    Reproduces GH #13387: ``/MarkInfo /Marked true`` is set, but the only
+    extractable content is a form-feed byte, not real text. Lives here
+    rather than in parsers/conftest.py so both parser tests and
+    paperless/tests/test_parser_utils.py can use it.
+
+    Returns
+    -------
+    Path
+        Absolute path to ``tesseract/tagged-but-no-text.pdf``.
+    """
+    return samples_dir / "tesseract" / "tagged-but-no-text.pdf"
+
+
 @pytest.fixture(autouse=True)
 def clean_registry() -> Generator[None, None, None]:
     """Reset the parser registry before and after every test.

src/paperless/tests/parsers/test_tesseract_parser.py

@@ -21,7 +21,7 @@ from documents.parsers import run_convert
 from paperless.models import ModeChoices
 from paperless.parsers import ParserProtocol
 from paperless.parsers.tesseract import RasterisedDocumentParser
-from paperless.parsers.tesseract import post_process_text
+from paperless.parsers.utils import is_tagged_pdf
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -151,36 +151,6 @@ class TestRasterisedDocumentParserLifecycle:
         assert tempdir is not None and not tempdir.exists()
 
 
-# ---------------------------------------------------------------------------
-# post_process_text
-# ---------------------------------------------------------------------------
-
-
-class TestPostProcessText:
-    @pytest.mark.parametrize(
-        ("source", "expected"),
-        [
-            pytest.param(
-                "simple     string",
-                "simple string",
-                id="collapse-spaces",
-            ),
-            pytest.param(
-                "simple    newline\n   testing string",
-                "simple newline\ntesting string",
-                id="preserve-newline",
-            ),
-            pytest.param(
-                "utf-8   строка с пробелами в конце  ",  # noqa: RUF001
-                "utf-8 строка с пробелами в конце",  # noqa: RUF001
-                id="utf8-trailing-spaces",
-            ),
-        ],
-    )
-    def test_post_process_text(self, source: str, expected: str) -> None:
-        assert post_process_text(source) == expected
-
-
 # ---------------------------------------------------------------------------
 # Page count
 # ---------------------------------------------------------------------------
@@ -910,25 +880,25 @@ class TestSkipArchive:
         self,
         mocker: MockerFixture,
         tesseract_parser: RasterisedDocumentParser,
-        tesseract_samples_dir: Path,
+        tagged_no_text_pdf_file: Path,
     ) -> None:
         """
         GIVEN:
-            - A PDF that reports itself as tagged (/MarkInfo /Marked true) but
-              has no actual extractable text (some scanner firmware produces
-              this — see GitHub issue #13349)
+            - A real PDF that reports itself as tagged (/MarkInfo /Marked
+              true) but whose only pdftotext output is layout padding (a
+              lone form-feed byte), not real text (see GitHub issue #13387,
+              originally reported against #13349's tagged-PDF handling)
             - Mode: auto, produce_archive=False
         WHEN:
             - Document is parsed
         THEN:
             - The tag alone is not trusted as "has text"; OCRmyPDF still runs
         """
+        assert is_tagged_pdf(tagged_no_text_pdf_file) is True
         tesseract_parser.settings.mode = ModeChoices.AUTO
-        mocker.patch("paperless.parsers.tesseract.is_tagged_pdf", return_value=True)
-        mocker.patch.object(tesseract_parser, "extract_text", return_value=None)
         mock_ocr = mocker.patch("ocrmypdf.ocr")
         tesseract_parser.parse(
-            tesseract_samples_dir / "multi-page-images.pdf",
+            tagged_no_text_pdf_file,
             "application/pdf",
             produce_archive=False,
         )

src/paperless/tests/test_parser_utils.py

@@ -4,10 +4,18 @@ from __future__ import annotations
 
 import codecs
 from pathlib import Path
+from typing import TYPE_CHECKING
+
+import pytest
 
 from paperless.parsers.utils import is_tagged_pdf
+from paperless.parsers.utils import pdf_born_digital_text
+from paperless.parsers.utils import post_process_text
 from paperless.parsers.utils import read_file_handle_unicode_errors
 
+if TYPE_CHECKING:
+    from pytest_mock import MockerFixture
+
 SAMPLES = Path(__file__).parent / "samples" / "tesseract"
 
 
@@ -60,3 +68,105 @@ class TestIsTaggedPdf:
         bad = tmp_path / "bad.pdf"
         bad.write_bytes(b"not a pdf")
         assert is_tagged_pdf(bad) is False
+
+
+class TestPostProcessText:
+    @pytest.mark.parametrize(
+        ("source", "expected"),
+        [
+            pytest.param(
+                "simple     string",
+                "simple string",
+                id="collapse-spaces",
+            ),
+            pytest.param(
+                "simple    newline\n   testing string",
+                "simple newline\ntesting string",
+                id="preserve-newline",
+            ),
+            pytest.param(
+                "utf-8   строка с пробелами в конце  ",  # noqa: RUF001
+                "utf-8 строка с пробелами в конце",  # noqa: RUF001
+                id="utf8-trailing-spaces",
+            ),
+            pytest.param(None, None, id="none-input"),
+            pytest.param("", None, id="empty-string"),
+            pytest.param("   \n\x0c  \n ", None, id="whitespace-and-formfeed-only"),
+        ],
+    )
+    def test_post_process_text(
+        self,
+        source: str | None,
+        expected: str | None,
+    ) -> None:
+        assert post_process_text(source) == expected
+
+
+class TestPdfBornDigitalText:
+    """Regression coverage for GH #13387.
+
+    should_produce_archive() and RasterisedDocumentParser.parse() must agree
+    on whether a PDF has real text, so both go through this one function.
+    """
+
+    @pytest.mark.parametrize(
+        ("extracted", "tagged", "expected_text", "expected_born_digital"),
+        [
+            pytest.param("tiny", True, "tiny", True, id="tagged-with-real-text"),
+            pytest.param("tiny", False, "tiny", False, id="untagged-below-min-length"),
+            pytest.param(
+                "x" * 51,
+                False,
+                "x" * 51,
+                True,
+                id="untagged-above-min-length",
+            ),
+            pytest.param(None, True, None, False, id="tagged-but-no-text"),
+        ],
+    )
+    def test_born_digital_decision(
+        self,
+        mocker: MockerFixture,
+        tmp_path: Path,
+        extracted: str | None,
+        tagged: bool,  # noqa: FBT001
+        expected_text: str | None,
+        expected_born_digital: bool,  # noqa: FBT001
+    ) -> None:
+        """
+        GIVEN:
+            - A PDF whose pdftotext output and /MarkInfo tag status vary
+        WHEN:
+            - pdf_born_digital_text() is called
+        THEN:
+            - The normalized text and born-digital verdict match; the tag
+              alone never counts as "has text"
+        """
+        mocker.patch(
+            "paperless.parsers.utils.extract_pdf_text",
+            return_value=extracted,
+        )
+        mocker.patch("paperless.parsers.utils.is_tagged_pdf", return_value=tagged)
+        text, born_digital = pdf_born_digital_text(tmp_path / "doc.pdf")
+        assert text == expected_text
+        assert born_digital is expected_born_digital
+
+    def test_tagged_but_textless_pdf_is_not_born_digital(
+        self,
+        tagged_no_text_pdf_file: Path,
+    ) -> None:
+        """
+        GIVEN:
+            - A real PDF that is tagged (/MarkInfo /Marked true) but whose
+              only "text" is layout padding (a stray form-feed byte)
+        WHEN:
+            - pdf_born_digital_text() is called with no mocking
+        THEN:
+            - The normalized text is None and the PDF is not treated as
+              born-digital. The raw, unnormalized pdftotext output is
+              non-empty for this file, which is exactly what caused the
+              archive decision to disagree with the OCR decision in #13387.
+        """
+        text, born_digital = pdf_born_digital_text(tagged_no_text_pdf_file)
+        assert text is None
+        assert born_digital is False

src/paperless_ai/indexing.py

@@ -1,5 +1,6 @@
 import logging
 from collections.abc import Iterable
+from collections.abc import Iterator
 from contextlib import contextmanager
 from datetime import timedelta
 from typing import TYPE_CHECKING
@@ -22,6 +23,7 @@ from paperless_ai.embedding import get_configured_model_name
 from paperless_ai.embedding import get_embedding_model
 
 if TYPE_CHECKING:
+    from django.db.models import QuerySet
     from llama_index.core.schema import BaseNode
 
     from paperless_ai.vector_store import PaperlessSqliteVecVectorStore
@@ -32,6 +34,35 @@ logger = logging.getLogger("paperless_ai.indexing")
 RAG_NUM_OUTPUT = 512
 RAG_CHUNK_OVERLAP = 200
 
+# update_llm_index(): row count per .iterator() batch when streaming
+# documents for a rebuild/update, matching _DocumentViewerStream's chunk
+# size in documents/search/_backend.py.
+_INDEX_STREAM_CHUNK_SIZE = 1000
+
+
+class _StreamedDocuments:
+    """A thin QuerySet wrapper that streams via ``.iterator()`` instead of
+    materializing every row (plus its ``content`` and prefetch caches) into
+    memory at once, while still supporting ``len()`` so ``iter_wrapper``'s
+    progress bar shows a real total instead of falling back to indeterminate.
+    Same shape as ``documents/search/_backend.py``'s ``_DocumentViewerStream``,
+    just without that class's extra per-batch permission lookup -- nothing
+    here needs one.
+    """
+
+    def __init__(self, documents: "QuerySet[Document]") -> None:
+        self._documents = documents
+
+    def __len__(self) -> int:
+        return self._documents.count()
+
+    def __iter__(self) -> Iterator[Document]:
+        # iterator(chunk_size=...) streams from a server-side cursor instead
+        # of materializing the whole queryset in memory; since Django 4.1 it
+        # still honours prefetch_related, running the prefetches one batch
+        # at a time.
+        return iter(self._documents.iterator(chunk_size=_INDEX_STREAM_CHUNK_SIZE))
+
 
 def queue_llm_index_update_if_needed(*, rebuild: bool, reason: str) -> bool:
     # NOTE: The check-then-enqueue sequence below is non-atomic (TOCTOU): two
@@ -144,24 +175,6 @@ def _exclude_readers():
         lock.close()
 
 
-def _with_exclusive_access(operation: str, fn):
-    """Run ``fn()`` with exclusive index access (see ``_exclude_readers()``),
-    for compaction/migration file swaps that must not run while readers are
-    active. Returns ``fn()``'s result, or None (after logging) if active
-    readers do not drain within ``LLM_INDEX_COMPACTION_LOCK_TIMEOUT`` --
-    callers skip the operation this run; it retries next time.
-    """
-    try:
-        with _exclude_readers():
-            return fn()
-    except Timeout:
-        logger.info(
-            "Skipping LLM index %s: index readers are active; will retry next run.",
-            operation,
-        )
-        return None
-
-
 @contextmanager
 def write_store(embed_model_name: str | None = None):
     """Acquire the write lock and yield the vector store.
@@ -186,21 +199,6 @@ def write_store(embed_model_name: str | None = None):
         yield store
 
 
-def _check_and_run_migrations(store: "PaperlessSqliteVecVectorStore") -> bool:
-    """Run any pending structural migrations, returning True if a pending
-    re-embed migration needs the caller to force a rebuild -- never
-    triggered automatically here. Safe to call before any write, including
-    delete()/upsert_document(): has_pending_migration() (see its docstring)
-    keeps this a no-op, with no exclusive access taken, once the store is
-    current.
-    """
-    if not store.has_pending_migration():
-        return False
-    return bool(
-        _with_exclusive_access("migration check", store.check_and_run_migrations),
-    )
-
-
 def _safe_related_name(document: Document, field: str) -> str | None:
     """
     Returns the ``name`` of a related object (correspondent, document_type,
@@ -372,7 +370,15 @@ def update_llm_index(
     happens, since a rebuild always covers the whole library regardless.
     """
     with write_store() as store:
-        needs_reembed = _check_and_run_migrations(store)
+        try:
+            with _exclude_readers():
+                needs_reembed = store.check_and_run_migrations()
+        except Timeout:
+            logger.info(
+                "Skipping LLM index migration check: index readers are active; "
+                "will retry next run.",
+            )
+            needs_reembed = False
         if needs_reembed:
             logger.warning(
                 "LLM index migration requires re-embedding; forcing rebuild.",
@@ -393,11 +399,12 @@ def update_llm_index(
     config = AIConfig()
     model_name = get_configured_model_name(config)
 
-    if not rebuild:
+    if not rebuild and llm_index_exists():
         with read_store() as store:
-            if store.table_exists() and store.config_mismatch(model_name):
-                logger.warning("Embedding model changed; forcing LLM index rebuild.")
-                rebuild = True
+            config_mismatch = store.config_mismatch(model_name)
+        if config_mismatch:
+            logger.warning("Embedding model changed; forcing LLM index rebuild.")
+            rebuild = True
 
     if no_documents:
         logger.warning("No documents found to index.")
@@ -409,7 +416,7 @@ def update_llm_index(
         if rebuild or not store.table_exists():
             logger.info("Rebuilding LLM index.")
             store.drop_table()
-            for document in iter_wrapper(documents):
+            for document in iter_wrapper(_StreamedDocuments(documents)):
                 nodes = build_document_node(document, chunk_size=chunk_size)
                 _embed_nodes(nodes, embed_model)
                 store.add(nodes)
@@ -422,7 +429,7 @@ def update_llm_index(
             )
             existing = store.get_modified_times()
             changed = 0
-            for document in iter_wrapper(scoped_documents):
+            for document in iter_wrapper(_StreamedDocuments(scoped_documents)):
                 doc_id = str(document.id)
                 if existing.get(doc_id) == document.modified.isoformat():
                     continue
@@ -436,7 +443,14 @@ def update_llm_index(
                 else "No changes detected in LLM index."
             )
 
-        _with_exclusive_access("compaction", store.compact)
+        try:
+            with _exclude_readers():
+                store.compact()
+        except Timeout:
+            logger.info(
+                "Skipping LLM index compaction: index readers are active; "
+                "will retry next run.",
+            )
     return msg
 
 
@@ -451,60 +465,25 @@ def llm_index_add_or_update_document(document: Document):
         _embed_nodes(new_nodes, get_embedding_model(config))
 
     with write_store(embed_model_name=get_configured_model_name(config)) as store:
-        needs_reembed = _check_and_run_migrations(store)
-        if needs_reembed:
-            logger.warning(
-                "Skipping incremental LLM index update for document %s: the "
-                "index requires re-embedding first. Run 'document_llmindex "
-                "rebuild' to resolve.",
-                document.id,
-            )
-            return
         store.upsert_document(str(document.id), new_nodes)
 
 
-def llm_index_migrate() -> None:
-    """Apply any pending LLM index schema migrations, with no reindex.
-
-    Intended to run unconditionally on every startup (see the
-    init-llmindex-migrate container step and the bare-metal upgrade docs):
-    has_pending_migration() short-circuits to a metadata-only read once the
-    store is current, so a healthy install pays almost nothing here. Only
-    ever applies structural migrations -- a pending re-embed migration is
-    left for the explicit, deliberate rebuild path (``document_llmindex
-    update``/``rebuild``) to resolve, since re-embedding can be slow and,
-    for a metered embedding backend, cost money.
-    """
-    if not AIConfig().llm_index_enabled:
-        return
-    with write_store() as store:
-        needs_reembed = _check_and_run_migrations(store)
-    if needs_reembed:
-        logger.warning(
-            "LLM index requires re-embedding, which this automatic migration "
-            "check will not do on its own -- it can be slow and, for a "
-            "metered embedding backend, cost money. Run "
-            "'document_llmindex rebuild' manually when ready.",
-        )
-
-
 def llm_index_compact() -> None:
     """Compact the index immediately, rebuilding the table to reclaim space."""
     with write_store() as store:
-        _with_exclusive_access("compaction", lambda: store.compact(force=True))
+        try:
+            with _exclude_readers():
+                store.compact(force=True)
+        except Timeout:
+            logger.info(
+                "Skipping LLM index compaction: index readers are active; "
+                "will retry next run.",
+            )
 
 
 def llm_index_remove_document(document: Document):
     """Remove a document's chunks from the LLM index."""
     with write_store() as store:
-        if _check_and_run_migrations(store):
-            logger.warning(
-                "Skipping removal of document %s from the LLM index: the "
-                "index requires re-embedding first. Run 'document_llmindex "
-                "rebuild' to resolve.",
-                document.id,
-            )
-            return
         store.delete(str(document.id))
 
 

src/paperless_ai/migrations/__init__.py

@@ -1,60 +0,0 @@
-"""Schema migrations for the sqlite-vec vector store.
-
-Each migration lives in its own module here, named ``mNNNN_description.py``
-(e.g. ``m0001_v1_to_v2.py`` -- a leading digit isn't a valid Python
-identifier, hence the ``m`` prefix, unlike Django's own numbered migrations,
-which load via a dynamic ``importlib.import_module()`` call rather than a
-static import statement), and registers itself into ``MIGRATIONS`` at import
-time. ``vector_store.py`` imports those modules at the bottom of the file,
-purely for that registration side effect, after ``PaperlessSqliteVecVectorStore``
-is fully defined -- migrations need it to implement ``apply()`` (see
-``Migration`` below).
-
-To add a new migration: add a new ``mNNNN_description.py`` module here that
-imports ``PaperlessSqliteVecVectorStore`` from ``paperless_ai.vector_store``,
-defines its ``apply()``, and appends a ``Migration`` to ``MIGRATIONS``; then
-import that module at the bottom of ``vector_store.py`` and bump
-``SCHEMA_VERSION`` there. A migration must freeze its own historical DDL for
-any side table its target version depends on (``DROP TABLE IF EXISTS`` +
-its own literal ``CREATE TABLE``/``CREATE INDEX`` statements) rather than
-delegating to any "current schema" helper -- see ``m0001_v1_to_v2.py`` for
-why and the worked example.
-"""
-
-import sqlite3
-from collections.abc import Callable
-from dataclasses import dataclass
-from dataclasses import field
-from typing import Literal
-
-
-@dataclass
-class Migration:
-    """A schema migration for the sqlite-vec vector store.
-
-    kind="structural": rows are copied into a new-schema file with no
-    re-embedding needed.  Supply ``apply(src_conn, dst_conn, dim)``, which
-    must create every table its target schema needs in ``dst_conn`` and copy
-    ``src_conn``'s rows and relevant ``index_meta`` keys into it.
-    ``schema_version`` is written by the migration runner after ``apply``
-    returns, not by ``apply`` itself.
-
-    kind="re-embed": the new schema requires fresh embeddings.
-    ``check_and_run_migrations()`` returns True when it encounters one of
-    these so the caller can force a full rebuild (which recreates the table
-    at the current SCHEMA_VERSION).
-    """
-
-    from_version: int
-    to_version: int
-    kind: Literal["structural", "re-embed"]
-    description: str
-    apply: Callable[[sqlite3.Connection, sqlite3.Connection, int], None] | None = field(
-        default=None,
-        repr=False,
-    )
-
-
-# Registry of all schema migrations in order, populated by each migration
-# module's import-time registration (see the module docstring above).
-MIGRATIONS: list[Migration] = []

src/paperless_ai/migrations/m0001_v1_to_v2.py

@@ -1,118 +0,0 @@
-import sqlite3
-
-from paperless_ai.migrations import MIGRATIONS
-from paperless_ai.migrations import Migration
-from paperless_ai.tables import ChunkRow
-from paperless_ai.tables import DocumentChunksTable
-from paperless_ai.tables import DocumentMetaRow
-from paperless_ai.tables import DocumentMetaTable
-from paperless_ai.tables import IndexMetaTable
-from paperless_ai.vector_store import COMPACT_BATCH_SIZE
-from paperless_ai.vector_store import DEFAULT_TABLE_NAME
-
-# v1's vec0 shape has never changed since it first shipped and is the ONLY
-# real upgrade path -- no store has ever existed at any intermediate
-# version, so this migration goes straight from that shipped shape to the
-# final v2 target in one pass.
-_V1_SELECT = (
-    "SELECT id, document_id, modified, node_content, embedding FROM "
-    + DEFAULT_TABLE_NAME
-)
-
-
-def _migrate_v1_to_v2(
-    src_conn: sqlite3.Connection,
-    dst_conn: sqlite3.Connection,
-    dim: int,
-) -> None:
-    """v1 -> v2: document_id TEXT -> INTEGER, modified moves out of vec0
-    into document_meta, document_chunks added for O(1) per-document delete.
-
-    Freezes its own v2-shaped vec0/document_chunks/document_meta DDL inline,
-    rather than delegating to the gateway "create table" helpers or the
-    store's own vec0-table builder (all of which always reflect the
-    *current* schema): a later schema version changing any of these tables'
-    shape must not silently change what this migration produces for someone
-    upgrading straight from v1.
-    _open_connection() already created document_chunks/document_meta on
-    dst_conn (reflecting current HEAD) as a side effect of opening it for
-    this migration's rebuild -- DROP them first so this migration's own
-    frozen CREATE TABLE isn't a silent no-op against that. Safe here because
-    dst_conn is a freshly opened, empty rebuild file with nothing written
-    yet.
-    """
-    dst_conn.execute(  # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query
-        "CREATE VIRTUAL TABLE "
-        + DEFAULT_TABLE_NAME
-        + " USING vec0("
-        + "id TEXT PRIMARY KEY,"
-        + " document_id INTEGER,"
-        + " +node_content TEXT,"
-        + " embedding float["
-        + str(int(dim))
-        + "] distance_metric=cosine"
-        + ")",
-    )
-    dst_conn.execute("DROP TABLE IF EXISTS document_chunks")
-    dst_conn.execute(
-        "CREATE TABLE document_chunks "
-        "(chunk_id TEXT PRIMARY KEY, document_id INTEGER NOT NULL)",
-    )
-    dst_conn.execute(
-        "CREATE INDEX idx_document_chunks_document_id ON document_chunks (document_id)",
-    )
-    dst_conn.execute("DROP TABLE IF EXISTS document_meta")
-    dst_conn.execute(
-        "CREATE TABLE document_meta "
-        "(document_id INTEGER PRIMARY KEY, modified TEXT NOT NULL)",
-    )
-
-    IndexMetaTable.set_dim(dst_conn, dim)
-    embed_model = IndexMetaTable.get_embed_model(src_conn)
-    if embed_model is not None:
-        IndexMetaTable.set_embed_model(dst_conn, embed_model)
-
-    dst_conn.execute("BEGIN IMMEDIATE")
-    src_cursor = src_conn.execute(_V1_SELECT)
-    live = 0
-    while batch := src_cursor.fetchmany(COMPACT_BATCH_SIZE):
-        vec0_rows = []
-        chunk_rows = []
-        meta_by_document: dict[int, str] = {}
-        for r in batch:
-            document_id = int(r["document_id"])
-            vec0_rows.append(
-                (r["id"], document_id, r["node_content"], bytes(r["embedding"])),
-            )
-            chunk_rows.append(ChunkRow(r["id"], document_id))
-            meta_by_document[document_id] = str(r["modified"] or "")
-        dst_conn.executemany(
-            "INSERT INTO "
-            + DEFAULT_TABLE_NAME
-            + " (id, document_id, node_content, embedding) VALUES (?, ?, ?, ?)",
-            vec0_rows,
-        )
-        DocumentChunksTable.insert_many(dst_conn, chunk_rows)
-        DocumentMetaTable.upsert_many(
-            dst_conn,
-            (DocumentMetaRow(doc_id, mod) for doc_id, mod in meta_by_document.items()),
-        )
-        live += len(batch)
-    # This migration only ever copies live rows (like compact()), so the
-    # cumulative counter resets to match -- the new file has no bloat yet.
-    IndexMetaTable.reset_total_inserts(dst_conn, live)
-    dst_conn.execute("COMMIT")
-
-
-MIGRATIONS.append(
-    Migration(
-        from_version=1,
-        to_version=2,
-        kind="structural",
-        description=(
-            "document_id TEXT -> INTEGER; move modified into document_meta; "
-            "add document_chunks for O(1) per-document delete"
-        ),
-        apply=_migrate_v1_to_v2,
-    ),
-)

src/paperless_ai/tables.py

@@ -1,245 +0,0 @@
-"""Thin gateways over the plain relational side tables that sit alongside the
-vec0 table. Each method takes the sqlite3.Connection to operate on
-explicitly, rather than owning one -- the store swaps connections during
-compact()/migration, and migrations always work across two connections
-(src_conn, dst_conn) at once.
-
-PRECONDITION: Callers must set conn.row_factory = sqlite3.Row before passing a
-connection to any of these gateways' read methods. The read methods across all
-three classes (DocumentChunksTable.chunk_ids_for_document, IndexMetaTable._get,
-DocumentMetaTable.all_modified_times, DocumentMetaTable.copy_all) use
-row["column_name"] dictionary-style indexing, which requires sqlite3.Row as the
-row factory -- without it, sqlite3.Row is not set, rows are returned as plain
-tuples, and tuple indices must be integers, raising TypeError.
-"""
-
-import sqlite3
-from collections.abc import Iterable
-from typing import NamedTuple
-
-
-class ChunkRow(NamedTuple):
-    chunk_id: str
-    document_id: int
-
-
-class DocumentMetaRow(NamedTuple):
-    document_id: int
-    modified: str
-
-
-class DocumentChunksTable:
-    """chunk_id -> document_id, indexed by document_id. Gives O(1)
-    per-document chunk lookup that vec0's own document_id metadata column
-    cannot (see PaperlessSqliteVecVectorStore._delete_chunks_by_document_id).
-    """
-
-    @staticmethod
-    def create(conn: sqlite3.Connection) -> None:
-        conn.execute(
-            "CREATE TABLE IF NOT EXISTS document_chunks "
-            "(chunk_id TEXT PRIMARY KEY, document_id INTEGER NOT NULL)",
-        )
-        conn.execute(
-            "CREATE INDEX IF NOT EXISTS idx_document_chunks_document_id "
-            "ON document_chunks (document_id)",
-        )
-
-    @staticmethod
-    def insert_many(conn: sqlite3.Connection, rows: Iterable[ChunkRow]) -> None:
-        """rows must already be batch-bounded by the caller (e.g. vec0's own
-        fetchmany() loop) -- this never reads, so it can't itself introduce
-        an unbounded scan, but a whole-table iterable defeats the point."""
-        conn.executemany(
-            "INSERT INTO document_chunks (chunk_id, document_id) VALUES (?, ?)",
-            rows,
-        )
-
-    @staticmethod
-    def chunk_ids_for_document(
-        conn: sqlite3.Connection,
-        document_id: int,
-    ) -> list[str]:
-        return [
-            row["chunk_id"]
-            for row in conn.execute(
-                "SELECT chunk_id FROM document_chunks WHERE document_id = ?",
-                (document_id,),
-            ).fetchall()
-        ]
-
-    @staticmethod
-    def delete_for_document(conn: sqlite3.Connection, document_id: int) -> None:
-        conn.execute(
-            "DELETE FROM document_chunks WHERE document_id = ?",
-            (document_id,),
-        )
-
-    @staticmethod
-    def delete_all(conn: sqlite3.Connection) -> None:
-        conn.execute("DELETE FROM document_chunks")
-
-    @staticmethod
-    def count(conn: sqlite3.Connection) -> int:
-        """Cheap stand-in for vec0's own row count -- see compact()."""
-        return conn.execute("SELECT count(*) FROM document_chunks").fetchone()[0]
-
-
-class DocumentMetaTable:
-    """document_id -> modified, one row per document. Lives outside vec0
-    because vec0 only inlines TEXT metadata up to 12 bytes and `modified`
-    (an ISO timestamp) is always longer.
-    """
-
-    @staticmethod
-    def create(conn: sqlite3.Connection) -> None:
-        conn.execute(
-            "CREATE TABLE IF NOT EXISTS document_meta "
-            "(document_id INTEGER PRIMARY KEY, modified TEXT NOT NULL)",
-        )
-
-    @staticmethod
-    def upsert_many(
-        conn: sqlite3.Connection,
-        rows: Iterable[DocumentMetaRow],
-    ) -> None:
-        conn.executemany(
-            "INSERT INTO document_meta (document_id, modified) VALUES (?, ?) "
-            "ON CONFLICT(document_id) DO UPDATE SET modified = excluded.modified",
-            rows,
-        )
-
-    @staticmethod
-    def delete_for_document(conn: sqlite3.Connection, document_id: int) -> None:
-        conn.execute(
-            "DELETE FROM document_meta WHERE document_id = ?",
-            (document_id,),
-        )
-
-    @staticmethod
-    def delete_all(conn: sqlite3.Connection) -> None:
-        conn.execute("DELETE FROM document_meta")
-
-    @staticmethod
-    def copy_all(
-        src_conn: sqlite3.Connection,
-        dst_conn: sqlite3.Connection,
-        batch_size: int,
-    ) -> None:
-        """Stream document_meta from src_conn into dst_conn in bounded
-        batches. The *only* sanctioned way to move this table across
-        connections (compact()/migrations) -- an unbounded fetchall here
-        would defeat the same OOM-avoidance the vec0 row copy already relies
-        on. batch_size has no default: forces the call site to think about
-        it (pass COMPACT_BATCH_SIZE)."""
-        cursor = src_conn.execute(
-            "SELECT document_id, modified FROM document_meta",
-        )
-        while batch := cursor.fetchmany(batch_size):
-            DocumentMetaTable.upsert_many(
-                dst_conn,
-                (DocumentMetaRow(r["document_id"], r["modified"]) for r in batch),
-            )
-
-    @staticmethod
-    def all_modified_times(conn: sqlite3.Connection) -> dict[str, str]:
-        """Full document_id -> modified map, for get_modified_times()'s
-        public API only. One unbounded read by design (existing behavior).
-        Never use this for cross-connection copying; see copy_all()."""
-        return {
-            str(row["document_id"]): str(row["modified"] or "")
-            for row in conn.execute(
-                "SELECT document_id, modified FROM document_meta",
-            )
-        }
-
-
-class IndexMetaTable:
-    """Typed accessors over index_meta's key/value rows -- replaces
-    PaperlessSqliteVecVectorStore._meta_get_on/_meta_set_on, which returned
-    untyped str | None regardless of whether the key held an int (dim,
-    schema_version, total_inserts) or a string (embed_model).
-    """
-
-    @staticmethod
-    def create(conn: sqlite3.Connection) -> None:
-        conn.execute(
-            "CREATE TABLE IF NOT EXISTS index_meta (key TEXT PRIMARY KEY, value TEXT)",
-        )
-
-    @staticmethod
-    def _get(conn: sqlite3.Connection, key: str) -> str | None:
-        row = conn.execute(
-            "SELECT value FROM index_meta WHERE key = ?",
-            (key,),
-        ).fetchone()
-        return row["value"] if row else None
-
-    @staticmethod
-    def _set(conn: sqlite3.Connection, key: str, value: str) -> None:
-        conn.execute(
-            "INSERT INTO index_meta (key, value) VALUES (?, ?) "
-            "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
-            (key, value),
-        )
-
-    @staticmethod
-    def get_dim(conn: sqlite3.Connection) -> int | None:
-        value = IndexMetaTable._get(conn, "dim")
-        return int(value) if value is not None else None
-
-    @staticmethod
-    def set_dim(conn: sqlite3.Connection, dim: int) -> None:
-        IndexMetaTable._set(conn, "dim", str(dim))
-
-    @staticmethod
-    def get_embed_model(conn: sqlite3.Connection) -> str | None:
-        return IndexMetaTable._get(conn, "embed_model")
-
-    @staticmethod
-    def set_embed_model(conn: sqlite3.Connection, name: str) -> None:
-        IndexMetaTable._set(conn, "embed_model", name)
-
-    @staticmethod
-    def get_schema_version(conn: sqlite3.Connection) -> int | None:
-        value = IndexMetaTable._get(conn, "schema_version")
-        return int(value) if value is not None else None
-
-    @staticmethod
-    def set_schema_version(conn: sqlite3.Connection, version: int) -> None:
-        IndexMetaTable._set(conn, "schema_version", str(version))
-
-    @staticmethod
-    def get_total_inserts(conn: sqlite3.Connection) -> int:
-        value = IndexMetaTable._get(conn, "total_inserts")
-        return int(value) if value is not None else 0
-
-    @staticmethod
-    def increment_total_inserts(conn: sqlite3.Connection, count: int) -> None:
-        """Add ``count`` to the stored counter in one SQL statement (INSERT
-        .. ON CONFLICT DO UPDATE with arithmetic), instead of a separate
-        read-then-write -- called once per add()/upsert_document(), so
-        halving the statement count here is a real, if small, per-call
-        saving. This only avoids a read-then-write race within this single
-        statement; it does not make the counter safe against concurrent
-        writers in general (callers still rely on the write FileLock for
-        that). index_meta.value has TEXT affinity, so the incremented
-        result is stored as its text representation -- get_total_inserts()
-        already expects that (int(value)), so this is not a behavior
-        change, only fewer statements.
-        """
-        conn.execute(
-            "INSERT INTO index_meta (key, value) VALUES ('total_inserts', ?) "
-            "ON CONFLICT(key) DO UPDATE SET value = "
-            "CAST(index_meta.value AS INTEGER) + CAST(excluded.value AS INTEGER)",
-            (str(count),),
-        )
-
-    @staticmethod
-    def reset_total_inserts(conn: sqlite3.Connection, count: int) -> None:
-        """Set total_inserts to an absolute value -- distinct from
-        increment_total_inserts(): used by compact()'s rebuild and by
-        m0001_v1_to_v2 after copying live rows into a fresh file, where
-        total_inserts must become exactly the live row count, not add to
-        whatever the source file's counter held."""
-        IndexMetaTable._set(conn, "total_inserts", str(count))

src/paperless_ai/tests/test_ai_indexing.py

@@ -1,4 +1,3 @@
-import logging
 from pathlib import Path
 from unittest.mock import MagicMock
 from unittest.mock import patch
@@ -187,6 +186,45 @@ def test_truncate_embedding_query_returns_single_chunk() -> None:
     assert "word199" not in result
 
 
+class TestStreamedDocuments:
+    """_StreamedDocuments streams via .iterator() instead of materializing
+    the whole queryset (plus its content and prefetch caches) in memory at
+    once, while still supporting len() so a progress bar wrapped around it
+    shows a real total.
+    """
+
+    def test_len_and_iter_delegate_to_streaming_queryset_methods(
+        self,
+        mocker: pytest_mock.MockerFixture,
+    ) -> None:
+        """
+        GIVEN:
+            - A mock queryset
+        WHEN:
+            - A _StreamedDocuments wrapping it is measured and iterated
+        THEN:
+            - len() uses count() (not a materializing len()), and iteration
+              uses .iterator(chunk_size=...) (not plain iteration, which
+              would materialize prefetches for the whole queryset at once)
+        """
+        mock_queryset = mocker.MagicMock()
+        mock_queryset.count.return_value = 42
+        mock_queryset.iterator.return_value = iter(["doc-1", "doc-2"])
+        streamed = indexing._StreamedDocuments(mock_queryset)
+
+        assert len(streamed) == 42
+        assert list(streamed) == ["doc-1", "doc-2"]
+        # count.call_count isn't asserted exactly: list()'s own size-hint
+        # optimization calls len(streamed) again internally, on top of the
+        # explicit len() call above -- both legitimately delegate to
+        # count(), so only the delegation itself (not the call count) is
+        # the thing being verified here.
+        mock_queryset.count.assert_called_with()
+        mock_queryset.iterator.assert_called_once_with(
+            chunk_size=indexing._INDEX_STREAM_CHUNK_SIZE,
+        )
+
+
 @pytest.mark.django_db
 def test_update_llm_index(
     temp_llm_index_dir: Path,
@@ -253,45 +291,6 @@ def test_update_llm_index_rebuilds_on_model_name_change(
         assert store.stored_model_name() == "model-b"
 
 
-@pytest.mark.django_db
-def test_update_llm_index_merges_exists_and_config_mismatch_reads(
-    temp_llm_index_dir: Path,
-    real_document: Document,
-    mock_embed_model: FakeEmbedding,
-) -> None:
-    # Build an initial index so the second call's table_exists()/
-    # config_mismatch() checks have something real to check against.
-    with patch("documents.models.Document.objects.all") as mock_all:
-        mock_queryset = MagicMock()
-        mock_queryset.exists.return_value = True
-        mock_queryset.__iter__.return_value = iter([real_document])
-        mock_queryset.select_related.return_value = mock_queryset
-        mock_queryset.prefetch_related.return_value = mock_queryset
-        mock_all.return_value = mock_queryset
-        indexing.update_llm_index(rebuild=True)
-
-    with patch("documents.models.Document.objects.all") as mock_all:
-        mock_queryset = MagicMock()
-        mock_queryset.exists.return_value = True
-        mock_queryset.__iter__.return_value = iter([real_document])
-        mock_queryset.select_related.return_value = mock_queryset
-        mock_queryset.prefetch_related.return_value = mock_queryset
-        mock_all.return_value = mock_queryset
-        with patch(
-            "paperless_ai.indexing.read_store",
-            wraps=indexing.read_store,
-        ) as read_store_spy:
-            indexing.update_llm_index(rebuild=False)
-
-    # Documents exist, so the fast-exit check's `no_documents and ...`
-    # short-circuits before ever calling llm_index_exists() -- the only
-    # read_store() call left in this path is the merged table_exists()/
-    # config_mismatch() check. Before this task's fix, that merged check
-    # was two separate read_store() calls (one inside llm_index_exists(),
-    # one for config_mismatch() right after) -- so this asserts 1, not 2.
-    assert read_store_spy.call_count == 1
-
-
 @pytest.mark.django_db
 def test_update_llm_index_partial_update(
     temp_llm_index_dir: Path,
@@ -777,7 +776,6 @@ class TestLlmIndexLocking:
         mocker: pytest_mock.MockerFixture,
     ) -> None:
         mock_store = MagicMock()
-        mock_store.has_pending_migration.return_value = False
         mocker.patch(
             "paperless_ai.indexing.write_store",
             return_value=mocker.MagicMock(
@@ -798,45 +796,12 @@ class TestLlmIndexLocking:
 
         mock_store.upsert_document.assert_called_once()
 
-    def test_add_or_update_document_skips_write_when_reembed_pending(
-        self,
-        temp_llm_index_dir: Path,
-        mock_embed_model: FakeEmbedding,
-        mocker: pytest_mock.MockerFixture,
-    ) -> None:
-        """A pending re-embed migration must block the incremental write,
-        not let it proceed against a schema that just changed underneath it.
-        """
-        mock_store = MagicMock()
-        mock_store.has_pending_migration.return_value = True
-        mock_store.check_and_run_migrations.return_value = True
-        mocker.patch(
-            "paperless_ai.indexing.write_store",
-            return_value=mocker.MagicMock(
-                __enter__=mocker.MagicMock(return_value=mock_store),
-                __exit__=mocker.MagicMock(return_value=False),
-            ),
-        )
-        mock_node = MagicMock()
-        mock_node.get_content.return_value = "fake node text"
-        mocker.patch(
-            "paperless_ai.indexing.build_document_node",
-            return_value=[mock_node],
-        )
-
-        doc = MagicMock(spec=Document)
-        doc.id = 1
-        indexing.llm_index_add_or_update_document(doc)
-
-        mock_store.upsert_document.assert_not_called()
-
     def test_remove_document_uses_write_store(
         self,
         temp_llm_index_dir: Path,
         mocker: pytest_mock.MockerFixture,
     ) -> None:
         mock_store = MagicMock()
-        mock_store.has_pending_migration.return_value = False
         mocker.patch(
             "paperless_ai.indexing.write_store",
             return_value=mocker.MagicMock(
@@ -851,31 +816,6 @@ class TestLlmIndexLocking:
 
         mock_store.delete.assert_called_once_with("1")
 
-    def test_remove_document_skips_write_when_reembed_pending(
-        self,
-        temp_llm_index_dir: Path,
-        mocker: pytest_mock.MockerFixture,
-    ) -> None:
-        """A pending re-embed migration must block the delete too, for the
-        same consistency reason as the incremental-update path.
-        """
-        mock_store = MagicMock()
-        mock_store.has_pending_migration.return_value = True
-        mock_store.check_and_run_migrations.return_value = True
-        mocker.patch(
-            "paperless_ai.indexing.write_store",
-            return_value=mocker.MagicMock(
-                __enter__=mocker.MagicMock(return_value=mock_store),
-                __exit__=mocker.MagicMock(return_value=False),
-            ),
-        )
-
-        doc = MagicMock(spec=Document)
-        doc.id = 1
-        indexing.llm_index_remove_document(doc)
-
-        mock_store.delete.assert_not_called()
-
     def test_update_llm_index_rebuild_uses_write_store(
         self,
         temp_llm_index_dir: Path,
@@ -948,76 +888,6 @@ class TestVectorStoreIndexing:
             assert rows >= 1
 
 
-class TestLlmIndexMigrate:
-    def test_noop_when_ai_disabled(self, mocker: pytest_mock.MockerFixture) -> None:
-        """
-        GIVEN:
-            - AI/LLM index support is disabled in configuration
-        WHEN:
-            - llm_index_migrate() is called
-        THEN:
-            - No store is opened and no migration check runs
-        """
-        mocker.patch(
-            "paperless_ai.indexing.AIConfig",
-            return_value=mocker.Mock(llm_index_enabled=False),
-        )
-        write_store_mock = mocker.patch("paperless_ai.indexing.write_store")
-        indexing.llm_index_migrate()
-        write_store_mock.assert_not_called()
-
-    def test_runs_pending_migration_when_enabled(
-        self,
-        mocker: pytest_mock.MockerFixture,
-    ) -> None:
-        """
-        GIVEN:
-            - AI/LLM index support is enabled
-        WHEN:
-            - llm_index_migrate() is called
-        THEN:
-            - The store is opened for write and a migration check runs
-        """
-        mocker.patch(
-            "paperless_ai.indexing.AIConfig",
-            return_value=mocker.Mock(llm_index_enabled=True),
-        )
-        store_mock = mocker.MagicMock()
-        store_mock.has_pending_migration.return_value = False
-        write_store_cm = mocker.patch("paperless_ai.indexing.write_store")
-        write_store_cm.return_value.__enter__.return_value = store_mock
-        indexing.llm_index_migrate()
-        store_mock.has_pending_migration.assert_called_once()
-
-    def test_logs_warning_when_reembed_needed(
-        self,
-        mocker: pytest_mock.MockerFixture,
-        caplog: pytest.LogCaptureFixture,
-    ) -> None:
-        """
-        GIVEN:
-            - AI/LLM index support is enabled
-            - A pending migration requires re-embedding
-        WHEN:
-            - llm_index_migrate() is called
-        THEN:
-            - A warning directs the operator to run a manual rebuild, since
-              this automatic check must never re-embed on its own
-        """
-        mocker.patch(
-            "paperless_ai.indexing.AIConfig",
-            return_value=mocker.Mock(llm_index_enabled=True),
-        )
-        store_mock = mocker.MagicMock()
-        store_mock.has_pending_migration.return_value = True
-        store_mock.check_and_run_migrations.return_value = True
-        write_store_cm = mocker.patch("paperless_ai.indexing.write_store")
-        write_store_cm.return_value.__enter__.return_value = store_mock
-        with caplog.at_level(logging.WARNING, logger="paperless_ai.indexing"):
-            indexing.llm_index_migrate()
-        assert "requires re-embedding" in caplog.text
-
-
 @pytest.mark.django_db
 class TestQuerySimilarDocuments:
     def test_query_similar_documents_respects_allowed_ids(

src/paperless_ai/tests/test_tables.py

@@ -1,340 +0,0 @@
-import sqlite3
-from collections.abc import Generator
-
-import pytest
-from pytest_mock import MockerFixture
-
-from paperless_ai.tables import ChunkRow
-from paperless_ai.tables import DocumentChunksTable
-from paperless_ai.tables import DocumentMetaRow
-from paperless_ai.tables import DocumentMetaTable
-from paperless_ai.tables import IndexMetaTable
-
-
-@pytest.fixture
-def conn() -> Generator[sqlite3.Connection, None, None]:
-    connection = sqlite3.connect(":memory:")
-    connection.row_factory = sqlite3.Row
-    try:
-        yield connection
-    finally:
-        connection.close()
-
-
-class TestDocumentChunksTable:
-    def test_create_is_idempotent(self, conn: sqlite3.Connection) -> None:
-        """
-        GIVEN:
-            - A bare sqlite3 connection
-        WHEN:
-            - create() is called, a row is inserted, then create() is called again
-        THEN:
-            - No error is raised and the row survives uncorrupted
-        """
-        DocumentChunksTable.create(conn)
-        DocumentChunksTable.insert_many(conn, [ChunkRow("c1", 1)])
-        DocumentChunksTable.create(conn)
-        assert DocumentChunksTable.chunk_ids_for_document(conn, 1) == ["c1"]
-
-    def test_insert_many_then_lookup_by_document_id(
-        self,
-        conn: sqlite3.Connection,
-    ) -> None:
-        """
-        GIVEN:
-            - An empty document_chunks table
-        WHEN:
-            - Two chunks for document 1 and one for document 2 are inserted
-        THEN:
-            - chunk_ids_for_document returns exactly the matching chunk ids
-        """
-        DocumentChunksTable.create(conn)
-        DocumentChunksTable.insert_many(
-            conn,
-            [ChunkRow("c1", 1), ChunkRow("c2", 1), ChunkRow("c3", 2)],
-        )
-        assert sorted(DocumentChunksTable.chunk_ids_for_document(conn, 1)) == [
-            "c1",
-            "c2",
-        ]
-        assert DocumentChunksTable.chunk_ids_for_document(conn, 2) == ["c3"]
-        assert DocumentChunksTable.chunk_ids_for_document(conn, 999) == []
-
-    def test_delete_for_document_removes_only_that_document(
-        self,
-        conn: sqlite3.Connection,
-    ) -> None:
-        """
-        GIVEN:
-            - Chunks for two different documents
-        WHEN:
-            - delete_for_document() is called for one of them
-        THEN:
-            - Only that document's chunks are removed
-        """
-        DocumentChunksTable.create(conn)
-        DocumentChunksTable.insert_many(
-            conn,
-            [ChunkRow("c1", 1), ChunkRow("c2", 2)],
-        )
-        DocumentChunksTable.delete_for_document(conn, 1)
-        assert DocumentChunksTable.chunk_ids_for_document(conn, 1) == []
-        assert DocumentChunksTable.chunk_ids_for_document(conn, 2) == ["c2"]
-
-    def test_delete_all_clears_every_row(self, conn: sqlite3.Connection) -> None:
-        """
-        GIVEN:
-            - Chunks for multiple documents
-        WHEN:
-            - delete_all() is called
-        THEN:
-            - count() returns 0
-        """
-        DocumentChunksTable.create(conn)
-        DocumentChunksTable.insert_many(
-            conn,
-            [ChunkRow("c1", 1), ChunkRow("c2", 2)],
-        )
-        DocumentChunksTable.delete_all(conn)
-        assert DocumentChunksTable.count(conn) == 0
-
-    def test_count_reflects_live_rows(self, conn: sqlite3.Connection) -> None:
-        """
-        GIVEN:
-            - An empty document_chunks table
-        WHEN:
-            - Rows are inserted then one document's rows are deleted
-        THEN:
-            - count() reflects the remaining row count
-        """
-        DocumentChunksTable.create(conn)
-        DocumentChunksTable.insert_many(
-            conn,
-            [ChunkRow("c1", 1), ChunkRow("c2", 1), ChunkRow("c3", 2)],
-        )
-        assert DocumentChunksTable.count(conn) == 3
-        DocumentChunksTable.delete_for_document(conn, 1)
-        assert DocumentChunksTable.count(conn) == 1
-
-
-class TestDocumentMetaTable:
-    def test_upsert_many_then_all_modified_times(
-        self,
-        conn: sqlite3.Connection,
-    ) -> None:
-        """
-        GIVEN:
-            - An empty document_meta table
-        WHEN:
-            - Two documents' modified timestamps are upserted
-        THEN:
-            - all_modified_times() returns both, keyed by str(document_id)
-        """
-        DocumentMetaTable.create(conn)
-        DocumentMetaTable.upsert_many(
-            conn,
-            [
-                DocumentMetaRow(1, "2026-01-01T00:00:00"),
-                DocumentMetaRow(2, "2026-02-02T00:00:00"),
-            ],
-        )
-        assert DocumentMetaTable.all_modified_times(conn) == {
-            "1": "2026-01-01T00:00:00",
-            "2": "2026-02-02T00:00:00",
-        }
-
-    def test_upsert_many_overwrites_existing_value(
-        self,
-        conn: sqlite3.Connection,
-    ) -> None:
-        """
-        GIVEN:
-            - A document_meta row for document 1
-        WHEN:
-            - upsert_many() is called again with a new modified value for
-              the same document_id
-        THEN:
-            - The stored value is replaced, not duplicated
-        """
-        DocumentMetaTable.create(conn)
-        DocumentMetaTable.upsert_many(conn, [DocumentMetaRow(1, "old")])
-        DocumentMetaTable.upsert_many(conn, [DocumentMetaRow(1, "new")])
-        assert DocumentMetaTable.all_modified_times(conn) == {"1": "new"}
-
-    def test_delete_for_document_removes_only_that_row(
-        self,
-        conn: sqlite3.Connection,
-    ) -> None:
-        """
-        GIVEN:
-            - document_meta rows for two documents
-        WHEN:
-            - delete_for_document() is called for one of them
-        THEN:
-            - Only that document's row is removed
-        """
-        DocumentMetaTable.create(conn)
-        DocumentMetaTable.upsert_many(
-            conn,
-            [DocumentMetaRow(1, "a"), DocumentMetaRow(2, "b")],
-        )
-        DocumentMetaTable.delete_for_document(conn, 1)
-        assert DocumentMetaTable.all_modified_times(conn) == {"2": "b"}
-
-    def test_delete_all_clears_every_row(self, conn: sqlite3.Connection) -> None:
-        """
-        GIVEN:
-            - document_meta rows for multiple documents
-        WHEN:
-            - delete_all() is called
-        THEN:
-            - all_modified_times() returns an empty dict
-        """
-        DocumentMetaTable.create(conn)
-        DocumentMetaTable.upsert_many(
-            conn,
-            [DocumentMetaRow(1, "a"), DocumentMetaRow(2, "b")],
-        )
-        DocumentMetaTable.delete_all(conn)
-        assert DocumentMetaTable.all_modified_times(conn) == {}
-
-    def test_copy_all_streams_every_row_to_destination(
-        self,
-        conn: sqlite3.Connection,
-    ) -> None:
-        """
-        GIVEN:
-            - A source connection with document_meta rows for 5 documents
-            - A separate, empty destination connection
-        WHEN:
-            - copy_all() is called with a batch size smaller than the row
-              count, forcing multiple fetchmany() cycles
-        THEN:
-            - Every row is present on the destination connection
-        """
-        DocumentMetaTable.create(conn)
-        DocumentMetaTable.upsert_many(
-            conn,
-            [DocumentMetaRow(i, f"modified-{i}") for i in range(5)],
-        )
-        dst_conn = sqlite3.connect(":memory:")
-        dst_conn.row_factory = sqlite3.Row
-        try:
-            DocumentMetaTable.create(dst_conn)
-            DocumentMetaTable.copy_all(conn, dst_conn, batch_size=2)
-            assert DocumentMetaTable.all_modified_times(dst_conn) == {
-                str(i): f"modified-{i}" for i in range(5)
-            }
-        finally:
-            dst_conn.close()
-
-
-class TestIndexMetaTable:
-    @pytest.mark.parametrize(
-        ("setter_name", "getter_name", "value"),
-        [
-            ("set_dim", "get_dim", 384),
-            ("set_embed_model", "get_embed_model", "model-a"),
-            ("set_schema_version", "get_schema_version", 2),
-        ],
-    )
-    def test_typed_accessor_roundtrip(
-        self,
-        conn: sqlite3.Connection,
-        setter_name: str,
-        getter_name: str,
-        value: int | str,
-    ) -> None:
-        """
-        GIVEN:
-            - An empty index_meta table
-        WHEN:
-            - A typed accessor's setter is called then the getter is read back
-        THEN:
-            - The same value is returned, correctly typed (int or str)
-        """
-        IndexMetaTable.create(conn)
-        getter = getattr(IndexMetaTable, getter_name)
-        setter = getattr(IndexMetaTable, setter_name)
-        assert getter(conn) is None
-        setter(conn, value)
-        assert getter(conn) == value
-
-    def test_total_inserts_starts_at_zero(self, conn: sqlite3.Connection) -> None:
-        """
-        GIVEN:
-            - An empty index_meta table
-        WHEN:
-            - get_total_inserts() is read before anything is set
-        THEN:
-            - 0 is returned
-        """
-        IndexMetaTable.create(conn)
-        assert IndexMetaTable.get_total_inserts(conn) == 0
-
-    def test_increment_total_inserts_accumulates(
-        self,
-        conn: sqlite3.Connection,
-    ) -> None:
-        """
-        GIVEN:
-            - An empty index_meta table
-        WHEN:
-            - increment_total_inserts() is called twice
-        THEN:
-            - get_total_inserts() returns the running sum
-        """
-        IndexMetaTable.create(conn)
-        IndexMetaTable.increment_total_inserts(conn, 5)
-        IndexMetaTable.increment_total_inserts(conn, 3)
-        assert IndexMetaTable.get_total_inserts(conn) == 8
-
-    def test_increment_total_inserts_is_a_single_statement(
-        self,
-        mocker: MockerFixture,
-    ) -> None:
-        """
-        GIVEN:
-            - An empty index_meta table
-        WHEN:
-            - increment_total_inserts() is called
-        THEN:
-            - Exactly one conn.execute() call is made (a single INSERT ...
-              ON CONFLICT DO UPDATE, not a separate read then write)
-        """
-
-        # sqlite3.Connection is an immutable C extension type with no
-        # instance __dict__, so mocker.spy(conn, "execute") can't shadow
-        # "execute" on a plain connection ("attribute 'execute' is
-        # read-only"). A trivial Python subclass gets a normal instance
-        # __dict__, making the instance spyable while still being a real,
-        # usable sqlite3.Connection.
-        class _SpyableConnection(sqlite3.Connection):
-            pass
-
-        conn = sqlite3.connect(":memory:", factory=_SpyableConnection)
-        try:
-            conn.row_factory = sqlite3.Row
-            IndexMetaTable.create(conn)
-            execute_spy = mocker.spy(conn, "execute")
-            IndexMetaTable.increment_total_inserts(conn, 5)
-            assert execute_spy.call_count == 1
-        finally:
-            conn.close()
-
-    def test_reset_total_inserts_sets_absolute_value(
-        self,
-        conn: sqlite3.Connection,
-    ) -> None:
-        """
-        GIVEN:
-            - A total_inserts counter already at a high value
-        WHEN:
-            - reset_total_inserts() is called with a lower value
-        THEN:
-            - get_total_inserts() returns exactly that value, not a sum
-        """
-        IndexMetaTable.create(conn)
-        IndexMetaTable.increment_total_inserts(conn, 100)
-        IndexMetaTable.reset_total_inserts(conn, 7)
-        assert IndexMetaTable.get_total_inserts(conn) == 7

src/paperless_ai/tests/test_vector_store.py

@@ -1,35 +1,28 @@
-import inspect
 import sqlite3
 from collections.abc import Generator
 from pathlib import Path
 
 import pytest
-import sqlite_vec
 from llama_index.core.schema import TextNode
 from llama_index.core.vector_stores.types import FilterOperator
 from llama_index.core.vector_stores.types import MetadataFilter
 from llama_index.core.vector_stores.types import MetadataFilters
 from llama_index.core.vector_stores.types import VectorStoreQuery
-from pytest_mock import MockerFixture
 
-from paperless_ai.migrations import MIGRATIONS
-from paperless_ai.migrations import Migration
-from paperless_ai.migrations import m0001_v1_to_v2
-from paperless_ai.tables import DocumentChunksTable
-from paperless_ai.tables import DocumentMetaTable
 from paperless_ai.vector_store import DB_FILENAME
 from paperless_ai.vector_store import DEFAULT_TABLE_NAME
+from paperless_ai.vector_store import MIGRATIONS
 from paperless_ai.vector_store import SCHEMA_VERSION
+from paperless_ai.vector_store import Migration
 from paperless_ai.vector_store import PaperlessSqliteVecVectorStore
 from paperless_ai.vector_store import _build_where
-from paperless_ai.vector_store import _pack
 
 DIM = 16
 
 
 def make_node(
     node_id: str,
-    document_id: int,
+    document_id: str,
     *,
     modified: str = "2026-06-10T00:00:00",
     seed: float = 0.0,
@@ -66,13 +59,13 @@ def _query(
     )
 
 
-def _eq_filter(key: str, value: int):
+def _eq_filter(key: str, value: str):
     return MetadataFilters(
         filters=[MetadataFilter(key=key, operator=FilterOperator.EQ, value=value)],
     )
 
 
-def _in_filter(document_ids: list[int]):
+def _in_filter(document_ids: list[str]):
     return MetadataFilters(
         filters=[
             MetadataFilter(
@@ -84,7 +77,7 @@ def _in_filter(document_ids: list[int]):
     )
 
 
-def _ne_filter(document_id: int):
+def _ne_filter(document_id: str):
     return MetadataFilters(
         filters=[
             MetadataFilter(
@@ -98,11 +91,11 @@ def _ne_filter(document_id: int):
 
 class TestCrud:
     def test_add_then_query_returns_node(self, store) -> None:
-        node = make_node("n1", 1)
+        node = make_node("n1", "1")
         assert store.add([node]) == ["n1"]
         result = _query(store, node.embedding, top_k=1)
         assert result.ids == ["n1"]
-        assert result.nodes[0].metadata["document_id"] == 1
+        assert result.nodes[0].metadata["document_id"] == "1"
         # cosine distance of the identical vector is 0 -> similarity 1
         assert result.similarities[0] == pytest.approx(1.0)
 
@@ -115,58 +108,58 @@ class TestCrud:
         assert not store.table_exists()
 
     def test_delete_removes_all_chunks_of_document(self, store) -> None:
-        store.add([make_node("a1", 1), make_node("a2", 1), make_node("b1", 2)])
-        store.delete(1)
+        store.add([make_node("a1", "1"), make_node("a2", "1"), make_node("b1", "2")])
+        store.delete("1")
         result = _query(store, [0.0] * DIM, top_k=10)
         assert result.ids == ["b1"]
 
     def test_query_with_in_filter_scopes_results(self, store) -> None:
         store.add(
             [
-                make_node("a1", 1, seed=0.0),
-                make_node("b1", 2, seed=1.0),
-                make_node("c1", 3, seed=2.0),
+                make_node("a1", "1", seed=0.0),
+                make_node("b1", "2", seed=1.0),
+                make_node("c1", "3", seed=2.0),
             ],
         )
-        result = _query(store, [0.0] * DIM, top_k=10, filters=_in_filter([2, 3]))
+        result = _query(store, [0.0] * DIM, top_k=10, filters=_in_filter(["2", "3"]))
         assert sorted(result.ids) == ["b1", "c1"]
 
     def test_query_respects_top_k_with_filter(self, store) -> None:
         # k semantics: global top-k even with IN filters (document_id is a
         # metadata column, not a partition key -- see design doc).
         store.add(
-            [make_node(f"n{i}", i % 4, seed=float(i)) for i in range(12)],
+            [make_node(f"n{i}", str(i % 4), seed=float(i)) for i in range(12)],
         )
         result = _query(
             store,
             [0.0] * DIM,
             top_k=3,
-            filters=_in_filter([0, 1, 2, 3]),
+            filters=_in_filter(["0", "1", "2", "3"]),
         )
         assert len(result.ids) == 3
         assert result.similarities == sorted(result.similarities, reverse=True)
 
     def test_get_nodes_filter_and_empty_paths(self, store) -> None:
-        assert store.get_nodes(filters=_in_filter([1])) == []  # no table yet
-        store.add([make_node("a1", 1), make_node("b1", 2)])
-        nodes = store.get_nodes(filters=_in_filter([1]))
+        assert store.get_nodes(filters=_in_filter(["1"])) == []  # no table yet
+        store.add([make_node("a1", "1"), make_node("b1", "2")])
+        nodes = store.get_nodes(filters=_in_filter(["1"]))
         assert [n.node_id for n in nodes] == ["a1"]
         assert nodes[0].embedding is not None
-        assert store.get_nodes(filters=_in_filter([999])) == []
+        assert store.get_nodes(filters=_in_filter(["999"])) == []
 
     def test_query_with_eq_filter_scopes_results(self, store) -> None:
         store.add(
             [
-                make_node("a1", 1, seed=0.0),
-                make_node("b1", 2, seed=1.0),
-                make_node("c1", 3, seed=2.0),
+                make_node("a1", "1", seed=0.0),
+                make_node("b1", "2", seed=1.0),
+                make_node("c1", "3", seed=2.0),
             ],
         )
         result = _query(
             store,
             [0.0] * DIM,
             top_k=10,
-            filters=_eq_filter("document_id", 2),
+            filters=_eq_filter("document_id", "2"),
         )
         assert result.ids == ["b1"]
 
@@ -175,7 +168,7 @@ class TestCrud:
             store.get_nodes(node_ids=["x"])
 
     def test_fresh_instance_sees_existing_table(self, store, tmp_path: Path) -> None:
-        store.add([make_node("a1", 1)])
+        store.add([make_node("a1", "1")])
         with PaperlessSqliteVecVectorStore(uri=str(tmp_path)) as reopened:
             assert reopened.table_exists()
             assert reopened.vector_dim() == DIM
@@ -183,98 +176,23 @@ class TestCrud:
 
     def test_table_exists_and_drop(self, store) -> None:
         assert not store.table_exists()
-        store.add([make_node("a1", 1)])
+        store.add([make_node("a1", "1")])
         assert store.table_exists()
         store.drop_table()
         assert not store.table_exists()
         assert store.vector_dim() is None
 
-    def test_document_id_stored_as_integer_in_vec0(
-        self,
-        store: PaperlessSqliteVecVectorStore,
-    ) -> None:
-        """
-        GIVEN:
-            - An empty vector store
-        WHEN:
-            - A node is added with an int document_id
-        THEN:
-            - vec0's own document_id column holds an INTEGER, not TEXT
-        """
-        store.add([make_node("a1", 1)])
-        row = store.client.execute(
-            "SELECT document_id FROM documents WHERE id = 'a1'",
-        ).fetchone()
-        assert isinstance(row["document_id"], int)
-
-    def test_drop_table_clears_modified_times(
-        self,
-        store: PaperlessSqliteVecVectorStore,
-    ) -> None:
-        """
-        GIVEN:
-            - A store with a tracked document's modified time
-        WHEN:
-            - drop_table() is called
-        THEN:
-            - document_meta and document_chunks are both cleared directly
-              (asserted against the tables themselves, not via
-              get_modified_times()/table_exists() -- those short-circuit on
-              the vec0 table being gone, which drop_table() does first, so
-              they would pass even if DocumentMetaTable.delete_all()/
-              DocumentChunksTable.delete_all() were never called)
-        """
-        store.add([make_node("a1", 1)])
-        store.drop_table()
-        assert (
-            store.client.execute(
-                "SELECT count(*) FROM document_meta",
-            ).fetchone()[0]
-            == 0
-        )
-        assert (
-            store.client.execute(
-                "SELECT count(*) FROM document_chunks",
-            ).fetchone()[0]
-            == 0
-        )
-
-    def test_upsert_document_checks_table_exists_once(
-        self,
-        store: PaperlessSqliteVecVectorStore,
-        mocker: MockerFixture,
-    ) -> None:
-        """
-        GIVEN:
-            - An existing store with one document already indexed
-        WHEN:
-            - upsert_document() replaces that document's chunks
-        THEN:
-            - table_exists() is queried at most once per call, not twice
-              (previously: once via _ensure_table(), once via the separate
-              `if self.table_exists():` delete-chunks guard)
-        """
-        store.add([make_node("a1", 1)])
-        # store is a pydantic model, whose __setattr__/__delattr__ reject
-        # arbitrary instance attributes ("object has no attribute
-        # 'table_exists'"), so mocker.spy(store, "table_exists") can't
-        # shadow the method on the instance. Spying on the class works
-        # (bound method lookup on the instance still resolves through it).
-        exists_spy = mocker.spy(PaperlessSqliteVecVectorStore, "table_exists")
-        store.upsert_document(1, [make_node("a2", 1)])
-        assert exists_spy.call_count == 1
-
 
 class TestBuildWhere:
     def test_ne_filter_translates_to_not_equal_clause(self) -> None:
-        where, params = _build_where(_ne_filter(1))
+        where, params = _build_where(_ne_filter("1"))
         assert where == "(document_id != ?)"
-        assert params == [1]
+        assert params == ["1"]
 
     def test_query_with_ne_filter_excludes_matching_document(self, store) -> None:
-        store.add([make_node("a1", 1), make_node("b1", 2)])
+        store.add([make_node("a1", "1"), make_node("b1", "2")])
         assert sorted(
-            _query(store, [0.0] * DIM, top_k=5, filters=_ne_filter(1)).ids,
+            _query(store, [0.0] * DIM, top_k=5, filters=_ne_filter("1")).ids,
         ) == [
             "b1",
         ]
@@ -288,7 +206,7 @@ class TestBuildWhere:
                 MetadataFilter(
                     key="document_id",
                     operator=FilterOperator.EQ,
-                    value=1,
+                    value="1",
                 ),
             ],
         )
@@ -297,13 +215,13 @@ class TestBuildWhere:
         assert params == []
 
     def test_query_with_untranslatable_filter_returns_no_rows(self, store) -> None:
-        store.add([make_node("a1", 1), make_node("b1", 2)])
+        store.add([make_node("a1", "1"), make_node("b1", "2")])
         nested = MetadataFilters(
             filters=[
                 MetadataFilter(
                     key="document_id",
                     operator=FilterOperator.EQ,
-                    value=1,
+                    value="1",
                 ),
             ],
         )
@@ -316,19 +234,19 @@ class TestBuildWhere:
 class TestUpsert:
     def test_upsert_replaces_and_prunes_stale_chunks(self, store) -> None:
         store.add(
-            [make_node("d1c1", 1), make_node("d1c2", 1), make_node("d2c1", 2)],
+            [make_node("d1c1", "1"), make_node("d1c2", "1"), make_node("d2c1", "2")],
         )
-        store.upsert_document(1, [make_node("d1new", 1)])
+        store.upsert_document("1", [make_node("d1new", "1")])
         result = _query(store, [0.0] * DIM, top_k=10)
         assert sorted(result.ids) == ["d1new", "d2c1"]
 
     def test_upsert_creates_table_when_missing(self, store) -> None:
-        store.upsert_document(1, [make_node("a1", 1)])
+        store.upsert_document("1", [make_node("a1", "1")])
         assert _query(store, [0.0] * DIM, top_k=1).ids == ["a1"]
 
     def test_upsert_empty_nodes_removes_document(self, store) -> None:
-        store.add([make_node("a1", 1), make_node("b1", 2)])
-        store.upsert_document(1, [])
+        store.add([make_node("a1", "1"), make_node("b1", "2")])
+        store.upsert_document("1", [])
         assert _query(store, [0.0] * DIM, top_k=10).ids == ["b1"]
 
     def test_upsert_is_atomic_for_concurrent_readers(
@@ -337,16 +255,16 @@ class TestUpsert:
         tmp_path: Path,
     ) -> None:
         """A second connection must never observe document 1 half-replaced."""
-        store.add([make_node("a1", 1), make_node("a2", 1)])
+        store.add([make_node("a1", "1"), make_node("a2", "1")])
         with PaperlessSqliteVecVectorStore(uri=str(tmp_path)) as reader:
-            store.upsert_document(1, [make_node("a3", 1)])
-            ids = [n.node_id for n in reader.get_nodes(filters=_in_filter([1]))]
+            store.upsert_document("1", [make_node("a3", "1")])
+            ids = [n.node_id for n in reader.get_nodes(filters=_in_filter(["1"]))]
             assert ids == ["a3"]
 
 
 class TestMetadataCoercion:
     def test_none_metadata_values_become_empty_strings(self, store) -> None:
-        node = make_node("a1", 1)
+        node = make_node("a1", "1")
         node.metadata["modified"] = None
         store.add([node])  # must not raise (vec0 rejects NULL metadata)
         assert store.get_modified_times() == {"1": ""}
@@ -365,7 +283,7 @@ class TestModelNameTracking:
             uri=str(tmp_path),
             embed_model_name="model-a",
         ) as store:
-            store.add([make_node("a1", 1)])
+            store.add([make_node("a1", "1")])
             assert store.stored_model_name() == "model-a"
         with PaperlessSqliteVecVectorStore(uri=str(tmp_path)) as reopened:
             assert reopened.stored_model_name() == "model-a"
@@ -376,7 +294,7 @@ class TestModelNameTracking:
             embed_model_name="model-a",
         ) as store:
             assert not store.config_mismatch("anything")  # no table yet
-            store.add([make_node("a1", 1)])
+            store.add([make_node("a1", "1")])
             assert not store.config_mismatch("model-a")
             assert store.config_mismatch("model-b")
 
@@ -385,7 +303,7 @@ class TestModelNameTracking:
         tmp_path: Path,
     ) -> None:
         with PaperlessSqliteVecVectorStore(uri=str(tmp_path)) as store:  # no model name
-            store.add([make_node("a1", 1)])
+            store.add([make_node("a1", "1")])
             assert not store.config_mismatch("model-a")
 
 
@@ -396,9 +314,9 @@ class TestGetModifiedTimes:
     def test_returns_one_entry_per_document(self, store) -> None:
         store.add(
             [
-                make_node("a1", 1, modified="2026-01-01T00:00:00"),
-                make_node("a2", 1, modified="2026-01-01T00:00:00"),
-                make_node("b1", 2, modified="2026-02-02T00:00:00"),
+                make_node("a1", "1", modified="2026-01-01T00:00:00"),
+                make_node("a2", "1", modified="2026-01-01T00:00:00"),
+                make_node("b1", "2", modified="2026-02-02T00:00:00"),
             ],
         )
         assert store.get_modified_times() == {
@@ -423,35 +341,37 @@ class TestCompact:
     def _churn(self, store, cycles: int) -> None:
         for i in range(cycles):
             store.upsert_document(
-                1,
-                [make_node(f"gen{i}-{j}", 1, seed=float(j)) for j in range(20)],
+                "1",
+                [make_node(f"gen{i}-{j}", "1", seed=float(j)) for j in range(20)],
             )
 
     def test_compact_noop_below_threshold(self, store) -> None:
-        store.add([make_node("a1", 1)])
+        store.add([make_node("a1", "1")])
         store.compact()
         assert _query(store, [0.0] * DIM, top_k=1).ids == ["a1"]
 
     def test_force_compact_preserves_rows_and_metadata(self, store) -> None:
-        store.add([make_node("a1", 1), make_node("b1", 2, seed=3.0)])
+        store.add([make_node("a1", "1"), make_node("b1", "2", seed=3.0)])
         self._churn(store, 5)
         before = {
-            n.node_id: n.metadata for n in store.get_nodes(filters=_in_filter([1, 2]))
+            n.node_id: n.metadata
+            for n in store.get_nodes(filters=_in_filter(["1", "2"]))
         }
         store.compact(force=True)
         after = {
-            n.node_id: n.metadata for n in store.get_nodes(filters=_in_filter([1, 2]))
+            n.node_id: n.metadata
+            for n in store.get_nodes(filters=_in_filter(["1", "2"]))
         }
         assert after == before
         assert self._bloat_ratio(store) == pytest.approx(1.0)
         # store remains fully usable after the rebuild; use a seed far from all
         # existing nodes (gen4-0..gen4-19 have seeds 0..19) so cosine KNN is
         # unambiguous at top_k=1.
-        store.upsert_document(3, [make_node("c1", 3, seed=100.0)])
+        store.upsert_document("3", [make_node("c1", "3", seed=100.0)])
         assert "c1" in _query(store, [100.0] * DIM, top_k=1).ids
 
     def test_auto_compact_triggers_on_churn(self, store) -> None:
-        store.add([make_node(f"s{j}", 1, seed=float(j)) for j in range(20)])
+        store.add([make_node(f"s{j}", "1", seed=float(j)) for j in range(20)])
         self._churn(store, 5)
         assert self._bloat_ratio(store) > 2
         store.compact()
@@ -473,7 +393,7 @@ class TestCompact:
         but a concurrent reader keeps -wal/-shm alive, so the cleanup must
         unlink them explicitly (as the structural-migration path does).
         """
-        store.add([make_node("a1", 1)])
+        store.add([make_node("a1", "1")])
         compact_path = str(tmp_path / DB_FILENAME) + ".compact"
         held: list[sqlite3.Connection] = []
 
@@ -509,75 +429,16 @@ class TestCompact:
         regression in the streaming loop (dropped tail, off-by-one) surfaces.
         """
         monkeypatch.setattr("paperless_ai.vector_store.COMPACT_BATCH_SIZE", 3)
-        store.add([make_node(f"n{i}", 1, seed=float(i)) for i in range(10)])
+        store.add([make_node(f"n{i}", "1", seed=float(i)) for i in range(10)])
         store.compact(force=True)
-        ids = {n.node_id for n in store.get_nodes(filters=_in_filter([1]))}
+        ids = {n.node_id for n in store.get_nodes(filters=_in_filter(["1"]))}
         assert ids == {f"n{i}" for i in range(10)}
         assert self._bloat_ratio(store) == pytest.approx(1.0)
 
-    def test_force_compact_preserves_modified_times(
-        self,
-        store: PaperlessSqliteVecVectorStore,
-    ) -> None:
-        """
-        GIVEN:
-            - A store with documents whose modified times are tracked
-        WHEN:
-            - compact(force=True) rebuilds the database file
-        THEN:
-            - get_modified_times() still returns every document's value
-              (document_meta must be copied across the file-swap, not just
-              the vec0 rows)
-        """
-        store.add(
-            [
-                make_node("a1", 1, modified="2026-01-01T00:00:00"),
-                make_node("b1", 2, modified="2026-02-02T00:00:00"),
-            ],
-        )
-        before = store.get_modified_times()
-        store.compact(force=True)
-        assert store.get_modified_times() == before
-
-    def test_compact_on_unmigrated_store_is_noop(
-        self,
-        store: PaperlessSqliteVecVectorStore,
-        mocker: MockerFixture,
-    ) -> None:
-        """
-        GIVEN:
-            - A store whose schema_version has been forced behind
-              SCHEMA_VERSION (has_pending_migration() is True)
-        WHEN:
-            - compact(force=True) is called directly, without the caller
-              having run check_and_run_migrations() first
-        THEN:
-            - compact() is a safe no-op: no file-swap rebuild is attempted
-              at all (asserted via a spy on _rebuild_into, since schema_version
-              alone is not a reliable signal -- a rebuild would otherwise
-              copy the stale schema_version across unchanged, making a
-              before/after equality check pass even when a rebuild *did*
-              happen). Rebuilding an unmigrated store would silently lose
-              document_meta and leave the swapped-in file claiming the old
-              schema_version -- see the class docstring rationale.
-        """
-        store.add([make_node("a1", 1)])
-        store.client.execute(
-            "UPDATE index_meta SET value = '0' WHERE key = 'schema_version'",
-        )
-        assert store.has_pending_migration() is True
-        rebuild_spy = mocker.spy(PaperlessSqliteVecVectorStore, "_rebuild_into")
-        store.compact(force=True)
-        rebuild_spy.assert_not_called()
-        row = store.client.execute(
-            "SELECT value FROM index_meta WHERE key = 'schema_version'",
-        ).fetchone()
-        assert int(row["value"]) == 0
-
 
 class TestDbFile:
     def test_single_db_file_in_index_dir(self, store, tmp_path: Path) -> None:
-        store.add([make_node("a1", 1)])
+        store.add([make_node("a1", "1")])
         assert (tmp_path / DB_FILENAME).exists()
 
     def test_wal_mode_enabled(self, store) -> None:
@@ -587,26 +448,7 @@ class TestDbFile:
 
 
 class TestMigrations:
-    """Tests for the schema migration machinery.
-
-    These tests exercise check_and_run_migrations()'s generic dispatch logic
-    (structural vs. re-embed, version-boundary stopping) using ad hoc test
-    migrations layered on top of SCHEMA_VERSION -- distinct from
-    TestV1ToV2Migration, which exercises the real, frozen m0001_v1_to_v2
-    migration. Test migrations use version numbers starting at
-    SCHEMA_VERSION (2) and above so they never collide with the real
-    from_version=1/to_version=2 migration already registered in MIGRATIONS.
-
-    The fake structural migrations' apply() fixtures (see
-    test_structural_migration_copies_rows_and_updates_version and
-    test_stop_at_reembed_boundary below) only populate the rebuilt vec0
-    table itself -- they never insert into document_chunks/document_meta on
-    the destination connection. That's fine here: these tests exist to
-    verify the generic dispatch mechanism (version bookkeeping, structural-
-    vs-reembed branching), not full schema correctness of a rebuilt store;
-    the real migration's data completeness is covered separately by
-    TestV1ToV2Migration.
-    """
+    """Tests for the schema migration machinery."""
 
     def _schema_version(self, store: PaperlessSqliteVecVectorStore) -> int | None:
         row = store.client.execute(
@@ -615,21 +457,21 @@ class TestMigrations:
         return int(row[0]) if row else None
 
     def test_new_table_records_schema_version(self, store) -> None:
-        store.add([make_node("a1", 1)])
+        store.add([make_node("a1", "1")])
         assert self._schema_version(store) == SCHEMA_VERSION
 
     def test_check_migrations_no_table_returns_false(self, store) -> None:
         assert store.check_and_run_migrations() is False
 
     def test_check_migrations_current_version_returns_false(self, store) -> None:
-        store.add([make_node("a1", 1)])
+        store.add([make_node("a1", "1")])
         assert store.check_and_run_migrations() is False
 
     def test_reembed_migration_returns_true(self, store, tmp_path: Path) -> None:
-        store.add([make_node("a1", 1)])
+        store.add([make_node("a1", "1")])
         migration = Migration(
-            from_version=SCHEMA_VERSION,
-            to_version=SCHEMA_VERSION + 1,
+            from_version=1,
+            to_version=2,
             kind="re-embed",
             description="test re-embed",
         )
@@ -638,7 +480,7 @@ class TestMigrations:
             from paperless_ai import vector_store as vs_mod
 
             original = vs_mod.SCHEMA_VERSION
-            vs_mod.SCHEMA_VERSION = SCHEMA_VERSION + 1
+            vs_mod.SCHEMA_VERSION = 2
             result = store.check_and_run_migrations()
         finally:
             MIGRATIONS.remove(migration)
@@ -650,7 +492,7 @@ class TestMigrations:
         store,
         tmp_path: Path,
     ) -> None:
-        store.add([make_node("a1", 1), make_node("b1", 2)])
+        store.add([make_node("a1", "1"), make_node("b1", "2")])
 
         def apply(
             src: sqlite3.Connection,
@@ -669,7 +511,7 @@ class TestMigrations:
                 (str(dim),),
             )
             rows = src.execute(
-                "SELECT id, document_id, node_content, embedding "
+                "SELECT id, document_id, modified, node_content, embedding "
                 f"FROM {DEFAULT_TABLE_NAME}",
             ).fetchall()
             dst.execute("BEGIN IMMEDIATE")
@@ -680,8 +522,8 @@ class TestMigrations:
                 [
                     (
                         r["id"],
-                        str(r["document_id"]),
-                        "",
+                        r["document_id"],
+                        r["modified"],
                         r["node_content"],
                         bytes(r["embedding"]),
                     )
@@ -696,8 +538,8 @@ class TestMigrations:
             dst.execute("COMMIT")
 
         migration = Migration(
-            from_version=SCHEMA_VERSION,
-            to_version=SCHEMA_VERSION + 1,
+            from_version=1,
+            to_version=2,
             kind="structural",
             description="test structural",
             apply=apply,
@@ -707,29 +549,28 @@ class TestMigrations:
             from paperless_ai import vector_store as vs_mod
 
             original = vs_mod.SCHEMA_VERSION
-            vs_mod.SCHEMA_VERSION = SCHEMA_VERSION + 1
+            vs_mod.SCHEMA_VERSION = 2
             result = store.check_and_run_migrations()
         finally:
             MIGRATIONS.remove(migration)
             vs_mod.SCHEMA_VERSION = original
 
         assert result is False
-        assert self._schema_version(store) == SCHEMA_VERSION + 1
+        assert self._schema_version(store) == 2
         ids = {n.node_id for n in store.get_nodes()}
         assert ids == {"a1", "b1"}
 
     def test_compact_preserves_schema_version(self, store) -> None:
-        store.add([make_node("a1", 1)])
+        store.add([make_node("a1", "1")])
         assert self._schema_version(store) == SCHEMA_VERSION
         store.compact(force=True)
         assert self._schema_version(store) == SCHEMA_VERSION
 
     def test_stop_at_reembed_boundary(self, store) -> None:
-        # Registry: structural v(N+1), re-embed v(N+2), structural v(N+3),
-        # where N = SCHEMA_VERSION. Only v(N+1) should apply; the re-embed
-        # boundary must stop execution before v(N+3) runs, and the stored
-        # version must stay at N+1.
-        store.add([make_node("a1", 1), make_node("b1", 2)])
+        # Registry: structural v2, re-embed v3, structural v4.
+        # Only v2 should apply; the re-embed boundary must stop execution
+        # before v4 runs, and the stored version must stay at 2.
+        store.add([make_node("a1", "1"), make_node("b1", "2")])
 
         def copy_apply(
             src: sqlite3.Connection,
@@ -748,7 +589,7 @@ class TestMigrations:
                 (str(dim),),
             )
             rows = src.execute(
-                "SELECT id, document_id, node_content, embedding "
+                "SELECT id, document_id, modified, node_content, embedding "
                 f"FROM {DEFAULT_TABLE_NAME}",
             ).fetchall()
             dst.execute("BEGIN IMMEDIATE")
@@ -759,8 +600,8 @@ class TestMigrations:
                 [
                     (
                         r["id"],
-                        str(r["document_id"]),
-                        "",
+                        r["document_id"],
+                        r["modified"],
                         r["node_content"],
                         bytes(r["embedding"]),
                     )
@@ -771,23 +612,23 @@ class TestMigrations:
 
         migrations = [
             Migration(
-                from_version=SCHEMA_VERSION,
-                to_version=SCHEMA_VERSION + 1,
+                from_version=1,
+                to_version=2,
                 kind="structural",
-                description="v(N+1) structural",
+                description="v2 structural",
                 apply=copy_apply,
             ),
             Migration(
-                from_version=SCHEMA_VERSION + 1,
-                to_version=SCHEMA_VERSION + 2,
+                from_version=2,
+                to_version=3,
                 kind="re-embed",
-                description="v(N+2) re-embed boundary",
+                description="v3 re-embed boundary",
             ),
             Migration(
-                from_version=SCHEMA_VERSION + 2,
-                to_version=SCHEMA_VERSION + 3,
+                from_version=3,
+                to_version=4,
                 kind="structural",
-                description="v(N+3) structural - must not run",
+                description="v4 structural - must not run",
                 apply=copy_apply,
             ),
         ]
@@ -796,7 +637,7 @@ class TestMigrations:
             from paperless_ai import vector_store as vs_mod
 
             original = vs_mod.SCHEMA_VERSION
-            vs_mod.SCHEMA_VERSION = SCHEMA_VERSION + 3
+            vs_mod.SCHEMA_VERSION = 4
             result = store.check_and_run_migrations()
         finally:
             for m in migrations:
@@ -804,226 +645,4 @@ class TestMigrations:
             vs_mod.SCHEMA_VERSION = original
 
         assert result is True
-        assert self._schema_version(store) == SCHEMA_VERSION + 1
-
-    def test_has_pending_migration_false_when_no_table(
-        self,
-        store: PaperlessSqliteVecVectorStore,
-    ) -> None:
-        """
-        GIVEN:
-            - A vector store with no table created yet
-        WHEN:
-            - has_pending_migration() is checked
-        THEN:
-            - False is returned (nothing to migrate before anything exists)
-        """
-        assert store.has_pending_migration() is False
-
-    def test_has_pending_migration_false_at_current_version(
-        self,
-        store: PaperlessSqliteVecVectorStore,
-    ) -> None:
-        """
-        GIVEN:
-            - A store at the current SCHEMA_VERSION
-        WHEN:
-            - has_pending_migration() is checked
-        THEN:
-            - False is returned
-        """
-        store.add([make_node("a1", 1)])
-        assert store.has_pending_migration() is False
-
-    def test_has_pending_migration_true_when_behind(
-        self,
-        store: PaperlessSqliteVecVectorStore,
-    ) -> None:
-        """
-        GIVEN:
-            - A store whose schema_version has been forced behind SCHEMA_VERSION
-        WHEN:
-            - has_pending_migration() is checked
-        THEN:
-            - True is returned
-        """
-        store.add([make_node("a1", 1)])
-        store.client.execute(
-            "UPDATE index_meta SET value = '0' WHERE key = 'schema_version'",
-        )
-        assert store.has_pending_migration() is True
-
-
-class TestV1ToV2Migration:
-    """m0001_v1_to_v2 migrates a real, historically-shaped v1 store. The
-    fixture below is a literal, hardcoded v1 DDL string -- NOT derived from
-    any current code -- so this test keeps testing the actual historical
-    shape even if vector_store.py's "current" schema changes again later.
-    """
-
-    def _build_v1_store(self, db_path: str, dim: int) -> None:
-        conn = sqlite3.connect(db_path)
-        conn.row_factory = sqlite3.Row
-        conn.enable_load_extension(True)  # noqa: FBT003
-        sqlite_vec.load(conn)
-        conn.enable_load_extension(False)  # noqa: FBT003
-        conn.execute("PRAGMA journal_mode=WAL")
-        conn.execute("PRAGMA synchronous=NORMAL")
-        conn.execute(
-            "CREATE TABLE IF NOT EXISTS index_meta (key TEXT PRIMARY KEY, value TEXT)",
-        )
-        conn.execute(  # nosemgrep
-            "CREATE VIRTUAL TABLE documents USING vec0("
-            "id TEXT PRIMARY KEY, document_id TEXT, modified TEXT,"
-            f" +node_content TEXT, embedding float[{dim}] distance_metric=cosine"
-            ")",
-        )
-        conn.execute(
-            "INSERT INTO index_meta (key, value) VALUES ('dim', ?)",
-            (str(dim),),
-        )
-        conn.execute(
-            "INSERT INTO index_meta (key, value) VALUES ('schema_version', '1')",
-        )
-        conn.execute(
-            "INSERT INTO index_meta (key, value) VALUES ('embed_model', 'model-a')",
-        )
-        rows = [
-            ("c1", "1", "2026-01-01T00:00:00", '{"text": "a"}', _pack([0.1] * dim)),
-            ("c2", "1", "2026-01-01T00:00:00", '{"text": "b"}', _pack([0.2] * dim)),
-            ("c3", "2", "2026-02-02T00:00:00", '{"text": "c"}', _pack([0.3] * dim)),
-        ]
-        conn.executemany(
-            "INSERT INTO documents (id, document_id, modified, node_content, embedding)"
-            " VALUES (?, ?, ?, ?, ?)",
-            rows,
-        )
-        conn.execute(
-            "INSERT INTO index_meta (key, value) VALUES ('total_inserts', '3')",
-        )
-        conn.commit()
-        conn.close()
-
-    def test_migration_converts_v1_store_to_v2(self, tmp_path: Path) -> None:
-        """
-        GIVEN:
-            - A real v1-shaped store (TEXT document_id, modified inline in
-              vec0, no document_chunks/document_meta) built from a literal,
-              hardcoded historical DDL
-        WHEN:
-            - A PaperlessSqliteVecVectorStore is opened against it
-        THEN:
-            - schema_version becomes 2, document_id values become int,
-              document_chunks/document_meta are backfilled once per chunk/
-              document respectively, and dim/embed_model survive
-        """
-        db_dir = tmp_path
-        self._build_v1_store(str(db_dir / DB_FILENAME), dim=16)
-        with PaperlessSqliteVecVectorStore(uri=str(db_dir)) as store:
-            assert store.check_and_run_migrations() is False
-            row = store.client.execute(
-                "SELECT value FROM index_meta WHERE key = 'schema_version'",
-            ).fetchone()
-            assert int(row["value"]) == 2
-            doc_id_row = store.client.execute(
-                "SELECT document_id FROM documents WHERE id = 'c1'",
-            ).fetchone()
-            assert isinstance(doc_id_row["document_id"], int)
-            assert doc_id_row["document_id"] == 1
-            chunk_ids = sorted(
-                r["chunk_id"]
-                for r in store.client.execute(
-                    "SELECT chunk_id FROM document_chunks",
-                ).fetchall()
-            )
-            assert chunk_ids == ["c1", "c2", "c3"]
-            assert store.get_modified_times() == {
-                "1": "2026-01-01T00:00:00",
-                "2": "2026-02-02T00:00:00",
-            }
-            assert store.stored_model_name() == "model-a"
-            assert store.vector_dim() == 16
-
-    def test_migration_raises_on_malformed_document_id(
-        self,
-        tmp_path: Path,
-    ) -> None:
-        """
-        GIVEN:
-            - A v1-shaped store with a corrupted, non-integer document_id
-              value on one row
-        WHEN:
-            - The migration runs
-        THEN:
-            - A ValueError is raised (fail loudly, no silent data loss) --
-              this matches the rest of vector_store.py, which has no
-              precedent for silently skipping malformed rows
-        """
-        db_dir = tmp_path
-        self._build_v1_store(str(db_dir / DB_FILENAME), dim=16)
-        conn = sqlite3.connect(str(db_dir / DB_FILENAME))
-        conn.enable_load_extension(True)  # noqa: FBT003
-        sqlite_vec.load(conn)
-        conn.enable_load_extension(False)  # noqa: FBT003
-        conn.execute(
-            "UPDATE documents SET document_id = 'not-an-int' WHERE id = 'c1'",
-        )
-        conn.commit()
-        conn.close()
-        with (
-            pytest.raises(ValueError),
-            PaperlessSqliteVecVectorStore(uri=str(db_dir)) as store,
-        ):
-            store.check_and_run_migrations()
-
-    def test_migration_never_delegates_to_current_schema_helpers(
-        self,
-        tmp_path: Path,
-        mocker: MockerFixture,
-    ) -> None:
-        """
-        GIVEN:
-            - A real v1-shaped store
-        WHEN:
-            - The migration runs, with DocumentChunksTable.create/
-              DocumentMetaTable.create/_create_vec_table spied on
-        THEN:
-            - None of those "current schema" helpers are ever called during
-              the migration -- it must freeze its own historical DDL, per
-              the DDL-freezing rule (see spec), so a future schema bump
-              can't silently corrupt this migration's output
-        """
-        db_dir = tmp_path
-        self._build_v1_store(str(db_dir / DB_FILENAME), dim=16)
-        chunks_create_spy = mocker.spy(DocumentChunksTable, "create")
-        meta_create_spy = mocker.spy(DocumentMetaTable, "create")
-        vec_table_spy = mocker.spy(
-            PaperlessSqliteVecVectorStore,
-            "_create_vec_table",
-        )
-        with PaperlessSqliteVecVectorStore(uri=str(db_dir)) as store:
-            store.check_and_run_migrations()
-        # _open_connection() legitimately calls create() three times across
-        # a structural migration: once for the store's own live connection
-        # (construction), once for the migration's temp rebuild file
-        # (_rebuild_file), and once more when _swap_in_compact() reopens the
-        # swapped-in file as self._conn. What matters is that
-        # m0001_v1_to_v2's apply() itself never calls these directly, which
-        # a source-text check alone can't prove (it can't tell "mentioned in
-        # a comment" from "actually called", and is trivially evadable by
-        # importing a symbol under an alias). Asserting the exact call count
-        # instead: exactly 3 calls to each create() (all from
-        # _open_connection, never a 4th from inside apply()), and zero calls
-        # to _create_vec_table (neither _open_connection nor apply() calls
-        # it -- apply() freezes its own literal CREATE VIRTUAL TABLE DDL
-        # instead).
-        assert chunks_create_spy.call_count == 3
-        assert meta_create_spy.call_count == 3
-        assert vec_table_spy.call_count == 0
-        # Cheap secondary signal, kept alongside the spy assertions above
-        # (not in place of them): the migration module's source should never
-        # even mention these "current schema" helpers by name.
-        source = inspect.getsource(m0001_v1_to_v2)
-        assert "DocumentChunksTable.create" not in source
-        assert "DocumentMetaTable.create" not in source
-        assert "_create_vec_table(" not in source
+        assert self._schema_version(store) == 2

src/paperless_ai/vector_store.py

@@ -2,13 +2,16 @@ import json
 import logging
 import sqlite3
 import struct
+from collections.abc import Callable
 from collections.abc import Iterator
 from collections.abc import Sequence
 from contextlib import contextmanager
+from dataclasses import dataclass
+from dataclasses import field
 from pathlib import Path
 from types import TracebackType
 from typing import Any
-from typing import NamedTuple
+from typing import Literal
 
 import sqlite_vec
 from llama_index.core.bridge.pydantic import PrivateAttr
@@ -23,14 +26,6 @@ from llama_index.core.vector_stores.types import VectorStoreQueryResult
 from llama_index.core.vector_stores.utils import metadata_dict_to_node
 from llama_index.core.vector_stores.utils import node_to_metadata_dict
 
-from paperless_ai.migrations import MIGRATIONS
-from paperless_ai.migrations import Migration
-from paperless_ai.tables import ChunkRow
-from paperless_ai.tables import DocumentChunksTable
-from paperless_ai.tables import DocumentMetaRow
-from paperless_ai.tables import DocumentMetaTable
-from paperless_ai.tables import IndexMetaTable
-
 logger = logging.getLogger("paperless_ai.vector_store")
 
 DB_FILENAME = "llmindex.db"
@@ -39,7 +34,7 @@ DEFAULT_TABLE_NAME = "documents"
 # Current schema version. Written to index_meta at table creation and bumped
 # whenever a Migration is added to MIGRATIONS. check_and_run_migrations() uses
 # this to decide which migrations to run on an existing store.
-SCHEMA_VERSION = 2
+SCHEMA_VERSION = 1
 
 # compact(): rebuild when the cumulative rowid count exceeds this multiple of
 # the live row count. DELETEs on vec0 tables never reclaim space (upstream
@@ -54,23 +49,40 @@ COMPACT_BATCH_SIZE = 500
 
 # Filterable vec0 metadata columns. _build_where() only ever receives filter
 # keys we construct ourselves, but allowlisting keeps SQL identifiers safe by
-# construction. "modified" is not here: it is never filtered on, and as of
-# schema v2 it isn't even a vec0 column anymore (see document_meta).
-_FILTER_COLUMNS = frozenset({"document_id"})
+# construction.
+_FILTER_COLUMNS = frozenset({"document_id", "modified"})
 
 
-class _Row(NamedTuple):
-    """One node, ready to write. ``modified`` is not a vec0 column (see
-    document_meta) -- it rides along here because every row-producing call
-    site needs both the vec0 insert values and the document_meta upsert
-    value from the same node.
+@dataclass
+class Migration:
+    """A schema migration for the sqlite-vec vector store.
+
+    kind="structural": rows are copied into a new-schema file with no
+    re-embedding needed.  Supply ``apply(src_conn, dst_conn, dim)`` which
+    must create the vec0 table in ``dst_conn``, copy all rows from
+    ``src_conn``, and write ``dim`` / ``embed_model`` / ``total_inserts`` to
+    ``dst_conn``'s ``index_meta``.  ``schema_version`` is written by the
+    migration runner after ``apply`` returns.
+
+    kind="re-embed": the new schema requires fresh embeddings.
+    ``check_and_run_migrations()`` returns True when it encounters one of
+    these so the caller can force a full rebuild (which recreates the table
+    at the current SCHEMA_VERSION).
     """
 
-    chunk_id: str
-    document_id: int
-    modified: str
-    node_content: str
-    embedding: bytes
+    from_version: int
+    to_version: int
+    kind: Literal["structural", "re-embed"]
+    description: str
+    apply: Callable[[sqlite3.Connection, sqlite3.Connection, int], None] | None = field(
+        default=None,
+        repr=False,
+    )
+
+
+# Registry of all schema migrations in order. Empty at v1 -- this is the
+# baseline. Add entries here (and bump SCHEMA_VERSION) when the schema changes.
+MIGRATIONS: list[Migration] = []
 
 
 def _pack(embedding: Sequence[float]) -> bytes:
@@ -81,30 +93,14 @@ def _unpack(blob: bytes) -> list[float]:
     return list(struct.unpack(f"{len(blob) // 4}f", blob))
 
 
-_INSERT = (
-    "INSERT INTO "
-    + DEFAULT_TABLE_NAME
-    + " (id, document_id, node_content, embedding) VALUES (?, ?, ?, ?)"
-)
-
-
-def _vec0_params(rows: list[_Row]) -> list[tuple[str, int, str, bytes]]:
-    """``rows``, minus the ``modified`` field vec0 no longer stores."""
-    return [(r.chunk_id, r.document_id, r.node_content, r.embedding) for r in rows]
-
-
-def _build_where(filters: MetadataFilters | None) -> tuple[str, list[int]]:
-    """Translate the EQ / IN / NE filters we use into a parameterized SQL
-    clause on vec0 metadata columns. Returns ("", []) when there is nothing
-    to filter. document_id is vec0's only filterable column and is INTEGER;
-    every value is coerced via int() here so callers (which today still pass
-    strings in places, e.g. indexing.py's MetadataFilter construction) don't
-    have to be individually correct -- vec0 doesn't coerce types itself.
+def _build_where(filters: MetadataFilters | None) -> tuple[str, list[str]]:
+    """Translate the EQ / IN / NE filters we use into a parameterized SQL clause
+    on vec0 metadata columns. Returns ("", []) when there is nothing to filter.
     """
     if filters is None or not filters.filters:
         return "", []
     clauses: list[str] = []
-    params: list[int] = []
+    params: list[str] = []
     for f in filters.filters:
         # filters.filters is Union[MetadataFilter, ExactMatchFilter, MetadataFilters];
         # we only build MetadataFilter entries, so skip anything else at runtime.
@@ -113,7 +109,7 @@ def _build_where(filters: MetadataFilters | None) -> tuple[str, list[int]]:
         if f.key not in _FILTER_COLUMNS:  # pragma: no cover - we build the keys
             raise NotImplementedError(f"Unsupported filter column: {f.key}")
         if f.operator == FilterOperator.IN:
-            values = [int(v) for v in f.value]  # type: ignore[union-attr]
+            values = [str(v) for v in f.value]  # type: ignore[union-attr]  # value is list when operator is IN
             if not values:  # pragma: no cover
                 clauses.append("1 = 0")
                 continue
@@ -122,10 +118,10 @@ def _build_where(filters: MetadataFilters | None) -> tuple[str, list[int]]:
             params.extend(values)
         elif f.operator == FilterOperator.EQ:
             clauses.append(f"{f.key} = ?")
-            params.append(int(f.value))
+            params.append(str(f.value))
         elif f.operator == FilterOperator.NE:
             clauses.append(f"{f.key} != ?")
-            params.append(int(f.value))
+            params.append(str(f.value))
         else:  # pragma: no cover - we only ever build EQ/IN/NE filters
             raise NotImplementedError(f"Unsupported filter operator: {f.operator}")
     if not clauses:
@@ -190,21 +186,9 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
         conn.enable_load_extension(False)  # noqa: FBT003
         conn.execute("PRAGMA journal_mode=WAL")
         conn.execute("PRAGMA synchronous=NORMAL")
-        IndexMetaTable.create(conn)
-        # vec0 metadata columns only get an efficient lookup path inside a
-        # KNN (MATCH) query; a plain `WHERE document_id = ?` is a full table
-        # scan regardless of index size. This plain, indexed table is how
-        # delete()/upsert_document() find a document's chunk ids without
-        # that scan.
-        DocumentChunksTable.create(conn)
-        # modified used to be a vec0 metadata column, but vec0 only inlines
-        # TEXT metadata up to 12 bytes -- an ISO timestamp is always longer,
-        # so every read recompiled and stepped a fresh SQL statement per row.
-        # It was never filtered on inside a KNN query either, so it never
-        # needed to be a vec0 column at all. One row per document here (not
-        # per chunk, like document_chunks), since every chunk of a document
-        # shares the same modified value -- see get_modified_times().
-        DocumentMetaTable.create(conn)
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS index_meta (key TEXT PRIMARY KEY, value TEXT)",
+        )
         return conn
 
     @property
@@ -239,6 +223,24 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
         else:
             self._conn.execute("COMMIT")
 
+    def _meta_get(self, key: str) -> str | None:
+        row = self._conn.execute(
+            "SELECT value FROM index_meta WHERE key = ?",
+            (key,),
+        ).fetchone()
+        return row["value"] if row else None
+
+    @staticmethod
+    def _meta_set_on(conn: sqlite3.Connection, key: str, value: str) -> None:
+        conn.execute(
+            "INSERT INTO index_meta (key, value) VALUES (?, ?) "
+            "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
+            (key, value),
+        )
+
+    def _meta_set(self, key: str, value: str) -> None:
+        self._meta_set_on(self._conn, key, value)
+
     def table_exists(self) -> bool:
         return (
             self._conn.execute(
@@ -251,19 +253,18 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
     def vector_dim(self) -> int | None:
         if not self.table_exists():
             return None
-        return IndexMetaTable.get_dim(self._conn)
+        value = self._meta_get("dim")
+        return int(value) if value else None
 
     def drop_table(self) -> None:
         self._conn.execute("DROP TABLE IF EXISTS " + DEFAULT_TABLE_NAME)
         self._conn.execute("DELETE FROM index_meta")
-        DocumentChunksTable.delete_all(self._conn)
-        DocumentMetaTable.delete_all(self._conn)
 
     def stored_model_name(self) -> str | None:
         """Return the embedding model name recorded at table creation, or None."""
         if not self.table_exists():
             return None
-        return IndexMetaTable.get_embed_model(self._conn)
+        return self._meta_get("embed_model")
 
     def config_mismatch(self, model_name: str) -> bool:
         """True when the stored model name differs from ``model_name``.
@@ -281,17 +282,14 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
         # document_id is deliberately a metadata column, NOT a partition key:
         # partition keys change KNN `k` to per-partition semantics under IN
         # filters (asg017/sqlite-vec#142); metadata columns give a correct
-        # global top-k. INTEGER (not TEXT, as in schema v1): EQ/NE/IN
-        # comparisons become a native i64 array compare instead of per-row
-        # strncmp against a 16-byte text view, and this drops the unused
-        # metadatatext shadow table TEXT columns carry. modified is not a
-        # column here at all as of v2 -- see document_meta.
+        # global top-k.
         conn.execute(  # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query
             "CREATE VIRTUAL TABLE "
             + DEFAULT_TABLE_NAME
             + " USING vec0("
             + "id TEXT PRIMARY KEY,"
-            + " document_id INTEGER,"
+            + " document_id TEXT,"
+            + " modified TEXT,"
             + " +node_content TEXT,"
             + " embedding float["
             + str(int(dim))
@@ -301,78 +299,37 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
 
     def _create_table(self, dim: int) -> None:
         self._create_vec_table(self._conn, dim)
-        IndexMetaTable.set_dim(self._conn, dim)
-        IndexMetaTable.set_schema_version(self._conn, SCHEMA_VERSION)
+        self._meta_set("dim", str(dim))
+        self._meta_set("schema_version", str(SCHEMA_VERSION))
         if self._embed_model_name:
-            IndexMetaTable.set_embed_model(self._conn, self._embed_model_name)
+            self._meta_set("embed_model", self._embed_model_name)
 
-    def _ensure_table(self, dim: int, *, table_exists: bool) -> None:
-        if not table_exists:
+    def _ensure_table(self, dim: int) -> None:
+        if not self.table_exists():
             self._create_table(dim)
 
-    def _row(self, node: BaseNode) -> _Row:
+    def _row(self, node: BaseNode) -> tuple[str, str, str, str, bytes]:
         meta = node_to_metadata_dict(
             node,
             remove_text=False,
             flat_metadata=self.flat_metadata,
         )
+        # vec0 metadata columns reject NULL (asg017/sqlite-vec#141): coerce
+        # every value to a string, with "" as the absent sentinel.
         document_id = node.ref_doc_id or node.metadata.get("document_id")
-        return _Row(
-            chunk_id=node.node_id,
-            # document_id is required -- int(None) raises TypeError and
-            # int("not-a-number") raises ValueError, both intentional:
-            # fail loudly on a malformed/missing document_id rather than
-            # silently indexing a chunk with no owning document. modified,
-            # below, still uses the str(x or "") sentinel pattern because a
-            # missing modified value is legitimate (vec0 no longer even
-            # stores it -- see document_meta), whereas document_id must
-            # always be present.
-            document_id=int(document_id),
-            modified=str(node.metadata.get("modified") or ""),
-            node_content=json.dumps(meta),
-            embedding=_pack(node.get_embedding()),
+        return (
+            node.node_id,
+            str(document_id or ""),
+            str(node.metadata.get("modified") or ""),
+            json.dumps(meta),
+            _pack(node.get_embedding()),
         )
 
-    def _index_chunks(self, rows: list[_Row]) -> None:
-        """Record each row's (chunk_id, document_id) in document_chunks, and
-        each row's (document_id, modified) in document_meta -- deduped
-        within the batch, since every chunk of a document shares the same
-        modified value -- kept in lockstep with every insert into the vec0
-        table.
-        """
-        DocumentChunksTable.insert_many(
-            self._conn,
-            (ChunkRow(r.chunk_id, r.document_id) for r in rows),
-        )
-        modified_by_document = {r.document_id: r.modified for r in rows}
-        DocumentMetaTable.upsert_many(
-            self._conn,
-            (
-                DocumentMetaRow(doc_id, mod)
-                for doc_id, mod in modified_by_document.items()
-            ),
-        )
-
-    def _delete_chunks_by_document_id(self, document_id: int) -> None:
-        """Delete all of a document's chunks via point-deletes on `id`.
-
-        vec0 has no efficient lookup on the document_id metadata column
-        outside a KNN query, so a plain `DELETE ... WHERE document_id = ?`
-        is a full table scan regardless of index size. Looking the chunk
-        ids up in document_chunks first (a real indexed lookup) and
-        deleting each by its `id` primary key instead turns that scan into
-        a handful of O(1) point deletes.
-        """
-        chunk_ids = DocumentChunksTable.chunk_ids_for_document(
-            self._conn,
-            document_id,
-        )
-        self._conn.executemany(
-            "DELETE FROM " + DEFAULT_TABLE_NAME + " WHERE id = ?",
-            [(chunk_id,) for chunk_id in chunk_ids],
-        )
-        DocumentChunksTable.delete_for_document(self._conn, document_id)
-        DocumentMetaTable.delete_for_document(self._conn, document_id)
+    _INSERT = (
+        "INSERT INTO "
+        + DEFAULT_TABLE_NAME
+        + " (id, document_id, modified, node_content, embedding) VALUES (?, ?, ?, ?, ?)"
+    )
 
     def _increment_total_inserts(self, count: int) -> None:
         """Increment the cumulative insert counter stored in index_meta.
@@ -382,56 +339,48 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
         live_rows exceeds COMPACT_BLOAT_RATIO the table has accumulated
         enough deleted-but-not-freed rows to warrant a rebuild.
         """
-        IndexMetaTable.increment_total_inserts(self._conn, count)
+        current = int(self._meta_get("total_inserts") or "0")
+        self._meta_set("total_inserts", str(current + count))
 
     def add(self, nodes: Sequence[BaseNode], **add_kwargs: Any) -> list[str]:
         if not nodes:
             return []
         rows = [self._row(node) for node in nodes]
         with self._transaction():
-            self._ensure_table(
-                len(nodes[0].get_embedding()),
-                table_exists=self.table_exists(),
-            )
-            self._conn.executemany(_INSERT, _vec0_params(rows))
-            self._index_chunks(rows)
+            self._ensure_table(len(nodes[0].get_embedding()))
+            self._conn.executemany(self._INSERT, rows)
             self._increment_total_inserts(len(rows))
         return [node.node_id for node in nodes]
 
-    def upsert_document(
-        self,
-        document_id: int | str,
-        nodes: list[BaseNode],
-    ) -> list[str]:
+    def upsert_document(self, document_id: str, nodes: list[BaseNode]) -> list[str]:
         """Atomically replace all stored chunks of ``document_id`` with ``nodes``.
 
         One transaction deletes the document's existing rows and inserts the
-        new set (vec0's INSERT OR REPLACE is broken upstream, so delete+insert
-        it is). WAL readers in other processes see either the old or the new
-        chunk set, never a partial state.
+        new set (vec0's INSERT OR REPLACE is broken upstream, #259, so
+        delete+insert it is). WAL readers in other processes see either the
+        old or the new chunk set, never a partial state.
         """
-        doc_id = int(document_id)
         rows = [self._row(node) for node in nodes]
         with self._transaction():
-            table_exists = self.table_exists()
-            if nodes and not table_exists:
-                self._ensure_table(
-                    len(nodes[0].get_embedding()),
-                    table_exists=False,
+            if nodes:
+                self._ensure_table(len(nodes[0].get_embedding()))
+            if self.table_exists():
+                self._conn.execute(
+                    "DELETE FROM " + DEFAULT_TABLE_NAME + " WHERE document_id = ?",
+                    (str(document_id),),
                 )
-                table_exists = True
-            if table_exists:
-                self._delete_chunks_by_document_id(doc_id)
             if rows:
-                self._conn.executemany(_INSERT, _vec0_params(rows))
-                self._index_chunks(rows)
+                self._conn.executemany(self._INSERT, rows)
                 self._increment_total_inserts(len(rows))
         return [node.node_id for node in nodes]
 
-    def delete(self, ref_doc_id: int | str, **delete_kwargs: Any) -> None:
+    def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
         if self.table_exists():
             with self._transaction():
-                self._delete_chunks_by_document_id(int(ref_doc_id))
+                self._conn.execute(
+                    "DELETE FROM " + DEFAULT_TABLE_NAME + " WHERE document_id = ?",
+                    (str(ref_doc_id),),
+                )
 
     def _rows_to_nodes(self, rows: list[sqlite3.Row]) -> list[BaseNode]:
         nodes: list[BaseNode] = []
@@ -501,66 +450,41 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
     def get_modified_times(self) -> dict[str, str]:
         """Return {document_id: stored_modified_isoformat} for all indexed documents.
 
-        document_meta already has exactly one row per document (not per
-        chunk, unlike the vec0 table), so no dedup is needed here.
+        All chunks of a document share the same ``modified`` value, so the
+        first row seen per document is sufficient.
         """
         if not self.table_exists():
             return {}
-        return DocumentMetaTable.all_modified_times(self._conn)
-
-    @property
-    def _db_path(self) -> str:
-        return str(Path(self._uri) / DB_FILENAME)
-
-    @contextmanager
-    def _rebuild_file(self) -> Iterator[sqlite3.Connection]:
-        """Open a fresh temp database file for a file-swap rebuild (compact
-        or structural migration), yielding its connection for the caller to
-        populate.
-
-        On success, swaps the temp file in as the live database (closing
-        this store's current connection first -- see _swap_in_compact()).
-        On any exception, discards the temp file, including its -wal/-shm,
-        instead, and this store's own connection is left untouched.
-        """
-        compact_path = self._db_path + ".compact"
-        new_conn = self._open_connection(compact_path)
-        try:
-            yield new_conn
-        except BaseException:
-            new_conn.close()
-            for suffix in ["", "-wal", "-shm"]:
-                Path(compact_path + suffix).unlink(missing_ok=True)
-            raise
-        else:
-            new_conn.close()
-            self._swap_in_compact(compact_path, self._db_path)
+        result: dict[str, str] = {}
+        for row in self._conn.execute(
+            "SELECT document_id, modified FROM " + DEFAULT_TABLE_NAME,
+        ):
+            doc_id = str(row["document_id"])
+            if doc_id not in result:
+                result[doc_id] = str(row["modified"] or "")
+        return result
 
     def compact(self, *, force: bool = False) -> None:
         """Rebuild the database file to reclaim space left behind by DELETEs.
 
         vec0 DELETE only invalidates rows; the vector data stays in the file
-        forever, and per-document re-indexing is a delete+insert. The
-        cumulative insert counter in ``index_meta`` tracks total rows ever
-        written; when that exceeds ``COMPACT_BLOAT_RATIO`` x the live row
-        count (or when forced), live rows are copied into a fresh database
-        file and swapped in via ``os.replace``.
+        forever (asg017/sqlite-vec#54), and per-document re-indexing is a
+        delete+insert. The cumulative insert counter in ``index_meta`` tracks
+        total rows ever written; when that exceeds ``COMPACT_BLOAT_RATIO`` x
+        the live row count (or when forced), live rows are copied into a fresh
+        database file and swapped in via ``os.replace``.
 
         Note: ``ALTER TABLE ... RENAME TO`` on vec0 virtual tables does NOT
-        rename the shadow tables (sqlite-vec upstream limitation), so an
-        in-place rename-based rebuild is not safe. The file-swap approach is
-        the maintainer-endorsed workaround.
+        rename the shadow tables (sqlite-vec upstream limitation), so
+        an in-place rename-based rebuild is not safe.  The file-swap approach
+        is the maintainer-endorsed workaround (asg017/sqlite-vec#205).
         """
         if not self.table_exists():
             return
-        if self.has_pending_migration():
-            logger.warning(
-                "Skipping compact: store has a pending schema migration; "
-                "run check_and_run_migrations() first",
-            )
-            return
-        live = DocumentChunksTable.count(self._conn)
-        total = IndexMetaTable.get_total_inserts(self._conn) or live
+        live = self._conn.execute(
+            "SELECT count(*) FROM " + DEFAULT_TABLE_NAME,
+        ).fetchone()[0]
+        total = int(self._meta_get("total_inserts") or str(live))
         if not force and total <= max(live, 1) * COMPACT_BLOAT_RATIO:
             return
         dim = self.vector_dim()
@@ -572,61 +496,50 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
             live,
             total,
         )
-        with self._rebuild_file() as new_conn:
-            self._rebuild_into(self._conn, new_conn, dim)
+        db_path = str(Path(self._uri) / DB_FILENAME)
+        compact_path = db_path + ".compact"
 
-    @staticmethod
-    def _rebuild_into(
-        src_conn: sqlite3.Connection,
-        dst_conn: sqlite3.Connection,
-        dim: int,
-    ) -> None:
-        """Create the vec0 table in ``dst_conn``, copy dim/embed_model from
-        ``src_conn``, and stream every live vec0 row, document_chunks row,
-        and document_meta row across. Used by compact() only --
-        m0001_v1_to_v2 freezes its own copy loop instead of calling this,
-        since this always reflects the *current* schema (see the migration
-        DDL-freezing rule in the spec).
-        """
-        PaperlessSqliteVecVectorStore._create_vec_table(dst_conn, dim)
-        dim_value = IndexMetaTable.get_dim(src_conn)
-        if dim_value is not None:
-            IndexMetaTable.set_dim(dst_conn, dim_value)
-        embed_model = IndexMetaTable.get_embed_model(src_conn)
-        if embed_model is not None:
-            IndexMetaTable.set_embed_model(dst_conn, embed_model)
-        schema_version = IndexMetaTable.get_schema_version(src_conn)
-        if schema_version is not None:
-            IndexMetaTable.set_schema_version(dst_conn, schema_version)
-
-        dst_conn.execute("BEGIN IMMEDIATE")
-        src_cursor = src_conn.execute(
-            "SELECT id, document_id, node_content, embedding FROM "
-            + DEFAULT_TABLE_NAME,
-        )
-        copied = 0
-        while batch := src_cursor.fetchmany(COMPACT_BATCH_SIZE):
-            dst_conn.executemany(
-                _INSERT,
-                [
-                    (
-                        r["id"],
-                        r["document_id"],
-                        r["node_content"],
-                        bytes(r["embedding"]),
-                    )
-                    for r in batch
-                ],
+        # Copy all live rows into a fresh database file.
+        new_conn = self._open_connection(compact_path)
+        try:
+            self._create_vec_table(new_conn, dim)
+            self._meta_set_on(new_conn, "dim", str(dim))
+            for key in ("embed_model", "schema_version"):
+                value = self._meta_get(key)
+                if value is not None:
+                    self._meta_set_on(new_conn, key, value)
+            src_cursor = self._conn.execute(
+                "SELECT id, document_id, modified, node_content, embedding "
+                "FROM " + DEFAULT_TABLE_NAME,
             )
-            DocumentChunksTable.insert_many(
-                dst_conn,
-                (ChunkRow(r["id"], r["document_id"]) for r in batch),
-            )
-            copied += len(batch)
-        DocumentMetaTable.copy_all(src_conn, dst_conn, COMPACT_BATCH_SIZE)
-        # Reset the cumulative counter: after a rebuild, total_inserts == live.
-        IndexMetaTable.reset_total_inserts(dst_conn, copied)
-        dst_conn.execute("COMMIT")
+            new_conn.execute("BEGIN IMMEDIATE")
+            # Stream rows from the source cursor in batches instead of
+            # materializing the whole table in memory, so a large index does
+            # not cause an OOM during routine maintenance compactions.
+            while batch := src_cursor.fetchmany(COMPACT_BATCH_SIZE):
+                new_conn.executemany(
+                    self._INSERT,
+                    [
+                        (
+                            r["id"],
+                            r["document_id"],
+                            r["modified"],
+                            r["node_content"],
+                            bytes(r["embedding"]),
+                        )
+                        for r in batch
+                    ],
+                )
+            # Reset the cumulative counter: after compact, total_inserts == live.
+            self._meta_set_on(new_conn, "total_inserts", str(live))
+            new_conn.execute("COMMIT")
+        except BaseException:
+            new_conn.close()
+            for p in [compact_path, compact_path + "-wal", compact_path + "-shm"]:
+                Path(p).unlink(missing_ok=True)
+            raise
+        new_conn.close()
+        self._swap_in_compact(compact_path, db_path)
 
     def _swap_in_compact(self, compact_path: str, db_path: str) -> None:
         """Atomically replace the live database with the compacted copy."""
@@ -638,31 +551,6 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
         Path(compact_path).replace(db_path)
         self._conn = self._open_connection(db_path)
 
-    def _stored_schema_version(self) -> int | None:
-        """The schema_version recorded in index_meta, or None if no table
-        exists. A missing key (a store predating version tracking) is
-        treated as SCHEMA_VERSION -- i.e. already current -- since no
-        migration in MIGRATIONS targets a version before tracking began.
-        """
-        if not self.table_exists():
-            return None
-        raw_version = IndexMetaTable.get_schema_version(self._conn)
-        return raw_version if raw_version is not None else SCHEMA_VERSION
-
-    def has_pending_migration(self) -> bool:
-        """Cheaply check whether a migration is pending, with no exclusive
-        access needed -- just a metadata read under the connection callers
-        already hold via the write FileLock.
-
-        Callers should only pay for check_and_run_migrations()'s exclusive
-        access (a structural migration's file swap must not run while
-        readers are active) when this returns True, so that the common
-        case -- already at SCHEMA_VERSION -- never contends with readers
-        or a concurrent compaction.
-        """
-        current = self._stored_schema_version()
-        return current is not None and current < SCHEMA_VERSION
-
     def check_and_run_migrations(self) -> bool:
         """Apply any pending schema migrations to the store.
 
@@ -671,13 +559,15 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
         this method returns True when one is encountered so the caller can
         force a full rebuild (which recreates the table at SCHEMA_VERSION).
 
-        Must be called under the write FileLock, with readers excluded (see
-        has_pending_migration() for a cheap pre-check that avoids paying for
-        that exclusion in the common case).  No-op when the table does not
-        exist or is already at SCHEMA_VERSION.
+        Must be called under the write FileLock.  No-op when the table does
+        not exist or is already at SCHEMA_VERSION.
         """
-        current = self._stored_schema_version()
-        if current is None or current >= SCHEMA_VERSION:
+        if not self.table_exists():
+            return False
+
+        raw = self._meta_get("schema_version")
+        current = int(raw) if raw is not None else SCHEMA_VERSION
+        if current >= SCHEMA_VERSION:
             return False
 
         pending = sorted(
@@ -689,7 +579,7 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
             if migration.kind == "re-embed":
                 logger.warning(
                     "LLM index schema v%d -> v%d requires re-embedding (%s); "
-                    "the caller must force a rebuild.",
+                    "forcing full rebuild.",
                     migration.from_version,
                     migration.to_version,
                     migration.description,
@@ -711,12 +601,16 @@ class PaperlessSqliteVecVectorStore(BasePydanticVectorStore):
         dim = self.vector_dim()
         if dim is None:  # pragma: no cover
             raise RuntimeError("Cannot migrate: no stored vector dimension")
-        with self._rebuild_file() as new_conn:
+        db_path = str(Path(self._uri) / DB_FILENAME)
+        compact_path = db_path + ".compact"
+        new_conn = self._open_connection(compact_path)
+        try:
             migration.apply(self._conn, new_conn, dim)
-            IndexMetaTable.set_schema_version(new_conn, migration.to_version)
-
-
-# Registers m0001_v1_to_v2 into MIGRATIONS; must be at the bottom (needs
-# PaperlessSqliteVecVectorStore fully defined) -- see
-# paperless_ai/migrations/__init__.py for the full procedure.
-from paperless_ai.migrations import m0001_v1_to_v2  # noqa: E402, F401
+            self._meta_set_on(new_conn, "schema_version", str(migration.to_version))
+        except BaseException:  # pragma: no cover
+            new_conn.close()
+            for p in [compact_path, compact_path + "-wal", compact_path + "-shm"]:
+                Path(p).unlink(missing_ok=True)
+            raise
+        new_conn.close()
+        self._swap_in_compact(compact_path, db_path)