From d175f894a42fa1f60e5bae33b5cb43e2798bfa1e Mon Sep 17 00:00:00 2001
From: Zohaib Hassnain <109234410+ZohaibHassan16@users.noreply.github.com>
Date: Mon, 31 Aug 2026 02:34:26 +0500
Subject: [PATCH 1/5] feat(vector_store): add iter_all enumeration and wire up
 qdrant migration

---
 semantica/cli.py                              |  31 +++-
 semantica/vector_store/qdrant_store.py        |  56 +++++++
 semantica/vector_store/vector_store.py        |  15 +-
 tests/test_cli_commands.py                    |  96 ++++++++++-
 tests/vector_store/test_qdrant_store.py       | 152 ++++++++++++++++++
 .../test_vector_manager_persistent.py         | 108 +++++++++++++
 6 files changed, 447 insertions(+), 11 deletions(-)
 create mode 100644 tests/vector_store/test_qdrant_store.py

diff --git a/semantica/cli.py b/semantica/cli.py
index d05a00f1..957878f4 100644
--- a/semantica/cli.py
+++ b/semantica/cli.py
@@ -5,6 +5,7 @@ This module provides the command-line interface for the Semantica framework,
 enabling users to interact with the framework via terminal commands.
 """
 
+import itertools
 import json
 import os
 import sys
@@ -3714,7 +3715,7 @@ def store_stats(cli_ctx: CLIContext, backend: str, fmt: str, local_json: bool) -
     _run_with_error_handling(_action)
 
 
-_MIGRATE_SUPPORTED_BACKENDS = {"faiss", "sqlite", "pgvector"}
+_MIGRATE_SUPPORTED_BACKENDS = {"faiss", "sqlite", "pgvector", "qdrant"}
 _MIGRATE_BATCH_SIZE = 500
 
 
@@ -3759,12 +3760,10 @@ def store_migrate(cli_ctx: CLIContext, from_backend: str, to_backend: str,
                   namespace: Optional[str], local_dry: bool, local_json: bool) -> None:
     """Migrate data between backends.
 
-    Direct migration is only wired up between faiss, sqlite, and pgvector -
-    these are the backends whose storage contract supports paging through
-    every stored vector. Migrating to or from qdrant, pinecone, milvus, or
-    weaviate still needs the export/reindex workaround below, since each of
-    those needs its own enumeration design (Qdrant scroll, Pinecone list,
-    etc.) that hasn't been built yet.
+    Wired up between faiss, sqlite, pgvector and qdrant. Migrating to or from
+    pinecone, milvus or weaviate still needs the export/reindex workaround
+    below, since each of those needs its own enumeration design (Pinecone
+    list, Milvus query_iterator, Weaviate cursor) that is not built yet.
 
     \b
     Example:
@@ -3804,7 +3803,18 @@ def store_migrate(cli_ctx: CLIContext, from_backend: str, to_backend: str,
         if source_index_path:
             source._backend_store.load_index(source_index_path)
 
+        # Pull the first record before building the destination: qdrant, milvus
+        # and weaviate expose no `dimension` attribute, so without this the
+        # destination falls back to the default 768 and rejects every insert.
+        # The data itself is the reliable source of truth.
+        source_iter = source.iter_vectors(batch_size=_MIGRATE_BATCH_SIZE)
+        first_item = next(source_iter, None)
+
         source_dimension = getattr(source._backend_store, "dimension", None)
+        if not source_dimension and first_item is not None:
+            first_vector = first_item.get("vector")
+            if first_vector is not None:
+                source_dimension = len(first_vector)
         if source_dimension and "dimension" not in dest_cfg:
             dest_cfg["dimension"] = source_dimension
 
@@ -3827,7 +3837,12 @@ def store_migrate(cli_ctx: CLIContext, from_backend: str, to_backend: str,
             metadata_batch.clear()
             ids_batch.clear()
 
-        for item in source.iter_vectors(batch_size=_MIGRATE_BATCH_SIZE):
+        items = (
+            source_iter
+            if first_item is None
+            else itertools.chain([first_item], source_iter)
+        )
+        for item in items:
             meta = dict(item.get("metadata") or {})
             if namespace and "namespace" not in meta:
                 meta["namespace"] = namespace
diff --git a/semantica/vector_store/qdrant_store.py b/semantica/vector_store/qdrant_store.py
index 1779a0bc..85ce7302 100644
--- a/semantica/vector_store/qdrant_store.py
+++ b/semantica/vector_store/qdrant_store.py
@@ -604,6 +604,62 @@ class QdrantStore:
             self.logger.warning(f"Failed to scroll Qdrant points by metadata filter: {e}")
             return []
 
+    def iter_all(self, batch_size: int = 500):
+        """
+        Iterate over every stored point using Qdrant's native scroll cursor.
+
+        Qdrant paginates by point-ID cursor, not by row offset, so this is
+        exposed instead of scan_vectors(offset, limit). An integer passed to
+        scroll()'s offset is a point ID rather than a rank, so there is no way
+        to seek to "the Nth record" without walking from the start.
+        VectorStore.iter_vectors() prefers this method when it is present.
+
+        Assumes a single unnamed vector per point, matching how insert_vectors()
+        writes them and how get_vector() reads them back. Collections configured
+        with named or multi-vectors are not handled here.
+
+        Args:
+            batch_size: Points to request per scroll call
+
+        Yields:
+            Result dicts with 'id', 'metadata', and 'vector', in scroll order
+
+        Raises:
+            ProcessingError: If the collection or client is not initialized.
+                Errors are raised rather than swallowed because a scan that
+                silently yields nothing is indistinguishable from an empty
+                source, which would let a caller such as `store migrate`
+                report success having copied nothing (issue #1083).
+        """
+        if self.collection is None or self.client is None or not QDRANT_AVAILABLE:
+            raise ProcessingError(
+                "Collection not initialized. Call create_collection() or get_collection() first."
+            )
+
+        next_offset = None
+        while True:
+            records, next_offset = self.client.scroll(
+                collection_name=self.collection.collection_name,
+                limit=batch_size,
+                offset=next_offset,
+                with_payload=True,
+                with_vectors=True,
+            )
+
+            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,
+                }
+
+            # The final page can carry records while already reporting no next
+            # cursor, so those records are yielded above before stopping here.
+            # Calling scroll() again with offset=None would restart from the
+            # beginning rather than continue past the end.
+            if next_offset is None or not records:
+                return
+
     def delete_vectors(
         self, point_ids: List[Union[str, int]], **options
     ) -> Dict[str, Any]:
diff --git a/semantica/vector_store/vector_store.py b/semantica/vector_store/vector_store.py
index d0257240..cd703eda 100644
--- a/semantica/vector_store/vector_store.py
+++ b/semantica/vector_store/vector_store.py
@@ -867,12 +867,25 @@ class VectorStore:
         """
         Iterate over every stored vector, one page at a time.
 
+        Backends whose native pagination is cursor based (Qdrant, Pinecone,
+        Milvus, Weaviate) cannot honestly implement the positional
+        scan_vectors(offset, limit) contract, so they expose iter_all()
+        instead and it is preferred here when present. Backends with real
+        positional access (inmemory, FAISS, SQLite-vec, PgVector) fall
+        through to the offset loop below.
+
         Args:
-            batch_size: Number of vectors to fetch per underlying scan_vectors() call
+            batch_size: Number of vectors to fetch per underlying call
 
         Yields:
             Result dicts with 'id', 'metadata', and 'vector', in scan order
         """
+        if self.backend != "inmemory" and self._backend_store is not None:
+            iter_all = getattr(self._backend_store, "iter_all", None)
+            if callable(iter_all):
+                yield from iter_all(batch_size=batch_size)
+                return
+
         offset = 0
         while True:
             page = self.scan_vectors(offset=offset, limit=batch_size)
diff --git a/tests/test_cli_commands.py b/tests/test_cli_commands.py
index 4a054be1..a90673c4 100644
--- a/tests/test_cli_commands.py
+++ b/tests/test_cli_commands.py
@@ -1360,10 +1360,11 @@ class TestStore:
         assert result.exit_code != 0
 
     def test_migrate_refuses_unsupported_backend_pair(self, runner):
+        # qdrant is wired up now, so weaviate stands in as the unsupported side.
         result = runner.invoke(cli_module.main, ["store", "migrate",
-                                      "--from", "faiss", "--to", "qdrant"])
+                                      "--from", "faiss", "--to", "weaviate"])
         assert result.exit_code != 0
