memind/session-snapshot.mjs

/**
 * Session snapshot cache service.
 *
 * Stores a DB-side copy of each session's userVisible messages so that
 * repeated opens of historical chats skip the Goose round-trip entirely.
 *
 * Rollback: set SESSION_SNAPSHOT_CACHE_ENABLED=0 in env → all reads/writes
 * are no-ops; behaviour falls back to the existing Goose path unchanged.
 *
 * The table (h5_session_snapshots) is additive-only and has no FK deps on
 * conversation content, so it can be dropped/truncated safely at any time.
 */

import { deriveUserFacingText } from './conversation-display.mjs';

const DERIVED_TITLE_MAX_LEN = 40;

function isGeneratedSessionName(name) {
  return /^20\d{6}(?:[_-]\d+)?$/.test(String(name ?? '').trim());
}

function isDefaultSessionName(name) {
  const trimmed = String(name ?? '').trim();
  return !trimmed || trimmed === 'New Chat' || trimmed === '新对话' || isGeneratedSessionName(trimmed);
}

/** Extract a plain-text snippet from a stored normalized message. */
function messageSnippet(message) {
  const fromContent = (message?.content ?? [])
    .filter((item) => item?.type === 'text' && typeof item.text === 'string')
    .map((item) => item.text)
    .join('')
    .trim();
  if (message?.role === 'user') {
    return deriveUserFacingText(fromContent);
  }
  if (fromContent) return fromContent;
  const displayText = message?.metadata?.displayText;
  return typeof displayText === 'string' ? displayText.trim() : '';
}

/** Build a one-line title from the first user message of a stored conversation. */
function deriveTitleFromMessages(messages) {
  if (!Array.isArray(messages)) return '';
  const firstUser = messages.find((m) => m?.role === 'user' && messageSnippet(m));
  const text = (firstUser ? messageSnippet(firstUser) : '')
    .replace(/\s+/g, ' ')
    .trim();
  if (!text) return '';
  return text.length > DERIVED_TITLE_MAX_LEN
    ? `${text.slice(0, DERIVED_TITLE_MAX_LEN)}…`
    : text;
}

function resolveDisplayTitle(session, messages) {
  if (session?.user_set_name) {
    const explicit = String(session.name ?? '').trim();
    if (explicit) return explicit;
  }

  const recipeTitle = String(session?.recipe?.title ?? '').trim();
  if (recipeTitle) return recipeTitle;

  const sessionName = String(session?.name ?? '').trim();
  if (sessionName && !isDefaultSessionName(sessionName)) return sessionName;

  return deriveTitleFromMessages(messages);
}

import { filterNonemptyUserVisibleMessages } from './conversation-transcript-persist.mjs';

async function syncConversationMemory({
  memoryV2 = null,
  conversationMemoryService = null,
  sessionId,
  userId,
  messages,
  logger = console,
} = {}) {
  if (memoryV2?.write) {
    const result = await memoryV2.write({ userId, sessionId, messages });
    if (result?.skipped && result?.reason === 'disabled') return result;
    if (!result?.ok && !result?.skipped) {
      logger?.warn?.(
        '[snapshot] memory v2 write degraded:',
        result?.reason ?? 'unknown',
      );
    }
    return result;
  }
  if (conversationMemoryService?.isEnabled?.()) {
    return conversationMemoryService.saveAndAnalyze(sessionId, userId, messages);
  }
  return null;
}

