wip: checkpoint full reindex parity and sqlite vector fixes

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

Author: phernandez

src/basic_memory/cli/commands/db.py

@@ -26,7 +26,7 @@ class EmbeddingProgress:
     """Typed CLI progress payload for embedding backfills."""
 
     entity_id: int
-    index: int
+    completed: int
     total: int
 
 
@@ -147,20 +147,30 @@ def reindex(
         False, "--embeddings", "-e", help="Rebuild vector embeddings (requires semantic search)"
     ),
     search: bool = typer.Option(False, "--search", "-s", help="Rebuild full-text search index"),
+    full: bool = typer.Option(
+        False,
+        "--full",
+        help="Force a full filesystem scan and file reindex instead of the default incremental scan",
+    ),
     project: str = typer.Option(
         None, "--project", "-p", help="Reindex a specific project (default: all)"
     ),
 ):  # pragma: no cover
     """Rebuild search indexes and/or vector embeddings without dropping the database.
 
-    By default rebuilds everything (search + embeddings if semantic is enabled).
-    Use --search or --embeddings to rebuild only one.
+    By default runs incremental search + embeddings (if semantic search is enabled).
+    Use --full to bypass incremental scan optimization, rebuild all file-backed search rows,
+    and re-embed all eligible notes.
+    Use --search or --embeddings to rebuild only one side.
 
     Examples:
-        bm reindex                  # Rebuild everything
+        bm reindex                  # Incremental search + embeddings
+        bm reindex --full           # Full search + full re-embed
         bm reindex --embeddings     # Only rebuild vector embeddings
         bm reindex --search         # Only rebuild FTS index
-        bm reindex -p claw          # Reindex only the 'claw' project
+        bm reindex --full --search  # Full search only
+        bm reindex --full --embeddings  # Full re-embed only
+        bm reindex -p claw --full   # Full reindex for only the 'claw' project
     """
     # If neither flag is set, do both
     if not embeddings and not search:
