memind/sse-event-taxonomy.test.mjs

import assert from 'node:assert/strict';
import { EventEmitter } from 'node:events';
import test from 'node:test';
import {
  classifyRunStreamTaxonomy,
  classifySessionEventTaxonomy,
  finalizeSessionStreamEvent,
  isSseEventTaxonomyEnabled,
  SSE_EVENT_TAXONOMY,
  wrapRunStreamPayload,
  writeSseErrorAndEnd,
} from './sse-event-taxonomy.mjs';

test('classifySessionEventTaxonomy maps goosed session events', () => {
  assert.equal(
    classifySessionEventTaxonomy({ type: 'Message', message: { role: 'assistant' } }),
    SSE_EVENT_TAXONOMY.DELTA,
  );
  assert.equal(
    classifySessionEventTaxonomy({ type: 'UpdateConversation', conversation: [] }),
    SSE_EVENT_TAXONOMY.CONTROL,
  );
  assert.equal(
    classifySessionEventTaxonomy({ type: 'Finish', reason: 'stop' }),
    SSE_EVENT_TAXONOMY.TERMINAL,
  );
  assert.equal(
    classifySessionEventTaxonomy({ type: 'Error', error: 'boom' }),
    SSE_EVENT_TAXONOMY.TERMINAL,
  );
  assert.equal(
    classifySessionEventTaxonomy({ type: 'Ping' }),
    SSE_EVENT_TAXONOMY.CONTROL,
  );
});

test('classifyRunStreamTaxonomy separates control and terminal run states', () => {
  assert.equal(
    classifyRunStreamTaxonomy('run', { run: { status: 'running' } }),
    SSE_EVENT_TAXONOMY.CONTROL,
  );
  assert.equal(
    classifyRunStreamTaxonomy('run', { run: { status: 'succeeded' } }),
    SSE_EVENT_TAXONOMY.TERMINAL,
  );
  assert.equal(
    classifyRunStreamTaxonomy('error', { message: 'failed' }),
    SSE_EVENT_TAXONOMY.TERMINAL,
  );
});

test('finalizeSessionStreamEvent attaches taxonomy only when flag enabled', () => {
  const event = { type: 'Finish', reason: 'stop' };
  const previous = process.env.MEMIND_SSE_EVENT_TAXONOMY;
  process.env.MEMIND_SSE_EVENT_TAXONOMY = '0';
  try {
    assert.deepEqual(finalizeSessionStreamEvent(event), event);
  } finally {
    if (previous == null) delete process.env.MEMIND_SSE_EVENT_TAXONOMY;
    else process.env.MEMIND_SSE_EVENT_TAXONOMY = previous;
  }

  process.env.MEMIND_SSE_EVENT_TAXONOMY = '1';
  try {
    assert.equal(isSseEventTaxonomyEnabled(), true);
    assert.deepEqual(finalizeSessionStreamEvent(event), {
      ...event,
      taxonomy: SSE_EVENT_TAXONOMY.TERMINAL,
    });
  } finally {
    if (previous == null) delete process.env.MEMIND_SSE_EVENT_TAXONOMY;
    else process.env.MEMIND_SSE_EVENT_TAXONOMY = previous;
  }
});

test('wrapRunStreamPayload keeps legacy payload when flag disabled', () => {
  const payload = { run: { id: 'run-1', status: 'running' } };
  const previous = process.env.MEMIND_SSE_EVENT_TAXONOMY;
  delete process.env.MEMIND_SSE_EVENT_TAXONOMY;
  try {
    assert.deepEqual(wrapRunStreamPayload('run', payload), payload);
  } finally {
    if (previous == null) delete process.env.MEMIND_SSE_EVENT_TAXONOMY;
    else process.env.MEMIND_SSE_EVENT_TAXONOMY = previous;
  }
});

test('writeSseErrorAndEnd uses SSE error frame after headers sent', () => {
  const res = new EventEmitter();
  res.headersSent = true;
  res.writableEnded = false;
  const chunks = [];
  res.write = (chunk) => {
    chunks.push(String(chunk));
    return true;
  };
  res.end = () => {
    res.writableEnded = true;
  };
  res.status = () => {
    throw new Error('status must not be called after SSE started');
  };
  res.json = () => {
    throw new Error('json must not be called after SSE started');
  };

  writeSseErrorAndEnd(res, 'stream failed');
  assert.equal(res.writableEnded, true);
  assert.match(chunks.join(''), /event: error/);
  assert.match(chunks.join(''), /stream failed/);
});