fix: upsert document vectors before pruning stale chunks

DocumentSyncService._index_vectors ran delete_by_filter on the
document's existing chunks FIRST and only then embedded; if the
embedding pass failed (Ollama down) the Paperless document was left
with zero vectors until the next successful sync - the same
zero-vector hazard already fixed for wiki pages in
VectorService.update_from_page.

Chunk ids are now deterministic uuid5 (document_{id}_chunk_{i}) so
re-upserting overwrites in place; new points are upserted first, then
stale points (including legacy random-uuid4 ones) are pruned via
scroll + delete_by_ids, and only after a successful upsert.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L

CHANGELOG.md

@@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed (Phase D review batch)
 
 - **Runtime prefetch registration is executable** - `SchedulerClient.register_volatile_fetch` (used by consolidation's `_register_prefetch`) registered tasks that were dead on arrival three ways: the JSON body sat under the ignored `body` key (`rest_api_executor` only reads `config["payload"]`), there was no `auth` block (the scheduled POST would 401 against library-desk's `verify_api_key`), and `user` was in the body while every `/volatile/fetch` endpoint requires it as a QUERY parameter (would 422). The config now puts `user` in the URL query string (URL-encoded), an empty `payload`, and `auth: {type: bearer, token: "${LIBRARY_API_KEY}"}` (substituted from the Scheduler's environment; never stored raw). `SchedulerClient` itself also sent no Authorization to the Scheduler API, so registration failed silently at consolidation time — it now sends `Authorization: Bearer <SCHEDULER_API_KEY>` (new `scheduler_api_key` setting; a client without a key logs a warning).
+- **Document sync uses delete-last reindex order** - `DocumentSyncService._index_vectors` deleted the document's existing chunks (`delete_by_filter`) BEFORE embedding, so a failed embedding pass (e.g. Ollama down) left the Paperless document with zero vectors until the next successful sync — the exact hazard already fixed for wiki pages in `VectorService.update_from_page`. Chunk point ids are now deterministic (uuid5 of `document_{id}_chunk_{i}`, replacing random uuid4) so re-upserting overwrites in place; new points are upserted first and stale points (including legacy uuid4 ones) are pruned afterwards, only after a successful upsert.
 - **Registrar authenticates to the Scheduler API** - `scripts/register_scheduler_tasks.py --execute` sent no `Authorization` header while the Scheduler's task-management endpoints are Bearer-guarded (`verify_api_key`: 401 on missing key), so the existence probes 401'd (misread as "task absent") and every registration failed; only the public `/health` gate passed. `--execute` now requires `SCHEDULER_API_KEY` in the environment (refuses to run without it, key is never stored) and sends `Authorization: Bearer $SCHEDULER_API_KEY` on all of its own HTTP calls. Deploy note updated alongside the existing `LIBRARY_API_KEY` requirement.
 
 ### Fixed (hazards batch)

src/services/document_sync_service.py

@@ -196,18 +196,6 @@ class DocumentSyncService:
         collection = get_qdrant_collection_name(user)
         await self.qdrant.ensure_collection(collection)
 
-        # Delete existing chunks for this document (via the async wrapper)
-        try:
-            await self.qdrant.delete_by_filter(
-                collection_name=collection,
-                filter_conditions={
-                    "doc_type": "document",
-                    "paperless_id": document_id,
-                }
-            )
-        except Exception as e:
-            logger.debug(f"No existing chunks to delete: {e}")
-
         # Chunk content
         chunks = self._chunk_text(content)
         if not chunks:
@@ -229,7 +217,11 @@ class DocumentSyncService:
                 )
                 continue
 