@@ -179,10 +189,19 @@ def reindex(
         if not search:
             raise typer.Exit(0)
 
-    run_with_cleanup(_reindex(app_config, search=search, embeddings=embeddings, project=project))
+    run_with_cleanup(
+        _reindex(app_config, search=search, embeddings=embeddings, full=full, project=project)
+    )
 
 
-async def _reindex(app_config, search: bool, embeddings: bool, project: str | None):
+async def _reindex(
+    app_config,
+    *,
+    search: bool,
+    embeddings: bool,
+    full: bool,
+    project: str | None,
+):
     """Run reindex operations."""
     from basic_memory.repository import EntityRepository
     from basic_memory.repository.search_repository import create_search_repository
@@ -220,6 +239,10 @@ async def _reindex(app_config, search: bool, embeddings: bool, project: str | No
             console.print(f"\n[bold]Project: [cyan]{proj.name}[/cyan][/bold]")
 
             if search:
+                search_mode_label = "full scan" if full else "incremental scan"
+                console.print(
+                    f"  Rebuilding full-text search index ([cyan]{search_mode_label}[/cyan])..."
+                )
                 sync_service = await get_sync_service(proj)
                 sync_dir = Path(proj.path)
                 with Progress(
@@ -244,14 +267,19 @@ async def on_index_progress(update: IndexProgress) -> None:
                     await sync_service.sync(
                         sync_dir,
                         project_name=proj.name,
+                        force_full=full,
+                        sync_embeddings=False,
                         progress_callback=on_index_progress,
                     )
                     progress.update(task, completed=progress.tasks[task].total or 1)
 
                 console.print("  [green]✓[/green] Full-text search index rebuilt")
 
             if embeddings:
-                console.print("  Building vector embeddings...")
+                embedding_mode_label = "full rebuild" if full else "incremental sync"
+                console.print(
+                    f"  Building vector embeddings ([cyan]{embedding_mode_label}[/cyan])..."
+                )
                 entity_repository = EntityRepository(session_maker, project_id=proj.id)
                 search_repository = create_search_repository(
                     session_maker, project_id=proj.id, app_config=app_config
@@ -274,16 +302,23 @@ async def on_index_progress(update: IndexProgress) -> None:
                     def on_progress(entity_id, index, total):
                         embedding_progress = EmbeddingProgress(
                             entity_id=entity_id,
-                            index=index,
+                            completed=index,
                             total=total,
                         )
+                        # Trigger: repository progress now reports terminal entity completion.
+                        # Why: operators need to see finished embedding work rather than
+                        # entities merely entering prepare.
+                        # Outcome: the CLI bar advances steadily with real completed work.
                         progress.update(
                             task,
                             total=embedding_progress.total,
-                            completed=embedding_progress.index,
+                            completed=embedding_progress.completed,
                         )
 
-                    stats = await search_service.reindex_vectors(progress_callback=on_progress)
+                    stats = await search_service.reindex_vectors(
+                        progress_callback=on_progress,
+                        force_full=full,
+                    )
                     progress.update(task, completed=stats["total_entities"])
 
                 console.print(

src/basic_memory/repository/search_repository_base.py

@@ -814,6 +814,22 @@ async def _sync_entity_vectors_internal(
         failed_entity_ids: set[int] = set()
         deferred_entity_ids: set[int] = set()
         synced_entity_ids: set[int] = set()
+        completed_entities = 0
+
+        def emit_progress(entity_id: int) -> None:
+            """Report terminal entity progress to callers such as the CLI.
+
+            Trigger: an entity reaches a terminal state in this sync run.
+            Why: operators need progress based on completed work, not the moment
+            an entity merely enters prepare.
+            Outcome: the progress bar advances when an entity is done for this
+            run, whether it synced, skipped, deferred, or failed.
+            """
+            nonlocal completed_entities
+            if progress_callback is None:
+                return
+            completed_entities += 1
+            progress_callback(entity_id, completed_entities, total_entities)
 
         prepare_window_size = self._vector_prepare_window_size()
         with telemetry.started_span(
@@ -826,13 +842,6 @@ async def _sync_entity_vectors_internal(
             for window_start in range(0, total_entities, prepare_window_size):
                 window_entity_ids = entity_ids[window_start : window_start + prepare_window_size]
 
-                if progress_callback is not None:
-                    # Trigger: prepare runs in bounded windows instead of strict one-by-one order.
-                    # Why: callbacks still need deterministic per-entity positions before the window starts.
-                    # Outcome: progress advances in prepare_window_size bursts.
-                    for offset, entity_id in enumerate(window_entity_ids, start=window_start):
-                        progress_callback(entity_id, offset, total_entities)
-
                 prepared_window = await self._prepare_entity_vector_jobs_window(window_entity_ids)
 
                 for entity_id, prepared in zip(window_entity_ids, prepared_window, strict=True):
@@ -847,6 +856,7 @@ async def _sync_entity_vectors_internal(
                             entity_id=entity_id,
                             error=str(prepared),
                         )
+                        emit_progress(entity_id)
                         continue
 
                     embedding_jobs_count = len(prepared.embedding_jobs)
@@ -886,6 +896,7 @@ async def _sync_entity_vectors_internal(
                             shard_count=prepared.shard_count,
                             remaining_jobs_after_shard=prepared.remaining_jobs_after_shard,
                         )
+                        emit_progress(entity_id)
                         continue
 
                     entity_runtime[entity_id] = _EntitySyncRuntime(
@@ -933,6 +944,7 @@ async def _sync_entity_vectors_internal(
                                 entity_runtime=entity_runtime,
                                 synced_entity_ids=synced_entity_ids,
                                 deferred_entity_ids=deferred_entity_ids,
+                                progress_callback=emit_progress,
                             )
                         except Exception as exc:
                             if not continue_on_error:
@@ -952,6 +964,8 @@ async def _sync_entity_vectors_internal(
                                 chunk_count=len(flush_jobs),
                                 error=str(exc),
                             )
+                            for failed_entity_id in affected_entity_ids:
+                                emit_progress(failed_entity_id)
 
             if pending_jobs:
                 flush_jobs = list(pending_jobs)
@@ -968,6 +982,7 @@ async def _sync_entity_vectors_internal(
                         entity_runtime=entity_runtime,
                         synced_entity_ids=synced_entity_ids,
                         deferred_entity_ids=deferred_entity_ids,
+                        progress_callback=emit_progress,
                     )
                 except Exception as exc:
                     if not continue_on_error:
@@ -987,6 +1002,8 @@ async def _sync_entity_vectors_internal(
                         chunk_count=len(flush_jobs),
                         error=str(exc),
                     )
+                    for failed_entity_id in affected_entity_ids:
+                        emit_progress(failed_entity_id)
 
         # Trigger: this should never happen after all flushes succeed.
         # Why: remaining jobs mean runtime tracking drifted from queued jobs.
@@ -1002,6 +1019,8 @@ async def _sync_entity_vectors_internal(
                 project_id=self.project_id,
                 unfinished_entities=orphan_runtime_entities,
             )
+            for failed_entity_id in orphan_runtime_entities:
+                emit_progress(failed_entity_id)
 
         # Keep result counters aligned with successful/failed terminal states.
         synced_entity_ids.difference_update(failed_entity_ids)
@@ -1527,6 +1546,7 @@ def _finalize_completed_entity_syncs(
         entity_runtime: dict[int, _EntitySyncRuntime],
         synced_entity_ids: set[int],
         deferred_entity_ids: set[int],
+        progress_callback: Callable[[int], None] | None = None,
     ) -> float:
         """Finalize completed entities and return cumulative queue wait seconds."""
         queue_wait_seconds_total = 0.0
@@ -1570,6 +1590,8 @@ def _finalize_completed_entity_syncs(
                 remaining_jobs_after_shard=runtime.remaining_jobs_after_shard,
             )
             entity_runtime.pop(entity_id, None)
+            if progress_callback is not None:
+                progress_callback(entity_id)
 
         return queue_wait_seconds_total
 

src/basic_memory/repository/sqlite_search_repository.py

@@ -56,7 +56,8 @@ def __init__(
             self._app_config.semantic_embedding_sync_batch_size
         )
         self._embedding_provider = embedding_provider
-        self._sqlite_vec_lock = asyncio.Lock()
+        self._sqlite_vec_load_lock = asyncio.Lock()
+        self._sqlite_prepare_write_lock = asyncio.Lock()
         self._vector_tables_initialized = False
         self._vector_dimensions = 384
 
@@ -357,7 +358,13 @@ async def _ensure_sqlite_vec_loaded(self, session) -> None:
                 "pip install -U basic-memory"
             ) from exc
 
-        async with self._sqlite_vec_lock:
+        # Trigger: sqlite-vec must be loaded on each SQLite connection before
+        # vec tables and functions are visible.
+        # Why: extension loading is connection-local, so we need one narrow
+        # critical section to avoid racing two coroutines on the same step.
+        # Outcome: connection setup stays serialized without blocking unrelated
+        # prepare work behind the write-side lock.
+        async with self._sqlite_vec_load_lock:
             try:
                 await session.execute(text("SELECT vec_version()"))
                 return
@@ -558,6 +565,76 @@ async def _delete_stale_chunks(
             stale_params,
         )
 
+    async def delete_entity_vector_rows(self, entity_id: int) -> None:
+        """Delete one entity's vec rows on a sqlite-vec-enabled connection."""
+        await self._ensure_vector_tables()
+
+        async with db.scoped_session(self.session_maker) as session:
+            await self._ensure_sqlite_vec_loaded(session)
+
+            # Constraint: sqlite-vec virtual tables are only visible after vec0 is
+            # loaded on this exact connection.
+            # Why: generic repository sessions can reach search_vector_chunks but still
+            #      fail with "no such module: vec0" when touching embeddings.
+            # Outcome: service-level cleanup routes vec-table deletes through this helper.
+            await self._delete_entity_chunks(session, entity_id)
+            await session.commit()
+
+    async def delete_project_vector_rows(self) -> None:
+        """Delete all vector rows for this project on a sqlite-vec-enabled connection."""
+        await self._ensure_vector_tables()
+
+        async with db.scoped_session(self.session_maker) as session:
+            await self._ensure_sqlite_vec_loaded(session)
+
+            # Constraint: sqlite-vec stores embeddings separately with no cascade delete.
+            # Why: full rebuild must clear embeddings before chunk rows or stale vectors remain.
+            # Outcome: the next sync recreates the project's derived vectors from scratch.
+            await session.execute(
+                text(
+                    "DELETE FROM search_vector_embeddings WHERE rowid IN ("
+                    "SELECT id FROM search_vector_chunks WHERE project_id = :project_id)"
+                ),
+                {"project_id": self.project_id},
+            )
+            await session.execute(
+                text("DELETE FROM search_vector_chunks WHERE project_id = :project_id"),
+                {"project_id": self.project_id},
+            )
+            await session.commit()
+
+    async def delete_stale_vector_rows(self) -> None:
+        """Delete vector rows whose source entities no longer exist."""
+        await self._ensure_vector_tables()
+
+        async with db.scoped_session(self.session_maker) as session:
+            await self._ensure_sqlite_vec_loaded(session)
+
+            stale_entity_filter = (
+                "entity_id NOT IN (SELECT id FROM entity WHERE project_id = :project_id)"
+            )
+            params = {"project_id": self.project_id}
+
+            # Trigger: deleted entities left behind derived vector rows.
+            # Why: sqlite-vec does not provide cascade cleanup from our chunk table.
+            # Outcome: stale vector state disappears before coverage stats or reindex runs.
+            await session.execute(
+                text(
+                    "DELETE FROM search_vector_embeddings WHERE rowid IN ("
+                    "SELECT id FROM search_vector_chunks "
+                    f"WHERE project_id = :project_id AND {stale_entity_filter})"
+                ),
+                params,
+            )
+            await session.execute(
+                text(
+                    "DELETE FROM search_vector_chunks "
+                    f"WHERE project_id = :project_id AND {stale_entity_filter}"
+                ),
+                params,
+            )
+            await session.commit()
+
     def _distance_to_similarity(self, distance: float) -> float:
         """Convert L2 distance to cosine similarity for normalized embeddings.
 
@@ -569,7 +646,12 @@ def _distance_to_similarity(self, distance: float) -> float:
     @asynccontextmanager
     async def _prepare_entity_write_scope(self):
         """SQLite keeps the shared read window, but funnels prepare writes through one lock."""
-        async with self._sqlite_vec_lock:
+        # Trigger: the shared prepare window fans out per entity after batched reads.
+        # Why: SQLite still benefits from shared reads, but write transactions do
+        # not get meaningfully faster when we open many at once.
+        # Outcome: one entity at a time mutates chunk rows, while vec extension
+        # loading uses its own separate lock and cannot deadlock this path.
+        async with self._sqlite_prepare_write_lock:
             yield
 
     def _prepare_window_existing_rows_sql(self, placeholders: str) -> str: