memind/conversation-repair.mjs

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