/** * 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 episodicMemoryService = options.episodicMemoryService ?? 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, ); }); } if (episodicMemoryService?.upsertSnapshot) { void episodicMemoryService.upsertSnapshot({ sessionId, userId, session: { ...session, display_title: sessionMeta.display_title, }, messages, }).catch((err) => { logger?.warn?.( '[snapshot] episodic memory index 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 { if (episodicMemoryService?.remove) { await episodicMemoryService.remove(sessionId).catch((err) => { logger?.warn?.( '[snapshot] episodic memory delete failed:', err instanceof Error ? err.message : err, ); }); } 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; 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 }; }