feat(memory-v2): restore personal shadow pipeline and candidate store

Recover WIP modules from stash so admin can persist capability config,
observe chat writes in shadow mode, and review persisted candidates.

Co-authored-by: Cursor <cursoragent@cursor.com>

memory-v2-admin-config.mjs

@@ -22,6 +22,59 @@ const FIELD_SPECS = [
   { env: 'MEMIND_CHAT_ROUTER_TIMEOUT_MS', group: 'chatIntentRouter', field: 'timeoutMs', type: 'number' },
   { env: 'MEMIND_CHAT_ROUTER_FALLBACK_ROUTE', group: 'chatIntentRouter', field: 'fallbackRoute', type: 'string' },
 
+  { env: 'MEMORY_CANDIDATE_ENABLED', group: 'candidateMemory', field: 'enabled', type: 'boolean' },
+  { env: 'MEMORY_CANDIDATE_MODE', group: 'candidateMemory', field: 'mode', type: 'string' },
+  { env: 'MEMORY_CANDIDATE_MIN_IMPORTANCE', group: 'candidateMemory', field: 'minImportance', type: 'number' },
+  { env: 'MEMORY_CANDIDATE_MIN_CONFIDENCE', group: 'candidateMemory', field: 'minConfidence', type: 'number' },
+  { env: 'MEMORY_CANDIDATE_MAX_PENDING', group: 'candidateMemory', field: 'maxPending', type: 'number' },
+  { env: 'MEMORY_CANDIDATE_PERSISTENCE_ENABLED', group: 'candidateMemory', field: 'persistenceEnabled', type: 'boolean' },
+
+  { env: 'MEMORY_POLICY_ENABLED', group: 'policy', field: 'enabled', type: 'boolean' },
+  { env: 'MEMORY_POLICY_SAVE_EXPLICIT', group: 'policy', field: 'saveExplicit', type: 'boolean' },
+  { env: 'MEMORY_POLICY_REJECT_SENSITIVE', group: 'policy', field: 'rejectSensitive', type: 'boolean' },
+  { env: 'MEMORY_POLICY_REQUIRE_EVIDENCE', group: 'policy', field: 'requireEvidence', type: 'boolean' },
+  { env: 'MEMORY_POLICY_RETENTION_DAYS', group: 'policy', field: 'retentionDays', type: 'number' },
+
+  { env: 'MEMORY_RETRIEVER_ENABLED', group: 'retriever', field: 'enabled', type: 'boolean' },
+  { env: 'MEMORY_RETRIEVER_EPISODIC_ENABLED', group: 'retriever', field: 'episodicEnabled', type: 'boolean' },
+  { env: 'MEMORY_RETRIEVER_SEMANTIC_ENABLED', group: 'retriever', field: 'semanticEnabled', type: 'boolean' },
+  { env: 'MEMORY_RETRIEVER_PREFERENCE_ENABLED', group: 'retriever', field: 'preferenceEnabled', type: 'boolean' },
+  { env: 'MEMORY_RETRIEVER_GOAL_ENABLED', group: 'retriever', field: 'goalEnabled', type: 'boolean' },
+  { env: 'MEMORY_RETRIEVER_LIMIT', group: 'retriever', field: 'limit', type: 'number' },
+  { env: 'MEMORY_RETRIEVER_TOKEN_BUDGET', group: 'retriever', field: 'tokenBudget', type: 'number' },
+  { env: 'MEMORY_RETRIEVER_TIMEOUT_MS', group: 'retriever', field: 'timeoutMs', type: 'number' },
+
+  { env: 'MEMORY_LIFECYCLE_ENABLED', group: 'lifecycle', field: 'enabled', type: 'boolean' },
+  { env: 'MEMORY_LIFECYCLE_DEDUPE_ENABLED', group: 'lifecycle', field: 'dedupeEnabled', type: 'boolean' },
+  { env: 'MEMORY_LIFECYCLE_CONFLICT_REVIEW', group: 'lifecycle', field: 'conflictReview', type: 'boolean' },
+  { env: 'MEMORY_LIFECYCLE_DECAY_ENABLED', group: 'lifecycle', field: 'decayEnabled', type: 'boolean' },
+  { env: 'MEMORY_LIFECYCLE_FORGETTING_ENABLED', group: 'lifecycle', field: 'forgettingEnabled', type: 'boolean' },
+  { env: 'MEMORY_LIFECYCLE_COMPACT_INTERVAL_HOURS', group: 'lifecycle', field: 'compactIntervalHours', type: 'number' },
+
+  { env: 'MEMORY_PERSONA_ENABLED', group: 'persona', field: 'enabled', type: 'boolean' },
+  { env: 'MEMORY_PERSONA_PROVIDER', group: 'persona', field: 'provider', type: 'string' },
+  { env: 'MEMORY_PERSONA_SHADOW_MODE', group: 'persona', field: 'shadowMode', type: 'boolean' },
+  { env: 'MEMORY_PERSONA_MAX_TOKENS', group: 'persona', field: 'maxTokens', type: 'number' },
+  { env: 'MEMORY_PERSONA_CACHE_TTL_SECONDS', group: 'persona', field: 'cacheTtlSeconds', type: 'number' },
+
+  { env: 'MEMORY_GRAPH_ENABLED', group: 'graph', field: 'enabled', type: 'boolean' },
+  { env: 'MEMORY_GRAPH_PROVIDER', group: 'graph', field: 'provider', type: 'string' },
+  { env: 'MEMORY_GRAPH_MAX_DEPTH', group: 'graph', field: 'maxDepth', type: 'number' },
+  { env: 'MEMORY_GRAPH_RELATION_LIMIT', group: 'graph', field: 'relationLimit', type: 'number' },
+
+  { env: 'MEMORY_USER_MANAGEMENT_ENABLED', group: 'userMemory', field: 'enabled', type: 'boolean' },
+  { env: 'MEMORY_USER_REVIEW_ENABLED', group: 'userMemory', field: 'reviewEnabled', type: 'boolean' },
+  { env: 'MEMORY_USER_CORRECTION_ENABLED', group: 'userMemory', field: 'correctionEnabled', type: 'boolean' },
+  { env: 'MEMORY_USER_PIN_ENABLED', group: 'userMemory', field: 'pinEnabled', type: 'boolean' },
+  { env: 'MEMORY_USER_FORGET_ENABLED', group: 'userMemory', field: 'forgetEnabled', type: 'boolean' },
+  { env: 'MEMORY_USER_DELETE_PROPAGATION', group: 'userMemory', field: 'deletePropagation', type: 'boolean' },
+
+  { env: 'MEMORY_PLUGIN_HEALTH_ENABLED', group: 'pluginHealth', field: 'enabled', type: 'boolean' },
+  { env: 'MEMORY_PLUGIN_HEALTH_INTERVAL_SECONDS', group: 'pluginHealth', field: 'intervalSeconds', type: 'number' },
+  { env: 'MEMORY_PLUGIN_HEALTH_TIMEOUT_MS', group: 'pluginHealth', field: 'timeoutMs', type: 'number' },
+  { env: 'MEMORY_PLUGIN_HEALTH_FAILURE_THRESHOLD', group: 'pluginHealth', field: 'failureThreshold', type: 'number' },
+  { env: 'MEMORY_PLUGIN_HEALTH_AUTO_FALLBACK', group: 'pluginHealth', field: 'autoFallback', type: 'boolean' },
+
   { env: 'MEMORY_PGVECTOR_ENABLED', group: 'pgvector', field: 'enabled', type: 'boolean' },
   { env: 'MEMORY_PGVECTOR_DATABASE_URL', group: 'pgvector', field: 'databaseUrl', type: 'secret' },
   { env: 'MEMORY_PGVECTOR_TABLE', group: 'pgvector', field: 'table', type: 'string' },
@@ -95,6 +148,14 @@ const FIELD_SPECS = [
 const GROUPS = [
   'global',
   'chatIntentRouter',
+  'candidateMemory',
+  'policy',
+  'retriever',
+  'lifecycle',
+  'persona',
+  'graph',
+  'userMemory',
+  'pluginHealth',
   'pgvector',
   'qdrant',
   'weaviate',

memory-v2-admin-config.test.mjs

@@ -81,6 +81,36 @@ test('memory v2 admin config service persists non-secret and secret patches', as
       url: 'http://127.0.0.1:6333',
       apiKey: 'secret-qdrant',
     },
+    candidateMemory: {
+      enabled: true,
+      mode: 'shadow',
+      minImportance: '0.75',
+      minConfidence: '0.85',
+      maxPending: '500',
+      persistenceEnabled: true,
+    },
+    policy: {
+      enabled: true,
+      saveExplicit: true,
+      rejectSensitive: true,
+      requireEvidence: true,
+      retentionDays: '365',
+    },
+    retriever: {
+      enabled: true,
+      episodicEnabled: true,
+      semanticEnabled: true,
+      preferenceEnabled: true,
+      goalEnabled: true,
+      limit: '12',
+      tokenBudget: '1800',
+      timeoutMs: '1200',
+    },
+    lifecycle: { enabled: true, dedupeEnabled: true, conflictReview: true },
+    persona: { enabled: true, provider: 'ai-mind', shadowMode: true },
+    graph: { enabled: true, provider: 'postgres', maxDepth: '2' },
+    userMemory: { enabled: true, reviewEnabled: true, forgetEnabled: true, deletePropagation: true },
+    pluginHealth: { enabled: true, intervalSeconds: '60', timeoutMs: '1500', failureThreshold: '3', autoFallback: true },
   }, { updatedBy: 'admin-1' });
 
   assert.equal(updated.config.global.enabled, true);
@@ -91,6 +121,15 @@ test('memory v2 admin config service persists non-secret and secret patches', as
   assert.equal(updated.config.qdrant.enabled, true);
   assert.equal(updated.config.qdrant.url, 'http://127.0.0.1:6333');
   assert.equal(updated.config.qdrant.apiKeyConfigured, true);
+  assert.equal(updated.config.candidateMemory.mode, 'shadow');
+  assert.equal(updated.config.candidateMemory.persistenceEnabled, true);
+  assert.equal(updated.config.policy.requireEvidence, true);
+  assert.equal(updated.config.retriever.tokenBudget, '1800');
+  assert.equal(updated.config.lifecycle.dedupeEnabled, true);
+  assert.equal(updated.config.persona.provider, 'ai-mind');
+  assert.equal(updated.config.graph.provider, 'postgres');
+  assert.equal(updated.config.userMemory.deletePropagation, true);
+  assert.equal(updated.config.pluginHealth.autoFallback, true);
   assert.equal(updated.updatedBy, 'admin-1');
 
   const runtimeState = await service.getRuntimeState();
@@ -107,6 +146,16 @@ test('memory v2 admin config service persists non-secret and secret patches', as
   assert.equal(runtimeState.overrides.MEMORY_QDRANT_ENABLED, '1');
   assert.equal(runtimeState.overrides.MEMORY_QDRANT_URL, 'http://127.0.0.1:6333');
   assert.equal(runtimeState.overrides.MEMORY_QDRANT_API_KEY, 'secret-qdrant');
+  assert.equal(runtimeState.overrides.MEMORY_CANDIDATE_ENABLED, '1');
+  assert.equal(runtimeState.overrides.MEMORY_CANDIDATE_MODE, 'shadow');
+  assert.equal(runtimeState.overrides.MEMORY_CANDIDATE_PERSISTENCE_ENABLED, '1');
+  assert.equal(runtimeState.overrides.MEMORY_POLICY_REQUIRE_EVIDENCE, '1');
+  assert.equal(runtimeState.overrides.MEMORY_RETRIEVER_TOKEN_BUDGET, '1800');
+  assert.equal(runtimeState.overrides.MEMORY_LIFECYCLE_DEDUPE_ENABLED, '1');
+  assert.equal(runtimeState.overrides.MEMORY_PERSONA_PROVIDER, 'ai-mind');
+  assert.equal(runtimeState.overrides.MEMORY_GRAPH_PROVIDER, 'postgres');
+  assert.equal(runtimeState.overrides.MEMORY_USER_DELETE_PROPAGATION, '1');
+  assert.equal(runtimeState.overrides.MEMORY_PLUGIN_HEALTH_AUTO_FALLBACK, '1');
 });
 
 test('memory v2 admin config internals flatten booleans and numbers consistently', () => {

memory-v2-personal-shadow.mjs

@@ -0,0 +1,232 @@
+import crypto from 'node:crypto';
+import { deriveUserFacingText } from './conversation-display.mjs';
+
+const TRUE_VALUES = new Set(['1', 'true', 'yes', 'on']);
+const FALSE_VALUES = new Set(['0', 'false', 'no', 'off']);
+const SECRET_PATTERNS = [
+  /\b(?:api[_-]?key|access[_-]?token|secret|password|passwd)\b\s*[:=]/i,
+  /\bsk-[a-z0-9_-]{12,}\b/i,
+  /-----BEGIN [A-Z ]+PRIVATE KEY-----/,
+  /(?:密码|口令|密钥|令牌)\s*[：:=]\s*\S+/i,
+];
+const TRIVIAL_PATTERNS = [
+  /^(?:你好|您好|谢谢|好的|可以|收到|再见|hi|hello|thanks)[！!。.\s]*$/i,
+  /^(?:查一下|搜一下|看看|继续|下一步)[。.!！?？\s]*$/i,
+];
+
+function readFlag(env, key, fallback = false) {
+  const raw = env?.[key];
+  if (raw == null || raw === '') return fallback;
+  const normalized = String(raw).trim().toLowerCase();
+  if (TRUE_VALUES.has(normalized)) return true;
+  if (FALSE_VALUES.has(normalized)) return false;
+  return fallback;
+}
+
+function readNumber(env, key, fallback, { min = -Infinity, max = Infinity } = {}) {
+  const value = Number(env?.[key]);
+  if (!Number.isFinite(value)) return fallback;
+  return Math.min(max, Math.max(min, value));
+}
+
+function normalizeText(value, maxLength = 2000) {
+  const text = String(value ?? '').replace(/\s+/g, ' ').trim();
+  if (!text) return '';
+  return text.length > maxLength ? text.slice(0, maxLength) : text;
+}
+
+function messageText(message) {
+  if (typeof message === 'string') return normalizeText(message);
+  if (!message || typeof message !== 'object') return '';
+  const displayText = message.displayText
+    ?? message.display_text
+    ?? message.metadata?.displayText
+    ?? message.metadata?.display_text;
+  if (typeof displayText === 'string' && displayText.trim()) {
+    return normalizeText(displayText);
+  }
+  const direct = message.text ?? message.content ?? message.message ?? '';
+  if (typeof direct === 'string') return normalizeText(deriveUserFacingText(direct));
+  if (Array.isArray(direct)) {
+    return normalizeText(deriveUserFacingText(
+      direct.map((item) => item?.text ?? item?.content ?? '').join('\n'),
+    ));
+  }
+  return '';
+}
+
+function stableHash(value) {
+  return crypto.createHash('sha256').update(String(value)).digest('hex');
+}
+
+function classify(text) {
+  const rules = [
+    { type: 'episodic', importance: 0.95, confidence: 0.98, pattern: /(?:请记住|记住|以后要|从现在起)/i, reason: 'explicit_memory_request' },
+    { type: 'episodic', importance: 0.9, confidence: 0.9, pattern: /(?:决定|确定|统一采用|必须|不允许|禁止|最终选择)/i, reason: 'decision_signal' },
+    { type: 'preference', importance: 0.8, confidence: 0.88, pattern: /(?:我喜欢|我偏好|我不喜欢|我的习惯|倾向于|更希望)/i, reason: 'preference_signal' },
+    { type: 'goal', importance: 0.85, confidence: 0.86, pattern: /(?:长期目标|当前目标|目标是|计划要|正在建设|准备实现|希望长期)/i, reason: 'goal_signal' },
+    { type: 'semantic', importance: 0.75, confidence: 0.82, pattern: /(?:技术栈|架构原则|项目使用|系统采用|仓库位于|运行在)/i, reason: 'stable_fact_signal' },
+  ];
+  return rules.find((rule) => rule.pattern.test(text)) ?? null;
+}
+
+export function resolvePersonalShadowConfig(env = process.env) {
+  const candidateEnabled = readFlag(env, 'MEMORY_CANDIDATE_ENABLED', false);
+  const requestedMode = normalizeText(env?.MEMORY_CANDIDATE_MODE || (candidateEnabled ? 'shadow' : 'off'), 20).toLowerCase();
+  return {
+    enabled: candidateEnabled && requestedMode !== 'off',
+    requestedMode,
+    effectiveMode: candidateEnabled && requestedMode !== 'off' ? 'shadow' : 'off',
+    minImportance: readNumber(env, 'MEMORY_CANDIDATE_MIN_IMPORTANCE', 0.7, { min: 0, max: 1 }),
+    minConfidence: readNumber(env, 'MEMORY_CANDIDATE_MIN_CONFIDENCE', 0.8, { min: 0, max: 1 }),
+    maxPending: Math.round(readNumber(env, 'MEMORY_CANDIDATE_MAX_PENDING', 500, { min: 1, max: 5000 })),
+    policyEnabled: readFlag(env, 'MEMORY_POLICY_ENABLED', true),
+    saveExplicit: readFlag(env, 'MEMORY_POLICY_SAVE_EXPLICIT', true),
+    rejectSensitive: readFlag(env, 'MEMORY_POLICY_REJECT_SENSITIVE', true),
+    requireEvidence: readFlag(env, 'MEMORY_POLICY_REQUIRE_EVIDENCE', true),
+    healthEnabled: readFlag(env, 'MEMORY_PLUGIN_HEALTH_ENABLED', true),
+  };
+}
+
+export function createPersonalMemoryShadowPipeline({ env = process.env, now = () => Date.now(), store = null } = {}) {
+  const config = resolvePersonalShadowConfig(env);
+  const candidates = [];
+  const candidateHashes = new Set();
+  const metrics = {
+    runs: 0,
+    observedMessages: 0,
+    accepted: 0,
+    rejected: 0,
+    deduped: 0,
+    errors: 0,
+    lastRunAt: null,
+    lastError: null,
+  };
+
+  function reject(reason) {
+    metrics.rejected += 1;
+    return { accepted: false, reason };
+  }
+
+  function evaluate({ userId, sessionId, message, index }) {
+    const text = messageText(message);
+    if (!text || text.length < 8) return reject('too_short');
+    if (TRIVIAL_PATTERNS.some((pattern) => pattern.test(text))) return reject('trivial');
+    if (config.rejectSensitive && SECRET_PATTERNS.some((pattern) => pattern.test(text))) {
+      return reject('sensitive_content');
+    }
+    const classification = classify(text);
+    if (!classification) return reject('no_durable_signal');
+    if (classification.reason === 'explicit_memory_request' && !config.saveExplicit) {
+      return reject('explicit_memory_disabled');
+    }
+    if (config.policyEnabled && classification.importance < config.minImportance) {
+      return reject('below_importance_threshold');
+    }
+    if (config.policyEnabled && classification.confidence < config.minConfidence) {
+      return reject('below_confidence_threshold');
+    }
+    const contentHash = stableHash(`${userId}\n${classification.type}\n${text.toLowerCase()}`);
+    if (candidateHashes.has(contentHash)) {
+      metrics.deduped += 1;
+      return { accepted: false, reason: 'duplicate' };
+    }
+    const capturedAt = now();
+    const evidence = {
+      sourceType: 'message',
+      sourceId: `${sessionId || 'unknown'}:${index}`,
+      evidenceHash: stableHash(`${sessionId || ''}\n${index}\n${text}`),
+      capturedAt,
+    };
+    if (config.requireEvidence && !evidence.sourceId) return reject('evidence_required');
+    const candidate = {
+      id: `pmc_${contentHash.slice(0, 24)}`,
+      userId: String(userId),
+      sessionId: sessionId ? String(sessionId) : null,
+      memoryType: classification.type,
+      content: text,
+      importance: classification.importance,
+      confidence: classification.confidence,
+      status: 'candidate',
+      policyReason: classification.reason,
+      evidence,
+      createdAt: capturedAt,
+    };
+    candidateHashes.add(contentHash);
+    candidates.push(candidate);
+    while (candidates.length > config.maxPending) {
+      const removed = candidates.shift();
+      if (removed) candidateHashes.delete(stableHash(`${removed.userId}\n${removed.memoryType}\n${removed.content.toLowerCase()}`));
+    }
+    metrics.accepted += 1;
+    return { accepted: true, candidate };
+  }
+
+  async function observeWrite({ userId, sessionId, messages = [] } = {}) {
+    if (!config.enabled) return { enabled: false, skipped: true, reason: 'disabled' };
+    metrics.runs += 1;
+    metrics.lastRunAt = now();
+    if (!userId || !Array.isArray(messages)) {
+      return { enabled: true, skipped: true, reason: 'invalid_input' };
+    }
+    try {
+      const results = [];
+      for (let index = 0; index < messages.length; index += 1) {
+        const message = messages[index];
+        const role = normalizeText(message?.role ?? message?.sender ?? '', 30).toLowerCase();
+        if (role && !['user', 'human'].includes(role)) continue;
+        metrics.observedMessages += 1;
+        const result = evaluate({ userId, sessionId, message, index });
+        if (result.accepted && store?.saveCandidate) {
+          const persisted = await store.saveCandidate(result.candidate);
+          if (persisted?.inserted === false) {
+            metrics.deduped += 1;
+          }
+        }
+        results.push(result);
+      }
+      return {
+        enabled: true,
+        skipped: false,
+        mode: config.effectiveMode,
+        observed: results.length,
+        accepted: results.filter((result) => result.accepted).length,
+        rejected: results.filter((result) => !result.accepted).length,
+      };
+    } catch (err) {
+      metrics.errors += 1;
+      metrics.lastError = err instanceof Error ? err.message : String(err);
+      return { enabled: true, skipped: true, reason: 'pipeline_failure' };
+    }
+  }
+
+  function getStatus() {
+    return {
+      enabled: config.enabled,
+      requestedMode: config.requestedMode,
+      effectiveMode: config.effectiveMode,
+      phase: 'shadow-candidate-v1',
+      injectionEnabled: false,
+      persistence: store?.saveCandidate ? 'mysql' : 'bounded-memory',
+      pendingCandidates: candidates.length,
+      health: {
+        enabled: config.healthEnabled,
+        state: metrics.errors > 0 ? 'degraded' : 'healthy',
+        ...metrics,
+      },
+      policy: {
+        enabled: config.policyEnabled,
+        minImportance: config.minImportance,
+        minConfidence: config.minConfidence,
+        rejectSensitive: config.rejectSensitive,
+        requireEvidence: config.requireEvidence,
+      },
+    };
+  }
+
+  function listCandidates({ limit = 50 } = {}) {
+    return candidates.slice(-Math.max(1, Math.min(200, Number(limit) || 50))).reverse();
+  }
+
+  return { config, observeWrite, getStatus, listCandidates };
+}

memory-v2-personal-shadow.test.mjs

@@ -0,0 +1,193 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import { createPersonalMemoryShadowPipeline, resolvePersonalShadowConfig } from './memory-v2-personal-shadow.mjs';
+import { createMemoryV2 } from './memory-v2.mjs';
+
+test('personal memory shadow config never activates response injection', () => {
+  const config = resolvePersonalShadowConfig({
+    MEMORY_CANDIDATE_ENABLED: '1',
+    MEMORY_CANDIDATE_MODE: 'active',
+  });
+  assert.equal(config.enabled, true);
+  assert.equal(config.requestedMode, 'active');
+  assert.equal(config.effectiveMode, 'shadow');
+});
+
+test('shadow pipeline extracts durable candidates with evidence and deduplicates', async () => {
+  let clock = 1000;
+  const pipeline = createPersonalMemoryShadowPipeline({
+    env: {
+      MEMORY_CANDIDATE_ENABLED: '1',
+      MEMORY_CANDIDATE_MODE: 'shadow',
+      MEMORY_POLICY_ENABLED: '1',
+      MEMORY_POLICY_REQUIRE_EVIDENCE: '1',
+      MEMORY_CANDIDATE_MIN_IMPORTANCE: '0.7',
+      MEMORY_CANDIDATE_MIN_CONFIDENCE: '0.8',
+    },
+    now: () => clock += 1,
+  });
+  const input = {
+    userId: 'u1',
+    sessionId: 's1',
+    messages: [
+      { role: 'user', text: '我决定所有生产发布必须从完整 main 分支打包。' },
+      { role: 'assistant', text: '收到。' },
+    ],
+  };
+  const first = await pipeline.observeWrite(input);
+  const second = await pipeline.observeWrite(input);
+
+  assert.equal(first.accepted, 1);
+  assert.equal(second.accepted, 0);
+  assert.equal(pipeline.listCandidates().length, 1);
+  assert.equal(pipeline.listCandidates()[0].memoryType, 'episodic');
+  assert.equal(pipeline.listCandidates()[0].evidence.sourceId, 's1:0');
+  assert.equal(pipeline.getStatus().health.deduped, 1);
+  assert.equal(pipeline.getStatus().injectionEnabled, false);
+});
+
+test('shadow policy rejects trivial and sensitive messages', async () => {
+  const pipeline = createPersonalMemoryShadowPipeline({
+    env: {
+      MEMORY_CANDIDATE_ENABLED: '1',
+      MEMORY_CANDIDATE_MODE: 'shadow',
+      MEMORY_POLICY_REJECT_SENSITIVE: '1',
+    },
+  });
+  const result = await pipeline.observeWrite({
+    userId: 'u1',
+    sessionId: 's1',
+    messages: [
+      { role: 'user', text: '谢谢' },
+      { role: 'user', text: '请记住 API_KEY=sk-this-is-a-secret-token' },
+      { role: 'user', text: '今天天气怎么样？' },
+    ],
+  });
+  assert.equal(result.accepted, 0);
+  assert.equal(result.rejected, 3);
+  assert.equal(pipeline.listCandidates().length, 0);
+});
+
+test('shadow pipeline prefers user-facing displayText over internal prompt content', async () => {
+  const pipeline = createPersonalMemoryShadowPipeline({
+    env: { MEMORY_CANDIDATE_ENABLED: '1', MEMORY_CANDIDATE_MODE: 'shadow' },
+  });
+  await pipeline.observeWrite({
+    userId: 'u1',
+    sessionId: 's1',
+    messages: [{
+      role: 'user',
+      content: [{ type: 'text', text: '[用户身份] 内部前缀\n我决定先测试再发布。' }],
+      metadata: { displayText: '我决定先测试再发布。' },
+    }],
+  });
+  assert.equal(pipeline.listCandidates().length, 1);
+  assert.equal(pipeline.listCandidates()[0].content, '我决定先测试再发布。');
+  assert.doesNotMatch(pipeline.listCandidates()[0].content, /用户身份/);
+});
+
+test('shadow pipeline strips protected user identity prefix when displayText is absent', async () => {
+  const pipeline = createPersonalMemoryShadowPipeline({
+    env: { MEMORY_CANDIDATE_ENABLED: '1', MEMORY_CANDIDATE_MODE: 'shadow' },
+  });
+  await pipeline.observeWrite({
+    userId: 'u1',
+    sessionId: 's1',
+    messages: [{
+      role: 'user',
+      content: [{
+        type: 'text',
+        text: '[用户身份]\n- 当前登录用户称呼：John\n- 内部提示\n\n我决定先测试再发布。',
+      }],
+    }],
+  });
+  assert.equal(pipeline.listCandidates().length, 1);
+  assert.equal(pipeline.listCandidates()[0].content, '我决定先测试再发布。');
+});
+
+test('shadow candidate pool remains bounded', async () => {
+  const pipeline = createPersonalMemoryShadowPipeline({
+    env: {
+      MEMORY_CANDIDATE_ENABLED: '1',
+      MEMORY_CANDIDATE_MODE: 'shadow',
+      MEMORY_CANDIDATE_MAX_PENDING: '2',
+    },
+  });
+  await pipeline.observeWrite({
+    userId: 'u1',
+    sessionId: 's1',
+    messages: [
+      { role: 'user', text: '我决定项目一使用完整 main 发布。' },
+      { role: 'user', text: '我决定项目二使用完整 main 发布。' },
+      { role: 'user', text: '我决定项目三使用完整 main 发布。' },
+    ],
+  });
+  assert.equal(pipeline.listCandidates().length, 2);
+  assert.match(pipeline.listCandidates()[0].content, /项目三/);
+});
+
+test('shadow pipeline persists accepted candidates when a store is configured', async () => {
+  const saved = [];
+  const pipeline = createPersonalMemoryShadowPipeline({
+    env: { MEMORY_CANDIDATE_ENABLED: '1', MEMORY_CANDIDATE_MODE: 'shadow' },
+    store: { async saveCandidate(candidate) { saved.push(candidate); return { inserted: true }; } },
+  });
+  await pipeline.observeWrite({
+    userId: 'u1',
+    sessionId: 's1',
+    messages: [{ role: 'user', text: '我的长期目标是建设一个持续成长的 Personal Agent。' }],
+  });
+  assert.equal(saved.length, 1);
+  assert.equal(saved[0].memoryType, 'goal');
+  assert.equal(pipeline.getStatus().persistence, 'mysql');
+});
+
+test('Memory V2 does not wait for the shadow pipeline before returning legacy write result', async () => {
+  let releaseShadow;
+  const shadowBlocked = new Promise((resolve) => { releaseShadow = resolve; });
+  const memory = createMemoryV2({
+    policy: { enabled: true, eventLogEnabled: true, failOpen: true, backend: 'legacy' },
+    backends: [{
+      name: 'legacy-conversation-memory',
+      isAvailable: () => true,
+      async write() { return { saved: 1, analyzed: 0, memories: 0 }; },
+    }],
+    personalShadowPipeline: {
+      config: { enabled: true },
+      observeWrite: () => shadowBlocked,
+      getStatus: () => ({ enabled: true }),
+    },
+  });
+  const result = await Promise.race([
+    memory.write({ userId: 'u1', sessionId: 's1', messages: [] }),
+    new Promise((_, reject) => setTimeout(() => reject(new Error('write waited for shadow')), 50)),
+  ]);
+  assert.equal(result.saved, 1);
+  releaseShadow();
+});
+
+test('Memory V2 exposes a shadow-only observation entry without calling legacy write', async () => {
+  let legacyWrites = 0;
+  const observed = [];
+  const memory = createMemoryV2({
+    policy: { enabled: true, eventLogEnabled: true, failOpen: true, backend: 'legacy' },
+    backends: [{
+      name: 'legacy-conversation-memory',
+      isAvailable: () => true,
+      async write() { legacyWrites += 1; return { saved: 1 }; },
+    }],
+    personalShadowPipeline: {
+      config: { enabled: true },
+      async observeWrite(input) { observed.push(input); return { accepted: 1 }; },
+      getStatus: () => ({ enabled: true }),
+    },
+  });
+  const result = await memory.observePersonalMemory({
+    userId: 'u1',
+    sessionId: 's1',
+    messages: [{ role: 'user', text: '我决定先完成测试。' }],
+  });
+  assert.equal(result.accepted, 1);
+  assert.equal(observed.length, 1);
+  assert.equal(legacyWrites, 0);
+});

memory-v2-personal-store.mjs

@@ -0,0 +1,127 @@
+const TABLE = 'h5_memory_v2_candidates';
+const ALLOWED_STATUSES = new Set(['candidate', 'accepted', 'rejected', 'forgotten']);
+
+export function buildPersonalMemoryCandidateSchemaSql({ table = TABLE } = {}) {
+  if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(table)) throw new Error('Invalid candidate table name');
+  return `CREATE TABLE IF NOT EXISTS \`${table}\` (
+    id VARCHAR(64) PRIMARY KEY,
+    user_id CHAR(36) NOT NULL,
+    session_id VARCHAR(191) NULL,
+    memory_type VARCHAR(32) NOT NULL,
+    content TEXT NOT NULL,
+    importance DECIMAL(5,4) NOT NULL,
+    confidence DECIMAL(5,4) NOT NULL,
+    status VARCHAR(24) NOT NULL DEFAULT 'candidate',
+    policy_reason VARCHAR(64) NOT NULL,
+    evidence_json JSON NOT NULL,
+    reviewed_by CHAR(36) NULL,
+    reviewed_at BIGINT NULL,
+    created_at BIGINT NOT NULL,
+    updated_at BIGINT NOT NULL,
+    UNIQUE KEY uniq_user_type_content (user_id, memory_type, id),
+    KEY idx_candidate_status_created (status, created_at),
+    KEY idx_candidate_user_status (user_id, status, created_at)
+  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`;
+}
+
+export async function ensurePersonalMemoryCandidateSchema(pool, options = {}) {
+  if (!pool?.query) throw new Error('Candidate schema requires a MySQL pool');
+  await pool.query(buildPersonalMemoryCandidateSchemaSql(options));
+}
+
+function parseJson(value, fallback = null) {
+  if (value == null) return fallback;
+  if (typeof value === 'object') return value;
+  try { return JSON.parse(value); } catch { return fallback; }
+}
+
+function normalizeRow(row) {
+  return {
+    id: String(row.id),
+    userId: String(row.user_id),
+    sessionId: row.session_id == null ? null : String(row.session_id),
+    memoryType: String(row.memory_type),
+    content: String(row.content),
+    importance: Number(row.importance),
+    confidence: Number(row.confidence),
+    status: String(row.status),
+    policyReason: String(row.policy_reason),
+    evidence: parseJson(row.evidence_json, {}),
+    reviewedBy: row.reviewed_by == null ? null : String(row.reviewed_by),
+    reviewedAt: row.reviewed_at == null ? null : Number(row.reviewed_at),
+    createdAt: Number(row.created_at),
+    updatedAt: Number(row.updated_at),
+  };
+}
+
+export function createPersonalMemoryCandidateStore(pool, { table = TABLE, now = () => Date.now() } = {}) {
+  if (!pool?.query) return null;
+  if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(table)) throw new Error('Invalid candidate table name');
+  const quoted = `\`${table}\``;
+
+  return {
+    async saveCandidate(candidate) {
+      const updatedAt = now();
+      const [result] = await pool.query(
+        `INSERT IGNORE INTO ${quoted}
+          (id, user_id, session_id, memory_type, content, importance, confidence, status,
+           policy_reason, evidence_json, created_at, updated_at)
+         VALUES (?, ?, ?, ?, ?, ?, ?, 'candidate', ?, ?, ?, ?)`,
+        [
+          candidate.id,
+          candidate.userId,
+          candidate.sessionId,
+          candidate.memoryType,
+          candidate.content,
+          candidate.importance,
+          candidate.confidence,
+          candidate.policyReason,
+          JSON.stringify(candidate.evidence ?? {}),
+          candidate.createdAt,
+          updatedAt,
+        ],
+      );
+      return { inserted: Number(result?.affectedRows ?? 0) > 0 };
+    },
+
+    async listCandidates({ status = 'candidate', userId = null, limit = 50, offset = 0 } = {}) {
+      const normalizedStatus = String(status || 'candidate');
+      if (!ALLOWED_STATUSES.has(normalizedStatus)) throw new Error('Invalid candidate status');
+      const safeLimit = Math.max(1, Math.min(200, Number(limit) || 50));
+      const safeOffset = Math.max(0, Number(offset) || 0);
+      const where = ['status = ?'];
+      const params = [normalizedStatus];
+      if (userId) {
+        where.push('user_id = ?');
+        params.push(String(userId));
+      }
+      params.push(safeLimit, safeOffset);
+      const [rows] = await pool.query(
+        `SELECT * FROM ${quoted} WHERE ${where.join(' AND ')}
+         ORDER BY created_at DESC LIMIT ? OFFSET ?`,
+        params,
+      );
+      return rows.map(normalizeRow);
+    },
+
+    async countByStatus() {
+      const [rows] = await pool.query(
+        `SELECT status, COUNT(*) AS count FROM ${quoted} GROUP BY status`,
+      );
+      return Object.fromEntries(rows.map((row) => [String(row.status), Number(row.count)]));
+    },
+
+    async reviewCandidate(id, status, { reviewedBy = null } = {}) {
+      if (!['accepted', 'rejected'].includes(status)) throw new Error('Invalid review status');
+      const reviewedAt = now();
+      const [result] = await pool.query(
+        `UPDATE ${quoted}
+            SET status = ?, reviewed_by = ?, reviewed_at = ?, updated_at = ?
+          WHERE id = ? AND status = 'candidate'`,
+        [status, reviewedBy, reviewedAt, reviewedAt, String(id)],
+      );
+      return { updated: Number(result?.affectedRows ?? 0) > 0, status, reviewedAt };
+    },
+  };
+}
+

memory-v2-personal-store.test.mjs

@@ -0,0 +1,47 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import {
+  buildPersonalMemoryCandidateSchemaSql,
+  createPersonalMemoryCandidateStore,
+  ensurePersonalMemoryCandidateSchema,
+} from './memory-v2-personal-store.mjs';
+
+test('candidate schema is explicit and idempotent', async () => {
+  const calls = [];
+  const pool = { async query(sql, params) { calls.push({ sql, params }); return [[], []]; } };
+  await ensurePersonalMemoryCandidateSchema(pool);
+  assert.match(calls[0].sql, /CREATE TABLE IF NOT EXISTS/);
+  assert.match(calls[0].sql, /h5_memory_v2_candidates/);
+  assert.throws(() => buildPersonalMemoryCandidateSchemaSql({ table: 'bad-name' }), /Invalid/);
+});
+
+test('candidate store saves, lists, counts, and reviews with parameterized SQL', async () => {
+  const calls = [];
+  const pool = {
+    async query(sql, params = []) {
+      calls.push({ sql, params });
+      if (sql.startsWith('INSERT')) return [{ affectedRows: 1 }, []];
+      if (sql.startsWith('SELECT *')) return [[{
+        id: 'pmc_1', user_id: 'u1', session_id: 's1', memory_type: 'episodic',
+        content: '决定使用 main 发布', importance: '0.9', confidence: '0.9', status: 'candidate',
+        policy_reason: 'decision_signal', evidence_json: '{"sourceId":"s1:0"}',
+        reviewed_by: null, reviewed_at: null, created_at: 1, updated_at: 1,
+      }], []];
+      if (sql.startsWith('SELECT status')) return [[{ status: 'candidate', count: 1 }], []];
+      if (sql.startsWith('UPDATE')) return [{ affectedRows: 1 }, []];
+      throw new Error(`Unexpected SQL: ${sql}`);
+    },
+  };
+  const store = createPersonalMemoryCandidateStore(pool, { now: () => 10 });
+  assert.deepEqual(await store.saveCandidate({
+    id: 'pmc_1', userId: 'u1', sessionId: 's1', memoryType: 'episodic',
+    content: '决定使用 main 发布', importance: 0.9, confidence: 0.9,
+    policyReason: 'decision_signal', evidence: { sourceId: 's1:0' }, createdAt: 1,
+  }), { inserted: true });
+  const rows = await store.listCandidates({ limit: 20 });
+  assert.equal(rows[0].evidence.sourceId, 's1:0');
+  assert.deepEqual(await store.countByStatus(), { candidate: 1 });
+  assert.equal((await store.reviewCandidate('pmc_1', 'accepted', { reviewedBy: 'admin-1' })).updated, true);
+  assert.ok(calls.every((call) => !call.sql.includes('admin-1')));
+});
+

memory-v2-runtime.mjs

@@ -4,6 +4,8 @@ import { createLangGraphHttpClient, createLangGraphMemoryBackend } from './memor
 import { createLegacyMemoryBackend, createMemoryV2, resolveMemoryV2Policy } from './memory-v2.mjs';
 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 { createMem0HttpClient, createMem0MemoryBackend } from './memory-v2-mem0.mjs';
 import { createNeo4jHttpClient, createNeo4jMemoryBackend } from './memory-v2-neo4j.mjs';
 import { createPgvectorMemoryBackend } from './memory-v2-pgvector.mjs';
@@ -400,11 +402,19 @@ export async function createMemoryV2Runtime({
     ],
   }));
 
