From bb0019845bf38af72bbfba69be31525f0ab72990 Mon Sep 17 00:00:00 2001 From: Nick Sullivan <142708+TechNickAI@users.noreply.github.com> Date: Fri, 12 Jun 2026 00:49:08 -0500 Subject: [PATCH] fix(stream): error on empty Claude SSE instead of synthetic success (#3689) Integrated into release/v3.8.23 --- open-sse/utils/stream.ts | 114 ++++--- .../claude-empty-stream-error-3685.test.ts | 278 ++++++++++++++++++ tests/unit/stream-utils.test.ts | 174 ++++++----- 3 files changed, 429 insertions(+), 137 deletions(-) create mode 100644 tests/unit/claude-empty-stream-error-3685.test.ts diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 4091d5cc38..48633fdc7c 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -109,7 +109,11 @@ function normalizeResponsesSseIds(payload: JsonRecord): boolean { } } - if (payload.response && typeof payload.response === "object" && !Array.isArray(payload.response)) { + if ( + payload.response && + typeof payload.response === "object" && + !Array.isArray(payload.response) + ) { const response = payload.response as JsonRecord; let responseChanged = false; const normalizedResponse = { ...response }; @@ -1016,6 +1020,32 @@ export function createSSEStream(options: StreamOptions = {}) { } }; + const emitClaudeEmptyStreamErrorAndAbort = ( + controller: TransformStreamDefaultController, + decrementPendingRequest = true + ) => { + clearIdleTimer(); + const msg = "Claude returned an empty response (no content block)"; + console.warn( + `[STREAM] Empty Claude stream at flush - emitting error (${provider || "provider"}:${model || "unknown"})` + ); + const errorBody = buildErrorBody(502, msg); + const errorEvent: Record = { type: "error", error: errorBody.error }; + const errOutput = formatSSE(errorEvent, FORMATS.CLAUDE); + reqLogger?.appendConvertedChunk?.(errOutput); + clientPayloadCollector.push(errorEvent); + controller.enqueue(encoder.encode(errOutput)); + if (onFailure) { + try { + void onFailure({ status: 502, message: msg, code: "empty_response" }); + } catch {} + } + if (decrementPendingRequest) { + trackPendingRequest(model, provider, connectionId, false); + } + controller.error(markPendingRequestCleared(new Error(msg))); + }; + const emitTranslatedClientItem = ( controller: TransformStreamDefaultController, item: Record @@ -1059,13 +1089,8 @@ export function createSSEStream(options: StreamOptions = {}) { sourceFormat === FORMATS.CLAUDE && shouldInjectClaudeEmptyResponseBeforeCurrentEvent(claudeEmptyResponseLifecycle, itemSanitized) ) { - const eventType = getClaudeEventType(itemSanitized); - emitSyntheticClaudeEmptyResponse(controller, { - includeContentBlock: true, - includeMessageDelta: - eventType === "message_stop" && !claudeEmptyResponseLifecycle.hasMessageDelta, - includeMessageStop: false, - }); + emitClaudeEmptyStreamErrorAndAbort(controller); + return; } if (sourceFormat === FORMATS.CLAUDE && isClaudeEventPayload(itemSanitized)) { @@ -1300,12 +1325,8 @@ export function createSSEStream(options: StreamOptions = {}) { type: eventType, }) ) { - emitSyntheticClaudeEmptyResponse(controller, { - includeContentBlock: true, - includeMessageDelta: - eventType === "message_stop" && !claudeEmptyResponseLifecycle.hasMessageDelta, - includeMessageStop: false, - }); + emitClaudeEmptyStreamErrorAndAbort(controller); + return; } pendingPassthroughEventLine = line; @@ -1337,7 +1358,8 @@ export function createSSEStream(options: StreamOptions = {}) { // clients like OpenCode, so drop it only for Responses-native consumers. const hasActiveDeltaValue = (value: unknown): boolean => { if (typeof value === "string") return value.length > 0; - if (Array.isArray(value)) return value.some((entry) => hasActiveDeltaValue(entry)); + if (Array.isArray(value)) + return value.some((entry) => hasActiveDeltaValue(entry)); if (value && typeof value === "object") { return Object.values(value).some((entry) => hasActiveDeltaValue(entry)); } @@ -1605,7 +1627,12 @@ export function createSSEStream(options: StreamOptions = {}) { parsed, passthroughResponsesOutputItems ); - if (stripped || backfilled || textualToolCallBackfilled || responsesIdsNormalized) { + if ( + stripped || + backfilled || + textualToolCallBackfilled || + responsesIdsNormalized + ) { output = `data: ${JSON.stringify(parsed)}\n`; injectedUsage = true; } @@ -1632,13 +1659,8 @@ export function createSSEStream(options: StreamOptions = {}) { parsed ) ) { - emitSyntheticClaudeEmptyResponse(controller, { - includeContentBlock: true, - includeMessageDelta: - parsed.type === "message_stop" && - !claudeEmptyResponseLifecycle.hasMessageDelta, - includeMessageStop: false, - }); + emitClaudeEmptyStreamErrorAndAbort(controller); + return; } updateClaudeEmptyResponseLifecycle(claudeEmptyResponseLifecycle, parsed); const restoredToolName = restoreClaudePassthroughToolUseName(parsed, toolNameMap); @@ -1708,14 +1730,16 @@ export function createSSEStream(options: StreamOptions = {}) { !parsed.choices[0].delta.reasoning_content ); const hadNonStringToolCallId = Array.isArray(parsed.choices) - ? parsed.choices.some((choice) => - Array.isArray(choice?.delta?.tool_calls) && - choice.delta.tool_calls.some( - (tc) => tc?.id != null && typeof tc.id !== "string" - ) + ? parsed.choices.some( + (choice) => + Array.isArray(choice?.delta?.tool_calls) && + choice.delta.tool_calls.some( + (tc) => tc?.id != null && typeof tc.id !== "string" + ) ) : false; - const hadNonStringTopLevelId = parsed?.id != null && typeof parsed.id !== "string"; + const hadNonStringTopLevelId = + parsed?.id != null && typeof parsed.id !== "string"; parsed = sanitizeStreamingChunk(parsed); if ( @@ -2148,13 +2172,8 @@ export function createSSEStream(options: StreamOptions = {}) { bufferedPayload ) ) { - const eventType = getClaudeEventType(bufferedPayload); - emitSyntheticClaudeEmptyResponse(controller, { - includeContentBlock: true, - includeMessageDelta: - eventType === "message_stop" && !claudeEmptyResponseLifecycle.hasMessageDelta, - includeMessageStop: false, - }); + emitClaudeEmptyStreamErrorAndAbort(controller, false); + return; } if (isClaudeEventPayload(bufferedPayload)) { updateClaudeEmptyResponseLifecycle(claudeEmptyResponseLifecycle, bufferedPayload); @@ -2164,7 +2183,8 @@ export function createSSEStream(options: StreamOptions = {}) { // Normalize numeric IDs for final buffered data: chunk (same as transform path) if (typeof bufferedPayload === "object" && !Array.isArray(bufferedPayload)) { const flushedParsed = bufferedPayload as JsonRecord; - const flushedType = typeof flushedParsed.type === "string" ? flushedParsed.type : ""; + const flushedType = + typeof flushedParsed.type === "string" ? flushedParsed.type : ""; const isResponses = flushedType.startsWith("response."); const isClaude = isClaudeEventPayload(flushedParsed); if (isResponses) { @@ -2181,7 +2201,9 @@ export function createSSEStream(options: StreamOptions = {}) { } if (Array.isArray(flushedParsed.choices)) { for (const choice of flushedParsed.choices as JsonRecord[]) { - const tcs = (choice as JsonRecord | undefined)?.delta as JsonRecord | undefined; + const tcs = (choice as JsonRecord | undefined)?.delta as + | JsonRecord + | undefined; if (Array.isArray(tcs?.tool_calls)) { for (const tc of tcs.tool_calls as JsonRecord[]) { if (tc?.id != null && typeof tc.id !== "string") { @@ -2208,11 +2230,8 @@ export function createSSEStream(options: StreamOptions = {}) { } if (shouldInjectClaudeEmptyResponseOnFlush(claudeEmptyResponseLifecycle)) { - emitSyntheticClaudeEmptyResponse(controller, { - includeContentBlock: true, - includeMessageDelta: !claudeEmptyResponseLifecycle.hasMessageDelta, - includeMessageStop: !claudeEmptyResponseLifecycle.hasMessageStop, - }); + emitClaudeEmptyStreamErrorAndAbort(controller, false); + return; } else if (shouldInjectClaudeMissingFinalizersOnFlush(claudeEmptyResponseLifecycle)) { emitSyntheticClaudeEmptyResponse(controller, { includeContentBlock: false, @@ -2489,11 +2508,8 @@ export function createSSEStream(options: StreamOptions = {}) { if (sourceFormat === FORMATS.CLAUDE) { if (shouldInjectClaudeEmptyResponseOnFlush(claudeEmptyResponseLifecycle)) { - emitSyntheticClaudeEmptyResponse(controller, { - includeContentBlock: true, - includeMessageDelta: !claudeEmptyResponseLifecycle.hasMessageDelta, - includeMessageStop: !claudeEmptyResponseLifecycle.hasMessageStop, - }); + emitClaudeEmptyStreamErrorAndAbort(controller, false); + return; } else if (shouldInjectClaudeMissingFinalizersOnFlush(claudeEmptyResponseLifecycle)) { emitSyntheticClaudeEmptyResponse(controller, { includeContentBlock: false, @@ -2631,7 +2647,7 @@ export function createSSEStream(options: StreamOptions = {}) { ); } -export default createSSEStream +export default createSSEStream; // Convenience functions for backward compatibility export function createSSETransformStreamWithLogger( diff --git a/tests/unit/claude-empty-stream-error-3685.test.ts b/tests/unit/claude-empty-stream-error-3685.test.ts new file mode 100644 index 0000000000..719f2cb638 --- /dev/null +++ b/tests/unit/claude-empty-stream-error-3685.test.ts @@ -0,0 +1,278 @@ +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"; + +// #3685 — When a Claude stream completes with lifecycle events (message_start / +// message_delta / message_stop) but zero content_block events, the router was +// injecting a synthetic success message instead of failing over. Fix: emit a +// real SSE error event and call controller.error() so the combo layer can retry. + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-3685-")); +process.env.DATA_DIR = TEST_DATA_DIR; +const core = await import("../../src/lib/db/core.ts"); +const { createSSEStream } = await import("../../open-sse/utils/stream.ts"); +const { FORMATS } = await import("../../open-sse/translator/formats.ts"); +const { getPendingRequests, clearPendingRequests } = + await import("../../src/lib/usage/usageHistory.ts"); + +const enc = new TextEncoder(); + +async function readTransformed(chunks: string[], options: Record) { + const source = new ReadableStream({ + start(c) { + for (const chunk of chunks) c.enqueue(enc.encode(chunk)); + c.close(); + }, + }); + return new Response(source.pipeThrough(createSSEStream(options as any))).text(); +} + +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 }); + } + } +}); + +// --- Golden path: empty content-block stream (the bug case) should now error --- + +test("#3685 passthrough: empty Claude SSE (no content_block) rejects the stream", async () => { + let failurePayload: Record | null = null; + await assert.rejects( + readTransformed( + [ + `event: message_start\ndata: ${JSON.stringify({ + type: "message_start", + message: { + id: "msg_3685", + type: "message", + role: "assistant", + model: "claude-sonnet-4-6", + content: [], + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 5, output_tokens: 0 }, + }, + })}\n\n`, + `event: message_delta\ndata: ${JSON.stringify({ + type: "message_delta", + delta: { stop_reason: "content_filter", stop_sequence: null }, + usage: { output_tokens: 1 }, + })}\n\n`, + `event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}\n\n`, + ], + { + mode: "passthrough", + sourceFormat: FORMATS.CLAUDE, + provider: "anthropic", + model: "claude-sonnet-4-6", + body: { messages: [{ role: "user", content: "hello" }] }, + onFailure(p: Record) { + failurePayload = p; + }, + } + ), + /empty response/i, + "stream should reject with empty-response error" + ); + assert.ok(failurePayload, "onFailure callback must be invoked"); + assert.equal((failurePayload as any).status, 502); + assert.match((failurePayload as any).message as string, /empty response/i); +}); + +test("#3685 passthrough: empty Claude SSE emits event: error SSE line before aborting", async () => { + const collected: string[] = []; + const source = new ReadableStream({ + start(c) { + const chunks = [ + `event: message_start\ndata: ${JSON.stringify({ + type: "message_start", + message: { + id: "msg_3685b", + type: "message", + role: "assistant", + model: "claude-sonnet-4-6", + content: [], + stop_reason: null, + usage: { input_tokens: 5, output_tokens: 0 }, + }, + })}\n\n`, + `event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}\n\n`, + ]; + for (const chunk of chunks) c.enqueue(enc.encode(chunk)); + c.close(); + }, + }); + const transformed = source.pipeThrough( + createSSEStream({ + mode: "passthrough", + sourceFormat: FORMATS.CLAUDE, + provider: "anthropic", + model: "claude-sonnet-4-6", + body: { messages: [{ role: "user", content: "hello" }] }, + } as any) + ); + const reader = transformed.getReader(); + const dec = new TextDecoder(); + let gotError = false; + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + collected.push(dec.decode(value)); + } + } catch { + gotError = true; + } + assert.ok(gotError, "stream reader should throw on error"); + const full = collected.join(""); + assert.match(full, /event: error/, "SSE error event must be emitted before abort"); + assert.doesNotMatch( + full, + /event: content_block_start/, + "no synthetic content_block must be emitted" + ); +}); + +// --- Regression guards: excluded cases must NOT be turned into errors --- + +test("#3685 regression: stream with content_block events is NOT turned into an error", async () => { + // A max_tokens:1 ping returns exactly 1 token → content_block events exist. + // hasContentBlock = true → shouldInjectClaudeEmptyResponseOnFlush = false → no error. + const text = await readTransformed( + [ + `event: message_start\ndata: ${JSON.stringify({ + type: "message_start", + message: { + id: "msg_ping", + type: "message", + role: "assistant", + model: "claude-haiku-4-5", + content: [], + stop_reason: 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: "Hi" }, + })}\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: "max_tokens", stop_sequence: null }, + usage: { output_tokens: 1 }, + })}\n\n`, + `event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}\n\n`, + ], + { + mode: "passthrough", + sourceFormat: FORMATS.CLAUDE, + provider: "anthropic", + model: "claude-haiku-4-5", + body: { messages: [{ role: "user", content: "ping" }] }, + } + ); + assert.match(text, /Hi/, "content must pass through untouched"); + assert.doesNotMatch(text, /event: error/, "must NOT emit an error event"); +}); + +test("#3685 pending request counter is decremented when empty-stream error fires", async () => { + // Regression guard for the bug caught by Cursor/Codex: emitClaudeEmptyStreamErrorAndAbort + // was marking the error with PENDING_REQUEST_CLEARED_MARKER but never calling + // trackPendingRequest(..., false). streamHandler.clearPendingRequest() trusts the marker + // and skips its own decrement, leaving the counter permanently inflated. + clearPendingRequests(); + const { trackPendingRequest } = await import("../../src/lib/usage/usageHistory.ts"); + + // Simulate the stream engine incrementing the counter at request start. + trackPendingRequest("claude-sonnet-4-6", "anthropic", "conn-test", true); + assert.equal( + getPendingRequests().byModel["claude-sonnet-4-6 (anthropic)"], + 1, + "pending count should start at 1 after request begins" + ); + + await assert.rejects( + readTransformed( + [ + `event: message_start\ndata: ${JSON.stringify({ + type: "message_start", + message: { + id: "msg_pending_test", + type: "message", + role: "assistant", + model: "claude-sonnet-4-6", + content: [], + stop_reason: null, + usage: { input_tokens: 3, output_tokens: 0 }, + }, + })}\n\n`, + `event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}\n\n`, + ], + { + mode: "passthrough", + sourceFormat: FORMATS.CLAUDE, + provider: "anthropic", + model: "claude-sonnet-4-6", + connectionId: "conn-test", + body: { messages: [{ role: "user", content: "hello" }] }, + } + ), + /empty response/i + ); + + // emitClaudeEmptyStreamErrorAndAbort must call trackPendingRequest(..., false) so the + // counter is back to 0 after the stream terminates. + assert.equal( + getPendingRequests().byModel["claude-sonnet-4-6 (anthropic)"], + 0, + "pending count must be 0 after empty-stream error — not left inflated" + ); +}); + +test("#3685 regression: upstream error event sets hasError=true and does NOT trigger empty-stream path", () => { + // If Claude itself emits type:error, lifecycle.hasError=true. + // shouldInjectClaudeEmptyResponseBeforeCurrentEvent / shouldInjectClaudeEmptyResponseOnFlush + // both check !lifecycle.hasError first — so neither our new error path nor the old synthetic + // path is triggered. Verified by inspecting the guard functions directly. + const lifecycle = { + hasMessageStart: true, + hasContentBlock: false, + hasMessageDelta: false, + hasMessageStop: false, + hasError: false, + syntheticContentInjected: false, + warningLogged: false, + }; + + // Simulate receiving an error event: sets hasError = true. + const lifecycleWithError = { ...lifecycle, hasError: true }; + + // shouldInjectClaudeEmptyResponseOnFlush equivalent: hasError blocks it + const wouldInjectOnFlush = + !lifecycleWithError.hasError && + !lifecycleWithError.hasContentBlock && + (lifecycleWithError.hasMessageStart || + lifecycleWithError.hasMessageDelta || + lifecycleWithError.hasMessageStop); + + assert.equal( + wouldInjectOnFlush, + false, + "hasError=true must prevent the empty-stream error path from firing" + ); +}); diff --git a/tests/unit/stream-utils.test.ts b/tests/unit/stream-utils.test.ts index 7447ddc7dd..933bc549d5 100644 --- a/tests/unit/stream-utils.test.ts +++ b/tests/unit/stream-utils.test.ts @@ -725,7 +725,11 @@ test("createSSEStream passthrough drops leaked empty chat bootstrap chunks for R created: 1, model: "gpt-5.4", choices: [ - { index: 0, delta: { role: "assistant", content: null, refusal: null }, finish_reason: null }, + { + index: 0, + delta: { role: "assistant", content: null, refusal: null }, + finish_reason: null, + }, ], })}\n\n`, `event: response.created\ndata: ${JSON.stringify({ @@ -1047,50 +1051,46 @@ test("createSSEStream passthrough merges Claude usage chunks and restores mapped assert.equal(onCompletePayload.responseBody.usage.total_tokens, 10); }); -test("createSSEStream passthrough injects a synthetic Claude text block for empty assistant SSE", async () => { - let onCompletePayload = null; - const text = await 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 }, +test("#3685 createSSEStream passthrough emits SSE error (not synthetic text) for empty Claude assistant SSE", async () => { + let failurePayload = 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" }], }, - })}\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" }], - }, - onComplete(payload) { - onCompletePayload = payload; - }, - } + onFailure(payload) { + failurePayload = payload; + }, + } + ), + /empty response/i ); - - assert.equal((text.match(/event: message_start/g) || []).length, 1); - assert.equal((text.match(/event: message_delta/g) || []).length, 1); - assert.match(text, /event: content_block_start/); - assert.match(text, /event: content_block_delta/); - assert.match(text, /event: message_stop/); - assert.ok(text.indexOf("event: content_block_start") > text.indexOf("event: message_start")); - assert.ok(text.indexOf("event: message_stop") > text.indexOf("event: content_block_stop")); - // SYNTHETIC_CLAUDE_EMPTY_RESPONSE_TEXT is "" so the accumulator produces null content (empty delta is falsy). - assert.equal(onCompletePayload.responseBody.choices[0].message.content, null); + assert.ok(failurePayload, "onFailure should be called"); + assert.equal(failurePayload.status, 502); + assert.match(failurePayload.message, /empty response/i); }); test("createSSEStream passthrough does not emit [DONE] for Claude SSE clients", async () => { @@ -1149,51 +1149,46 @@ test("createSSEStream passthrough does not emit [DONE] for Claude SSE clients", assert.doesNotMatch(text, /\[DONE\]/); }); -test("createSSEStream translate mode injects a synthetic Claude text block when OpenAI finishes empty", async () => { - let onCompletePayload = null; - const text = await readTransformed( - [ - `data: ${JSON.stringify({ - id: "chatcmpl_empty_1", - object: "chat.completion.chunk", - created: 1, +test("#3685 createSSEStream translate mode emits SSE error (not synthetic text) when OpenAI upstream finishes empty for Claude client", async () => { + let failurePayload = 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", - 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" }], - }, - onComplete(payload) { - onCompletePayload = payload; - }, - } + body: { + messages: [{ role: "user", content: "hello" }], + }, + onFailure(payload) { + failurePayload = payload; + }, + } + ), + /empty response/i ); - - assert.equal((text.match(/event: message_start/g) || []).length, 1); - assert.match(text, /event: content_block_start/); - assert.match(text, /event: content_block_delta/); - assert.match(text, /event: message_delta/); - assert.match(text, /event: message_stop/); - assert.ok(text.indexOf("event: content_block_start") > text.indexOf("event: message_start")); - assert.ok(text.indexOf("event: message_delta") > text.indexOf("event: content_block_stop")); - // SYNTHETIC_CLAUDE_EMPTY_RESPONSE_TEXT is "" so the accumulator produces null content (empty delta is falsy). - assert.equal(onCompletePayload.responseBody.choices[0].message.content, null); - assert.equal(onCompletePayload.responseBody.usage.total_tokens, 3); + assert.ok(failurePayload, "onFailure should be called"); + assert.equal(failurePayload.status, 502); + assert.match(failurePayload.message, /empty response/i); }); test("createSSETransformStreamWithLogger flushes a trailing Claude usage event without a newline", async () => { @@ -1741,7 +1736,7 @@ test("createSSEStream passthrough logs empty response after tool_calls completio index: 0, id: "call_tc", type: "function", - function: { name: "task_complete", arguments: '{}' }, + function: { name: "task_complete", arguments: "{}" }, }, ], }, @@ -1771,7 +1766,10 @@ test("createSSEStream passthrough logs empty response after tool_calls completio 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"); + 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); });