memind/session-stream.mjs

export const SESSION_STREAM_TERMINAL_TYPES = new Set(['Finish', 'Error']);

function envFlag(value, fallback = false) {
  const raw = String(value ?? '').trim().toLowerCase();
  if (!raw) return fallback;
  return ['1', 'true', 'yes', 'on'].includes(raw);
}

export function isSessionStreamReplayEnabled(env = process.env) {
  return envFlag(env.MEMIND_SESSION_STREAM_REPLAY, false);
}

export function parseSessionStreamLastEventId(value) {
  const id = String(value ?? '').trim();
  return id || null;
}

export function isTerminalSessionEvent(event) {
  const type = String(event?.type ?? '').trim();
  return SESSION_STREAM_TERMINAL_TYPES.has(type);
}

export function shouldPersistSessionStreamEvent(event) {
  const type = String(event?.type ?? '').trim();
  if (!type) return false;
  return type !== 'Ping';
}

export function parseSessionSseBlock(block) {
  const lines = String(block ?? '').split('\n');
  let id = null;
  let eventName = null;
  let data = null;
  for (const line of lines) {
    if (line.startsWith('id:')) id = line.slice(3).trim() || null;
    if (line.startsWith('event:')) eventName = line.slice(6).trim() || null;
    if (line.startsWith('data:')) data = line.slice(5).trim();
  }
  return { id, eventName, data };
}

export function formatSessionStreamSseChunk({ id = null, event = null, data }) {
  const lines = [];
  if (id) lines.push(`id: ${id}`);
  if (event) lines.push(`event: ${event}`);
  lines.push(`data: ${JSON.stringify(data)}`);
  return `${lines.join('\n')}\n\n`;
}

export function shouldSkipUpstreamAfterSessionReplay(events, { cursorMiss = false } = {}) {
  if (!Array.isArray(events) || events.length === 0) {
    return cursorMiss ? false : false;
  }
  const last = events[events.length - 1];
  return isTerminalSessionEvent(last?.payload ?? last);
}