Report source embedding progress

packages/rag/src/source-rag-pipeline.ts

@@ -29,10 +29,16 @@ export class SourceRagPipeline {
     documents: ArchivedSourceDocument[],
     options?: {
       forceReindex?: boolean;
+      onEmbeddingProgress?: (completedChunks: number, totalChunks: number) => void;
     },
   ): Promise<SourceRagIngestResult> {
     const mutation = await this.archiveRepository.planRepoDocuments(repoName, documents);
-    await this.replaceChangedChunks(repoName, mutation, options?.forceReindex ?? false);
+    await this.replaceChangedChunks(
+      repoName,
+      mutation,
+      options?.forceReindex ?? false,
+      options?.onEmbeddingProgress,
+    );
     await this.archiveRepository.commitRepoDocuments(repoName, mutation);
 
     return {
@@ -48,6 +54,7 @@ export class SourceRagPipeline {
     repoName: string,
     mutation: SourceDocumentSyncResult,
     forceReindex: boolean,
+    onEmbeddingProgress?: (completedChunks: number, totalChunks: number) => void,
   ): Promise<void> {
     const documentsToUpsert = forceReindex ? mutation.activeDocuments : mutation.changedDocuments;
 
@@ -74,6 +81,7 @@ export class SourceRagPipeline {
       }
 
       await this.vectorStore.upsert(batch);
+      onEmbeddingProgress?.(Math.min(index + batch.length, chunks.length), chunks.length);
     }
   }
 }

packages/rag/src/source-repo-indexing.ts

@@ -95,6 +95,9 @@ export async function indexSourceRepos(
     });
     const result = await pipeline.syncRepoDocuments(repo.repoName, scan.documents, {
       forceReindex: options.forceReindex,
+      onEmbeddingProgress: (completedChunks, totalChunks) => options.logger?.(
+        `Repo ${repo.repoName}: embedded_chunks=${completedChunks}/${totalChunks}`,
+      ),
     });
     const repoResult: IndexedRepoResult = {
       repoName: repo.repoName,

scripts/smoke-source-rag-commit-boundary.ts

@@ -54,7 +54,11 @@ async function main(): Promise<void> {
       new SourceDocumentIngester(),
       new InMemoryVectorStore(),
     );
-    await workingPipeline.syncRepoDocuments("Fixture", [updated]);
+    let embeddedChunks = 0;
+    await workingPipeline.syncRepoDocuments("Fixture", [updated], {
+      onEmbeddingProgress: (completedChunks) => { embeddedChunks = completedChunks; },
+    });
+    assertEqual(embeddedChunks, 1, "embedding progress did not report the committed batch");
     assertEqual((await archive.get(sourceId))?.content, updated.content, "successful retry did not commit archive");
 
     const manifest = JSON.parse(await readFile(archivePath, "utf8")) as { repos?: unknown[] };