memind/session-reply-wait.mjs

import { Readable } from 'node:stream';

function parseSseDataLines(frame) {
  let data = '';
  for (const line of String(frame ?? '').split('\n')) {
    if (line.startsWith('data:')) data += line.slice(5).trim();
  }
  return data;
}

export function parseSessionStreamEvent(frame) {
  const data = parseSseDataLines(frame);
  if (!data) return null;
  try {
    return JSON.parse(data);
  } catch {
    return null;
  }
}

export function eventMatchesRequest(event, requestId) {
  if (!requestId) return true;
  const routingId = event?.chat_request_id ?? event?.request_id ?? null;
  return !routingId || routingId === requestId;
}

export async function consumeSessionEventsUntilFinish(
  body,
  {
    requestId = null,
    timeoutMs = 15 * 60 * 1000,
    onEvent = null,
  } = {},
) {
  if (!body) {
    const err = new Error('session event stream unavailable');
    err.code = 'SESSION_EVENT_STREAM_UNAVAILABLE';
    throw err;
  }

  const reader = Readable.fromWeb(body);
  const decoder = new TextDecoder();
  let buffer = '';
  const deadline = Date.now() + Math.max(1, Number(timeoutMs) || 1);

  for await (const chunk of reader) {
    if (Date.now() > deadline) {
      const err = new Error(`session reply timed out after ${timeoutMs}ms`);
      err.code = 'SESSION_REPLY_TIMEOUT';
      throw err;
    }

    buffer += decoder.decode(chunk, { stream: true });
    const frames = buffer.split('\n\n');
    buffer = frames.pop() ?? '';
    for (const frame of frames) {
      const trimmed = frame.trim();
      if (!trimmed || trimmed.startsWith(':')) continue;
      const event = parseSessionStreamEvent(trimmed);
      if (!event) continue;
      if (!eventMatchesRequest(event, requestId)) continue;
      onEvent?.(event);
      if (event.type === 'Error') {
        const err = new Error(String(event.error ?? 'session reply failed'));
        err.code = 'SESSION_REPLY_ERROR';
        throw err;
      }
      if (event.type === 'Finish') {
        return {
          finishEvent: event,
          tokenState: event.token_state ?? null,
        };
      }
    }
  }

  const err = new Error('session event stream ended before Finish');
  err.code = 'SESSION_REPLY_INCOMPLETE';
  throw err;
}