feat(core): report work units on indexing and embedding results

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,

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

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,
     )
 
 

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,
         )
 

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, ...] = ()

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(
     *,

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."""
         ...
 

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,

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: ...

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,

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),

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")