+  const personalCandidateStore = readFlag(env, 'MEMORY_CANDIDATE_PERSISTENCE_ENABLED', false)
+    ? createPersonalMemoryCandidateStore(mysqlPool)
+    : null;
+  const personalShadowPipeline = createPersonalMemoryShadowPipeline({
+    env,
+    store: personalCandidateStore,
+  });
   const memory = createMemoryV2({
     legacyMemoryService,
     backends,
     env,
     logger,
+    personalShadowPipeline,
   });
 
   const pgBackfillEnabled = readFlag(env, 'MEMORY_PGVECTOR_BACKFILL_ENABLED', pgvectorEnabled);
@@ -593,6 +603,11 @@ export async function createManagedMemoryV2Runtime({
       return runtime.compact(input);
     },
 
+    async observePersonalMemory(input = {}) {
+      const runtime = await ensureRuntime();
+      return runtime.observePersonalMemory(input);
+    },
+
     async close() {
       const runtime = activeRuntime;
       activeRuntime = null;

memory-v2.mjs

@@ -1,4 +1,5 @@
 import { createMemoryV2PluginBackends } from './memory-v2-plugin-backends.mjs';
+import { createPersonalMemoryShadowPipeline } from './memory-v2-personal-shadow.mjs';
 
 const TRUE_VALUES = new Set(['1', 'true', 'yes', 'on']);
 const FALSE_VALUES = new Set(['0', 'false', 'no', 'off']);
@@ -231,12 +232,14 @@ export function createMemoryV2({
   policy = null,
   env = process.env,
   logger = console,
+  personalShadowPipeline = null,
 } = {}) {
   const resolvedPolicy = policy ?? resolveMemoryV2Policy({ env });
   const resolvedBackends = backends ?? [
     createLegacyMemoryBackend(legacyMemoryService),
     ...createMemoryV2PluginBackends(),
   ];
+  const shadowPipeline = personalShadowPipeline ?? createPersonalMemoryShadowPipeline({ env });
 
   async function failOpen(operation, err, fallback) {
     logger?.warn?.(
@@ -278,6 +281,13 @@ export function createMemoryV2({
     if (!backend?.write) return skippedWriteResult('no_backend');
     try {
       const result = await backend.write(input);
+      if (shadowPipeline?.config?.enabled) {
+        void Promise.resolve(shadowPipeline.observeWrite(input)).catch((err) => {
+          logger?.warn?.(
+            `[memory-v2] personal shadow pipeline skipped: ${err instanceof Error ? err.message : err}`,
+          );
+        });
+      }
       return {
         ok: true,
         enabled: true,
@@ -334,10 +344,28 @@ export function createMemoryV2({
     }
   }
 
+  async function observePersonalMemory(input = {}) {
+    if (!shadowPipeline?.config?.enabled) {
+      return { enabled: false, skipped: true, reason: 'disabled' };
+    }
+    try {
+      return await shadowPipeline.observeWrite(input);
+    } catch (err) {
+      logger?.warn?.(
+        `[memory-v2] personal shadow observation skipped: ${err instanceof Error ? err.message : err}`,
+      );
+      return {
+        enabled: true,
+        skipped: true,
+        reason: 'pipeline_failure',
+      };
+    }
+  }
+
   function getStatus() {
     const backends = resolvedBackends.map((backend) => backendStatus(backend));
     const selected = selectBackend(resolvedBackends, resolvedPolicy, 'resolve');
-    return {
+    const status = {
       enabled: Boolean(resolvedPolicy.enabled),
       backend: resolvedPolicy.backend,
       selectedBackend: selected?.name ?? null,
@@ -347,6 +375,10 @@ export function createMemoryV2({
       failOpen: Boolean(resolvedPolicy.failOpen),
       backends,
     };
+    if (shadowPipeline?.config?.enabled) {
+      status.personalMemory = shadowPipeline.getStatus();
+    }
+    return status;
   }
 
   return {
@@ -355,5 +387,6 @@ export function createMemoryV2({
     resolve,
     write,
     compact,
+    observePersonalMemory,
   };
 }

scripts/setup-memory-v2-personal-schema.mjs

@@ -0,0 +1,15 @@
+import { createDbPool } from '../db.mjs';
+import { ensurePersonalMemoryCandidateSchema } from '../memory-v2-personal-store.mjs';
+
+if (process.env.MEMORY_PERSONAL_SCHEMA_CONFIRM !== 'local-only') {
+  throw new Error('Refusing schema change: set MEMORY_PERSONAL_SCHEMA_CONFIRM=local-only explicitly');
+}
+
+const pool = createDbPool();
+try {
+  await ensurePersonalMemoryCandidateSchema(pool);
+  console.log('Memory V2 personal candidate schema ready.');
+} finally {
+  await pool.end();
+}
+