From 4ecfdbb23898f8ffac1819ef93dcd7171d8d37bc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E5=86=A0=E8=BE=B0?= Date: Sat, 27 Jun 2026 17:46:46 +0800 Subject: [PATCH] fix(pipeline): reindex embeddings after config drift --- src/utils/pipeline-factory.test.ts | 121 +++++++++++++++++++++++++++++ src/utils/pipeline-factory.ts | 48 ++++++++++++ 2 files changed, 169 insertions(+) create mode 100644 src/utils/pipeline-factory.test.ts diff --git a/src/utils/pipeline-factory.test.ts b/src/utils/pipeline-factory.test.ts new file mode 100644 index 00000000..14f19dcf --- /dev/null +++ b/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, 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(); + } + }); +}); diff --git a/src/utils/pipeline-factory.ts b/src/utils/pipeline-factory.ts index ce87801c..e9c3af10 100644 --- a/src/utils/pipeline-factory.ts +++ b/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 { + 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 // ============================