mirror of
https://github.com/openclaw/openclaw.git
synced 2026-10-03 09:39:25 +00:00
fix(memory): reuse completed embeddings after failed rebuilds (#144404)
This commit is contained in:
Josh Lehman
GitHub
parent
782a994014
commit
cd9be2bbe7
20 changed files with 854 additions and 257 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<ForgetDatabase>(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) {
|
||||
|
|
|
|||
|
|
@ -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 = ?")
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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<number> };
|
||||
};
|
||||
|
||||
/** 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<EmbeddingCacheDatabase>(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<MemoryEmbeddingCacheRow>((parameter) =>
|
||||
getNodeSqliteKysely<EmbeddingCacheDatabase>(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<string>();
|
||||
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<EmbeddingCacheDatabase>(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<T extends Pick<MemoryChunk, "hash">>(params: {
|
||||
chunks: T[];
|
||||
cached: Map<string, number[]>;
|
||||
|
|
|
|||
|
|
@ -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<T> =
|
||||
|
|
@ -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<number[][]> {
|
||||
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<string, unknown> = {},
|
||||
): Promise<number[][]> {
|
||||
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<string | EmbeddingInput>,
|
||||
generation?: MemorySemanticProviderGeneration,
|
||||
cacheCandidates?: MemoryEmbeddingCacheCandidate[],
|
||||
): Promise<number[][]> {
|
||||
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<void> {
|
||||
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<MemoryIndexEntry, boolean>();
|
||||
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<T>;
|
||||
fallback: () => Promise<number[][]>;
|
||||
}): Promise<T | number[][]> {
|
||||
}): 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 (
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -176,12 +176,14 @@ export async function runMemoryEmbeddingBatchRetryWithSplit<TInput, TOutput>(par
|
|||
profile: MemoryEmbeddingRetryProfileName;
|
||||
items: TInput[];
|
||||
run: (items: TInput[]) => Promise<TOutput[]>;
|
||||
onSuccess?: (items: TInput[], outputs: TOutput[]) => void | Promise<void>;
|
||||
isSplittable: (message: string) => boolean;
|
||||
waitForRetry: (delayMs: number) => Promise<void>;
|
||||
onSplit?: (info: { itemCount: number; splitAt: number; message: string }) => void;
|
||||
}): Promise<TOutput[]> {
|
||||
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<TInput, TOutput>(par
|
|||
});
|
||||
return [...left, ...right];
|
||||
}
|
||||
await params.onSuccess?.(params.items, outputs);
|
||||
return outputs;
|
||||
}
|
||||
|
||||
export function buildTextEmbeddingInputs(chunks: MemoryEmbeddingChunk[]): EmbeddingInput[] {
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
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<void>((resolve) => {
|
||||
setImmediate(resolve);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
protected ensureSchema() {
|
||||
const result = ensureMemoryIndexSchema({
|
||||
db: this.db,
|
||||
|
|
|
|||
|
|
@ -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({
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
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<void>((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();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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 = () => {
|
||||
|
|
|
|||
|
|
@ -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<ReturnType<typeof createServer>>();
|
||||
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
|
@ -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<string
|
|||
return truncated ? `${text}${EMBEDDING_ERROR_TRUNCATED_SUFFIX}` : text;
|
||||
}
|
||||
|
||||
async function createEmbeddingHttpError(response: Response): Promise<Error> {
|
||||
async function createEmbeddingHttpError(
|
||||
response: Response,
|
||||
requestHeaders: HeadersInit,
|
||||
): Promise<Error> {
|
||||
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(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue