/** * Repair Goose conversations from h5_conversation_messages when agent sync * left empty placeholders (portal direct chat → goosed escalation). */ import { extractConversationMessageText } from './conversation-memory.mjs'; export function parseStoredConversationRow(row) { if (!row) return null; const rawJson = row.raw_json ?? row.rawJson; if (typeof rawJson === 'string' && rawJson.trim()) { try { const parsed = JSON.parse(rawJson); if (parsed && typeof parsed === 'object') { return { ...parsed, metadata: { userVisible: true, agentVisible: true, ...(parsed.metadata && typeof parsed.metadata === 'object' ? parsed.metadata : {}), }, }; } } catch { // fall through to text reconstruction } } const text = String(row.text ?? '').trim(); if (!text) return null; const messageKey = String(row.message_key ?? row.messageKey ?? '').trim(); const createdMs = Number(row.created_at ?? row.createdAt ?? 0) || Date.now(); return { id: messageKey || undefined, role: String(row.role ?? 'user').trim() || 'user', created: Math.floor(createdMs / 1000), content: [{ type: 'text', text }], metadata: { userVisible: true, agentVisible: true, ...(String(row.role ?? '') === 'assistant' ? { source: 'conversation-db-repair' } : {}), }, }; } export function filterUserVisibleConversation(messages) { if (!Array.isArray(messages)) return []; return messages.filter((message) => message?.metadata?.userVisible !== false); } export function parseAgentRunUserMessage(row) { const raw = row?.user_message_json ?? row?.userMessageJson; let parsed = raw; if (typeof raw === 'string') { try { parsed = JSON.parse(raw); } catch { return null; } } if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) return null; const message = { ...parsed, role: 'user', metadata: { ...(parsed.metadata && typeof parsed.metadata === 'object' ? parsed.metadata : {}), userVisible: true, agentVisible: true, memoryInputSource: 'agent-run-original', }, }; return extractConversationMessageText(message) ? message : null; } export function restoreConversationUserMessagesFromAgentRunRows(messages, runRows) { const conversation = Array.isArray(messages) ? messages : []; const originals = (Array.isArray(runRows) ? runRows : []) .map((row) => parseAgentRunUserMessage(row)) .filter(Boolean); if (originals.length === 0) return conversation; const userIndexes = conversation .map((message, index) => (message?.role === 'user' ? index : -1)) .filter((index) => index >= 0); if (userIndexes.length === 0) return conversation; const restored = [...conversation]; let userOffset = userIndexes.length - 1; let originalOffset = originals.length - 1; while (userOffset >= 0 && originalOffset >= 0) { const messageIndex = userIndexes[userOffset]; const current = restored[messageIndex] ?? {}; const original = originals[originalOffset]; restored[messageIndex] = { ...current, role: 'user', content: original.content, metadata: { ...(current.metadata && typeof current.metadata === 'object' ? current.metadata : {}), ...(original.metadata && typeof original.metadata === 'object' ? original.metadata : {}), userVisible: current?.metadata?.userVisible ?? true, agentVisible: current?.metadata?.agentVisible ?? true, memoryInputSource: 'agent-run-original', }, }; userOffset -= 1; originalOffset -= 1; } return restored; } export async function loadSuccessfulAgentRunUserMessageRows( pool, sessionId, userId, { limit = 200 } = {}, ) { if (!pool || !sessionId || !userId) return []; const resolvedLimit = Math.max(1, Math.min(200, Number(limit) || 200)); const [rows] = await pool.query( `SELECT id, user_message_json, created_at FROM h5_agent_runs WHERE agent_session_id = ? AND user_id = ? AND status = 'succeeded' ORDER BY created_at DESC, id DESC LIMIT ?`, [sessionId, userId, resolvedLimit], ); return [...rows].reverse(); } export async function restoreConversationUserMessagesFromAgentRuns(pool, messages, sessionId, userId) { const userMessageCount = Array.isArray(messages) ? messages.filter((message) => message?.role === 'user').length : 0; if (userMessageCount === 0) return Array.isArray(messages) ? messages : []; const rows = await loadSuccessfulAgentRunUserMessageRows(pool, sessionId, userId, { limit: userMessageCount, }); return restoreConversationUserMessagesFromAgentRunRows(messages, rows); } export async function restoreConversationUserMessagesFromAgentRunsFailOpen( pool, messages, sessionId, userId, { logger = console } = {}, ) { try { return await restoreConversationUserMessagesFromAgentRuns(pool, messages, sessionId, userId); } catch (err) { logger?.warn?.( '[memory-v2] original agent-run transcript unavailable; using visible session transcript:', err instanceof Error ? err.message : err, ); return Array.isArray(messages) ? messages : []; } } export function countNonEmptyConversationMessages(messages) { if (!Array.isArray(messages)) return 0; return messages.filter((message) => { if (message?.metadata?.userVisible === false) return false; return Boolean(extractConversationMessageText(message)); }).length; } export function buildConversationFromDbRows(dbRows) { if (!Array.isArray(dbRows) || dbRows.length === 0) return []; const byKey = new Map(); for (const row of dbRows) { const parsed = parseStoredConversationRow(row); if (!parsed) continue; const key = String(row.message_key ?? row.messageKey ?? parsed.id ?? '').trim() || `${parsed.role}:${extractConversationMessageText(parsed).slice(0, 48)}`; byKey.set(key, { message: parsed, sequenceNo: Number(row.sequence_no ?? row.sequenceNo ?? 0), createdAt: Number(row.created_at ?? row.createdAt ?? 0), }); } return [...byKey.values()] .sort((left, right) => { if (left.sequenceNo !== right.sequenceNo) return left.sequenceNo - right.sequenceNo; if (left.createdAt !== right.createdAt) return left.createdAt - right.createdAt; return String(left.message.id ?? '').localeCompare(String(right.message.id ?? '')); }) .map((entry) => entry.message); } export function shouldRepairConversationFromDb(gooseMessages, dbRows) { const dbBuilt = buildConversationFromDbRows(dbRows); const dbCount = countNonEmptyConversationMessages(dbBuilt); if (dbCount === 0) return false; const gooseVisible = Array.isArray(gooseMessages) ? gooseMessages.filter((message) => message?.metadata?.userVisible !== false) : []; const gooseNonEmpty = countNonEmptyConversationMessages(gooseVisible); const gooseEmptyVisible = gooseVisible.some( (message) => !extractConversationMessageText(message), ); if (gooseNonEmpty === 0) return true; if (dbCount > gooseNonEmpty) return true; if (gooseEmptyVisible && dbCount >= gooseNonEmpty) return true; return false; } export function repairConversationFromDbRows(gooseMessages, dbRows) { const dbBuilt = buildConversationFromDbRows(dbRows); if (!shouldRepairConversationFromDb(gooseMessages, dbRows)) { return Array.isArray(gooseMessages) ? gooseMessages : []; } if (dbBuilt.length === 0) return Array.isArray(gooseMessages) ? gooseMessages : []; return dbBuilt; } export async function loadSessionConversationRows(pool, sessionId, userId) { if (!pool || !sessionId || !userId) return []; const [rows] = await pool.query( `SELECT message_key, sequence_no, role, text, raw_json, created_at FROM h5_conversation_messages WHERE agent_session_id = ? AND user_id = ? ORDER BY sequence_no ASC, created_at ASC`, [sessionId, userId], ); return rows; } export async function repairSessionConversationFromDb(pool, gooseSession, sessionId, userId) { if (!gooseSession || !pool || !sessionId || !userId) return gooseSession; const dbRows = await loadSessionConversationRows(pool, sessionId, userId); const conversation = Array.isArray(gooseSession.conversation) ? gooseSession.conversation : []; if (!shouldRepairConversationFromDb(conversation, dbRows)) return gooseSession; return { ...gooseSession, conversation: repairConversationFromDbRows(conversation, dbRows), }; }