-        assert "faiss, pgvector, sqlite" in result.output
+        assert "faiss, pgvector, qdrant, sqlite" in result.output
 
     def _fake_migrate_store_module(self, source_items, stored, dest_configs=None):
         class _FakeBackendStore:
@@ -1445,6 +1446,97 @@ class TestStore:
         assert result.exit_code != 0
         assert "index_path" in result.output
 
+    def _fake_dimensionless_store_module(self, source_items, dest_configs, stored):
+        """Fake VectorStore whose backend store has NO `dimension` attribute.
+
+        Matches qdrant/milvus/weaviate, none of which expose one.
+        """
+
+        class _DimensionlessBackendStore:
+            pass
+
+        class _FakeStore:
+            def __init__(self, backend, config=None, **kw):
+                self.backend = backend
+                self._config = config or {}
+                self._backend_store = _DimensionlessBackendStore()
+                dest_configs[backend] = dict(self._config)
+
+            def iter_vectors(self, batch_size=500):
+                if self.backend == "qdrant":
+                    yield from source_items
+                    return
+                return
+                yield  # pragma: no cover - makes this a generator
+
+            def store_vectors(self, vectors, metadata, ids=None):
+                for vec_id, vec in zip(ids, vectors):
+                    stored[vec_id] = vec
+
+        return _fake_module(VectorStore=_FakeStore)
+
+    def test_migrate_infers_dimension_from_first_vector(self, runner, monkeypatch):
+        source_items = [
+            {"id": "a", "vector": [0.1, 0.2, 0.3], "metadata": {}},
+            {"id": "b", "vector": [0.4, 0.5, 0.6], "metadata": {}},
+        ]
+        dest_configs, stored = {}, {}
+        fake_vs = self._fake_dimensionless_store_module(source_items, dest_configs, stored)
+        monkeypatch.setitem(__import__("sys").modules, "semantica.vector_store", fake_vs)
+
+        result = runner.invoke(cli_module.main, ["store", "migrate",
+                                      "--from", "qdrant", "--to", "sqlite", "--json"])
+        _ok(result)
+        assert dest_configs["sqlite"].get("dimension") == 3
+
+    def test_migrate_does_not_drop_the_peeked_first_record(self, runner, monkeypatch):
+        """The first record is consumed to infer dimension, so it must be
+        chained back into the migration loop rather than lost."""
+        source_items = [
+            {"id": "a", "vector": [0.1, 0.2, 0.3], "metadata": {}},
+            {"id": "b", "vector": [0.4, 0.5, 0.6], "metadata": {}},
+        ]
+        dest_configs, stored = {}, {}
+        fake_vs = self._fake_dimensionless_store_module(source_items, dest_configs, stored)
+        monkeypatch.setitem(__import__("sys").modules, "semantica.vector_store", fake_vs)
+
+        result = runner.invoke(cli_module.main, ["store", "migrate",
+                                      "--from", "qdrant", "--to", "sqlite", "--json"])
+        _ok(result)
+        assert _json_output(result)["migrated"] == 2
+        assert sorted(stored) == ["a", "b"]
+
+    def test_migrate_keeps_explicit_dest_dimension_over_inference(self, runner, monkeypatch):
+        source_items = [{"id": "a", "vector": [0.1, 0.2, 0.3], "metadata": {}}]
+        dest_configs, stored = {}, {}
+        fake_vs = self._fake_dimensionless_store_module(source_items, dest_configs, stored)
+        monkeypatch.setitem(__import__("sys").modules, "semantica.vector_store", fake_vs)
+        monkeypatch.setattr(
+            cli_module.Config, "to_dict",
+            lambda self: {"vector_store": {"qdrant": {}, "sqlite": {"dimension": 128}}},
+        )
+
+        result = runner.invoke(cli_module.main, ["store", "migrate",
+                                      "--from", "qdrant", "--to", "sqlite", "--json"])
+        _ok(result)
+        assert dest_configs["sqlite"]["dimension"] == 128
+
+    def test_migrate_qdrant_is_now_supported(self, runner, monkeypatch):
+        dest_configs, stored = {}, {}
+        fake_vs = self._fake_dimensionless_store_module([], dest_configs, stored)
+        monkeypatch.setitem(__import__("sys").modules, "semantica.vector_store", fake_vs)
+
+        result = runner.invoke(cli_module.main, ["store", "migrate",
+                                      "--from", "qdrant", "--to", "pgvector", "--json"])
+        _ok(result)
+        assert _json_output(result)["migrated"] == 0
+
+    def test_migrate_still_refuses_milvus(self, runner):
+        result = runner.invoke(cli_module.main, ["store", "migrate",
+                                      "--from", "qdrant", "--to", "milvus"])
+        assert result.exit_code != 0
+        assert "milvus" in result.output
+
     def test_migrate_faiss_dest_requires_index_path(self, runner, monkeypatch):
         fake_vs = _fake_module(VectorStore=lambda **kw: MagicMock())
         monkeypatch.setitem(__import__("sys").modules, "semantica.vector_store", fake_vs)
