diff --git a/config/assertion-safety-baseline.txt b/config/assertion-safety-baseline.txt index a5b2067b8739..2feeb8cfb372 100644 --- a/config/assertion-safety-baseline.txt +++ b/config/assertion-safety-baseline.txt @@ -727,7 +727,7 @@ extensions/memory-core/src/memory/manager-embedding-ops.ts 1 extensions/memory-core/src/memory/manager-keyword-retrieval.ts 1 extensions/memory-core/src/memory/manager-search-orchestration.ts 1 extensions/memory-core/src/memory/manager-search.ts 6 -extensions/memory-core/src/memory/manager-sync-base.ts 4 +extensions/memory-core/src/memory/manager-sync-base.ts 3 extensions/memory-core/src/memory/manager-vector-rebuild-state.ts 1 extensions/memory-core/src/memory/manager-watch-ops.ts 4 extensions/memory-core/src/memory/manager.ts 2 diff --git a/docs/cli/memory.md b/docs/cli/memory.md index 1ac8efcb2ec3..729b7691718d 100644 --- a/docs/cli/memory.md +++ b/docs/cli/memory.md @@ -282,9 +282,16 @@ review what remains. The purge removes matching promotion-marker entries and session-reference sections from scanned memory files, selected session-corpus lines, and selected-session transcript index chunks. It clears associated full-text and -vector rows, cached embeddings, matching short-term state, ingestion seen-hash -scopes, and origin rows. Matching content is scrubbed from dreaming rewrite -backups, rather than deleting every backup. +vector rows, matching short-term state, ingestion seen-hash scopes, and origin +rows. Matching content is scrubbed from dreaming rewrite backups, rather than +deleting every backup. + +For a nonempty session selection, Forget also clears the selected agent's entire +embedding cache, including results retained from unfinished index rebuilds. Those +results may not yet be linked to published index chunks. Unrelated published +index entries remain usable, but later indexing may need to regenerate their +cached embeddings. The dry-run report includes the whole-cache removal count; +a dry run or an empty session selection does not clear the cache. Consolidation preserves origins for replaced promotion markers while retained rewrite preimages reference them. Those origins are pruned only after live diff --git a/docs/concepts/memory-provenance.md b/docs/concepts/memory-provenance.md index 3a2b5e53446f..d0ff55363662 100644 --- a/docs/concepts/memory-provenance.md +++ b/docs/concepts/memory-provenance.md @@ -192,9 +192,14 @@ merged prose. Surviving sources may support a new entry later, but automatic reconstruction is not guaranteed. The cleanup covers matching promoted entries, session-corpus lines, memory -index chunks and their full-text/vector rows, cached embeddings, short-term -state, ingestion deduplication state, and dreaming rewrite preimages. It also -removes whole lines containing exact selected corpus snippets from scanned +index chunks and their full-text/vector rows, short-term state, ingestion +deduplication state, and dreaming rewrite preimages. A nonempty session selection +also clears the selected agent's entire embedding cache, including results from +unfinished rebuilds that are not yet linked to published chunks. Unrelated +published index entries remain usable; later indexing may need to regenerate +cached embeddings. Dry runs report this removal without applying it. + +The purge also removes whole lines containing exact selected corpus snippets from scanned memory files and dream diaries. The [command reference](/cli/memory#memory-forget) describes the counters and selection limits. diff --git a/extensions/memory-core/src/memory-forget.test.ts b/extensions/memory-core/src/memory-forget.test.ts index b53bfe682fce..f736aee54f77 100644 --- a/extensions/memory-core/src/memory-forget.test.ts +++ b/extensions/memory-core/src/memory-forget.test.ts @@ -48,6 +48,27 @@ describe("memory forget", () => { }), ); + it.each([true, false])( + "reports no cache deletion for an empty selection (dryRun=%s)", + async (dryRun) => { + const db = openOpenClawAgentDatabase({ agentId: "main" }).db; + db.prepare(`INSERT INTO memory_embedding_cache + (provider, model, provider_key, hash, embedding, dims, updated_at) + VALUES ('test', 'test', 'test', 'unrelated', '[1,0]', 2, 1)`).run(); + const report = await forgetMemoryEntries({ + cfg, + agentId: "main", + hookSources: ["no-matching-source"], + dryRun, + }); + expect(report.sessionIds).toEqual([]); + expect(report.artifacts.embeddingCacheRows).toBe(0); + expect(db.prepare("SELECT hash FROM memory_embedding_cache").all()).toEqual([ + { hash: "unrelated" }, + ]); + }, + ); + it.each([ { label: "session ID", selector: "archived" }, { label: "session key", selector: "agent:main:archived" }, @@ -236,6 +257,11 @@ describe("memory forget", () => { VALUES ('unrelated', 'MEMORY.md', 'memory', 1, 2, 'unrelated-hash', 'test', 'Keep this.', '[1,0]', 1)`, ).run(); + db.prepare( + `INSERT INTO memory_embedding_cache + (provider, model, provider_key, hash, embedding, dims, updated_at) + VALUES ('test', 'test', 'test', 'unrelated-hash', '[1,0]', 2, 1)`, + ).run(); const preview = await forgetMemoryEntries({ cfg, @@ -246,8 +272,13 @@ describe("memory forget", () => { expect(preview).toMatchObject({ sessionIds: ["unknown-session"], sessionResolutions: [{ sessionId: "unknown-session", source: "unresolved" }], + artifacts: { embeddingCacheRows: 1 }, }); - expect(Object.values(preview.artifacts).every((count) => count === 0)).toBe(true); + expect( + Object.entries(preview.artifacts) + .filter(([name]) => name !== "embeddingCacheRows") + .every(([, count]) => count === 0), + ).toBe(true); expect(listMemorySessionTombstones({ agentId: "main" })).toEqual([]); const report = await forgetMemoryEntries({ @@ -260,10 +291,14 @@ describe("memory forget", () => { expect(tombstones).toMatchObject([{ sessionId: "unknown-session", reason: "forgotten" }]); expect( await forgetMemoryEntries({ cfg, agentId: "main", sessionIds: ["unknown-session"] }), - ).toEqual(report); + ).toEqual({ + ...report, + artifacts: { ...report.artifacts, embeddingCacheRows: 0 }, + }); expect(listMemorySessionTombstones({ agentId: "main" })).toEqual(tombstones); expect(await fs.readFile(memoryPath, "utf8")).toBe(content); expect(db.prepare("SELECT id FROM memory_index_chunks").all()).toEqual([{ id: "unrelated" }]); + expect(db.prepare("SELECT hash FROM memory_embedding_cache").all()).toEqual([]); expect( await readMemoryCoreWorkspaceEntries({ namespace: DREAMING_MEMORY_BACKUP_NAMESPACE, @@ -665,7 +700,7 @@ describe("memory forget", () => { indexSources: 3, ftsRows: 4, vectorRows: 4, - embeddingCacheRows: 4, + embeddingCacheRows: 5, shortTermEntries: 1, seenHashScopes: 1, backups: 1, @@ -764,12 +799,18 @@ describe("memory forget", () => { "memory_index_chunks_fts", "memory_index_chunks_vec", "memory_index_chunk_provenance", - "memory_embedding_cache", ]) { expect( (db.prepare(`SELECT COUNT(*) AS count FROM ${table}`).get() as { count: number }).count, ).toBe(1); } + expect( + ( + db.prepare("SELECT COUNT(*) AS count FROM memory_embedding_cache").get() as { + count: number; + } + ).count, + ).toBe(0); expect( (db.prepare("SELECT path FROM memory_index_sources").all() as Array<{ path: string }>).map( (row) => row.path, diff --git a/extensions/memory-core/src/memory-forget.ts b/extensions/memory-core/src/memory-forget.ts index c19b69d31dec..99d0ae6e35ca 100644 --- a/extensions/memory-core/src/memory-forget.ts +++ b/extensions/memory-core/src/memory-forget.ts @@ -238,7 +238,6 @@ async function planMemoryIndex(params: { (source.source === "sessions" && removedSessionPaths.has(source.path)), ); const chunkIds = chunks.map((chunk) => chunk.id); - const chunkHashes = [...new Set(chunks.map((chunk) => chunk.hash))]; const ftsRows = chunkIds.length > 0 && tableExists(db, "memory_index_chunks_fts") ? executeSqliteQuerySync( @@ -247,16 +246,14 @@ async function planMemoryIndex(params: { ).rows.length : 0; const hasVectorTable = tableExists(db, "memory_index_chunks_vec"); - const embeddingCacheRows = - chunkHashes.length > 0 && tableExists(db, "memory_embedding_cache") - ? executeSqliteQuerySync( - db, - kysely - .selectFrom("memory_embedding_cache") - .select("hash") - .where("hash", "in", chunkHashes), - ).rows.length - : 0; + let embeddingCacheRows = 0; + if (params.sessionIds.size > 0 && tableExists(db, "memory_embedding_cache")) { + const cacheCount = db + .prepare("SELECT COUNT(*) AS count FROM memory_embedding_cache") + // SAFETY: the aggregate query always returns one row with the declared count alias. + .get() as { count?: unknown }; + embeddingCacheRows = Number(cacheCount.count ?? 0); + } return { chunks, sources, ftsRows, embeddingCacheRows, hasVectorTable, databasePath }; }, { agentId: params.agentId }, @@ -598,7 +595,6 @@ async function forgetWorkspaceMemory( try { const kysely = getNodeSqliteKysely(db); const chunkIds = indexPlan.chunks.map((chunk) => chunk.id); - const chunkHashes = [...new Set(indexPlan.chunks.map((chunk) => chunk.hash))]; if (chunkIds.length > 0 && indexPlan.hasVectorTable) { const loaded = await loadSqliteVecExtension({ db }); if (!loaded.ok) { @@ -614,9 +610,9 @@ async function forgetWorkspaceMemory( agentId: params.agentId, sessionIds: [...sessionIds], }); - if (recorded === 0 && changedPaths.size > 0) { - // Repeating a partial purge can still rewrite an unindexed file. Fence - // pending shadow rebuilds before any filesystem mutation, even on failure. + if (recorded === 0) { + // Every explicit purge invalidates in-flight cache work, including a + // repeated purge whose selected source was already scrubbed. executeSqliteQuerySync( db, kysely @@ -657,11 +653,8 @@ async function forgetWorkspaceMemory( .where("source", "=", source.source), ); } - if (indexPlan.embeddingCacheRows > 0) { - executeSqliteQuerySync( - db, - kysely.deleteFrom("memory_embedding_cache").where("hash", "in", chunkHashes), - ); + if (tableExists(db, "memory_embedding_cache")) { + executeSqliteQuerySync(db, kysely.deleteFrom("memory_embedding_cache")); } }); if (retainedShortTerm.length !== shortTermEntries.length) { diff --git a/extensions/memory-core/src/memory/manager-chunk-writer.test.ts b/extensions/memory-core/src/memory/manager-chunk-writer.test.ts index 72b6285f6d8f..457b06ef3dfc 100644 --- a/extensions/memory-core/src/memory/manager-chunk-writer.test.ts +++ b/extensions/memory-core/src/memory/manager-chunk-writer.test.ts @@ -126,11 +126,13 @@ describe("memory chunk publication", () => { [ "memory_index_sources", ...CHUNK_WRITE_TABLES, - "memory_embedding_cache", "memory_index_chunks_fts", "memory_index_state", ].map((table) => db.prepare(`SELECT * FROM ${table} ORDER BY rowid`).all()); const before = snapshot(); + const cacheSnapshot = () => + db.prepare("SELECT * FROM memory_embedding_cache ORDER BY rowid").all(); + const cacheBefore = cacheSnapshot(); expect(before[1]?.some((row) => String(row.text).includes("Alpha memory line."))).toBe( true, ); @@ -158,9 +160,16 @@ describe("memory chunk publication", () => { prepare.mockRestore(); } expect(snapshot()).toEqual(before); + // Completed provider work is durable even when index publication rolls back. + const retainedCache = cacheSnapshot(); + expect(retainedCache).toEqual(expect.arrayContaining(cacheBefore)); + expect(retainedCache).toHaveLength(cacheBefore.length + 1); + const completedRequests = fixture.provider.embedBatchCalls; db.exec("DROP TRIGGER fail_chunk_publication"); await manager.sync({ reason: "retry" }); + expect(fixture.provider.embedBatchCalls).toBe(completedRequests); + expect(cacheSnapshot()).toEqual(retainedCache); expect( db .prepare("SELECT text FROM memory_index_chunks WHERE path LIKE ? AND source = ?") diff --git a/extensions/memory-core/src/memory/manager-db.test.ts b/extensions/memory-core/src/memory/manager-db.test.ts index 471aff3722c5..439dcac196c8 100644 --- a/extensions/memory-core/src/memory/manager-db.test.ts +++ b/extensions/memory-core/src/memory/manager-db.test.ts @@ -486,21 +486,28 @@ describe("memory manager database publication", () => { } }); - it("preserves the live embedding cache when the shadow index has caching disabled", async () => { + it("preserves the live embedding cache instead of publishing the shadow cache", async () => { const targetPath = path.join(fixtureRoot, "target.sqlite"); const sourcePath = path.join(fixtureRoot, "source.sqlite"); const targetDb = new DatabaseSync(targetPath); const sourceDb = new DatabaseSync(sourcePath); try { ensureTestMemorySchema(targetDb); - ensureTestMemorySchema(sourceDb, false); + ensureTestMemorySchema(sourceDb); targetDb .prepare( `INSERT INTO memory_embedding_cache ( provider, model, provider_key, hash, embedding, dims, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?)`, ) - .run("test", "model", "key", "hash", "[]", 0, 1); + .run("test", "model", "key", "live-hash", "[]", 0, 1); + sourceDb + .prepare( + `INSERT INTO memory_embedding_cache ( + provider, model, provider_key, hash, embedding, dims, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?)`, + ) + .run("test", "model", "key", "shadow-hash", "[1]", 1, 2); sourceDb.close(); await publishMemoryDatabaseTables({ @@ -512,7 +519,7 @@ describe("memory manager database publication", () => { }); expect(targetDb.prepare("SELECT hash FROM memory_embedding_cache").all()).toEqual([ - { hash: "hash" }, + { hash: "live-hash" }, ]); } finally { try { diff --git a/extensions/memory-core/src/memory/manager-db.ts b/extensions/memory-core/src/memory/manager-db.ts index f64f1b91db59..234d091cc672 100644 --- a/extensions/memory-core/src/memory/manager-db.ts +++ b/extensions/memory-core/src/memory/manager-db.ts @@ -299,17 +299,6 @@ export async function publishMemoryDatabaseTables(params: { FROM ${MEMORY_REINDEX_SCHEMA}.memory_index_chunk_provenance; `); - if (tableExists(params.targetDb, MEMORY_REINDEX_SCHEMA, "memory_embedding_cache")) { - params.targetDb.exec(` - DELETE FROM main.memory_embedding_cache; - INSERT INTO main.memory_embedding_cache ( - provider, model, provider_key, hash, embedding, dims, updated_at - ) - SELECT provider, model, provider_key, hash, embedding, dims, updated_at - FROM ${MEMORY_REINDEX_SCHEMA}.memory_embedding_cache; - `); - } - replaceVirtualTable({ db: params.targetDb, tableName: "memory_index_chunks_fts", diff --git a/extensions/memory-core/src/memory/manager-embedding-cache.test.ts b/extensions/memory-core/src/memory/manager-embedding-cache.test.ts index e83acf572348..7370f3ab83fa 100644 --- a/extensions/memory-core/src/memory/manager-embedding-cache.test.ts +++ b/extensions/memory-core/src/memory/manager-embedding-cache.test.ts @@ -77,6 +77,49 @@ describe("memory embedding cache", () => { } }); + it("reserves space before replacing cached vectors at capacity", () => { + const db = createDb(); + const provider = { id: "local", model: "fixture" }; + try { + upsertMemoryEmbeddingCache({ + db, + enabled: true, + provider, + providerKey: "fixture", + entries: [ + { hash: "a", embedding: [1] }, + { hash: "b", embedding: [2] }, + ], + now: 1, + }); + db.exec(`CREATE TEMP TRIGGER reject_cache_overflow BEFORE INSERT ON memory_embedding_cache + WHEN (SELECT COUNT(*) FROM memory_embedding_cache) >= 2 + BEGIN SELECT RAISE(ABORT, 'cache overflow'); END;`); + db.exec("BEGIN IMMEDIATE"); + upsertMemoryEmbeddingCache({ + db, + enabled: true, + provider, + providerKey: "fixture", + maxEntries: 2, + entries: [ + { hash: "a", embedding: [3] }, + { hash: "a", embedding: [4] }, + ], + now: 2, + }); + db.exec("COMMIT"); + expect( + db.prepare("SELECT hash, embedding FROM memory_embedding_cache ORDER BY hash").all(), + ).toEqual([ + { hash: "a", embedding: "[4]" }, + { hash: "b", embedding: "[2]" }, + ]); + } finally { + db.close(); + } + }); + it("loads provider-declared alias cache rows without accepting arbitrary identities", () => { const db = createDb(); try { diff --git a/extensions/memory-core/src/memory/manager-embedding-cache.ts b/extensions/memory-core/src/memory/manager-embedding-cache.ts index 87bf28b038bf..3ea481d0ca94 100644 --- a/extensions/memory-core/src/memory/manager-embedding-cache.ts +++ b/extensions/memory-core/src/memory/manager-embedding-cache.ts @@ -6,12 +6,14 @@ import { } from "openclaw/plugin-sdk/memory-core-host-engine-storage"; import { compileSqliteQueryBindings, + executeSqliteQuerySync, + type Generated, getNodeSqliteKysely, iterateSqliteQuerySync, } from "openclaw/plugin-sdk/sqlite-runtime"; import type { MemoryIndexProviderIdentity } from "./manager-reindex-state.js"; -export type MemoryEmbeddingCacheRow = { +type MemoryEmbeddingCacheRow = { provider: string; model: string; provider_key: string; @@ -22,9 +24,19 @@ export type MemoryEmbeddingCacheRow = { }; type EmbeddingCacheDatabase = { - memory_embedding_cache: MemoryEmbeddingCacheRow; + memory_embedding_cache: MemoryEmbeddingCacheRow & { rowid: Generated }; }; +/** Require a finite, nonempty vector compatible with the active embedding dimensions. */ +export function isValidMemoryEmbedding(embedding: number[], dimensions?: number): boolean { + return ( + Array.isArray(embedding) && + embedding.length > 0 && + (dimensions === undefined || embedding.length === dimensions) && + embedding.every((coordinate) => typeof coordinate === "number" && Number.isFinite(coordinate)) + ); +} + export function loadMemoryEmbeddingCache(params: { db: DatabaseSync; enabled: boolean; @@ -58,7 +70,8 @@ export function loadMemoryEmbeddingCache(params: { .where("hash", "in", batch); for (const row of iterateSqliteQuerySync(params.db, query)) { // The first stored row wins even when its vector needs to be regenerated. - out.set(row.hash, parseEmbedding(row.embedding)); + const embedding = parseEmbedding(row.embedding); + out.set(row.hash, isValidMemoryEmbedding(embedding) ? embedding : []); unresolved.delete(row.hash); } } @@ -66,7 +79,25 @@ export function loadMemoryEmbeddingCache(params: { return out; } -export function prepareMemoryEmbeddingCacheUpsert(db: DatabaseSync) { +/** Discard ambiguous vector spaces without removing unrelated provider caches or index rows. */ +export function clearMemoryEmbeddingCacheIdentities( + database: DatabaseSync, + identities: MemoryIndexProviderIdentity[], +): void { + const db = getNodeSqliteKysely(database); + for (const identity of identities) { + executeSqliteQuerySync( + database, + db + .deleteFrom("memory_embedding_cache") + .where("provider", "=", identity.provider) + .where("model", "=", identity.model) + .where("provider_key", "=", identity.providerKey), + ); + } +} + +function prepareMemoryEmbeddingCacheUpsert(db: DatabaseSync) { const { compiled, bind } = compileSqliteQueryBindings((parameter) => getNodeSqliteKysely(db) .insertInto("memory_embedding_cache") @@ -98,15 +129,46 @@ export function upsertMemoryEmbeddingCache(params: { provider: { id: string; model: string } | null; providerKey: string | null; entries: Array<{ hash: string; embedding: number[] }>; + maxEntries?: number; now?: number; }): void { const provider = params.provider; if (!params.enabled || !provider || !params.providerKey || params.entries.length === 0) { return; } + const seenHashes = new Set(); + const uniqueEntries: Array<{ hash: string; embedding: number[] }> = []; + for (let index = params.entries.length - 1; index >= 0; index -= 1) { + const entry = params.entries[index]; + if (entry && !seenHashes.has(entry.hash)) { + seenHashes.add(entry.hash); + uniqueEntries.push(entry); + } + } + uniqueEntries.reverse(); + const maxEntries = + typeof params.maxEntries === "number" && + Number.isFinite(params.maxEntries) && + params.maxEntries > 0 + ? Math.floor(params.maxEntries) + : undefined; + const retainedEntries = + maxEntries === undefined ? uniqueEntries : uniqueEntries.slice(-maxEntries); + if (retainedEntries.length === 0) { + return; + } + if (maxEntries !== undefined) { + reserveMemoryEmbeddingCacheCapacity({ + db: params.db, + provider, + providerKey: params.providerKey, + hashes: retainedEntries.map((entry) => entry.hash), + maxEntries, + }); + } const now = params.now ?? Date.now(); const upsert = prepareMemoryEmbeddingCacheUpsert(params.db); - for (const entry of params.entries) { + for (const entry of retainedEntries) { const embedding = entry.embedding ?? []; upsert({ provider: provider.id, @@ -120,6 +182,44 @@ export function upsertMemoryEmbeddingCache(params: { } } +function reserveMemoryEmbeddingCacheCapacity(params: { + db: DatabaseSync; + provider: { id: string; model: string }; + providerKey: string; + hashes: string[]; + maxEntries: number; +}): void { + const db = getNodeSqliteKysely(params.db); + // The caller's transaction replaces incoming rows and reserves space before + // inserting vectors, so even a transient row-count overflow is impossible. + for (let start = 0; start < params.hashes.length; start += 400) { + executeSqliteQuerySync( + params.db, + db + .deleteFrom("memory_embedding_cache") + .where("provider", "=", params.provider.id) + .where("model", "=", params.provider.model) + .where("provider_key", "=", params.providerKey) + .where("hash", "in", params.hashes.slice(start, start + 400)), + ); + } + // SQLite performs eviction without materializing the full cache in JavaScript. + executeSqliteQuerySync( + params.db, + db.deleteFrom("memory_embedding_cache").where( + "rowid", + "in", + db + .selectFrom("memory_embedding_cache") + .select("rowid") + .orderBy("updated_at", "desc") + .orderBy("rowid", "desc") + .limit(-1) + .offset(params.maxEntries - params.hashes.length), + ), + ); +} + export function collectMemoryCachedEmbeddings>(params: { chunks: T[]; cached: Map; diff --git a/extensions/memory-core/src/memory/manager-embedding-ops.ts b/extensions/memory-core/src/memory/manager-embedding-ops.ts index f4e2f53f4ce1..b80f79851da8 100644 --- a/extensions/memory-core/src/memory/manager-embedding-ops.ts +++ b/extensions/memory-core/src/memory/manager-embedding-ops.ts @@ -32,15 +32,21 @@ import { } from "openclaw/plugin-sdk/memory-core-host-engine-storage"; import { MAX_TIMER_TIMEOUT_MS, resolveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime"; import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; -import { runSqliteImmediateTransaction } from "openclaw/plugin-sdk/sqlite-runtime"; +import { + runSqliteImmediateTransaction, + runSqliteImmediateTransactionSync, +} from "openclaw/plugin-sdk/sqlite-runtime"; import { chunkItems } from "openclaw/plugin-sdk/text-chunking"; import { hasMemorySessionTombstone } from "../memory-entry-origins.js"; import { withMemoryWorkspaceLock } from "../memory-workspace-lock.js"; import { readSessionResetRecallCutoffMetadata } from "../session-reset-recall-metadata.js"; import type { EmbeddingProvider } from "./embeddings.js"; import { createMemoryChunkWriter, type IndexedMemoryChunk } from "./manager-chunk-writer.js"; +import { readMemoryDatabaseRevision } from "./manager-db.js"; import { + clearMemoryEmbeddingCacheIdentities, collectMemoryCachedEmbeddings, + isValidMemoryEmbedding, loadMemoryEmbeddingCache, upsertMemoryEmbeddingCache, } from "./manager-embedding-cache.js"; @@ -103,6 +109,12 @@ type PreparedMemoryIndexEntry = { structuredInputBytes?: number; }; +type MemoryEmbeddingCacheCandidate = { + chunk: IndexedMemoryChunk; + entry: MemoryIndexEntry; + source: MemorySource; +}; + // Retry attempts are host control state. Provider-thrown values stay opaque so // they cannot override the counter or break accounting when they are immutable. type MemoryBatchRetryResult = @@ -330,16 +342,27 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { identities.at(0), "primary memory provider identity", ).providerKey; + const database = this.database; this.syncProviderGeneration = provider ? { kind: "semantic", - database: this.db, + database, + databaseRevision: readMemoryDatabaseRevision(database.db), + cacheWritesInvalidated: false, provider, ...(runtime ? { runtime } : {}), providerKey, identities, } - : { kind: "fts-only", database: this.db, provider: null, providerKey, identities }; + : { + kind: "fts-only", + database, + databaseRevision: readMemoryDatabaseRevision(database.db), + cacheWritesInvalidated: false, + provider: null, + providerKey, + identities, + }; this.syncProviderGenerationRelease = provider ? this.acquireProviderUse(provider) : null; this.syncProviderGenerationOwners = 1; } @@ -382,9 +405,10 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { } private async embedChunksInBatches( - chunks: IndexedMemoryChunk[], + candidates: MemoryEmbeddingCacheCandidate[], generation: MemorySemanticProviderGeneration, ): Promise { + const chunks = candidates.map((candidate) => candidate.chunk); if (chunks.length === 0) { return []; } @@ -394,24 +418,31 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { return embeddings; } - const missingChunks = missing.map((m) => m.chunk); - const batches = buildMemoryEmbeddingBatches(missingChunks, EMBEDDING_BATCH_MAX_TOKENS); + const missingCandidates = missing.map((item) => + expectDefined(candidates[item.index], "missing memory embedding candidate"), + ); + const batches = buildMemoryEmbeddingBatches( + missingCandidates.map((candidate) => candidate.chunk), + EMBEDDING_BATCH_MAX_TOKENS, + ); let cursor = 0; - for (const batch of batches) { - const inputs = buildTextEmbeddingInputs(batch); + for (const batchChunks of batches) { + const batchCandidates = missingCandidates.slice(cursor, cursor + batchChunks.length); + const inputs = buildTextEmbeddingInputs(batchChunks); const hasStructuredInputs = inputs.some((input) => hasNonTextEmbeddingParts(input)); const batchEmbeddings = await this.embedBatchWithRetry( - hasStructuredInputs ? inputs : batch.map((chunk) => chunk.text), + hasStructuredInputs ? inputs : batchChunks.map((chunk) => chunk.text), generation, + batchCandidates, ); - for (let i = 0; i < batch.length; i += 1) { + for (let i = 0; i < batchChunks.length; i += 1) { const item = missing[cursor + i]; const embedding = batchEmbeddings[i] ?? []; if (item) { embeddings[item.index] = embedding; } } - cursor += batch.length; + cursor += batchChunks.length; } return embeddings; } @@ -446,15 +477,16 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { } private async embedChunksWithBatch( - chunks: IndexedMemoryChunk[], + candidates: MemoryEmbeddingCacheCandidate[], source: string, generation: MemorySemanticProviderGeneration, debugContext: Record = {}, ): Promise { + const chunks = candidates.map((candidate) => candidate.chunk); const provider = generation.provider; const batchEmbed = generation.runtime?.batchEmbed; if (!batchEmbed) { - return this.embedChunksInBatches(chunks, generation); + return this.embedChunksInBatches(candidates, generation); } if (chunks.length === 0) { return []; @@ -464,7 +496,10 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { return embeddings; } - const missingChunks = missing.map((item) => item.chunk); + const missingCandidates = missing.map((item) => + expectDefined(candidates[item.index], "missing memory embedding candidate"), + ); + const missingChunks = missingCandidates.map((candidate) => candidate.chunk); const batchResult = await this.runBatchWithFallback({ provider: provider.id, run: async () => @@ -477,14 +512,18 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { timeoutMs: this.batch.timeoutMs, debug: this.buildBatchDebug(source, chunks, debugContext), }), - fallback: async () => await this.embedChunksInBatches(missingChunks, generation), + fallback: async () => await this.embedChunksInBatches(missingCandidates, generation), }); - if (!batchResult) { - return this.embedChunksInBatches(chunks, generation); + const batchEmbeddings = batchResult.value; + if (!batchEmbeddings) { + return this.embedChunksInBatches(candidates, generation); + } + if (batchResult.kind === "batch") { + await this.persistGeneratedEmbeddings(missingCandidates, batchEmbeddings, generation); } for (let index = 0; index < missing.length; index += 1) { const item = missing[index]; - const embedding = batchResult[index] ?? []; + const embedding = batchEmbeddings[index] ?? []; if (!item) { continue; } @@ -500,20 +539,27 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { embeddings: number[][]; missing: Array<{ index: number; chunk: IndexedMemoryChunk }>; } { - return collectMemoryCachedEmbeddings({ - chunks, - cached: loadMemoryEmbeddingCache({ - db: this.db, - enabled: this.cache.enabled, - providerIdentities: generation.identities, - hashes: chunks.map((chunk) => chunk.hash), - }), + const cached = loadMemoryEmbeddingCache({ + db: generation.database.db, + enabled: this.cache.enabled, + providerIdentities: generation.identities, + hashes: chunks.map((chunk) => chunk.hash), }); + // Cache hits and new batches must inhabit the same vector space during a sync. + for (const [hash, embedding] of cached) { + if (!isValidMemoryEmbedding(embedding, generation.embeddingDimensions)) { + cached.delete(hash); + } else { + generation.embeddingDimensions ??= embedding.length; + } + } + return collectMemoryCachedEmbeddings({ chunks, cached }); } protected async embedBatchWithRetry( inputs: Array, generation?: MemorySemanticProviderGeneration, + cacheCandidates?: MemoryEmbeddingCacheCandidate[], ): Promise { if (inputs.length === 0) { return []; @@ -524,13 +570,17 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { } const structured = inputs.some((input) => typeof input !== "string"); const label = structured ? "structured batch" : "batch"; + const requestItems = inputs.map((input, index) => ({ + input, + cacheCandidate: cacheCandidates?.[index], + })); try { return await this.withProviderUse( provider, async () => await runMemoryEmbeddingBatchRetryWithSplit({ profile: "index", - items: inputs, + items: requestItems, run: async (batchItems) => { const timeoutMs = this.resolveEmbeddingTimeout( "batch", @@ -546,7 +596,10 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { timeoutMs, message: `memory embeddings batch timed out after ${Math.round(timeoutMs / 1000)}s`, run: async (signal) => - await provider.embedBatch(batchItems, { signal, inputType: "document" }), + await provider.embedBatch( + batchItems.map((item) => item.input), + { signal, inputType: "document" }, + ), }); if (!structured) { log.debug("memory embeddings: batch completed", { @@ -556,6 +609,18 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { } return result; }, + onSuccess: async (batchItems, batchEmbeddings) => { + if (!generation) { + return; + } + const batchCandidates = batchItems.flatMap((item) => + item.cacheCandidate ? [item.cacheCandidate] : [], + ); + if (batchCandidates.length !== batchItems.length) { + return; + } + await this.persistGeneratedEmbeddings(batchCandidates, batchEmbeddings, generation); + }, isSplittable: isSplittableMemoryEmbeddingBatchError, waitForRetry: async (delayMs) => { await this.waitForEmbeddingRetry( @@ -586,6 +651,123 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { } } + private async persistGeneratedEmbeddings( + candidates: MemoryEmbeddingCacheCandidate[], + embeddings: number[][], + generation: MemorySemanticProviderGeneration, + ): Promise { + if ( + !this.cache.enabled || + candidates.length === 0 || + this.syncProviderGeneration !== generation || + generation.cacheWritesInvalidated || + generation.database.closed + ) { + return; + } + // Validate the whole provider response before retaining any vectors. Index + // insertion can fail later, but must never leave a reusable malformed batch. + const dimensions = generation.embeddingDimensions ?? embeddings[0]?.length; + if ( + embeddings.length !== candidates.length || + !embeddings.every((embedding) => isValidMemoryEmbedding(embedding, dimensions)) + ) { + if ( + generation.embeddingDimensions !== undefined && + embeddings.some( + (embedding) => + isValidMemoryEmbedding(embedding) && + embedding.length !== generation.embeddingDimensions, + ) + ) { + // Separate successful batches can disagree. Neither dimension is authoritative; + // discard this identity's ambiguous cache so retries can recover after restart. + await withMemoryWorkspaceLock(this.workspaceDir, async () => { + if ( + this.syncProviderGeneration !== generation || + generation.cacheWritesInvalidated || + generation.database.closed + ) { + return; + } + runSqliteImmediateTransactionSync(generation.database.db, () => { + if ( + readMemoryDatabaseRevision(generation.database.db) === generation.databaseRevision + ) { + clearMemoryEmbeddingCacheIdentities(generation.database.db, generation.identities); + } + generation.cacheWritesInvalidated = true; + }); + }); + } + throw new Error( + "memory embeddings: malformed vector response (count, dimensions, or coordinates)", + ); + } + generation.embeddingDimensions = dimensions; + await withMemoryWorkspaceLock(this.workspaceDir, async () => { + if ( + this.syncProviderGeneration !== generation || + generation.cacheWritesInvalidated || + generation.database.closed + ) { + return; + } + const entryValidity = new Map(); + const accepted: Array<{ hash: string; embedding: number[] }> = []; + for (const [index, candidate] of candidates.entries()) { + let valid = entryValidity.get(candidate.entry); + if (valid === undefined) { + if (candidate.source === "memory") { + const current = await buildFileEntry( + candidate.entry.absPath, + this.workspaceDir, + this.settings.multimodal, + ); + valid = current?.hash === candidate.entry.hash; + } else { + const sessionId = candidate.entry.sessionId; + valid = Boolean( + sessionId && + !hasMemorySessionTombstone(generation.database.db, this.agentId, sessionId), + ); + } + entryValidity.set(candidate.entry, valid); + } + if (valid) { + accepted.push({ + hash: candidate.chunk.hash, + embedding: embeddings[index] ?? [], + }); + } + } + if (accepted.length === 0) { + return; + } + runSqliteImmediateTransactionSync(generation.database.db, () => { + if ( + this.syncProviderGeneration !== generation || + generation.cacheWritesInvalidated || + generation.database.closed + ) { + return; + } + if (readMemoryDatabaseRevision(generation.database.db) !== generation.databaseRevision) { + generation.cacheWritesInvalidated = true; + return; + } + upsertMemoryEmbeddingCache({ + db: generation.database.db, + enabled: true, + provider: generation.provider, + providerKey: generation.providerKey, + entries: accepted, + maxEntries: this.cache.maxEntries, + }); + }); + }); + } + private async waitForEmbeddingRetry( delayMs: number, action: string, @@ -696,9 +878,9 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { provider: string; run: () => Promise; fallback: () => Promise; - }): Promise { + }): Promise<{ kind: "batch"; value: T } | { kind: "fallback"; value: number[][] }> { if (!this.batch.enabled) { - return await params.fallback(); + return { kind: "fallback", value: await params.fallback() }; } const result = await this.runBatchWithTimeoutRetry({ provider: params.provider, @@ -711,7 +893,7 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { } // An in-flight success clears failures without re-enabling disabled batching. this.batchFailure = { count: 0 }; - return result.value; + return { kind: "batch", value: result.value }; } const message = formatErrorMessage(result.error); @@ -726,7 +908,7 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { log.warn( `memory embeddings: ${params.provider} batch failed (${this.batchFailure.count}/${this.batchFailureLimit}); ${suffix}; falling back to non-batch embeddings: ${message}`, ); - return await params.fallback(); + return { kind: "fallback", value: await params.fallback() }; } protected getIndexConcurrency(): number { @@ -785,7 +967,7 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { // Embedding and vector setup may await while a purge completes. Read the // live owner, never the shadow index, immediately before publishing. if ( - hasMemorySessionTombstone(generation?.database ?? this.db, this.agentId, sessionId) + hasMemorySessionTombstone(generation?.database.db ?? this.db, this.agentId, sessionId) ) { this.markFailedFullReindexRetry({ memory: false, sessions: true }); throw new Error( @@ -823,17 +1005,6 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { .run(chunk.text, id, entry.path, source, model, chunk.startLine, chunk.endLine); } } - upsertMemoryEmbeddingCache({ - db: this.db, - enabled: this.cache.enabled, - provider: generation?.provider ?? null, - providerKey: generation?.providerKey ?? null, - entries: chunks.map((chunk, index) => ({ - hash: chunk.hash, - embedding: embeddings[index] ?? [], - })), - now, - }); this.upsertFileRecord(entry, source); if (needsVectorRebuild) { this.markVectorRebuildRequired(); @@ -844,6 +1015,9 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { if (!published) { return; } + if (generation && this.db === generation.database.db) { + generation.databaseRevision = readMemoryDatabaseRevision(generation.database.db); + } this.database.vectorDegradedWriteWarningShown = logMemoryVectorDegradedWrite({ vectorEnabled: this.vector.enabled, vectorReady, @@ -1068,11 +1242,17 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { return; } const current = prepared; - const chunks = current.flatMap((item) => item.chunks); + const candidates = current.flatMap((item) => + item.chunks.map((chunk) => ({ chunk, entry: item.entry, source: item.source })), + ); + const chunks = candidates.map((candidate) => candidate.chunk); const sourceCounts = countBatchSources(current); const source = formatBatchSourceLabel(sourceCounts); sourceWideBatchGroup += 1; - const chunkBatches = splitSourceWideEmbeddingChunks(chunks, SOURCE_WIDE_BATCH_MAX_REQUESTS); + const chunkBatches = splitSourceWideEmbeddingChunks( + candidates, + SOURCE_WIDE_BATCH_MAX_REQUESTS, + ); log.debug( `memory embeddings: source-wide batch submit group=${sourceWideBatchGroup} source=${source} files=${current.length} chunks=${chunks.length} requests=${chunkBatches.length} sources=${formatBatchSourceCounts( sourceCounts, @@ -1180,8 +1360,23 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps { let embeddings: number[][]; try { embeddings = this.batch.enabled - ? await this.embedChunksWithBatch(prepared.chunks, options.source, generation) - : await this.embedChunksInBatches(prepared.chunks, generation); + ? await this.embedChunksWithBatch( + prepared.chunks.map((chunk) => ({ + chunk, + entry: prepared.entry, + source: prepared.source, + })), + options.source, + generation, + ) + : await this.embedChunksInBatches( + prepared.chunks.map((chunk) => ({ + chunk, + entry: prepared.entry, + source: prepared.source, + })), + generation, + ); } catch (err) { const message = formatErrorMessage(err); if ( diff --git a/extensions/memory-core/src/memory/manager-embedding-policy.test.ts b/extensions/memory-core/src/memory/manager-embedding-policy.test.ts index 959108b6a85a..1e98ff12ffd7 100644 --- a/extensions/memory-core/src/memory/manager-embedding-policy.test.ts +++ b/extensions/memory-core/src/memory/manager-embedding-policy.test.ts @@ -464,6 +464,7 @@ describe("memory embedding policy", () => { }); it("splits OpenAI 431 oversized embedding batches without retrying the same request", async () => { + const completed: string[][] = []; const run = vi.fn(async (items: string[]) => { if (items.length > 1) { throw new Error( @@ -477,11 +478,15 @@ describe("memory embedding policy", () => { profile: "index", items: ["a", "b", "c", "d"], run, + onSuccess: (items) => { + completed.push(items); + }, isSplittable: isSplittableMemoryEmbeddingBatchError, waitForRetry: async () => {}, }); expect(result).toEqual([[97], [98], [99], [100]]); + expect(completed).toEqual([["a"], ["b"], ["c"], ["d"]]); expect(run.mock.calls.map(([items]) => items.length)).toEqual([4, 2, 1, 1, 2, 1, 1]); expect(isSplittableMemoryEmbeddingBatchError("431 request_headers_too_large")).toBe(true); expect(isSplittableMemoryEmbeddingBatchError("embedding validation failed at item 4312")).toBe( diff --git a/extensions/memory-core/src/memory/manager-embedding-policy.ts b/extensions/memory-core/src/memory/manager-embedding-policy.ts index 538561b0874e..7d521c52c0ea 100644 --- a/extensions/memory-core/src/memory/manager-embedding-policy.ts +++ b/extensions/memory-core/src/memory/manager-embedding-policy.ts @@ -176,12 +176,14 @@ export async function runMemoryEmbeddingBatchRetryWithSplit(par profile: MemoryEmbeddingRetryProfileName; items: TInput[]; run: (items: TInput[]) => Promise; + onSuccess?: (items: TInput[], outputs: TOutput[]) => void | Promise; isSplittable: (message: string) => boolean; waitForRetry: (delayMs: number) => Promise; onSplit?: (info: { itemCount: number; splitAt: number; message: string }) => void; }): Promise { + let outputs: TOutput[]; try { - return await runMemoryEmbeddingRetryLoop({ + outputs = await runMemoryEmbeddingRetryLoop({ profile: params.profile, run: async () => await params.run(params.items), waitForRetry: params.waitForRetry, @@ -204,6 +206,8 @@ export async function runMemoryEmbeddingBatchRetryWithSplit(par }); return [...left, ...right]; } + await params.onSuccess?.(params.items, outputs); + return outputs; } export function buildTextEmbeddingInputs(chunks: MemoryEmbeddingChunk[]): EmbeddingInput[] { diff --git a/extensions/memory-core/src/memory/manager-memory-source-race.test.ts b/extensions/memory-core/src/memory/manager-memory-source-race.test.ts index 813ce46af534..1ae03039b714 100644 --- a/extensions/memory-core/src/memory/manager-memory-source-race.test.ts +++ b/extensions/memory-core/src/memory/manager-memory-source-race.test.ts @@ -1,6 +1,7 @@ import fs from "node:fs/promises"; import path from "node:path"; import type { DatabaseSync } from "node:sqlite"; +import { hashText } from "openclaw/plugin-sdk/memory-core-host-engine-storage"; import { describe, expect, it } from "vitest"; import { createManagerIndexFixture } from "./manager-index.test-support.js"; @@ -29,18 +30,20 @@ describe("memory source changes during indexing", () => { async ({ provider, force, mutation, maintenance }) => { const changingFile = path.join(fixture.paths.memory, "changing.md"); const siblingFile = path.join(fixture.paths.memory, "sibling.md"); + const obsoleteContent = "Obsolete alpha source awaiting embeddings."; + const latestContent = "Latest alpha source after the concurrent edit."; await fs.writeFile(changingFile, "Original alpha source."); await fs.writeFile(siblingFile, "Original beta sibling."); const cfg = fixture.createConfig({ provider, batchEnabled: true, + cacheEnabled: true, vectorEnabled: false, sources: ["memory"], }); - cfg.memory = { ...cfg.memory, search: { ...cfg.memory?.search, cache: { enabled: false } } }; const manager = await fixture.getFreshManager(cfg, "cli"); await manager.sync({ reason: "baseline", force: true }); - await fs.writeFile(changingFile, "Obsolete alpha source awaiting embeddings."); + await fs.writeFile(changingFile, obsoleteContent); await fs.writeFile(siblingFile, "Updated beta sibling survives the concurrent edit."); Reflect.set(manager, "dirty", true); let releaseEmbedding = () => {}; @@ -77,7 +80,7 @@ describe("memory source changes during indexing", () => { if (mutation === "delete") { await fs.unlink(changingFile); } else { - await fs.writeFile(changingFile, "Latest alpha source after the concurrent edit."); + await fs.writeFile(changingFile, latestContent); } releaseEmbedding(); await expect(activeSync).resolves.toBeUndefined(); @@ -90,6 +93,11 @@ describe("memory source changes during indexing", () => { .join("\n"); expect(indexedText()).toContain("Updated beta sibling"); expect(indexedText()).not.toContain("Obsolete alpha"); + expect( + db + .prepare("SELECT hash FROM memory_embedding_cache WHERE hash = ?") + .get(hashText(obsoleteContent)), + ).toBeUndefined(); expect(manager.status().dirty).toBe(true); expect(Reflect.get(manager, "memoryFullRetryDirty")).toBe(false); @@ -106,6 +114,11 @@ describe("memory source changes during indexing", () => { ).toBeUndefined(); } else { expect(indexedText()).toContain("Latest alpha source"); + expect( + db + .prepare("SELECT hash FROM memory_embedding_cache WHERE hash = ?") + .get(hashText(latestContent)), + ).toEqual({ hash: hashText(latestContent) }); } expect(fixture.provider.providerRuntimeBatchCalls.flat().join("\n")).not.toContain( "beta sibling", diff --git a/extensions/memory-core/src/memory/manager-sync-base.ts b/extensions/memory-core/src/memory/manager-sync-base.ts index 127712497d5c..03704050cbb4 100644 --- a/extensions/memory-core/src/memory/manager-sync-base.ts +++ b/extensions/memory-core/src/memory/manager-sync-base.ts @@ -1,5 +1,4 @@ // Memory Core plugin module owns shared manager synchronization state. -import type { DatabaseSync } from "node:sqlite"; import type { FSWatcher } from "chokidar"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import { @@ -12,7 +11,6 @@ import { import { ensureMemoryIndexSchema, loadSqliteVecExtension, - MEMORY_EMBEDDING_CACHE_TABLE, MEMORY_INDEX_VECTOR_TABLE, type MemorySessionSyncTarget, type MemoryEntryProvenance, @@ -20,7 +18,6 @@ import { type MemorySyncParams, type MemorySyncProgressUpdate, } from "openclaw/plugin-sdk/memory-core-host-engine-storage"; -import { runSqliteImmediateTransactionSync } from "openclaw/plugin-sdk/sqlite-runtime"; import type { MemoryCoreAcquireLocalService } from "./embedding-local-service.js"; import { resolveEmbeddingProviderIndexIdentity, @@ -29,10 +26,6 @@ import { type EmbeddingProviderRuntime, } from "./embeddings.js"; import { MemoryManagerDatabaseContext } from "./manager-database-context.js"; -import { - prepareMemoryEmbeddingCacheUpsert, - type MemoryEmbeddingCacheRow, -} from "./manager-embedding-cache.js"; import { resolveMemoryPrimaryProviderRequest, type MemoryProviderLifecycleState, @@ -100,10 +93,6 @@ export const MEMORY_INDEX_META_KEY = "memory_index_meta_v1"; const META_KEY = MEMORY_INDEX_META_KEY; const VECTOR_TABLE = MEMORY_INDEX_VECTOR_TABLE; const LEGACY_VECTOR_TABLE = "chunks_vec"; -const EMBEDDING_CACHE_TABLE = MEMORY_EMBEDDING_CACHE_TABLE; -// Production embeddings measured ~28 KB/row; 1,000-row synchronous commits -// blocked the event loop for seconds. Keep each commit small between yields. -const EMBEDDING_CACHE_SEED_BATCH_SIZE = 100; const VECTOR_LOAD_TIMEOUT_MS = 30_000; const log = createSubsystemLogger("memory"); @@ -628,46 +617,6 @@ export abstract class MemoryManagerSyncBase extends MemoryManagerDatabaseContext return buildMemorySourceFilter(alias, sources); } - protected async seedEmbeddingCache(sourceDb: DatabaseSync): Promise { - if (!this.cache.enabled) { - return; - } - type CacheRow = MemoryEmbeddingCacheRow & { rowid: number }; - const selectBatch = sourceDb.prepare( - `SELECT rowid, provider, model, provider_key, hash, embedding, dims, updated_at - FROM ${EMBEDDING_CACHE_TABLE} - WHERE rowid > ? - ORDER BY rowid - LIMIT ?`, - ); - const upsert = prepareMemoryEmbeddingCacheUpsert(this.db); - let lastRowid = 0; - while (true) { - // Materialize each source page so neither a read cursor nor a write - // transaction remains open when control returns to the event loop. - const batch = selectBatch.all(lastRowid, EMBEDDING_CACHE_SEED_BATCH_SIZE) as CacheRow[]; - if (batch.length === 0) { - return; - } - runSqliteImmediateTransactionSync( - this.db, - () => { - for (const row of batch) { - upsert(row); - } - }, - { operationLabel: "memory.embedding-cache.seed" }, - ); - lastRowid = batch[batch.length - 1]?.rowid ?? lastRowid; - if (batch.length < EMBEDDING_CACHE_SEED_BATCH_SIZE) { - return; - } - await new Promise((resolve) => { - setImmediate(resolve); - }); - } - } - protected ensureSchema() { const result = ensureMemoryIndexSchema({ db: this.db, diff --git a/extensions/memory-core/src/memory/manager-sync-ops.ts b/extensions/memory-core/src/memory/manager-sync-ops.ts index 21aa6d74fa1c..9b774001ae39 100644 --- a/extensions/memory-core/src/memory/manager-sync-ops.ts +++ b/extensions/memory-core/src/memory/manager-sync-ops.ts @@ -1,6 +1,5 @@ // Memory Core plugin module coordinates synchronization and shadow reindexing. import { randomUUID } from "node:crypto"; -import type { DatabaseSync } from "node:sqlite"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import { createSubsystemLogger, @@ -55,7 +54,9 @@ import { markMemoryVectorIndexClean } from "./manager-vector-rebuild-state.js"; export type { MemoryIndexWorkItem } from "./manager-sync-base.js"; type MemorySyncProviderGenerationBase = { - database: DatabaseSync; + database: MemoryIndexDatabase; + databaseRevision: number; + cacheWritesInvalidated: boolean; providerKey: string; identities: MemoryIndexProviderIdentity[]; }; @@ -64,6 +65,7 @@ export type MemorySyncProviderGeneration = | (MemorySyncProviderGenerationBase & { kind: "fts-only"; provider: null }) | (MemorySyncProviderGenerationBase & { kind: "semantic"; + embeddingDimensions?: number; provider: EmbeddingProvider; runtime?: EmbeddingProviderRuntime; }); @@ -536,7 +538,6 @@ export abstract class MemoryManagerSyncOps extends MemoryManagerSourceSyncOps { const rebuilt = await this.withReindexDatabase(shadow, async () => { try { this.ensureSchema(); - await this.seedEmbeddingCache(originalDb); const shouldSyncMemory = shouldRetryMemoryOnFailure; const shouldSyncSessions = shouldRetrySessionsOnFailure; @@ -599,9 +600,6 @@ export abstract class MemoryManagerSyncOps extends MemoryManagerSourceSyncOps { } this.writeMeta(nextMeta); - // Bound the cache before copying it into the shared agent database; - // deleting overflow afterward does not undo primary-file growth. - await this.pruneEmbeddingCacheIfNeeded(); return { nextMeta, vectorIndexComplete, @@ -636,6 +634,9 @@ export abstract class MemoryManagerSyncOps extends MemoryManagerSourceSyncOps { this.fts.available = shadow.fts.available; this.fts.loadError = shadow.fts.loadError; this.vector.dims = rebuilt.nextMeta.vectorDims; + // Cache-only rebuilds bypass insertion-time eviction; prune the canonical + // cache only after successful publication so failed rebuilds retain their work. + await this.pruneEmbeddingCacheIfNeeded(); } catch (err) { this.restoreReindexRetryState(originalRetryState); this.markFailedFullReindexRetry({ diff --git a/extensions/memory-core/src/memory/manager-sync-yield.test.ts b/extensions/memory-core/src/memory/manager-sync-yield.test.ts index c6aa550eb381..0b45e24d7456 100644 --- a/extensions/memory-core/src/memory/manager-sync-yield.test.ts +++ b/extensions/memory-core/src/memory/manager-sync-yield.test.ts @@ -190,18 +190,6 @@ class SessionSyncYieldHarness extends MemoryManagerSyncOps { } } -class EmbeddingCacheSeedHarness extends SessionSyncYieldHarness { - protected override readonly cache = { enabled: true }; - - constructor(db: DatabaseSync) { - super(db, () => {}); - } - - async seedCache(sourceDb: DatabaseSync): Promise { - await this.seedEmbeddingCache(sourceDb); - } -} - describe("session sync responsiveness", () => { beforeEach(() => { setSyncYieldStateDir(); @@ -253,81 +241,3 @@ describe("session sync responsiveness", () => { } }); }); - -describe("embedding cache seed responsiveness", () => { - function countCacheRows(db: DatabaseSync): number { - const row = db.prepare("SELECT count(*) AS count FROM memory_embedding_cache").get() as { - count: number; - }; - return row.count; - } - - it("commits each materialized page before yielding", async () => { - const sourceDb = createDb(); - const targetDb = createDb(); - const { StatementSync } = requireNodeSqlite(); - const prepare = vi.spyOn(targetDb, "prepare"); - const columns = vi.spyOn(StatementSync.prototype, "columns"); - try { - const insert = sourceDb.prepare( - `INSERT INTO memory_embedding_cache - (provider, model, provider_key, hash, embedding, dims, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?)`, - ); - const rawLargeEmbedding = ` ${JSON.stringify(Array.from({ length: 4096 }, () => 0.1234567890123456))}\n`; - sourceDb.exec("BEGIN"); - for (let index = 0; index < 101; index += 1) { - insert.run( - "test", - "model", - "key", - `hash-${index}`, - index === 0 ? " malformed JSON \n" : index === 1 ? rawLargeEmbedding : "[ 0.5 ]", - index === 0 ? null : index === 1 ? 4096 : 1, - index - 1, - ); - } - sourceDb.exec("COMMIT"); - - let duringYield: { - sourceInTransaction: boolean; - targetInTransaction: boolean; - rows: number; - } | null = null; - const observedYield = new Promise((resolve, reject) => { - setImmediate(() => { - try { - duringYield = { - sourceInTransaction: sourceDb.isTransaction, - targetInTransaction: targetDb.isTransaction, - rows: countCacheRows(targetDb), - }; - resolve(); - } catch (error) { - reject(error instanceof Error ? error : new Error(String(error))); - } - }); - }); - - await new EmbeddingCacheSeedHarness(targetDb).seedCache(sourceDb); - await observedYield; - - expect(duringYield).toEqual({ - sourceInTransaction: false, - targetInTransaction: false, - rows: 100, - }); - expect(countCacheRows(targetDb)).toBe(101); - expect(prepare.mock.calls.filter(([sql]) => /^insert/i.test(sql))).toHaveLength(1); - expect(columns).not.toHaveBeenCalled(); - const readCache = (db: DatabaseSync) => - db.prepare("SELECT * FROM memory_embedding_cache ORDER BY hash").all(); - expect(readCache(targetDb)).toEqual(readCache(sourceDb)); - } finally { - prepare.mockRestore(); - columns.mockRestore(); - sourceDb.close(); - targetDb.close(); - } - }); -}); diff --git a/extensions/memory-core/src/memory/manager.reindex-recovery.test.ts b/extensions/memory-core/src/memory/manager.reindex-recovery.test.ts index 44e9ac35da0c..bceb5fa7ae2a 100644 --- a/extensions/memory-core/src/memory/manager.reindex-recovery.test.ts +++ b/extensions/memory-core/src/memory/manager.reindex-recovery.test.ts @@ -41,8 +41,12 @@ describe("memory manager reindex recovery", () => { let workspaceDir = ""; let memoryDir = ""; let manager: MemoryIndexManager | null = null; + let embeddingCalls: unknown[][] = []; + let batchEmbeddingCalls: string[][] = []; beforeEach(async () => { + embeddingCalls = []; + batchEmbeddingCalls = []; // Register the fixture at the same boundary used by config and provider creation. registerEmbeddingProvider({ id: "openai", @@ -53,10 +57,36 @@ describe("memory manager reindex recovery", () => { model: "mock-embed", maxInputTokens: 8192, embed: async () => [0, 1, 0], - embedBatch: async (inputs) => inputs.map(() => [0, 1, 0]), + embedBatch: async (inputs) => { + embeddingCalls.push(inputs); + return inputs.map(() => [0, 1, 0]); + }, }, }), }); + for (const id of ["batch-test", "batch-wide-test"] as const) { + registerEmbeddingProvider({ + id, + transport: "remote", + create: async () => ({ + provider: { + id, + model: "mock-embed", + maxInputTokens: 8192, + embed: async () => [0, 1, 0], + embedBatch: async (inputs) => inputs.map(() => [0, 1, 0]), + }, + runtime: { + id, + ...(id === "batch-wide-test" ? { sourceWideBatchEmbed: true } : {}), + batchEmbed: async (batch: { chunks: Array<{ text: string }> }) => { + batchEmbeddingCalls.push(batch.chunks.map((chunk) => chunk.text)); + return batch.chunks.map(() => [0, 1, 0]); + }, + }, + }), + }); + } fixtureRoot = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-mem-reindex-recovery-")); workspaceDir = path.join(fixtureRoot, "workspace"); memoryDir = path.join(workspaceDir, "memory"); @@ -84,6 +114,7 @@ describe("memory manager reindex recovery", () => { provider?: string; sources?: Array<"memory" | "sessions">; cacheEnabled?: boolean; + batchEnabled?: boolean; }): OpenClawConfig { return isolateMemoryManagerTestConfig({ memory: { @@ -91,6 +122,7 @@ describe("memory manager reindex recovery", () => { provider: params.provider ?? "openai", model: "mock-embed", store: { vector: {} }, + remote: params.batchEnabled ? { batch: { enabled: true } } : undefined, cache: { enabled: params.cacheEnabled ?? false }, sources: params.sources, rememberAcrossConversations: params.sources?.includes("sessions") ?? false, @@ -205,6 +237,194 @@ describe("memory manager reindex recovery", () => { ).toEqual(publishedRows); }); + it.each([ + { name: "inconsistent dimensions", invalid: [0, 1] }, + { name: "nonfinite coordinates", invalid: [0, Number.NaN, 0] }, + ])("does not retain $name after rejected provider output", async ({ invalid }) => { + const memoryManager = await openManager(createCfg({ sources: ["memory"], cacheEnabled: true })); + await memoryManager.sync({ reason: "cli", force: true }); + await fs.writeFile( + path.join(memoryDir, "alpha.md"), + Array.from( + { length: 80 }, + (_, index) => `Fact ${index}: keep independent reusable memory content.`, + ).join("\n"), + ); + // SAFETY: the fixture owns this manager and its registered embedding provider. + const harness = memoryManager as unknown as ReindexHarness; + if (!harness.provider) { + throw new Error("fixture provider missing"); + } + const embed = vi + .spyOn(harness.provider, "embedBatch") + .mockImplementationOnce(async (inputs) => { + expect(inputs.length).toBeGreaterThan(1); + return inputs.map((_, index) => (index === 0 ? [0, 1, 0] : invalid)); + }); + await expect(memoryManager.sync({ reason: "cli", force: true })).rejects.toThrow(); + expect(harness.db.prepare("SELECT hash FROM memory_embedding_cache").all()).toEqual([]); + await memoryManager.sync({ reason: "cli", force: true }); + expect(embed).toHaveBeenCalledTimes(2); + expect( + harness.db.prepare("SELECT hash FROM memory_embedding_cache").all().length, + ).toBeGreaterThan(0); + }); + + it("recovers when separate provider batches disagree on dimensions", async () => { + const cfg = createCfg({ sources: ["memory"], cacheEnabled: true }); + const memoryManager = await openManager(cfg); + await memoryManager.sync({ reason: "cli", force: true }); + const harness = memoryManager as unknown as ReindexHarness; + if (!harness.provider) { + throw new Error("fixture provider missing"); + } + await fs.writeFile( + path.join(memoryDir, "large.md"), + `${"first ".repeat(3500)}\n${"second ".repeat(3500)}`, + ); + harness.db + .prepare(`INSERT INTO memory_embedding_cache + (provider, model, provider_key, hash, embedding, dims, updated_at) + VALUES ('unrelated', 'unrelated', 'unrelated', 'keep', '[1,0]', 2, 1)`) + .run(); + const embed = vi + .spyOn(harness.provider, "embedBatch") + .mockImplementationOnce(async (inputs) => inputs.map(() => [0, 1])); + await expect(memoryManager.sync({ reason: "cli", force: true })).rejects.toThrow( + "malformed vector response", + ); + expect(embed.mock.calls.length).toBeGreaterThan(1); + expect(harness.db.prepare("SELECT provider FROM memory_embedding_cache").all()).toEqual([ + { provider: "unrelated" }, + ]); + await memoryManager.close(); + manager = null; + const reopened = await openManager(cfg); + await expect(reopened.sync({ reason: "cli", force: true })).resolves.toBeUndefined(); + expect( + (reopened as unknown as ReindexHarness).db + .prepare("SELECT DISTINCT dims FROM memory_embedding_cache WHERE provider = 'openai'") + .all(), + ).toEqual([{ dims: 3 }]); + }); + + it("retains completed embeddings across a failed rebuild and manager restart", async () => { + const cfg = createCfg({ sources: ["memory"], cacheEnabled: true }); + const memoryPath = path.join(memoryDir, "alpha.md"); + await fs.writeFile(memoryPath, "published alpha"); + const memoryManager = await openManager(cfg); + await memoryManager.sync({ reason: "cli", force: true }); + const harness = memoryManager as unknown as ReindexHarness; + const published = harness.db.prepare("SELECT text FROM memory_index_chunks").all(); + await fs.writeFile(memoryPath, "replacement beta"); + const metadata = vi.spyOn(harness, "writeMeta").mockImplementationOnce(() => { + throw new Error("late shadow failure"); + }); + + await expect(memoryManager.sync({ reason: "cli", force: true })).rejects.toThrow( + "late shadow failure", + ); + expect(harness.db.prepare("SELECT text FROM memory_index_chunks").all()).toEqual(published); + metadata.mockRestore(); + const paidInputs = embeddingCalls.flat(); + expect(paidInputs).toContain("replacement beta"); + await memoryManager.close(); + manager = null; + const reopened = await openManager(cfg); + embeddingCalls = []; + + await reopened.sync({ reason: "cli", force: true }); + + expect(embeddingCalls.flat()).toEqual([]); + expect( + (reopened as unknown as ReindexHarness).db + .prepare("SELECT text FROM memory_index_chunks") + .all(), + ).toEqual([{ text: "replacement beta" }]); + }); + + it("retains successful batches when a later batch in the same file fails", async () => { + const cfg = createCfg({ sources: ["memory"], cacheEnabled: true }); + await fs.writeFile(path.join(memoryDir, "alpha.md"), "published alpha"); + const memoryManager = await openManager(cfg); + await memoryManager.sync({ reason: "cli", force: true }); + const harness = memoryManager as unknown as ReindexHarness; + const provider = harness.provider; + if (!provider) { + throw new Error("expected the test embedding provider"); + } + await fs.writeFile( + path.join(memoryDir, "large.md"), + `${"first ".repeat(3500)}\n${"second ".repeat(3500)}`, + ); + let completed: unknown[] = []; + const requests = vi.spyOn(provider, "embedBatch").mockImplementation(async (inputs) => { + if (completed.length > 0) { + throw new Error("permanent embedding failure"); + } + completed = inputs; + return inputs.map(() => [0, 1, 0]); + }); + + await expect(memoryManager.sync({ reason: "cli", force: true })).rejects.toThrow( + "permanent embedding failure", + ); + expect(completed.length).toBeGreaterThan(0); + expect(harness.db.prepare("SELECT text FROM memory_index_chunks").all()).toEqual([ + { text: "published alpha" }, + ]); + requests.mockRestore(); + embeddingCalls = []; + + await memoryManager.sync({ reason: "cli", force: true }); + + expect(embeddingCalls.flat().length).toBeGreaterThan(0); + for (const input of completed) { + expect(embeddingCalls.flat()).not.toContain(input); + } + }); + + it.each(["batch-test", "batch-wide-test"] as const)( + "retains completed %s runtime batches after a late rebuild failure", + async (provider) => { + const cfg = createCfg({ + provider, + sources: ["memory"], + cacheEnabled: true, + batchEnabled: true, + }); + const alphaPath = path.join(memoryDir, "alpha.md"); + const betaPath = path.join(memoryDir, "beta.md"); + await fs.writeFile(alphaPath, "published alpha"); + await fs.writeFile(betaPath, "published beta"); + const memoryManager = await openManager(cfg); + await memoryManager.sync({ reason: "cli", force: true }); + const harness = memoryManager as unknown as ReindexHarness; + await fs.writeFile(alphaPath, "replacement alpha"); + await fs.writeFile(betaPath, "replacement beta"); + batchEmbeddingCalls = []; + const metadata = vi.spyOn(harness, "writeMeta").mockImplementationOnce(() => { + throw new Error("late runtime batch failure"); + }); + + await expect(memoryManager.sync({ reason: "cli", force: true })).rejects.toThrow( + "late runtime batch failure", + ); + expect(batchEmbeddingCalls.flat()).toEqual( + expect.arrayContaining(["replacement alpha", "replacement beta"]), + ); + metadata.mockRestore(); + await memoryManager.close(); + manager = null; + const reopened = await openManager(cfg); + batchEmbeddingCalls = []; + + await reopened.sync({ reason: "cli", force: true }); + + expect(batchEmbeddingCalls).toEqual([]); + }, + ); + it("bounds the shadow cache before any entries reach the primary", async () => { const memoryManager = await openManager(createCfg({ sources: ["memory"], cacheEnabled: true })); const harness = memoryManager as unknown as ReindexHarness; @@ -230,6 +450,21 @@ describe("memory manager reindex recovery", () => { }); }); + it("bounds the canonical cache after a successful all-cache-hit rebuild", async () => { + const { memoryManager, harness, newest } = await createOversizedPublishedCache(); + embeddingCalls = []; + + await memoryManager.sync({ reason: "cli", force: true }); + + expect(embeddingCalls).toEqual([]); + expect(harness.db.prepare("SELECT * FROM memory_embedding_cache ORDER BY hash").all()).toEqual( + newest, + ); + expect(harness.db.prepare("SELECT text FROM memory_index_chunks").all()).toEqual([ + { text: "published alpha" }, + ]); + }); + it("leaves even an oversized published cache untouched when a full rebuild fails", async () => { const { memoryManager, harness, before } = await createOversizedPublishedCache(); harness.writeMeta = () => { diff --git a/src/plugins/openai-compatible-embedding-provider.http-error.test.ts b/src/plugins/openai-compatible-embedding-provider.http-error.test.ts new file mode 100644 index 000000000000..6afa27330ca7 --- /dev/null +++ b/src/plugins/openai-compatible-embedding-provider.http-error.test.ts @@ -0,0 +1,71 @@ +import { once } from "node:events"; +import { createServer } from "node:http"; +import type { AddressInfo } from "node:net"; +import { afterEach, describe, expect, it } from "vitest"; +import { ProviderHttpError } from "../agents/provider-http-errors.js"; +import { openAICompatibleEmbeddingProviderAdapter } from "./openai-compatible-embedding-provider.js"; + +const servers = new Set>(); + +afterEach(async () => { + await Promise.all( + Array.from(servers, async (server) => { + server.closeAllConnections(); + server.close(); + await once(server, "close"); + servers.delete(server); + }), + ); +}); + +describe("OpenAI-compatible embedding HTTP errors", () => { + it("preserves retry metadata while redacting reflected request credentials", async () => { + const token = "secret-reflected-token"; + const server = createServer((request, response) => { + request.resume(); + request.once("end", () => { + expect(request.headers.authorization).toBe(`Bearer ${token}`); + response.writeHead(429, { + "content-type": "application/json", + "retry-after": "4", + }); + response.end( + JSON.stringify({ + error: { + message: `Quota exhausted for ${token}`, + type: "rate_limit_error", + code: "rate_limit_exceeded", + }, + }), + ); + }); + }); + servers.add(server); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const address = server.address() as AddressInfo; + + const result = await openAICompatibleEmbeddingProviderAdapter.create({ + config: {}, + provider: "openai-compatible", + model: "text-embedding-bge-m3", + remote: { baseUrl: `http://127.0.0.1:${address.port}/v1`, apiKey: token }, + }); + if (!result.provider) { + throw new Error("expected OpenAI-compatible embedding provider"); + } + + const error = await result.provider.embed("hello").catch((cause: unknown) => cause); + + expect(error).toBeInstanceOf(ProviderHttpError); + expect(error).toMatchObject({ + status: 429, + code: "rate_limit_exceeded", + errorType: "rate_limit_error", + retryAfterMs: 4_000, + }); + expect((error as Error).message).toContain("openai-compatible embeddings failed: HTTP 429"); + expect((error as Error).message).not.toContain(token); + expect((error as ProviderHttpError).errorBody).not.toContain(token); + }); +}); diff --git a/src/plugins/openai-compatible-embedding-provider.ts b/src/plugins/openai-compatible-embedding-provider.ts index 643ab2d2ac47..da60df3be323 100644 --- a/src/plugins/openai-compatible-embedding-provider.ts +++ b/src/plugins/openai-compatible-embedding-provider.ts @@ -8,11 +8,15 @@ import { MEMORY_SEARCH_DEADLINE_CONTROL, type MemorySearchDeadlineControl, } from "../../packages/memory-host-sdk/src/host/search-deadline-control.js"; -import { readProviderJsonArrayFieldResponse } from "../agents/provider-http-errors.js"; +import { + createProviderHttpError, + readProviderJsonArrayFieldResponse, +} from "../agents/provider-http-errors.js"; import type { AcquireConfiguredProviderLocalService, ConfiguredProviderLocalServiceTarget, } from "../agents/provider-local-service-target.js"; +import { redactProviderResponseErrorText } from "../agents/provider-request-header-redaction.js"; import type { ModelProviderLocalServiceConfig } from "../config/types.models.js"; import { normalizeResolvedSecretInputString } from "../config/types.secrets.js"; import { readResponseTextPrefix } from "../infra/http-body.js"; @@ -305,11 +309,27 @@ async function readEmbeddingErrorBodySnippet(response: Response): Promise { +async function createEmbeddingHttpError( + response: Response, + requestHeaders: HeadersInit, +): Promise { const snippet = await readEmbeddingErrorBodySnippet(response); - return new Error( - `openai-compatible embeddings failed: HTTP ${response.status}${snippet ? `: ${snippet}` : ""}`, + const error = await createProviderHttpError( + new Response(snippet, { + status: response.status, + statusText: response.statusText, + headers: response.headers, + }), + "openai-compatible embeddings failed", + { requestHeaders }, ); + const safeSnippet = snippet + ? redactProviderResponseErrorText(snippet, requestHeaders, { + sourceTruncated: snippet.endsWith(EMBEDDING_ERROR_TRUNCATED_SUFFIX), + }) + : undefined; + error.message = `openai-compatible embeddings failed: HTTP ${response.status}${safeSnippet ? `: ${safeSnippet}` : ""}`; + return error; } async function postEmbeddingRequest(params: { @@ -355,7 +375,7 @@ async function postEmbeddingRequest(params: { auditContext: "embedding-provider:openai-compatible", onResponse: async (response) => { if (!response.ok) { - throw await createEmbeddingHttpError(response); + throw await createEmbeddingHttpError(response, client.headers); } return readEmbeddingVectors( await readProviderJsonArrayFieldResponse(