export function createSessionSnapshotService(pool, options = {}) {
  const conversationMemoryService = options.conversationMemoryService ?? null;
  const memoryV2 = options.memoryV2 ?? null;
  const logger = options.logger ?? console;

  function isEnabled() {
    return process.env.SESSION_SNAPSHOT_CACHE_ENABLED !== '0';
  }

  async function persistDisplayTitle(sessionId, title) {
    const normalized = String(title ?? '').trim();
    if (!normalized || !pool) return;
    try {
      await pool.query(
        `UPDATE h5_session_snapshots
            SET display_title = ?
          WHERE agent_session_id = ?
            AND (display_title IS NULL OR display_title = '')`,
        [normalized, sessionId],
      );
    } catch (err) {
      console.warn('[snapshot] persistDisplayTitle failed:', err instanceof Error ? err.message : err);
    }
  }

  /**
   * Persist a snapshot for a session.
   * @param {string} sessionId
   * @param {string} userId
   * @param {object} session  – raw Goose session object (includes message_count, updated_at, etc.)
   * @param {Array}  messages – already-normalized userVisible Message[]
   */
  async function save(sessionId, userId, session, messages) {
    if (!isEnabled() || !pool) return;
    try {
      const now = Date.now();
      const sessionMeta = {
        name: session.name ?? '',
        display_title: resolveDisplayTitle(session, messages),
        working_dir: session.working_dir ?? '',
        created_at_str: session.created_at ?? '',
        updated_at_str: session.updated_at ?? '',
        user_set_name: session.user_set_name ? 1 : 0,
        recipe_json: session.recipe != null ? JSON.stringify(session.recipe) : null,
      };
      await pool.query(
        `INSERT INTO h5_session_snapshots
           (agent_session_id, user_id, name, display_title, working_dir,
            created_at_str, updated_at_str, user_set_name, recipe_json,
            synced_msg_count, source_updated_at,
            messages_json, synced_at)
         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
         ON DUPLICATE KEY UPDATE
           name = VALUES(name),
           display_title = VALUES(display_title),
           working_dir = VALUES(working_dir),
           created_at_str = VALUES(created_at_str),
           updated_at_str = VALUES(updated_at_str),
           user_set_name = VALUES(user_set_name),
           recipe_json = VALUES(recipe_json),
           synced_msg_count = VALUES(synced_msg_count),
           source_updated_at = VALUES(source_updated_at),
           messages_json = VALUES(messages_json),
           synced_at = VALUES(synced_at)`,
        [
          sessionId,
          userId,
          sessionMeta.name,
          sessionMeta.display_title,
          sessionMeta.working_dir,
          sessionMeta.created_at_str,
          sessionMeta.updated_at_str,
          sessionMeta.user_set_name,
          sessionMeta.recipe_json,
          session.message_count ?? 0,
          session.updated_at ?? '',
          JSON.stringify(messages),
          now,
        ],
      );
      if (memoryV2?.write || conversationMemoryService?.isEnabled?.()) {
        void syncConversationMemory({
          memoryV2,
          conversationMemoryService,
          sessionId,
          userId,
          messages,
          logger,
        }).catch((err) => {
          logger?.warn?.(
            '[snapshot] conversation memory sync failed:',
            err instanceof Error ? err.message : err,
          );
        });
      }
    } catch (err) {
      // Non-fatal: log and continue; caller falls back to Goose.
      console.warn('[snapshot] save failed:', err instanceof Error ? err.message : err);
    }
  }

  /**
   * Read a snapshot. Returns null if not found or cache is disabled.
   * The returned object reconstructs what `loadSessionDetail` returns.
   */
  async function get(sessionId) {
    if (!isEnabled() || !pool) return null;
    try {
      const [rows] = await pool.query(
        `SELECT agent_session_id, user_id, name, display_title, working_dir,
                created_at_str, updated_at_str, user_set_name, recipe_json,
                synced_msg_count, source_updated_at, messages_json, synced_at
         FROM h5_session_snapshots
         WHERE agent_session_id = ?
         LIMIT 1`,
        [sessionId],
      );
      if (!rows.length) return null;
      const row = rows[0];
      const parsedMessages = JSON.parse(row.messages_json);
      const storedDisplayTitle = String(row.display_title ?? '').trim();
      const rawName = String(row.name ?? '').trim();
      const fallbackDisplayTitle =
        storedDisplayTitle ||
        resolveDisplayTitle(
          {
            name: row.name,
            recipe: row.recipe_json ? JSON.parse(row.recipe_json) : null,
            user_set_name: row.user_set_name === 1,
          },
          parsedMessages,
        );
      if (!storedDisplayTitle && fallbackDisplayTitle) {
        void persistDisplayTitle(sessionId, fallbackDisplayTitle);
      }
      return {
        session: {
          id: sessionId,
          name: fallbackDisplayTitle || rawName,
          working_dir: row.working_dir,
          message_count: row.synced_msg_count,
          created_at: row.created_at_str || undefined,
          updated_at: row.updated_at_str || undefined,
          user_set_name: row.user_set_name === 1,
          recipe: row.recipe_json ? JSON.parse(row.recipe_json) : null,
          conversation: null,
        },
        messages: parsedMessages,
        meta: {
          synced_msg_count: row.synced_msg_count,
          source_updated_at: row.source_updated_at,
          synced_at: row.synced_at,
        },
      };
    } catch (err) {
      console.warn('[snapshot] get failed:', err instanceof Error ? err.message : err);
      return null;
    }
  }

  /**
   * Delete a snapshot (called when the session itself is deleted).
   */
  async function remove(sessionId) {
    if (!pool) return;
    try {
      await pool.query(
        `DELETE FROM h5_session_snapshots WHERE agent_session_id = ?`,
        [sessionId],
      );
    } catch (err) {
      console.warn('[snapshot] remove failed:', err instanceof Error ? err.message : err);
    }
  }

  /**
   * Fetch full session from Goose and write/update the snapshot.
   * Fire-and-forget safe: errors are swallowed.
   */
  async function refresh(sessionId, userId, apiFetchFn) {
    if (!isEnabled() || !pool) return;
    try {
      const res = await apiFetchFn(`/sessions/${encodeURIComponent(sessionId)}`, {
        method: 'GET',
      });
      if (!res.ok) return;
      const gooseSession = await res.json();
      // Store userVisible messages with content — skip empty Goose placeholders.
      const messages = filterNonemptyUserVisibleMessages(gooseSession.conversation ?? []);
      await save(sessionId, userId, gooseSession, messages);
    } catch (err) {
      console.warn('[snapshot] refresh failed:', sessionId, err instanceof Error ? err.message : err);
    }
  }

  /**
   * Derive display titles from the first user message for the given session ids.
   * Used to give unnamed/default sessions a meaningful label in the history list.
   * Returns a Map<sessionId, title>; sessions without a usable snapshot are omitted.
   */
  async function getDerivedTitles(sessionIds) {
    if (!isEnabled() || !pool) return new Map();
    const ids = [...new Set((sessionIds ?? []).filter(Boolean))];
    if (ids.length === 0) return new Map();
    const titles = new Map();
    try {
      const placeholders = ids.map(() => '?').join(', ');
      const [rows] = await pool.query(
        `SELECT agent_session_id, display_title, name, recipe_json, user_set_name, messages_json
           FROM h5_session_snapshots
          WHERE agent_session_id IN (${placeholders})`,
        ids,
      );
      for (const row of rows) {
        try {
          const parsedMessages = JSON.parse(row.messages_json);
          const title =
            String(row.display_title ?? '').trim() ||
            resolveDisplayTitle(
              {
                name: row.name,
                recipe: row.recipe_json ? JSON.parse(row.recipe_json) : null,
                user_set_name: row.user_set_name === 1,
              },
              parsedMessages,
            );
          if (title) {
            titles.set(row.agent_session_id, title);
            if (!String(row.display_title ?? '').trim()) {
              void persistDisplayTitle(row.agent_session_id, title);
            }
          }
        } catch {
          // Skip rows with unparseable message payloads.
        }
      }
    } catch (err) {
      console.warn('[snapshot] getDerivedTitles failed:', err instanceof Error ? err.message : err);
    }
    return titles;
  }

  async function getHistorySummaries(sessionIds) {
    if (!isEnabled() || !pool) return new Map();
    const ids = [...new Set((sessionIds ?? []).filter(Boolean))];
    if (ids.length === 0) return new Map();
    const summaries = new Map();
    try {
      const placeholders = ids.map(() => '?').join(', ');
      const [rows] = await pool.query(
        `SELECT agent_session_id, display_title, name, recipe_json, user_set_name,
                synced_msg_count, messages_json
           FROM h5_session_snapshots
          WHERE agent_session_id IN (${placeholders})`,
        ids,
      );
      for (const row of rows) {
        try {
          const parsedMessages = JSON.parse(row.messages_json);
          const title =
            String(row.display_title ?? '').trim() ||
            resolveDisplayTitle(
              {
                name: row.name,
                recipe: row.recipe_json ? JSON.parse(row.recipe_json) : null,
                user_set_name: row.user_set_name === 1,
              },
              parsedMessages,
            );
          const messageCount = Number(row.synced_msg_count ?? 0);
          summaries.set(row.agent_session_id, {
            title,
            messageCount,
          });
          if (title && !String(row.display_title ?? '').trim()) {
            void persistDisplayTitle(row.agent_session_id, title);
          }
        } catch {
          // Skip rows with unparseable payloads.
        }
      }
    } catch (err) {
      console.warn('[snapshot] getHistorySummaries failed:', err instanceof Error ? err.message : err);
    }
    return summaries;
  }

  return { save, get, remove, refresh, getDerivedTitles, getHistorySummaries, isEnabled };
}