diegosouzapw--omniroute/tests/unit/stream-utilities.test.ts

import test from "node:test";
import assert from "node:assert/strict";

import {
  pipeWithDisconnect,
  createStreamController,
  createDisconnectAwareStream,
} from "../../open-sse/utils/streamHandler.ts";
import { createPassthroughStreamWithLogger } from "../../open-sse/utils/stream.ts";

import { wantsProgress, createProgressTransform } from "../../open-sse/utils/progressTracker.ts";

test("wantsProgress detects X-OmniRoute-Progress header correctly", () => {
  const headersObj = { "x-omniroute-progress": "true" };
  assert.equal(wantsProgress(new Headers(headersObj)), true);

  const headersMap = new Map([["x-omniroute-progress", "true"]]);
  assert.equal(wantsProgress(headersMap), true);

  const headersPlain = { "x-omniroute-progress": "true" };
  assert.equal(wantsProgress(headersPlain), true);

  assert.equal(wantsProgress({ "x-omniroute-progress": "false" }), false);
  assert.equal(wantsProgress(null), false);
  assert.equal(wantsProgress({}), false);
});

test("createProgressTransform maps SSE text output to valid byte stream with progress", async () => {
  const transform = createProgressTransform({ signal: new AbortController().signal });
  const writer = transform.writable.getWriter();

  writer.write(new TextEncoder().encode('data: {"chunk":"one"}\n\n'));
  writer.write(new TextEncoder().encode('data: {"chunk":"two"}\n\n'));
  writer.write(new TextEncoder().encode("data: [DONE]\n\n"));
  writer.close();

  const reader = transform.readable.getReader();
  const decoder = new TextDecoder();
  let result = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    result += decoder.decode(value);
  }

  assert.match(result, /data: \{"chunk":"one"\}/);
  assert.match(result, /data: \{"chunk":"two"\}/);
  assert.match(result, /data: \[DONE\]/);
  assert.match(result, /event: progress/); // progress check
  assert.match(result, /done":true/);
});

test("createPassthroughStreamWithLogger omits [DONE] for Responses clients", async () => {
  const transform = createPassthroughStreamWithLogger(
    "codex",
    null,
    null,
    "gpt-5.5-low",
    null,
    null,
    null,
    null,
    null,
    "openai-responses"
  );

  const writer = transform.writable.getWriter();
  await writer.write(
    new TextEncoder().encode(
      [
        "event: response.completed",
        'data: {"type":"response.completed","response":{"id":"resp_1","model":"gpt-5.5-low","status":"completed","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}',
        "",
      ].join("\n")
    )
  );
  await writer.close();

  const reader = transform.readable.getReader();
  const decoder = new TextDecoder();
  let result = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    result += decoder.decode(value);
  }

  assert.match(result, /event: response\.completed/);
  assert.doesNotMatch(result, /data: \[DONE\]/);
});

test("createPassthroughStreamWithLogger synthesizes reasoning summary events from reasoning output items", async () => {
  const transform = createPassthroughStreamWithLogger(
    "codex",
    null,
    null,
    "gpt-5.5-low",
    null,
    null,
    null,
    null,
    null,
    "openai-responses"
  );

  const writer = transform.writable.getWriter();
  await writer.write(
    new TextEncoder().encode(
      [
        "event: response.output_item.done",
        'data: {"type":"response.output_item.done","response_id":"resp_reasoning_1","output_index":0,"item":{"id":"rs_resp_reasoning_1_0","type":"reasoning","summary":[{"type":"summary_text","text":"Reasoning summary text"}]}}',
        "",
      ].join("\n")
    )
  );
  await writer.close();

  const reader = transform.readable.getReader();
  const decoder = new TextDecoder();
  let result = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    result += decoder.decode(value);
  }

  assert.match(result, /event: response\.reasoning_summary_text\.delta/);
  assert.match(result, /"delta":"Reasoning summary text"/);
  assert.match(result, /event: response\.reasoning_summary_part\.done/);
  assert.match(result, /event: response\.output_item\.done/);
});

