memind/agent-run-stream.mjs

export const RUN_STREAM_STATUS_EVENTS = new Set([
  'queued',
  'running',
  'retryable',
  'succeeded',
  'failed',
]);

export const RUN_STREAM_TERMINAL_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 isRunStreamReplayEnabled(env = process.env) {
  return envFlag(env.MEMIND_RUN_STREAM_REPLAY, false);
}

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

export function extractRunFromStreamEvent(event) {
  if (!event) return null;
  if (event.eventType === 'run_snapshot' && event.data?.run) {
    return event.data.run;
  }
  return null;
}

export function shouldEmitRunUpdateForStreamEvent(eventType) {
  return RUN_STREAM_STATUS_EVENTS.has(String(eventType ?? '').trim())
    || String(eventType ?? '').trim() === 'run_snapshot';
}

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

export function projectRunStreamEvents(events, { fallbackRun = null } = {}) {
  let latestRun = fallbackRun;
  const emittedRuns = [];
  for (const event of Array.isArray(events) ? events : []) {
    const snapshotRun = extractRunFromStreamEvent(event);
    if (snapshotRun) {
      latestRun = snapshotRun;
      emittedRuns.push({ eventId: event.id, run: snapshotRun });
      continue;
    }
    if (shouldEmitRunUpdateForStreamEvent(event.eventType) && latestRun) {
      emittedRuns.push({ eventId: event.id, run: latestRun });
    }
  }
  return { latestRun, emittedRuns };
}

export function isTerminalRunStatus(status) {
  return RUN_STREAM_TERMINAL_STATUSES.has(String(status ?? '').trim());
}