diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index 9274807..7ceea2c 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -565,6 +565,40 @@ export function createAgentRunGateway({ const toolGatewayStatus = toolGateway?.getStatus ? toolGateway.getStatus() : null; const routing = await resolveRunRouting(row, userMessage, runOptions); const routingDecision = resolveLegacyRouteFromClassification(routing) ?? routing?.route ?? null; + let agentMemoryContext = null; + if (routingDecision === CHAT_INTENT_ROUTE.AGENT && chatIntentRouter?.resolveAgentMemoryContext) { + const displayText = userMessage?.metadata?.displayText + ?? userMessage?.content?.find?.((item) => item?.type === 'text')?.text + ?? ''; + try { + agentMemoryContext = await chatIntentRouter.resolveAgentMemoryContext({ + userId: row.user_id, + sessionId: row.agent_session_id ?? null, + text: displayText, + forceDeepReasoning: runOptions.forceDeepReasoning, + }); + if (agentMemoryContext?.enabled && agentMemoryContext.mode !== 'off') { + await appendEvent(runId, 'agent_memory_resolved', { + mode: agentMemoryContext.mode, + injectionEnabled: Boolean(agentMemoryContext.injectionEnabled), + source: agentMemoryContext.source ?? null, + memoryCount: Array.isArray(agentMemoryContext.memories) + ? agentMemoryContext.memories.length + : 0, + skipped: Boolean(agentMemoryContext.skipped), + degraded: Boolean(agentMemoryContext.degraded), + reason: agentMemoryContext.reason ?? null, + latencyMs: Number(agentMemoryContext.latencyMs ?? 0), + }); + } + } catch (err) { + console.warn( + '[AgentRun] agent memory shadow resolve skipped:', + err instanceof Error ? err.message : err, + ); + agentMemoryContext = null; + } + } if (routing) { logRouterDecisionShadow(routing, { requestId: row.request_id ?? null, @@ -573,7 +607,10 @@ export function createAgentRunGateway({ await appendEvent(runId, 'intent_routed', routing); if (routingDecision === CHAT_INTENT_ROUTE.AGENT && chatIntentRouter?.applyAgentOrchestration) { const grantedSkills = await resolveGrantedSkills(row.user_id); - userMessage = chatIntentRouter.applyAgentOrchestration(userMessage, routing, { grantedSkills }); + userMessage = chatIntentRouter.applyAgentOrchestration(userMessage, routing, { + grantedSkills, + memoryContext: agentMemoryContext, + }); } } const preferDirectChat = diff --git a/agent-run-gateway.test.mjs b/agent-run-gateway.test.mjs index faa1300..2df0068 100644 --- a/agent-run-gateway.test.mjs +++ b/agent-run-gateway.test.mjs @@ -1001,8 +1001,21 @@ test('agent run uses chat intent router to enrich agent orchestration messages', source: 'llm', }; }, - applyAgentOrchestration(userMessage, classification, { grantedSkills = [] }) { + async resolveAgentMemoryContext() { + return { + enabled: true, + mode: 'shadow', + injectionEnabled: false, + skipped: false, + memories: [{ label: 'preference', text: '用户喜欢完整方案' }], + source: 'legacy-conversation-memory', + latencyMs: 1, + }; + }, + applyAgentOrchestration(userMessage, classification, { grantedSkills = [], memoryContext = null }) { const displayText = userMessage?.metadata?.displayText ?? userMessage?.content?.[0]?.text ?? ''; + assert.equal(memoryContext?.mode, 'shadow'); + assert.equal(memoryContext?.injectionEnabled, false); return { ...userMessage, content: [{ @@ -1033,6 +1046,7 @@ test('agent run uses chat intent router to enrich agent orchestration messages', assert.match(submitted[0].userMessage.content[0].text, /Memind 任务编排/); assert.match(submitted[0].userMessage.content[0].text, /帮我做一个页面/); assert.ok(pool.events.some((event) => event.eventType === 'intent_routed')); + assert.ok(pool.events.some((event) => event.eventType === 'agent_memory_resolved')); }); test('agent run escalates direct sessions to a new backend session when forced', async () => { diff --git a/chat-intent-router.mjs b/chat-intent-router.mjs index e9acdff..bbf2d8a 100644 --- a/chat-intent-router.mjs +++ b/chat-intent-router.mjs @@ -650,6 +650,18 @@ export function resolveChatIntentRouterPolicy({ env = process.env, overrides = { }; } +function normalizeAgentInjectionMode(value) { + const mode = String(value ?? 'off').trim().toLowerCase(); + return ['off', 'shadow', 'canary', 'active'].includes(mode) ? mode : 'off'; +} + +function normalizeAgentCanaryUserIds(value) { + return [...new Set(String(value ?? '') + .split(/[\s,]+/u) + .map((item) => item.trim()) + .filter(Boolean))].slice(0, 1000); +} + function normalizeClassification(raw, { source, fallbackRoute = CHAT_INTENT_ROUTE.AGENT } = {}) { const route = normalizeRoute(raw?.route) ?? fallbackRoute; const confidenceRaw = Number(raw?.confidence); @@ -688,14 +700,26 @@ export function buildAgentOrchestrationAgentText({ displayText, classification, skillPrompt = '', + memoryContext = null, }) { const taskBody = String(displayText ?? '').trim(); + const memoryLines = memoryContext?.injectionEnabled + ? (Array.isArray(memoryContext.memories) ? memoryContext.memories : []) + .map((item) => normalizeMemoryText(item)) + .map(({ label, text }) => { + const clipped = truncateText(text, 120); + return clipped ? `- ${label ? `[${label}] ` : ''}${clipped}` : ''; + }) + .filter(Boolean) + .slice(0, 8) + : []; const lines = [ `${AGENT_ORCHESTRATION_HEADER}以下为用户任务，请使用工具与技能实际执行并产出结果，不要只做文字描述。`, `路由判定：${classification.reason}`, classification.agentBrief ? `执行要点：${classification.agentBrief}` : '', classification.suggestedSkill ? `建议 skill：${classification.suggestedSkill}` : '', skillPrompt, + memoryLines.length ? ['[Memory Context]', ...memoryLines].join('\n') : '', '', '用户任务：', taskBody, @@ -703,13 +727,18 @@ export function buildAgentOrchestrationAgentText({ return lines.join('\n'); } -export function applyAgentOrchestrationToUserMessage(userMessage, classification, { grantedSkills = [] } = {}) { +export function applyAgentOrchestrationToUserMessage( + userMessage, + classification, + { grantedSkills = [], memoryContext = null } = {}, +) { const displayText = messageDisplayText(userMessage); const skillPrompt = resolveSkillPrompt(classification?.suggestedSkill, grantedSkills, displayText); const agentText = buildAgentOrchestrationAgentText({ displayText, classification, skillPrompt, + memoryContext, }); const content = Array.isArray(userMessage?.content) ? userMessage.content.map((item, index) => { @@ -885,6 +914,22 @@ export function createChatIntentRouter(options = {}) { 'fallbackRoute', ]), }); + const agentMemoryPolicy = { + enabled: envFlag(env?.MEMORY_AGENT_RESOLVE_ENABLED, false), + mode: normalizeAgentInjectionMode(env?.MEMORY_AGENT_INJECTION_MODE), + canaryUserIds: normalizeAgentCanaryUserIds(env?.MEMORY_AGENT_CANARY_USER_IDS), + limit: Math.round(boundedNumber(env?.MEMORY_AGENT_RESOLVE_LIMIT, 3, { min: 1, max: 50 })), + timeoutMs: Math.round(boundedNumber(env?.MEMORY_AGENT_RESOLVE_TIMEOUT_MS, 1200, { min: 0, max: 30_000 })), + }; + const agentMemoryMetrics = { + resolveStarted: 0, + resolved: 0, + skipped: 0, + degraded: 0, + injected: 0, + lastLatencyMs: null, + lastReason: null, + }; function getStatus() { return { @@ -902,6 +947,11 @@ export function createChatIntentRouter(options = {}) { fallbackRoute: policy.fallbackRoute, normalizedDecisionEnabled: isNormalizedRouterDecisionEnabled(env), normalizedDecisionMode: resolveNormalizedRouterDecisionMode(env), + agentMemory: { + ...agentMemoryPolicy, + injectionEnabled: agentMemoryPolicy.mode === 'active', + metrics: { ...agentMemoryMetrics }, + }, }; } @@ -966,6 +1016,76 @@ export function createChatIntentRouter(options = {}) { return buildRouterContext({ memories, source: 'router-resolve' }); } + async function resolveAgentMemoryContext({ userId, sessionId, text, forceDeepReasoning = false } = {}) { + const injectionEnabled = agentMemoryPolicy.mode === 'active' + || (agentMemoryPolicy.mode === 'canary' && agentMemoryPolicy.canaryUserIds.includes(String(userId ?? '').trim())); + const base = { + enabled: agentMemoryPolicy.enabled, + mode: agentMemoryPolicy.mode, + injectionEnabled, + skipped: true, + degraded: false, + reason: null, + source: null, + memories: [], + latencyMs: 0, + }; + agentMemoryMetrics.resolveStarted += 1; + if (!agentMemoryPolicy.enabled || agentMemoryPolicy.mode === 'off' || !userId || !memoryV2?.resolve) { + agentMemoryMetrics.skipped += 1; + agentMemoryMetrics.lastReason = 'disabled'; + return base; + } + const intervention = resolveMemoryInterventionMode({ + forceDeepReasoning, + recallQuestion: isMemoryRecallQuestion(text), + context: 'agent', + }); + const limit = Math.min( + agentMemoryPolicy.limit, + memoryLimitForIntervention(intervention, { context: 'agent' }), + ); + if (limit <= 0) { + agentMemoryMetrics.skipped += 1; + agentMemoryMetrics.lastReason = 'intervention_skip'; + return { ...base, reason: 'intervention_skip' }; + } + const startedAt = Date.now(); + try { + const resolved = await withTimeout( + memoryV2.resolve({ userId, sessionId, query: text, limit }), + agentMemoryPolicy.timeoutMs, + 'Memory V2 agent resolve', + ); + const result = { + ...base, + skipped: false, + source: resolved?.source ?? null, + degraded: Boolean(resolved?.degraded), + reason: resolved?.reason ?? null, + memories: Array.isArray(resolved?.memories) ? resolved.memories.slice(0, limit) : [], + latencyMs: Date.now() - startedAt, + }; + agentMemoryMetrics.resolved += 1; + agentMemoryMetrics.lastLatencyMs = result.latencyMs; + agentMemoryMetrics.lastReason = result.reason; + if (result.degraded) agentMemoryMetrics.degraded += 1; + if (result.injectionEnabled && result.memories.length) agentMemoryMetrics.injected += 1; + return result; + } catch (err) { + const result = { + ...base, + degraded: true, + reason: err?.code === 'CHAT_INTENT_ROUTER_TIMEOUT' ? 'timeout' : 'resolve_failed', + latencyMs: Date.now() - startedAt, + }; + agentMemoryMetrics.degraded += 1; + agentMemoryMetrics.lastLatencyMs = result.latencyMs; + agentMemoryMetrics.lastReason = result.reason; + return result; + } + } + async function classify({ userId = null, userMessage, @@ -1022,6 +1142,7 @@ export function createChatIntentRouter(options = {}) { getStatus, isEnabled, classify, + resolveAgentMemoryContext, applyAgentOrchestration: applyAgentOrchestrationToUserMessage, }; } @@ -1114,6 +1235,11 @@ export function createManagedChatIntentRouter({ return router.classify(input); }, + async resolveAgentMemoryContext(input = {}) { + const router = await ensureRouter(); + return router.resolveAgentMemoryContext(input); + }, + applyAgentOrchestration: applyAgentOrchestrationToUserMessage, }; } diff --git a/chat-intent-router.test.mjs b/chat-intent-router.test.mjs index 1cbdc30..9d5e38b 100644 --- a/chat-intent-router.test.mjs +++ b/chat-intent-router.test.mjs @@ -1078,6 +1078,83 @@ test('createChatIntentRouter timeout fallback prefers agent even when policy req assert.equal(result.source, 'fallback'); }); +test('agent memory shadow resolve is independently gated and never marked for injection', async () => { + const calls = []; + const router = createChatIntentRouter({ + env: { + MEMORY_AGENT_RESOLVE_ENABLED: '1', + MEMORY_AGENT_INJECTION_MODE: 'shadow', + MEMORY_AGENT_RESOLVE_LIMIT: '3', + MEMORY_AGENT_RESOLVE_TIMEOUT_MS: '100', + }, + memoryV2: { + async resolve(input) { + calls.push(input); + return { source: 'legacy-conversation-memory', memories: [{ label: 'preference', text: '喜欢完整方案' }] }; + }, + }, + }); + + const result = await router.resolveAgentMemoryContext({ + userId: 'u1', + sessionId: 's1', + text: '帮我设计一个商城', + }); + + assert.equal(result.mode, 'shadow'); + assert.equal(result.injectionEnabled, false); + assert.equal(result.skipped, false); + assert.equal(result.memories.length, 1); + assert.deepEqual(calls[0], { userId: 'u1', sessionId: 's1', query: '帮我设计一个商城', limit: 3 }); +}); + +test('agent memory canary injects only for configured user ids', async () => { + const router = createChatIntentRouter({ + env: { + MEMORY_AGENT_RESOLVE_ENABLED: '1', + MEMORY_AGENT_INJECTION_MODE: 'canary', + MEMORY_AGENT_CANARY_USER_IDS: 'user-canary, user-other', + }, + memoryV2: { + async resolve() { + return { memories: [{ label: 'goal', text: '当前项目是 TKMind' }] }; + }, + }, + }); + + const canary = await router.resolveAgentMemoryContext({ userId: 'user-canary', text: '继续项目' }); + const control = await router.resolveAgentMemoryContext({ userId: 'user-control', text: '继续项目' }); + assert.equal(canary.injectionEnabled, true); + assert.equal(control.injectionEnabled, false); + assert.equal(canary.memories.length, 1); + assert.equal(control.memories.length, 1); +}); + +test('active agent memory context is hidden from displayText but available to orchestration envelope', () => { + const enriched = applyAgentOrchestrationToUserMessage( + { + role: 'user', + content: [{ type: 'text', text: '帮我设计一个商城' }], + }, + { + route: CHAT_INTENT_ROUTE.AGENT, + reason: '任务执行', + agentBrief: '', + suggestedSkill: null, + }, + { + memoryContext: { + injectionEnabled: true, + memories: [{ label: 'preference', text: '用户喜欢完整方案' }], + }, + }, + ); + + assert.equal(enriched.metadata.displayText, '帮我设计一个商城'); + assert.match(enriched.content[0].text, /Memory Context/); + assert.match(enriched.content[0].text, /用户喜欢完整方案/); +}); + test('logRouterDecisionShadow emits payload only in shadow mode', () => { const previous = process.env.MEMIND_ROUTER_NORMALIZED_DECISION; const lines = []; diff --git a/docs/memory-v2/README.md b/docs/memory-v2/README.md index 977b3a1..fd7ae27 100644 --- a/docs/memory-v2/README.md +++ b/docs/memory-v2/README.md @@ -46,6 +46,41 @@ The first implementation is intentionally narrow: - Runtime startup does not create schema. - No goosed or SSE behavior is changed. +### Runtime control flags + +The `memind_adm` Memory V2 page exposes runtime controls separately from +`MEMORY_ENABLED`. They are disabled by default. Agent resolve currently +supports `shadow` observation and explicit `active` hidden-context injection; +promotion, Compact V2, and Reflection remain guarded for later stages: + +| Admin field | Environment override | Default | +| --- | --- | --- | +| `runtimeControl.agentResolveEnabled` | `MEMORY_AGENT_RESOLVE_ENABLED` | `0` | +| `runtimeControl.agentInjectionMode` | `MEMORY_AGENT_INJECTION_MODE` | `off` | +| `runtimeControl.agentCanaryUserIds` | `MEMORY_AGENT_CANARY_USER_IDS` | empty | +| `runtimeControl.agentResolveLimit` | `MEMORY_AGENT_RESOLVE_LIMIT` | `3` | +| `runtimeControl.agentResolveTimeoutMs` | `MEMORY_AGENT_RESOLVE_TIMEOUT_MS` | `1200` | +| `runtimeControl.promotionEnabled` | `MEMORY_PROMOTION_ENABLED` | `0` | +| `runtimeControl.compactionV2Enabled` | `MEMORY_COMPACTION_V2_ENABLED` | `0` | +| `runtimeControl.reflectionEnabled` | `MEMORY_REFLECTION_ENABLED` | `0` | +| `runtimeControl.lifecycleWorkerEnabled` | `MEMORY_LIFECYCLE_WORKER_ENABLED` | `0` | +| `runtimeControl.lifecycleRolloutMode` | `MEMORY_LIFECYCLE_ROLLOUT_MODE` | `off` | +| `runtimeControl.lifecycleRolloutUserIds` | `MEMORY_LIFECYCLE_ROLLOUT_USER_IDS` | empty | + +The controls are reported in `memoryV2.getStatus().runtimeControl`. Turning +them off must leave the existing legacy conversation-memory path unchanged. + +The additive user-scoped management API is: + +- `GET /user-memory/v1/items` to list active (or requested-status) items. +- `DELETE /user-memory/v1/items/:memoryId` to forget one item when lifecycle + forgetting is enabled. + +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. + 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. The server runtime uses `createMemoryV2Runtime(...)`. It keeps pgvector dormant unless all of these are true: diff --git a/memory-v2-admin-config.mjs b/memory-v2-admin-config.mjs index c8c9068..07fa03c 100644 --- a/memory-v2-admin-config.mjs +++ b/memory-v2-admin-config.mjs @@ -29,6 +29,18 @@ const FIELD_SPECS = [ { env: 'MEMORY_CANDIDATE_MAX_PENDING', group: 'candidateMemory', field: 'maxPending', type: 'number' }, { env: 'MEMORY_CANDIDATE_PERSISTENCE_ENABLED', group: 'candidateMemory', field: 'persistenceEnabled', type: 'boolean' }, + { env: 'MEMORY_AGENT_RESOLVE_ENABLED', group: 'runtimeControl', field: 'agentResolveEnabled', type: 'boolean' }, + { env: 'MEMORY_AGENT_INJECTION_MODE', group: 'runtimeControl', field: 'agentInjectionMode', type: 'string' }, + { env: 'MEMORY_AGENT_CANARY_USER_IDS', group: 'runtimeControl', field: 'agentCanaryUserIds', type: 'string' }, + { env: 'MEMORY_AGENT_RESOLVE_LIMIT', group: 'runtimeControl', field: 'agentResolveLimit', type: 'number' }, + { env: 'MEMORY_AGENT_RESOLVE_TIMEOUT_MS', group: 'runtimeControl', field: 'agentResolveTimeoutMs', type: 'number' }, + { env: 'MEMORY_PROMOTION_ENABLED', group: 'runtimeControl', field: 'promotionEnabled', type: 'boolean' }, + { env: 'MEMORY_COMPACTION_V2_ENABLED', group: 'runtimeControl', field: 'compactionV2Enabled', type: 'boolean' }, + { env: 'MEMORY_REFLECTION_ENABLED', group: 'runtimeControl', field: 'reflectionEnabled', type: 'boolean' }, + { env: 'MEMORY_LIFECYCLE_WORKER_ENABLED', group: 'runtimeControl', field: 'lifecycleWorkerEnabled', type: 'boolean' }, + { env: 'MEMORY_LIFECYCLE_ROLLOUT_MODE', group: 'runtimeControl', field: 'lifecycleRolloutMode', type: 'string' }, + { env: 'MEMORY_LIFECYCLE_ROLLOUT_USER_IDS', group: 'runtimeControl', field: 'lifecycleRolloutUserIds', type: 'string' }, + { 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' }, @@ -149,6 +161,7 @@ const GROUPS = [ 'global', 'chatIntentRouter', 'candidateMemory', + 'runtimeControl', 'policy', 'retriever', 'lifecycle', diff --git a/memory-v2-admin-config.test.mjs b/memory-v2-admin-config.test.mjs index 5c639c1..5be0c0f 100644 --- a/memory-v2-admin-config.test.mjs +++ b/memory-v2-admin-config.test.mjs @@ -89,6 +89,16 @@ test('memory v2 admin config service persists non-secret and secret patches', as maxPending: '500', persistenceEnabled: true, }, + runtimeControl: { + agentResolveEnabled: true, + agentInjectionMode: 'shadow', + agentCanaryUserIds: 'user-1,user-2', + agentResolveLimit: '3', + agentResolveTimeoutMs: '1200', + promotionEnabled: false, + compactionV2Enabled: false, + reflectionEnabled: false, + }, policy: { enabled: true, saveExplicit: true, @@ -123,6 +133,10 @@ test('memory v2 admin config service persists non-secret and secret patches', as 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.runtimeControl.agentResolveEnabled, true); + assert.equal(updated.config.runtimeControl.agentInjectionMode, 'shadow'); + assert.equal(updated.config.runtimeControl.agentCanaryUserIds, 'user-1,user-2'); + assert.equal(updated.config.runtimeControl.agentResolveLimit, '3'); assert.equal(updated.config.policy.requireEvidence, true); assert.equal(updated.config.retriever.tokenBudget, '1800'); assert.equal(updated.config.lifecycle.dedupeEnabled, true); @@ -149,6 +163,12 @@ test('memory v2 admin config service persists non-secret and secret patches', as 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_AGENT_RESOLVE_ENABLED, '1'); + assert.equal(runtimeState.overrides.MEMORY_AGENT_INJECTION_MODE, 'shadow'); + assert.equal(runtimeState.overrides.MEMORY_AGENT_CANARY_USER_IDS, 'user-1,user-2'); + assert.equal(runtimeState.overrides.MEMORY_AGENT_RESOLVE_LIMIT, '3'); + assert.equal(runtimeState.overrides.MEMORY_AGENT_RESOLVE_TIMEOUT_MS, '1200'); + assert.equal(runtimeState.overrides.MEMORY_PROMOTION_ENABLED, '0'); 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'); diff --git a/memory-v2-lifecycle.mjs b/memory-v2-lifecycle.mjs new file mode 100644 index 0000000..1d37d77 --- /dev/null +++ b/memory-v2-lifecycle.mjs @@ -0,0 +1,150 @@ +import crypto from 'node:crypto'; + +const MEMORY_TABLE = 'h5_user_memory_items'; +const CANDIDATE_TABLE = 'h5_memory_v2_candidates'; + +function bounded(value, fallback, min, max) { + const parsed = Number(value); + if (!Number.isFinite(parsed)) return fallback; + return Math.min(max, Math.max(min, parsed)); +} + +function flag(value, fallback = false) { + if (value == null || value === '') return fallback; + return ['1', 'true', 'yes', 'on'].includes(String(value).trim().toLowerCase()) + ? true + : ['0', 'false', 'no', 'off'].includes(String(value).trim().toLowerCase()) ? false : fallback; +} + +export function resolveMemoryV2LifecyclePolicy(env = process.env) { + return { + enabled: flag(env.MEMORY_LIFECYCLE_ENABLED, false), + dedupeEnabled: flag(env.MEMORY_LIFECYCLE_DEDUPE_ENABLED, true), + decayEnabled: flag(env.MEMORY_LIFECYCLE_DECAY_ENABLED, false), + forgettingEnabled: flag(env.MEMORY_LIFECYCLE_FORGETTING_ENABLED, false), + retentionDays: Math.round(bounded(env.MEMORY_POLICY_RETENTION_DAYS, 365, 1, 3650)), + compactIntervalHours: Math.round(bounded(env.MEMORY_LIFECYCLE_COMPACT_INTERVAL_HOURS, 24, 1, 720)), + promotionEnabled: flag(env.MEMORY_PROMOTION_ENABLED, false), + compactionEnabled: flag(env.MEMORY_COMPACTION_V2_ENABLED, false), + reflectionEnabled: flag(env.MEMORY_REFLECTION_ENABLED, false), + rolloutMode: String(env.MEMORY_LIFECYCLE_ROLLOUT_MODE ?? 'off').trim().toLowerCase() || 'off', + rolloutUserIds: String(env.MEMORY_LIFECYCLE_ROLLOUT_USER_IDS ?? '') + .split(/[\s,]+/u).map((item) => item.trim()).filter(Boolean).slice(0, 1000), + }; +} + +function normalizeItem(row) { + return { + id: String(row.id), + userId: String(row.user_id), + type: String(row.label ?? 'fact'), + content: String(row.memory_text ?? ''), + confidence: Number(row.confidence ?? 0), + status: String(row.status ?? 'active'), + sourceSessionId: row.source_session_id == null ? null : String(row.source_session_id), + evidenceMessageId: row.evidence_message_id == null ? null : String(row.evidence_message_id), + createdAt: Number(row.created_at ?? 0), + updatedAt: Number(row.updated_at ?? 0), + }; +} + +export function createMemoryV2LifecycleService({ pool = null, env = process.env, now = () => Date.now(), logger = console } = {}) { + const policy = resolveMemoryV2LifecyclePolicy(env); + const metrics = { list: 0, forget: 0, expire: 0, compact: 0, promote: 0, reflect: 0, errors: 0, lastRunAt: null, lastError: null }; + const canRun = (userId = null) => { + if (!policy.enabled) return false; + if (policy.rolloutMode === 'active') return true; + if (policy.rolloutMode === 'canary') return Boolean(userId && policy.rolloutUserIds.includes(String(userId))); + return false; + }; + + async function listMemories({ userId, status = 'active', limit = 100, offset = 0 } = {}) { + if (!pool?.query || !userId) return []; + metrics.list += 1; + const safeLimit = Math.max(1, Math.min(200, Number(limit) || 100)); + const safeOffset = Math.max(0, Number(offset) || 0); + const [rows] = await pool.query( + `SELECT * FROM ${MEMORY_TABLE} WHERE user_id = ? AND status = ? ORDER BY updated_at DESC LIMIT ? OFFSET ?`, + [String(userId), String(status), safeLimit, safeOffset], + ); + return rows.map(normalizeItem); + } + + async function forgetMemory({ userId, memoryId } = {}) { + if (!canRun(userId) && !policy.forgettingEnabled) 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'`, + [now(), String(memoryId), String(userId)], + ); + metrics.forget += 1; + return { ok: true, updated: Number(result?.affectedRows ?? 0) > 0 }; + } + + async function expire({ userId = null } = {}) { + if (!pool?.query || !policy.forgettingEnabled) return { ok: true, skipped: true, reason: 'disabled', expired: 0 }; + const cutoff = now() - policy.retentionDays * 86400000; + const params = [cutoff]; + let scope = ''; + if (userId) { scope = ' AND user_id = ?'; params.push(String(userId)); } + const [result] = await pool.query( + `UPDATE ${MEMORY_TABLE} SET status = 'archived', updated_at = ? WHERE updated_at < ? AND status = 'active'${scope}`, + [now(), cutoff, ...params.slice(1)], + ); + metrics.expire += 1; + return { ok: true, skipped: false, expired: Number(result?.affectedRows ?? 0), cutoff }; + } + + async function compact({ userId = null } = {}) { + if (!policy.compactionEnabled || !canRun(userId)) return { ok: true, skipped: true, reason: 'disabled', analyzed: 0, memories: 0 }; + metrics.compact += 1; metrics.lastRunAt = now(); + // Compaction is deliberately conservative: it reports eligible material and + // never overwrites source memories until a backend-specific compactor is enabled. + const items = userId ? await listMemories({ userId, limit: 200 }) : []; + return { ok: true, skipped: false, analyzed: items.length, memories: 0, mode: 'candidate-only' }; + } + + async function promote({ userId = null, limit = 50 } = {}) { + if (!pool?.query || !policy.promotionEnabled || !canRun(userId)) return { ok: true, skipped: true, reason: 'disabled', promoted: 0 }; + const params = []; + let scope = ''; + if (userId) { scope = ' AND user_id = ?'; params.push(String(userId)); } + params.push(Math.max(1, Math.min(200, Number(limit) || 50))); + const [rows] = await pool.query( + `SELECT * FROM ${CANDIDATE_TABLE} WHERE status = 'accepted'${scope} ORDER BY updated_at ASC LIMIT ?`, + params, + ); + let promoted = 0; + for (const row of rows) { + const id = crypto.createHash('sha256').update(`${row.user_id}\n${row.memory_type}\n${row.content}`).digest('hex'); + const hash = crypto.createHash('sha256').update(`${row.user_id}\n${row.content}`).digest('hex'); + const [result] = await pool.query( + `INSERT IGNORE INTO ${MEMORY_TABLE} + (id,user_id,label,memory_hash,memory_text,evidence_message_id,source_session_id,confidence,status,raw_json,created_at,updated_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?)`, + [id, row.user_id, row.memory_type === 'semantic' ? 'knowledge' : row.memory_type, hash, row.content, null, row.session_id, row.confidence, 'active', row.evidence_json, row.created_at, now()], + ); + if (Number(result?.affectedRows ?? 0) > 0) promoted += 1; + } + metrics.promote += 1; + return { ok: true, skipped: false, promoted }; + } + + async function reflect({ userId = null } = {}) { + if (!policy.reflectionEnabled || !canRun(userId)) return { ok: true, skipped: true, reason: 'disabled', updated: 0 }; + metrics.reflect += 1; metrics.lastRunAt = now(); + return { ok: true, skipped: false, updated: 0, mode: 'observation-only' }; + } + + return { + policy, + listMemories, + forgetMemory, + expire, + compact, + promote, + reflect, + canRun, + getStatus() { return { policy, metrics: { ...metrics } }; }, + }; +} diff --git a/memory-v2-lifecycle.test.mjs b/memory-v2-lifecycle.test.mjs new file mode 100644 index 0000000..8021548 --- /dev/null +++ b/memory-v2-lifecycle.test.mjs @@ -0,0 +1,47 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { createMemoryV2LifecycleService, resolveMemoryV2LifecyclePolicy } from './memory-v2-lifecycle.mjs'; + +test('lifecycle policy defaults to safe disabled workers', () => { + const policy = resolveMemoryV2LifecyclePolicy({}); + assert.equal(policy.enabled, false); + assert.equal(policy.promotionEnabled, false); + assert.equal(policy.compactionEnabled, false); + assert.equal(policy.reflectionEnabled, false); + assert.equal(policy.rolloutMode, 'off'); +}); + +test('lifecycle canary only runs for configured users', async () => { + const lifecycle = createMemoryV2LifecycleService({ + env: { + MEMORY_LIFECYCLE_ENABLED: '1', + MEMORY_LIFECYCLE_ROLLOUT_MODE: 'canary', + MEMORY_LIFECYCLE_ROLLOUT_USER_IDS: 'u-1', + MEMORY_PROMOTION_ENABLED: '1', + }, + }); + assert.equal(lifecycle.canRun('u-1'), true); + assert.equal(lifecycle.canRun('u-2'), false); + assert.equal((await lifecycle.promote({ userId: 'u-2' })).skipped, true); +}); + +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' }), { + ok: true, updated: false, skipped: true, reason: 'disabled', + }); +}); + +test('list and forget use user-scoped SQL', async () => { + const calls = []; + const pool = { query: async (sql, params) => { + calls.push({ sql, params }); + 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' } }); + 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 = \?/); +}); diff --git a/memory-v2-runtime.mjs b/memory-v2-runtime.mjs index a28e2f5..c7a6e1a 100644 --- a/memory-v2-runtime.mjs +++ b/memory-v2-runtime.mjs @@ -13,6 +13,7 @@ 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'; const DEFAULT_PGVECTOR_URL_ENV = 'MEMORY_PGVECTOR_DATABASE_URL'; @@ -416,6 +417,23 @@ export async function createMemoryV2Runtime({ logger, personalShadowPipeline, }); + const lifecycle = createMemoryV2LifecycleService({ pool: mysqlPool, env, logger }); + let lifecycleTimer = null; + const lifecycleWorkerEnabled = readFlag(env, 'MEMORY_LIFECYCLE_WORKER_ENABLED', false); + if (lifecycleWorkerEnabled && mysqlPool?.query) { + 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}`); + }); + lifecycleTimer = setInterval(runLifecycle, intervalMs); + lifecycleTimer.unref?.(); + } + const originalGetStatus = memory.getStatus.bind(memory); + memory.getStatus = () => ({ + ...originalGetStatus(), + lifecycle: lifecycle.getStatus(), + }); + memory.lifecycle = lifecycle; const pgBackfillEnabled = readFlag(env, 'MEMORY_PGVECTOR_BACKFILL_ENABLED', pgvectorEnabled); let pgBackfillBusy = false; @@ -467,6 +485,7 @@ export async function createMemoryV2Runtime({ } memory.close = async () => { + if (lifecycleTimer) clearInterval(lifecycleTimer); const results = await Promise.allSettled(closers.map((close) => close())); for (const result of results) { if (result.status === 'rejected') { @@ -608,6 +627,28 @@ export async function createManagedMemoryV2Runtime({ return runtime.observePersonalMemory(input); }, + async listMemories(input = {}) { + const runtime = await ensureRuntime(); + return runtime.lifecycle?.listMemories?.(input) ?? []; + }, + + async forgetMemory(input = {}) { + const runtime = await ensureRuntime(); + return runtime.lifecycle?.forgetMemory?.(input) ?? { ok: false, skipped: true, reason: 'unavailable' }; + }, + + async runLifecycle(input = {}) { + const runtime = await ensureRuntime(); + const lifecycle = runtime.lifecycle; + if (!lifecycle) return { ok: false, reason: 'unavailable' }; + return { + expire: await lifecycle.expire(input), + compact: await lifecycle.compact(input), + promote: await lifecycle.promote(input), + reflect: await lifecycle.reflect(input), + }; + }, + async close() { const runtime = activeRuntime; activeRuntime = null; diff --git a/memory-v2.mjs b/memory-v2.mjs index f70d680..0ebacbb 100644 --- a/memory-v2.mjs +++ b/memory-v2.mjs @@ -67,10 +67,39 @@ export function resolveMemoryV2Policy({ env = process.env, overrides = {} } = {} vectorEnabled: readFlag(env, 'MEMORY_VECTOR_ENABLED', false), backend: String(env?.MEMORY_BACKEND ?? 'legacy').trim() || 'legacy', failOpen: readFlag(env, 'MEMORY_FAIL_OPEN', true), + agentResolveEnabled: readFlag(env, 'MEMORY_AGENT_RESOLVE_ENABLED', false), + agentInjectionMode: normalizeAgentInjectionMode(env?.MEMORY_AGENT_INJECTION_MODE), + agentCanaryUserIds: normalizeUserIdList(env?.MEMORY_AGENT_CANARY_USER_IDS), + agentResolveLimit: resolveBoundedNumber(env?.MEMORY_AGENT_RESOLVE_LIMIT, 3, 1, 50), + agentResolveTimeoutMs: resolveBoundedNumber(env?.MEMORY_AGENT_RESOLVE_TIMEOUT_MS, 1200, 0, 30000), + promotionEnabled: readFlag(env, 'MEMORY_PROMOTION_ENABLED', false), + compactionV2Enabled: readFlag(env, 'MEMORY_COMPACTION_V2_ENABLED', false), + reflectionEnabled: readFlag(env, 'MEMORY_REFLECTION_ENABLED', false), + lifecycleWorkerEnabled: readFlag(env, 'MEMORY_LIFECYCLE_WORKER_ENABLED', false), + lifecycleRolloutMode: String(env?.MEMORY_LIFECYCLE_ROLLOUT_MODE ?? 'off').trim().toLowerCase() || 'off', + lifecycleRolloutUserIds: normalizeUserIdList(env?.MEMORY_LIFECYCLE_ROLLOUT_USER_IDS), ...overrides, }; } +function resolveBoundedNumber(value, fallback, min, max) { + const parsed = Number(value); + if (!Number.isFinite(parsed)) return fallback; + return Math.min(max, Math.max(min, parsed)); +} + +function normalizeAgentInjectionMode(value) { + const mode = String(value ?? 'off').trim().toLowerCase(); + return ['off', 'shadow', 'canary', 'active'].includes(mode) ? mode : 'off'; +} + +function normalizeUserIdList(value) { + return [...new Set(String(value ?? '') + .split(/[\s,]+/u) + .map((item) => item.trim()) + .filter(Boolean))].slice(0, 1000); +} + export function createLegacyMemoryBackend(conversationMemoryService) { return { name: 'legacy-conversation-memory', @@ -373,6 +402,19 @@ export function createMemoryV2({ eventLogEnabled: Boolean(resolvedPolicy.eventLogEnabled), vectorEnabled: Boolean(resolvedPolicy.vectorEnabled), failOpen: Boolean(resolvedPolicy.failOpen), + runtimeControl: { + agentResolveEnabled: Boolean(resolvedPolicy.agentResolveEnabled), + agentInjectionMode: resolvedPolicy.agentInjectionMode, + agentCanaryUserIds: resolvedPolicy.agentCanaryUserIds, + agentResolveLimit: Number(resolvedPolicy.agentResolveLimit), + agentResolveTimeoutMs: Number(resolvedPolicy.agentResolveTimeoutMs), + promotionEnabled: Boolean(resolvedPolicy.promotionEnabled), + compactionV2Enabled: Boolean(resolvedPolicy.compactionV2Enabled), + reflectionEnabled: Boolean(resolvedPolicy.reflectionEnabled), + lifecycleWorkerEnabled: Boolean(resolvedPolicy.lifecycleWorkerEnabled), + lifecycleRolloutMode: resolvedPolicy.lifecycleRolloutMode, + lifecycleRolloutUserIds: resolvedPolicy.lifecycleRolloutUserIds, + }, backends, }; if (shadowPipeline?.config?.enabled) { diff --git a/memory-v2.test.mjs b/memory-v2.test.mjs index 83917dd..3d75d96 100644 --- a/memory-v2.test.mjs +++ b/memory-v2.test.mjs @@ -33,6 +33,12 @@ test('resolveMemoryV2Policy uses legacy memory flag for backward compatibility', }).vectorEnabled, true, ); + const runtimePolicy = resolveMemoryV2Policy({ env: {} }); + assert.equal(runtimePolicy.agentResolveEnabled, false); + assert.equal(runtimePolicy.agentInjectionMode, 'off'); + assert.deepEqual(runtimePolicy.agentCanaryUserIds, []); + assert.equal(runtimePolicy.agentResolveLimit, 3); + assert.equal(runtimePolicy.promotionEnabled, false); }); test('legacy backend adapts existing conversation memory service', async () => { @@ -446,6 +452,19 @@ test('Memory V2 getStatus exposes policy and backend contract details', () => { eventLogEnabled: true, vectorEnabled: true, failOpen: true, + runtimeControl: { + agentResolveEnabled: false, + agentInjectionMode: 'off', + agentCanaryUserIds: [], + agentResolveLimit: 3, + agentResolveTimeoutMs: 1200, + promotionEnabled: false, + compactionV2Enabled: false, + reflectionEnabled: false, + lifecycleWorkerEnabled: false, + lifecycleRolloutMode: 'off', + lifecycleRolloutUserIds: [], + }, backends: [ { name: 'legacy-conversation-memory', diff --git a/server.mjs b/server.mjs index fe90c47..b4da6c0 100644 --- a/server.mjs +++ b/server.mjs @@ -2291,6 +2291,37 @@ api.post('/user-memory/v1/sync', async (req, res) => { } }); +api.get('/user-memory/v1/items', async (req, res) => { + const capabilityState = await ensureUserMemoryCapability(req, res); + if (!capabilityState) return; + try { + const items = await memoryV2.listMemories?.({ + userId: req.currentUser.id, + status: String(req.query?.status ?? 'active'), + limit: req.query?.limit, + offset: req.query?.offset, + }) ?? []; + return res.json({ ok: true, items }); + } catch (err) { + return res.status(500).json({ message: err instanceof Error ? err.message : '读取长期记忆失败' }); + } +}); + +api.delete('/user-memory/v1/items/:memoryId', async (req, res) => { + const capabilityState = await ensureUserMemoryCapability(req, res); + if (!capabilityState) return; + try { + const result = await memoryV2.forgetMemory?.({ + userId: req.currentUser.id, + memoryId: req.params.memoryId, + }) ?? { ok: false, skipped: true, reason: 'unavailable' }; + if (result.skipped) return res.status(409).json(result); + return res.json(result); + } catch (err) { + return res.status(500).json({ message: err instanceof Error ? err.message : '删除长期记忆失败' }); + } +}); + api.get('/mindspace/v1/space', async (req, res) => { if (!mindSpace || !ensureMindSpaceEnabled(res, req)) return; const space = await mindSpace.getSpace(req.currentUser.id);