diff --git a/next.config.mjs b/next.config.mjs index 220e747be0..24bbc30de5 100644 --- a/next.config.mjs +++ b/next.config.mjs @@ -82,6 +82,16 @@ const minimalBuildAliases = isMinimalBuild } : {}; +function readTimeoutMs(...values) { + for (const value of values) { + const normalized = typeof value === "string" ? value.trim() : value; + if (normalized == null || normalized === "") continue; + const parsed = Number(normalized); + if (Number.isFinite(parsed) && parsed >= 0) return Math.floor(parsed); + } + return 600_000; +} + /** @type {import('next').NextConfig} */ const nextConfig = { distDir, @@ -114,6 +124,10 @@ const nextConfig = { // uploads (OpenAI-compatible /v1/files) routinely exceed this. Match the // 512 MB server-side cap; tune via env if needed. proxyClientMaxBodySize: process.env.NEXT_PROXY_BODY_LIMIT || "512mb", + // Next's internal router proxy defaults to 30s when this is unset. OmniRoute + // can legitimately hold non-streaming chat requests open for minutes while an + // upstream provider finishes, so reuse the existing request-timeout knobs. + proxyTimeout: readTimeoutMs(process.env.REQUEST_TIMEOUT_MS, process.env.FETCH_TIMEOUT_MS), // PR-2 of diegosouzapw/OmniRoute#3932: tree-shake barrel re-exports so // route bundles don't pull in 14 locale files, every lucide-react icon, // or the full date-fns surface when only one helper is used. diff --git a/open-sse/handlers/chatCore/nonStreamingResponseBody.ts b/open-sse/handlers/chatCore/nonStreamingResponseBody.ts index 632ccab987..da6e8c6acc 100644 --- a/open-sse/handlers/chatCore/nonStreamingResponseBody.ts +++ b/open-sse/handlers/chatCore/nonStreamingResponseBody.ts @@ -43,10 +43,7 @@ const DEFAULT_MAX_NONSTREAMING_RESPONSE_BYTES = 64 * 1024 * 1024; // 64 MB * Override with `OMNIROUTE_MAX_NONSTREAMING_RESPONSE_BYTES`. */ export const MAX_NONSTREAMING_RESPONSE_BYTES = (() => { - const parsed = Number.parseInt( - String(process.env.OMNIROUTE_MAX_NONSTREAMING_RESPONSE_BYTES), - 10 - ); + const parsed = Number.parseInt(String(process.env.OMNIROUTE_MAX_NONSTREAMING_RESPONSE_BYTES), 10); return Number.isFinite(parsed) && parsed > 0 ? parsed : DEFAULT_MAX_NONSTREAMING_RESPONSE_BYTES; })(); @@ -77,10 +74,18 @@ export async function readNonStreamingResponseBody( * by `maxBytes` (cancels the upstream and throws {@link NonStreamingResponseTooLargeError} * past the cap) and by the body timeout, cancelling early on a terminal SSE signal. */ -type NonStreamingChunk = - | { kind: "done" } - | { kind: "skip" } - | { kind: "chunk"; value: Uint8Array }; +type NonStreamingChunk = { kind: "done" } | { kind: "skip" } | { kind: "chunk"; value: Uint8Array }; + +function cancelNonStreamingReader( + reader: ReadableStreamDefaultReader, + reason: unknown +): void { + try { + void reader.cancel(reason).catch(() => {}); + } catch { + // The caller is already unwinding or returning a complete terminal response. + } +} /** Read the next chunk under the body-timeout deadline, normalizing end/empty cases. */ async function readNextNonStreamingChunk( @@ -110,6 +115,12 @@ async function drainNonStreamingSseBody( let rawBody = ""; let bytesSeen = 0; const deadline = FETCH_BODY_TIMEOUT_MS > 0 ? Date.now() + FETCH_BODY_TIMEOUT_MS : 0; + let cancelRequested = false; + const requestCancel = (reason: unknown) => { + if (cancelRequested) return; + cancelRequested = true; + cancelNonStreamingReader(reader, reason); + }; try { while (true) { @@ -121,19 +132,19 @@ async function drainNonStreamingSseBody( // growing `rawBody` until the V8 heap is exhausted. bytesSeen += next.value.byteLength; if (bytesSeen > maxBytes) { - await reader.cancel("non-streaming response exceeded byte cap").catch(() => {}); + requestCancel("non-streaming response exceeded byte cap"); throw new NonStreamingResponseTooLargeError(bytesSeen, maxBytes); } const decodedChunk = decoder.decode(next.value, { stream: true }); rawBody += decodedChunk; if (appendNonStreamingSseTerminalSignal(terminalState, decodedChunk)) { - await reader.cancel("non-streaming bridge consumed terminal SSE event").catch(() => {}); + requestCancel("non-streaming bridge consumed terminal SSE event"); break; } } } catch (error) { - await reader.cancel(error).catch(() => {}); + requestCancel(error); throw error; } finally { rawBody += decoder.decode(); diff --git a/open-sse/handlers/chatCore/nonStreamingSse.ts b/open-sse/handlers/chatCore/nonStreamingSse.ts index 23e3da9517..5d7e642fc6 100644 --- a/open-sse/handlers/chatCore/nonStreamingSse.ts +++ b/open-sse/handlers/chatCore/nonStreamingSse.ts @@ -64,7 +64,9 @@ export function isTruthyStreamBody(body: unknown): boolean { return !!body && typeof body === "object" && (body as { stream?: unknown }).stream === true; } -export function isEventStreamAccepted(headers: Record | Headers | null | undefined) { +export function isEventStreamAccepted( + headers: Record | Headers | null | undefined +) { return (getHeaderValueCaseInsensitive(headers, "accept") || "") .toLowerCase() .includes("text/event-stream"); @@ -88,19 +90,32 @@ const NON_STREAMING_SSE_TERMINAL_TYPES = new Set([ "response.incomplete", ]); +function isNonStreamingSseTerminalType(eventType: string): boolean { + return NON_STREAMING_SSE_TERMINAL_TYPES.has(eventType); +} + export type NonStreamingSseTerminalState = { currentEvent: string; pendingLine: string; }; +function hasClaudeTerminalMessageDelta(parsed: unknown, eventType: string): boolean { + if (eventType !== "message_delta" || !parsed || typeof parsed !== "object") return false; + const delta = (parsed as { delta?: unknown }).delta; + if (!delta || typeof delta !== "object") return false; + const stopReason = (delta as { stop_reason?: unknown }).stop_reason; + return typeof stopReason === "string" ? stopReason.length > 0 : stopReason != null; +} + function processNonStreamingSseTerminalLine( state: NonStreamingSseTerminalState, rawLine: string ): boolean { const trimmed = rawLine.trim(); if (!trimmed || trimmed.startsWith(":")) { + const terminalEventOnly = !trimmed && isNonStreamingSseTerminalType(state.currentEvent); if (!trimmed) state.currentEvent = ""; - return false; + return terminalEventOnly; } if (trimmed.startsWith("event:")) { @@ -119,8 +134,11 @@ function processNonStreamingSseTerminalLine( // terminate with `[DONE]` (handled above), so parsing every one of them here is pure // waste that compounds into the CPU-runaway on large buffered responses. Skip the // JSON.parse unless the line could actually be a typed terminal. - if (!data.includes('"type"')) { - return NON_STREAMING_SSE_TERMINAL_TYPES.has(state.currentEvent); + if ( + !data.includes('"type"') && + !(state.currentEvent === "message_delta" && data.includes("stop_reason")) + ) { + return isNonStreamingSseTerminalType(state.currentEvent); } try { @@ -129,7 +147,9 @@ function processNonStreamingSseTerminalLine( parsed && typeof parsed === "object" && typeof parsed.type === "string" ? parsed.type : state.currentEvent; - return NON_STREAMING_SSE_TERMINAL_TYPES.has(eventType); + return ( + isNonStreamingSseTerminalType(eventType) || hasClaudeTerminalMessageDelta(parsed, eventType) + ); } catch { // Keep reading malformed data so the parser can report a useful upstream error. return false; diff --git a/open-sse/handlers/sseParser.ts b/open-sse/handlers/sseParser.ts index 94a14e810e..d2e634e12a 100644 --- a/open-sse/handlers/sseParser.ts +++ b/open-sse/handlers/sseParser.ts @@ -370,8 +370,7 @@ export function parseSSEToClaudeResponse(rawSSE, fallbackModel) { type: "thinking", index, thinking: toString(contentBlock.thinking), - signature: - typeof contentBlock.signature === "string" ? contentBlock.signature : undefined, + signature: toString(contentBlock.signature) || undefined, }); } else if (blockType === "tool_use") { blocks.set(index, { @@ -416,12 +415,17 @@ export function parseSSEToClaudeResponse(rawSSE, fallbackModel) { continue; } - if (deltaType === "thinking_delta" || typeof delta.thinking === "string") { + const isThinkingDelta = deltaType === "thinking_delta" || typeof delta.thinking === "string"; + const isSignatureDelta = + deltaType === "signature_delta" || typeof delta.signature === "string"; + if (isThinkingDelta || isSignatureDelta) { const thinking = existing && existing.type === "thinking" ? existing : { type: "thinking", index, thinking: "", signature: undefined }; - thinking.thinking += toString(delta.thinking); + if (isThinkingDelta) thinking.thinking += toString(delta.thinking); + const signature = toString(delta.signature); + if (signature) thinking.signature = `${thinking.signature || ""}${signature}`; blocks.set(index, thinking); continue; } @@ -454,35 +458,27 @@ export function parseSSEToClaudeResponse(rawSSE, fallbackModel) { if (!sawClaudeEvent) return null; - const content = [...blocks.values()] - .sort((a, b) => a.index - b.index) - .flatMap((block) => { - if (block.type === "text") { - return block.text ? [{ type: "text", text: block.text }] : []; - } - if (block.type === "thinking") { - return block.thinking - ? [ - { - type: "thinking", - thinking: block.thinking, - ...(block.signature ? { signature: block.signature } : {}), - }, - ] - : []; + const content = []; + for (const block of [...blocks.values()].sort((a, b) => a.index - b.index)) { + if (block.type === "text") { + if (block.text) content.push({ type: "text", text: block.text }); + continue; + } + if (block.type === "thinking") { + const hasSignature = typeof block.signature === "string" && block.signature.length > 0; + if (block.thinking || hasSignature) { + content.push({ + type: "thinking", + thinking: block.thinking || "", + ...(hasSignature ? { signature: block.signature } : {}), + }); } + continue; + } - const parsedInput = - block.inputJson.trim().length > 0 ? tryParseJson(block.inputJson) : block.input; - return [ - { - type: "tool_use", - id: block.id, - name: block.name, - input: parsedInput, - }, - ]; - }); + const input = block.inputJson.trim().length > 0 ? tryParseJson(block.inputJson) : block.input; + content.push({ type: "tool_use", id: block.id, name: block.name, input }); + } return { id: messageId || `msg_${Date.now()}`, diff --git a/src/domain/omnirouteResponseMeta.ts b/src/domain/omnirouteResponseMeta.ts index 9abe49f327..aac287a213 100644 --- a/src/domain/omnirouteResponseMeta.ts +++ b/src/domain/omnirouteResponseMeta.ts @@ -17,6 +17,40 @@ function toNonNegativeInteger(value: unknown): number { return Math.max(0, Math.round(toFiniteNumber(value))); } +const INVALID_HEADER_VALUE_CONTROL_CHARS = /[\u0000-\u001f\u007f]/g; +const ASCII_HEADER_VALUE_PATTERN = /^[\u0020-\u007e]*$/; + +function toWellFormedUnicode(value: string): string { + let result = ""; + + for (let i = 0; i < value.length; i += 1) { + const code = value.charCodeAt(i); + if (code >= 0xd800 && code <= 0xdbff) { + const next = value.charCodeAt(i + 1); + if (next >= 0xdc00 && next <= 0xdfff) { + result += value[i] + value[i + 1]; + i += 1; + } else { + result += "\uFFFD"; + } + continue; + } + if (code >= 0xdc00 && code <= 0xdfff) { + result += "\uFFFD"; + continue; + } + result += value[i]; + } + + return result; +} + +function toHeaderValue(value: string): string { + const withoutControls = value.replace(INVALID_HEADER_VALUE_CONTROL_CHARS, ""); + if (ASCII_HEADER_VALUE_PATTERN.test(withoutControls)) return withoutControls; + return encodeURIComponent(toWellFormedUnicode(withoutControls)); +} + export function getOmniRouteTokenCounts(usage: UsageLike): { input: number; output: number } { if (!usage || typeof usage !== "object") { return { input: 0, output: 0 }; @@ -75,35 +109,37 @@ export function buildOmniRouteResponseMetaHeaders({ }): Record { const tokens = getOmniRouteTokenCounts(usage); const headers: Record = { - [OMNIROUTE_RESPONSE_HEADERS.cacheHit]: String(cacheHit), - [OMNIROUTE_RESPONSE_HEADERS.latencyMs]: String(toNonNegativeInteger(latencyMs)), - [OMNIROUTE_RESPONSE_HEADERS.responseCost]: formatOmniRouteCost(costUsd), - [OMNIROUTE_RESPONSE_HEADERS.tokensIn]: String(tokens.input), - [OMNIROUTE_RESPONSE_HEADERS.tokensOut]: String(tokens.output), - [OMNIROUTE_RESPONSE_HEADERS.version]: APP_CONFIG.version, + [OMNIROUTE_RESPONSE_HEADERS.cacheHit]: toHeaderValue(String(cacheHit)), + [OMNIROUTE_RESPONSE_HEADERS.latencyMs]: toHeaderValue(String(toNonNegativeInteger(latencyMs))), + [OMNIROUTE_RESPONSE_HEADERS.responseCost]: toHeaderValue(formatOmniRouteCost(costUsd)), + [OMNIROUTE_RESPONSE_HEADERS.tokensIn]: toHeaderValue(String(tokens.input)), + [OMNIROUTE_RESPONSE_HEADERS.tokensOut]: toHeaderValue(String(tokens.output)), + [OMNIROUTE_RESPONSE_HEADERS.version]: toHeaderValue(APP_CONFIG.version), }; if (typeof model === "string" && model.trim().length > 0) { - headers[OMNIROUTE_RESPONSE_HEADERS.model] = model; + headers[OMNIROUTE_RESPONSE_HEADERS.model] = toHeaderValue(model); } if (typeof requestId === "string" && requestId.trim().length > 0) { - headers[OMNIROUTE_RESPONSE_HEADERS.requestId] = requestId; + headers[OMNIROUTE_RESPONSE_HEADERS.requestId] = toHeaderValue(requestId); } if (typeof provider === "string" && provider.trim().length > 0) { - headers[OMNIROUTE_RESPONSE_HEADERS.provider] = getProviderAlias(provider); + headers[OMNIROUTE_RESPONSE_HEADERS.provider] = toHeaderValue(getProviderAlias(provider)); } // Cache-saved cost: emitted only when the caller passes a value (cache HITs), so // non-cache responses keep their existing header shape. `0` is a valid saved cost. if (costSavedUsd != null) { - headers[OMNIROUTE_RESPONSE_HEADERS.costSaved] = formatOmniRouteCost(costSavedUsd); + headers[OMNIROUTE_RESPONSE_HEADERS.costSaved] = toHeaderValue( + formatOmniRouteCost(costSavedUsd) + ); } const attempts = toNonNegativeInteger(fallbackAttempts); if (attempts > 0) { - headers[OMNIROUTE_RESPONSE_HEADERS.fallbackAttempts] = String(attempts); + headers[OMNIROUTE_RESPONSE_HEADERS.fallbackAttempts] = toHeaderValue(String(attempts)); } return headers; diff --git a/tests/unit/chatcore-non-streaming-response-body.test.ts b/tests/unit/chatcore-non-streaming-response-body.test.ts index 7e9807cd1d..a82d11ef19 100644 --- a/tests/unit/chatcore-non-streaming-response-body.test.ts +++ b/tests/unit/chatcore-non-streaming-response-body.test.ts @@ -24,8 +24,8 @@ test("drains an SSE stream chunk-by-chunk and concatenates until close", async ( const enc = new TextEncoder(); const body = new ReadableStream({ start(controller) { - controller.enqueue(enc.encode("data: {\"a\":1}\n\n")); - controller.enqueue(enc.encode("data: {\"b\":2}\n\n")); + controller.enqueue(enc.encode('data: {"a":1}\n\n')); + controller.enqueue(enc.encode('data: {"b":2}\n\n')); controller.close(); }, }); @@ -35,6 +35,33 @@ test("drains an SSE stream chunk-by-chunk and concatenates until close", async ( assert.ok(out.includes('"b":2')); }); +test("returns after terminal SSE even when underlying cancel never resolves", async () => { + const enc = new TextEncoder(); + let cancelled = false; + const body = new ReadableStream({ + start(controller) { + controller.enqueue(enc.encode('data: {"id":"x","choices":[{"delta":{"content":"ok"}}]}\n\n')); + controller.enqueue(enc.encode("data: [DONE]\n\n")); + }, + cancel() { + cancelled = true; + return new Promise(() => {}); + }, + }); + const response = new Response(body, { headers: { "Content-Type": "text/event-stream" } }); + + const out = await Promise.race([ + readNonStreamingResponseBody(response, "text/event-stream", true), + new Promise((_, reject) => + setTimeout(() => reject(new Error("read timed out after terminal SSE")), 1000) + ), + ]); + + assert.ok(out.includes('"content":"ok"')); + assert.ok(out.includes("[DONE]")); + assert.equal(cancelled, true); +}); + // #5152: bound the non-streaming buffer so a runaway upstream body cannot fill the V8 heap. test("aborts and throws when an SSE stream exceeds the byte cap (no unbounded string)", async () => { diff --git a/tests/unit/chatcore-non-streaming-sse.test.ts b/tests/unit/chatcore-non-streaming-sse.test.ts index 2f4965e1c8..e467405e47 100644 --- a/tests/unit/chatcore-non-streaming-sse.test.ts +++ b/tests/unit/chatcore-non-streaming-sse.test.ts @@ -55,6 +55,61 @@ test("appendNonStreamingSseTerminalSignal detects [DONE] and terminal event type const stop: NonStreamingSseTerminalState = { currentEvent: "", pendingLine: "" }; assert.equal(appendNonStreamingSseTerminalSignal(stop, "event: message_stop\ndata: {}\n"), true); + const responseTerminalEvents = [ + "response.completed", + "response.done", + "response.cancelled", + "response.canceled", + "response.failed", + "response.incomplete", + ]; + for (const eventType of responseTerminalEvents) { + const typed: NonStreamingSseTerminalState = { currentEvent: "", pendingLine: "" }; + assert.equal( + appendNonStreamingSseTerminalSignal(typed, `data: {"type":"${eventType}"}\n`), + true, + eventType + ); + + const eventOnly: NonStreamingSseTerminalState = { currentEvent: "", pendingLine: "" }; + assert.equal( + appendNonStreamingSseTerminalSignal(eventOnly, `event: ${eventType}\n\n`), + true, + eventType + ); + } + + const splitTerminalData: NonStreamingSseTerminalState = { currentEvent: "", pendingLine: "" }; + assert.equal( + appendNonStreamingSseTerminalSignal(splitTerminalData, "event: response.completed\n"), + false + ); + assert.equal( + appendNonStreamingSseTerminalSignal( + splitTerminalData, + 'data: {"response":{"status":"completed"}}\n' + ), + true + ); + + const messageDeltaWithType: NonStreamingSseTerminalState = { currentEvent: "", pendingLine: "" }; + assert.equal( + appendNonStreamingSseTerminalSignal( + messageDeltaWithType, + 'data: {"type":"message_delta","delta":{"stop_reason":"end_turn"}}\n' + ), + true + ); + + const messageDeltaFromEvent: NonStreamingSseTerminalState = { currentEvent: "", pendingLine: "" }; + assert.equal( + appendNonStreamingSseTerminalSignal( + messageDeltaFromEvent, + 'event: message_delta\ndata: {"delta":{"stop_reason":"end_turn"}}\n' + ), + true + ); + const delta: NonStreamingSseTerminalState = { currentEvent: "", pendingLine: "" }; assert.equal( appendNonStreamingSseTerminalSignal(delta, 'data: {"type":"content_block_delta"}\n'), @@ -62,6 +117,20 @@ test("appendNonStreamingSseTerminalSignal detects [DONE] and terminal event type ); }); +test("parseNonStreamingSSEPayload still parses Claude event/data buffers", () => { + const raw = + 'event: message_start\ndata: {"type":"message_start","message":{"id":"msg_1","model":"claude","role":"assistant","usage":{"input_tokens":1}}}\n\n' + + 'event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}\n\n' + + 'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"hi"}}\n\n' + + 'event: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":1}}\n\n' + + 'event: message_stop\ndata: {"type":"message_stop"}\n\n'; + const result = parseNonStreamingSSEPayload(raw, FORMATS.CLAUDE, "claude"); + assert.ok(result !== null); + assert.equal(result?.format, FORMATS.CLAUDE); + assert.deepEqual(result?.body.content, [{ type: "text", text: "hi" }]); + assert.equal(result?.body.stop_reason, "end_turn"); +}); + test("parseNonStreamingSSEPayload parses an OpenAI-format SSE buffer", () => { const raw = 'data: {"id":"x","choices":[{"delta":{"content":"hi"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n'; diff --git a/tests/unit/omniroute-response-meta.test.ts b/tests/unit/omniroute-response-meta.test.ts index 5956411e8d..0b7bbd8e2b 100644 --- a/tests/unit/omniroute-response-meta.test.ts +++ b/tests/unit/omniroute-response-meta.test.ts @@ -50,6 +50,40 @@ test("buildOmniRouteResponseMetaHeaders formats provider alias, tokens, latency, assert.equal(headers["X-OmniRoute-Response-Cost"], "0.0012345679"); }); +test("buildOmniRouteResponseMetaHeaders keeps ASCII model header values unchanged", () => { + const headers = buildOmniRouteResponseMetaHeaders({ + provider: "openai", + model: "gpt-4o-mini", + }); + + assert.equal(headers[OMNIROUTE_RESPONSE_HEADERS.model], "gpt-4o-mini"); +}); + +test("buildOmniRouteResponseMetaHeaders percent-encodes non-ASCII model header values", () => { + const model = "free-mix/[假流式]gemini-3.5-flash"; + const headers = buildOmniRouteResponseMetaHeaders({ + provider: "openai", + model, + }); + + assert.equal(headers[OMNIROUTE_RESPONSE_HEADERS.model], encodeURIComponent(model)); + assert.doesNotThrow(() => new Headers(headers)); +}); + +test("buildOmniRouteResponseMetaHeaders strips control characters from string header values", () => { + const headers = buildOmniRouteResponseMetaHeaders({ + provider: "openai", + model: "free\r\nX-Injected: yes\u0000-model", + requestId: "req-1\nreq-2\rreq-3\u0007", + }); + + assert.doesNotMatch(headers[OMNIROUTE_RESPONSE_HEADERS.model], /[\r\n\u0000-\u001f\u007f]/); + assert.doesNotMatch(headers[OMNIROUTE_RESPONSE_HEADERS.requestId], /[\r\n\u0000-\u001f\u007f]/); + assert.equal(headers[OMNIROUTE_RESPONSE_HEADERS.model], "freeX-Injected: yes-model"); + assert.equal(headers[OMNIROUTE_RESPONSE_HEADERS.requestId], "req-1req-2req-3"); + assert.doesNotThrow(() => new Headers(headers)); +}); + test("buildOmniRouteResponseMetaHeaders always emits X-OmniRoute-Version", () => { const headers = buildOmniRouteResponseMetaHeaders({ provider: "openai", model: "gpt" }); assert.equal(headers[OMNIROUTE_RESPONSE_HEADERS.version], APP_CONFIG.version); diff --git a/tests/unit/sse-parser.test.ts b/tests/unit/sse-parser.test.ts index 9a524e21a8..7511834df0 100644 --- a/tests/unit/sse-parser.test.ts +++ b/tests/unit/sse-parser.test.ts @@ -134,6 +134,49 @@ test("parseSSEToClaudeResponse tolerates event-only types and missing blank sepa assert.deepEqual(parsed.usage, { input_tokens: 3, output_tokens: 2 }); }); +test("parseSSEToClaudeResponse merges signature_delta into an existing thinking block", () => { + const rawSSE = [ + "event: message_start", + 'data: {"type":"message_start","message":{"id":"msg_thinking_sig","model":"claude-sonnet-4-6","role":"assistant"}}', + "", + "event: content_block_delta", + 'data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"first "}}', + "", + "event: content_block_delta", + 'data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"second"}}', + "", + "event: content_block_delta", + 'data: {"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig-1"}}', + "", + ].join("\n"); + + const parsed = parseSSEToClaudeResponse(rawSSE, "fallback-model"); + + assert.equal((parsed.content[0] as any).type, "thinking"); + assert.equal((parsed.content[0] as any).thinking, "first second"); + assert.equal((parsed.content[0] as any).signature, "sig-1"); +}); + +test("parseSSEToClaudeResponse preserves signature_delta when it arrives before thinking_delta", () => { + const rawSSE = [ + "event: message_start", + 'data: {"type":"message_start","message":{"id":"msg_sig_first","model":"claude-sonnet-4-6","role":"assistant"}}', + "", + "event: content_block_delta", + 'data: {"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig-before"}}', + "", + "event: content_block_delta", + 'data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"later thinking"}}', + "", + ].join("\n"); + + const parsed = parseSSEToClaudeResponse(rawSSE, "fallback-model"); + + assert.equal((parsed.content[0] as any).type, "thinking"); + assert.equal((parsed.content[0] as any).thinking, "later thinking"); + assert.equal((parsed.content[0] as any).signature, "sig-before"); +}); + test("parseSSEToClaudeResponse ignores malformed payloads and returns null when nothing valid remains", () => { const parsed = parseSSEToClaudeResponse( ["event: content_block_delta", "data: not-json", "", "data: [DONE]"].join("\n"),