diff --git a/tests/vector_store/test_qdrant_store.py b/tests/vector_store/test_qdrant_store.py
new file mode 100644
index 00000000..89a9fee3
--- /dev/null
+++ b/tests/vector_store/test_qdrant_store.py
@@ -0,0 +1,152 @@
+"""Tests for QdrantStore.iter_all() cursor enumeration.
+
+Qdrant is not installed in this environment, so these drive the real
+QdrantStore against a MagicMock standing in for the qdrant_client, following
+the pattern already used for qdrant in test_backend_metadata_filtering.py.
+"""
+
+from unittest.mock import MagicMock, patch
+
+import numpy as np
+import pytest
+
+from semantica.utils.exceptions import ProcessingError
+from semantica.vector_store.qdrant_store import QdrantStore
+
+
+def _record(point_id, payload=None, vector=None):
+    """Build a stand-in for a qdrant_client Record."""
+    rec = MagicMock()
+    rec.id = point_id
+    rec.payload = payload
+    rec.vector = vector
+    return rec
+
+
+def _store_with_scroll(*pages):
+    """QdrantStore whose client.scroll() returns the given (records, cursor) pages."""
+    store = QdrantStore()
+    store.client = MagicMock()
+    store.client.scroll.side_effect = list(pages)
+    store.collection = MagicMock()
+    store.collection.collection_name = "test_collection"
+    return store
+
+
+@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
+def test_iter_all_threads_cursor_across_pages():
+    """The next call must continue from the previous page's next_page_offset."""
+    store = _store_with_scroll(
+        ([_record(1), _record(2)], "cursor-1"),
+        ([_record(3)], None),
+    )
+
+    result = list(store.iter_all(batch_size=2))
+
+    assert [item["id"] for item in result] == ["1", "2", "3"]
+    calls = store.client.scroll.call_args_list
+    assert len(calls) == 2
+    assert calls[0][1]["offset"] is None
+    assert calls[0][1]["limit"] == 2
+    assert calls[1][1]["offset"] == "cursor-1"
+
+
+@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
+def test_iter_all_yields_final_page_that_reports_no_next_cursor():
+    """Qdrant can return records and a null cursor on the same page.
+
+    Those records must still be yielded. Treating a null cursor as "stop
+    before this page" would silently drop the tail of every scan.
+    """
+    store = _store_with_scroll(([_record(1), _record(2)], None))
+
+    result = list(store.iter_all(batch_size=10))
+
+    assert [item["id"] for item in result] == ["1", "2"]
+    assert store.client.scroll.call_count == 1
+
+
+@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
+def test_iter_all_converts_records_to_the_shared_result_shape():
+    store = _store_with_scroll(
+        ([_record(7, payload={"tag": "x"}, vector=[0.1, 0.2, 0.3])], None),
+    )
+
+    item = list(store.iter_all())[0]
+
+    assert item["id"] == "7"
+    assert item["metadata"] == {"tag": "x"}
+    np.testing.assert_allclose(item["vector"], np.array([0.1, 0.2, 0.3]))
+
+
+@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
+def test_iter_all_handles_missing_payload_and_vector():
+    store = _store_with_scroll(([_record(1, payload=None, vector=None)], None))
+
+    item = list(store.iter_all())[0]
+
+    assert item["metadata"] == {}
+    assert item["vector"] is None
+
+
+@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
+def test_iter_all_empty_collection_yields_nothing():
+    store = _store_with_scroll(([], None))
+
+    assert list(store.iter_all()) == []
+
+
+@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
+def test_iter_all_stops_on_empty_page_even_with_a_cursor():
+    """Defensive: an empty page ends the scan rather than looping forever."""
+    store = _store_with_scroll(([], "cursor-that-never-clears"))
+
+    assert list(store.iter_all()) == []
+    assert store.client.scroll.call_count == 1
+
+
+@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
+def test_iter_all_raises_when_collection_not_initialized():
+    """Must fail loudly, not yield nothing.
+
+    An empty scan is indistinguishable from an empty source, which would let
+    `store migrate` report success having copied nothing (issue #1083).
+    """
+    store = QdrantStore()
+
+    with pytest.raises(ProcessingError, match="Collection not initialized"):
+        list(store.iter_all())
+
+
+@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", False)
+def test_iter_all_raises_when_qdrant_unavailable():
+    store = QdrantStore()
+    store.client = MagicMock()
+    store.collection = MagicMock()
+
+    with pytest.raises(ProcessingError):
+        list(store.iter_all())
+
+
+@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
+def test_iter_all_propagates_scroll_errors():
+    store = QdrantStore()
+    store.client = MagicMock()
+    store.client.scroll.side_effect = RuntimeError("connection reset")
+    store.collection = MagicMock()
+    store.collection.collection_name = "test_collection"
+
+    with pytest.raises(RuntimeError, match="connection reset"):
+        list(store.iter_all())
+
+
+@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
+def test_iter_all_requests_payload_and_vectors():
+    store = _store_with_scroll(([], None))
+
+    list(store.iter_all())
+
+    kwargs = store.client.scroll.call_args[1]
+    assert kwargs["with_payload"] is True
+    assert kwargs["with_vectors"] is True
+    assert kwargs["collection_name"] == "test_collection"
diff --git a/tests/vector_store/test_vector_manager_persistent.py b/tests/vector_store/test_vector_manager_persistent.py
index 6a432939..f0586ad2 100644
--- a/tests/vector_store/test_vector_manager_persistent.py
+++ b/tests/vector_store/test_vector_manager_persistent.py
@@ -26,6 +26,7 @@ from unittest.mock import MagicMock, patch
 
 import numpy as np
 
