From 3fa4a8742060ac54b9f0e27b8d8751db76fd3a8c Mon Sep 17 00:00:00 2001
From: Meta Vens <rwvens@hotmail.com>
Date: Sun, 30 Aug 2026 23:35:57 +0200
Subject: [PATCH] Report source embedding progress

---
 packages/rag/src/source-rag-pipeline.ts     | 10 +++++++++-
 packages/rag/src/source-repo-indexing.ts    |  3 +++
 scripts/smoke-source-rag-commit-boundary.ts |  6 +++++-
 3 files changed, 17 insertions(+), 2 deletions(-)

diff --git a/packages/rag/src/source-rag-pipeline.ts b/packages/rag/src/source-rag-pipeline.ts
index 7b59c58..5e94238 100644
--- a/packages/rag/src/source-rag-pipeline.ts
+++ b/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);
     }
   }
 }
diff --git a/packages/rag/src/source-repo-indexing.ts b/packages/rag/src/source-repo-indexing.ts
index 530ea22..9cc1128 100644
--- a/packages/rag/src/source-repo-indexing.ts
+++ b/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,
diff --git a/scripts/smoke-source-rag-commit-boundary.ts b/scripts/smoke-source-rag-commit-boundary.ts
index b2d976e..4937d37 100644
--- a/scripts/smoke-source-rag-commit-boundary.ts
+++ b/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[] };