Make source indexing retry-safe

notes/fresh-workspace-handoff.md

@@ -19,6 +19,7 @@ Git history and smoke artifacts carry routine proof. `state/evidence.jsonl` shou
 - Disaster recovery is active on trusted old Yggdrasil (`94.130.75.184`). Big Yggdrasil (`94.130.223.131`) is retired and its recycled address is not a GameCult host. The surviving host did not retain the former VoidBot runtime state or credentials; the current body was rebuilt only from admitted recovery material.
 - Raven's verified offsite archive `20260603-043304-offsite-auto.zip` is staged on old Yggdrasil. Its SHA-256 is `47572df766e63212012eb75762359113a11c8bbf149b497e120fc4fda7191dfc`; the newer June 4 archive is corrupt. The payload contains PostgreSQL state, six Qdrant snapshots, and exact RAG archives, but no secrets.
 - The standalone read-only resident MCP was recovered first. `voidbot-retrieval.service` owns only loopback `127.0.0.1:17875`, exact archives, VoidBot's history/source Qdrant collections, and the matching embedding model. Discord, worker jobs, Personas, moderation, Bifrost crossings, and delivery remain stopped until their missing authority and credentials are rebuilt.
+- The operator authorized bringing source indexing back up on 2026-08-30. This admits a new source-refresh writer, not mutation authority for `voidbot-retrieval.service`. Before its first write, name one owner, repository inputs, coordinated archive/Qdrant commit and rollback boundary, trigger, progress witness, and deployment path; then prove the smallest incremental catch-up before broad reconciliation.
 - `apps/worker/src/mcp-http-main.ts` is the standalone HTTP entrypoint. It delegates to the existing MCP context and HTTP server; it is not a second retrieval implementation.
 - The current `voidbot.service` runs only the typed `voidbot.swarm` Eve publisher. The repo Face/native Persona scheduler, worker queue, Discord bot, Postgres body, and resident cognition organs described later in this handoff are source architecture or retired-runtime history, not current service-health claims. Restore none of them by implication from the surviving swarm projection.
 - The teardown authority remains `notes/agent-swarm-teardown-plan-2026-05-21.md`, but it is now a restart checklist and smell ledger rather than a stop sign. Keep using it when a Face/Bifrost/CTB boundary smells like prompt/parser/transport Jenga.

notes/voidbot-implementation-plan.md

@@ -10,6 +10,12 @@ This is the current forward plan for the next larger organs. It is not a changel
 > decision. Do not treat the roadmap below as evidence that those processes are
 > live.
 
+Source-index refresh restoration is now operator-authorized. It must return as
+a separate writer authority; the read-only retrieval MCP keeps no crawler,
+mutation tool, archive writer, or Qdrant write lifecycle. The immediate work is
+to name that writer's inputs, atomicity/rollback boundary, schedule, progress
+witness, and deployment owner, then prove a bounded incremental catch-up.
+
 ## Current Aim
 
 The typed moderation, mood, sleep, agency, and private self-state boundary is live, but the repo Face / CTB / Bifrost swarm loop is under teardown. Do not let ordinary swarm scheduling run. The active plan is `notes/agent-swarm-teardown-plan-2026-05-21.md`: the current Face loop mixed scheduling, prompt law, public speech, identity, governance, Discord mirroring, Bifrost dispatch, and GitHub proposal behavior into one prompt-and-parser machine. That foundation is no longer eligible for feature work.

package.json

@@ -72,6 +72,7 @@
     "rag:migrate-qdrant": "tsx scripts/migrate-local-vectors-to-qdrant.ts",
     "rag:migrate-vector-layout": "tsx scripts/migrate-vector-store-layout.ts",
     "rag:rebuild": "tsx scripts/rebuild-rag-index.ts",