+from semantica.utils.exceptions import ProcessingError
 from semantica.vector_store.vector_store import VectorStore, VectorManager
 
 
@@ -138,6 +139,38 @@ class _NonScanningBackendStore:
     """Fake persistent backend store without any scan capability."""
 
 
+class _IterAllBackendStore:
+    """Fake cursor-based backend store exposing iter_all() but not scan_vectors().
+
+    Mirrors qdrant/pinecone/milvus/weaviate, which cannot honour a positional
+    offset and therefore expose native iteration instead.
+    """
+
+    def __init__(self, items):
+        self._items = items
+        self.batch_sizes = []
+
+    def iter_all(self, batch_size=500):
+        self.batch_sizes.append(batch_size)
+        for item in self._items:
+            yield item
+
+    def scan_vectors(self, offset=0, limit=100):
+        raise AssertionError("scan_vectors() must not be called when iter_all() exists")
+
+
+class _MisShapedIterAllBackendStore:
+    """Backend store whose ``iter_all`` attribute is not callable."""
+
+    iter_all = 42  # plain attribute, not a method
+
+    def __init__(self, items):
+        self._items = items
+
+    def scan_vectors(self, offset=0, limit=100):
+        return self._items[offset:offset + limit]
+
+
 class VectorStoreScanVectorsTests(unittest.TestCase):
     """VectorStore.scan_vectors() / iter_vectors() backend-agnostic accessors."""
 
@@ -192,6 +225,81 @@ class VectorStoreScanVectorsTests(unittest.TestCase):
         self.assertEqual(list(store.iter_vectors(batch_size=2)), [])
 
 
+# ---------------------------------------------------------------------------
+# VectorStore.iter_vectors() preference for a native iter_all()
+# ---------------------------------------------------------------------------
+
+class VectorStoreIterAllDispatchTests(unittest.TestCase):
+    """iter_vectors() prefers a backend's native iter_all() when present.
+
+    Cursor-based backends cannot implement scan_vectors(offset, limit)
+    honestly, so they expose iter_all() instead and iter_vectors() routes to
+    it rather than walking offsets.
+    """
+
+    def _persistent_store(self, backend_store, backend_name="qdrant"):
+        store = VectorStore(backend="inmemory", dimension=2)
+        store.backend = backend_name
+        store._backend_store = backend_store
+        return store
+
+    def test_iter_vectors_uses_iter_all_when_available(self):
+        items = [
+            {"id": "a", "vector": None, "metadata": {"n": 1}},
+            {"id": "b", "vector": None, "metadata": {"n": 2}},
+        ]
+        backend = _IterAllBackendStore(items)
+        store = self._persistent_store(backend)
+
+        self.assertEqual(list(store.iter_vectors(batch_size=7)), items)
+
+    def test_iter_vectors_forwards_batch_size_to_iter_all(self):
+        backend = _IterAllBackendStore([])
+        store = self._persistent_store(backend)
+
+        list(store.iter_vectors(batch_size=32))
+
+        self.assertEqual(backend.batch_sizes, [32])
+
+    def test_iter_vectors_falls_back_to_scan_vectors_without_iter_all(self):
+        items = [{"id": "a", "vector": None, "metadata": {}}]
+        store = self._persistent_store(_ScanningBackendStore(items))
+
+        self.assertEqual(list(store.iter_vectors(batch_size=2)), items)
+
+    def test_iter_vectors_falls_back_when_iter_all_not_callable(self):
+        # A mis-shaped adapter exposing a non-callable ``iter_all`` must not be
+        # invoked; the offset path still has to work. Mirrors the count()
+        # precedent in _MisShapedBackendStore.
+        items = [{"id": "a", "vector": None, "metadata": {}}]
+        store = self._persistent_store(_MisShapedIterAllBackendStore(items))
+
+        self.assertEqual(list(store.iter_vectors(batch_size=2)), items)
+
+    def test_iter_vectors_inmemory_ignores_iter_all(self):
+        store = VectorStore(backend="inmemory", dimension=2)
+        store.store_vectors([np.array([1.0, 0.0])], [{"type": "a"}])
+        store._backend_store = _IterAllBackendStore([{"id": "wrong"}])
+
+        collected = list(store.iter_vectors(batch_size=2))
+
+        self.assertEqual([item["metadata"] for item in collected], [{"type": "a"}])
+
+    def test_iter_vectors_propagates_iter_all_errors(self):
+        # A scan that silently yields nothing is indistinguishable from an
+        # empty source, which would let `store migrate` report success having
+        # copied nothing (issue #1083).
+        class _FailingIterAll:
+            def iter_all(self, batch_size=500):
+                raise ProcessingError("backend unreachable")
+                yield  # pragma: no cover - makes this a generator
+
+        store = self._persistent_store(_FailingIterAll())
+
+        with self.assertRaises(ProcessingError):
+            list(store.iter_vectors(batch_size=2))
+
+
 # ---------------------------------------------------------------------------
 # VectorManager tests — inmemory backend
 # ---------------------------------------------------------------------------

