memind/memory-v2-lifecycle.mjs

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

export function resolveMemoryV2LifecycleWorkerScopes(policy = {}) {
  if (!policy.enabled) return [];
  if (policy.rolloutMode === 'active') return [null];
  if (policy.rolloutMode === 'canary') {
    return [...new Set((policy.rolloutUserIds ?? []).map(String).filter(Boolean))];
  }
  return [];
}

function normalizeItem(row) {
  return {
    id: String(row.id),
    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 (!policy.forgettingEnabled || !canRun(userId)) return { ok: true, updated: false, skipped: true, reason: 'disabled' };
    if (!pool?.query || !userId || !memoryId) return { ok: false, updated: false, reason: 'invalid_input' };
    const [result] = await pool.query(
      `UPDATE ${MEMORY_TABLE} SET status = 'deleted', updated_at = ? WHERE id = ? AND user_id = ? AND status <> 'deleted'`,
      [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 || !canRun(userId)) {
      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;
    const promotedUserIds = new Set();
    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;
        promotedUserIds.add(String(row.user_id));
      }
    }
    metrics.promote += 1;
    return { ok: true, skipped: false, promoted, promotedUserIds: [...promotedUserIds] };
  }

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