fix(core): batch SqliteVecIndex.delete under SQLite expression-depth limit

pstack-evidence.md

@@ -0,0 +1,9 @@
+# pstack evidence — basic-memory #1850
+
+## Principles applied
+- Laziness Protocol: batch the existing OR-chain delete; no API change.
+- Prove It Works: regression deletes 1200 pending chunks and asserts both tables empty.
+- Make Operations Idempotent: batches preserve pending-only match semantics.
+
+## Repro → fix
+Unbatched OR of 4-term ANDs exceeds SQLITE_MAX_EXPR_DEPTH (~995). `_DELETE_BATCH_SIZE = 200` splits UPDATE/DELETE work; one commit after all batches.

src/basic_memory/repository/sqlite_vec_index.py

@@ -252,45 +252,58 @@ async def upsert(self, project_id: int, records: Sequence[VectorRecord]) -> None
             )
             await session.commit()
 
+    # SQLite caps expression-tree depth at SQLITE_MAX_EXPR_DEPTH (1000). Each
+    # pending-row predicate is a 4-term AND under an OR chain, so a single
+    # statement fails above ~995 deletions (#1850). Batch below that ceiling.
+    _DELETE_BATCH_SIZE = 200
+
     async def delete(self, project_id: int, records: Sequence[VectorDeletion]) -> None:
         if not records:
             return
         await self.initialize()
         async with db.scoped_session(self._session_maker) as session:
             await self._ensure_loaded(session)