From 5e80ebd837b46831d321d201c0cc7d9a08b0a6fd Mon Sep 17 00:00:00 2001
From: Zohaib Hassnain <109234410+ZohaibHassan16@users.noreply.github.com>
Date: Mon, 31 Aug 2026 02:54:29 +0500
Subject: [PATCH 2/5] fix(vector_store): drop qdrant migrate wiring, keep
 iter_all only

VectorStore cannot actually migrate to or from qdrant yet. _init_backend_store constructs QdrantStore without connecting or selecting a collection, so reads raise a Collection not initialized error, and the facade store_vectors dispatches only to add/add_vectors while QdrantStore exposes insert_vectors, so writes raise NotImplementedError.

Both are pre-existing facade gaps that nothing had exposed, since migrate previously only allowed faiss/sqlite/pgvector. Adding qdrant to the allowlist claimed support that does not work end to end, so it is removed along with the dimension inference that only fires for backends missing a .dimension attribute. Tracked separately; this PR keeps just the iter_all primitive.
---
 semantica/cli.py           | 31 ++++--------
 tests/test_cli_commands.py | 96 +-------------------------------------
 2 files changed, 10 insertions(+), 117 deletions(-)

diff --git a/semantica/cli.py b/semantica/cli.py
index 957878f4..d05a00f1 100644
--- a/semantica/cli.py
+++ b/semantica/cli.py
@@ -5,7 +5,6 @@ This module provides the command-line interface for the Semantica framework,
 enabling users to interact with the framework via terminal commands.
 """
 
-import itertools
 import json
 import os
 import sys
@@ -3715,7 +3714,7 @@ def store_stats(cli_ctx: CLIContext, backend: str, fmt: str, local_json: bool) -
     _run_with_error_handling(_action)
 
 
-_MIGRATE_SUPPORTED_BACKENDS = {"faiss", "sqlite", "pgvector", "qdrant"}
+_MIGRATE_SUPPORTED_BACKENDS = {"faiss", "sqlite", "pgvector"}
 _MIGRATE_BATCH_SIZE = 500
 
 
@@ -3760,10 +3759,12 @@ def store_migrate(cli_ctx: CLIContext, from_backend: str, to_backend: str,
                   namespace: Optional[str], local_dry: bool, local_json: bool) -> None:
     """Migrate data between backends.
 
-    Wired up between faiss, sqlite, pgvector and qdrant. Migrating to or from
-    pinecone, milvus or weaviate still needs the export/reindex workaround
-    below, since each of those needs its own enumeration design (Pinecone
-    list, Milvus query_iterator, Weaviate cursor) that is not built yet.
+    Direct migration is only wired up between faiss, sqlite, and pgvector -
+    these are the backends whose storage contract supports paging through
+    every stored vector. Migrating to or from qdrant, pinecone, milvus, or
+    weaviate still needs the export/reindex workaround below, since each of
+    those needs its own enumeration design (Qdrant scroll, Pinecone list,
+    etc.) that hasn't been built yet.
 
     \b
     Example:
@@ -3803,18 +3804,7 @@ def store_migrate(cli_ctx: CLIContext, from_backend: str, to_backend: str,
         if source_index_path:
             source._backend_store.load_index(source_index_path)
 
-        # Pull the first record before building the destination: qdrant, milvus
-        # and weaviate expose no `dimension` attribute, so without this the
-        # destination falls back to the default 768 and rejects every insert.
-        # The data itself is the reliable source of truth.
-        source_iter = source.iter_vectors(batch_size=_MIGRATE_BATCH_SIZE)
-        first_item = next(source_iter, None)
-
         source_dimension = getattr(source._backend_store, "dimension", None)
-        if not source_dimension and first_item is not None:
-            first_vector = first_item.get("vector")
-            if first_vector is not None:
-                source_dimension = len(first_vector)
         if source_dimension and "dimension" not in dest_cfg:
             dest_cfg["dimension"] = source_dimension
 
@@ -3837,12 +3827,7 @@ def store_migrate(cli_ctx: CLIContext, from_backend: str, to_backend: str,
             metadata_batch.clear()
             ids_batch.clear()
 
-        items = (
-            source_iter
-            if first_item is None
-            else itertools.chain([first_item], source_iter)
-        )
-        for item in items:
+        for item in source.iter_vectors(batch_size=_MIGRATE_BATCH_SIZE):
             meta = dict(item.get("metadata") or {})
             if namespace and "namespace" not in meta:
                 meta["namespace"] = namespace
diff --git a/tests/test_cli_commands.py b/tests/test_cli_commands.py
index a90673c4..4a054be1 100644
--- a/tests/test_cli_commands.py
+++ b/tests/test_cli_commands.py
@@ -1360,11 +1360,10 @@ class TestStore:
         assert result.exit_code != 0
 
     def test_migrate_refuses_unsupported_backend_pair(self, runner):
-        # qdrant is wired up now, so weaviate stands in as the unsupported side.
         result = runner.invoke(cli_module.main, ["store", "migrate",
-                                      "--from", "faiss", "--to", "weaviate"])
+                                      "--from", "faiss", "--to", "qdrant"])
         assert result.exit_code != 0
