fix: reindex embeddings after config drift

src/utils/pipeline-factory.test.ts

@@ -0,0 +1,121 @@
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+import { parseConfig } from "../config.js";
+import type { IMemoryStore } from "../core/store/types.js";
+import type { EmbeddingService } from "../core/store/embedding.js";
+import { createPipeline, resetStores } from "./pipeline-factory.js";
+
+const createStoreBundleMock = vi.hoisted(() => vi.fn());
+
+vi.mock("../core/store/factory.js", () => ({
+  createStoreBundle: createStoreBundleMock,
+}));
+
+function createLogger() {
+  return {
+    debug: vi.fn(),
+    info: vi.fn(),
+    warn: vi.fn(),
+    error: vi.fn(),
+  };
+}
+
+function createEmbeddingService(): EmbeddingService {
+  return {
+    embed: vi.fn(async () => new Float32Array([1, 0])),
+    embedBatch: vi.fn(async (texts: string[]) => texts.map(() => new Float32Array([1, 0]))),
+    getDimensions: vi.fn(() => 2),
+    getProviderInfo: vi.fn(() => ({ provider: "openai", model: "test-embedding" })),
+    isReady: vi.fn(() => true),
+    startWarmup: vi.fn(),
+    close: vi.fn(),
+  };
+}
+
+function createMemoryStore(): IMemoryStore {
+  return {
+    init: vi.fn(async () => ({ needsReindex: true, reason: "embedding model changed" })),
+    isDegraded: vi.fn(() => false),
+    getCapabilities: vi.fn(() => ({
+      vectorSearch: true,
+      ftsSearch: true,
+      nativeHybridSearch: false,
+      sparseVectors: false,
+    })),
+    close: vi.fn(),
+    upsertL1: vi.fn(),
+    deleteL1: vi.fn(),
+    deleteL1Batch: vi.fn(),
+    deleteL1Expired: vi.fn(),
+    countL1: vi.fn(),
+    queryL1Records: vi.fn(),
+    getAllL1Texts: vi.fn(),
+    searchL1Vector: vi.fn(),
+    searchL1Fts: vi.fn(),
+    upsertL0: vi.fn(),
+    deleteL0: vi.fn(),
+    deleteL0Expired: vi.fn(),
+    countL0: vi.fn(),
+    queryL0ForL1: vi.fn(),
+    queryL0GroupedBySessionId: vi.fn(),
+    getAllL0Texts: vi.fn(),
+    searchL0Vector: vi.fn(),
+    searchL0Fts: vi.fn(),
+    reindexAll: vi.fn(async (embedFn: (text: string) => Promise<Float32Array>, onProgress) => {
+      await embedFn("existing L1 memory");
+      onProgress?.(1, 1, "L1");
+      return { l1Count: 1, l0Count: 0 };
+    }),
+    isFtsAvailable: vi.fn(() => true),
+  } as unknown as IMemoryStore;
+}
+
+describe("createPipeline", () => {
+  let tempDir: string | undefined;
+
+  afterEach(() => {
+    resetStores();
+    if (tempDir) {
+      rmSync(tempDir, { recursive: true, force: true });
+      tempDir = undefined;
+    }
+  });
+
+  it("reindexes existing records when store init reports embedding drift", async () => {
+    tempDir = mkdtempSync(path.join(tmpdir(), "pipeline-factory-"));
+    const store = createMemoryStore();
+    const embeddingService = createEmbeddingService();
+    createStoreBundleMock.mockReturnValue({
+      store,
+      embedding: embeddingService,
+      storeSnapshot: { type: "sqlite", sqlitePath: "vectors.db" },
+    });
+
+    const cfg = parseConfig({
+      embedding: {
+        provider: "openai",
+        baseUrl: "https://embedding.example/v1",
+        apiKey: "test-key",
+        model: "test-embedding",
+        dimensions: 2,
+      },
+    });
+
+    const pipeline = await createPipeline({
+      pluginDataDir: tempDir,
+      cfg,
+      openclawConfig: {},
+      logger: createLogger(),
+    });
+
+    try {
+      expect(store.reindexAll).toHaveBeenCalledTimes(1);
+      expect(embeddingService.embed).toHaveBeenCalledWith("existing L1 memory");
+    } finally {
+      await pipeline.destroy();
+    }
+  });
+});

src/utils/pipeline-factory.ts

@@ -229,6 +229,18 @@ async function _doInitStores(
       } catch (err) {
         logger.warn(`${TAG} Failed to read/write manifest (non-fatal): ${err instanceof Error ? err.message : String(err)}`);
       }
+
+      const reindexed = await reindexStoreIfNeeded({
+        vectorStore,
+        embeddingService,
+        needsReindex,
+        reindexReason,
+        logger,
+      });
+      if (reindexed) {
+        needsReindex = false;
+        reindexReason = undefined;
+      }
     }
   } catch (err) {
     logger.warn(
@@ -241,6 +253,42 @@ async function _doInitStores(
   return { vectorStore, embeddingService, needsReindex, reindexReason };
 }
 
+async function reindexStoreIfNeeded(opts: {
+  vectorStore: IMemoryStore | undefined;
+  embeddingService: EmbeddingService | undefined;
+  needsReindex: boolean;
+  reindexReason?: string;
+  logger: PipelineLogger;
+}): Promise<boolean> {
+  const { vectorStore, embeddingService, needsReindex, reindexReason, logger } = opts;
+  if (!needsReindex) return false;
+
+  const reason = reindexReason ?? "embedding config changed";
+  if (!vectorStore || !embeddingService) {
+    logger.warn(
+      `${TAG} Reindex needed (${reason}) but store or embedding service is unavailable; existing vectors were not refreshed`,
+    );
+    return false;
+  }
+
+  try {
+    logger.info(`${TAG} Reindex triggered: ${reason}. Re-embedding existing records...`);
+    const result = await vectorStore.reindexAll(
+      (text) => embeddingService.embed(text),
+      (done, total, layer) => {
+        if (total > 0 && (done === total || done % 20 === 0)) {
+          logger.debug?.(`${TAG} Reindex progress: ${layer} ${done}/${total}`);
+        }
+      },
+    );
+    logger.info(`${TAG} Reindex complete: L1=${result.l1Count}, L0=${result.l0Count}`);
+    return true;
+  } catch (err) {
+    logger.warn(`${TAG} Reindex failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`);
+    return false;
+  }
+}
+
 // ============================
 // L1 Runner factory
 // ============================