fix(core): parallelize stale vector cleanup

Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/services/search_service.py

@@ -469,25 +469,31 @@ async def sync_entity_vectors_batch(
                 *(self._clear_entity_vectors(entity_id) for entity_id in opted_out_ids)
             )
 
-        repository_results: list[VectorSyncBatchResult] = []
-        if unknown_ids:
-            # Trigger: a caller passes entity IDs that were deleted after the batch was built.
-            # Why: repository sync still owns stale chunk cleanup for IDs with no source rows.
-            # Outcome: deleted entities do not silently keep orphaned vector rows forever.
-            repository_results.append(await self.repository.sync_entity_vectors_batch(unknown_ids))
-
         eligible_entity_ids = [
             entity_id
             for entity_id in entity_ids
             if entity_id in entities_by_id and entity_id not in opted_out_ids
         ]
-        if eligible_entity_ids:
-            repository_results.append(
-                await self.repository.sync_entity_vectors_batch(
-                    eligible_entity_ids,
-                    progress_callback=progress_callback,
-                )
+
+        cleanup_task = (
+            self.repository.sync_entity_vectors_batch(unknown_ids) if unknown_ids else None
+        )
+        eligible_task = (
+            self.repository.sync_entity_vectors_batch(
+                eligible_entity_ids,
+                progress_callback=progress_callback,
             )
+            if eligible_entity_ids
+            else None
+        )
+        repository_results = [
+            result
+            for result in await asyncio.gather(
+                cleanup_task if cleanup_task is not None else asyncio.sleep(0, result=None),
+                eligible_task if eligible_task is not None else asyncio.sleep(0, result=None),
+            )
+            if result is not None
+        ]
 
         if not repository_results:
             return VectorSyncBatchResult(
@@ -503,7 +509,9 @@ async def sync_entity_vectors_batch(
             entities_failed=sum(result.entities_failed for result in repository_results),
             entities_deferred=sum(result.entities_deferred for result in repository_results),
             entities_skipped=(
-                len(opted_out_ids) + sum(result.entities_skipped for result in repository_results)
+                len(opted_out_ids)
+                + sum(result.entities_skipped for result in repository_results)
+                - len(unknown_ids)
             ),
             failed_entity_ids=[
                 failed_entity_id

tests/services/test_semantic_search.py

@@ -273,6 +273,7 @@ async def test_semantic_vector_sync_batch_cleans_up_unknown_ids(search_service,
                 entities_total=1,
                 entities_synced=1,
                 entities_failed=0,
+                entities_skipped=1,
             ),
             VectorSyncBatchResult(
                 entities_total=1,
@@ -282,12 +283,19 @@ async def test_semantic_vector_sync_batch_cleans_up_unknown_ids(search_service,
         ]
     )
     monkeypatch.setattr(repository, "sync_entity_vectors_batch", sync_batch)
+    progress_callback = AsyncMock()
 
-    result = await search_service.sync_entity_vectors_batch([41, 42])
+    result = await search_service.sync_entity_vectors_batch([41, 42], progress_callback)
 
     assert sync_batch.await_count == 2
-    assert sync_batch.await_args_list[0].args[0] == [41]
-    assert sync_batch.await_args_list[1].args[0] == [42]
+    called_entity_ids = {tuple(call.args[0]) for call in sync_batch.await_args_list}
+    assert called_entity_ids == {(41,), (42,)}
+    progress_callback_calls = [
+        call for call in sync_batch.await_args_list if call.kwargs.get("progress_callback") is not None
+    ]
+    assert len(progress_callback_calls) == 1
+    assert progress_callback_calls[0].args[0] == [42]
+    assert progress_callback_calls[0].kwargs["progress_callback"] is progress_callback
     assert result.entities_total == 2
     assert result.entities_synced == 2
     assert result.entities_failed == 0