-        assert "faiss, pgvector, qdrant, sqlite" in result.output
+        assert "faiss, pgvector, sqlite" in result.output
 
     def _fake_migrate_store_module(self, source_items, stored, dest_configs=None):
         class _FakeBackendStore:
@@ -1446,97 +1445,6 @@ class TestStore:
         assert result.exit_code != 0
         assert "index_path" in result.output
 
-    def _fake_dimensionless_store_module(self, source_items, dest_configs, stored):
-        """Fake VectorStore whose backend store has NO `dimension` attribute.
-
-        Matches qdrant/milvus/weaviate, none of which expose one.
-        """
-
-        class _DimensionlessBackendStore:
-            pass
-
-        class _FakeStore:
-            def __init__(self, backend, config=None, **kw):
-                self.backend = backend
-                self._config = config or {}
-                self._backend_store = _DimensionlessBackendStore()
-                dest_configs[backend] = dict(self._config)
-
-            def iter_vectors(self, batch_size=500):
-                if self.backend == "qdrant":
-                    yield from source_items
-                    return
-                return
-                yield  # pragma: no cover - makes this a generator
-
-            def store_vectors(self, vectors, metadata, ids=None):
-                for vec_id, vec in zip(ids, vectors):
-                    stored[vec_id] = vec
-
-        return _fake_module(VectorStore=_FakeStore)
-
-    def test_migrate_infers_dimension_from_first_vector(self, runner, monkeypatch):
-        source_items = [
-            {"id": "a", "vector": [0.1, 0.2, 0.3], "metadata": {}},
-            {"id": "b", "vector": [0.4, 0.5, 0.6], "metadata": {}},
-        ]
-        dest_configs, stored = {}, {}
-        fake_vs = self._fake_dimensionless_store_module(source_items, dest_configs, stored)
-        monkeypatch.setitem(__import__("sys").modules, "semantica.vector_store", fake_vs)
-
-        result = runner.invoke(cli_module.main, ["store", "migrate",
-                                      "--from", "qdrant", "--to", "sqlite", "--json"])
-        _ok(result)
-        assert dest_configs["sqlite"].get("dimension") == 3
-
-    def test_migrate_does_not_drop_the_peeked_first_record(self, runner, monkeypatch):
-        """The first record is consumed to infer dimension, so it must be
-        chained back into the migration loop rather than lost."""
-        source_items = [
-            {"id": "a", "vector": [0.1, 0.2, 0.3], "metadata": {}},
-            {"id": "b", "vector": [0.4, 0.5, 0.6], "metadata": {}},
-        ]
-        dest_configs, stored = {}, {}
-        fake_vs = self._fake_dimensionless_store_module(source_items, dest_configs, stored)
-        monkeypatch.setitem(__import__("sys").modules, "semantica.vector_store", fake_vs)
-
-        result = runner.invoke(cli_module.main, ["store", "migrate",
-                                      "--from", "qdrant", "--to", "sqlite", "--json"])
-        _ok(result)
-        assert _json_output(result)["migrated"] == 2
-        assert sorted(stored) == ["a", "b"]
-
-    def test_migrate_keeps_explicit_dest_dimension_over_inference(self, runner, monkeypatch):
-        source_items = [{"id": "a", "vector": [0.1, 0.2, 0.3], "metadata": {}}]
-        dest_configs, stored = {}, {}
-        fake_vs = self._fake_dimensionless_store_module(source_items, dest_configs, stored)
-        monkeypatch.setitem(__import__("sys").modules, "semantica.vector_store", fake_vs)
-        monkeypatch.setattr(
-            cli_module.Config, "to_dict",
-            lambda self: {"vector_store": {"qdrant": {}, "sqlite": {"dimension": 128}}},
-        )
-
-        result = runner.invoke(cli_module.main, ["store", "migrate",
-                                      "--from", "qdrant", "--to", "sqlite", "--json"])
-        _ok(result)
-        assert dest_configs["sqlite"]["dimension"] == 128
-
-    def test_migrate_qdrant_is_now_supported(self, runner, monkeypatch):
-        dest_configs, stored = {}, {}
-        fake_vs = self._fake_dimensionless_store_module([], dest_configs, stored)
-        monkeypatch.setitem(__import__("sys").modules, "semantica.vector_store", fake_vs)
-
-        result = runner.invoke(cli_module.main, ["store", "migrate",
-                                      "--from", "qdrant", "--to", "pgvector", "--json"])
-        _ok(result)
-        assert _json_output(result)["migrated"] == 0
-
-    def test_migrate_still_refuses_milvus(self, runner):
-        result = runner.invoke(cli_module.main, ["store", "migrate",
-                                      "--from", "qdrant", "--to", "milvus"])
-        assert result.exit_code != 0
-        assert "milvus" in result.output
-
     def test_migrate_faiss_dest_requires_index_path(self, runner, monkeypatch):
         fake_vs = _fake_module(VectorStore=lambda **kw: MagicMock())
         monkeypatch.setitem(__import__("sys").modules, "semantica.vector_store", fake_vs)

