memind/sse-event-taxonomy.mjs

export const SSE_EVENT_TAXONOMY = {
  DELTA: 'delta',
  CONTROL: 'control',
  TERMINAL: 'terminal',
};

const TERMINAL_RUN_STATUSES = new Set(['succeeded', 'failed']);

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

export function isSseEventTaxonomyEnabled(env = process.env) {
  return envFlag(env.MEMIND_SSE_EVENT_TAXONOMY, false);
}

export function classifySessionEventTaxonomy(event) {
  const type = String(event?.type ?? '').trim();
  if (!type) return null;
  switch (type) {
    case 'Message':
      return SSE_EVENT_TAXONOMY.DELTA;
    case 'Finish':
    case 'Error':
      return SSE_EVENT_TAXONOMY.TERMINAL;
    case 'UpdateConversation':
    case 'Ping':
    case 'ActiveRequests':
      return SSE_EVENT_TAXONOMY.CONTROL;
    default:
      break;
  }
  const normalized = type.toLowerCase();
  if (normalized === 'message') return SSE_EVENT_TAXONOMY.DELTA;
  if (normalized === 'finish' || normalized === 'error') return SSE_EVENT_TAXONOMY.TERMINAL;
  if (
    normalized === 'updateconversation' ||
    normalized === 'ping' ||
    normalized === 'activerequests'
  ) {
    return SSE_EVENT_TAXONOMY.CONTROL;
  }
  return null;
}

export function classifyRunStreamTaxonomy(eventName, payload = {}) {
  const name = String(eventName ?? '').trim().toLowerCase();
  if (name === 'error') return SSE_EVENT_TAXONOMY.TERMINAL;
  if (name !== 'run') return null;
  const status = String(payload?.run?.status ?? payload?.status ?? '').trim().toLowerCase();
  if (TERMINAL_RUN_STATUSES.has(status)) return SSE_EVENT_TAXONOMY.TERMINAL;
  if (status) return SSE_EVENT_TAXONOMY.CONTROL;
  return null;
}

export function finalizeSessionStreamEvent(event, env = process.env) {
  if (!event || typeof event !== 'object' || Array.isArray(event)) return event;
  if (!isSseEventTaxonomyEnabled(env)) return event;
  const taxonomy = classifySessionEventTaxonomy(event);
  if (!taxonomy || event.taxonomy === taxonomy) return event;
  return { ...event, taxonomy };
}

export function wrapRunStreamPayload(eventName, data, env = process.env) {
  if (!data || typeof data !== 'object' || Array.isArray(data)) return data;
  if (!isSseEventTaxonomyEnabled(env)) return data;
  const taxonomy = classifyRunStreamTaxonomy(eventName, data);
  if (!taxonomy || data.taxonomy === taxonomy) return data;
  return { ...data, taxonomy };
}

export function writeSseErrorAndEnd(res, message, { status = 502 } = {}) {
  if (res.writableEnded) return;
  if (res.headersSent) {
    try {
      res.write(`event: error\ndata: ${JSON.stringify({ message })}\n\n`);
    } catch {
      // Client may already be gone.
    }
    res.end();
    return;
  }
  res.status(status).json({ message });
}