import test from "node:test"; import assert from "node:assert/strict"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-stream-utils-")); process.env.DATA_DIR = TEST_DATA_DIR; const core = await import("../../src/lib/db/core.ts"); const { createSSEStream, createSSETransformStreamWithLogger, createPassthroughStreamWithLogger } = await import("../../open-sse/utils/stream.ts"); const { buildStreamSummaryFromEvents, compactStructuredStreamPayload, createStructuredSSECollector, } = await import("../../open-sse/utils/streamPayloadCollector.ts"); const { FORMATS } = await import("../../open-sse/translator/formats.ts"); const { createRequestLogger } = await import("../../open-sse/utils/requestLogger.ts"); // Retained stream chunks are prefixed with a fixed-width per-chunk arrival timestamp // ("[HH:MM:SS.mmm] ", 15 chars) by the request logger for streaming-latency // observability (added in #5834). Strip it before comparing raw chunk payloads. const STREAM_CHUNK_TS_PREFIX_LEN = "[HH:MM:SS.mmm] ".length; const stripChunkTs = (chunk: string): string => chunk.replace(/^\[\d{2}:\d{2}:\d{2}\.\d{3}\] /, ""); const textEncoder = new TextEncoder(); // PR #3399 intentionally changed the synthetic empty-response text to "" so that // proxy internals no longer leak into chat history. Tests assert on the new behavior. const SYNTHETIC_CLAUDE_EMPTY_RESPONSE_TEXT = ""; async function readTransformed(chunks, options) { const source = new ReadableStream({ start(controller) { for (const chunk of chunks) { controller.enqueue(textEncoder.encode(chunk)); } controller.close(); }, }); return new Response(source.pipeThrough(createSSEStream(options))).text(); } async function readWithTransform(chunks, transformStream) { const source = new ReadableStream({ start(controller) { for (const chunk of chunks) { controller.enqueue(textEncoder.encode(chunk)); } controller.close(); }, }); return new Response(source.pipeThrough(transformStream)).text(); } function multilineDataEvent(payload, splitBeforeKey) { const json = JSON.stringify(payload); const splitAt = json.indexOf(`"${splitBeforeKey}"`) - 1; assert.ok(splitAt > 0, `split key ${splitBeforeKey} must exist in payload`); return `data: ${json.slice(0, splitAt)}\ndata: ${json.slice(splitAt)}\n\n`; } function parseJsonDataPayloads(text) { return text .split("\n") .filter((line) => line.startsWith("data: ")) .map((line) => line.slice(6)) .filter((data) => data && data !== "[DONE]") .map((data) => JSON.parse(data)); } function parseSillyTavernCustomOpenAIStream(text) { const events = []; let reasoning = ""; let content = ""; for (const payload of parseJsonDataPayloads(text)) { const choices = Array.isArray(payload.choices) ? payload.choices : []; const reasoningDelta = choices.find((choice) => choice?.delta?.reasoning_content)?.delta?.reasoning_content ?? choices.find((choice) => choice?.delta?.reasoning)?.delta?.reasoning ?? ""; const contentDelta = choices[0]?.delta?.content ?? choices[0]?.message?.content ?? choices[0]?.text ?? ""; reasoning += reasoningDelta; content += contentDelta; events.push({ reasoningDelta, contentDelta, reasoning, content }); } return { reasoning, content, events }; } test("createSSEStream leaves successful pending requests for onComplete finalization", async () => { const usageHistory = await import("../../src/lib/usage/usageHistory.ts"); usageHistory.clearPendingRequests(); const model = "gpt-4"; const provider = "openai"; const connectionId = "stream-on-complete-finalize"; const requestId = usageHistory.trackPendingRequest(model, provider, connectionId, true); let finalizedInOnComplete = false; await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_pending_finalize", object: "chat.completion.chunk", created: 1, model, choices: [{ index: 0, delta: { content: "hello" }, finish_reason: null }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_pending_finalize", object: "chat.completion.chunk", created: 1, model, choices: [{ index: 0, delta: {}, finish_reason: "stop" }], usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 }, })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider, model, connectionId, body: { messages: [{ role: "user", content: "hello" }] }, onComplete(payload) { finalizedInOnComplete = usageHistory.finalizePendingRequestById(requestId, { providerResponse: payload.providerPayload, clientResponse: payload.clientPayload, }); }, } ); assert.equal(finalizedInOnComplete, true); assert.ok(usageHistory.getCompletedDetails().has(requestId)); assert.equal(usageHistory.getPendingById().has(requestId), false); usageHistory.clearPendingRequests(); }); test.after(() => { core.resetDbInstance(); if (fs.existsSync(TEST_DATA_DIR)) { for (const entry of fs.readdirSync(TEST_DATA_DIR)) { fs.rmSync(path.join(TEST_DATA_DIR, entry), { recursive: true, force: true }); } } }); test("createSSEStream passthrough normalizes tool-call finishes and reports the assembled response", async () => { let onCompletePayload = null; const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_1", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [{ index: 0, delta: { role: "assistant", content: "Hello " } }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_1", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [ { index: 0, delta: { tool_calls: [ { index: 0, id: "call_1", type: "function", function: { name: "read_file", arguments: '{"path":"/tmp/a"}', }, }, ], }, }, ], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_1", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "openai", model: "gpt-4.1-mini", body: { messages: [{ role: "user", content: "hello" }], }, onComplete(payload) { onCompletePayload = payload; }, } ); assert.match(text, /"content":"Hello "/); assert.match(text, /"name":"read_file"/); assert.match(text, /"finish_reason":"tool_calls"/); assert.equal(onCompletePayload.status, 200); assert.equal(onCompletePayload.responseBody.choices[0].finish_reason, "tool_calls"); assert.equal(onCompletePayload.responseBody.choices[0].message.tool_calls[0].id, "call_1"); assert.equal(onCompletePayload.responseBody.choices[0].message.content, "Hello"); assert.equal(onCompletePayload.clientPayload._streamed, true); }); test("createSSEStream passthrough converts textual tool-call content into structured call log tool_calls", async () => { let onCompletePayload = null; const toolArgs = JSON.stringify({ command: 'sqlite3 /root/.o\u200dmniroute/omniroute.db ".tables"', }); const toolText = `[Tool call: terminal]\nArguments: ${toolArgs}`; const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_textual_tool", object: "chat.completion.chunk", created: 1, model: "antigravity/gemini-3.5-flash-low", choices: [{ index: 0, delta: { role: "assistant", content: toolText } }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_textual_tool", object: "chat.completion.chunk", created: 1, model: "antigravity/gemini-3.5-flash-low", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "antigravity", model: "antigravity/gemini-3.5-flash-low", body: { messages: [{ role: "user", content: "inspect db" }], }, onComplete(payload) { onCompletePayload = payload; }, } ); assert.equal(onCompletePayload.status, 200); assert.match(text, /"tool_calls":\[/); assert.match(text, /"name":"terminal"/); assert.doesNotMatch(text, /"content":"\[Tool call: terminal/); const choice = onCompletePayload.responseBody.choices[0]; assert.equal(choice.finish_reason, "tool_calls"); assert.equal(choice.message.content, null); assert.equal(choice.message.tool_calls[0].function.name, "terminal"); assert.deepEqual(JSON.parse(choice.message.tool_calls[0].function.arguments), { command: 'sqlite3 /root/.omniroute/omniroute.db ".tables"', }); assert.doesNotMatch(text, /\[Tool call: terminal\]/); }); test("createSSEStream passthrough converts split textual tool-call content at completion", async () => { let onCompletePayload = null; const splitToolArgs = JSON.stringify({ command: 'sqlite3 ~/.o\u200dmniroute/o\u200dmniroute.db ".tables"', }); const chunks = ["[Tool call: terminal]\n", `Arguments: ${splitToolArgs}`]; const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_split_textual_tool", object: "chat.completion.chunk", created: 1, model: "antigravity/gemini-3.5-flash-low", choices: [{ index: 0, delta: { role: "assistant", content: chunks[0] } }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_split_textual_tool", object: "chat.completion.chunk", created: 1, model: "antigravity/gemini-3.5-flash-low", choices: [{ index: 0, delta: { content: chunks[1] } }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_split_textual_tool", object: "chat.completion.chunk", created: 1, model: "antigravity/gemini-3.5-flash-low", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "antigravity", model: "antigravity/gemini-3.5-flash-low", body: { messages: [{ role: "user", content: "inspect db" }] }, onComplete(payload) { onCompletePayload = payload; }, } ); assert.match(text, /"tool_calls":\[/); assert.match(text, /"name":"terminal"/); assert.doesNotMatch(text, /"content":"\[Tool call: terminal/); const choice = onCompletePayload.responseBody.choices[0]; assert.equal(choice.finish_reason, "tool_calls"); assert.equal(choice.message.content, null); assert.equal(choice.message.tool_calls[0].function.name, "terminal"); assert.deepEqual(JSON.parse(choice.message.tool_calls[0].function.arguments), { command: 'sqlite3 ~/.omniroute/omniroute.db ".tables"', }); assert.doesNotMatch(text, /\[Tool call: terminal\]/); }); test("createSSEStream passthrough handles textual tool-call content split inside the prefix [Tool call: across chunks", async () => { let onCompletePayload = null; const splitToolArgs = JSON.stringify({ command: "whoami", }); const chunks = ["[Tool", " call: terminal]\n", `Arguments: ${splitToolArgs}`]; const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_split_prefix_textual_tool", object: "chat.completion.chunk", created: 1, model: "antigravity/gemini-3.5-flash-low", choices: [{ index: 0, delta: { role: "assistant", content: chunks[0] } }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_split_prefix_textual_tool", object: "chat.completion.chunk", created: 1, model: "antigravity/gemini-3.5-flash-low", choices: [{ index: 0, delta: { content: chunks[1] } }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_split_prefix_textual_tool", object: "chat.completion.chunk", created: 1, model: "antigravity/gemini-3.5-flash-low", choices: [{ index: 0, delta: { content: chunks[2] } }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_split_prefix_textual_tool", object: "chat.completion.chunk", created: 1, model: "antigravity/gemini-3.5-flash-low", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "antigravity", model: "antigravity/gemini-3.5-flash-low", body: { messages: [{ role: "user", content: "inspect db" }] }, onComplete(payload) { onCompletePayload = payload; }, } ); assert.doesNotMatch(text, /"content":"\[Tool/); assert.doesNotMatch(text, /"content":" call:/); assert.match(text, /"tool_calls":\[/); assert.match(text, /"name":"terminal"/); const choice = onCompletePayload.responseBody.choices[0]; assert.equal(choice.finish_reason, "tool_calls"); assert.equal(choice.message.content, null); assert.equal(choice.message.tool_calls[0].function.name, "terminal"); assert.deepEqual(JSON.parse(choice.message.tool_calls[0].function.arguments), { command: "whoami", }); }); test("createSSEStream passthrough buffers fragmented textual tool-call JSON before emitting", async () => { let onCompletePayload = null; const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_fragmented_live_shape", object: "chat.completion.chunk", created: 1, model: "MainAgent", choices: [ { index: 0, delta: { role: "assistant", content: '[Tool call: terminal]\nArguments: {"' }, }, ], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_fragmented_live_shape", object: "chat.completion.chunk", created: 1, model: "MainAgent", choices: [ { index: 0, delta: { content: 'command":"echo live_shape","timeout":10}' }, }, ], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_fragmented_live_shape", object: "chat.completion.chunk", created: 1, model: "MainAgent", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "omniroute", model: "MainAgent", body: { messages: [{ role: "user", content: "inspect" }] }, onComplete(payload) { onCompletePayload = payload; }, } ); assert.doesNotMatch(text, /\[Tool call:/); assert.doesNotMatch(text, /Arguments:/); assert.match(text, /"tool_calls":\[/); assert.match(text, /"finish_reason":"tool_calls"/); const choice = onCompletePayload.responseBody.choices[0]; assert.equal(choice.finish_reason, "tool_calls"); assert.equal(choice.message.content, null); assert.equal(choice.message.tool_calls[0].function.name, "terminal"); assert.deepEqual(JSON.parse(choice.message.tool_calls[0].function.arguments), { command: "echo live_shape", timeout: 10, }); }); test("createSSEStream passthrough suppresses trailing prose plus textual tool call", async () => { let onCompletePayload = null; const toolArgs = JSON.stringify({ command: "echo should_not_leak", timeout: 10, }); const toolText = `Вот оно! Статические файлы Next.js отдают 404. Чанки не найдены.\n\n[Tool call: terminal]\nArguments: ${toolArgs}`; const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_trailing_textual_tool", object: "chat.completion.chunk", created: 1, model: "MainAgent", choices: [{ index: 0, delta: { role: "assistant", content: toolText } }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_trailing_textual_tool", object: "chat.completion.chunk", created: 1, model: "MainAgent", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "omniroute", model: "MainAgent", body: { messages: [{ role: "user", content: "inspect static files" }] }, onComplete(payload) { onCompletePayload = payload; }, } ); assert.equal(onCompletePayload.status, 200); const choice = onCompletePayload.responseBody.choices[0]; assert.equal(choice.finish_reason, "tool_calls"); assert.equal(choice.message.content, null); assert.equal(choice.message.tool_calls[0].function.name, "terminal"); assert.deepEqual(JSON.parse(choice.message.tool_calls[0].function.arguments), { command: "echo should_not_leak", timeout: 10, }); assert.doesNotMatch(text, /\[Tool call: terminal\]/); assert.doesNotMatch(text, /Arguments:/); assert.doesNotMatch(JSON.stringify(onCompletePayload.responseBody), /\[Tool call: terminal\]/); assert.doesNotMatch(JSON.stringify(onCompletePayload.responseBody), /Arguments:/); }); test("createSSEStream passthrough suppresses textual tool calls for unknown tools", async () => { let onCompletePayload = null; const toolText = `[Tool call: search_files_ide] Arguments: {"path":"/opt/OmniRoute/src","target":"files"}`; const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_unknown_textual_tool", object: "chat.completion.chunk", created: 1, model: "antigravity/gemini-3.5-flash-low", choices: [{ index: 0, delta: { role: "assistant", content: toolText } }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_unknown_textual_tool", object: "chat.completion.chunk", created: 1, model: "antigravity/gemini-3.5-flash-low", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "antigravity", model: "antigravity/gemini-3.5-flash-low", body: { messages: [{ role: "user", content: "inspect files" }], tools: [ { type: "function", function: { name: "search_files", parameters: { type: "object" } } }, ], }, onComplete(payload) { onCompletePayload = payload; }, } ); const choice = onCompletePayload.responseBody.choices[0]; assert.equal(choice.finish_reason, "stop"); assert.equal(choice.message.content, null); assert.equal(choice.message.tool_calls, undefined); assert.doesNotMatch(text, /search_files_ide/); assert.doesNotMatch(JSON.stringify(onCompletePayload.responseBody), /search_files_ide/); }); test("createSSEStream passthrough suppresses malformed textual tool-call content", async () => { let onCompletePayload = null; const malformedToolText = `(empty)[Tool call: terminal]\nArguments: {"command":"sqlite3 /opt/O\u200dmniRoute/data/o\u200dmniroute.`; const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_malformed_textual_tool", object: "chat.completion.chunk", created: 1, model: "antigravity/gemini-3.5-flash-low", choices: [{ index: 0, delta: { role: "assistant", content: malformedToolText } }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_malformed_textual_tool", object: "chat.completion.chunk", created: 1, model: "antigravity/gemini-3.5-flash-low", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "antigravity", model: "antigravity/gemini-3.5-flash-low", body: { messages: [{ role: "user", content: "inspect db" }] }, onComplete(payload) { onCompletePayload = payload; }, } ); const choice = onCompletePayload.responseBody.choices[0]; assert.equal(choice.finish_reason, "stop"); assert.equal(choice.message.content, null); assert.equal(choice.message.tool_calls, undefined); // PR #3355 bug-2 fix: flush now always emits the buffer as plain text (not swallowed). assert.match(text, /\[Tool call: terminal\]/); }); test("createSSEStream suppresses malformed compact textual tool-call content", async () => { let onCompletePayload = null; const text = await readTransformed( [ `data: ${JSON.stringify({ candidates: [ { content: { parts: [ { text: "[Tool call: search_files_ide{file_glob:*combos*.ts,path:/opt/OmniRoute,target:files}]", }, ], }, }, ], })}\n\n`, `data: ${JSON.stringify({ candidates: [{ finishReason: "STOP" }] })}\n\n`, ], { mode: "translate", targetFormat: FORMATS.ANTIGRAVITY, sourceFormat: FORMATS.OPENAI, provider: "antigravity", model: "antigravity/gemini-3.5-flash-low", body: { messages: [{ role: "user", content: "inspect files" }] }, onComplete(payload) { onCompletePayload = payload; }, } ); const choice = onCompletePayload.responseBody.choices[0]; assert.equal(choice.finish_reason, "stop"); assert.equal(choice.message.content, null); assert.equal(choice.message.tool_calls, undefined); assert.doesNotMatch(JSON.stringify(onCompletePayload.responseBody), /\[Tool call:/); }); test("createSSEStream passthrough flushes a buffered final line without a trailing newline", async () => { const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_2", object: "chat.completion.chunk", created: 2, model: "gpt-4.1-mini", choices: [{ index: 0, delta: { role: "assistant", content: "tail chunk" } }], })}`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "openai", model: "gpt-4.1-mini", body: { messages: [{ role: "user", content: "hello" }], }, } ); assert.match(text, /tail chunk/); assert.equal(text.includes("data: "), true); }); test("createSSEStream passthrough merges multi-line SSE data before forwarding", async () => { const text = await readTransformed( [ multilineDataEvent( { id: "chatcmpl_multiline", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [{ index: 0, delta: { content: "hello" }, finish_reason: null }], }, "choices" ), `data: [DONE]\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "openai", model: "gpt-4.1-mini", body: { messages: [{ role: "user", content: "hello" }], }, } ); assert.match(text, /"content":"hello"/); assert.doesNotMatch(text, /\ndata:\s*,/); const firstPayload = text .split("\n") .find((line) => line.startsWith("data: ") && !line.includes("[DONE]")); assert.ok(firstPayload, "merged SSE payload should be forwarded as a single JSON data line"); assert.equal(JSON.parse(firstPayload.slice(6)).choices[0].delta.content, "hello"); }); test("createSSEStream passthrough forwards data only after the complete SSE event boundary", async () => { const text = await readTransformed( [ [ `data: ${JSON.stringify({ delta: "hello" })}`, "event: response.output_text.delta", "", "", ].join("\n"), ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "openai", model: "responses-model", body: { messages: [{ role: "user", content: "hello" }], }, } ); assert.match(text, /^event: response\.output_text\.delta\ndata: /); assert.doesNotMatch(text, /^data: .*?\n\nevent:/s); }); test("createSSEStream passthrough preserves event metadata in a single SSE event", async () => { const text = await readTransformed( [ [ ": upstream-note", "id: 42", "trace: upstream-abc", `data: ${JSON.stringify({ id: "chatcmpl_meta", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [{ index: 0, delta: { content: "metadata content" } }], })}`, "", "", ].join("\n"), ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "openai", model: "gpt-4.1-mini", body: { messages: [{ role: "user", content: "hello" }], }, } ); assert.match(text, /^: upstream-note\nid: 42\ntrace: upstream-abc\ndata: /); assert.doesNotMatch(text, /^: upstream-note\n\nid: 42/s); assert.doesNotMatch(text, /\ntrace: upstream-abc\n\n/s); assert.match(text, /metadata content/); }); test("createSSEStream translate mode converts Claude SSE into OpenAI chunks and completion payload", async () => { let onCompletePayload = null; const text = await readTransformed( [ `data: ${JSON.stringify({ type: "message_start", message: { id: "msg_1", model: "claude-sonnet-4", role: "assistant", usage: { input_tokens: 3 }, }, })}\n\n`, `data: ${JSON.stringify({ type: "content_block_start", index: 0, content_block: { type: "text", text: "" }, })}\n\n`, `data: ${JSON.stringify({ type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "Hello Claude" }, })}\n\n`, `data: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 4 }, })}\n\n`, `data: ${JSON.stringify({ type: "message_stop", })}\n\n`, ], { mode: "translate", targetFormat: FORMATS.CLAUDE, sourceFormat: FORMATS.OPENAI, provider: "claude", model: "claude-sonnet-4", body: { messages: [{ role: "user", content: "hello" }], }, onComplete(payload) { onCompletePayload = payload; }, } ); assert.match(text, /"content":"Hello Claude"/); assert.match(text, /\[DONE\]/); assert.equal(onCompletePayload.status, 200); assert.equal(onCompletePayload.responseBody.choices[0].message.content, "Hello Claude"); assert.equal(onCompletePayload.responseBody.usage.prompt_tokens, 3); assert.equal(onCompletePayload.responseBody.usage.completion_tokens, 4); assert.equal(onCompletePayload.responseBody.usage.total_tokens, 7); }); test("createSSEStream translate mode preserves Claude text_delta thinking tags as content", async () => { let onCompletePayload = null; const text = await readTransformed( [ `data: ${JSON.stringify({ type: "message_start", message: { id: "msg_text_thinking_tag", model: "claude-opus-4-6", role: "assistant", usage: { input_tokens: 3 }, }, })}\n\n`, `data: ${JSON.stringify({ type: "content_block_start", index: 0, content_block: { type: "text", text: "" }, })}\n\n`, `data: ${JSON.stringify({ type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "\n[metacognition" }, })}\n\n`, `data: ${JSON.stringify({ type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "]\n\nVisible answer" }, })}\n\n`, `data: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 4 }, })}\n\n`, `data: ${JSON.stringify({ type: "message_stop", })}\n\n`, ], { mode: "translate", targetFormat: FORMATS.CLAUDE, sourceFormat: FORMATS.OPENAI, provider: "claude", model: "claude-opus-4-6", body: { messages: [{ role: "user", content: "hello" }], }, onComplete(payload) { onCompletePayload = payload; }, } ); const deltas = parseJsonDataPayloads(text) .map((payload) => payload.choices?.[0]?.delta) .filter(Boolean); assert.deepEqual( deltas.filter((delta) => typeof delta.content === "string").map((delta) => delta.content), ["\n[metacognition", "]\n\nVisible answer"] ); assert.equal( deltas.some((delta) => delta.reasoning_content !== undefined), false ); assert.equal( onCompletePayload.responseBody.choices[0].message.content, "\n[metacognition]\n\nVisible answer" ); }); test("createSSEStream translate mode keeps native Claude thinking_delta as reasoning_content", async () => { const text = await readTransformed( [ `data: ${JSON.stringify({ type: "message_start", message: { id: "msg_native_thinking", model: "claude-opus-4-6", role: "assistant", usage: { input_tokens: 3 }, }, })}\n\n`, `data: ${JSON.stringify({ type: "content_block_start", index: 0, content_block: { type: "thinking", thinking: "" }, })}\n\n`, `data: ${JSON.stringify({ type: "content_block_delta", index: 0, delta: { type: "thinking_delta", thinking: "Native plan" }, })}\n\n`, `data: ${JSON.stringify({ type: "content_block_stop", index: 0, })}\n\n`, `data: ${JSON.stringify({ type: "content_block_start", index: 1, content_block: { type: "text", text: "" }, })}\n\n`, `data: ${JSON.stringify({ type: "content_block_delta", index: 1, delta: { type: "text_delta", text: "Final answer" }, })}\n\n`, `data: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 4 }, })}\n\n`, `data: ${JSON.stringify({ type: "message_stop", })}\n\n`, ], { mode: "translate", targetFormat: FORMATS.CLAUDE, sourceFormat: FORMATS.OPENAI, provider: "claude", model: "claude-opus-4-6", body: { messages: [{ role: "user", content: "hello" }], }, } ); const deltas = parseJsonDataPayloads(text) .map((payload) => payload.choices?.[0]?.delta) .filter(Boolean); assert.ok(deltas.some((delta) => delta.reasoning_content === "Native plan")); assert.ok(deltas.some((delta) => delta.content === "Final answer")); }); test("createSSEStream translate mode parses multi-line SSE data events", async () => { let onCompletePayload = null; const text = await readTransformed( [ `data: ${JSON.stringify({ type: "message_start", message: { id: "msg_multiline", model: "claude-sonnet-4", role: "assistant", usage: { input_tokens: 3 }, }, })}\n\n`, `data: ${JSON.stringify({ type: "content_block_start", index: 0, content_block: { type: "text", text: "" }, })}\n\n`, multilineDataEvent( { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "Hello from multiline SSE" }, }, "delta" ), `data: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 5 }, })}\n\n`, `data: ${JSON.stringify({ type: "message_stop", })}\n\n`, ], { mode: "translate", targetFormat: FORMATS.CLAUDE, sourceFormat: FORMATS.OPENAI, provider: "claude", model: "claude-sonnet-4", body: { messages: [{ role: "user", content: "hello" }], }, onComplete(payload) { onCompletePayload = payload; }, } ); assert.match(text, /"content":"Hello from multiline SSE"/); assert.equal( onCompletePayload.responseBody.choices[0].message.content, "Hello from multiline SSE" ); }); test("createSSEStream Responses passthrough converts textual tool-call deltas before streaming", async () => { let onCompletePayload = null; const toolText = `[Tool call: terminal] Arguments: {"command":"systemctl status omniroute"}`; const text = await readTransformed( [ `data: ${JSON.stringify({ type: "response.output_text.delta", delta: toolText, })} `, `data: ${JSON.stringify({ type: "response.completed", response: { id: "resp_textual_tool", object: "response", model: "antigravity/gemini-3.5-flash-low", status: "completed", output: [], usage: { input_tokens: 10, output_tokens: 4, total_tokens: 14 }, }, })} `, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI_RESPONSES, clientResponseFormat: FORMATS.OPENAI_RESPONSES, provider: "antigravity", model: "antigravity/gemini-3.5-flash-low", body: { input: "check service", tools: [{ type: "function", name: "terminal", parameters: { type: "object" } }], }, onComplete(payload) { onCompletePayload = payload; }, } ); assert.doesNotMatch(text, /\[Tool call: terminal\]/); assert.doesNotMatch(text, /Arguments:/); assert.match(text, /response.output_item.added/); assert.match(text, /response.function_call_arguments.done/); assert.match(text, /"name":"terminal"/); assert.equal(onCompletePayload.responseBody.choices[0].finish_reason, "tool_calls"); assert.equal(onCompletePayload.responseBody.choices[0].message.content, null); assert.equal( onCompletePayload.responseBody.choices[0].message.tool_calls[0].function.name, "terminal" ); assert.doesNotMatch(JSON.stringify(onCompletePayload.clientPayload), /\[Tool call: terminal\]/); }); test("createSSEStream passthrough preserves Responses API events and completion summaries", async () => { let onCompletePayload = null; const text = await readTransformed( [ `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "Hello ", })}\n\n`, `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "world", })}\n\n`, `data: ${JSON.stringify({ type: "response.completed", response: { id: "resp_1", object: "response", model: "gpt-4.1-mini", status: "completed", usage: { input_tokens: 2, output_tokens: 3, total_tokens: 5 }, }, })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI_RESPONSES, provider: "openai", model: "gpt-4.1-mini", body: { input: "hello" }, onComplete(payload) { onCompletePayload = payload; }, } ); assert.match(text, /response.output_text.delta/); assert.match(text, /response.completed/); assert.equal(onCompletePayload.responseBody.usage.total_tokens, 5); assert.equal(onCompletePayload.providerPayload.summary.object, "response"); }); test("createSSEStream passthrough drops leaked empty chat bootstrap chunks for Responses clients", async () => { const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl-dummy", object: "chat.completion.chunk", created: 1, model: "gpt-5.4", choices: [ { index: 0, delta: { role: "assistant", content: null, refusal: null }, finish_reason: null, }, ], })}\n\n`, `event: response.created\ndata: ${JSON.stringify({ type: "response.created", response: { id: "resp_1", object: "response", model: "gpt-5.4", status: "in_progress", output: [], }, })}\n\n`, `event: response.in_progress\ndata: ${JSON.stringify({ type: "response.in_progress", response: { id: "resp_1", object: "response", model: "gpt-5.4", status: "in_progress", output: [], }, })}\n\n`, `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "OK", })}\n\n`, `data: ${JSON.stringify({ type: "response.completed", response: { id: "resp_1", object: "response", model: "gpt-5.4", status: "completed", output: [], usage: { input_tokens: 2, output_tokens: 1, total_tokens: 3 }, }, })}\n\n`, `data: [DONE]\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI_RESPONSES, clientResponseFormat: FORMATS.OPENAI_RESPONSES, provider: "openai", model: "gpt-5.4", body: { input: "hello" }, } ); assert.doesNotMatch(text, /chatcmpl-dummy/); assert.match(text, /response\.created/); assert.match(text, /response\.output_text\.delta/); assert.match(text, /"delta":"OK"/); }); test("buildStreamSummaryFromEvents falls back to response.output_text.delta when completed output is empty", () => { const summary = buildStreamSummaryFromEvents( [ { index: 0, data: { type: "response.output_text.delta", delta: "Hello ", }, }, { index: 1, data: { type: "response.output_text.delta", delta: "world", }, }, { index: 2, data: { type: "response.completed", response: { id: "resp_fallback", object: "response", model: "gpt-5.4", status: "completed", output: [], usage: { output_tokens: 2 }, }, }, }, ], FORMATS.OPENAI_RESPONSES, "gpt-5.4" ); assert.equal((summary as any).object, "response"); assert.equal((summary as any).output[0].type, "message"); assert.equal((summary as any).output[0].content[0].type, "output_text"); assert.equal((summary as any).output[0].content[0].text, "Hello world"); assert.equal((summary as any).usage.output_tokens, 2); }); test("createSSEStream translate mode aborts on Responses failure with rate limit error", async () => { let onCompletePayload = null; await assert.rejects( readTransformed( [ `data: ${JSON.stringify({ type: "response.created", response: { id: "resp_fail", object: "response", model: "gpt-5.4", status: "in_progress", output: [], }, })}\n\n`, `data: ${JSON.stringify({ type: "response.failed", response: { id: "resp_fail", object: "response", model: "gpt-5.4", status: "failed", error: { message: "Rate limit reached for gpt-5.4", code: "rate_limit_exceeded", }, }, })}\n\n`, `data: [DONE]\n\n`, ], { mode: "translate", targetFormat: FORMATS.OPENAI_RESPONSES, sourceFormat: FORMATS.OPENAI, provider: "codex", model: "gpt-5.4", body: { messages: [{ role: "user", content: "hello" }] }, onComplete(payload) { onCompletePayload = payload; }, } ), /Rate limit reached for gpt-5\.4|Upstream failure/ ); assert.ok(onCompletePayload, "should capture completion payload before aborting"); assert.equal(onCompletePayload.status, 429); assert.equal(onCompletePayload.responseBody.error.type, "rate_limit_error"); assert.equal(onCompletePayload.responseBody.error.code, "rate_limit_exceeded"); assert.match(onCompletePayload.responseBody.error.message, /Rate limit reached/); }); test("createSSEStream passthrough restores Claude tool names from the mapping table", async () => { const toolNameMap = new Map([["tool_alias", "read_file"]]); const text = await readTransformed( [ `data: ${JSON.stringify({ type: "content_block_start", index: 0, content_block: { type: "tool_use", id: "tool_1", name: "tool_alias", input: { path: "/tmp/a" }, }, })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.CLAUDE, provider: "claude", model: "claude-sonnet-4", toolNameMap, body: { messages: [{ role: "user", content: "hello" }] }, } ); assert.match(text, /"name":"read_file"/); assert.equal(text.includes("tool_alias"), false); }); test("createSSEStream passthrough fixes generic ids and preserves readable reasoning aliases", async () => { const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chat", object: "chat.completion.chunk", created: 1, model: "kimi-k2.5", choices: [ { index: 0, delta: { reasoning: "Let me think first", }, }, ], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "openai", model: "kimi-k2.5", body: { messages: [{ role: "user", content: "hello" }] }, } ); assert.match(text, /"id":"chatcmpl-/); assert.match(text, /"reasoning":"Let me think first"/); assert.doesNotMatch(text, /"reasoning_content":"Let me think first"/); }); test("createSSEStream passthrough mirrors unsupported reasoning aliases with valid ids", async () => { const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_reasoning_alias", object: "chat.completion.chunk", created: 1, model: "kimi-k2.5", choices: [ { index: 0, delta: { reasoning_text: "Alias-only reasoning", }, }, ], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "openai", model: "kimi-k2.5", body: { messages: [{ role: "user", content: "hello" }] }, } ); assert.match(text, /"reasoning_content":"Alias-only reasoning"/); }); test("createSSEStream passthrough preserves OpenAI content thinking tags as content", async () => { let onCompletePayload = null; const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_text_tag", object: "chat.completion.chunk", created: 1, model: "openai-compatible-model", choices: [ { index: 0, delta: { role: "assistant", content: "\nVisible prompt tag" }, }, ], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_text_tag", object: "chat.completion.chunk", created: 1, model: "openai-compatible-model", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "openai-compatible", model: "openai-compatible-model", body: { messages: [{ role: "user", content: "hello" }] }, onComplete(payload) { onCompletePayload = payload; }, } ); const deltas = parseJsonDataPayloads(text) .map((payload) => payload.choices?.[0]?.delta) .filter(Boolean); assert.ok(deltas.some((delta) => delta.content === "\nVisible prompt tag")); assert.equal( deltas.some((delta) => delta.reasoning_content !== undefined), false ); assert.equal( onCompletePayload.responseBody.choices[0].message.content, "\nVisible prompt tag" ); }); test("createSSEStream passthrough splits mixed reasoning and content deltas and estimates usage", async () => { let onCompletePayload = null; const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_reasoning", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [ { index: 0, delta: { reasoning_content: "First think", content: "Then answer", }, }, ], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_reasoning", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "openai", model: "gpt-4.1-mini", body: { messages: [{ role: "user", content: "hello world" }], }, onComplete(payload) { onCompletePayload = payload; }, } ); const reasoningIndex = text.indexOf('"reasoning_content":"First think"'); const contentIndex = text.indexOf('"content":"Then answer"'); assert.ok(reasoningIndex >= 0); assert.ok(contentIndex > reasoningIndex); assert.match(text, /"total_tokens":\d+/); assert.equal(onCompletePayload.responseBody.choices[0].message.reasoning_content, "First think"); assert.equal(onCompletePayload.responseBody.choices[0].message.content, "Then answer"); assert.ok(onCompletePayload.responseBody.usage.total_tokens > 0); }); test("createSSEStream passthrough output is consumable by SillyTavern-style reasoning parser", async () => { const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_reasoning_st", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [ { index: 0, delta: { reasoning_content: "First think", content: "Then answer", }, }, ], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_reasoning_st", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "openai-compatible", model: "gpt-4.1-mini", body: { messages: [{ role: "user", content: "hello world" }], }, } ); const parsed = parseSillyTavernCustomOpenAIStream(text); const reasoningEventIndex = parsed.events.findIndex((event) => event.reasoningDelta); const contentEventIndex = parsed.events.findIndex((event) => event.contentDelta); assert.equal(parsed.reasoning, "First think"); assert.equal(parsed.content, "Then answer"); assert.ok(reasoningEventIndex >= 0); assert.ok(contentEventIndex > reasoningEventIndex); }); test("createSSEStream passthrough writes complete SSE events per converted chunk", async () => { const convertedChunks = []; await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_reasoning", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [ { index: 0, delta: { reasoning_content: "First think", content: "Then answer", }, }, ], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_reasoning", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "openai", model: "gpt-4.1-mini", body: { messages: [{ role: "user", content: "hello world" }], }, reqLogger: { appendConvertedChunk(value) { convertedChunks.push(value); }, }, } ); assert.equal(convertedChunks.includes("\n"), false); for (const chunk of convertedChunks.filter((value) => value.startsWith("data:"))) { assert.equal(chunk.endsWith("\n\n"), true); } }); test("createSSEStream passthrough merges Claude usage chunks and restores mapped tool names", async () => { let onCompletePayload = null; const text = await readTransformed( [ `data: ${JSON.stringify({ type: "message_start", message: { id: "msg_passthrough", model: "claude-sonnet-4", role: "assistant", usage: { input_tokens: 6 }, }, })}\n\n`, `data: ${JSON.stringify({ type: "content_block_start", index: 0, content_block: { type: "tool_use", id: "tool_1", name: "tool_alias", input: { path: "/tmp/a" }, }, })}\n\n`, `data: ${JSON.stringify({ type: "content_block_delta", index: 1, delta: { text: "Claude says hi" }, })}\n\n`, `data: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 4 }, })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.CLAUDE, provider: "claude", model: "claude-sonnet-4", toolNameMap: new Map([["tool_alias", "read_file"]]), body: { messages: [{ role: "user", content: "hello" }], }, onComplete(payload) { onCompletePayload = payload; }, } ); assert.match(text, /"name":"read_file"/); assert.equal(text.includes('"name":"tool_alias"'), false); assert.equal(onCompletePayload.responseBody.choices[0].message.content, "Claude says hi"); assert.equal(onCompletePayload.responseBody.usage.prompt_tokens, 6); assert.equal(onCompletePayload.responseBody.usage.completion_tokens, 4); assert.equal(onCompletePayload.responseBody.usage.total_tokens, 10); }); test("#3685 createSSEStream passthrough emits SSE error (not synthetic text) for empty Claude assistant SSE", async () => { let failurePayload = null; let completePayload = null; await assert.rejects( readTransformed( [ `event: message_start\ndata: ${JSON.stringify({ type: "message_start", message: { id: "msg_empty_passthrough", type: "message", role: "assistant", model: "claude-sonnet-4", content: [], stop_reason: null, stop_sequence: null, usage: { input_tokens: 7, output_tokens: 0 }, }, })}\n\n`, `event: message_stop\ndata: ${JSON.stringify({ type: "message_stop", })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.CLAUDE, provider: "claude", model: "claude-sonnet-4", body: { messages: [{ role: "user", content: "hello" }], }, onFailure(payload) { failurePayload = payload; }, onComplete(payload) { completePayload = payload; }, } ), /empty response/i ); assert.ok(failurePayload, "onFailure should be called"); assert.equal(failurePayload.status, 502); assert.match(failurePayload.message, /empty response/i); assert.equal(failurePayload.code, "empty_response", "code must identify the failure kind"); assert.equal(typeof failurePayload.code, "string", "code must be a string"); assert.ok(failurePayload.message.length > 0, "message must be non-empty"); assert.ok(failurePayload.status >= 500, "status must be a server error (5xx)"); assert.equal(completePayload, null, "onComplete must not fire when stream is empty"); }); test("createSSEStream passthrough does not emit [DONE] for Claude SSE clients", async () => { const text = await readTransformed( [ `event: message_start\ndata: ${JSON.stringify({ type: "message_start", message: { id: "msg_claude_done_gate", type: "message", role: "assistant", model: "claude-sonnet-4", content: [], stop_reason: null, stop_sequence: null, usage: { input_tokens: 3, output_tokens: 0 }, }, })}\n\n`, `event: content_block_start\ndata: ${JSON.stringify({ type: "content_block_start", index: 0, content_block: { type: "text", text: "" }, })}\n\n`, `event: content_block_delta\ndata: ${JSON.stringify({ type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "Claude client stream" }, })}\n\n`, `event: content_block_stop\ndata: ${JSON.stringify({ type: "content_block_stop", index: 0, })}\n\n`, `event: message_delta\ndata: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "end_turn", stop_sequence: null }, usage: { output_tokens: 3 }, })}\n\n`, `event: message_stop\ndata: ${JSON.stringify({ type: "message_stop", })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.CLAUDE, clientResponseFormat: FORMATS.CLAUDE, provider: "claude", model: "claude-sonnet-4", body: { messages: [{ role: "user", content: "hello" }], }, } ); assert.match(text, /event: message_stop/); assert.match(text, /Claude client stream/); assert.doesNotMatch(text, /\[DONE\]/); }); test("#3685 createSSEStream translate mode emits SSE error (not synthetic text) when OpenAI upstream finishes empty for Claude client", async () => { let failurePayload = null; let completePayload = null; await assert.rejects( readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_empty_1", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [{ index: 0, delta: { role: "assistant" } }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_empty_1", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], usage: { prompt_tokens: 3, completion_tokens: 0, total_tokens: 3 }, })}\n\n`, ], { mode: "translate", targetFormat: FORMATS.OPENAI, sourceFormat: FORMATS.CLAUDE, provider: "openai", model: "gpt-4.1-mini", body: { messages: [{ role: "user", content: "hello" }], }, onFailure(payload) { failurePayload = payload; }, onComplete(payload) { completePayload = payload; }, } ), /empty response/i ); assert.ok(failurePayload, "onFailure should be called"); assert.equal(failurePayload.status, 502); assert.match(failurePayload.message, /empty response/i); assert.equal(failurePayload.code, "empty_response", "code must identify the failure kind"); assert.ok(failurePayload.message.length > 0, "message must be non-empty"); assert.ok(failurePayload.status >= 500, "status must be a server error (5xx)"); assert.equal(completePayload, null, "onComplete must not fire when stream is empty"); }); test("createSSETransformStreamWithLogger flushes a trailing Claude usage event without a newline", async () => { let onCompletePayload = null; const text = await readWithTransform( [ `data: ${JSON.stringify({ type: "message_start", message: { id: "msg_tail", model: "claude-sonnet-4", role: "assistant", usage: { input_tokens: 3 }, }, })}\n\n`, `data: ${JSON.stringify({ type: "content_block_start", index: 0, content_block: { type: "text", text: "" }, })}\n\n`, `data: ${JSON.stringify({ type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "Buffered tail" }, })}\n\n`, `data: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 5 }, })}`, ], createSSETransformStreamWithLogger( FORMATS.CLAUDE, FORMATS.OPENAI, "claude", null, null, "claude-sonnet-4", null, { messages: [{ role: "user", content: "hello" }] }, (payload) => { onCompletePayload = payload; } ) ); assert.match(text, /Buffered tail/); assert.match(text, /\[DONE\]/); assert.equal(onCompletePayload.responseBody.choices[0].message.content, "Buffered tail"); assert.equal(onCompletePayload.responseBody.usage.prompt_tokens, 3); assert.equal(onCompletePayload.responseBody.usage.completion_tokens, 5); assert.equal(onCompletePayload.responseBody.usage.total_tokens, 8); }); test("buildStreamSummaryFromEvents compacts Responses API deltas into a synthetic response", () => { const summary = buildStreamSummaryFromEvents( [ { index: 0, data: { type: "response.output_text.delta", delta: "Hello " } }, { index: 1, data: { type: "response.output_text.delta", delta: "world" } }, { index: 2, data: { type: "response.output_text.done", usage: { input_tokens: 2, output_tokens: 3, total_tokens: 5 }, }, }, ], FORMATS.OPENAI_RESPONSES, "gpt-4.1-mini" ); assert.equal((summary as any).object, "response"); assert.equal((summary as any).model, "gpt-4.1-mini"); assert.equal((summary as any).output[0].content[0].text, "Hello world"); assert.deepEqual((summary as any).usage, { input_tokens: 2, output_tokens: 3, total_tokens: 5 }); }); test("buildStreamSummaryFromEvents preserves Gemini thought parts and function calls", () => { const summary = buildStreamSummaryFromEvents( [ { index: 0, data: { modelVersion: "gemini-2.5-pro", candidates: [ { content: { role: "model", parts: [ { text: "Thinking", thought: true }, { text: " aloud", thought: true }, ], }, }, ], }, }, { index: 1, data: { candidates: [ { content: { role: "model", parts: [ { text: "Done." }, { functionCall: { name: "read_file", args: { path: "/tmp/a" } } }, ], }, finishReason: "STOP", }, ], usageMetadata: { promptTokenCount: 4, candidatesTokenCount: 5, totalTokenCount: 9, }, }, }, ], FORMATS.GEMINI, "gemini-2.5-pro" ); assert.equal((summary as any).modelVersion, "gemini-2.5-pro"); assert.equal((summary as any).candidates[0].content.parts[0].text, "Thinking aloud"); assert.equal((summary as any).candidates[0].content.parts[0].thought, true); assert.deepEqual((summary as any).candidates[0].content.parts[2], { functionCall: { name: "read_file", args: { path: "/tmp/a" } }, }); assert.deepEqual((summary as any).usageMetadata, { promptTokenCount: 4, candidatesTokenCount: 5, totalTokenCount: 9, }); }); test("compactStructuredStreamPayload wraps primitive summaries with Omniroute stream metadata", () => { const compact = compactStructuredStreamPayload({ _streamed: true, _format: "sse-json", _stage: "client_response", _eventCount: 2, summary: "done", }); assert.deepEqual(compact, { summary: "done", _omniroute_stream: { format: "sse-json", stage: "client_response", eventCount: 2, }, }); }); test("createSSETransformStreamWithLogger flushes Responses API terminal events on stream end", async () => { const text = await readWithTransform( [ `data: ${JSON.stringify({ id: "chatcmpl_flush", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [{ index: 0, delta: { role: "assistant", content: "Hello" } }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_flush", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], usage: { prompt_tokens: 2, completion_tokens: 3, total_tokens: 5 }, })}\n\n`, ], createSSETransformStreamWithLogger( FORMATS.OPENAI, FORMATS.OPENAI_RESPONSES, "openai", null, null, "gpt-4.1-mini", null, { messages: [{ role: "user", content: "hello" }] } ) ); assert.match(text, /response\.created/); assert.match(text, /response\.completed/); assert.doesNotMatch(text, /\[DONE\]/); }); test("createPassthroughStreamWithLogger reuses passthrough mode helpers", async () => { const text = await readWithTransform( [ `data: ${JSON.stringify({ id: "chatcmpl_passthrough", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [{ index: 0, delta: { role: "assistant", content: "Hello again" } }], })}\n\n`, "data: [DONE]\n\n", ], createPassthroughStreamWithLogger("openai", null, null, "gpt-4.1-mini", null, { messages: [{ role: "user", content: "hello" }], }) ); assert.match(text, /Hello again/); assert.match(text, /\[DONE\]/); }); test("createStructuredSSECollector drops excess events and compactStructuredStreamPayload preserves metadata for object summaries", () => { const collector = createStructuredSSECollector({ stage: "client_response", maxEvents: 1, maxBytes: 512, }); collector.push({ type: "response.output_text.delta", delta: "one" }); collector.push({ type: "response.output_text.delta", delta: "two" }); const built = collector.build( { object: "response", status: "completed", }, { includeEvents: false } ); const compact = compactStructuredStreamPayload(built); assert.equal(built._truncated, true); assert.equal(built._droppedEvents, 1); assert.equal(built._eventCount, 2); assert.deepEqual(compact, { object: "response", status: "completed", _omniroute_stream: { format: "sse-json", stage: "client_response", eventCount: 2, truncated: true, droppedEvents: 1, }, }); }); test("createSSEStream passthrough drops keepalive event blocks without losing Responses deltas", async () => { const text = await readTransformed( [ "event: keepalive\ndata:\n\n", `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "Hello keepalive-safe", })}\n\n`, `data: ${JSON.stringify({ type: "response.completed", response: { id: "resp_keepalive", object: "response", model: "gpt-4.1-mini", status: "completed", usage: { input_tokens: 2, output_tokens: 1, total_tokens: 3 }, }, })}\n\n`, "data: [DONE]\n\n", ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI_RESPONSES, provider: "openai", model: "gpt-4.1-mini", body: { input: "hello" }, } ); assert.equal(text.includes("event: keepalive"), false); assert.equal(text.includes("data:\n\n"), false); assert.match(text, /response\.output_text\.delta/); assert.match(text, /Hello keepalive-safe/); assert.match(text, /data: \[DONE\]/); }); test("createSSEStream passthrough aborts on Responses usage-limit failures and reports 429", async () => { let failurePayload = null; await assert.rejects( readTransformed( [ `data: ${JSON.stringify({ type: "response.failed", response: { id: "resp_usage_limit", object: "response", model: "gpt-5.5", status: "failed", error: { code: "usage_limit_reached", message: "Your weekly usage limit has been reached", }, }, })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI_RESPONSES, provider: "codex", model: "gpt-5.5", body: { input: "hello" }, onFailure(payload) { failurePayload = payload; }, } ), /weekly usage limit|Upstream failure/ ); assert.ok(failurePayload, "should report the stream failure before aborting"); assert.equal(failurePayload.status, 429); assert.equal(failurePayload.code, "usage_limit_reached"); }); test("createRequestLogger skips disabled logs and caps retained stream chunk bytes", async () => { const disabled = await createRequestLogger("openai", "openai", "gpt-test", { enabled: false, }); disabled.logClientRawRequest("/v1/chat/completions", { prompt: "hello" }); disabled.appendProviderChunk("x".repeat(32)); assert.equal(disabled.getPipelinePayloads(), null); const logger = await createRequestLogger("openai", "openai", "gpt-test", { enabled: true, captureStreamChunks: true, // The byte budget now also counts the 15-char timestamp prefix, so size it as // prefix + 5 to keep the original intent: cap the *payload* to 5 bytes ("abcde"). maxStreamChunkBytes: STREAM_CHUNK_TS_PREFIX_LEN + 5, }); logger.appendProviderChunk("abcdef"); logger.appendProviderChunk("ghijkl"); const payloads = logger.getPipelinePayloads(); // First chunk retained but its payload truncated to 5 bytes; second replaced by marker. assert.equal(stripChunkTs(payloads.streamChunks.provider[0]), "abcde"); assert.equal( payloads.streamChunks.provider[1], `[stream chunk log truncated after ${STREAM_CHUNK_TS_PREFIX_LEN + 5} bytes]` ); assert.equal(payloads.streamChunks.provider.length, 2); }); test("createRequestLogger caps retained stream chunk item count", async () => { const logger = await createRequestLogger("openai", "openai", "gpt-test", { enabled: true, captureStreamChunks: true, maxStreamChunkBytes: 1024, maxStreamChunkItems: 2, }); logger.appendProviderChunk("one"); logger.appendProviderChunk("two"); logger.appendProviderChunk("three"); const payloads = logger.getPipelinePayloads(); assert.equal(stripChunkTs(payloads.streamChunks.provider[0]), "one"); assert.equal(payloads.streamChunks.provider[1], "[stream chunk log truncated after 2 chunks]"); assert.equal(payloads.streamChunks.provider.length, 2); }); // T-VERIFY: passthrough mode failure decrements pending requests // Regression test for missing trackPendingRequest(false) on passthrough failure import { getPendingRequests, clearPendingRequests } from "../../src/lib/usage/usageHistory.ts"; test("createSSEStream passthrough mode decrements pending requests on failure", async () => { // Clear any existing pending requests first clearPendingRequests(); const initial = getPendingRequests(); assert.equal(Object.keys(initial.byModel).length, 0, "should start with no pending requests"); let failurePayload = null; const testProvider = "openai-compatible-test-failure"; const testModel = "gpt-test"; const testConnectionId = "test-conn-123"; await assert.rejects( readTransformed( [ `data: ${JSON.stringify({ type: "response.failed", response: { id: "resp_failed_test", object: "response", model: testModel, status: "failed", error: { code: "test_failure", message: "Test failure for pending request tracking", }, }, })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI_RESPONSES, provider: testProvider, model: testModel, connectionId: testConnectionId, body: { input: "hello" }, onFailure(payload) { failurePayload = payload; }, } ), /Test failure|Upstream failure/ ); assert.ok(failurePayload, "should report the stream failure"); // Verify pending requests are properly decremented after failure const pending = getPendingRequests(); const modelKey = `${testModel} (${testProvider})`; const count = pending.byModel[modelKey] || 0; assert.equal( count, 0, `pending request count for ${modelKey} should be 0 after failure, got ${count}` ); }); test("createSSEStream passthrough drops empty choices array chunks", async () => { let onCompletePayload = null; const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_empty", object: "chat.completion.chunk", created: 1, model: "kimi-k2.6", choices: [], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_empty", object: "chat.completion.chunk", created: 1, model: "kimi-k2.6", choices: [{ index: 0, delta: { role: "assistant", content: "Hello" } }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_empty", object: "chat.completion.chunk", created: 1, model: "kimi-k2.6", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "opencode-go", model: "kimi-k2.6", body: { messages: [{ role: "user", content: "hello" }] }, onComplete(payload) { onCompletePayload = payload; }, } ); // Empty choices WITHOUT usage are DROPPED, never replaced with a synthetic // "[OmniRoute] Upstream returned an empty response. Please retry." chunk. That // injection (reintroduced by #3422) was fed back by clients as a turn and caused // the retry loop #3388/#3502, which #3400 had fixed by dropping the chunk. assert.doesNotMatch(text, /\[OmniRoute\] Upstream returned an empty response/); // Subsequent valid chunks must still pass through untouched. assert.match(text, /"content":"Hello"/); assert.match(text, /"finish_reason":"stop"/); assert.equal(onCompletePayload.status, 200); assert.equal(onCompletePayload.responseBody.choices[0].message.content, "Hello"); }); test("createSSEStream passthrough forwards OpenAI usage-only empty choices chunks", async () => { let onCompletePayload = null; const usage = { prompt_tokens: 7, completion_tokens: 3, total_tokens: 10 }; const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_usage_only", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [{ index: 0, delta: { role: "assistant", content: "Hello" } }], usage: null, })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_usage_only", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], usage: null, })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_usage_only", object: "chat.completion.chunk", created: 1, model: "gpt-4.1-mini", choices: [], usage, })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "openai-compatible", model: "gpt-4.1-mini", body: { messages: [{ role: "user", content: "hello" }], stream_options: { include_usage: true }, }, onComplete(payload) { onCompletePayload = payload; }, } ); assert.doesNotMatch(text, /\[OmniRoute\] Upstream returned an empty response/); assert.match(text, /"choices":\[\]/); assert.match(text, /"usage":\{"prompt_tokens":7,"completion_tokens":3,"total_tokens":10\}/); assert.equal(onCompletePayload.status, 200); assert.equal(onCompletePayload.usage.prompt_tokens, 7); assert.equal(onCompletePayload.usage.completion_tokens, 3); assert.equal(onCompletePayload.responseBody.choices[0].message.content, "Hello"); assert.deepEqual(onCompletePayload.responseBody.usage, usage); }); test("createSSEStream passthrough logs empty response after tool_calls completion", async () => { let onCompletePayload = null; const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_tool_then_empty", object: "chat.completion.chunk", created: 1, model: "gpt-5.5-xhigh", choices: [ { index: 0, delta: { tool_calls: [ { index: 0, id: "call_tc", type: "function", function: { name: "task_complete", arguments: "{}" }, }, ], }, }, ], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_tool_then_empty", object: "chat.completion.chunk", created: 1, model: "gpt-5.5-xhigh", choices: [{ index: 0, delta: {}, finish_reason: "tool_calls" }], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "codex", model: "gpt-5.5-xhigh", body: { messages: [{ role: "user", content: "do task" }] }, onComplete(payload) { onCompletePayload = payload; }, } ); assert.match(text, /"finish_reason":"tool_calls"/); assert.equal(onCompletePayload.status, 200); assert.equal(onCompletePayload.responseBody.choices[0].finish_reason, "tool_calls"); assert.equal( onCompletePayload.responseBody.choices[0].message.tool_calls[0].function.name, "task_complete" ); // Content should be null (empty) since no text was generated assert.equal(onCompletePayload.responseBody.choices[0].message.content, null); }); test("createSSEStream passthrough does not swallow false positive textual tool call", async () => { let onCompletePayload = null; const sentence = "Checking: [Tool call: terminal] was executed successfully."; const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_false_positive_textual_tool", object: "chat.completion.chunk", created: 1, model: "MainAgent", choices: [{ index: 0, delta: { role: "assistant", content: sentence } }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_false_positive_textual_tool", object: "chat.completion.chunk", created: 1, model: "MainAgent", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "omniroute", model: "MainAgent", body: { messages: [{ role: "user", content: "inspect status" }] }, onComplete(payload) { onCompletePayload = payload; }, } ); const choice = onCompletePayload.responseBody.choices[0]; assert.equal(choice.finish_reason, "stop"); assert.equal(choice.message.content, sentence); assert.equal(choice.message.tool_calls, undefined); assert.match(text, /\[Tool call: terminal\] was executed successfully/); }); test("createSSEStream passthrough does not swallow false positive textual tool call starting chunk", async () => { let onCompletePayload = null; const chunk1 = "[Tool call: terminal]"; const chunk2 = " was skipped."; const text = await readTransformed( [ `data: ${JSON.stringify({ id: "chatcmpl_false_positive_textual_tool_start", object: "chat.completion.chunk", created: 1, model: "MainAgent", choices: [{ index: 0, delta: { role: "assistant", content: chunk1 } }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_false_positive_textual_tool_start", object: "chat.completion.chunk", created: 1, model: "MainAgent", choices: [{ index: 0, delta: { content: chunk2 } }], })}\n\n`, `data: ${JSON.stringify({ id: "chatcmpl_false_positive_textual_tool_start", object: "chat.completion.chunk", created: 1, model: "MainAgent", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], })}\n\n`, ], { mode: "passthrough", sourceFormat: FORMATS.OPENAI, provider: "omniroute", model: "MainAgent", body: { messages: [{ role: "user", content: "inspect status" }] }, onComplete(payload) { onCompletePayload = payload; }, } ); const choice = onCompletePayload.responseBody.choices[0]; assert.equal(choice.finish_reason, "stop"); assert.equal(choice.message.content, chunk1 + chunk2); assert.equal(choice.message.tool_calls, undefined); assert.match(text, /\[Tool call: terminal\] was skipped/); });