-            params: dict[str, object] = {"project_id": project_id}
-            predicates: list[str] = []
-            for index, record in enumerate(records):
-                params[f"entity_id_{index}"] = record.key.entity_id
-                params[f"chunk_key_{index}"] = record.key.chunk_key
-                params[f"source_hash_{index}"] = record.source_hash
-                predicates.append(
-                    f"(entity_id = :entity_id_{index} "
-                    f"AND chunk_key = :chunk_key_{index} "
-                    f"AND source_hash = :source_hash_{index} "
-                    "AND embedding_status = 'pending')"
-                )
-            result = await session.execute(
-                text(
-                    "UPDATE search_vector_chunks SET source_hash = source_hash "
-                    "WHERE project_id = :project_id AND ("
-                    + " OR ".join(predicates)
-                    + ") RETURNING id"
-                ),
-                params,
-            )
-            rowids = [int(row_id) for row_id in result.scalars().all()]
-            if rowids:
-                params = {f"rowid_{index}": rowid for index, rowid in enumerate(rowids)}
-                placeholders = ", ".join(f":rowid_{index}" for index in range(len(rowids)))
-                await session.execute(
-                    text(f"DELETE FROM search_vector_embeddings WHERE rowid IN ({placeholders})"),
-                    params,
-                )
-                await session.execute(
-                    text(f"DELETE FROM search_vector_chunks WHERE id IN ({placeholders})"),
+            all_rowids: list[int] = []
+            for start in range(0, len(records), self._DELETE_BATCH_SIZE):
+                batch = records[start : start + self._DELETE_BATCH_SIZE]
+                params: dict[str, object] = {"project_id": project_id}
+                predicates: list[str] = []
+                for index, record in enumerate(batch):
+                    params[f"entity_id_{index}"] = record.key.entity_id
+                    params[f"chunk_key_{index}"] = record.key.chunk_key
+                    params[f"source_hash_{index}"] = record.source_hash
+                    predicates.append(
+                        f"(entity_id = :entity_id_{index} "
+                        f"AND chunk_key = :chunk_key_{index} "
+                        f"AND source_hash = :source_hash_{index} "
+                        "AND embedding_status = 'pending')"
+                    )
+                result = await session.execute(
+                    text(
+                        "UPDATE search_vector_chunks SET source_hash = source_hash "
+                        "WHERE project_id = :project_id AND ("
+                        + " OR ".join(predicates)
+                        + ") RETURNING id"
+                    ),
                     params,
                 )
+                all_rowids.extend(int(row_id) for row_id in result.scalars().all())
+            if all_rowids:
+                # IN-lists are shallow; still batch to keep statement size bounded.
+                for start in range(0, len(all_rowids), self._DELETE_BATCH_SIZE):
+                    batch_ids = all_rowids[start : start + self._DELETE_BATCH_SIZE]
+                    params = {f"rowid_{index}": rowid for index, rowid in enumerate(batch_ids)}
+                    placeholders = ", ".join(f":rowid_{index}" for index in range(len(batch_ids)))
+                    await session.execute(
+                        text(
+                            f"DELETE FROM search_vector_embeddings WHERE rowid IN ({placeholders})"
+                        ),
+                        params,
+                    )
+                    await session.execute(
+                        text(f"DELETE FROM search_vector_chunks WHERE id IN ({placeholders})"),
+                        params,
+                    )
                 await session.commit()
 
     async def delete_entity(self, project_id: int, entity_id: int) -> None:

tests/repository/test_sqlite_vector_search_repository.py

@@ -1087,6 +1087,78 @@ async def test_sqlite_vec_delete_requires_pending_source_generation(search_repos
 
 
 @pytest.mark.asyncio
+async def test_sqlite_vec_delete_batches_above_expression_depth_limit(search_repository):
+    """Deleting more than ~995 pending chunks must not hit SQLITE_MAX_EXPR_DEPTH (#1850)."""
+    if not isinstance(search_repository, SQLiteSearchRepository):
+        pytest.skip("sqlite-vec deletion behavior is local SQLite-only.")
+
+    _enable_semantic(search_repository)
+    await search_repository.init_search_index()
+    index = cast(SQLiteVecIndex, search_repository._semantic_vector_index)
+    project_id = search_repository.project_id
+    count = 1200
+    deletions: list[VectorDeletion] = []
+
+    async with db.scoped_session(search_repository.session_maker) as session:
+        await index._ensure_loaded(session)
+        for i in range(count):
+            chunk_key = f"entity:1850:{i}"
+            row_id = 1850000 + i
+            await session.execute(
+                text(
+                    "INSERT INTO search_vector_chunks ("
+                    "id, entity_id, project_id, chunk_key, source_type, source_row_id, "
+                    "chunk_index, chunk_text, source_hash, entity_fingerprint, "
+                    "embedding_model, vector_index, embedding_status"
+                    ") VALUES ("
+                    ":id, 1850, :project_id, :chunk_key, 'entity', 1850, :chunk_index, "
+                    "'text', 'hash-1850', 'fingerprint', :embedding_model, "
+                    "'sqlite-vec', 'pending')"
+                ),
+                {
+                    "id": row_id,
+                    "project_id": project_id,
+                    "chunk_key": chunk_key,
+                    "chunk_index": i,
+                    "embedding_model": search_repository._embedding_model_key(),
+                },
+            )
+            await session.execute(
+                text(
+                    "INSERT INTO search_vector_embeddings (rowid, embedding, source_hash) "
+                    "VALUES (:rowid, :embedding, 'hash-1850')"
+                ),
+                {"rowid": row_id, "embedding": "[1.0, 0.0, 0.0, 0.0]"},
+            )
+            deletions.append(
+                VectorDeletion(
+                    key=VectorKey(entity_id=1850, chunk_key=chunk_key),
+                    source_hash="hash-1850",
+                )
+            )
+        await session.commit()
+
+    await index.delete(project_id, deletions)
+
+    async with db.scoped_session(search_repository.session_maker) as session:
+        remaining_chunks = await session.scalar(
+            text(
+                "SELECT COUNT(*) FROM search_vector_chunks "
+                "WHERE project_id = :project_id AND entity_id = 1850"
+            ),
+            {"project_id": project_id},
+        )
+        remaining_embeddings = await session.scalar(
+            text(
+                "SELECT COUNT(*) FROM search_vector_embeddings "
+                "WHERE rowid BETWEEN :lo AND :hi"
+            ),
+            {"lo": 1850000, "hi": 1850000 + count - 1},
+        )
+    assert remaining_chunks == 0
+    assert remaining_embeddings == 0
+
+
 async def test_sqlite_chunk_upsert_and_delete_lifecycle(search_repository):
     """sync_entity_vectors updates changed chunks and clears vectors when source rows disappear."""
     if not isinstance(search_repository, SQLiteSearchRepository):