fix(vector_store): dedupe qdrant record conversion, don't abort iter_all on a live cursor with an empty page

semantica/vector_store/qdrant_store.py

@@ -590,20 +590,19 @@ class QdrantStore:
                 with_payload=True,
                 with_vectors=True,
             )
-            results = []
-            for rec in records:
-                results.append(
-                    {
-                        "id": str(rec.id),
-                        "metadata": rec.payload or {},
-                        "vector": np.array(rec.vector) if rec.vector is not None else None,
-                    }
-                )
-            return results
+            return [self._record_to_result(rec) for rec in records]
         except Exception as e:
             self.logger.warning(f"Failed to scroll Qdrant points by metadata filter: {e}")
             return []
 
+    @staticmethod
+    def _record_to_result(rec: Any) -> Dict[str, Any]:
+        return {
+            "id": str(rec.id),
+            "metadata": rec.payload or {},
+            "vector": np.array(rec.vector) if rec.vector is not None else None,
+        }
+
     def iter_all(self, batch_size: int = 500):
         """
         Iterate over every stored point using Qdrant's scroll cursor.
@@ -624,7 +623,7 @@ class QdrantStore:
 
         Raises:
             ProcessingError: If the collection or client is not initialized, or
-                if the scan cannot advance past a full page.
+                if the cursor stops advancing before the scan completes.
         """
         if self.collection is None or self.client is None or not QDRANT_AVAILABLE:
             raise ProcessingError(
@@ -632,6 +631,7 @@ class QdrantStore:
             )
 
         next_offset = None
+        last_offset = object()
         while True:
             records, next_offset = self.client.scroll(
                 collection_name=self.collection.collection_name,
@@ -642,11 +642,7 @@ class QdrantStore:
             )
 
             for rec in records:
-                yield {
-                    "id": str(rec.id),
-                    "metadata": rec.payload or {},
-                    "vector": np.array(rec.vector) if rec.vector is not None else None,
-                }
+                yield self._record_to_result(rec)
 
             # A final page can carry records alongside a null cursor, so they
             # are yielded above before stopping. Passing offset=None back to
@@ -654,14 +650,16 @@ class QdrantStore:
             if next_offset is None:
                 return
 
-            # A cursor without records means the scan cannot advance, which is
-            # truncation rather than completion.
-            if not records:
+            # An empty page with a live cursor isn't necessarily truncation —
+            # a batch window that lands entirely on deleted points comes back
+            # this way too, and there's more to scan past it. Only treat it as
+            # stuck if the cursor itself stops moving.
+            if not records and next_offset == last_offset:
                 raise ProcessingError(
-                    "Qdrant returned an empty page alongside a continuation "
-                    "cursor, so the scan cannot advance. Refusing to return a "
-                    "truncated scan."
+                    "Qdrant scroll cursor stopped advancing without reaching "
+                    "the end of the collection, so the scan cannot complete."
                 )
+            last_offset = next_offset
 
     def delete_vectors(
         self, point_ids: List[Union[str, int]], **options

tests/vector_store/test_qdrant_store.py

@@ -94,12 +94,26 @@ def test_iter_all_empty_collection_yields_nothing():
 
 
 @patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
-def test_iter_all_raises_on_empty_page_with_a_cursor():
-    """An empty page with a cursor left set cannot advance, so returning here
-    would hand back a partial scan that reads as a complete one."""
-    store = _store_with_scroll(([], "cursor-that-never-clears"))
+def test_iter_all_continues_past_empty_page_with_advancing_cursor():
+    store = _store_with_scroll(
+        ([], "cursor-1"),
+        ([_record(1)], None),
+    )
 
-    with pytest.raises(ProcessingError, match="cannot advance"):
+    result = list(store.iter_all())
+
+    assert [item["id"] for item in result] == ["1"]
+    assert store.client.scroll.call_count == 2
+
+
+@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
+def test_iter_all_raises_when_cursor_stops_advancing():
+    store = _store_with_scroll(
+        ([], "stuck-cursor"),
+        ([], "stuck-cursor"),
+    )
+
+    with pytest.raises(ProcessingError, match="stopped advancing"):
         list(store.iter_all())