From f2e7b9ed75cfc4148d5046c5b705a037afc408cd Mon Sep 17 00:00:00 2001
From: Zohaib Hassnain <109234410+ZohaibHassan16@users.noreply.github.com>
Date: Mon, 31 Aug 2026 13:39:46 +0500
Subject: [PATCH 3/5] fix(vector_store): raise instead of truncating when a
 qdrant scan cannot advance

---
 semantica/vector_store/qdrant_store.py  | 41 +++++++++++++------------
 tests/vector_store/test_qdrant_store.py | 24 ++++++---------
 2 files changed, 31 insertions(+), 34 deletions(-)

diff --git a/semantica/vector_store/qdrant_store.py b/semantica/vector_store/qdrant_store.py
index 85ce7302..5b057ca5 100644
--- a/semantica/vector_store/qdrant_store.py
+++ b/semantica/vector_store/qdrant_store.py
@@ -606,17 +606,15 @@ class QdrantStore:
 
     def iter_all(self, batch_size: int = 500):
         """
-        Iterate over every stored point using Qdrant's native scroll cursor.
+        Iterate over every stored point using Qdrant's scroll cursor.
 
-        Qdrant paginates by point-ID cursor, not by row offset, so this is
-        exposed instead of scan_vectors(offset, limit). An integer passed to
-        scroll()'s offset is a point ID rather than a rank, so there is no way
-        to seek to "the Nth record" without walking from the start.
-        VectorStore.iter_vectors() prefers this method when it is present.
+        Paginates by point-ID cursor rather than row offset, which is why this
+        exists instead of scan_vectors(offset, limit). An integer offset is a
+        point ID, not a rank.
 
-        Assumes a single unnamed vector per point, matching how insert_vectors()
-        writes them and how get_vector() reads them back. Collections configured
-        with named or multi-vectors are not handled here.
+        Assumes a single unnamed vector per point, as insert_vectors() and
+        get_vector() already do. Named and multi-vector collections are not
+        handled.
 
         Args:
             batch_size: Points to request per scroll call
@@ -625,11 +623,8 @@ class QdrantStore:
             Result dicts with 'id', 'metadata', and 'vector', in scroll order
 
         Raises:
-            ProcessingError: If the collection or client is not initialized.
-                Errors are raised rather than swallowed because a scan that
-                silently yields nothing is indistinguishable from an empty
-                source, which would let a caller such as `store migrate`
-                report success having copied nothing (issue #1083).
+            ProcessingError: If the collection or client is not initialized, or
+                if the scan cannot advance past a full page.
         """
         if self.collection is None or self.client is None or not QDRANT_AVAILABLE:
             raise ProcessingError(
@@ -653,13 +648,21 @@ class QdrantStore:
                     "vector": np.array(rec.vector) if rec.vector is not None else None,
                 }
 
-            # The final page can carry records while already reporting no next
-            # cursor, so those records are yielded above before stopping here.
-            # Calling scroll() again with offset=None would restart from the
-            # beginning rather than continue past the end.
-            if next_offset is None or not records:
+            # A final page can carry records alongside a null cursor, so they
+            # are yielded above before stopping. Passing offset=None back to
+            # scroll() would restart from the beginning, not continue.
+            if next_offset is None:
                 return
 
+            # A cursor without records means the scan cannot advance, which is
+            # truncation rather than completion.
+            if not records:
+                raise ProcessingError(
+                    "Qdrant returned an empty page alongside a continuation "
+                    "cursor, so the scan cannot advance. Refusing to return a "
+                    "truncated scan."
+                )
+
     def delete_vectors(
         self, point_ids: List[Union[str, int]], **options
     ) -> Dict[str, Any]:
diff --git a/tests/vector_store/test_qdrant_store.py b/tests/vector_store/test_qdrant_store.py
index 89a9fee3..8b4ef049 100644
--- a/tests/vector_store/test_qdrant_store.py
+++ b/tests/vector_store/test_qdrant_store.py
@@ -35,7 +35,7 @@ def _store_with_scroll(*pages):
 
 @patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
 def test_iter_all_threads_cursor_across_pages():