test("createPassthroughStreamWithLogger shows a placeholder for encrypted reasoning items", async () => {
  const transform = createPassthroughStreamWithLogger(
    "codex",
    null,
    null,
    "gpt-5.5-low",
    null,
    null,
    null,
    null,
    null,
    "openai-responses"
  );

  const writer = transform.writable.getWriter();
  await writer.write(
    new TextEncoder().encode(
      [
        "event: response.output_item.done",
        'data: {"type":"response.output_item.done","response_id":"resp_reasoning_2","output_index":0,"item":{"id":"rs_resp_reasoning_2_0","type":"reasoning","encrypted_content":"enc_opaque_state"}}',
        "",
      ].join("\n")
    )
  );
  await writer.close();

  const reader = transform.readable.getReader();
  const decoder = new TextDecoder();
  let result = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    result += decoder.decode(value);
  }

  assert.match(result, /event: response\.reasoning_summary_text\.delta/);
  assert.match(result, /Codex is reasoning/);
  assert.match(result, /"summary":\[\{"type":"summary_text","text":"Codex is reasoning/);
  assert.match(result, /event: response\.output_item\.done/);
});

test("createPassthroughStreamWithLogger backfills completed output from function_call arguments events", async () => {
  const transform = createPassthroughStreamWithLogger(
    "codex",
    null,
    null,
    "gpt-5.5-low",
    null,
    null,
    null,
    null,
    null,
    "openai-responses"
  );

  const writer = transform.writable.getWriter();
  await writer.write(
    new TextEncoder().encode(
      [
        'data: {"type":"response.created","response":{"id":"resp_fc_1"}}',
        "event: response.output_item.added",
        'data: {"type":"response.output_item.added","response_id":"resp_fc_1","output_index":0,"item":{"id":"fc_call_1","type":"function_call","call_id":"call_1","name":"workspace_read_file","arguments":""}}',
        "event: response.function_call_arguments.done",
        'data: {"type":"response.function_call_arguments.done","response_id":"resp_fc_1","output_index":0,"item_id":"fc_call_1","arguments":"{\\"path\\":\\"README.md\\"}"}',
        "event: response.completed",
        'data: {"type":"response.completed","response":{"id":"resp_fc_1","status":"completed","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}',
        "",
      ].join("\n")
    )
  );
  await writer.close();

  const reader = transform.readable.getReader();
  const decoder = new TextDecoder();
  let result = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    result += decoder.decode(value);
  }

  assert.match(result, /event: response\.completed/);
  assert.match(result, /"type":"function_call"/);
  assert.match(result, /"call_id":"call_1"/);
  assert.ok(result.includes('"arguments":"{\\"path\\":\\"README.md\\"}"'));
});

test("createPassthroughStreamWithLogger keeps reasoning deltas out of logged assistant content", async () => {
  let completePayload = null;
  const transform = createPassthroughStreamWithLogger(
    "codex",
    null,
    null,
    "gpt-5.5-low",
    null,
    {
      input: [
        {
          type: "message",
          role: "user",
          content: [{ type: "input_text", text: "Say OK." }],
        },
      ],
    },
    (payload) => {
      completePayload = payload;
    },
    null,
    null,
    "openai-responses"
  );

  const writer = transform.writable.getWriter();
  await writer.write(
    new TextEncoder().encode(
      [
        'data: {"type":"response.created","response":{"id":"resp_reasoning_delta_1"}}',
        "event: response.reasoning_summary_text.delta",
        'data: {"type":"response.reasoning_summary_text.delta","response_id":"resp_reasoning_delta_1","delta":"Internal reasoning should not be content."}',
        "event: response.function_call_arguments.delta",
        'data: {"type":"response.function_call_arguments.delta","response_id":"resp_reasoning_delta_1","delta":"{\\"path\\":\\"secret.txt\\"}"}',
        "event: response.output_text.delta",
        'data: {"type":"response.output_text.delta","response_id":"resp_reasoning_delta_1","delta":"OK"}',
        "event: response.completed",
        'data: {"type":"response.completed","response":{"id":"resp_reasoning_delta_1","status":"completed","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}',
        "",
      ].join("\n")
    )
  );
  await writer.close();

  const reader = transform.readable.getReader();
  while (true) {
    const { done } = await reader.read();
    if (done) break;
  }

  assert.ok(completePayload, "expected onComplete payload");
  assert.equal(completePayload.responseBody.choices[0].message.content, "OK");
  assert.equal(
    JSON.stringify(completePayload.responseBody).includes(
      "Internal reasoning should not be content"
    ),
    false
  );
  assert.equal(JSON.stringify(completePayload.responseBody).includes("secret.txt"), false);
});

test("createStreamController returns valid controller", () => {
  let completeLogged = false;
  let disconnectLogged = false;

  const originalLog = console.log;
  console.log = (msg) => {
    if (msg.includes("complete")) completeLogged = true;
    if (msg.includes("disconnect")) disconnectLogged = true;
  };

  const sc = createStreamController({
    provider: "test",
    model: "conn_1",
  });

  assert.equal(typeof sc.signal, "object");

  sc.handleComplete();
  assert.equal(completeLogged, true);
  assert.equal(sc.isConnected(), false);

  sc.handleDisconnect();
  assert.equal(disconnectLogged, false);

  console.log = originalLog;
});