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 } }; }, }; }