From bfb1f6fea9448f7bfdb444b602d74b97a5757cdc Mon Sep 17 00:00:00 2001 From: john Date: Tue, 21 Jul 2026 22:55:19 +0800 Subject: [PATCH] fix(memory): initialize candidates and enforce lifecycle rollout --- AGENTS.md | 1 + docs/memory-v2/README.md | 11 +++- docs/regression-guards/README.md | 1 + .../memory-v2-candidate-and-lifecycle.md | 45 ++++++++++++++++ memory-v2-lifecycle.mjs | 15 +++++- memory-v2-lifecycle.test.mjs | 54 ++++++++++++++++++- memory-v2-runtime.mjs | 43 +++++++++++---- memory-v2-runtime.test.mjs | 51 ++++++++++++++++++ 8 files changed, 207 insertions(+), 14 deletions(-) create mode 100644 docs/regression-guards/memory-v2-candidate-and-lifecycle.md diff --git a/AGENTS.md b/AGENTS.md index 8506cd2..d0d306c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -30,6 +30,7 @@ bash scripts/check-release-ready.sh | MindSpace remote 页面 sync + storage 缺失缩略图 fallback | [docs/regression-guards/mindspace-remote-page-sync-and-thumbnail.md](docs/regression-guards/mindspace-remote-page-sync-and-thumbnail.md) | | Page Data 数据集注册、绑定与交付验收 | [docs/regression-guards/page-data-delivery-contract.md](docs/regression-guards/page-data-delivery-contract.md) | | H5 SSE 断线续播、Portal/Goose 游标映射与 Finish 终态恢复 | [docs/regression-guards/h5-session-stream-replay.md](docs/regression-guards/h5-session-stream-replay.md) | +| Memory V2 候选表初始化与生命周期灰度作用域 | [docs/regression-guards/memory-v2-candidate-and-lifecycle.md](docs/regression-guards/memory-v2-candidate-and-lifecycle.md) | 索引：[docs/regression-guards/README.md](docs/regression-guards/README.md) diff --git a/docs/memory-v2/README.md b/docs/memory-v2/README.md index fd7ae27..019ddf9 100644 --- a/docs/memory-v2/README.md +++ b/docs/memory-v2/README.md @@ -79,7 +79,16 @@ The additive user-scoped management API is: Lifecycle workers are disabled by default. When explicitly enabled they run expiration, conservative compaction observation, candidate promotion, and reflection observation according to the rollout mode; none of these operations -blocks the chat path. +blocks the chat path. `off` creates no worker scope and performs no lifecycle +mutation, `canary` is always user-scoped to the configured rollout IDs, and only +`active` permits an unscoped global worker run. + +When candidate persistence is enabled, the Portal runtime idempotently creates +`h5_memory_v2_candidates` before enabling the MySQL candidate store. DDL failure +is fail-open for Portal chat and falls back to bounded in-memory candidates; +`memind_adm` uses the same schema helper during bootstrap and fails startup rather +than serving a permanently broken candidate API. The table is additive and must +not be dropped as part of an application rollback. The pgvector adapter does not create tables or generate embeddings. It only defines the adapter contract for a future semantic memory backend and requires explicit `enabled: true`, an injected PostgreSQL pool, and either an input embedding or an injected `embedQuery(...)` function. diff --git a/docs/regression-guards/README.md b/docs/regression-guards/README.md index f0f4c6e..e15f550 100644 --- a/docs/regression-guards/README.md +++ b/docs/regression-guards/README.md @@ -10,6 +10,7 @@ | [mindspace-remote-page-sync-and-thumbnail.md](./mindspace-remote-page-sync-and-thumbnail.md) | ① remote 模式 public HTML 入库 sync ② storage 缺失时缩略图/读页回退 workspace HTML | | [page-data-delivery-contract.md](./page-data-delivery-contract.md) | 数据集注册、绑定、真实 page UUID 与交付验收 | | [h5-session-stream-replay.md](./h5-session-stream-replay.md) | Portal/Goose SSE 游标映射、断线续播与 Finish 终态恢复 | +| [memory-v2-candidate-and-lifecycle.md](./memory-v2-candidate-and-lifecycle.md) | 候选记忆表幂等初始化、Portal fail-open、生命周期 off/canary/active 作用域 | ## 自动化 diff --git a/docs/regression-guards/memory-v2-candidate-and-lifecycle.md b/docs/regression-guards/memory-v2-candidate-and-lifecycle.md new file mode 100644 index 0000000..9441dcf --- /dev/null +++ b/docs/regression-guards/memory-v2-candidate-and-lifecycle.md @@ -0,0 +1,45 @@ +# Memory V2 候选表与生命周期灰度守卫 + +## 已知故障 + +生产曾出现 `h5_memory_v2_candidates` 不存在：候选记忆写入和后台查询报错， +但 Agent 召回仍在 shadow 模式执行，因此表现为“有召回事件、回答未注入、个人记忆状态降级”。 + +同时，生命周期 worker 只检查了 worker 开关；`expire()` 没有检查 rollout +作用域。若忘却开关开启，即使 `MEMORY_LIFECYCLE_ROLLOUT_MODE=off`，定时器也可能 +归档全量用户的过期记忆。 + +## 必须保留的行为 + +1. `MEMORY_CANDIDATE_PERSISTENCE_ENABLED=1` 且 MySQL 可用时，Portal 必须先执行 + `ensurePersonalMemoryCandidateSchema()`，再创建候选记忆 store。 +2. 候选表 DDL 失败不能阻塞 Portal 聊天启动；Portal 应记录告警并退回 bounded-memory。 +3. memind_adm 必须在创建候选 store 前完成同一幂等建表；初始化失败时后台启动失败， + 不允许以“页面可用但候选接口持续报错”的半初始化状态运行。 +4. 生命周期 rollout 语义必须严格一致： + - `off`：不得执行任何写操作，也不得启动 worker 定时器。 + - `canary`：只允许 `MEMORY_LIFECYCLE_ROLLOUT_USER_IDS` 中的用户。 + - `active`：才允许无 user scope 的全局任务。 +5. `forgetMemory()` 和 `expire()` 都必须同时满足功能开关与 rollout 作用域。 + +## 回归检查 + +```bash +node --test memory-v2-personal-store.test.mjs memory-v2-lifecycle.test.mjs memory-v2-runtime.test.mjs +npm test +``` + +memind_adm 同时执行： + +```bash +node --test server/personal-memory-candidate-store.test.mjs +npm test +``` + +灰度上线后还必须验证：候选表存在、候选写入成功、后台可查询，以及一个测试用户完成 +“保存 → 候选 → 晋升 → 新会话召回 → 回答注入”的闭环。未通过前不得切换全量 active。 + +## 回滚 + +代码可以按标准 runtime/release 回滚；候选表是加法结构，回滚时保留空表或已有数据， +不得为了代码回滚直接删除表。先把候选、生命周期和 Agent 注入模式切回 shadow/off。 diff --git a/memory-v2-lifecycle.mjs b/memory-v2-lifecycle.mjs index 1d37d77..24cfd0f 100644 --- a/memory-v2-lifecycle.mjs +++ b/memory-v2-lifecycle.mjs @@ -33,6 +33,15 @@ export function resolveMemoryV2LifecyclePolicy(env = process.env) { }; } +export function resolveMemoryV2LifecycleWorkerScopes(policy = {}) { + if (!policy.enabled) return []; + if (policy.rolloutMode === 'active') return [null]; + if (policy.rolloutMode === 'canary') { + return [...new Set((policy.rolloutUserIds ?? []).map(String).filter(Boolean))]; + } + return []; +} + function normalizeItem(row) { return { id: String(row.id), @@ -71,7 +80,7 @@ export function createMemoryV2LifecycleService({ pool = null, env = process.env, } async function forgetMemory({ userId, memoryId } = {}) { - if (!canRun(userId) && !policy.forgettingEnabled) return { ok: true, updated: false, skipped: true, reason: 'disabled' }; + if (!policy.forgettingEnabled || !canRun(userId)) return { ok: true, updated: false, skipped: true, reason: 'disabled' }; if (!pool?.query || !userId || !memoryId) return { ok: false, updated: false, reason: 'invalid_input' }; const [result] = await pool.query( `UPDATE ${MEMORY_TABLE} SET status = 'deleted', updated_at = ? WHERE id = ? AND user_id = ? AND status <> 'deleted'`, @@ -82,7 +91,9 @@ export function createMemoryV2LifecycleService({ pool = null, env = process.env, } async function expire({ userId = null } = {}) { - if (!pool?.query || !policy.forgettingEnabled) return { ok: true, skipped: true, reason: 'disabled', expired: 0 }; + if (!pool?.query || !policy.forgettingEnabled || !canRun(userId)) { + return { ok: true, skipped: true, reason: 'disabled', expired: 0 }; + } const cutoff = now() - policy.retentionDays * 86400000; const params = [cutoff]; let scope = ''; diff --git a/memory-v2-lifecycle.test.mjs b/memory-v2-lifecycle.test.mjs index 8021548..edeb868 100644 --- a/memory-v2-lifecycle.test.mjs +++ b/memory-v2-lifecycle.test.mjs @@ -1,6 +1,10 @@ import test from 'node:test'; import assert from 'node:assert/strict'; -import { createMemoryV2LifecycleService, resolveMemoryV2LifecyclePolicy } from './memory-v2-lifecycle.mjs'; +import { + createMemoryV2LifecycleService, + resolveMemoryV2LifecyclePolicy, + resolveMemoryV2LifecycleWorkerScopes, +} from './memory-v2-lifecycle.mjs'; test('lifecycle policy defaults to safe disabled workers', () => { const policy = resolveMemoryV2LifecyclePolicy({}); @@ -25,6 +29,16 @@ test('lifecycle canary only runs for configured users', async () => { assert.equal((await lifecycle.promote({ userId: 'u-2' })).skipped, true); }); +test('lifecycle worker scopes match off, canary, and active rollout modes', () => { + assert.deepEqual(resolveMemoryV2LifecycleWorkerScopes({ enabled: true, rolloutMode: 'off' }), []); + assert.deepEqual(resolveMemoryV2LifecycleWorkerScopes({ + enabled: true, + rolloutMode: 'canary', + rolloutUserIds: ['u-1', 'u-1', 'u-2'], + }), ['u-1', 'u-2']); + assert.deepEqual(resolveMemoryV2LifecycleWorkerScopes({ enabled: true, rolloutMode: 'active' }), [null]); +}); + test('forget is fail-safe when lifecycle is disabled', async () => { const lifecycle = createMemoryV2LifecycleService({ env: { MEMORY_LIFECYCLE_ENABLED: '0' } }); assert.deepEqual(await lifecycle.forgetMemory({ userId: 'u-1', memoryId: 'm-1' }), { @@ -39,9 +53,45 @@ test('list and forget use user-scoped SQL', async () => { if (sql.startsWith('SELECT')) return [[{ id: 'm-1', user_id: 'u-1', label: 'fact', memory_text: 'x', status: 'active', confidence: 0.9, created_at: 1, updated_at: 2 }]]; return [{ affectedRows: 1 }]; } }; - const lifecycle = createMemoryV2LifecycleService({ pool, env: { MEMORY_LIFECYCLE_ENABLED: '1', MEMORY_LIFECYCLE_FORGETTING_ENABLED: '1' } }); + const lifecycle = createMemoryV2LifecycleService({ pool, env: { + MEMORY_LIFECYCLE_ENABLED: '1', + MEMORY_LIFECYCLE_FORGETTING_ENABLED: '1', + MEMORY_LIFECYCLE_ROLLOUT_MODE: 'active', + } }); assert.equal((await lifecycle.listMemories({ userId: 'u-1' }))[0].id, 'm-1'); assert.equal((await lifecycle.forgetMemory({ userId: 'u-1', memoryId: 'm-1' })).updated, true); assert.match(calls[0].sql, /user_id = \?/); assert.match(calls[1].sql, /user_id = \?/); }); + +test('forget and expire never mutate data while lifecycle rollout is off', async () => { + const calls = []; + const pool = { query: async (...args) => { calls.push(args); return [{ affectedRows: 1 }]; } }; + const lifecycle = createMemoryV2LifecycleService({ pool, env: { + MEMORY_LIFECYCLE_ENABLED: '1', + MEMORY_LIFECYCLE_FORGETTING_ENABLED: '1', + MEMORY_LIFECYCLE_ROLLOUT_MODE: 'off', + } }); + + assert.equal((await lifecycle.forgetMemory({ userId: 'u-1', memoryId: 'm-1' })).skipped, true); + assert.equal((await lifecycle.expire()).skipped, true); + assert.equal(calls.length, 0); +}); + +test('expire is limited to the configured canary user', async () => { + const calls = []; + const pool = { query: async (sql, params) => { calls.push({ sql, params }); return [{ affectedRows: 2 }]; } }; + const lifecycle = createMemoryV2LifecycleService({ pool, now: () => 1_000_000, env: { + MEMORY_LIFECYCLE_ENABLED: '1', + MEMORY_LIFECYCLE_FORGETTING_ENABLED: '1', + MEMORY_LIFECYCLE_ROLLOUT_MODE: 'canary', + MEMORY_LIFECYCLE_ROLLOUT_USER_IDS: 'u-1', + MEMORY_POLICY_RETENTION_DAYS: '1', + } }); + + assert.equal((await lifecycle.expire({ userId: 'u-2' })).skipped, true); + assert.equal((await lifecycle.expire({ userId: 'u-1' })).expired, 2); + assert.equal(calls.length, 1); + assert.match(calls[0].sql, /user_id = \?/); + assert.equal(calls[0].params.at(-1), 'u-1'); +}); diff --git a/memory-v2-runtime.mjs b/memory-v2-runtime.mjs index c7a6e1a..f447dc0 100644 --- a/memory-v2-runtime.mjs +++ b/memory-v2-runtime.mjs @@ -5,7 +5,10 @@ import { createLegacyMemoryBackend, createMemoryV2, resolveMemoryV2Policy } from import { createLettaHttpClient, createLettaMemoryBackend } from './memory-v2-letta.mjs'; import { createMemoryV2PluginBackends } from './memory-v2-plugin-backends.mjs'; import { createPersonalMemoryShadowPipeline } from './memory-v2-personal-shadow.mjs'; -import { createPersonalMemoryCandidateStore } from './memory-v2-personal-store.mjs'; +import { + createPersonalMemoryCandidateStore, + ensurePersonalMemoryCandidateSchema, +} from './memory-v2-personal-store.mjs'; import { createMem0HttpClient, createMem0MemoryBackend } from './memory-v2-mem0.mjs'; import { createNeo4jHttpClient, createNeo4jMemoryBackend } from './memory-v2-neo4j.mjs'; import { createPgvectorMemoryBackend } from './memory-v2-pgvector.mjs'; @@ -13,7 +16,10 @@ import { backfillLegacyMemoriesToPgvector } from './memory-v2-pgvector-backfill. 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'; -import { createMemoryV2LifecycleService } from './memory-v2-lifecycle.mjs'; +import { + createMemoryV2LifecycleService, + resolveMemoryV2LifecycleWorkerScopes, +} from './memory-v2-lifecycle.mjs'; const DEFAULT_PGVECTOR_URL_ENV = 'MEMORY_PGVECTOR_DATABASE_URL'; @@ -403,9 +409,17 @@ export async function createMemoryV2Runtime({ ], })); - const personalCandidateStore = readFlag(env, 'MEMORY_CANDIDATE_PERSISTENCE_ENABLED', false) - ? createPersonalMemoryCandidateStore(mysqlPool) - : null; + let personalCandidateStore = null; + if (readFlag(env, 'MEMORY_CANDIDATE_PERSISTENCE_ENABLED', false) && mysqlPool?.query) { + try { + await ensurePersonalMemoryCandidateSchema(mysqlPool); + personalCandidateStore = createPersonalMemoryCandidateStore(mysqlPool); + } catch (err) { + logger?.warn?.( + `[memory-v2] candidate persistence unavailable: ${err instanceof Error ? err.message : err}`, + ); + } + } const personalShadowPipeline = createPersonalMemoryShadowPipeline({ env, store: personalCandidateStore, @@ -420,11 +434,22 @@ export async function createMemoryV2Runtime({ const lifecycle = createMemoryV2LifecycleService({ pool: mysqlPool, env, logger }); let lifecycleTimer = null; const lifecycleWorkerEnabled = readFlag(env, 'MEMORY_LIFECYCLE_WORKER_ENABLED', false); - if (lifecycleWorkerEnabled && mysqlPool?.query) { + const lifecycleWorkerScopes = resolveMemoryV2LifecycleWorkerScopes(lifecycle.policy); + if (lifecycleWorkerEnabled && mysqlPool?.query && lifecycleWorkerScopes.length > 0) { const intervalMs = Math.max(60_000, Number(lifecycle.policy.compactIntervalHours ?? 24) * 3_600_000); - const runLifecycle = () => lifecycle.expire().then(() => lifecycle.compact()).then(() => lifecycle.promote()).then(() => lifecycle.reflect()).catch((err) => { - logger?.warn?.(`[memory-v2] lifecycle worker skipped: ${err instanceof Error ? err.message : err}`); - }); + const runLifecycle = async () => { + try { + for (const userId of lifecycleWorkerScopes) { + const input = userId == null ? {} : { userId }; + await lifecycle.expire(input); + await lifecycle.compact(input); + await lifecycle.promote(input); + await lifecycle.reflect(input); + } + } catch (err) { + logger?.warn?.(`[memory-v2] lifecycle worker skipped: ${err instanceof Error ? err.message : err}`); + } + }; lifecycleTimer = setInterval(runLifecycle, intervalMs); lifecycleTimer.unref?.(); } diff --git a/memory-v2-runtime.test.mjs b/memory-v2-runtime.test.mjs index 148f444..69e8608 100644 --- a/memory-v2-runtime.test.mjs +++ b/memory-v2-runtime.test.mjs @@ -76,6 +76,57 @@ test('createMemoryV2Runtime does not open pg pool when embedding module is missi assert.equal(pgvector.reason, 'embedding_module_not_configured'); }); +test('createMemoryV2Runtime ensures candidate schema before enabling MySQL persistence', async () => { + const calls = []; + const mysqlPool = { + async query(sql, params = []) { + calls.push({ sql, params }); + if (sql.startsWith('CREATE TABLE')) return [[], []]; + if (sql.startsWith('INSERT')) return [{ affectedRows: 1 }, []]; + throw new Error(`Unexpected SQL: ${sql}`); + }, + }; + const memory = await createMemoryV2Runtime({ + logger: silentLogger(), + legacyMemoryService: legacyService(), + mysqlPool, + env: { + MEMORY_CANDIDATE_ENABLED: '1', + MEMORY_CANDIDATE_MODE: 'shadow', + MEMORY_CANDIDATE_PERSISTENCE_ENABLED: '1', + }, + }); + + assert.equal(memory.getStatus().personalMemory.persistence, 'mysql'); + assert.match(calls[0].sql, /CREATE TABLE IF NOT EXISTS `h5_memory_v2_candidates`/); + const observed = await memory.observePersonalMemory({ + userId: 'u-1', + sessionId: 's-1', + messages: [{ role: 'user', content: '请记住我偏好使用简洁的中文回答' }], + }); + assert.equal(observed.accepted, 1); + assert.equal(calls[1].sql.startsWith('INSERT'), true); + await memory.close(); +}); + +test('createMemoryV2Runtime fails open when candidate schema initialization fails', async () => { + const logger = silentLogger(); + const memory = await createMemoryV2Runtime({ + logger, + legacyMemoryService: legacyService(), + mysqlPool: { async query() { throw new Error('ddl denied'); } }, + env: { + MEMORY_CANDIDATE_ENABLED: '1', + MEMORY_CANDIDATE_MODE: 'shadow', + MEMORY_CANDIDATE_PERSISTENCE_ENABLED: '1', + }, + }); + + assert.equal(memory.getStatus().personalMemory.persistence, 'bounded-memory'); + assert.match(logger.warnings[0], /candidate persistence unavailable: ddl denied/); + await memory.close(); +}); + test('createMemoryV2Runtime selects pgvector only when pool and embedding are configured', async () => { const queries = []; let poolEnded = false;