diff --git a/src/basic_memory/index/local_dependencies.py b/src/basic_memory/index/local_dependencies.py index c01e97104..aca00d920 100644 --- a/src/basic_memory/index/local_dependencies.py +++ b/src/basic_memory/index/local_dependencies.py @@ -393,6 +393,7 @@ async def index_markdown_file( checksum=synced.checksum, content_checksum=await compute_checksum(synced.markdown_content), operation=operation, + indexed_bytes=synced.indexed_bytes, content_superseded=content_superseded, ) @@ -464,6 +465,7 @@ async def index_regular_file( # A regular file carries no note provenance to validate. content_checksum=None, operation=operation, + indexed_bytes=indexed.indexed_bytes, ) async def publish_relation_generation( @@ -478,6 +480,7 @@ async def publish_relation_generation( entity_id=synced.entity.id, permalink=synced.entity.permalink, checksum=synced.checksum, + indexed_bytes=synced.indexed_bytes, content_type=synced.content_type, markdown_content=synced.markdown_content, observations=synced.observations, @@ -564,6 +567,8 @@ async def index_current_markdown_file( content_type=self.file_service.content_type(path), updated_at=file_metadata.modified_at, size=file_metadata.size, + # Unchanged bytes were not re-indexed, so this pass did no indexing work. + indexed_bytes=0, relations=(), resolve_relations=resolve_relations, ) @@ -614,6 +619,7 @@ async def refresh_unchanged_markdown_file( content_type=self.file_service.content_type(input_file.path), updated_at=file_metadata.modified_at, size=file_metadata.size, + indexed_bytes=indexed.indexed_bytes, observations=indexed.observations, sections=indexed.sections, relations=indexed.relations, @@ -684,6 +690,7 @@ async def index_changed_markdown_file( content_type=self.file_service.content_type(input_file.path), updated_at=file_metadata.modified_at, size=file_metadata.size, + indexed_bytes=indexed.indexed_bytes, observations=indexed.observations, sections=indexed.sections, relations=indexed.relations, diff --git a/src/basic_memory/indexing/batch_indexer.py b/src/basic_memory/indexing/batch_indexer.py index 75cead3e4..3eafa5142 100644 --- a/src/basic_memory/indexing/batch_indexer.py +++ b/src/basic_memory/indexing/batch_indexer.py @@ -268,6 +268,7 @@ async def index_files( worker=lambda path: self._refresh_search_index( prepared_entities[path], entities_by_id[prepared_entities[path].entity_id], + indexed_bytes=_input_content_bytes(files[path]), ), ) error_by_path.update(refresh_errors) @@ -336,13 +337,18 @@ async def index_markdown_file( ) if index_search: - return await self._refresh_search_index(prepared_entity, entity) + return await self._refresh_search_index( + prepared_entity, + entity, + indexed_bytes=_input_content_bytes(file), + ) return IndexedEntity( path=prepared_entity.path, entity_id=entity.id, permalink=entity.permalink, checksum=prepared_entity.checksum, + indexed_bytes=_input_content_bytes(file), content_type=prepared_entity.content_type, markdown_content=prepared_entity.markdown_content, observations=prepared_entity.observations, @@ -930,7 +936,13 @@ async def refresh_indexed_entity_search( relations=indexed.relations, resolve_relations=indexed.resolve_relations, ) - refreshed = await self._refresh_search_index(prepared, refresh.entity) + # Re-rendering search for an already indexed snapshot reads no new + # content, so the snapshot keeps the byte count of its original read. + refreshed = await self._refresh_search_index( + prepared, + refresh.entity, + indexed_bytes=indexed.indexed_bytes, + ) async with db.scoped_session(self.session_maker) as session: # Trigger: N+1 can be accepted after N loaded its coherent snapshot but # before N finishes the external search write. @@ -986,7 +998,7 @@ async def _find_unresolved_relations_for_entity(self, entity_id: int): # --- Search refresh --- async def _refresh_search_index( - self, prepared: _PreparedEntity, entity: Entity + self, prepared: _PreparedEntity, entity: Entity, *, indexed_bytes: int ) -> IndexedEntity: if prepared.refresh_search: try: @@ -1005,6 +1017,9 @@ async def _refresh_search_index( entity_id=entity.id, permalink=entity.permalink, checksum=prepared.checksum, + # A guarded no-op (a writer that lost a note/resource race) indexed nothing, + # so it bills nothing, whatever it read. + indexed_bytes=indexed_bytes if prepared.refresh_search else 0, content_type=prepared.content_type, markdown_content=prepared.markdown_content, observations=prepared.observations, @@ -1240,3 +1255,13 @@ async def run(path: str) -> None: await asyncio.gather(*(run(path) for path in paths)) return results, errors + + +def _input_content_bytes(file: IndexInputFile) -> int: + """Byte length of the loaded content the indexer parses for one input file. + + Markdown input always carries content (preparation rejects it otherwise). A + regular file loaded without content is indexed from metadata alone, so it + reports 0 bytes. + """ + return len(file.content) if file.content is not None else 0 diff --git a/src/basic_memory/indexing/embedding_index_planning.py b/src/basic_memory/indexing/embedding_index_planning.py index e354918bf..590cbfb1d 100644 --- a/src/basic_memory/indexing/embedding_index_planning.py +++ b/src/basic_memory/indexing/embedding_index_planning.py @@ -204,6 +204,9 @@ class EmbeddingIndexResult: entity_id: int status: EmbeddingIndexStatus reason: str + # Chunks actually sent to the embedder. Chunks found unchanged are skipped + # and do not count, so an immediate re-sync reports zero. + chunks_embedded: int @dataclass(frozen=True, slots=True) @@ -240,6 +243,8 @@ class EmbeddingIndexBatchResult: failed_entities: int deferred_entities: int reason: str + # Sum of chunks actually sent to the embedder across the batch. + chunks_embedded: int @classmethod def no_entities(cls) -> "EmbeddingIndexBatchResult": @@ -252,6 +257,7 @@ def no_entities(cls) -> "EmbeddingIndexBatchResult": failed_entities=0, deferred_entities=0, reason="no entities", + chunks_embedded=0, ) @@ -334,11 +340,12 @@ async def run_embedding_index( vector_sync: EntityVectorSync, ) -> EmbeddingIndexResult: """Run one embedding index request through a concrete vector sync backend.""" - await vector_sync.sync_entity_vectors(request.entity_id) + sync_result = await vector_sync.sync_entity_vectors(request.entity_id) return EmbeddingIndexResult( entity_id=request.entity_id, status=EmbeddingIndexStatus.processed, reason=f"entity embeddings indexed: {request.entity_id}", + chunks_embedded=sync_result.embedding_jobs_total, ) @@ -370,6 +377,7 @@ def summarize_embedding_index_batch_result( failed_entities=batch_result.entities_failed, deferred_entities=batch_result.entities_deferred, reason=f"entity embedding batch indexed: {plan.unique_entities} entities", + chunks_embedded=batch_result.embedding_jobs_total, ) diff --git a/src/basic_memory/indexing/file_indexer.py b/src/basic_memory/indexing/file_indexer.py index 4d2b6144d..6a9a06f49 100644 --- a/src/basic_memory/indexing/file_indexer.py +++ b/src/basic_memory/indexing/file_indexer.py @@ -268,6 +268,7 @@ async def index_markdown_file( checksum=synced.checksum, content_checksum=await compute_checksum(synced.markdown_content), operation=operation, + indexed_bytes=synced.indexed_bytes, content_superseded=content_superseded, ) diff --git a/src/basic_memory/indexing/models.py b/src/basic_memory/indexing/models.py index 51ec7e5a0..0e2606f38 100644 --- a/src/basic_memory/indexing/models.py +++ b/src/basic_memory/indexing/models.py @@ -183,6 +183,9 @@ class IndexedEntity: entity_id: int permalink: str | None checksum: str + # Byte length of the stored content this pass read and parsed (the loaded + # input file's bytes, before any frontmatter normalization rewrite). + indexed_bytes: int content_type: str | None = None markdown_content: str | None = None observations: tuple[IndexedObservation, ...] = () @@ -233,6 +236,9 @@ class FileIndexResult: # None for regular (non-note) files. content_checksum: RuntimeNoteContentChecksum | None operation: FileIndexOperation + # Byte length of the stored content the indexer read and indexed. For both + # Markdown and regular files this is the full stored object read from storage. + indexed_bytes: int # The indexed snapshot committed successfully, but a newer accepted note # generation won before derived relations could be published. The next # coalesced write owns convergence; callers must not enqueue stale followups. @@ -250,6 +256,7 @@ def from_fields( checksum: str, content_checksum: RuntimeNoteContentChecksum | None, operation: FileIndexOperation, + indexed_bytes: int, content_superseded: bool = False, ) -> FileIndexResult: """Validate entity fields loaded for a completed file-index result. @@ -279,6 +286,7 @@ def from_fields( checksum=checksum, content_checksum=content_checksum, operation=operation, + indexed_bytes=indexed_bytes, content_superseded=content_superseded, ) @@ -343,6 +351,10 @@ class IndexFileJobResult: # landed mid-job, so this result describes superseded content (issue #1445). # That write's own storage notification indexes the newer content. content_superseded: bool = False + # Bytes of stored content this job actually indexed, for work-unit billing. + # Only a processed, non-superseded file reports bytes; current, missing, + # failed, and superseded outcomes report 0. + indexed_bytes: int = 0 @dataclass(frozen=True, slots=True) @@ -448,6 +460,9 @@ def index_file_job_result_from_indexed_file( operation = indexed_file.operation if live_update_plan is not None and live_update_plan.operation is not None: operation = live_update_plan.operation + content_superseded = indexed_file.content_superseded or ( + live_update_plan.content_superseded if live_update_plan is not None else False + ) return IndexFileJobResult( status=IndexFileJobStatus.processed, @@ -467,10 +482,10 @@ def index_file_job_result_from_indexed_file( live_update_plan.live_update_source if live_update_plan is not None else None ), db_version=live_update_plan.db_version if live_update_plan is not None else None, - content_superseded=( - indexed_file.content_superseded - or (live_update_plan.content_superseded if live_update_plan is not None else False) - ), + content_superseded=content_superseded, + # A superseded result describes content a newer write replaced; that + # write's own job bills its bytes, so this one reports none. + indexed_bytes=0 if content_superseded else indexed_file.indexed_bytes, ) @@ -755,6 +770,11 @@ class IndexFileBatchJobResult: file_results: tuple[IndexFileJobResult, ...] vector_targets: tuple[EmbeddingIndexTarget, ...] + @property + def indexed_bytes(self) -> int: + """Stored bytes actually indexed across this batch's file results.""" + return sum(result.indexed_bytes for result in self.file_results) + def project_index_file_outcome_from_job_result( result: IndexFileJobResult, @@ -828,6 +848,7 @@ def build_index_file_batch_job_result( reason=f"file indexed: {file_path}", entity_id=indexed.entity_id, entity_checksum=indexed.checksum, + indexed_bytes=indexed.indexed_bytes, ) ) else: @@ -860,6 +881,9 @@ class SyncedMarkdownFile: content_type: str updated_at: datetime size: int + # Byte length of the stored content this pass read and indexed; 0 when the + # file was found unchanged and nothing was re-indexed. + indexed_bytes: int observations: tuple[IndexedObservation, ...] = () sections: tuple[IndexedSection, ...] = () relations: tuple[IndexedRelation, ...] = () diff --git a/src/basic_memory/indexing/project_index_coordinator.py b/src/basic_memory/indexing/project_index_coordinator.py index 0317ffb1a..6b01f22ac 100644 --- a/src/basic_memory/indexing/project_index_coordinator.py +++ b/src/basic_memory/indexing/project_index_coordinator.py @@ -184,6 +184,14 @@ class ProjectIndexCoordinatorResult: batch_results: tuple[IndexFileBatchJobResult, ...] = () completion: ProjectIndexCompletion | None = None + @property + def indexed_bytes(self) -> int: + """Stored bytes indexed by batches this run executed inline. + + Queued batches run as their own jobs and report their own bytes. + """ + return sum(batch_result.indexed_bytes for batch_result in self.batch_results) + def build_project_index_batch_job_plan( *, diff --git a/src/basic_memory/repository/search_repository.py b/src/basic_memory/repository/search_repository.py index fb554a5f9..dabcdc00e 100644 --- a/src/basic_memory/repository/search_repository.py +++ b/src/basic_memory/repository/search_repository.py @@ -170,7 +170,7 @@ async def purge_stale_search_rows(self) -> int: """Delete full-text rows whose backing database row no longer exists.""" ... - async def sync_entity_vectors(self, entity_id: int) -> None: + async def sync_entity_vectors(self, entity_id: int) -> VectorSyncBatchResult: """Sync semantic vector chunks for an entity.""" ... diff --git a/src/basic_memory/repository/search_repository_base.py b/src/basic_memory/repository/search_repository_base.py index 87cf3c553..6cf433cab 100644 --- a/src/basic_memory/repository/search_repository_base.py +++ b/src/basic_memory/repository/search_repository_base.py @@ -1568,9 +1568,9 @@ def _split_text_into_chunks(self, text_value: str) -> list[str]: # Shared semantic search: sync_entity_vectors orchestration # ------------------------------------------------------------------ - async def sync_entity_vectors(self, entity_id: int) -> None: + async def sync_entity_vectors(self, entity_id: int) -> VectorSyncBatchResult: """Sync semantic chunk rows + embeddings for a single entity.""" - await self._sync_entity_vectors_internal( + return await self._sync_entity_vectors_internal( [entity_id], progress_callback=None, continue_on_error=False, diff --git a/src/basic_memory/runtime/vector_sync.py b/src/basic_memory/runtime/vector_sync.py index 4e44bc08b..3438868f4 100644 --- a/src/basic_memory/runtime/vector_sync.py +++ b/src/basic_memory/runtime/vector_sync.py @@ -38,4 +38,4 @@ class VectorSyncBatchResult: class EntityVectorSync(Protocol): """Capability that refreshes semantic vectors for one entity.""" - async def sync_entity_vectors(self, entity_id: EntityId) -> None: ... + async def sync_entity_vectors(self, entity_id: EntityId) -> VectorSyncBatchResult: ... diff --git a/src/basic_memory/services/search_service.py b/src/basic_memory/services/search_service.py index 559f203fa..287fad32c 100644 --- a/src/basic_memory/services/search_service.py +++ b/src/basic_memory/services/search_service.py @@ -668,19 +668,24 @@ async def index_entity_data( ) raise # pragma: no cover - async def sync_entity_vectors(self, entity_id: int) -> None: + async def sync_entity_vectors(self, entity_id: int) -> VectorSyncBatchResult: """Refresh vector chunks for one entity in repositories that support semantic indexing.""" async with db.scoped_session(self.session_maker) as session: entity = await self.entity_repository.find_by_id(session, entity_id) if entity is None: await self._clear_entity_vectors(entity_id) - return + # Clearing vectors embeds nothing, so the result reports zero chunks. + # A missing entity is not counted as skipped, matching the batch path. + return VectorSyncBatchResult(entities_total=1, entities_synced=0, entities_failed=0) if not entity_embeddings_enabled(entity): await self._clear_entity_vectors(entity_id) - return + # Opted-out entities count as skipped, matching the batch path. + return VectorSyncBatchResult( + entities_total=1, entities_synced=0, entities_failed=0, entities_skipped=1 + ) - await self.repository.sync_entity_vectors(entity_id) + return await self.repository.sync_entity_vectors(entity_id) async def sync_entity_vectors_batch( self, diff --git a/test-int/read_cache/test_materialization_invalidation.py b/test-int/read_cache/test_materialization_invalidation.py index 2c2ac9a3c..564a35008 100644 --- a/test-int/read_cache/test_materialization_invalidation.py +++ b/test-int/read_cache/test_materialization_invalidation.py @@ -61,6 +61,7 @@ async def index_file(self, file_path: str, *, source: str) -> FileIndexResult: if checksum is None: raise AssertionError("seeded materialization entity must carry a checksum") return FileIndexResult( + indexed_bytes=0, file_path=file_path, entity_id=self.entity.id, external_id=str(self.entity.external_id), diff --git a/test-int/read_cache/test_search_read_cache.py b/test-int/read_cache/test_search_read_cache.py index e21fbc54b..076e43108 100644 --- a/test-int/read_cache/test_search_read_cache.py +++ b/test-int/read_cache/test_search_read_cache.py @@ -13,6 +13,7 @@ from basic_memory.models import Project from basic_memory.read_cache import ReadCacheKey, ReadCacheOperation from basic_memory.read_cache.redis import RedisReadCache +from basic_memory.runtime.vector_sync import VectorSyncBatchResult class RedisCacheHarness(Protocol): @@ -51,7 +52,7 @@ class PartiallyFailingVectorSync: def __init__(self) -> None: self.synced_entity_ids: list[int] = [] - async def sync_entity_vectors(self, entity_id: int) -> None: + async def sync_entity_vectors(self, entity_id: int) -> VectorSyncBatchResult: self.synced_entity_ids.append(entity_id) raise RuntimeError("vector publication failed") diff --git a/test-int/test_indexed_bytes_work_units.py b/test-int/test_indexed_bytes_work_units.py new file mode 100644 index 000000000..b6379ad49 --- /dev/null +++ b/test-int/test_indexed_bytes_work_units.py @@ -0,0 +1,84 @@ +"""Indexing reports the stored bytes it actually indexed, for work-unit billing. + +Cloud bills per-tenant work units from these counts, so a file the checker finds +current must report zero: only content the indexer read and parsed counts. +""" + +import pytest + +from basic_memory.file_utils import compute_checksum +from basic_memory.index.local_project import LocalProjectIndexRuntimeFactory +from basic_memory.indexing.models import IndexFileJobStatus +from basic_memory.models import Project +from basic_memory.runtime.jobs import RuntimeIndexFileBatchJobRequest, RuntimeObservedIndexFile +from basic_memory.runtime.projects import ProjectRuntimeReference + +# Complete frontmatter, so indexing never rewrites the file and the stored bytes +# stay exactly what the test wrote. +NOTE_CONTENT = """--- +title: Billing Note +type: note +permalink: notes/billing-note +--- + +# Billing Note + +Unicode counts as bytes, not characters: café, naïve, 漢字. + +- [fact] Work units bill on bytes indexed +""" + + +@pytest.mark.asyncio +async def test_batch_index_reports_bytes_indexed_and_zero_for_current_file( + test_project: Project, + project_config, +) -> None: + path = "notes/billing-note.md" + (project_config.home / "notes").mkdir(parents=True, exist_ok=True) + (project_config.home / path).write_text(NOTE_CONTENT, encoding="utf-8") + expected_bytes = len(NOTE_CONTENT.encode("utf-8")) + runtime = await LocalProjectIndexRuntimeFactory().runtime_for_project(test_project) + # The observed checksum lets the checker compare against the index and + # decide the second pass is current without reading content. + request = RuntimeIndexFileBatchJobRequest( + project=ProjectRuntimeReference.from_project(test_project), + batch_index=0, + batch_count=1, + observed_files=( + RuntimeObservedIndexFile( + path=path, + checksum=await compute_checksum(NOTE_CONTENT), + size=expected_bytes, + ), + ), + ) + + first = await runtime.batch_enqueuer.enqueue_index_file_batch(request) + second = await runtime.batch_enqueuer.enqueue_index_file_batch(request) + # The local enqueuer runs batches inline, so both calls return results. + assert first is not None + assert second is not None + + assert (project_config.home / path).read_text(encoding="utf-8") == NOTE_CONTENT + assert first.file_results[0].status == IndexFileJobStatus.processed + assert first.file_results[0].indexed_bytes == expected_bytes + assert first.indexed_bytes == expected_bytes + assert second.file_results[0].status == IndexFileJobStatus.current + assert second.file_results[0].indexed_bytes == 0 + assert second.indexed_bytes == 0 + + +@pytest.mark.asyncio +async def test_single_file_index_reports_bytes_indexed( + test_project: Project, + project_config, +) -> None: + path = "notes/billing-note.md" + (project_config.home / "notes").mkdir(parents=True, exist_ok=True) + (project_config.home / path).write_text(NOTE_CONTENT, encoding="utf-8") + dependencies = await LocalProjectIndexRuntimeFactory().dependencies_for_project(test_project) + + result = await dependencies.file_indexer.index_file(path, source="index") + + assert result.indexed_bytes == len(NOTE_CONTENT.encode("utf-8")) diff --git a/tests/cloud/test_note_content_materialization.py b/tests/cloud/test_note_content_materialization.py index 018f5eec9..cb4f677f5 100644 --- a/tests/cloud/test_note_content_materialization.py +++ b/tests/cloud/test_note_content_materialization.py @@ -59,6 +59,7 @@ def __init__(self) -> None: async def index_file(self, file_path: str, *, source: str) -> FileIndexResult: self.calls.append((file_path, source)) return FileIndexResult( + indexed_bytes=0, file_path=file_path, entity_id=42, external_id="note-1", diff --git a/tests/index/test_inline_storage_event_processor.py b/tests/index/test_inline_storage_event_processor.py index ac2840a30..8a1bf2dbc 100644 --- a/tests/index/test_inline_storage_event_processor.py +++ b/tests/index/test_inline_storage_event_processor.py @@ -104,6 +104,7 @@ def __init__(self) -> None: async def index_file(self, file_path: str, *, source: str) -> FileIndexResult: self.calls.append((file_path, source)) return FileIndexResult( + indexed_bytes=0, file_path=file_path, entity_id=42, external_id="note-42", diff --git a/tests/index/test_local_markdown_file_indexer.py b/tests/index/test_local_markdown_file_indexer.py index 9ccd6b3bf..a3483c674 100644 --- a/tests/index/test_local_markdown_file_indexer.py +++ b/tests/index/test_local_markdown_file_indexer.py @@ -54,6 +54,7 @@ async def test_local_file_indexer_logs_brace_path_through_retry( entity.observations = [] entity.relations = [] synced_file = SyncedMarkdownFile( + indexed_bytes=0, entity=entity, checksum="abc123", markdown_content="# {AG} Plan\n", @@ -130,6 +131,7 @@ async def test_local_file_indexer_treats_empty_markdown_basename_as_regular_file note_content_reconciler=cast(IndexMarkdownNoteContentReconciler, Mock()), ) regular_result = FileIndexResult( + indexed_bytes=0, file_path="_phase7_import/.md", entity_id=42, external_id="resource-42", diff --git a/tests/index/test_local_project_index.py b/tests/index/test_local_project_index.py index da57e7faa..81b5b1c55 100644 --- a/tests/index/test_local_project_index.py +++ b/tests/index/test_local_project_index.py @@ -1118,6 +1118,7 @@ async def index_files( return IndexingBatchResult( indexed=[ IndexedEntity( + indexed_bytes=0, path=file_path, entity_id=index + 1, permalink=None, @@ -2521,6 +2522,7 @@ class RecordingMarkdownFileIndexer: async def index_file(self, file_path: str, *, source: str) -> FileIndexResult: self.indexed_paths.append(file_path) return FileIndexResult.from_fields( + indexed_bytes=0, file_path=file_path, entity_id=99, external_id="note-99", diff --git a/tests/index/test_local_schedulers.py b/tests/index/test_local_schedulers.py index 8c17a03e4..580c06b77 100644 --- a/tests/index/test_local_schedulers.py +++ b/tests/index/test_local_schedulers.py @@ -14,6 +14,7 @@ drain_background_tasks, ) from basic_memory.read_cache import ReadCacheInvalidationStatus +from basic_memory.runtime.vector_sync import VectorSyncBatchResult PROJECT_EXTERNAL_ID = "00000000-0000-0000-0000-000000000013" @@ -37,8 +38,9 @@ def __init__(self) -> None: self.vector_synced: list[int] = [] self.reindexed_project = False - async def sync_entity_vectors(self, entity_id: int) -> None: + async def sync_entity_vectors(self, entity_id: int) -> VectorSyncBatchResult: self.vector_synced.append(entity_id) + return VectorSyncBatchResult(entities_total=1, entities_synced=1, entities_failed=0) async def reindex_all(self) -> None: self.reindexed_project = True @@ -78,7 +80,7 @@ async def test_entity_vector_scheduler_invalidates_after_partial_failure(): class FailingSearchService(StubSearchService): @override - async def sync_entity_vectors(self, entity_id: int) -> None: + async def sync_entity_vectors(self, entity_id: int) -> VectorSyncBatchResult: self.vector_synced.append(entity_id) raise RuntimeError("vector publication failed") diff --git a/tests/indexing/test_embedding_index_planning.py b/tests/indexing/test_embedding_index_planning.py index be8f0934a..6be985203 100644 --- a/tests/indexing/test_embedding_index_planning.py +++ b/tests/indexing/test_embedding_index_planning.py @@ -26,8 +26,16 @@ class SingleVectorSync: def __init__(self) -> None: self.synced_entity_ids: list[int] = [] - async def sync_entity_vectors(self, entity_id: int) -> None: + async def sync_entity_vectors(self, entity_id: int) -> VectorSyncBatchResult: self.synced_entity_ids.append(entity_id) + return VectorSyncBatchResult( + entities_total=1, + entities_synced=1, + entities_failed=0, + chunks_total=5, + chunks_skipped=2, + embedding_jobs_total=3, + ) class BatchVectorSync: @@ -42,6 +50,7 @@ async def sync_entity_vectors_batch(self, entity_ids: list[int]) -> VectorSyncBa entities_failed=0, entities_skipped=1, entities_deferred=1, + embedding_jobs_total=4, ) @@ -197,6 +206,7 @@ def test_embedding_index_batch_result_summarizes_plan_and_sync_counts() -> None: entities_failed=0, entities_skipped=1, entities_deferred=1, + embedding_jobs_total=7, ) assert summarize_embedding_index_batch_result(plan, batch_result) == ( @@ -208,6 +218,7 @@ def test_embedding_index_batch_result_summarizes_plan_and_sync_counts() -> None: failed_entities=0, deferred_entities=1, reason="entity embedding batch indexed: 2 entities", + chunks_embedded=7, ) ) with pytest.raises(FrozenInstanceError): @@ -223,6 +234,7 @@ def test_embedding_index_batch_result_handles_empty_batches() -> None: failed_entities=0, deferred_entities=0, reason="no entities", + chunks_embedded=0, ) @@ -231,10 +243,12 @@ def test_embedding_index_result_describes_one_entity_outcome() -> None: entity_id=42, status=EmbeddingIndexStatus.processed, reason="entity embeddings indexed: 42", + chunks_embedded=3, ) == EmbeddingIndexResult( entity_id=42, status=EmbeddingIndexStatus.processed, reason="entity embeddings indexed: 42", + chunks_embedded=3, ) @@ -254,6 +268,7 @@ async def test_run_embedding_index_syncs_one_entity_and_returns_result() -> None entity_id=42, status=EmbeddingIndexStatus.processed, reason="entity embeddings indexed: 42", + chunks_embedded=3, ) @@ -281,6 +296,7 @@ async def test_run_embedding_index_batch_dedupes_and_summarizes_vector_sync() -> failed_entities=0, deferred_entities=1, reason="entity embedding batch indexed: 2 entities", + chunks_embedded=4, ) diff --git a/tests/indexing/test_file_batch_runner.py b/tests/indexing/test_file_batch_runner.py index ab332cbef..7df5f88c8 100644 --- a/tests/indexing/test_file_batch_runner.py +++ b/tests/indexing/test_file_batch_runner.py @@ -129,6 +129,7 @@ async def index_files( return IndexingBatchResult( indexed=[ IndexedEntity( + indexed_bytes=0, path="notes/a.md", entity_id=42, permalink="notes/a", diff --git a/tests/indexing/test_file_indexer.py b/tests/indexing/test_file_indexer.py index 3bcdacec0..1d01c6e3c 100644 --- a/tests/indexing/test_file_indexer.py +++ b/tests/indexing/test_file_indexer.py @@ -64,6 +64,7 @@ def _synced_file( ) -> SyncedMarkdownFile: """Create the canonical markdown index result consumed by FileIndexer.""" return SyncedMarkdownFile( + indexed_bytes=0, entity=entity or _entity(), checksum=checksum, markdown_content=CANONICAL_MARKDOWN, @@ -91,6 +92,7 @@ def _file_indexer( markdown_indexer.publish_relation_generation = AsyncMock(return_value=True) markdown_indexer.index_file = AsyncMock( return_value=FileIndexResult( + indexed_bytes=0, file_path="notes/note.md", entity_id=42, external_id="note-42", diff --git a/tests/indexing/test_index_batch_runtime.py b/tests/indexing/test_index_batch_runtime.py index 6658f57b1..bfabb720d 100644 --- a/tests/indexing/test_index_batch_runtime.py +++ b/tests/indexing/test_index_batch_runtime.py @@ -181,6 +181,7 @@ async def test_index_batch_runtime_indexes_loaded_files_and_reconciles_note_cont result=IndexingBatchResult( indexed=[ IndexedEntity( + indexed_bytes=0, path="ok.md", entity_id=10, permalink="ok", @@ -189,6 +190,7 @@ async def test_index_batch_runtime_indexes_loaded_files_and_reconciles_note_cont markdown_content="# OK\n", ), IndexedEntity( + indexed_bytes=0, path="bad.md", entity_id=20, permalink="bad", @@ -197,6 +199,7 @@ async def test_index_batch_runtime_indexes_loaded_files_and_reconciles_note_cont markdown_content="# Bad\n", ), IndexedEntity( + indexed_bytes=0, path="image.png", entity_id=30, permalink=None, @@ -299,6 +302,7 @@ async def test_rewrite_between_scan_and_reconcile_publishes_no_stale_relations( result=IndexingBatchResult( indexed=[ IndexedEntity( + indexed_bytes=0, path="rewritten.md", entity_id=10, permalink="rewritten", @@ -352,6 +356,7 @@ def test_count_search_indexed_entities_uses_markdown_content_presence() -> None: count_search_indexed_entities( [ IndexedEntity( + indexed_bytes=0, path="note.md", entity_id=1, permalink="note", @@ -359,6 +364,7 @@ def test_count_search_indexed_entities_uses_markdown_content_presence() -> None: markdown_content="# Note\n", ), IndexedEntity( + indexed_bytes=0, path="image.png", entity_id=2, permalink=None, diff --git a/tests/indexing/test_index_file_runner.py b/tests/indexing/test_index_file_runner.py index 678f5e1d0..5e0d72702 100644 --- a/tests/indexing/test_index_file_runner.py +++ b/tests/indexing/test_index_file_runner.py @@ -174,6 +174,7 @@ def read_plan() -> FileIndexPlan: def indexed_file() -> FileIndexResult: return FileIndexResult( + indexed_bytes=0, file_path="notes/a.md", entity_id=42, external_id="note-42", diff --git a/tests/indexing/test_models.py b/tests/indexing/test_models.py index 9de9fbdc9..727300992 100644 --- a/tests/indexing/test_models.py +++ b/tests/indexing/test_models.py @@ -65,6 +65,7 @@ def test_file_index_result_is_a_frozen_success_value(): result = FileIndexResult( + indexed_bytes=0, file_path="notes/a.md", entity_id=42, external_id="note-42", @@ -83,6 +84,7 @@ def test_file_index_result_is_a_frozen_success_value(): def test_file_index_result_from_fields_validates_required_entity_text(): result = FileIndexResult.from_fields( + indexed_bytes=0, file_path="notes/a.md", entity_id=42, external_id=" note-42 ", @@ -94,6 +96,7 @@ def test_file_index_result_from_fields_validates_required_entity_text(): ) assert result == FileIndexResult( + indexed_bytes=0, file_path="notes/a.md", entity_id=42, external_id="note-42", @@ -106,6 +109,7 @@ def test_file_index_result_from_fields_validates_required_entity_text(): with pytest.raises(RuntimeError, match="Indexed entity for notes/a.md is missing title"): FileIndexResult.from_fields( + indexed_bytes=0, file_path="notes/a.md", entity_id=42, external_id="note-42", @@ -119,6 +123,7 @@ def test_file_index_result_from_fields_validates_required_entity_text(): def test_file_index_result_from_fields_validates_optional_permalink_text(): result = FileIndexResult.from_fields( + indexed_bytes=0, file_path="notes/a.md", entity_id=42, external_id="note-42", @@ -133,6 +138,7 @@ def test_file_index_result_from_fields_validates_optional_permalink_text(): with pytest.raises(RuntimeError, match="Indexed entity for notes/a.md has invalid permalink"): FileIndexResult.from_fields( + indexed_bytes=0, file_path="notes/a.md", entity_id=42, external_id="note-42", @@ -145,6 +151,7 @@ def test_file_index_result_from_fields_validates_optional_permalink_text(): with pytest.raises(RuntimeError, match="Indexed entity for notes/a.md has blank permalink"): FileIndexResult.from_fields( + indexed_bytes=0, file_path="notes/a.md", entity_id=42, external_id="note-42", @@ -215,8 +222,40 @@ def test_index_file_job_result_from_read_decision_fails_fast(): index_file_job_result_from_decision(decision) +@pytest.mark.parametrize( + ("indexed_superseded", "plan_superseded", "expected_bytes"), + [(False, False, 120), (True, False, 0), (False, True, 0)], +) +def test_index_file_job_result_bills_bytes_only_for_current_content( + indexed_superseded: bool, + plan_superseded: bool, + expected_bytes: int, +): + """A superseded result describes replaced content; the newer write bills its bytes.""" + indexed_file = FileIndexResult( + file_path="notes/a.md", + entity_id=42, + external_id="note-42", + title="A Note", + permalink="notes/a-note", + checksum="checksum-1", + content_checksum="content-1", + operation=FileIndexOperation.updated, + indexed_bytes=120, + content_superseded=indexed_superseded, + ) + + result = index_file_job_result_from_indexed_file( + indexed_file, + live_update_plan=IndexedFileLiveUpdatePlan(content_superseded=plan_superseded), + ) + + assert result.indexed_bytes == expected_bytes + + def test_index_file_job_result_from_indexed_file_uses_trusted_live_update_plan(): indexed_file = FileIndexResult( + indexed_bytes=0, file_path="notes/a.md", entity_id=42, external_id="note-42", @@ -588,12 +627,14 @@ def test_build_index_file_batch_job_result_preserves_order_and_embedding_targets }, indexed_files=( IndexedEntity( + indexed_bytes=0, path="notes/processed.md", entity_id=42, permalink="notes/processed", checksum="checksum-processed", ), IndexedEntity( + indexed_bytes=0, path="notes/failed.md", entity_id=43, permalink="notes/failed", @@ -875,6 +916,7 @@ def test_plan_current_materialized_note_result_requests_entity_when_metadata_is_ def indexed_note(*, checksum: str = "etag-1", content_checksum: str | None = "content-1"): return FileIndexResult( + indexed_bytes=0, file_path="notes/a.md", entity_id=42, external_id="note-42", diff --git a/tests/indexing/test_note_content_batch_reconciliation.py b/tests/indexing/test_note_content_batch_reconciliation.py index 76cb7509c..6e5243412 100644 --- a/tests/indexing/test_note_content_batch_reconciliation.py +++ b/tests/indexing/test_note_content_batch_reconciliation.py @@ -147,6 +147,7 @@ async def reconcile_note_content( reconciliation = await reconcile_indexed_note_content_batch( [ IndexedEntity( + indexed_bytes=0, path="ok.md", entity_id=42, permalink="ok", @@ -154,6 +155,7 @@ async def reconcile_note_content( markdown_content="# OK\n", ), IndexedEntity( + indexed_bytes=0, path="missing.md", entity_id=404, permalink="missing", @@ -161,6 +163,7 @@ async def reconcile_note_content( markdown_content="# Missing\n", ), IndexedEntity( + indexed_bytes=0, path="bad.md", entity_id=43, permalink="bad", @@ -168,6 +171,7 @@ async def reconcile_note_content( markdown_content="# Bad\n", ), IndexedEntity( + indexed_bytes=0, path="binary.png", entity_id=44, permalink=None, @@ -234,6 +238,7 @@ def test_indexed_note_content_observed_at_uses_original_timestamp_when_unchanged result = indexed_note_content_observed_at( IndexedEntity( + indexed_bytes=0, path="ok.md", entity_id=42, permalink="ok", @@ -258,6 +263,7 @@ def test_indexed_note_content_observed_at_uses_now_when_frontmatter_rewrites_fil result = indexed_note_content_observed_at( IndexedEntity( + indexed_bytes=0, path="ok.md", entity_id=42, permalink="ok", @@ -273,6 +279,7 @@ def test_indexed_note_content_observed_at_uses_now_when_frontmatter_rewrites_fil def test_indexed_note_content_observed_at_handles_missing_file_info() -> None: result = indexed_note_content_observed_at( IndexedEntity( + indexed_bytes=0, path="ok.md", entity_id=42, permalink="ok", @@ -307,6 +314,7 @@ async def fake_scoped_session( reconciliation = await reconcile_indexed_note_content_batch( [ IndexedEntity( + indexed_bytes=0, path="gone.md", entity_id=42, permalink="gone", @@ -384,6 +392,7 @@ async def test_batch_reader_reconciles_fresh_content_not_scan_snapshot( reconciliation = await reconcile_indexed_note_content_batch( [ IndexedEntity( + indexed_bytes=0, path=sample_entity.file_path, entity_id=sample_entity.id, permalink=sample_entity.permalink, @@ -462,6 +471,7 @@ async def test_batch_without_reader_reverts_to_scan_snapshot( reconciliation = await reconcile_indexed_note_content_batch( [ IndexedEntity( + indexed_bytes=0, path=sample_entity.file_path, entity_id=sample_entity.id, permalink=sample_entity.permalink, diff --git a/tests/indexing/test_regular_file_classification.py b/tests/indexing/test_regular_file_classification.py index 7ceb41130..106838cd4 100644 --- a/tests/indexing/test_regular_file_classification.py +++ b/tests/indexing/test_regular_file_classification.py @@ -433,6 +433,8 @@ async def stale_resource_snapshot( assert result.errors == [] assert result.indexed[0].content_type == RUNTIME_MARKDOWN_CONTENT_TYPE assert result.indexed[0].checksum == "new-markdown-checksum" + # The stale pass published nothing, so it bills none of the bytes it read. + assert result.indexed[0].indexed_bytes == 0 async with db.scoped_session(search_service.session_maker) as session: preserved = await entity_repository.get_by_id(session, note.id) note_content = await NoteContentRepository(project_id=project_id).get_by_entity_id( diff --git a/tests/repository/test_sqlite_vector_search_repository.py b/tests/repository/test_sqlite_vector_search_repository.py index 297ce4b92..bff60463c 100644 --- a/tests/repository/test_sqlite_vector_search_repository.py +++ b/tests/repository/test_sqlite_vector_search_repository.py @@ -13,6 +13,10 @@ from basic_memory import db from basic_memory.config import BasicMemoryConfig, DatabaseBackend +from basic_memory.indexing.embedding_index_planning import ( + EmbeddingIndexJobRequest, + run_embedding_index, +) from basic_memory.repository.embedding_provider import EmbeddingProvider from basic_memory.repository.litellm_provider import LiteLLMEmbeddingProvider from basic_memory.repository.prefixing_provider import PrefixingEmbeddingProvider @@ -1836,3 +1840,34 @@ async def fake_scoped_session(_session_maker): await index.search(query_embedding, limit=500, projects=search_repository.scope) assert captured_params[0]["vector_k"] == 500 assert captured_params[0]["limit"] == 500 + + +@pytest.mark.asyncio +async def test_embedding_index_reports_only_chunks_actually_embedded(search_repository): + """A first sync reports embedded chunks; an unchanged re-sync reports zero. + + Work units bill on chunks sent to the embedder, so chunks the sync skips as + unchanged must not count. + """ + if not isinstance(search_repository, SQLiteSearchRepository): + pytest.skip("sqlite-vec repository behavior is local SQLite-only.") + + _enable_semantic(search_repository) + await search_repository.init_search_index() + await search_repository.index_item( + _entity_row( + project_id=search_repository.project_id, + row_id=121, + entity_id=121, + title="Worker Queue", + permalink="specs/worker-queue", + content_stems="async worker queue task scheduling and retries", + ) + ) + request = EmbeddingIndexJobRequest(project_id=search_repository.project_id, entity_id=121) + + first = await run_embedding_index(request, vector_sync=search_repository) + second = await run_embedding_index(request, vector_sync=search_repository) + + assert first.chunks_embedded > 0 + assert second.chunks_embedded == 0