-    """The next call must continue from the previous page's next_page_offset."""
+    """The next call continues from the previous page's cursor."""
     store = _store_with_scroll(
         ([_record(1), _record(2)], "cursor-1"),
         ([_record(3)], None),
@@ -53,11 +53,8 @@ def test_iter_all_threads_cursor_across_pages():
 
 @patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
 def test_iter_all_yields_final_page_that_reports_no_next_cursor():
-    """Qdrant can return records and a null cursor on the same page.
-
-    Those records must still be yielded. Treating a null cursor as "stop
-    before this page" would silently drop the tail of every scan.
-    """
+    """Records and a null cursor can arrive together; those records must still
+    be yielded or every scan loses its tail."""
     store = _store_with_scroll(([_record(1), _record(2)], None))
 
     result = list(store.iter_all(batch_size=10))
@@ -97,21 +94,18 @@ def test_iter_all_empty_collection_yields_nothing():
 
 
 @patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
-def test_iter_all_stops_on_empty_page_even_with_a_cursor():
-    """Defensive: an empty page ends the scan rather than looping forever."""
+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"))
 
-    assert list(store.iter_all()) == []
-    assert store.client.scroll.call_count == 1
+    with pytest.raises(ProcessingError, match="cannot advance"):
+        list(store.iter_all())
 
 
 @patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
 def test_iter_all_raises_when_collection_not_initialized():
-    """Must fail loudly, not yield nothing.
-
-    An empty scan is indistinguishable from an empty source, which would let
-    `store migrate` report success having copied nothing (issue #1083).
-    """
+    """Must fail loudly: an empty scan reads the same as an empty source."""
     store = QdrantStore()
 
     with pytest.raises(ProcessingError, match="Collection not initialized"):

From fcdad56893d9cb0a20184fa4a73de93044a6577e Mon Sep 17 00:00:00 2001
From: Zohaib Hassnain <109234410+ZohaibHassan16@users.noreply.github.com>
Date: Mon, 31 Aug 2026 13:42:24 +0500
Subject: [PATCH 4/5] making it clean

---
 semantica/vector_store/vector_store.py        |  9 +++-----
 .../test_vector_manager_persistent.py         | 21 ++++---------------
 2 files changed, 7 insertions(+), 23 deletions(-)

diff --git a/semantica/vector_store/vector_store.py b/semantica/vector_store/vector_store.py
index cd703eda..e504a455 100644
--- a/semantica/vector_store/vector_store.py
+++ b/semantica/vector_store/vector_store.py
@@ -867,12 +867,9 @@ class VectorStore:
         """
         Iterate over every stored vector, one page at a time.
 
-        Backends whose native pagination is cursor based (Qdrant, Pinecone,
-        Milvus, Weaviate) cannot honestly implement the positional
-        scan_vectors(offset, limit) contract, so they expose iter_all()
-        instead and it is preferred here when present. Backends with real
-        positional access (inmemory, FAISS, SQLite-vec, PgVector) fall
-        through to the offset loop below.
+        Cursor-based backends expose iter_all() because they cannot support a
+        positional offset; it takes precedence when present. Everything else
+        falls through to the scan_vectors() offset loop.
 
         Args:
             batch_size: Number of vectors to fetch per underlying call
diff --git a/tests/vector_store/test_vector_manager_persistent.py b/tests/vector_store/test_vector_manager_persistent.py
index f0586ad2..147fa669 100644
--- a/tests/vector_store/test_vector_manager_persistent.py
+++ b/tests/vector_store/test_vector_manager_persistent.py
@@ -140,11 +140,7 @@ class _NonScanningBackendStore:
 
 
 class _IterAllBackendStore:
-    """Fake cursor-based backend store exposing iter_all() but not scan_vectors().
-
-    Mirrors qdrant/pinecone/milvus/weaviate, which cannot honour a positional
-    offset and therefore expose native iteration instead.
-    """
+    """Fake cursor-based store: iter_all() only, no usable scan_vectors()."""
 
     def __init__(self, items):
         self._items = items
@@ -230,12 +226,7 @@ class VectorStoreScanVectorsTests(unittest.TestCase):
 # ---------------------------------------------------------------------------
 
 class VectorStoreIterAllDispatchTests(unittest.TestCase):
-    """iter_vectors() prefers a backend's native iter_all() when present.
-
-    Cursor-based backends cannot implement scan_vectors(offset, limit)
-    honestly, so they expose iter_all() instead and iter_vectors() routes to
-    it rather than walking offsets.
-    """
+    """iter_vectors() prefers a backend's native iter_all() when present."""
 
     def _persistent_store(self, backend_store, backend_name="qdrant"):
         store = VectorStore(backend="inmemory", dimension=2)
@@ -268,9 +259,7 @@ class VectorStoreIterAllDispatchTests(unittest.TestCase):
         self.assertEqual(list(store.iter_vectors(batch_size=2)), items)
 
     def test_iter_vectors_falls_back_when_iter_all_not_callable(self):
-        # A mis-shaped adapter exposing a non-callable ``iter_all`` must not be
-        # invoked; the offset path still has to work. Mirrors the count()
-        # precedent in _MisShapedBackendStore.
+        # Mirrors the count() precedent in _MisShapedBackendStore.
         items = [{"id": "a", "vector": None, "metadata": {}}]
         store = self._persistent_store(_MisShapedIterAllBackendStore(items))
 
@@ -286,9 +275,7 @@ class VectorStoreIterAllDispatchTests(unittest.TestCase):
         self.assertEqual([item["metadata"] for item in collected], [{"type": "a"}])
 
     def test_iter_vectors_propagates_iter_all_errors(self):
-        # A scan that silently yields nothing is indistinguishable from an
-        # empty source, which would let `store migrate` report success having
-        # copied nothing (issue #1083).
+        # Silently yielding nothing would read as an empty source (#1083).
         class _FailingIterAll:
             def iter_all(self, batch_size=500):
                 raise ProcessingError("backend unreachable")

From fa967983e613f909af8e7983bb777d61f1d9e2ce Mon Sep 17 00:00:00 2001
From: Zohaib Hassnain <109234410+ZohaibHassan16@users.noreply.github.com>
Date: Tue, 1 Sep 2026 22:49:55 +0500
Subject: [PATCH 5/5] 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  | 42 ++++++++++++-------------
 tests/vector_store/test_qdrant_store.py | 24 +++++++++++---
 2 files changed, 39 insertions(+), 27 deletions(-)

diff --git a/semantica/vector_store/qdrant_store.py b/semantica/vector_store/qdrant_store.py
index 5b057ca5..ac55f553 100644
--- a/semantica/vector_store/qdrant_store.py
+++ b/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
diff --git a/tests/vector_store/test_qdrant_store.py b/tests/vector_store/test_qdrant_store.py
index 8b4ef049..8958d8c1 100644
--- a/tests/vector_store/test_qdrant_store.py
+++ b/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())