fix: await ensure_collection and survive partial embed failures in document sync

- ensure_collection was called without await, so the coroutine never ran
  and fresh tenants had no collection when the upsert hit Qdrant
- a single None entry from embed_batch poisoned the point batch and
  aborted the whole document upsert; failed chunks are now skipped with
  a warning (all-failed raises and the IndexResult reports failure)
- raw client.delete/client.upsert calls now go through the async wrapper
  (delete_by_filter and the new batch upsert_points method)

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

CHANGELOG.md

@@ -20,6 +20,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed (performance)
 
+- **Document sync indexing fixed** - `DocumentSyncService._index_vectors` now awaits `ensure_collection` (the coroutine was created but never ran, so fresh tenants had no collection at upsert time), filters out `None` entries from `embed_batch` so one failed chunk embedding no longer aborts the whole document upsert (all-failed still reports failure), and routes the raw `client.delete`/`client.upsert` calls through the async wrapper (`delete_by_filter` / new batch `upsert_points`). Offline unit tests added.
 - **Async Qdrant client** - `QdrantClientWrapper` now uses `AsyncQdrantClient` with an explicit timeout (`QDRANT_TIMEOUT`, default 30s). Every vector call previously ran on the synchronous client inside async wrapper methods, blocking the FastAPI event loop for the duration of each Qdrant round-trip. The wrapper API is unchanged (all methods were already `async`), so call sites only gained real awaits. The HybridRAG document leg was moved off the deprecated raw `client.search` onto the wrapper's `search_vectors` (fixing a latent `AttributeError`: it called the nonexistent `ollama.embed_text`, so the leg always reported `failed`), and the health check awaits `get_collections`.
 
 ### Changed

src/clients/qdrant_client.py

@@ -467,6 +467,43 @@ class QdrantClientWrapper:
             logger.error(f"Failed to upsert vector: {e}", exc_info=True)
             return False
 
+    async def upsert_points(
+        self,
+        collection_name: str,
+        points: List[Dict[str, Any]]
+    ) -> int:
+        """
+        Upsert a batch of vector points in a single request.
+
+        Args:
+            collection_name: Collection name
+            points: List of {"id": str, "vector": List[float], "payload": dict}
+
+        Returns:
+            Number of points upserted
+
+        Raises:
+            Exception: If the upsert fails (callers decide how to degrade)
+        """
+        if not points:
+            return 0
+
+        structs = [
+            PointStruct(
+                id=p["id"],
+                vector=p["vector"],
+                payload=p.get("payload", {})
+            )
+            for p in points
+        ]
+
+        await self.client.upsert(
+            collection_name=collection_name,
+            points=structs
+        )
+        logger.info(f"Upserted {len(structs)} points into {collection_name}")
+        return len(structs)
+
     async def delete_by_filter(
         self,
         collection_name: str,

src/services/document_sync_service.py

@@ -194,19 +194,15 @@ class DocumentSyncService:
     ) -> int:
         """Create vector embeddings for document content."""
         collection = get_qdrant_collection_name(user)
-        self.qdrant.ensure_collection(collection)
+        await self.qdrant.ensure_collection(collection)
 
-        # Delete existing chunks for this document
+        # Delete existing chunks for this document (via the async wrapper)
         try:
-            await self.qdrant.client.delete(
+            await self.qdrant.delete_by_filter(
                 collection_name=collection,
-                points_selector={
-                    "filter": {
-                        "must": [
-                            {"key": "doc_type", "match": {"value": "document"}},
-                            {"key": "paperless_id", "match": {"value": document_id}},
-                        ]
-                    }
+                filter_conditions={
+                    "doc_type": "document",
+                    "paperless_id": document_id,
                 }
             )
         except Exception as e:
@@ -217,12 +213,22 @@ class DocumentSyncService:
         if not chunks:
             return 0
 
-        # Generate embeddings
+        # Generate embeddings (embed_batch returns None for failed chunks)
         embeddings = await self.ollama.embed_batch(chunks)
 
-        # Build points
+        # Build points, skipping chunks whose embedding failed. Previously a
+        # single None embedding poisoned the batch and aborted the whole
+        # document upsert.
         points = []
+        skipped = 0
         for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
+            if embedding is None:
+                skipped += 1
+                logger.warning(
+                    f"Skipping chunk {i} of document {document_id}: embedding failed"
+                )
+                continue
+
             point_id = str(uuid.uuid4())
             content_hash = hashlib.md5(chunk.encode()).hexdigest()
 
@@ -240,9 +246,19 @@ class DocumentSyncService:
                 }
             })
 
