diff --git a/memory-v2-admin-config.mjs b/memory-v2-admin-config.mjs index e38c81a..c8c9068 100644 --- a/memory-v2-admin-config.mjs +++ b/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', diff --git a/memory-v2-admin-config.test.mjs b/memory-v2-admin-config.test.mjs index e2c3c19..5c639c1 100644 --- a/memory-v2-admin-config.test.mjs +++ b/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', () => { diff --git a/memory-v2-personal-shadow.mjs b/memory-v2-personal-shadow.mjs new file mode 100644 index 0000000..c389f2e --- /dev/null +++ b/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 }; +} diff --git a/memory-v2-personal-shadow.test.mjs b/memory-v2-personal-shadow.test.mjs new file mode 100644 index 0000000..be272be --- /dev/null +++ b/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); +}); diff --git a/memory-v2-personal-store.mjs b/memory-v2-personal-store.mjs new file mode 100644 index 0000000..402b971 --- /dev/null +++ b/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 }; + }, + }; +} + diff --git a/memory-v2-personal-store.test.mjs b/memory-v2-personal-store.test.mjs new file mode 100644 index 0000000..60f1ae1 --- /dev/null +++ b/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'))); +}); + diff --git a/memory-v2-runtime.mjs b/memory-v2-runtime.mjs index 3e877b2..a28e2f5 100644 --- a/memory-v2-runtime.mjs +++ b/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; diff --git a/memory-v2.mjs b/memory-v2.mjs index afc8b7c..f70d680 100644 --- a/memory-v2.mjs +++ b/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, }; } diff --git a/scripts/setup-memory-v2-personal-schema.mjs b/scripts/setup-memory-v2-personal-schema.mjs new file mode 100644 index 0000000..6c09701 --- /dev/null +++ b/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(); +} +