From ecc85d2c2db7b05a8f0a82fb427f679a0e4e53b9 Mon Sep 17 00:00:00 2001 From: phernandez Date: Tue, 11 Aug 2026 18:51:24 -0500 Subject: [PATCH 1/2] fix(cli): surface reindex embedding failures and index identity Signed-off-by: phernandez --- src/basic_memory/cli/commands/db.py | 30 ++- .../repository/semantic_vector_sync.py | 35 ++- src/basic_memory/runtime/vector_sync.py | 5 + src/basic_memory/services/search_service.py | 55 +++-- tests/cli/test_db_reindex.py | 205 +++++++++++++++++- tests/repository/test_semantic_vector_sync.py | 40 ++++ tests/services/test_semantic_search.py | 65 +++++- 7 files changed, 402 insertions(+), 33 deletions(-) diff --git a/src/basic_memory/cli/commands/db.py b/src/basic_memory/cli/commands/db.py index e5e18db64..3ccba1f57 100644 --- a/src/basic_memory/cli/commands/db.py +++ b/src/basic_memory/cli/commands/db.py @@ -13,6 +13,7 @@ import typer from loguru import logger from rich.console import Console +from rich.markup import escape from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskProgressColumn from basic_memory.cli.app import app @@ -20,6 +21,15 @@ from basic_memory.config import ConfigManager, ProjectMode console = Console() +REINDEX_ERROR_SUMMARY_MAX_LENGTH = 240 + + +def _reindex_error_summary(message: str) -> str: + """Collapse one error to a bounded single-line CLI summary.""" + single_line = " ".join(message.split()) + if len(single_line) <= REINDEX_ERROR_SUMMARY_MAX_LENGTH: + return single_line + return f"{single_line[: REINDEX_ERROR_SUMMARY_MAX_LENGTH - 3].rstrip()}..." def _is_basic_memory_mcp(cmdline: list[str]) -> bool: @@ -364,6 +374,8 @@ async def _reindex( console.print(f"[red]Project '{project}' not found.[/red]") raise typer.Exit(1) + embedding_entities_total = 0 + embedding_errors_total = 0 for proj in projects: console.print(f"\n[bold]Project: [cyan]{proj.name}[/cyan][/bold]") @@ -455,11 +467,20 @@ def on_progress(entity_id, index, total): progress.update(task, completed=stats["total_entities"]) console.print( - f" [green]done[/green] Embeddings complete: " + " [green]done[/green] Embeddings complete " + f"([cyan]index={escape(stats['vector_index'])}[/cyan], " + f"[cyan]model={escape(stats['embedding_model'])}[/cyan]): " f"{stats['embedded']} entities embedded, " f"{stats['skipped']} skipped, " f"{stats['errors']} errors" ) + if stats["sample_errors"]: + console.print( + " [yellow]Representative error:[/yellow] " + f"{escape(_reindex_error_summary(stats['sample_errors'][0]))}" + ) + embedding_entities_total += stats["total_entities"] + embedding_errors_total += stats["errors"] if stats["total_entities"] == 0 and not search: # Trigger: embeddings-only mode found no database entities. # Why: this mode rebuilds derived vectors; it does not discover files. @@ -471,6 +492,13 @@ def on_progress(entity_id, index, total): "or start the MCP server and retry after its initial index completes." ) + # Trigger: every entity attempted across the selected projects failed to embed. + # Why: requested search work and other project summaries must still finish first. + # Outcome: the command preserves useful output but no longer reports false success. + if embedding_entities_total > 0 and embedding_errors_total == embedding_entities_total: + console.print("\n[red]Reindex failed: all vector embedding attempts failed.[/red]") + raise typer.Exit(code=1) + console.print("\n[green]Reindex complete![/green]") finally: await db.shutdown_db() diff --git a/src/basic_memory/repository/semantic_vector_sync.py b/src/basic_memory/repository/semantic_vector_sync.py index c2297d0e6..63c88cfa0 100644 --- a/src/basic_memory/repository/semantic_vector_sync.py +++ b/src/basic_memory/repository/semantic_vector_sync.py @@ -14,7 +14,10 @@ from basic_memory import db from basic_memory.repository.semantic_chunking import VectorChunkRecord -from basic_memory.runtime.vector_sync import VectorSyncBatchResult +from basic_memory.runtime.vector_sync import ( + VECTOR_SYNC_SAMPLE_ERROR_LIMIT, + VectorSyncBatchResult, +) from basic_memory.schemas.search import SearchItemType if TYPE_CHECKING: # pragma: no cover - import cycle exists only for static analysis @@ -38,6 +41,19 @@ class _VectorSyncBatchAccumulator: queue_wait_seconds_total: float = 0.0 embed_seconds_total: float = 0.0 write_seconds_total: float = 0.0 + sample_errors: list[str] = field(default_factory=list) + + def record_error(self, error: BaseException) -> None: + """Retain a small, distinct sample without flooding callers.""" + message = " ".join(str(error).split()) + if not message: + message = type(error).__name__ + if ( + message in self.sample_errors + or len(self.sample_errors) >= VECTOR_SYNC_SAMPLE_ERROR_LIMIT + ): + return + self.sample_errors.append(message) def freeze( self, @@ -46,6 +62,8 @@ def freeze( synced_entity_ids: set[int], deferred_entity_ids: set[int], failed_entity_ids: set[int], + vector_index: str, + embedding_model: str, ) -> VectorSyncBatchResult: """Return the immutable result after entity terminal states settle.""" return VectorSyncBatchResult( @@ -55,6 +73,9 @@ def freeze( entities_deferred=len(deferred_entity_ids), entities_skipped=self.entities_skipped, failed_entity_ids=tuple(sorted(failed_entity_ids)), + sample_errors=tuple(self.sample_errors), + vector_index=vector_index, + embedding_model=embedding_model, chunks_total=self.chunks_total, chunks_skipped=self.chunks_skipped, embedding_jobs_total=self.embedding_jobs_total, @@ -274,11 +295,15 @@ async def sync_entity_vectors_internal( assert repository._embedding_provider is not None total_entities = len(entity_ids) + vector_index = repository._semantic_vector_index_name + embedding_model = repository._embedding_model_key() if total_entities == 0: return VectorSyncBatchResult( entities_total=0, entities_synced=0, entities_failed=0, + vector_index=vector_index, + embedding_model=embedding_model, ) batch_start = time.perf_counter() backend_name = type(repository).__name__.removesuffix("SearchRepository").lower() @@ -336,6 +361,7 @@ def emit_progress(entity_id: int) -> None: if not continue_on_error: raise prepared failed_entity_ids.add(entity_id) + batch_counters.record_error(prepared) logger.warning( "Vector batch sync entity prepare failed: project_id={project_id} " "entity_id={entity_id} error={error}", @@ -431,6 +457,7 @@ def emit_progress(entity_id: int) -> None: except Exception as exc: if not continue_on_error: raise + batch_counters.record_error(exc) affected_entity_ids = sorted({job.entity_id for job in flush_jobs}) failed_entity_ids.update(affected_entity_ids) synced_entity_ids.difference_update(affected_entity_ids) @@ -471,6 +498,7 @@ def emit_progress(entity_id: int) -> None: except Exception as exc: if not continue_on_error: raise + batch_counters.record_error(exc) affected_entity_ids = sorted({job.entity_id for job in flush_jobs}) failed_entity_ids.update(affected_entity_ids) synced_entity_ids.difference_update(affected_entity_ids) @@ -494,6 +522,9 @@ def emit_progress(entity_id: int) -> None: # Outcome: fail-safe marks these entities as failed to avoid false positives. if entity_runtime: orphan_runtime_entities = sorted(entity_runtime.keys()) + batch_counters.record_error( + RuntimeError("Vector sync left unfinished entities after flushes.") + ) failed_entity_ids.update(orphan_runtime_entities) synced_entity_ids.difference_update(orphan_runtime_entities) deferred_entity_ids.difference_update(orphan_runtime_entities) @@ -514,6 +545,8 @@ def emit_progress(entity_id: int) -> None: synced_entity_ids=synced_entity_ids, deferred_entity_ids=deferred_entity_ids, failed_entity_ids=failed_entity_ids, + vector_index=vector_index, + embedding_model=embedding_model, ) logger.info( diff --git a/src/basic_memory/runtime/vector_sync.py b/src/basic_memory/runtime/vector_sync.py index 326ae447c..7bae86e5c 100644 --- a/src/basic_memory/runtime/vector_sync.py +++ b/src/basic_memory/runtime/vector_sync.py @@ -5,6 +5,8 @@ type EntityId = int +VECTOR_SYNC_SAMPLE_ERROR_LIMIT = 3 + @dataclass(frozen=True, slots=True) class VectorSyncBatchResult: @@ -16,6 +18,9 @@ class VectorSyncBatchResult: entities_deferred: int = 0 entities_skipped: int = 0 failed_entity_ids: tuple[EntityId, ...] = () + sample_errors: tuple[str, ...] = () + vector_index: str = "" + embedding_model: str = "" chunks_total: int = 0 chunks_skipped: int = 0 embedding_jobs_total: int = 0 diff --git a/src/basic_memory/services/search_service.py b/src/basic_memory/services/search_service.py index 5f1df14ef..fdba54991 100644 --- a/src/basic_memory/services/search_service.py +++ b/src/basic_memory/services/search_service.py @@ -26,7 +26,10 @@ from basic_memory.repository.search_query import relaxed_query_words from basic_memory.schemas.base import normalize_note_type from basic_memory.schemas.search import SearchQuery, SearchItemType, SearchRetrievalMode -from basic_memory.runtime.vector_sync import VectorSyncBatchResult +from basic_memory.runtime.vector_sync import ( + VECTOR_SYNC_SAMPLE_ERROR_LIMIT, + VectorSyncBatchResult, +) from basic_memory.services import FileService # Maximum size for content_stems field to stay under Postgres's 8KB index row limit. @@ -516,11 +519,7 @@ async def sync_entity_vectors_batch( ) -> VectorSyncBatchResult: """Refresh vector chunks for a batch of entities.""" if not entity_ids: - return VectorSyncBatchResult( - entities_total=0, - entities_synced=0, - entities_failed=0, - ) + return await self.repository.sync_entity_vectors_batch([]) async with db.scoped_session(self.session_maker) as session: entities_by_id = { @@ -550,31 +549,19 @@ async def sync_entity_vectors_batch( 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 + eligible_task = self.repository.sync_entity_vectors_batch( + eligible_entity_ids, + progress_callback=progress_callback, ) 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), + eligible_task, ) if result is not None ] - if not repository_results: - return VectorSyncBatchResult( - entities_total=len(entity_ids), - entities_synced=0, - entities_failed=0, - entities_skipped=len(opted_out_ids), - ) - batch_result = VectorSyncBatchResult( entities_total=len(entity_ids), entities_synced=sum(result.entities_synced for result in repository_results), @@ -590,6 +577,25 @@ async def sync_entity_vectors_batch( for result in repository_results for failed_entity_id in result.failed_entity_ids ), + sample_errors=tuple( + dict.fromkeys( + error + for result in repository_results + for error in result.sample_errors + ) + )[:VECTOR_SYNC_SAMPLE_ERROR_LIMIT], + vector_index=next( + (result.vector_index for result in repository_results if result.vector_index), + "", + ), + embedding_model=next( + ( + result.embedding_model + for result in repository_results + if result.embedding_model + ), + "", + ), chunks_total=sum(result.chunks_total for result in repository_results), chunks_skipped=sum(result.chunks_skipped for result in repository_results), embedding_jobs_total=sum(result.embedding_jobs_total for result in repository_results), @@ -616,7 +622,7 @@ async def reindex_vectors( eligible entity re-embeds from scratch. Returns: - dict with stats: total_entities, embedded, skipped, errors + dict with counts, sampled errors, and the active vector index/model identity """ async with db.scoped_session(self.session_maker) as session: entities = await self.entity_repository.find_all(session) @@ -638,6 +644,9 @@ async def reindex_vectors( "embedded": batch_result.entities_synced, "skipped": batch_result.entities_skipped, "errors": batch_result.entities_failed, + "sample_errors": batch_result.sample_errors, + "vector_index": batch_result.vector_index, + "embedding_model": batch_result.embedding_model, } for failed_entity_id in batch_result.failed_entity_ids: diff --git a/tests/cli/test_db_reindex.py b/tests/cli/test_db_reindex.py index d679d11f5..caca24e29 100644 --- a/tests/cli/test_db_reindex.py +++ b/tests/cli/test_db_reindex.py @@ -1,6 +1,7 @@ """Tests for `bm reindex` CLI wiring.""" import asyncio +from collections.abc import Mapping from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock @@ -16,6 +17,25 @@ runner = CliRunner() +def _vector_stats( + *, + total_entities: int, + embedded: int, + skipped: int, + errors: int, + sample_errors: tuple[str, ...] = (), +) -> dict[str, object]: + return { + "total_entities": total_entities, + "embedded": embedded, + "skipped": skipped, + "errors": errors, + "sample_errors": sample_errors, + "vector_index": "milvus", + "embedding_model": "FastEmbedEmbeddingProvider:BAAI/bge-small-en-v1.5", + } + + def _stub_app_config(*, semantic_search_enabled: bool = True) -> SimpleNamespace: """Build the minimal config surface the CLI reindex path expects.""" return SimpleNamespace( @@ -44,6 +64,98 @@ def _configure_reindex_cli(monkeypatch, app_config: SimpleNamespace) -> None: ) +def _configure_embedding_runtime( + monkeypatch, + session_maker, + stats: Mapping[str, object], +) -> tuple[SimpleNamespace, AsyncMock, list[str]]: + """Install the runtime boundaries needed to exercise the real reindex command.""" + app_config = _stub_app_config() + project = SimpleNamespace(id=1, name="foo", path="/tmp/foo") + printed_lines: list[str] = [] + project_index = AsyncMock( + return_value=SimpleNamespace( + total_files=1, + enqueued_files=1, + enqueued_batches=1, + deleted_files=0, + ) + ) + + class StubProjectRepository: + async def get_active_projects(self, session): + return [project] + + class StubSearchService: + def __init__(self, *args, **kwargs) -> None: + pass + + async def reindex_vectors(self, *, progress_callback=None, force_full: bool = False): + return dict(stats) + + class SilentProgress: + def __init__(self, *args, **kwargs) -> None: + pass + + def __enter__(self): + return self + + def __exit__(self, *args) -> bool: + return False + + def add_task(self, *args, **kwargs) -> int: + return 1 + + def update(self, *args, **kwargs) -> None: + pass + + _configure_reindex_cli(monkeypatch, app_config) + monkeypatch.setattr(db_cmd, "run_with_cleanup", lambda coro: asyncio.run(coro)) + monkeypatch.setattr( + "basic_memory.services.initialization.reconcile_projects_with_config", + AsyncMock(), + ) + monkeypatch.setattr( + "basic_memory.services.initialization.recover_project_materializations", + AsyncMock(), + ) + monkeypatch.setattr( + "basic_memory.db.get_or_create_db", + AsyncMock(return_value=(None, session_maker)), + ) + monkeypatch.setattr("basic_memory.db.shutdown_db", AsyncMock()) + monkeypatch.setattr("basic_memory.repository.ProjectRepository", StubProjectRepository) + monkeypatch.setattr( + "basic_memory.index.local_project.run_local_project_index_for_project", + project_index, + ) + monkeypatch.setattr( + "basic_memory.repository.search_repository.create_search_repository", + lambda *args, **kwargs: object(), + ) + monkeypatch.setattr("basic_memory.repository.EntityRepository", lambda *a, **k: object()) + monkeypatch.setattr( + "basic_memory.markdown.entity_parser.EntityParser", + lambda *a, **k: object(), + ) + monkeypatch.setattr( + "basic_memory.markdown.markdown_processor.MarkdownProcessor", + lambda *a, **k: object(), + ) + monkeypatch.setattr( + "basic_memory.services.file_service.FileService", + lambda *a, **k: object(), + ) + monkeypatch.setattr("basic_memory.services.search_service.SearchService", StubSearchService) + monkeypatch.setattr(db_cmd, "Progress", SilentProgress) + monkeypatch.setattr( + db_cmd.console, + "print", + lambda message="", *args, **kwargs: printed_lines.append(str(message)), + ) + return app_config, project_index, printed_lines + + def test_reindex_defaults_to_incremental_search_and_embeddings(monkeypatch): app_config = _stub_app_config() _configure_reindex_cli(monkeypatch, app_config) @@ -251,7 +363,7 @@ async def reindex_vectors(self, *, progress_callback=None, force_full: bool = Fa "force_full": force_full, } ) - return {"total_entities": 2, "embedded": 2, "skipped": 0, "errors": 0} + return _vector_stats(total_entities=2, embedded=2, skipped=0, errors=0) class SilentProgress: def __init__(self, *args, **kwargs): @@ -341,7 +453,7 @@ def __init__(self, search_repository, entity_repository, file_service, *, sessio pass async def reindex_vectors(self, *, progress_callback=None, force_full: bool = False): - return {"total_entities": 0, "embedded": 0, "skipped": 0, "errors": 0} + return _vector_stats(total_entities=0, embedded=0, skipped=0, errors=0) class SilentProgress: def __init__(self, *args, **kwargs): @@ -483,7 +595,7 @@ def __init__(self, *args, **kwargs) -> None: async def reindex_vectors(self, *, progress_callback=None, force_full: bool = False): vector_reindex_calls.append({"force_full": force_full}) - return {"total_entities": 1, "embedded": 1, "skipped": 0, "errors": 0} + return _vector_stats(total_entities=1, embedded=1, skipped=0, errors=0) class SilentProgress: def __init__(self, *args, **kwargs) -> None: @@ -537,3 +649,90 @@ def update(self, *args, **kwargs) -> None: assert index_call.kwargs["embeddings"] is False assert len(vector_reindex_calls) == 1 assert vector_reindex_calls[0]["force_full"] is True + + +def test_reindex_total_embedding_failure_surfaces_error_and_exits_one( + monkeypatch, + session_maker, +): + representative_error = "first line\nsecond line " + ("x" * 300) + stats = _vector_stats( + total_entities=3, + embedded=0, + skipped=0, + errors=3, + sample_errors=(representative_error,), + ) + _app_config, project_index, printed_lines = _configure_embedding_runtime( + monkeypatch, + session_maker, + stats, + ) + + result = runner.invoke(app, ["reindex"]) + + assert result.exit_code == 1 + project_index.assert_awaited_once() + output = "\n".join(printed_lines) + assert ( + "Embeddings complete ([cyan]index=milvus[/cyan], " + "[cyan]model=FastEmbedEmbeddingProvider:BAAI/bge-small-en-v1.5[/cyan]): " + "0 entities embedded, 0 skipped, 3 errors" + ) in output + representative_line = next( + line for line in printed_lines if "Representative error:" in line + ) + assert "first line second line" in representative_line + assert representative_line.endswith("...") + assert "Reindex failed: all vector embedding attempts failed." in output + assert "Reindex complete!" not in output + + +def test_reindex_partial_embedding_failure_surfaces_error_and_exits_zero( + monkeypatch, + session_maker, +): + stats = _vector_stats( + total_entities=3, + embedded=2, + skipped=0, + errors=1, + sample_errors=("Milvus is unavailable",), + ) + _app_config, _project_index, printed_lines = _configure_embedding_runtime( + monkeypatch, + session_maker, + stats, + ) + + result = runner.invoke(app, ["reindex", "--embeddings"]) + + assert result.exit_code == 0 + output = "\n".join(printed_lines) + assert "2 entities embedded, 0 skipped, 1 errors" in output + assert "Representative error:[/yellow] Milvus is unavailable" in output + assert "Reindex complete!" in output + + +def test_reindex_embedding_success_reports_index_and_model_and_exits_zero( + monkeypatch, + session_maker, +): + stats = _vector_stats(total_entities=2, embedded=2, skipped=0, errors=0) + _app_config, _project_index, printed_lines = _configure_embedding_runtime( + monkeypatch, + session_maker, + stats, + ) + + result = runner.invoke(app, ["reindex", "--embeddings"]) + + assert result.exit_code == 0 + output = "\n".join(printed_lines) + assert ( + "Embeddings complete ([cyan]index=milvus[/cyan], " + "[cyan]model=FastEmbedEmbeddingProvider:BAAI/bge-small-en-v1.5[/cyan]): " + "2 entities embedded, 0 skipped, 0 errors" + ) in output + assert "Representative error:" not in output + assert "Reindex complete!" in output diff --git a/tests/repository/test_semantic_vector_sync.py b/tests/repository/test_semantic_vector_sync.py index 8b89cea2d..4bba2e200 100644 --- a/tests/repository/test_semantic_vector_sync.py +++ b/tests/repository/test_semantic_vector_sync.py @@ -129,6 +129,8 @@ def _batch_repository( repository = _TestRepository() repository._semantic_embedding_sync_batch_size = batch_size monkeypatch.setattr(repository, "_embedding_provider", object()) + monkeypatch.setattr(repository, "_embedding_model_key", Mock(return_value="test-model")) + repository._semantic_vector_index_name = "sqlite-vec" monkeypatch.setattr(repository, "_assert_semantic_available", Mock()) monkeypatch.setattr(repository, "_ensure_vector_tables", AsyncMock()) monkeypatch.setattr(repository, "_vector_prepare_window_size", Mock(return_value=8)) @@ -166,6 +168,8 @@ async def test_vector_sync_handles_empty_batches_and_deferred_empty_entities( ) assert empty_result.entities_total == 0 + assert empty_result.vector_index == "sqlite-vec" + assert empty_result.embedding_model == "test-model" deferred_repository = _batch_repository( monkeypatch, @@ -182,6 +186,38 @@ async def test_vector_sync_handles_empty_batches_and_deferred_empty_entities( assert deferred_result.entities_synced == 0 +@pytest.mark.asyncio +async def test_vector_sync_samples_distinct_errors_with_a_small_cap( + monkeypatch: pytest.MonkeyPatch, +) -> None: + repository = _batch_repository( + monkeypatch, + [ + RuntimeError("first failure\nwith context"), + RuntimeError("first failure with context"), + ValueError("second failure"), + RuntimeError(), + RuntimeError("fourth distinct failure"), + ], + ) + + result = await semantic_vector_sync.sync_entity_vectors_internal( + repository, + [1, 2, 3, 4, 5], + progress_callback=None, + continue_on_error=True, + ) + + assert result.entities_failed == 5 + assert result.sample_errors == ( + "first failure with context", + "second failure", + "RuntimeError", + ) + assert result.vector_index == "sqlite-vec" + assert result.embedding_model == "test-model" + + @pytest.mark.asyncio async def test_vector_sync_propagates_prepare_and_threshold_flush_errors( monkeypatch: pytest.MonkeyPatch, @@ -236,6 +272,7 @@ async def test_vector_sync_handles_final_flush_errors_and_orphan_runtime( ) assert failed_result.failed_entity_ids == (1,) + assert failed_result.sample_errors == ("final flush failed",) strict_repository = _batch_repository( monkeypatch, @@ -267,6 +304,9 @@ async def test_vector_sync_handles_final_flush_errors_and_orphan_runtime( ) assert orphan_result.failed_entity_ids == (1,) + assert orphan_result.sample_errors == ( + "Vector sync left unfinished entities after flushes.", + ) def test_vector_shard_planning_and_logging_edges(monkeypatch) -> None: diff --git a/tests/services/test_semantic_search.py b/tests/services/test_semantic_search.py index bf892ebae..e5e16edd5 100644 --- a/tests/services/test_semantic_search.py +++ b/tests/services/test_semantic_search.py @@ -186,6 +186,28 @@ async def test_semantic_vector_sync_batch_skips_embed_opt_out_and_reports_skips( delete_entity_vectors.assert_awaited_once_with(41) +@pytest.mark.asyncio +async def test_semantic_vector_sync_empty_batch_preserves_index_identity( + search_service, + monkeypatch, +): + repository = _sqlite_repo(search_service) + expected = VectorSyncBatchResult( + entities_total=0, + entities_synced=0, + entities_failed=0, + vector_index="sqlite-vec", + embedding_model="FastEmbedEmbeddingProvider:test-model", + ) + sync_batch = AsyncMock(return_value=expected) + monkeypatch.setattr(repository, "sync_entity_vectors_batch", sync_batch) + + result = await search_service.sync_entity_vectors_batch([]) + + assert result is expected + sync_batch.assert_awaited_once_with([]) + + @pytest.mark.asyncio async def test_embed_opt_out_note_still_participates_in_fts( search_service, session_maker, test_project @@ -240,9 +262,12 @@ async def test_reindex_vectors_respects_embed_opt_out(search_service, monkeypatc sync_batch = AsyncMock( return_value=VectorSyncBatchResult( entities_total=2, - entities_synced=1, - entities_failed=0, + entities_synced=0, + entities_failed=1, entities_skipped=1, + sample_errors=("embedding service unavailable",), + vector_index="sqlite-vec", + embedding_model="FastEmbedEmbeddingProvider:test-model", ) ) monkeypatch.setattr(search_service, "_purge_stale_search_rows", purge_stale_rows) @@ -257,9 +282,12 @@ async def test_reindex_vectors_respects_embed_opt_out(search_service, monkeypatc reconcile.assert_awaited_once() assert stats == { "total_entities": 2, - "embedded": 1, + "embedded": 0, "skipped": 1, - "errors": 0, + "errors": 1, + "sample_errors": ("embedding service unavailable",), + "vector_index": "sqlite-vec", + "embedding_model": "FastEmbedEmbeddingProvider:test-model", } @@ -283,7 +311,13 @@ async def sync_entity_vectors_batch(entity_ids, progress_callback=None): assert entity_ids == [42] assert progress_callback is None calls.append("sync") - return VectorSyncBatchResult(entities_total=1, entities_synced=1, entities_failed=0) + return VectorSyncBatchResult( + entities_total=1, + entities_synced=1, + entities_failed=0, + vector_index="sqlite-vec", + embedding_model="FastEmbedEmbeddingProvider:test-model", + ) monkeypatch.setattr(repository, "delete_stale_vector_rows", delete_stale_vector_rows) monkeypatch.setattr(search_service, "sync_entity_vectors_batch", sync_entity_vectors_batch) @@ -299,6 +333,9 @@ async def sync_entity_vectors_batch(entity_ids, progress_callback=None): "embedded": 1, "skipped": 0, "errors": 0, + "sample_errors": (), + "vector_index": "sqlite-vec", + "embedding_model": "FastEmbedEmbeddingProvider:test-model", } @@ -383,6 +420,8 @@ async def test_reindex_vectors_force_full_clears_project_vectors_before_resync( entities_total=2, entities_synced=2, entities_failed=0, + vector_index="sqlite-vec", + embedding_model="FastEmbedEmbeddingProvider:test-model", ) ) monkeypatch.setattr(search_service, "_purge_stale_search_rows", purge_stale_rows) @@ -402,6 +441,9 @@ async def test_reindex_vectors_force_full_clears_project_vectors_before_resync( "embedded": 2, "skipped": 0, "errors": 0, + "sample_errors": (), + "vector_index": "sqlite-vec", + "embedding_model": "FastEmbedEmbeddingProvider:test-model", } @@ -423,11 +465,17 @@ async def test_semantic_vector_sync_batch_cleans_up_unknown_ids(search_service, entities_synced=1, entities_failed=0, entities_skipped=1, + sample_errors=("shared failure", "cleanup failure"), + vector_index="sqlite-vec", + embedding_model="FastEmbedEmbeddingProvider:test-model", ), VectorSyncBatchResult( entities_total=1, entities_synced=1, entities_failed=0, + sample_errors=("shared failure", "embedding failure", "extra failure"), + vector_index="sqlite-vec", + embedding_model="FastEmbedEmbeddingProvider:test-model", ), ] ) @@ -451,3 +499,10 @@ async def test_semantic_vector_sync_batch_cleans_up_unknown_ids(search_service, assert result.entities_synced == 2 assert result.entities_failed == 0 assert result.entities_skipped == 0 + assert result.sample_errors == ( + "shared failure", + "cleanup failure", + "embedding failure", + ) + assert result.vector_index == "sqlite-vec" + assert result.embedding_model == "FastEmbedEmbeddingProvider:test-model" From 9cf618f308e19dd15f8b2a69153581954f348dc2 Mon Sep 17 00:00:00 2001 From: phernandez Date: Tue, 11 Aug 2026 19:24:01 -0500 Subject: [PATCH 2/2] test(core): give vector sync stubs a model identity Signed-off-by: phernandez --- tests/repository/test_semantic_search_base.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/tests/repository/test_semantic_search_base.py b/tests/repository/test_semantic_search_base.py index 0d5f83afe..3e63186af 100644 --- a/tests/repository/test_semantic_search_base.py +++ b/tests/repository/test_semantic_search_base.py @@ -961,7 +961,7 @@ async def test_sync_entity_vectors_batch_flushes_at_configured_threshold(monkeyp """Batch sync should flush queued jobs at semantic_embedding_sync_batch_size boundaries.""" repo = _ConcreteRepo() repo._semantic_enabled = True - repo._embedding_provider = object() + repo._embedding_provider = SimpleNamespace(model_name="stub", dimensions=4) repo._semantic_embedding_sync_batch_size = 2 prepared_by_entity = { @@ -1006,7 +1006,7 @@ async def test_sync_entity_vectors_batch_skip_only_has_zero_queue_wait(monkeypat """Skip-only batches should not accumulate synthetic queue wait.""" repo = _ConcreteRepo() repo._semantic_enabled = True - repo._embedding_provider = object() + repo._embedding_provider = SimpleNamespace(model_name="stub", dimensions=4) async def _stub_prepare_window(entity_ids: list[int]): return [ @@ -1039,7 +1039,7 @@ async def test_sync_entity_vectors_batch_progress_tracks_terminal_entities(monke """Progress callback should advance on terminal entity completion, not prepare entry.""" repo = _ConcreteRepo() repo._semantic_enabled = True - repo._embedding_provider = object() + repo._embedding_provider = SimpleNamespace(model_name="stub", dimensions=4) repo._semantic_embedding_sync_batch_size = 2 prepared_by_entity = { @@ -1083,7 +1083,7 @@ async def test_sync_entity_vectors_batch_continue_on_error(monkeypatch): """Batch sync should continue after per-entity and per-flush failures.""" repo = _ConcreteRepo() repo._semantic_enabled = True - repo._embedding_provider = object() + repo._embedding_provider = SimpleNamespace(model_name="stub", dimensions=4) repo._semantic_embedding_sync_batch_size = 1 async def _stub_prepare_window(entity_ids: list[int]): @@ -1130,7 +1130,7 @@ async def test_sync_entity_vectors_batch_only_attributes_queue_wait_to_flushed_e """Mixed batches should only charge queue wait to entities that entered flush work.""" repo = _ConcreteRepo() repo._semantic_enabled = True - repo._embedding_provider = object() + repo._embedding_provider = SimpleNamespace(model_name="stub", dimensions=4) repo._semantic_embedding_sync_batch_size = 2 async def _stub_prepare_window(entity_ids: list[int]): @@ -1193,7 +1193,7 @@ async def test_sync_entity_vectors_batch_tracks_prepare_and_queue_wait_seconds(m """Queue wait should be reported separately from prepare/embed/write timings.""" repo = _ConcreteRepo() repo._semantic_enabled = True - repo._embedding_provider = object() + repo._embedding_provider = SimpleNamespace(model_name="stub", dimensions=4) repo._semantic_embedding_sync_batch_size = 2 async def _stub_prepare_window(entity_ids: list[int]): @@ -1309,7 +1309,7 @@ async def test_sync_entity_vectors_batch_records_entity_granularity_histograms(m """Entity timing histograms should emit one sample per finalized entity.""" repo = _ConcreteRepo() repo._semantic_enabled = True - repo._embedding_provider = object() + repo._embedding_provider = SimpleNamespace(model_name="stub", dimensions=4) repo._semantic_embedding_sync_batch_size = 2 async def _stub_prepare_window(entity_ids: list[int]):