+    "rag:commit-boundary-smoke": "tsx scripts/smoke-source-rag-commit-boundary.ts",
     "rag:exact-source-mcp-smoke": "tsx scripts/smoke-exact-source-document-mcp.ts"
   },
   "devDependencies": {

packages/rag/src/source-document-archive.ts

@@ -52,6 +52,10 @@ export interface SourceDocumentSyncResult {
   activeDocuments: ArchivedSourceDocumentRecord[];
 }
 
+export interface SourceDocumentSyncPlan extends SourceDocumentSyncResult {
+  previousDocuments: ArchivedSourceDocumentRecord[];
+}
+
 export class FileSourceDocumentArchiveRepository {
   private readonly manifestStore: SerializedFileStore<SourceDocumentArchiveManifest>;
   private readonly shardRoot: string;
@@ -72,78 +76,42 @@ export class FileSourceDocumentArchiveRepository {
     repoName: string,
     documents: ArchivedSourceDocument[],
   ): Promise<SourceDocumentSyncResult> {
+    const plan = await this.planRepoDocuments(repoName, documents);
+    await this.commitRepoDocuments(repoName, plan);
+    return plan;
+  }
+
+  public async planRepoDocuments(
+    repoName: string,
+    documents: ArchivedSourceDocument[],
+  ): Promise<SourceDocumentSyncPlan> {
     await this.ensureInitialized();
 
     const shardStore = this.getRepoShardStore(repoName);
-    const result = await shardStore.mutate((store) => {
-      const now = new Date().toISOString();
-      const previousRepoSourceIds = store.documents.map((document) => document.id);
-      const nextDocuments = documents.map((document) => toArchivedSourceDocumentRecord(document, now));
-      const existingById = new Map(store.documents.map((document, index) => [document.id, index]));
-      const nextIds = new Set(nextDocuments.map((document) => document.id));
-      const changedSourceIds: string[] = [];
-      const changedDocuments: ArchivedSourceDocumentRecord[] = [];
-      let created = 0;
-      let updated = 0;
-      let unchanged = 0;
-      let deleted = 0;
+    const store = await shardStore.snapshot();
+    return buildSyncPlan(store.documents, documents);
+  }
 
-      for (const document of nextDocuments) {
-        const position = existingById.get(document.id);
+  public async commitRepoDocuments(
+    repoName: string,
+    plan: SourceDocumentSyncPlan,
+  ): Promise<void> {
+    await this.ensureInitialized();
 
-        if (position === undefined) {
-          store.documents.push(document);
-          existingById.set(document.id, store.documents.length - 1);
-          changedSourceIds.push(document.id);
-          changedDocuments.push(document);
-          created += 1;
-          continue;
-        }
-
-        const existing = store.documents[position];
-
-        if (areEquivalent(existing, document)) {
-          unchanged += 1;
-          continue;
-        }
-
-        store.documents[position] = document;
-        changedSourceIds.push(document.id);
-        changedDocuments.push(document);
-        updated += 1;
+    const shardStore = this.getRepoShardStore(repoName);
+    await shardStore.mutate((store) => {
+      if (!sameDocumentSet(store.documents, plan.previousDocuments)) {
+        throw new Error(`Source archive changed while ${repoName} was being indexed.`);
       }
 
-      const retainedDocuments = store.documents.filter((document) => {
-        if (nextIds.has(document.id)) {
-          return true;
-        }
-
-        changedSourceIds.push(document.id);
-        deleted += 1;
-        return false;
-      });
-
-      store.documents = retainedDocuments;
-
-      return {
-        created,
-        updated,
-        unchanged,
-        deleted,
-        previousRepoSourceIds,
-        changedSourceIds,
-        changedDocuments,
-        activeDocuments: retainedDocuments.slice().sort(compareDocuments),
-      };
+      store.documents = plan.activeDocuments.slice();
     });
 
-    await this.syncRepoSummary(repoName, result.activeDocuments.length);
+    await this.syncRepoSummary(repoName, plan.activeDocuments.length);
 
-    if (result.activeDocuments.length === 0) {
+    if (plan.activeDocuments.length === 0) {
       await this.deleteRepoShard(repoName);
     }
-
-    return result;
   }
 
   public async get(sourceId: string): Promise<ArchivedSourceDocumentRecord | undefined> {
@@ -399,6 +367,65 @@ function areEquivalent(
   return JSON.stringify(comparableFields(left)) === JSON.stringify(comparableFields(right));
 }
 
+function buildSyncPlan(
+  previousDocuments: ArchivedSourceDocumentRecord[],
+  documents: ArchivedSourceDocument[],
+): SourceDocumentSyncPlan {
+  const now = new Date().toISOString();
+  const nextDocuments = documents.map((document) => toArchivedSourceDocumentRecord(document, now));
+  const existingById = new Map(previousDocuments.map((document) => [document.id, document]));
+  const nextIds = new Set(nextDocuments.map((document) => document.id));
+  const changedSourceIds: string[] = [];
+  const changedDocuments: ArchivedSourceDocumentRecord[] = [];
+  const activeDocuments: ArchivedSourceDocumentRecord[] = [];
+  let created = 0;
+  let updated = 0;
+  let unchanged = 0;
+
+  for (const document of nextDocuments) {
+    const existing = existingById.get(document.id);
+
+    if (!existing) {
+      activeDocuments.push(document);
+      changedSourceIds.push(document.id);
+      changedDocuments.push(document);
+      created += 1;
+    } else if (areEquivalent(existing, document)) {
+      activeDocuments.push(existing);
+      unchanged += 1;
+    } else {
+      activeDocuments.push(document);
+      changedSourceIds.push(document.id);
+      changedDocuments.push(document);
+      updated += 1;
+    }
+  }
+
+  const deletedSourceIds = previousDocuments
+    .filter((document) => !nextIds.has(document.id))
+    .map((document) => document.id);
+  changedSourceIds.push(...deletedSourceIds);
+
+  return {
+    created,
+    updated,
+    unchanged,
+    deleted: deletedSourceIds.length,
+    previousRepoSourceIds: previousDocuments.map((document) => document.id),
+    changedSourceIds,
+    changedDocuments,
+    activeDocuments: activeDocuments.sort(compareDocuments),
+    previousDocuments: previousDocuments.slice(),
+  };
+}
+
+function sameDocumentSet(
+  left: ArchivedSourceDocumentRecord[],
+  right: ArchivedSourceDocumentRecord[],
+): boolean {
+  return JSON.stringify(left) === JSON.stringify(right);
+}
+
 function comparableFields(document: ArchivedSourceDocumentRecord): Record<string, unknown> {
   return {
     id: document.id,

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

@@ -31,8 +31,9 @@ export class SourceRagPipeline {
       forceReindex?: boolean;
     },
   ): Promise<SourceRagIngestResult> {
-    const mutation = await this.archiveRepository.syncRepoDocuments(repoName, documents);
+    const mutation = await this.archiveRepository.planRepoDocuments(repoName, documents);
     await this.replaceChangedChunks(repoName, mutation, options?.forceReindex ?? false);
+    await this.archiveRepository.commitRepoDocuments(repoName, mutation);
 
     return {
       createdDocuments: mutation.created,

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

@@ -0,0 +1,78 @@
+import { mkdtemp, readFile, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+import {
+  FileSourceDocumentArchiveRepository,
+  InMemoryVectorStore,
+  SourceDocumentIngester,
+  SourceRagPipeline,
+  type ArchivedSourceDocument,
+} from "@voidbot/rag";
+
+class FailingVectorStore extends InMemoryVectorStore {
+  public override async upsert(): Promise<void> {
+    throw new Error("fixture vector failure");
+  }
+}
+
+const sourceId = "Fixture:README.md";
+
+async function main(): Promise<void> {
+  const root = await mkdtemp(join(tmpdir(), "voidbot-source-commit-"));
+  const archivePath = join(root, "source-documents.json");
+
+  try {
+    const archive = new FileSourceDocumentArchiveRepository(archivePath);
+    const original = document("original body");
+    const updated = document("updated body");
+    await archive.syncRepoDocuments("Fixture", [original]);
+
+    const failingPipeline = new SourceRagPipeline(
+      archive,
+      new SourceDocumentIngester(),
+      new FailingVectorStore(),
+    );
+
+    await failingPipeline.syncRepoDocuments("Fixture", [updated]).then(
+      () => { throw new Error("Expected vector failure did not occur."); },
+      () => undefined,
+    );
+
+    assertEqual((await archive.get(sourceId))?.content, original.content, "archive advanced after vector failure");
+
+    const workingPipeline = new SourceRagPipeline(
+      archive,
+      new SourceDocumentIngester(),
+      new InMemoryVectorStore(),
+    );
+    await workingPipeline.syncRepoDocuments("Fixture", [updated]);
+    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[] };
+    assertEqual(manifest.repos?.length, 1, "manifest summary was not committed");
+    console.log("Source RAG commit-boundary smoke passed.");
+  } finally {
+    await rm(root, { recursive: true, force: true });
+  }
+}
+
+void main().catch((error) => {
+  console.error(error);
+  process.exitCode = 1;
+});
+
+function document(content: string): ArchivedSourceDocument {
+  return {
+    id: sourceId,
+    repoName: "Fixture",
+    path: "README.md",
+    content,
+  };
+}
+
+function assertEqual(actual: unknown, expected: unknown, message: string): void {
+  if (actual !== expected) {
+    throw new Error(`${message}: expected ${String(expected)}, got ${String(actual)}`);
+  }
+}

state/map.yaml

@@ -15,10 +15,11 @@ objective:
     safely.
   current: >
     Keep the recovered read-only archive/source retrieval organ healthy on
-    surviving old Yggdrasil, and design an admitted source-index refresh owner.
-    The query path is live, but no current runtime owns incremental source
-    crawling or coordinated archive/Qdrant writes, so most source shards remain
-    at the 2026-06-03 recovery snapshot. Keep Discord, worker jobs, Personas,
+    surviving old Yggdrasil, and restore source-index refresh under a newly
+    admitted writer authority. The operator has authorized bringing indexing
+    back up. The query path remains live and read-only; the new writer must own
+    incremental source crawling plus coordinated archive/Qdrant writes without
+    entering the retrieval MCP boundary. Keep Discord, worker jobs, Personas,
     moderation, and delivery stopped until their missing authority and
     credentials are deliberately rebuilt.
 
@@ -33,6 +34,7 @@ constraints:
 
 invariants:
   - The recovery MCP organ is read-only and loopback-only. It may consume exact archives, VoidBot-owned vector collections, and the configured embedding provider; it cannot start Discord, worker-job, Persona, moderation, Bifrost, Idunn, or canon mutation paths.
+  - Source refresh is a separate writer authority from `voidbot-retrieval.service`. Restoring indexing must not add mutation tools, crawler lifecycle, or archive/Qdrant write ownership to the read-only MCP process.
   - Deployment follows one path: push the intended upstream commit, then let Idunn deploy VoidBot on Yggdrasil. Do not create, restore, or search for a local VoidBot deployment path.
   - The map is canonical and slow-changing.
   - Scratch is temporary and disposable.
@@ -293,4 +295,4 @@ rejected_paths:
 
 current_status:
   summary: Yggdrasil release `11568671a0d72c4e273c2b76d0a2a642788cc347` runs the typed `voidbot.swarm` Eve publisher and the separate read-only `voidbot-retrieval.service`. The retrieval MCP and Qdrant are healthy: 49 source repositories are visible, source search succeeds, and the green source collection contains 98,170 points. The recovery body has no source-index writer, repo crawler, hook consumer, or reconciliation scheduler; retrieval availability therefore does not imply corpus freshness. The Discord bot, worker, Persona scheduler, Postgres, and resident cognition organs remain absent.
-  next_action: Produce and approve an authority map for source refresh before starting any writer: one owner, its exact repository inputs, archive/Qdrant commit boundary, failure rollback, scheduling trigger, progress surface, and deployment path. Do not use a manual full reindex or revive the retired workstation stack as a diagnostic shortcut.
+  next_action: Restore source refresh under one explicit writer owner. Map its exact repository inputs, archive/Qdrant commit boundary, failure rollback, scheduling trigger, progress surface, and deployment path before the first write, then run the smallest bounded incremental catch-up that proves archive and vectors advance together. Keep `voidbot-retrieval.service` read-only; do not use a manual full reindex or revive the retired workstation stack.

state/scratch.md

@@ -2,8 +2,8 @@
 
 ## Current Subgoal
 
-Inspect the health of the deliberately narrow Yggdrasil VoidBot body, with
-source retrieval/index availability as the priority failure.
+Restore source indexing on Yggdrasil under the operator's explicit 2026-08-30
+authorization while preserving the recovered read-only retrieval boundary.
 
 ## Working Notes
 
@@ -15,7 +15,13 @@ source retrieval/index availability as the priority failure.
   projection.
 - The recovery corpus is read-only. Diagnose retrieval service, tunnel,
   embedding provider, Qdrant collections, and recovered archive independently;
-  do not start indexing writers as a health-check side effect.
+  the new indexing writer must remain a separate authority and lifecycle.
+- Before the first write, the source-refresh owner must name exact repository
+  inputs, coordinated archive/Qdrant commit and rollback behavior, scheduling
+  trigger, durable progress witness, and deployment path.
+- Start with the smallest bounded incremental catch-up that proves archive and
+  vectors advance together. A manual full reindex and the retired workstation
+  stack remain rejected shortcuts.
 - Last durable retrieval proof is the 2026-08-16 exact-document witness for
   release `20260816T193315Z-c88c71909dbc`; it is historical evidence, not a
   present-health result.