fix(chat): harden non-streaming SSE aggregation (#5746)

2026-08-20 14:12:59 +03:00 · 2026-06-30 23:55:58 -04:00
parent b7fc200e52
commit 9a03840adf
9 changed files with 310 additions and 60 deletions
--- 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.
--- 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<Uint8Array>,
+  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();
--- 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<string, unknown> | Headers | null | undefined) {
+export function isEventStreamAccepted(
+  headers: Record<string, unknown> | 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;
--- 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()}`,
--- 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<string, string> {
  const tokens = getOmniRouteTokenCounts(usage);
  const headers: Record<string, string> = {
-    [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;
--- 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<string>((_, 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 () => {
--- 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';
--- 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);
--- 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"),