memind/scripts/goose-v149-sse.mjs

/**
 * Shared SSE helpers for local Goose v1.49 smoke scripts.
 */
import { randomUUID } from 'node:crypto';
import { Readable } from 'node:stream';
import { Agent, fetch } from 'undici';

const blockedHosts = new Set(['58.38.22.103', '120.26.184.105']);

export function createV149Client(options = {}) {
  const host = options.host ?? process.env.GOOSE_V149_HOST ?? '127.0.0.1';
  const port = options.port ?? process.env.GOOSE_V149_PORT ?? '18049';
  const secret = options.secret ?? process.env.GOOSE_SERVER__SECRET_KEY ?? 'local-v149-dev-secret';
  if (blockedHosts.has(host)) {
    throw new Error(`Refusing production host ${host}`);
  }
  const base = `https://${host}:${port}`;
  const dispatcher = new Agent({ connect: { rejectUnauthorized: false } });

  async function apiFetch(pathname, init = {}) {
    const headers = { ...(init.headers ?? {}), 'X-Secret-Key': secret };
    if (init.body && !headers['Content-Type']) headers['Content-Type'] = 'application/json';
    return fetch(`${base}${pathname}`, { ...init, headers, dispatcher });
  }

  async function apiJson(pathname, body, init = {}) {
    const response = await apiFetch(pathname, {
      method: init.method ?? 'POST',
      body: body == null ? undefined : JSON.stringify(body),
      ...init,
    });
    const text = await response.text();
    if (!response.ok) {
      throw new Error(`${pathname} ${response.status}: ${text.slice(0, 800)}`);
    }
    return text.trim() ? JSON.parse(text) : {};
  }

  return { base, apiFetch, apiJson };
}

function parseSseData(frame) {
  let data = '';
  for (const line of frame.split('\n')) {
    if (line.startsWith('data:')) data += line.slice(5).trim();
  }
  if (!data) return null;
  try {
    return JSON.parse(data);
  } catch {
    return null;
  }
}

export async function collectReplyEvents({
  client,
  sessionId,
  requestId = randomUUID(),
  userMessage,
  timeoutMs = Number(process.env.GOOSE_V149_REPLY_TIMEOUT_MS || 90_000),
  triggerReply = true,
}) {
  const events = [];
  const eventsResponse = await client.apiFetch(`/sessions/${sessionId}/events`, {
    method: 'GET',
    headers: { Accept: 'text/event-stream' },
  });
  if (!eventsResponse.ok || !eventsResponse.body) {
    const text = await eventsResponse.text().catch(() => '');
    throw new Error(`events ${eventsResponse.status}: ${text.slice(0, 400)}`);
  }

  if (triggerReply) {
    const replyResponse = await client.apiFetch(`/sessions/${sessionId}/reply`, {
      method: 'POST',
      body: JSON.stringify({
        request_id: requestId,
        user_message: userMessage,
      }),
    });
    if (!replyResponse.ok) {
      const text = await replyResponse.text().catch(() => '');
      throw new Error(`reply ${replyResponse.status}: ${text.slice(0, 400)}`);
    }
    replyResponse.body?.cancel?.();
  }

  const reader = Readable.fromWeb(eventsResponse.body);
  const decoder = new TextDecoder();
  let buffer = '';
  const deadline = Date.now() + timeoutMs;
  let finishEvent = null;
  let errorEvent = null;

  for await (const chunk of reader) {
    if (Date.now() > deadline) break;
    buffer += decoder.decode(chunk, { stream: true });
    const frames = buffer.split('\n\n');
    buffer = frames.pop() ?? '';
    for (const frame of frames) {
      const event = parseSseData(frame);
      if (!event) continue;
      const routingId = event.chat_request_id ?? event.request_id;
      if (routingId && routingId !== requestId) continue;
      events.push(event);
      if (event.type === 'Finish') finishEvent = event;
      if (event.type === 'Error') errorEvent = event;
      if (finishEvent || errorEvent) {
        reader.cancel?.().catch(() => {});
        break;
      }
    }
    if (finishEvent || errorEvent) break;
  }

  return {
    requestId,
    events,
    finishEvent,
    errorEvent,
    outcome: finishEvent ? 'finish' : errorEvent ? 'error' : 'timeout',
  };
}

export function extractTokenStateFromEvents(events = []) {
  for (let i = events.length - 1; i >= 0; i -= 1) {
    const event = events[i];
    if (event?.type === 'Finish' && event.token_state) return event.token_state;
    if (event?.type === 'Message' && event.token_state) return event.token_state;
  }
  return null;
}