-            point_id = str(uuid.uuid4())
+            # Deterministic id: re-upserting the same document overwrites
+            # its previous chunks in place (enables delete-last below).
+            point_id = str(
+                uuid.uuid5(uuid.NAMESPACE_DNS, f"document_{document_id}_chunk_{i}")
+            )
             content_hash = hashlib.md5(chunk.encode()).hexdigest()
 
             points.append({
@@ -256,13 +248,38 @@ class DocumentSyncService:
                 f"(embedding failures); indexing the remaining {len(points)}"
             )
 
-        # Upsert to Qdrant in one batch via the async wrapper
+        # Upsert BEFORE pruning stale chunks (same order as the wiki
+        # reindex fix in VectorService.update_from_page): the old
+        # delete-first order left the document with ZERO vectors until the
+        # next successful sync whenever the embedding pass failed after the
+        # delete (e.g. Ollama down). Deterministic uuid5 ids make the
+        # in-place overwrite safe.
         if points:
             await self.qdrant.upsert_points(
                 collection_name=collection,
                 points=points
             )
 
+            # Prune chunks left over from a previous version of the document
+            # (indexes beyond the new count, or legacy random-uuid4 points).
+            # Only prune after a successful upsert - a fully failed embedding
+            # pass must not wipe the old vectors.
+            new_ids = {p["id"] for p in points}
+            existing = await self.qdrant.scroll_all_points(
+                collection_name=collection,
+                filter_conditions={
+                    "doc_type": "document",
+                    "paperless_id": document_id,
+                },
+                with_payload=False,
+            )
+            stale_ids = [pt["id"] for pt in existing if pt["id"] not in new_ids]
+            if stale_ids:
+                await self.qdrant.delete_by_ids(
+                    collection_name=collection,
+                    point_ids=stale_ids,
+                )
+
         return len(points)
 
     async def _index_graph(

tests/test_document_sync.py

@@ -6,10 +6,13 @@ Pins the Phase D fixes:
   that never ran, so fresh tenants had no collection at upsert time)
 - None entries from embed_batch are filtered out instead of poisoning
   the whole batch upsert (one failed chunk aborted the document)
-- raw client.delete/client.upsert calls are routed through the async
-  wrapper methods (delete_by_filter / upsert_points)
+- delete-LAST reindex order (same as the wiki fix): new points are
+  upserted with deterministic uuid5 ids BEFORE stale points are pruned,
+  so a failed embedding pass can no longer leave a document with zero
+  vectors (the old order ran delete_by_filter first)
 """
 
+import uuid
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
@@ -24,8 +27,9 @@ TENANT_COLLECTION = "library_desk_llm_tester"
 def mock_qdrant():
     qdrant = MagicMock()
     qdrant.ensure_collection = AsyncMock()
-    qdrant.delete_by_filter = AsyncMock(return_value=0)
     qdrant.upsert_points = AsyncMock(side_effect=lambda collection_name, points: len(points))
+    qdrant.scroll_all_points = AsyncMock(return_value=[])
+    qdrant.delete_by_ids = AsyncMock(side_effect=lambda collection_name, point_ids: len(point_ids))
     return qdrant
 
 
@@ -75,14 +79,46 @@ class TestIndexVectors:
         assert result.success is True
         mock_qdrant.ensure_collection.assert_awaited_once_with(TENANT_COLLECTION)
 
-    async def test_delete_routed_through_wrapper(self, sync_service, mock_qdrant):
+    async def test_point_ids_are_deterministic(self, sync_service, mock_qdrant):
         await sync_service.index_document(
             document_id=7, user=TENANT, content="hello world", title="Doc"
         )
 
-        mock_qdrant.delete_by_filter.assert_awaited_once_with(
+        points = mock_qdrant.upsert_points.await_args.kwargs["points"]
+        expected = str(uuid.uuid5(uuid.NAMESPACE_DNS, "document_7_chunk_0"))
+        assert points[0]["id"] == expected
+
+    async def test_stale_chunks_pruned_after_upsert(self, sync_service, mock_qdrant):
+        """Old points not in the new set are deleted AFTER the new upsert."""
+        call_order = []
+        mock_qdrant.upsert_points = AsyncMock(
+            side_effect=lambda collection_name, points: (
+                call_order.append("upsert"), len(points))[1]
+        )
+        stale_id = str(uuid.uuid4())  # legacy random-uuid4 point
+        kept_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, "document_7_chunk_0"))
+        mock_qdrant.scroll_all_points = AsyncMock(
+            return_value=[{"id": stale_id, "payload": {}},
+                          {"id": kept_id, "payload": {}}]
+        )
+        mock_qdrant.delete_by_ids = AsyncMock(
+            side_effect=lambda collection_name, point_ids: (
+                call_order.append("delete"), len(point_ids))[1]
+        )
+
+        await sync_service.index_document(
+            document_id=7, user=TENANT, content="hello world", title="Doc"
+        )
+
+        assert call_order == ["upsert", "delete"]
+        mock_qdrant.delete_by_ids.assert_awaited_once_with(
+            collection_name=TENANT_COLLECTION,
+            point_ids=[stale_id],
+        )
+        mock_qdrant.scroll_all_points.assert_awaited_once_with(
             collection_name=TENANT_COLLECTION,
             filter_conditions={"doc_type": "document", "paperless_id": 7},
+            with_payload=False,
         )
 
     async def test_upsert_routed_through_wrapper(self, sync_service, mock_qdrant):
@@ -134,3 +170,7 @@ class TestIndexVectors:
         assert result.success is False
         assert "embed" in (result.error or "").lower()
         mock_qdrant.upsert_points.assert_not_awaited()
+        # Delete-last: a fully failed embedding pass must leave the old
+        # vectors untouched (previously delete_by_filter ran first, leaving
+        # the document with zero vectors until the next successful sync)
+        mock_qdrant.delete_by_ids.assert_not_awaited()