diff --git a/docs/memory-v2/README.md b/docs/memory-v2/README.md index 019ddf9..f79e802 100644 --- a/docs/memory-v2/README.md +++ b/docs/memory-v2/README.md @@ -394,6 +394,14 @@ Optional table name for pgvector retrieval. Defaults to `memory_embeddings`. Optional PostgreSQL pool size for Memory V2 pgvector retrieval. Defaults to `5`. +`MEMORY_PGVECTOR_SYNC_USER_LIMIT` + +Maximum number of recent active memories synchronously upserted for the current +user after a successful `write(...)`, `compact(...)`, or candidate promotion. +Defaults to `50`. When a session ID is available the sync is additionally scoped +to that session. This immediate, idempotent sync keeps newly saved memories +recallable without waiting for the separate global checkpoint backfill. + ## Current Integration Points Memory V2 is only allowed to sit above existing memory code. diff --git a/docs/regression-guards/memory-v2-candidate-and-lifecycle.md b/docs/regression-guards/memory-v2-candidate-and-lifecycle.md index 9441dcf..3feba84 100644 --- a/docs/regression-guards/memory-v2-candidate-and-lifecycle.md +++ b/docs/regression-guards/memory-v2-candidate-and-lifecycle.md @@ -9,6 +9,10 @@ 作用域。若忘却开关开启，即使 `MEMORY_LIFECYCLE_ROLLOUT_MODE=off`，定时器也可能 归档全量用户的过期记忆。 +灰度还曾出现 MySQL 已成功保存新记忆，但 pgvector 仍只包含旧数据：runtime 从全局 +`updated_at=0` 游标开始做有限批次 backfill，新写入行可能长期排在批次之外。结果是 +`agent_memory_resolved` 显示已注入，但回答只拿到旧的无关记忆。 + ## 必须保留的行为 1. `MEMORY_CANDIDATE_PERSISTENCE_ENABLED=1` 且 MySQL 可用时，Portal 必须先执行 @@ -21,11 +25,15 @@ - `canary`：只允许 `MEMORY_LIFECYCLE_ROLLOUT_USER_IDS` 中的用户。 - `active`：才允许无 user scope 的全局任务。 5. `forgetMemory()` 和 `expire()` 都必须同时满足功能开关与 rollout 作用域。 +6. 用户记忆 `write/compact` 成功后，必须在返回前按 `userId + sessionId` 将本次活跃记忆 + 幂等 upsert 到 pgvector；候选晋升成功后也必须按实际晋升用户同步。不得依赖从零开始 + 的全局有限批次 backfill 来保证新记忆可立即召回。 ## 回归检查 ```bash -node --test memory-v2-personal-store.test.mjs memory-v2-lifecycle.test.mjs memory-v2-runtime.test.mjs +node --test memory-v2-personal-store.test.mjs memory-v2-lifecycle.test.mjs \ + memory-v2-pgvector-backfill.test.mjs memory-v2-runtime.test.mjs npm test ``` diff --git a/memory-v2-lifecycle.mjs b/memory-v2-lifecycle.mjs index 24cfd0f..1cc7af0 100644 --- a/memory-v2-lifecycle.mjs +++ b/memory-v2-lifecycle.mjs @@ -126,6 +126,7 @@ export function createMemoryV2LifecycleService({ pool = null, env = process.env, params, ); let promoted = 0; + const promotedUserIds = new Set(); for (const row of rows) { const id = crypto.createHash('sha256').update(`${row.user_id}\n${row.memory_type}\n${row.content}`).digest('hex'); const hash = crypto.createHash('sha256').update(`${row.user_id}\n${row.content}`).digest('hex'); @@ -135,10 +136,13 @@ export function createMemoryV2LifecycleService({ pool = null, env = process.env, VALUES (?,?,?,?,?,?,?,?,?,?,?,?)`, [id, row.user_id, row.memory_type === 'semantic' ? 'knowledge' : row.memory_type, hash, row.content, null, row.session_id, row.confidence, 'active', row.evidence_json, row.created_at, now()], ); - if (Number(result?.affectedRows ?? 0) > 0) promoted += 1; + if (Number(result?.affectedRows ?? 0) > 0) { + promoted += 1; + promotedUserIds.add(String(row.user_id)); + } } metrics.promote += 1; - return { ok: true, skipped: false, promoted }; + return { ok: true, skipped: false, promoted, promotedUserIds: [...promotedUserIds] }; } async function reflect({ userId = null } = {}) { diff --git a/memory-v2-lifecycle.test.mjs b/memory-v2-lifecycle.test.mjs index edeb868..66e2040 100644 --- a/memory-v2-lifecycle.test.mjs +++ b/memory-v2-lifecycle.test.mjs @@ -95,3 +95,35 @@ test('expire is limited to the configured canary user', async () => { assert.match(calls[0].sql, /user_id = \?/); assert.equal(calls[0].params.at(-1), 'u-1'); }); + +test('promotion reports the users whose memories were inserted', async () => { + const pool = { + async query(sql) { + if (sql.startsWith('SELECT * FROM h5_memory_v2_candidates')) { + return [[{ + user_id: 'u-1', + memory_type: 'episodic', + content: '记住灰度代号', + session_id: 's-1', + confidence: 0.9, + evidence_json: '{}', + created_at: 100, + }]]; + } + if (sql.startsWith('INSERT IGNORE INTO h5_user_memory_items')) { + return [{ affectedRows: 1 }]; + } + throw new Error(`Unexpected SQL: ${sql}`); + }, + }; + const lifecycle = createMemoryV2LifecycleService({ pool, env: { + MEMORY_LIFECYCLE_ENABLED: '1', + MEMORY_PROMOTION_ENABLED: '1', + MEMORY_LIFECYCLE_ROLLOUT_MODE: 'canary', + MEMORY_LIFECYCLE_ROLLOUT_USER_IDS: 'u-1', + } }); + + const result = await lifecycle.promote({ userId: 'u-1' }); + assert.equal(result.promoted, 1); + assert.deepEqual(result.promotedUserIds, ['u-1']); +}); diff --git a/memory-v2-pgvector-backfill.mjs b/memory-v2-pgvector-backfill.mjs index dc09116..79c5749 100644 --- a/memory-v2-pgvector-backfill.mjs +++ b/memory-v2-pgvector-backfill.mjs @@ -60,61 +60,16 @@ function normalizeLegacyMemory(row) { }; } -export async function loadLegacyMemoryBackfillBatch(mysqlPool, { cursor = {}, limit = DEFAULT_LIMIT } = {}) { - if (!mysqlPool?.query) { - throw new Error('loadLegacyMemoryBackfillBatch requires a MySQL pool with query(sql, params)'); - } - const resolvedCursor = normalizeCursor(cursor); - const resolvedLimit = normalizeLimit(limit); - const [rows] = await mysqlPool.query( - `SELECT id, user_id, label, memory_text, evidence_message_id, source_session_id, - confidence, created_at, updated_at - FROM h5_user_memory_items - WHERE status = 'active' - AND (updated_at > ? OR (updated_at = ? AND id > ?)) - ORDER BY updated_at ASC, id ASC - LIMIT ?`, - [resolvedCursor.updatedAt, resolvedCursor.updatedAt, resolvedCursor.id, resolvedLimit], - ); - const memories = (rows ?? []).map((row) => normalizeLegacyMemory(row)).filter(Boolean); - const last = memories.at(-1); - return { - memories, - nextCursor: last ? { updatedAt: last.updatedAt, id: last.id } : resolvedCursor, - hasMore: memories.length === resolvedLimit, - }; -} - -export async function backfillLegacyMemoriesToPgvector({ - mysqlPool, - pgPool = null, - embedMemory = null, - tableName = DEFAULT_TABLE, - cursor = {}, - limit = DEFAULT_LIMIT, - dryRun = true, -} = {}) { - const table = quoteIdent(tableName); - const batch = await loadLegacyMemoryBackfillBatch(mysqlPool, { cursor, limit }); - if (dryRun) { - return { - ok: true, - mode: 'dry-run', - scanned: batch.memories.length, - inserted: 0, - nextCursor: batch.nextCursor, - hasMore: batch.hasMore, - }; - } +async function upsertLegacyMemories({ memories, pgPool, embedMemory, tableName }) { if (!pgPool?.query) { - throw new Error('backfill apply requires a PostgreSQL pool with query(sql, params)'); + throw new Error('pgvector sync requires a PostgreSQL pool with query(sql, params)'); } if (typeof embedMemory !== 'function') { - throw new Error('backfill apply requires embedMemory(memory) => number[]'); + throw new Error('pgvector sync requires embedMemory(memory) => number[]'); } - + const table = quoteIdent(tableName); let inserted = 0; - for (const memory of batch.memories) { + for (const memory of memories) { const embedding = normalizeEmbedding(await embedMemory(memory)); if (!embedding) continue; await pgPool.query( @@ -146,6 +101,109 @@ export async function backfillLegacyMemoriesToPgvector({ ); inserted += 1; } + return inserted; +} + +export async function loadLegacyMemoryBackfillBatch(mysqlPool, { cursor = {}, limit = DEFAULT_LIMIT } = {}) { + if (!mysqlPool?.query) { + throw new Error('loadLegacyMemoryBackfillBatch requires a MySQL pool with query(sql, params)'); + } + const resolvedCursor = normalizeCursor(cursor); + const resolvedLimit = normalizeLimit(limit); + const [rows] = await mysqlPool.query( + `SELECT id, user_id, label, memory_text, evidence_message_id, source_session_id, + confidence, created_at, updated_at + FROM h5_user_memory_items + WHERE status = 'active' + AND (updated_at > ? OR (updated_at = ? AND id > ?)) + ORDER BY updated_at ASC, id ASC + LIMIT ?`, + [resolvedCursor.updatedAt, resolvedCursor.updatedAt, resolvedCursor.id, resolvedLimit], + ); + const memories = (rows ?? []).map((row) => normalizeLegacyMemory(row)).filter(Boolean); + const last = memories.at(-1); + return { + memories, + nextCursor: last ? { updatedAt: last.updatedAt, id: last.id } : resolvedCursor, + hasMore: memories.length === resolvedLimit, + }; +} + +export async function loadLegacyUserMemorySyncBatch( + mysqlPool, + { userId, sessionId = null, limit = DEFAULT_LIMIT } = {}, +) { + if (!mysqlPool?.query) { + throw new Error('loadLegacyUserMemorySyncBatch requires a MySQL pool with query(sql, params)'); + } + const resolvedUserId = String(userId ?? '').trim(); + if (!resolvedUserId) throw new Error('loadLegacyUserMemorySyncBatch requires userId'); + const resolvedSessionId = String(sessionId ?? '').trim(); + const resolvedLimit = normalizeLimit(limit); + const sessionScope = resolvedSessionId ? ' AND source_session_id = ?' : ''; + const params = resolvedSessionId + ? [resolvedUserId, resolvedSessionId, resolvedLimit] + : [resolvedUserId, resolvedLimit]; + const [rows] = await mysqlPool.query( + `SELECT id, user_id, label, memory_text, evidence_message_id, source_session_id, + confidence, created_at, updated_at + FROM h5_user_memory_items + WHERE status = 'active' + AND user_id = ?${sessionScope} + ORDER BY updated_at DESC, id DESC + LIMIT ?`, + params, + ); + return (rows ?? []).map((row) => normalizeLegacyMemory(row)).filter(Boolean); +} + +export async function syncLegacyUserMemoriesToPgvector({ + mysqlPool, + pgPool, + embedMemory, + tableName = DEFAULT_TABLE, + userId, + sessionId = null, + limit = DEFAULT_LIMIT, +} = {}) { + const memories = await loadLegacyUserMemorySyncBatch(mysqlPool, { userId, sessionId, limit }); + const inserted = await upsertLegacyMemories({ memories, pgPool, embedMemory, tableName }); + return { + ok: true, + mode: 'user-sync', + scanned: memories.length, + inserted, + userId: String(userId), + sessionId: sessionId == null ? null : String(sessionId), + }; +} + +export async function backfillLegacyMemoriesToPgvector({ + mysqlPool, + pgPool = null, + embedMemory = null, + tableName = DEFAULT_TABLE, + cursor = {}, + limit = DEFAULT_LIMIT, + dryRun = true, +} = {}) { + const batch = await loadLegacyMemoryBackfillBatch(mysqlPool, { cursor, limit }); + if (dryRun) { + return { + ok: true, + mode: 'dry-run', + scanned: batch.memories.length, + inserted: 0, + nextCursor: batch.nextCursor, + hasMore: batch.hasMore, + }; + } + const inserted = await upsertLegacyMemories({ + memories: batch.memories, + pgPool, + embedMemory, + tableName, + }); return { ok: true, diff --git a/memory-v2-pgvector-backfill.test.mjs b/memory-v2-pgvector-backfill.test.mjs index 1ce1714..2ec72e1 100644 --- a/memory-v2-pgvector-backfill.test.mjs +++ b/memory-v2-pgvector-backfill.test.mjs @@ -3,6 +3,8 @@ import test from 'node:test'; import { backfillLegacyMemoriesToPgvector, loadLegacyMemoryBackfillBatch, + loadLegacyUserMemorySyncBatch, + syncLegacyUserMemoriesToPgvector, } from './memory-v2-pgvector-backfill.mjs'; function createMysqlPool(rows) { @@ -162,3 +164,55 @@ test('backfillLegacyMemoriesToPgvector apply requires pg pool and embedding func /embedMemory/, ); }); + +test('loadLegacyUserMemorySyncBatch scopes recent memories to one user and session', async () => { + const calls = []; + const mysqlPool = { + async query(sql, params) { + calls.push({ sql, params }); + return [[legacyRows[1]]]; + }, + }; + + const memories = await loadLegacyUserMemorySyncBatch(mysqlPool, { + userId: 'user-1', + sessionId: 'session-2', + limit: 25, + }); + + assert.equal(memories.length, 1); + assert.equal(memories[0].id, 'mem-2'); + assert.match(calls[0].sql, /user_id = \?/); + assert.match(calls[0].sql, /source_session_id = \?/); + assert.match(calls[0].sql, /ORDER BY updated_at DESC/); + assert.deepEqual(calls[0].params, ['user-1', 'session-2', 25]); +}); + +test('syncLegacyUserMemoriesToPgvector immediately upserts the scoped memory', async () => { + const pgCalls = []; + const mysqlPool = { + async query() { + return [[legacyRows[0]]]; + }, + }; + + const result = await syncLegacyUserMemoriesToPgvector({ + mysqlPool, + pgPool: { + async query(sql, params) { + pgCalls.push({ sql, params }); + return { rows: [] }; + }, + }, + embedMemory: async () => [0.2, 0.4], + userId: 'user-1', + sessionId: 'session-1', + limit: 10, + }); + + assert.equal(result.mode, 'user-sync'); + assert.equal(result.scanned, 1); + assert.equal(result.inserted, 1); + assert.match(pgCalls[0].sql, /INSERT INTO "memory_embeddings"/); + assert.equal(pgCalls[0].params[4], 'mem-1'); +}); diff --git a/memory-v2-runtime.mjs b/memory-v2-runtime.mjs index f447dc0..617a8a8 100644 --- a/memory-v2-runtime.mjs +++ b/memory-v2-runtime.mjs @@ -12,7 +12,10 @@ import { import { createMem0HttpClient, createMem0MemoryBackend } from './memory-v2-mem0.mjs'; import { createNeo4jHttpClient, createNeo4jMemoryBackend } from './memory-v2-neo4j.mjs'; import { createPgvectorMemoryBackend } from './memory-v2-pgvector.mjs'; -import { backfillLegacyMemoriesToPgvector } from './memory-v2-pgvector-backfill.mjs'; +import { + backfillLegacyMemoriesToPgvector, + syncLegacyUserMemoriesToPgvector, +} from './memory-v2-pgvector-backfill.mjs'; import { createQdrantHttpClient, createQdrantMemoryBackend } from './memory-v2-qdrant.mjs'; import { createRedisStreamsClient, createRedisStreamsMemoryBackend } from './memory-v2-redis-streams.mjs'; import { createWeaviateHttpClient, createWeaviateMemoryBackend } from './memory-v2-weaviate.mjs'; @@ -464,8 +467,26 @@ export async function createMemoryV2Runtime({ let pgBackfillBusy = false; let pgBackfillCursor = { updatedAt: 0, id: '' }; - async function syncPgvectorFromLegacy() { + async function syncPgvectorFromLegacy({ userId = null, sessionId = null } = {}) { if (!pgBackfillEnabled || !pgPoolRef || !pgEmbedQuery || !mysqlPool?.query) return; + if (userId) { + try { + await syncLegacyUserMemoriesToPgvector({ + mysqlPool, + pgPool: pgPoolRef, + embedMemory: (item) => pgEmbedQuery(item.text, item), + tableName: env?.MEMORY_PGVECTOR_TABLE, + userId, + sessionId, + limit: Math.max(1, Number(env?.MEMORY_PGVECTOR_SYNC_USER_LIMIT ?? 50) || 50), + }); + } catch (err) { + logger?.warn?.( + `[memory-v2] pgvector user sync skipped: ${err instanceof Error ? err.message : err}`, + ); + } + return; + } if (pgBackfillBusy) return; pgBackfillBusy = true; try { @@ -493,12 +514,29 @@ export async function createMemoryV2Runtime({ } } + const originalLifecyclePromote = lifecycle.promote.bind(lifecycle); + lifecycle.promote = async (input = {}) => { + const result = await originalLifecyclePromote(input); + if (Number(result?.promoted ?? 0) > 0) { + const promotedUserIds = Array.isArray(result?.promotedUserIds) + ? result.promotedUserIds + : input?.userId ? [input.userId] : []; + for (const userId of promotedUserIds) { + await syncPgvectorFromLegacy({ userId }); + } + } + return result; + }; + function wrapMemorySideEffect(methodName) { const original = memory[methodName].bind(memory); memory[methodName] = async (input = {}) => { const result = await original(input); - if ((result?.memories ?? 0) > 0 || (result?.analyzed ?? 0) > 0) { - void syncPgvectorFromLegacy(); + if (input?.userId || (result?.memories ?? 0) > 0 || (result?.analyzed ?? 0) > 0) { + await syncPgvectorFromLegacy({ + userId: input?.userId ?? null, + sessionId: input?.sessionId ?? null, + }); } return result; }; diff --git a/memory-v2-runtime.test.mjs b/memory-v2-runtime.test.mjs index 69e8608..1128adc 100644 --- a/memory-v2-runtime.test.mjs +++ b/memory-v2-runtime.test.mjs @@ -127,6 +127,172 @@ test('createMemoryV2Runtime fails open when candidate schema initialization fail await memory.close(); }); +test('createMemoryV2Runtime syncs a successful user memory write into pgvector before returning', async () => { + const mysqlCalls = []; + const pgCalls = []; + class FakePool { + async query(sql, params) { + pgCalls.push({ sql, params }); + return { rows: [] }; + } + async end() {} + } + const mysqlPool = { + async query(sql, params) { + mysqlCalls.push({ sql, params }); + if (sql.includes('FROM h5_user_memory_items')) { + return [[{ + id: 'mem-new', + user_id: 'u-1', + label: 'fact', + memory_text: '灰度代号 MEM-NEW', + evidence_message_id: 'msg-1', + source_session_id: 's-new', + confidence: 0.9, + status: 'active', + created_at: 100, + updated_at: 200, + }]]; + } + throw new Error(`Unexpected MySQL query: ${sql}`); + }, + }; + const memory = await createMemoryV2Runtime({ + logger: silentLogger(), + mysqlPool, + legacyMemoryService: legacyService(), + env: { + MEMORY_ENABLED: '1', + MEMORY_EVENT_LOG_ENABLED: '1', + MEMORY_BACKEND: 'pgvector', + MEMORY_VECTOR_ENABLED: '1', + MEMORY_PGVECTOR_DATABASE_URL: 'postgresql://local/memory', + MEMORY_PGVECTOR_EMBEDDING_MODULE: './fake-embed.mjs', + }, + async importPg() { return { Pool: FakePool }; }, + async importModule() { return { embedQuery: async () => [0.1, 0.2] }; }, + }); + + const result = await memory.write({ userId: 'u-1', sessionId: 's-new', messages: [] }); + assert.equal(result.memories, 1); + assert.match(mysqlCalls[0].sql, /user_id = \?/); + assert.match(mysqlCalls[0].sql, /source_session_id = \?/); + assert.deepEqual(mysqlCalls[0].params, ['u-1', 's-new', 50]); + const insert = pgCalls.find((call) => call.sql.includes('INSERT INTO "memory_embeddings"')); + assert.ok(insert); + assert.equal(insert.params[1], '灰度代号 MEM-NEW'); + await memory.close(); +}); + +test('createMemoryV2Runtime retries scoped pgvector sync after a deduplicated memory write', async () => { + const pgCalls = []; + class FakePool { + async query(sql, params) { + pgCalls.push({ sql, params }); + return { rows: [] }; + } + async end() {} + } + const mysqlPool = { + async query(sql) { + if (sql.includes('FROM h5_user_memory_items')) { + return [[{ + id: 'mem-existing', + user_id: 'u-1', + label: 'fact', + memory_text: '需要重试同步的既有记忆', + source_session_id: 's-existing', + confidence: 0.9, + created_at: 100, + updated_at: 200, + }]]; + } + throw new Error(`Unexpected MySQL query: ${sql}`); + }, + }; + const memory = await createMemoryV2Runtime({ + logger: silentLogger(), + mysqlPool, + legacyMemoryService: { + ...legacyService(), + async saveAndAnalyze() { + return { saved: 0, analyzed: 0, memories: 0 }; + }, + }, + env: { + MEMORY_ENABLED: '1', + MEMORY_BACKEND: 'pgvector', + MEMORY_VECTOR_ENABLED: '1', + MEMORY_PGVECTOR_DATABASE_URL: 'postgresql://local/memory', + MEMORY_PGVECTOR_EMBEDDING_MODULE: './fake-embed.mjs', + }, + async importPg() { return { Pool: FakePool }; }, + async importModule() { return { embedQuery: async () => [0.1, 0.2] }; }, + }); + + const result = await memory.write({ userId: 'u-1', sessionId: 's-existing', messages: [] }); + assert.equal(result.memories, 0); + const insert = pgCalls.find((call) => call.sql.includes('INSERT INTO "memory_embeddings"')); + assert.ok(insert); + assert.equal(insert.params[1], '需要重试同步的既有记忆'); + await memory.close(); +}); + +test('createMemoryV2Runtime syncs promoted canary candidates into pgvector', async () => { + const pgCalls = []; + class FakePool { + async query(sql, params) { + pgCalls.push({ sql, params }); + return { rows: [] }; + } + async end() {} + } + const mysqlPool = { + async query(sql) { + if (sql.startsWith('SELECT * FROM h5_memory_v2_candidates')) { + return [[{ + user_id: 'u-1', memory_type: 'episodic', content: '候选灰度代号', + session_id: 's-1', confidence: 0.9, evidence_json: '{}', created_at: 100, + }]]; + } + if (sql.startsWith('INSERT IGNORE INTO h5_user_memory_items')) return [{ affectedRows: 1 }]; + if (sql.includes('FROM h5_user_memory_items')) { + return [[{ + id: 'mem-promoted', user_id: 'u-1', label: 'episodic', memory_text: '候选灰度代号', + source_session_id: 's-1', confidence: 0.9, status: 'active', created_at: 100, updated_at: 200, + }]]; + } + throw new Error(`Unexpected MySQL query: ${sql}`); + }, + }; + const memory = await createMemoryV2Runtime({ + logger: silentLogger(), + mysqlPool, + legacyMemoryService: legacyService(), + env: { + MEMORY_ENABLED: '1', + MEMORY_BACKEND: 'pgvector', + MEMORY_VECTOR_ENABLED: '1', + MEMORY_PGVECTOR_DATABASE_URL: 'postgresql://local/memory', + MEMORY_PGVECTOR_EMBEDDING_MODULE: './fake-embed.mjs', + MEMORY_LIFECYCLE_ENABLED: '1', + MEMORY_PROMOTION_ENABLED: '1', + MEMORY_LIFECYCLE_ROLLOUT_MODE: 'canary', + MEMORY_LIFECYCLE_ROLLOUT_USER_IDS: 'u-1', + }, + async importPg() { return { Pool: FakePool }; }, + async importModule() { return { embedQuery: async () => [0.3, 0.4] }; }, + }); + + const result = await memory.lifecycle.promote({ userId: 'u-1' }); + assert.equal(result.promoted, 1); + assert.deepEqual(result.promotedUserIds, ['u-1']); + const insert = pgCalls.find((call) => call.sql.includes('INSERT INTO "memory_embeddings"')); + assert.ok(insert); + assert.equal(insert.params[1], '候选灰度代号'); + await memory.close(); +}); + test('createMemoryV2Runtime selects pgvector only when pool and embedding are configured', async () => { const queries = []; let poolEnded = false;