fix(memory): initialize candidates and enforce lifecycle rollout

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)
 

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.
 

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 作用域 |
 
 ## 自动化
 

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。

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 = '';

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');
+});

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?.();
   }

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;