-        # Upsert to Qdrant
+        if skipped and not points:
+            raise RuntimeError(
+                f"All {skipped} chunk embeddings failed for document {document_id}"
+            )
+        if skipped:
+            logger.warning(
+                f"Document {document_id}: {skipped}/{len(chunks)} chunks skipped "
+                f"(embedding failures); indexing the remaining {len(points)}"
+            )
+
+        # Upsert to Qdrant in one batch via the async wrapper
         if points:
-            await self.qdrant.client.upsert(
+            await self.qdrant.upsert_points(
                 collection_name=collection,
                 points=points
             )

tests/test_document_sync.py

@@ -0,0 +1,135 @@
+"""
+Offline unit tests for DocumentSyncService vector indexing fixes.
+
+Pins the Phase D fixes:
+- ensure_collection is actually awaited (it used to be a bare coroutine
+  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)
+"""
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from src.services.document_sync_service import DocumentSyncService
+
+TENANT = "llm_tester"
+TENANT_COLLECTION = "library_desk_llm_tester"
+
+
+@pytest.fixture
+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))
+    return qdrant
+
+
+@pytest.fixture
+def mock_ollama():
+    ollama = MagicMock()
+    ollama.embed_batch = AsyncMock(
+        side_effect=lambda texts: [[0.1] * 768 for _ in texts]
+    )
+    return ollama
+
+
+@pytest.fixture
+def mock_neo4j():
+    neo4j = MagicMock()
+    neo4j.execute_query = AsyncMock(return_value=[])
+    return neo4j
+
+
+@pytest.fixture
+def mock_paperless():
+    paperless = MagicMock()
+    paperless.get_custom_field_by_name = AsyncMock(return_value=None)
+    return paperless
+
+
+@pytest.fixture
+def sync_service(mock_paperless, mock_qdrant, mock_ollama, mock_neo4j):
+    return DocumentSyncService(
+        paperless_client=mock_paperless,
+        qdrant_client=mock_qdrant,
+        ollama_client=mock_ollama,
+        neo4j_client=mock_neo4j,
+        wiki_client=MagicMock(),
+        settings=MagicMock(),
+    )
+
+
+@pytest.mark.unit
+class TestIndexVectors:
+    async def test_ensure_collection_is_awaited(self, sync_service, mock_qdrant):
+        result = await sync_service.index_document(
+            document_id=1, user=TENANT, content="hello world", title="Doc"
+        )
+
+        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):
+        await sync_service.index_document(
+            document_id=7, user=TENANT, content="hello world", title="Doc"
+        )
+
+        mock_qdrant.delete_by_filter.assert_awaited_once_with(
+            collection_name=TENANT_COLLECTION,
+            filter_conditions={"doc_type": "document", "paperless_id": 7},
+        )
+
+    async def test_upsert_routed_through_wrapper(self, sync_service, mock_qdrant):
+        result = await sync_service.index_document(
+            document_id=1, user=TENANT, content="hello world", title="Doc"
+        )
+
+        assert result.chunks_created == 1
+        kwargs = mock_qdrant.upsert_points.await_args.kwargs
+        assert kwargs["collection_name"] == TENANT_COLLECTION
+        payload = kwargs["points"][0]["payload"]
+        assert payload["doc_type"] == "document"
+        assert payload["paperless_id"] == 1
+
+    async def test_failed_chunk_embedding_is_skipped_not_fatal(
+        self, sync_service, mock_qdrant, mock_ollama
+    ):
+        # Three chunks; the middle embedding fails
+        long_content = " ".join(f"word{i}" for i in range(1200))
+        mock_ollama.embed_batch = AsyncMock(
+            side_effect=lambda texts: [
+                [0.1] * 768 if i != 1 else None for i in range(len(texts))
+            ]
+        )
+
+        result = await sync_service.index_document(
+            document_id=2, user=TENANT, content=long_content, title="Doc"
+        )
+
+        assert result.success is True
+        points = mock_qdrant.upsert_points.await_args.kwargs["points"]
+        assert result.chunks_created == len(points)
+        # The failed chunk (index 1) is absent, the others kept their index
+        indices = [p["payload"]["chunk_index"] for p in points]
+        assert 1 not in indices
+        assert len(indices) >= 2
+
+    async def test_all_embeddings_failed_reports_failure(
+        self, sync_service, mock_qdrant, mock_ollama
+    ):
+        mock_ollama.embed_batch = AsyncMock(
+            side_effect=lambda texts: [None for _ in texts]
+        )
+
+        result = await sync_service.index_document(
+            document_id=3, user=TENANT, content="hello world", title="Doc"
+        )
+
+        assert result.success is False
+        assert "embed" in (result.error or "").lower()
+        mock_qdrant.upsert_points.assert_not_awaited()