fix: stream routed SSE chunks promptly (#3759)

Reworks stream readiness as a ping/zombie filter instead of a semantic content gate: downstream streaming is released as soon as any structured non-ping SSE event arrives, replaying the buffered prefix. Removes the fixed 2s first-byte cap (readiness now inherits REQUEST_TIMEOUT_MS unless STREAM_READINESS_TIMEOUT_MS is set) so slow first-byte reasoning providers no longer false-504. Combo stream quality stays strict: validateResponseQuality still requires an actual content_block / known non-Claude payload before accepting a routed target. Also normalizes multi-line data: framing, metadata-prefixed events, and final events that arrive without a trailing blank line. Integrated into release/v3.8.25. Co-authored-by: R.D. <rogerproself@gmail.com>
2026-07-26 09:52:11 +03:00 · 2026-06-14 00:10:24 -04:00
parent cccb087011
commit 6781a843f2
14 changed files with 1940 additions and 432 deletions
--- a/docs/reference/ENVIRONMENT.md
+++ b/docs/reference/ENVIRONMENT.md
@@ -543,6 +543,7 @@ REQUEST_TIMEOUT_MS (global override)
 │   ├── FETCH_CONNECT_TIMEOUT_MS (independent, default: 30000)
 │   └── FETCH_KEEPALIVE_TIMEOUT_MS (independent, default: 4000)
 ├─→ STREAM_IDLE_TIMEOUT_MS (inherits from REQUEST_TIMEOUT_MS, default: 600000)
+├─→ STREAM_READINESS_TIMEOUT_MS (inherits from REQUEST_TIMEOUT_MS, default: 80000)
 └─→ API_BRIDGE_PROXY_TIMEOUT_MS (inherits from REQUEST_TIMEOUT_MS, default: 30000)
    ├─→ API_BRIDGE_SERVER_REQUEST_TIMEOUT_MS (derived, default: 300000)
    ├── API_BRIDGE_SERVER_HEADERS_TIMEOUT_MS (default: 60000)
@@ -555,6 +556,7 @@ REQUEST_TIMEOUT_MS (global override)
 | `REQUEST_TIMEOUT_MS`                     | _(unset)_            | Global shortcut — overrides both `FETCH_TIMEOUT_MS` and `STREAM_IDLE_TIMEOUT_MS` defaults.                  |
 | `FETCH_TIMEOUT_MS`                       | `600000`             | Total HTTP request timeout for upstream provider calls.                                                     |
 | `STREAM_IDLE_TIMEOUT_MS`                 | `600000`             | Max silence between SSE chunks before aborting. Extended-thinking models rarely pause >90s.                 |
+| `STREAM_READINESS_TIMEOUT_MS`            | `80000`              | Time to receive the first non-ping SSE event. Inherits `REQUEST_TIMEOUT_MS` when set.                       |
 | `FETCH_HEADERS_TIMEOUT_MS`               | = `FETCH_TIMEOUT_MS` | Time to receive response headers.                                                                           |
 | `FETCH_BODY_TIMEOUT_MS`                  | = `FETCH_TIMEOUT_MS` | Time to receive the full response body.                                                                     |
 | `FETCH_CONNECT_TIMEOUT_MS`               | `30000`              | TCP connection establishment timeout.                                                                       |
--- a/open-sse/config/constants.ts
+++ b/open-sse/config/constants.ts
@@ -17,9 +17,9 @@ export const FETCH_TIMEOUT_MS = upstreamTimeouts.fetchTimeoutMs;
 // idle for this duration. Override with STREAM_IDLE_TIMEOUT_MS env var.
 export const STREAM_IDLE_TIMEOUT_MS = upstreamTimeouts.streamIdleTimeoutMs;

-// Timeout for the first useful SSE event. Keep this much shorter than the
-// post-start idle timeout so slow-thinking models can keep streaming after the
-// first token, while dead 200 OK streams fail fast enough for combo fallback.
+// Timeout for the first non-ping SSE event. Inherits REQUEST_TIMEOUT_MS when
+// set, unless STREAM_READINESS_TIMEOUT_MS is specified directly. This must stay
+// conservative for large prompts and slow first-byte reasoning providers.
 export const STREAM_READINESS_TIMEOUT_MS = upstreamTimeouts.streamReadinessTimeoutMs;

 // Error code used when an upstream Antigravity request stalls before response
--- a/open-sse/handlers/chatCore.ts
+++ b/open-sse/handlers/chatCore.ts
@@ -83,6 +83,7 @@ import {

 import {
  getCallLogPipelineCaptureStreamChunks,
+  getCallLogPipelineMaxSizeBytes,
  getChatLogTextLimit,
  getChatLogArrayTailItems,
  getChatLogMaxDepth,
@@ -2302,6 +2303,7 @@ export async function handleChatCore({
  const reqLogger = await createRequestLogger(sourceFormat, targetFormat, model, {
    enabled: detailedLoggingEnabled,
    captureStreamChunks: capturePipelineStreamChunks,
+    maxStreamChunkBytes: getCallLogPipelineMaxSizeBytes(),
    // Provide model/provider/connectionId so streamChunks can be attached to the
    // in-memory pending request record before final call-log persistence.
    model,
--- a/open-sse/services/combo.ts
+++ b/open-sse/services/combo.ts
@@ -22,6 +22,10 @@ import {
 import { FETCH_TIMEOUT_MS, RateLimitReason } from "../config/constants.ts";
 import { errorResponse, unavailableResponse } from "../utils/error.ts";
 import { clamp01 } from "../utils/number.ts";
+import {
+  createSSEDataLineNormalizer,
+  isKnownNonClaudeStreamPayload,
+} from "../utils/streamHelpers.ts";
 import {
  recordComboIntent,
  recordComboRequest,
@@ -490,19 +494,10 @@ export async function validateResponseQuality(
  // detect the empty-content-block pattern (content_filter stop_reason with
  // no content_block_* events) WITHOUT de-streaming non-empty responses.
  //
-  // Strategy:
-  // - Read chunks from response.body one at a time, accumulating raw bytes.
-  // - Parse SSE events incrementally.
-  // - If a content_block_* event appears → stream HAS content. Stop buffering.
-  //   Return a clonedResponse whose body replays buffered bytes then pipes the
-  //   remainder of the original reader. Only the chunks up to the first content
-  //   block were held in memory — the rest stream normally.
-  // - If the stream ends with a complete Claude lifecycle but NO content_block
-  //   → return invalid (combo failover). The empty lifecycle is tiny so fully
-  //   reading it is acceptable.
-  // - If the stream ends without a recognisable complete Claude lifecycle →
-  //   return valid with a clonedResponse replaying all buffered bytes (don't
-  //   misclassify non-Claude or partial streams as empty).
+  // Parse SSE events incrementally. Stop buffering once a content_block_* event
+  // or a known non-Claude SSE payload appears, replay the buffered prefix, then
+  // pipe the original reader so the rest of the stream keeps flowing normally.
+  // Only fail over when a complete Claude lifecycle ends without content_block.
  //
  // Non-text/event-stream streaming responses are not buffered at all.
  if (isStreaming) {
@@ -530,7 +525,7 @@ export async function validateResponseQuality(
    let hasMessageStart = false;
    let hasContentBlock = false;
    let hasLifecycleEnd = false;
-    // `event:` type line seen before the next `data:` line in the same event.
+    const sseLineNormalizer = createSSEDataLineNormalizer();
    let pendingEventType = "";

    /**
@@ -546,8 +541,8 @@ export async function validateResponseQuality(
      // Retain the potentially-incomplete trailing fragment.
      decodedSoFar = lines[lines.length - 1];

-      for (let i = 0; i < lines.length - 1; i++) {
-        const trimmed = lines[i].trim();
+      for (const line of sseLineNormalizer.normalize(lines.slice(0, -1))) {
+        const trimmed = line.trim();

        if (trimmed.startsWith("event:")) {
          pendingEventType = trimmed.slice(6).trim();
@@ -573,6 +568,10 @@ export async function validateResponseQuality(
          (typeof parsed.type === "string" ? parsed.type : null) || pendingEventType || "";
        pendingEventType = "";

+        if (isKnownNonClaudeStreamPayload(parsed, eventType)) {
+          return true;
+        }
+
        switch (eventType) {
          case "message_start":
            hasMessageStart = true;
@@ -651,6 +650,7 @@ export async function validateResponseQuality(
          // Stream finished — flush the TextDecoder and parse any remaining text.
          const tail = decoder.decode(undefined, { stream: false });
          if (tail) decodedSoFar += tail;
+          if (decodedSoFar.trim()) decodedSoFar += "\n\n";
          parseAccumulatedSse();

          if (hasMessageStart && hasLifecycleEnd && !hasContentBlock) {
--- a/open-sse/utils/passthroughTailProcessor.ts
+++ b/open-sse/utils/passthroughTailProcessor.ts
@@ -0,0 +1,303 @@
+import { extractUsage } from "./usageTracking.ts";
+import { parseSSEDataPayload } from "./streamHelpers.ts";
+import {
+  backfillResponsesCompletedOutput,
+  normalizeResponsesSseIds,
+  pushUniqueResponsesOutputItems,
+  stringifyIdValue,
+  stripResponsesLifecycleEcho,
+} from "./responsesStreamHelpers.ts";
+
+type JsonRecord = Record<string, unknown>;
+
+type EventPrefixBuffer = {
+  eventType: () => string;
+  flush: () => string;
+  prefixData: (output: string, line: string) => string;
+  remember: (line: string) => void;
+};
+
+export type PassthroughTailProcessorContext = {
+  getSkipPassthroughEvent: () => boolean;
+  setSkipPassthroughEvent: (value: boolean) => void;
+  clearPendingPassthroughEvent: () => void;
+  shouldAbortOnClaudeLifecycle: (payload: unknown) => boolean;
+  emitClaudeEmptyStreamErrorAndAbort: () => void;
+  isClaudeEventPayload: (payload: unknown) => boolean;
+  updateClaudeEmptyResponseLifecycle: (payload: unknown) => void;
+  passthroughEventPrefix: EventPrefixBuffer;
+  emitConvertedOutput: (output: string) => void;
+  pushProviderPayload: (payload: unknown) => void;
+  pushClientPayload: (payload: unknown) => void;
+  setPassthroughResponsesId: (value: string) => void;
+  setUsage: (value: unknown) => void;
+  addTotalContentLength: (value: number) => void;
+  appendPassthroughContent: (value: string) => void;
+  appendPassthroughReasoning: (value: string) => void;
+  getResponsesReasoningKey: (payload: Record<string, unknown>) => string | null;
+  markResponsesReasoningSummarySeen: (key: string) => void;
+  ensureVisibleResponsesReasoningSummary: (payload: Record<string, unknown>) => boolean;
+  emitSyntheticResponsesReasoningSummary: (payload: Record<string, unknown>) => void;
+  passthroughResponsesOutputItems: unknown[];
+  passthroughResponsesPendingFunctionCalls: Map<string, JsonRecord>;
+  getPassthroughResponsesCurrentFunctionCallKey: () => string | null;
+  setPassthroughResponsesCurrentFunctionCallKey: (value: string | null) => void;
+  hasPassthroughToolCalls: () => boolean;
+  toResponsesCompletedWithToolCalls: (parsed: JsonRecord) => JsonRecord;
+};
+
+function asRecord(value: unknown): JsonRecord {
+  return value && typeof value === "object" && !Array.isArray(value) ? (value as JsonRecord) : {};
+}
+
+function getFunctionCallPendingKey(item: JsonRecord | null): string | null {
+  if (!item) return null;
+  if (typeof item.id === "string") return item.id;
+  if (typeof item.call_id === "string") return item.call_id;
+  return null;
+}
+
+function handleResponsesTailPayload(
+  parsed: JsonRecord,
+  output: string,
+  context: PassthroughTailProcessorContext
+): string {
+  const responsesIdsNormalized = normalizeResponsesSseIds(parsed);
+  const parsedResponse = asRecord(parsed.response);
+  const responseId =
+    (parsedResponse ? stringifyIdValue(parsedResponse.id) : null) ||
+    stringifyIdValue(parsed.response_id);
+  if (responseId) {
+    context.setPassthroughResponsesId(responseId);
+  }
+
+  const extracted = extractUsage(parsed);
+  if (extracted) {
+    context.setUsage(extracted);
+  }
+
+  if (typeof parsed.delta === "string") {
+    context.addTotalContentLength(parsed.delta.length);
+  }
+  if (parsed.type === "response.output_text.delta" && typeof parsed.delta === "string") {
+    context.appendPassthroughContent(parsed.delta);
+  }
+  if (
+    parsed.type === "response.reasoning_summary_text.delta" ||
+    parsed.type === "response.reasoning_summary_text.done" ||
+    parsed.type === "response.reasoning_summary_part.done"
+  ) {
+    const reasoningKey = context.getResponsesReasoningKey(parsed);
+    if (reasoningKey) {
+      context.markResponsesReasoningSummarySeen(reasoningKey);
+    }
+  }
+  if (
+    parsed.type === "response.output_item.added" &&
+    asRecord(parsed.item).type === "function_call"
+  ) {
+    const item = { ...(parsed.item as JsonRecord) };
+    const pendingKey = getFunctionCallPendingKey(item);
+    if (pendingKey) {
+      if (typeof item.arguments !== "string") {
+        item.arguments = "";
+      }
+      context.passthroughResponsesPendingFunctionCalls.set(pendingKey, item);
+      context.setPassthroughResponsesCurrentFunctionCallKey(pendingKey);
+    }
+  }
+  if (parsed.type === "response.function_call_arguments.delta") {
+    const pendingKey =
+      typeof parsed.item_id === "string"
+        ? parsed.item_id
+        : context.getPassthroughResponsesCurrentFunctionCallKey();
+    const pending = pendingKey
+      ? context.passthroughResponsesPendingFunctionCalls.get(pendingKey)
+      : undefined;
+    if (pending && typeof parsed.delta === "string") {
+      const previousArgs = typeof pending.arguments === "string" ? pending.arguments : "";
+      pending.arguments = previousArgs + parsed.delta;
+    }
+  }
+  if (parsed.type === "response.function_call_arguments.done") {
+    const pendingKey =
+      typeof parsed.item_id === "string"
+        ? parsed.item_id
+        : context.getPassthroughResponsesCurrentFunctionCallKey();
+    const pending = pendingKey
+      ? context.passthroughResponsesPendingFunctionCalls.get(pendingKey)
+      : undefined;
+    if (pending) {
+      if (typeof parsed.arguments === "string") {
+        pending.arguments = parsed.arguments;
+      }
+      pushUniqueResponsesOutputItems(context.passthroughResponsesOutputItems, [pending]);
+    }
+  }
+  if (parsed.type === "response.output_item.done" && parsed.item) {
+    const reasoningSummaryInjected = context.ensureVisibleResponsesReasoningSummary(parsed);
+    context.emitSyntheticResponsesReasoningSummary(parsed);
+    pushUniqueResponsesOutputItems(context.passthroughResponsesOutputItems, [parsed.item]);
+    if (reasoningSummaryInjected) {
+      output = `data: ${JSON.stringify(parsed)}\n\n`;
+    }
+    const item = asRecord(parsed.item);
+    if (item.type === "function_call") {
+      const pendingKey = getFunctionCallPendingKey(item);
+      if (pendingKey) {
+        context.passthroughResponsesPendingFunctionCalls.delete(pendingKey);
+        if (context.getPassthroughResponsesCurrentFunctionCallKey() === pendingKey) {
+          context.setPassthroughResponsesCurrentFunctionCallKey(null);
+        }
+      }
+    }
+  }
+  if (
+    parsed.type === "response.completed" &&
+    Array.isArray(asRecord(parsed.response).output) &&
+    (asRecord(parsed.response).output as unknown[]).length > 0
+  ) {
+    pushUniqueResponsesOutputItems(
+      context.passthroughResponsesOutputItems,
+      asRecord(parsed.response).output as unknown[]
+    );
+  }
+  if (
+    parsed.type === "response.completed" &&
+    context.passthroughResponsesPendingFunctionCalls.size > 0
+  ) {
+    pushUniqueResponsesOutputItems(context.passthroughResponsesOutputItems, [
+      ...context.passthroughResponsesPendingFunctionCalls.values(),
+    ]);
+    context.passthroughResponsesPendingFunctionCalls.clear();
+    context.setPassthroughResponsesCurrentFunctionCallKey(null);
+  }
+
+  const textualToolCallBackfilled =
+    parsed.type === "response.completed" && context.hasPassthroughToolCalls();
+  const outputPayload = textualToolCallBackfilled
+    ? context.toResponsesCompletedWithToolCalls(parsed)
+    : parsed;
+  const stripped = stripResponsesLifecycleEcho(outputPayload);
+  const backfilled = backfillResponsesCompletedOutput(
+    outputPayload,
+    context.passthroughResponsesOutputItems
+  );
+
+  if (stripped || backfilled || textualToolCallBackfilled || responsesIdsNormalized) {
+    output = `data: ${JSON.stringify(outputPayload)}\n\n`;
+  }
+
+  return output;
+}
+
+function handleOpenAiTailPayload(parsed: JsonRecord, context: PassthroughTailProcessorContext) {
+  const firstChoice = Array.isArray(parsed.choices)
+    ? (parsed.choices[0] as JsonRecord | undefined)
+    : undefined;
+  const delta = asRecord(firstChoice?.delta);
+  if (typeof delta.content === "string") {
+    context.appendPassthroughContent(delta.content);
+    context.addTotalContentLength(delta.content.length);
+  }
+  const reasoningDelta =
+    typeof delta.reasoning_content === "string"
+      ? delta.reasoning_content
+      : typeof delta.reasoning === "string"
+        ? delta.reasoning
+        : "";
+  if (reasoningDelta) {
+    context.appendPassthroughReasoning(reasoningDelta);
+  }
+}
+
+export function processBufferedPassthroughLine(
+  line: string,
+  context: PassthroughTailProcessorContext
+): boolean {
+  const trimmed = line.trim();
+
+  if (context.getSkipPassthroughEvent()) {
+    if (!trimmed) {
+      context.setSkipPassthroughEvent(false);
+      context.clearPendingPassthroughEvent();
+    }
+    return false;
+  }
+
+  if (/^event:\s*keepalive\b/i.test(trimmed)) {
+    context.setSkipPassthroughEvent(true);
+    context.clearPendingPassthroughEvent();
+    return false;
+  }
+
+  if (/^event:/i.test(trimmed)) {
+    const eventType = trimmed.replace(/^event:\s*/i, "");
+    if (context.shouldAbortOnClaudeLifecycle({ type: eventType })) {
+      context.emitClaudeEmptyStreamErrorAndAbort();
+      return true;
+    }
+    context.passthroughEventPrefix.remember(line);
+    return false;
+  }
+
+  if (/^(?::|id:|retry:)/i.test(trimmed)) {
+    context.passthroughEventPrefix.remember(line);
+    return false;
+  }
+
+  if (!trimmed) {
+    const pendingOutput = context.passthroughEventPrefix.flush();
+    if (pendingOutput) {
+      context.emitConvertedOutput(pendingOutput);
+    }
+    context.clearPendingPassthroughEvent();
+    return false;
+  }
+
+  if (!trimmed.startsWith("data:")) {
+    context.passthroughEventPrefix.remember(line);
+    return false;
+  }
+
+  const parsedPassthroughData = parseSSEDataPayload(trimmed.slice(5), {
+    eventType: context.passthroughEventPrefix.eventType(),
+  });
+  if (parsedPassthroughData?.done === true) {
+    return false;
+  }
+
+  let output =
+    line.startsWith("data:") && !line.startsWith("data: ")
+      ? `data: ${line.slice(5)}\n\n`
+      : `${line}\n\n`;
+
+  if (parsedPassthroughData) {
+    context.pushProviderPayload(parsedPassthroughData);
+    if (context.shouldAbortOnClaudeLifecycle(parsedPassthroughData)) {
+      context.emitClaudeEmptyStreamErrorAndAbort();
+      return true;
+    }
+
+    if (context.isClaudeEventPayload(parsedPassthroughData)) {
+      context.updateClaudeEmptyResponseLifecycle(parsedPassthroughData);
+    }
+
+    const parsed = parsedPassthroughData as JsonRecord;
+    const parsedType = typeof parsed.type === "string" ? parsed.type : "";
+    const isResponses = parsedType.startsWith("response.");
+    const isClaude = context.isClaudeEventPayload(parsed);
+
+    if (isResponses) {
+      output = handleResponsesTailPayload(parsed, output, context);
+    } else if (!isClaude) {
+      handleOpenAiTailPayload(parsed, context);
+    }
+
+    context.pushClientPayload(parsed);
+  }
+
+  output = context.passthroughEventPrefix.prefixData(output, line);
+  context.emitConvertedOutput(output);
+  return false;
+}
--- a/open-sse/utils/responsesStreamHelpers.ts
+++ b/open-sse/utils/responsesStreamHelpers.ts
@@ -0,0 +1,166 @@
+type JsonRecord = Record<string, unknown>;
+
+export function stringifyIdValue(value: unknown): string | null {
+  return value === null || value === undefined ? null : String(value);
+}
+
+function normalizeResponsesOutputItemIds(item: unknown): unknown {
+  if (!item || typeof item !== "object" || Array.isArray(item)) {
+    return item;
+  }
+
+  const record = item as JsonRecord;
+  let changed = false;
+  const normalized = { ...record };
+
+  const id = stringifyIdValue(record.id);
+  if (id !== null && record.id !== id) {
+    normalized.id = id;
+    changed = true;
+  }
+
+  const callId = stringifyIdValue(record.call_id);
+  if (callId !== null && record.call_id !== callId) {
+    normalized.call_id = callId;
+    changed = true;
+  }
+
+  return changed ? normalized : item;
+}
+
+export function normalizeResponsesSseIds(payload: JsonRecord): boolean {
+  let changed = false;
+
+  for (const key of ["response_id", "item_id", "call_id"] as const) {
+    const value = stringifyIdValue(payload[key]);
+    if (value !== null && payload[key] !== value) {
+      payload[key] = value;
+      changed = true;
+    }
+  }
+
+  if (payload.item && typeof payload.item === "object" && !Array.isArray(payload.item)) {
+    const normalizedItem = normalizeResponsesOutputItemIds(payload.item);
+    if (normalizedItem !== payload.item) {
+      payload.item = normalizedItem;
+      changed = true;
+    }
+  }
+
+  if (
+    payload.response &&
+    typeof payload.response === "object" &&
+    !Array.isArray(payload.response)
+  ) {
+    const response = payload.response as JsonRecord;
+    let responseChanged = false;
+    const normalizedResponse = { ...response };
+
+    const responseId = stringifyIdValue(response.id);
+    if (responseId !== null && response.id !== responseId) {
+      normalizedResponse.id = responseId;
+      responseChanged = true;
+    }
+
+    if (Array.isArray(response.output)) {
+      const normalizedOutput = response.output.map(normalizeResponsesOutputItemIds);
+      if (normalizedOutput.some((item, index) => item !== response.output[index])) {
+        normalizedResponse.output = normalizedOutput;
+        responseChanged = true;
+      }
+    }
+
+    if (responseChanged) {
+      payload.response = normalizedResponse;
+      changed = true;
+    }
+  }
+
+  return changed;
+}
+
+function buildResponsesOutputItemKey(item: unknown): string | null {
+  if (!item || typeof item !== "object" || Array.isArray(item)) {
+    return null;
+  }
+
+  const record = item as JsonRecord;
+  const type = typeof record.type === "string" ? record.type : "";
+  const id = stringifyIdValue(record.id) ?? "";
+  const callId = stringifyIdValue(record.call_id) ?? "";
+  const outputIndex = typeof record.output_index === "number" ? record.output_index : "";
+  const name = typeof record.name === "string" ? record.name : "";
+
+  if (!type && !id && !callId) {
+    return null;
+  }
+
+  return `${type}:${id}:${callId}:${outputIndex}:${name}`;
+}
+
+export function pushUniqueResponsesOutputItems(target: unknown[], items: readonly unknown[]) {
+  const seen = new Set<string>();
+
+  for (const existingItem of target) {
+    const key = buildResponsesOutputItemKey(existingItem);
+    if (key) {
+      seen.add(key);
+    }
+  }
+
+  for (const item of items) {
+    const key = buildResponsesOutputItemKey(item);
+    if (key && seen.has(key)) {
+      continue;
+    }
+
+    target.push(item);
+    if (key) {
+      seen.add(key);
+    }
+  }
+}
+
+export function backfillResponsesCompletedOutput(
+  parsed: unknown,
+  collectedItems: readonly unknown[]
+): boolean {
+  if (!collectedItems.length) return false;
+  if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return false;
+  const obj = parsed as Record<string, unknown>;
+  if (obj.type !== "response.completed") return false;
+  const resp = obj.response;
+  if (!resp || typeof resp !== "object" || Array.isArray(resp)) return false;
+  const r = resp as Record<string, unknown>;
+  const existing = r.output;
+  if (Array.isArray(existing) && existing.length > 0) return false;
+  r.output = collectedItems.slice();
+  return true;
+}
+
+const RESPONSES_LIFECYCLE_EVENT_TYPES = new Set([
+  "response.created",
+  "response.in_progress",
+  "response.completed",
+]);
+
+export function stripResponsesLifecycleEcho(parsed: unknown): boolean {
+  if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return false;
+  const obj = parsed as Record<string, unknown>;
+  if (typeof obj.type !== "string" || !RESPONSES_LIFECYCLE_EVENT_TYPES.has(obj.type)) {
+    return false;
+  }
+  const resp = obj.response;
+  if (!resp || typeof resp !== "object" || Array.isArray(resp)) return false;
+  const r = resp as Record<string, unknown>;
+  let changed = false;
+  if ("instructions" in r) {
+    delete r.instructions;
+    changed = true;
+  }
+  if ("tools" in r) {
+    delete r.tools;
+    changed = true;
+  }
+  return changed;
+}
--- a/open-sse/utils/stream.ts
+++ b/open-sse/utils/stream.ts
@@ -12,6 +12,9 @@ import {
 } from "./usageTracking.ts";
 import {
  parseSSELine,
+  parseSSEDataPayload,
+  createSSEDataLineNormalizer,
+  createSSEEventPrefixBuffer,
  hasValuableContent,
  fixInvalidId,
  formatSSE,
@@ -27,7 +30,6 @@ import { STREAM_IDLE_TIMEOUT_MS, FETCH_BODY_TIMEOUT_MS, HTTP_STATUS } from "../c
 import {
  OMIT_STREAMING_CHUNK_MARKER,
  sanitizeStreamingChunk,
-  extractThinkingFromContent,
 } from "../handlers/responseSanitizer.ts";
 import { buildErrorBody } from "./error.ts";
 import { parseTextualToolCallCandidate, isValidToolCallHeaderPrefix } from "./textualToolCall.ts";
@@ -37,6 +39,14 @@ import {
  markToolFinish,
  consumeToolFinishTime,
 } from "../services/sessionManager.ts";
+import {
+  backfillResponsesCompletedOutput,
+  normalizeResponsesSseIds,
+  pushUniqueResponsesOutputItems,
+  stringifyIdValue,
+  stripResponsesLifecycleEcho,
+} from "./responsesStreamHelpers.ts";
+import { processBufferedPassthroughLine } from "./passthroughTailProcessor.ts";

 /**
 * Race a response body read against a timeout.
@@ -59,88 +69,10 @@ export function withBodyTimeout<T>(
 }

 export { COLORS, formatSSE };
+export { backfillResponsesCompletedOutput, stripResponsesLifecycleEcho };

 type JsonRecord = Record<string, unknown>;

-function stringifyIdValue(value: unknown): string | null {
-  return value === null || value === undefined ? null : String(value);
-}
-
-function normalizeResponsesOutputItemIds(item: unknown): unknown {
-  if (!item || typeof item !== "object" || Array.isArray(item)) {
-    return item;
-  }
-
-  const record = item as JsonRecord;
-  let changed = false;
-  const normalized = { ...record };
-
-  const id = stringifyIdValue(record.id);
-  if (id !== null && record.id !== id) {
-    normalized.id = id;
-    changed = true;
-  }
-
-  const callId = stringifyIdValue(record.call_id);
-  if (callId !== null && record.call_id !== callId) {
-    normalized.call_id = callId;
-    changed = true;
-  }
-
-  return changed ? normalized : item;
-}
-
-function normalizeResponsesSseIds(payload: JsonRecord): boolean {
-  let changed = false;
-
-  for (const key of ["response_id", "item_id", "call_id"] as const) {
-    const value = stringifyIdValue(payload[key]);
-    if (value !== null && payload[key] !== value) {
-      payload[key] = value;
-      changed = true;
-    }
-  }
-
-  if (payload.item && typeof payload.item === "object" && !Array.isArray(payload.item)) {
-    const normalizedItem = normalizeResponsesOutputItemIds(payload.item);
-    if (normalizedItem !== payload.item) {
-      payload.item = normalizedItem;
-      changed = true;
-    }
-  }
-
-  if (
-    payload.response &&
-    typeof payload.response === "object" &&
-    !Array.isArray(payload.response)
-  ) {
-    const response = payload.response as JsonRecord;
-    let responseChanged = false;
-    const normalizedResponse = { ...response };
-
-    const responseId = stringifyIdValue(response.id);
-    if (responseId !== null && response.id !== responseId) {
-      normalizedResponse.id = responseId;
-      responseChanged = true;
-    }
-
-    if (Array.isArray(response.output)) {
-      const normalizedOutput = response.output.map(normalizeResponsesOutputItemIds);
-      if (normalizedOutput.some((item, index) => item !== response.output[index])) {
-        normalizedResponse.output = normalizedOutput;
-        responseChanged = true;
-      }
-    }
-
-    if (responseChanged) {
-      payload.response = normalizedResponse;
-      changed = true;
-    }
-  }
-
-  return changed;
-}
-
 export const PENDING_REQUEST_CLEARED_MARKER = "__omniroutePendingRequestCleared";

 function markPendingRequestCleared(error: Error): Error {
@@ -148,48 +80,6 @@ function markPendingRequestCleared(error: Error): Error {
  return error;
 }

-function buildResponsesOutputItemKey(item: unknown): string | null {
-  if (!item || typeof item !== "object" || Array.isArray(item)) {
-    return null;
-  }
-
-  const record = item as JsonRecord;
-  const type = typeof record.type === "string" ? record.type : "";
-  const id = stringifyIdValue(record.id) ?? "";
-  const callId = stringifyIdValue(record.call_id) ?? "";
-  const outputIndex = typeof record.output_index === "number" ? record.output_index : "";
-  const name = typeof record.name === "string" ? record.name : "";
-
-  if (!type && !id && !callId) {
-    return null;
-  }
-
-  return `${type}:${id}:${callId}:${outputIndex}:${name}`;
-}
-
-function pushUniqueResponsesOutputItems(target: unknown[], items: readonly unknown[]) {
-  const seen = new Set<string>();
-
-  for (const existingItem of target) {
-    const key = buildResponsesOutputItemKey(existingItem);
-    if (key) {
-      seen.add(key);
-    }
-  }
-
-  for (const item of items) {
-    const key = buildResponsesOutputItemKey(item);
-    if (key && seen.has(key)) {
-      continue;
-    }
-
-    target.push(item);
-    if (key) {
-      seen.add(key);
-    }
-  }
-}
-
 type StreamLogger = {
  appendProviderChunk?: (value: string) => void;
  appendConvertedChunk?: (value: string) => void;
@@ -394,7 +284,7 @@ function buildResponsesFunctionCallEvents(toolCall: ToolCall) {
 }

 function formatSSEDataEvents(events: unknown[]) {
-  return events.map((event) => `data: ${JSON.stringify(event)}\n`).join("\n");
+  return events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join("");
 }

 function toChatCompletionChunkWithToolCall(base: JsonRecord, toolCall: ToolCall) {
@@ -684,80 +574,6 @@ const STREAM_MODE = {
  PASSTHROUGH: "passthrough", // No translation, normalize output, extract usage
 };

-/**
- * Lifecycle event types in OpenAI Responses API streams whose `response`
- * payload is a snapshot of the request (echoes back `instructions` + `tools`).
- */
-const RESPONSES_LIFECYCLE_EVENT_TYPES = new Set([
-  "response.created",
-  "response.in_progress",
-  "response.completed",
-]);
-
-/**
- * Backfill `parsed.response.output` on a `response.completed` event from the
- * snapshots accumulated as the stream progressed (`response.output_item.done`).
- *
- * Why: when the upstream request runs with `store: false`, OpenAI's Responses
- * API leaves `response.output` empty in the final `response.completed`
- * snapshot — clients that rebuild assistant messages from that snapshot
- * (notably the GitHub Copilot CLI 1.0.36) end up with `choices: []` and never
- * trigger tool execution. Codex CLI and others that consume per-item events
- * are unaffected; backfilling the array makes both styles work.
- *
- * Returns true when `parsed.response.output` was empty and got replaced, so
- * the caller can re-serialize.
- */
-export function backfillResponsesCompletedOutput(
-  parsed: unknown,
-  collectedItems: readonly unknown[]
-): boolean {
-  if (!collectedItems.length) return false;
-  if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return false;
-  const obj = parsed as Record<string, unknown>;
-  if (obj.type !== "response.completed") return false;
-  const resp = obj.response;
-  if (!resp || typeof resp !== "object" || Array.isArray(resp)) return false;
-  const r = resp as Record<string, unknown>;
-  const existing = r.output;
-  if (Array.isArray(existing) && existing.length > 0) return false;
-  r.output = collectedItems.slice();
-  return true;
-}
-
-/**
- * Strip the request echo (`instructions`, `tools`) from `parsed.response`
- * on Responses API lifecycle events.
- *
- * Why: those fields can balloon the SSE message past 100 KB when the request
- * carries large tool definitions / instructions. Some clients (notably the
- * GitHub Copilot CLI) cannot process oversized SSE events and stop rendering
- * mid-stream. The fields are pure echo of the original request — clients
- * already hold the original locally — so removing them is observably safe.
- *
- * Returns true when the payload was modified and must be re-serialized.
- */
-export function stripResponsesLifecycleEcho(parsed: unknown): boolean {
-  if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return false;
-  const obj = parsed as Record<string, unknown>;
-  if (typeof obj.type !== "string" || !RESPONSES_LIFECYCLE_EVENT_TYPES.has(obj.type)) {
-    return false;
-  }
-  const resp = obj.response;
-  if (!resp || typeof resp !== "object" || Array.isArray(resp)) return false;
-  const r = resp as Record<string, unknown>;
-  let changed = false;
-  if ("instructions" in r) {
-    delete r.instructions;
-    changed = true;
-  }
-  if ("tools" in r) {
-    delete r.tools;
-    changed = true;
-  }
-  return changed;
-}
-
 /**
 * Create unified SSE transform stream with idle timeout protection.
 * If the upstream provider stops sending data for STREAM_IDLE_TIMEOUT_MS,
@@ -894,8 +710,8 @@ export function createSSEStream(options: StreamOptions = {}) {
  let idleTimer: ReturnType<typeof setInterval> | null = null;
  let streamTimedOut = false;
  const claudeEmptyResponseLifecycle = createClaudeEmptyResponseLifecycle();
-  let pendingPassthroughEventLine: string | null = null;
-  let pendingPassthroughEventEmitted = false;
+  const passthroughEventPrefix = createSSEEventPrefixBuffer();
+  const multilineSseDataLineNormalizer = createSSEDataLineNormalizer();

  const clearIdleTimer = () => {
    if (idleTimer) {
@@ -905,19 +721,7 @@ export function createSSEStream(options: StreamOptions = {}) {
  };

  const clearPendingPassthroughEvent = () => {
-    pendingPassthroughEventLine = null;
-    pendingPassthroughEventEmitted = false;
-  };
-
-  const maybePrefixPendingPassthroughEvent = (output: string, line: string) => {
-    if (!pendingPassthroughEventLine || !line.startsWith("data:")) {
-      return output;
-    }
-    if (!pendingPassthroughEventEmitted) {
-      pendingPassthroughEventEmitted = true;
-      return `${pendingPassthroughEventLine}\n${output}`;
-    }
-    return output;
+    passthroughEventPrefix.clear();
  };

  const applyTextualToolCallStreamingGuard = (parsed: Record<string, unknown>) => {
@@ -1054,15 +858,6 @@ export function createSSEStream(options: StreamOptions = {}) {
    const isResponsesEvent = typeof item?.event === "string" && item.event.startsWith("response.");
    if (sourceFormat === FORMATS.OPENAI && !isResponsesEvent) {
      itemSanitized = sanitizeStreamingChunk(itemSanitized) as Record<string, unknown>;
-
-      const delta = itemSanitized?.choices?.[0]?.delta;
-      if (delta?.content && typeof delta.content === "string") {
-        const { content, thinking } = extractThinkingFromContent(delta.content);
-        delta.content = content;
-        if (thinking && !delta.reasoning_content) {
-          delta.reasoning_content = thinking;
-        }
-      }
    }

    if (!hasValuableContent(itemSanitized, sourceFormat)) {
@@ -1286,7 +1081,7 @@ export function createSSEStream(options: StreamOptions = {}) {
        const lines = buffer.split("\n");
        buffer = lines.pop() || "";

-        for (const line of lines) {
+        for (const line of multilineSseDataLineNormalizer.normalize(lines)) {
          const trimmed = line.trim();

          // Passthrough mode: normalize and forward
@@ -1313,12 +1108,6 @@ export function createSSEStream(options: StreamOptions = {}) {
            }

            if (/^event:/i.test(trimmed)) {
-              if (pendingPassthroughEventLine && !pendingPassthroughEventEmitted) {
-                const pendingOutput = `${pendingPassthroughEventLine}\n`;
-                reqLogger?.appendConvertedChunk?.(pendingOutput);
-                controller.enqueue(encoder.encode(pendingOutput));
-              }
-
              const eventType = trimmed.replace(/^event:\s*/i, "");
              if (
                shouldInjectClaudeEmptyResponseBeforeCurrentEvent(claudeEmptyResponseLifecycle, {
@@ -1329,13 +1118,38 @@ export function createSSEStream(options: StreamOptions = {}) {
                return;
              }

-              pendingPassthroughEventLine = line;
-              pendingPassthroughEventEmitted = false;
+              passthroughEventPrefix.remember(line);
              continue;
            }

+            if (/^(?::|id:|retry:)/i.test(trimmed)) {
+              passthroughEventPrefix.remember(line);
+              continue;
+            }
+
+            if (!trimmed) {
+              const pendingOutput = passthroughEventPrefix.flush();
+              if (pendingOutput) {
+                reqLogger?.appendConvertedChunk?.(pendingOutput);
+                controller.enqueue(encoder.encode(pendingOutput));
+              }
+              clearPendingPassthroughEvent();
+              continue;
+            }
+
+            if (!trimmed.startsWith("data:")) {
+              passthroughEventPrefix.remember(line);
+              continue;
+            }
+
+            const parsedPassthroughData = trimmed.startsWith("data:")
+              ? parseSSEDataPayload(trimmed.slice(5), {
+                  eventType: passthroughEventPrefix.eventType(),
+                })
+              : null;
+
            if (trimmed.startsWith("data:")) {
-              const providerPayload = parseSSELine(trimmed);
+              const providerPayload = parsedPassthroughData ?? parseSSELine(trimmed);
              if (providerPayload) {
                providerPayloadCollector.push(providerPayload);
                if ((providerPayload as { done?: unknown }).done === true) {
@@ -1350,7 +1164,7 @@ export function createSSEStream(options: StreamOptions = {}) {

            if (trimmed.startsWith("data:") && trimmed.slice(5).trim() !== "[DONE]") {
              try {
-                let parsed = JSON.parse(trimmed.slice(5).trim());
+                let parsed = parsedPassthroughData ?? JSON.parse(trimmed.slice(5).trim());

                // Some upstream Responses-compatible providers leak an initial Chat Completions
                // bootstrap chunk (assistant role + empty content) before emitting proper
@@ -1464,8 +1278,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                          controller.enqueue(encoder.encode(output));
                          injectedUsage = true;
                        } else {
-                          output = `data: ${JSON.stringify(parsed)}
-`;
+                          output = `data: ${JSON.stringify(parsed)}\n\n`;
                          injectedUsage = true;
                        }
                        passthroughBufferedTextualToolCallContent = "";
@@ -1476,12 +1289,12 @@ export function createSSEStream(options: StreamOptions = {}) {
                          incomingDelta
                        );
                        parsed.delta = "";
-                        output = `data: ${JSON.stringify(parsed)}\n`;
+                        output = `data: ${JSON.stringify(parsed)}\n\n`;
                        injectedUsage = true;
                      } else {
                        if (passthroughBufferedTextualToolCallContent) {
                          parsed.delta = passthroughBufferedTextualToolCallContent + incomingDelta;
-                          output = `data: ${JSON.stringify(parsed)}\n`;
+                          output = `data: ${JSON.stringify(parsed)}\n\n`;
                          injectedUsage = true;
                        }
                        passthroughAccumulatedContent = appendBoundedText(
@@ -1569,7 +1382,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                    emitSyntheticResponsesReasoningSummary(controller, parsed);
                    pushUniqueResponsesOutputItems(passthroughResponsesOutputItems, [parsed.item]);
                    if (reasoningSummaryInjected) {
-                      output = `data: ${JSON.stringify(parsed)}\n`;
+                      output = `data: ${JSON.stringify(parsed)}\n\n`;
                      injectedUsage = true;
                    }
                    if (parsed.item?.type === "function_call") {
@@ -1633,7 +1446,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                    textualToolCallBackfilled ||
                    responsesIdsNormalized
                  ) {
-                    output = `data: ${JSON.stringify(parsed)}\n`;
+                    output = `data: ${JSON.stringify(parsed)}\n\n`;
                    injectedUsage = true;
                  }
                } else if (isClaudeSSE) {
@@ -1680,7 +1493,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                    );
                  }
                  if (restoredToolName) {
-                    output = `data: ${JSON.stringify(parsed)}\n`;
+                    output = `data: ${JSON.stringify(parsed)}\n\n`;
                    injectedUsage = true;
                  }
                } else {
@@ -1708,7 +1521,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                    const emptyChoicesUsage = extractUsage(parsed) ?? parsed.usage;
                    if (hasValidUsage(emptyChoicesUsage)) {
                      usage = emptyChoicesUsage;
-                      output = `data: ${JSON.stringify(parsed)}\n`;
+                      output = `data: ${JSON.stringify(parsed)}\n\n`;
                      injectedUsage = true;
                      clientPayload = parsed;
                      clientPayloadCollector.push(clientPayload);
@@ -1723,12 +1536,6 @@ export function createSSEStream(options: StreamOptions = {}) {
                    continue;
                  }

-                  // Detect reasoning alias before sanitization strips it
-                  const hadReasoningAlias = !!(
-                    parsed.choices?.[0]?.delta?.reasoning &&
-                    typeof parsed.choices[0].delta.reasoning === "string" &&
-                    !parsed.choices[0].delta.reasoning_content
-                  );
                  const hadNonStringToolCallId = Array.isArray(parsed.choices)
                    ? parsed.choices.some(
                        (choice) =>
@@ -1740,6 +1547,11 @@ export function createSSEStream(options: StreamOptions = {}) {
                    : false;
                  const hadNonStringTopLevelId =
                    parsed?.id != null && typeof parsed.id !== "string";
+                  const hadReasoningAlias = !!(
+                    parsed.choices?.[0]?.delta?.reasoning &&
+                    typeof parsed.choices[0].delta.reasoning === "string" &&
+                    !parsed.choices[0].delta.reasoning_content
+                  );

                  parsed = sanitizeStreamingChunk(parsed);
                  if (
@@ -1761,15 +1573,6 @@ export function createSSEStream(options: StreamOptions = {}) {
                  let textualToolCallConverted = false;
                  let toolCallIdCoerced = false;

-                  // Extract <think> tags from streaming content
-                  if (delta?.content && typeof delta.content === "string") {
-                    const { content, thinking } = extractThinkingFromContent(delta.content);
-                    delta.content = content;
-                    if (thinking && !delta.reasoning_content) {
-                      delta.reasoning_content = thinking;
-                    }
-                  }
-
                  // Split combined reasoning+content deltas into separate SSE events.
                  // Standard OpenAI streaming never mixes both fields in one delta;
                  // clients (e.g. LobeChat) may skip content when reasoning_content
@@ -1780,7 +1583,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                    delete rDelta.content;
                    reasoningChunk.choices[0].finish_reason = null;
                    delete reasoningChunk.usage;
-                    const rOutput = `data: ${JSON.stringify(reasoningChunk)}\n`;
+                    const rOutput = `data: ${JSON.stringify(reasoningChunk)}\n\n`;
                    passthroughAccumulatedReasoning = appendBoundedText(
                      passthroughAccumulatedReasoning,
                      delta.reasoning_content
@@ -1789,7 +1592,6 @@ export function createSSEStream(options: StreamOptions = {}) {
                    clientPayloadCollector.push(reasoningChunk);
                    reqLogger?.appendConvertedChunk?.(rOutput);
                    controller.enqueue(encoder.encode(rOutput));
-                    controller.enqueue(encoder.encode("\n"));
                    delete delta.reasoning_content;
                  }

@@ -1898,7 +1700,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                    parsed.choices[0].finish_reason = "tool_calls";
                    // If we modify it, we must output the modified object
                    if (!injectedUsage && hasValidUsage(parsed.usage)) {
-                      output = `data: ${JSON.stringify(parsed)}\n`;
+                      output = `data: ${JSON.stringify(parsed)}\n\n`;
                      injectedUsage = true;
                    }
                  }
@@ -1909,16 +1711,16 @@ export function createSSEStream(options: StreamOptions = {}) {
                  ) {
                    const estimated = estimateUsage(body, totalContentLength, FORMATS.OPENAI);
                    parsed.usage = filterUsageForFormat(estimated, FORMATS.OPENAI);
-                    output = `data: ${JSON.stringify(parsed)}\n`;
+                    output = `data: ${JSON.stringify(parsed)}\n\n`;
                    usage = estimated;
                    injectedUsage = true;
                  } else if (isFinishChunk && usage) {
                    const buffered = addBufferToUsage(usage);
                    parsed.usage = filterUsageForFormat(buffered, FORMATS.OPENAI);
-                    output = `data: ${JSON.stringify(parsed)}\n`;
+                    output = `data: ${JSON.stringify(parsed)}\n\n`;
                    injectedUsage = true;
                  } else if (textualToolCallConverted) {
-                    output = `data: ${JSON.stringify(parsed)}\n`;
+                    output = `data: ${JSON.stringify(parsed)}\n\n`;
                    injectedUsage = true;
                  } else if (
                    idFixed ||
@@ -1927,7 +1729,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                    hadNonStringToolCallId ||
                    hadNonStringTopLevelId
                  ) {
-                    output = `data: ${JSON.stringify(parsed)}\n`;
+                    output = `data: ${JSON.stringify(parsed)}\n\n`;
                    injectedUsage = true;
                  }
                }
@@ -1938,18 +1740,13 @@ export function createSSEStream(options: StreamOptions = {}) {

            if (!injectedUsage) {
              if (line.startsWith("data:") && !line.startsWith("data: ")) {
-                output = "data: " + line.slice(5) + "\n";
+                output = "data: " + line.slice(5) + "\n\n";
              } else {
-                output = line + "\n";
+                output = line + "\n\n";
              }
            }

-            if (!trimmed && pendingPassthroughEventLine && !pendingPassthroughEventEmitted) {
-              output = `${pendingPassthroughEventLine}\n${output}`;
-              pendingPassthroughEventEmitted = true;
-            }
-
-            output = maybePrefixPendingPassthroughEvent(output, line);
+            output = passthroughEventPrefix.prefixData(output, line);

            if (clientPayload) {
              clientPayloadCollector.push(clientPayload);
@@ -2152,8 +1949,85 @@ export function createSSEStream(options: StreamOptions = {}) {
        try {
          const remaining = decoder.decode();
          if (remaining) buffer += remaining;
+          let normalizedTailLines: string[] = [];
+          if (multilineSseDataLineNormalizer.hasPending()) {
+            const tailLines = buffer ? [buffer, ""] : [""];
+            normalizedTailLines = multilineSseDataLineNormalizer.normalize(tailLines);
+            buffer = "";
+          }

          if (mode === STREAM_MODE.PASSTHROUGH) {
+            const tailProcessorContext = {
+              getSkipPassthroughEvent: () => skipPassthroughEvent,
+              setSkipPassthroughEvent: (value: boolean) => {
+                skipPassthroughEvent = value;
+              },
+              clearPendingPassthroughEvent,
+              shouldAbortOnClaudeLifecycle: (payload: unknown) =>
+                shouldInjectClaudeEmptyResponseBeforeCurrentEvent(
+                  claudeEmptyResponseLifecycle,
+                  payload
+                ),
+              emitClaudeEmptyStreamErrorAndAbort: () =>
+                emitClaudeEmptyStreamErrorAndAbort(controller, false),
+              isClaudeEventPayload,
+              updateClaudeEmptyResponseLifecycle: (payload: unknown) =>
+                updateClaudeEmptyResponseLifecycle(claudeEmptyResponseLifecycle, payload),
+              passthroughEventPrefix,
+              emitConvertedOutput: (output: string) => {
+                reqLogger?.appendConvertedChunk?.(output);
+                controller.enqueue(encoder.encode(output));
+              },
+              pushProviderPayload: (payload: unknown) => providerPayloadCollector.push(payload),
+              pushClientPayload: (payload: unknown) => clientPayloadCollector.push(payload),
+              setPassthroughResponsesId: (value: string) => {
+                passthroughResponsesId = value;
+              },
+              setUsage: (value: unknown) => {
+                usage = value as UsageTokenRecord;
+              },
+              addTotalContentLength: (value: number) => {
+                totalContentLength += value;
+              },
+              appendPassthroughContent: (value: string) => {
+                passthroughAccumulatedContent = appendBoundedText(
+                  passthroughAccumulatedContent,
+                  value
+                );
+              },
+              appendPassthroughReasoning: (value: string) => {
+                passthroughAccumulatedReasoning = appendBoundedText(
+                  passthroughAccumulatedReasoning,
+                  value
+                );
+              },
+              getResponsesReasoningKey,
+              markResponsesReasoningSummarySeen: (key: string) => {
+                passthroughResponsesReasoningSummarySeen.add(key);
+              },
+              ensureVisibleResponsesReasoningSummary,
+              emitSyntheticResponsesReasoningSummary: (payload: Record<string, unknown>) =>
+                emitSyntheticResponsesReasoningSummary(controller, payload),
+              passthroughResponsesOutputItems,
+              passthroughResponsesPendingFunctionCalls,
+              getPassthroughResponsesCurrentFunctionCallKey: () =>
+                passthroughResponsesCurrentFunctionCallKey,
+              setPassthroughResponsesCurrentFunctionCallKey: (value: string | null) => {
+                passthroughResponsesCurrentFunctionCallKey = value;
+              },
+              hasPassthroughToolCalls: () => passthroughToolCalls.size > 0,
+              toResponsesCompletedWithToolCalls: (parsed: JsonRecord) =>
+                toResponsesCompletedWithToolCalls(parsed, [
+                  ...passthroughToolCalls.values(),
+                ]) as JsonRecord,
+            };
+
+            for (const line of normalizedTailLines) {
+              if (processBufferedPassthroughLine(line, tailProcessorContext)) {
+                return;
+              }
+            }
+
            const bufferedLine = buffer.trim();
            if (skipPassthroughEvent || /^event:\s*keepalive\b/i.test(bufferedLine)) {
              skipPassthroughEvent = false;
@@ -2189,7 +2063,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                  const isClaude = isClaudeEventPayload(flushedParsed);
                  if (isResponses) {
                    if (normalizeResponsesSseIds(flushedParsed)) {
-                      output = `data: ${JSON.stringify(flushedParsed)}\n`;
+                      output = `data: ${JSON.stringify(flushedParsed)}\n\n`;
                    }
                  } else if (!isClaude) {
                    let flushChanged = false;
@@ -2215,16 +2089,16 @@ export function createSSEStream(options: StreamOptions = {}) {
                      }
                    }
                    if (flushChanged) {
-                      output = `data: ${JSON.stringify(flushedParsed)}\n`;
+                      output = `data: ${JSON.stringify(flushedParsed)}\n\n`;
                    }
                  }
                }
              }
-              if (!bufferedLine && pendingPassthroughEventLine && !pendingPassthroughEventEmitted) {
-                output = `${pendingPassthroughEventLine}\n${output}`;
-                pendingPassthroughEventEmitted = true;
+              if (!bufferedLine) output = passthroughEventPrefix.flush() || output;
+              output = passthroughEventPrefix.prefixData(output, buffer);
+              if (output && !output.endsWith("\n\n")) {
+                output = output.endsWith("\n") ? `${output}\n` : `${output}\n\n`;
              }
-              output = maybePrefixPendingPassthroughEvent(output, buffer);
              reqLogger?.appendConvertedChunk?.(output);
              controller.enqueue(encoder.encode(output));
            }
--- a/open-sse/utils/streamHelpers.ts
+++ b/open-sse/utils/streamHelpers.ts
@@ -13,29 +13,302 @@

 import { FORMATS } from "../translator/formats.ts";

-// Parse SSE data line
-export function parseSSELine(line) {
-  if (!line) return null;
+type SSEPayloadOptions = {
+  eventType?: string;
+  logWarning?: boolean;
+};

-  // Trim leading whitespace before checking prefix character
-  const trimmed = line.trimStart();
-  if (!trimmed || trimmed.charCodeAt(0) !== 100) return null; // 'd' = 100
+type SSEChoicePayload = {
+  delta?: Record<string, unknown> & { tool_calls?: unknown };
+  finish_reason?: unknown;
+  [key: string]: unknown;
+};

-  const data = trimmed.slice(5).trim();
-  if (data === "[DONE]") return { done: true };
+type SSEJsonPayload = Record<string, unknown> & {
+  done?: boolean;
+  choices?: SSEChoicePayload[];
+};

+type SSEDataLineNormalizer = {
+  hasPending: () => boolean;
+  normalize: (lines: string[]) => string[];
+};
+
+type SSEEventPrefixBuffer = {
+  clear: () => void;
+  eventType: () => string;
+  flush: () => string;
+  prefixData: (output: string, line: string) => string;
+  remember: (line: string) => void;
+};
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return !!value && typeof value === "object" && !Array.isArray(value);
+}
+
+export function parseSSEDataPayload(
+  data: unknown,
+  options: SSEPayloadOptions = {}
+): SSEJsonPayload | null {
+  const payload = String(data ?? "").trim();
+  if (!payload) return null;
+  if (payload === "[DONE]") return { done: true };
  try {
-    return JSON.parse(data);
+    const parsed = JSON.parse(payload) as unknown;
+    const eventType = options.eventType;
+    if (eventType && isRecord(parsed) && typeof parsed.type !== "string") {
+      return { ...parsed, type: eventType } as SSEJsonPayload;
+    }
+    return parsed as SSEJsonPayload;
  } catch (error) {
-    if (data.length > 0) {
+    if (options.logWarning !== false && payload.length > 0) {
      console.log(
-        `[WARN] Failed to parse SSE line (${data.length} chars): ${data.substring(0, 200)}...`
+        `[WARN] Failed to parse SSE payload (${payload.length} chars): ${payload.substring(0, 200)}...`
      );
    }
    return null;
  }
 }

+export function parseSSEDataLines(
+  dataLines: string[],
+  options: SSEPayloadOptions = {}
+): SSEJsonPayload | null {
+  return parseSSEDataPayload(dataLines.join("\n"), options);
+}
+
+// Parse SSE data line
+export function parseSSELine(line: string): SSEJsonPayload | null {
+  if (!line) return null;
+
+  // Trim leading whitespace before checking field name.
+  const trimmed = line.trimStart();
+  if (!trimmed.startsWith("data:")) return null;
+
+  return parseSSEDataPayload(trimmed.slice(5));
+}
+
+function extractSseDataLine(line: string): string | null {
+  const trimmed = line.trimStart().replace(/\r$/, "");
+  if (!trimmed.startsWith("data:")) return null;
+  return trimmed.slice(5).trimStart();
+}
+
+export function createSSEDataLineNormalizer(): SSEDataLineNormalizer {
+  let pendingEventLines: string[] = [];
+
+  const getPendingDataLines = () =>
+    pendingEventLines
+      .map((line) => extractSseDataLine(line))
+      .filter((line): line is string => line !== null);
+
+  const hasSelfDescribingPendingDataPayload = () => {
+    const dataLines = getPendingDataLines();
+    const parsed =
+      dataLines.length > 0 ? parseSSEDataLines(dataLines, { logWarning: false }) : null;
+    if (!parsed) return false;
+    return (
+      parsed.done === true ||
+      typeof parsed.type === "string" ||
+      typeof parsed.object === "string" ||
+      Array.isArray(parsed.choices) ||
+      Array.isArray(parsed.candidates) ||
+      isRecord(parsed.response)
+    );
+  };
+
+  const flush = (output: string[]) => {
+    if (pendingEventLines.length === 0) return;
+
+    const eventLines = pendingEventLines.filter((line) => line.trim().length > 0);
+    const dataLines: string[] = [];
+    const passthroughLines: string[] = [];
+
+    for (const line of eventLines) {
+      const dataLine = extractSseDataLine(line);
+      if (dataLine !== null) {
+        dataLines.push(dataLine);
+      } else {
+        passthroughLines.push(line);
+      }
+    }
+
+    output.push(...passthroughLines);
+    if (dataLines.length > 0) {
+      const parsed = parseSSEDataLines(dataLines, { logWarning: false });
+      if (parsed) {
+        output.push(parsed.done === true ? "data: [DONE]" : `data: ${JSON.stringify(parsed)}`);
+      } else {
+        output.push(...eventLines.filter((line) => extractSseDataLine(line) !== null));
+      }
+    } else {
+      output.push(...eventLines.filter((line) => extractSseDataLine(line) !== null));
+    }
+
+    output.push("");
+    pendingEventLines = [];
+  };
+
+  return {
+    hasPending() {
+      return pendingEventLines.length > 0;
+    },
+    normalize(lines: string[]) {
+      const output: string[] = [];
+      for (const line of lines) {
+        const normalizedLine = line.replace(/\r$/, "");
+        const trimmed = normalizedLine.trim();
+
+        if (
+          trimmed &&
+          /^(?:event:|id:|retry:|:)/i.test(trimmed) &&
+          hasSelfDescribingPendingDataPayload()
+        ) {
+          flush(output);
+        }
+
+        pendingEventLines.push(normalizedLine);
+        if (!trimmed) {
+          flush(output);
+        }
+      }
+      return output;
+    },
+  };
+}
+
+export function createSSEEventPrefixBuffer(): SSEEventPrefixBuffer {
+  let lines: string[] = [];
+  let emitted = false;
+  const hasUnemitted = () => lines.length > 0 && !emitted;
+  const prefix = (output: string) => {
+    if (!hasUnemitted()) return output;
+    emitted = true;
+    return `${lines.join("\n")}\n${output}`;
+  };
+  return {
+    clear() {
+      lines = [];
+      emitted = false;
+    },
+    eventType() {
+      for (let i = lines.length - 1; i >= 0; i--) {
+        const match = lines[i].trim().match(/^event:\s*(.+)$/i);
+        if (match) return match[1].trim();
+      }
+      return "";
+    },
+    flush() {
+      return hasUnemitted() ? prefix("\n") : "";
+    },
+    prefixData(output, line) {
+      return line.startsWith("data:") ? prefix(output) : output;
+    },
+    remember(line) {
+      lines.push(line);
+      emitted = false;
+    },
+  };
+}
+
+function hasOpenAICompatibleStreamValue(parsed: Record<string, unknown>): boolean {
+  if (!Array.isArray(parsed.choices)) return false;
+
+  return parsed.choices.some((choice) => {
+    if (!isRecord(choice)) return false;
+
+    const delta = isRecord(choice.delta) ? choice.delta : null;
+    if (!delta) return false;
+    if (typeof delta.content === "string" && delta.content.length > 0) return true;
+    if (typeof delta.reasoning_content === "string" && delta.reasoning_content.length > 0) {
+      return true;
+    }
+    if (typeof delta.reasoning_text === "string" && delta.reasoning_text.length > 0) {
+      return true;
+    }
+    return Array.isArray(delta.tool_calls) && delta.tool_calls.length > 0;
+  });
+}
+
+function hasResponsesStreamValue(parsed: Record<string, unknown>, eventType = ""): boolean {
+  const type = typeof parsed.type === "string" ? parsed.type : eventType;
+  if (!type.startsWith("response.")) return false;
+
+  if (
+    type === "response.output_text.delta" ||
+    type === "response.reasoning_text.delta" ||
+    type === "response.reasoning_summary_text.delta" ||
+    type === "response.function_call_arguments.delta"
+  ) {
+    return (
+      (typeof parsed.delta === "string" && parsed.delta.length > 0) ||
+      (typeof parsed.text === "string" && parsed.text.length > 0) ||
+      (typeof parsed.arguments === "string" && parsed.arguments.length > 0)
+    );
+  }
+
+  if (type === "response.output_item.added" || type === "response.output_item.done") {
+    return isRecord(parsed.item);
+  }
+
+  if (type === "response.content_part.added") {
+    return isRecord(parsed.part);
+  }
+
+  if (type === "response.completed" && isRecord(parsed.response)) {
+    const output = parsed.response.output;
+    return Array.isArray(output) && output.length > 0;
+  }
+
+  return false;
+}
+
+function hasGeminiCandidateStreamValue(parsed: Record<string, unknown>): boolean {
+  const candidates = Array.isArray(parsed.candidates)
+    ? parsed.candidates
+    : isRecord(parsed.response) && Array.isArray(parsed.response.candidates)
+      ? parsed.response.candidates
+      : [];
+
+  return candidates.some((candidate) => {
+    if (!isRecord(candidate)) return false;
+    const content = isRecord(candidate.content) ? candidate.content : null;
+    const parts = Array.isArray(content?.parts) ? content.parts : [];
+    return parts.some((part) => {
+      if (!isRecord(part)) return false;
+      if (typeof part.text === "string" && part.text.length > 0) return true;
+      return isRecord(part.functionCall) || isRecord(part.executableCode);
+    });
+  });
+}
+
+export function isKnownNonClaudeStreamPayload(
+  parsed: Record<string, unknown>,
+  eventType = ""
+): boolean {
+  if (Array.isArray(parsed.choices)) {
+    return hasOpenAICompatibleStreamValue(parsed);
+  }
+
+  const objectType = typeof parsed.object === "string" ? parsed.object : "";
+  if (
+    objectType === "chat.completion.chunk" ||
+    objectType === "text_completion" ||
+    objectType.endsWith(".completion.chunk")
+  ) {
+    return hasOpenAICompatibleStreamValue(parsed);
+  }
+
+  const type = typeof parsed.type === "string" ? parsed.type : eventType;
+  if (type.startsWith("response.")) return hasResponsesStreamValue(parsed, eventType);
+  if (Array.isArray(parsed.candidates)) return hasGeminiCandidateStreamValue(parsed);
+
+  const response = parsed.response;
+  return isRecord(response) && Array.isArray(response.candidates)
+    ? hasGeminiCandidateStreamValue(parsed)
+    : false;
+}
+
 // Check if chunk has valuable content (not empty)
 export function hasValuableContent(chunk, format) {
  // OpenAI format
--- a/open-sse/utils/streamReadiness.ts
+++ b/open-sse/utils/streamReadiness.ts
@@ -71,93 +71,22 @@ function hasUsefulJsonPayload(payload: unknown): boolean {
  return hasUsefulValue(payload);
 }

-function hasOpenAIResponseLifecyclePayload(
-  payload: Record<string, unknown>,
-  type: string
-): boolean {
-  if (type === "response.created" || type === "response.in_progress") {
-    const response = payload.response;
-    if (!isRecord(response)) return false;
-
-    return (
-      hasNonEmptyString(response.id) ||
-      hasNonEmptyString(response.object) ||
-      hasNonEmptyString(response.status) ||
-      typeof response.created_at === "number"
-    );
-  }
-
-  if (type === "response.output_item.added") {
-    const item = payload.item;
-    if (!isRecord(item)) return false;
-
-    return (
-      hasNonEmptyString(item.id) ||
-      hasNonEmptyString(item.type) ||
-      hasNonEmptyString(item.status) ||
-      Array.isArray(item.content) ||
-      isRecord(item.content)
-    );
-  }
-
-  return false;
+function isPingEventType(type: string): boolean {
+  return /^(?:ping|keepalive|heartbeat)$/i.test(type);
 }

-function hasChatCompletionToolCallStart(value: unknown): boolean {
-  // Accept a tool_call item if it has a non-empty id (fully-formed chunk) OR if it
-  // carries a numeric index (first OpenAI streaming chunk — id/name arrive in later
-  // chunks). Either form signals that a tool call has started. (#3612)
-  const hasToolCallStart = (item: unknown) =>
-    isRecord(item) && (hasNonEmptyString(item.id) || typeof item.index === "number");
-  if (Array.isArray(value)) return value.some(hasToolCallStart);
-  return hasToolCallStart(value);
+function getPayloadType(payload: unknown, eventType = ""): string {
+  if (!isRecord(payload)) return eventType;
+  const type = payload.type ?? payload.event ?? payload.object;
+  return typeof type === "string" ? type : eventType;
 }

-function hasChatCompletionFunctionCallStart(value: unknown): boolean {
-  return isRecord(value) && hasNonEmptyString(value.name);
-}
-
-function hasChatCompletionChunkStartPayload(payload: Record<string, unknown>): boolean {
-  // If object or type is PRESENT and is clearly a non-chat-chunk type, reject early.
-  // If both are absent (many OA-compatible backends omit object), we fall through and
-  // let the choices structure decide. (#3612)
-  const obj = payload.object;
-  const typ = payload.type;
-  const hasForeignObject =
-    (typeof obj === "string" && obj !== "chat.completion.chunk") ||
-    (typeof typ === "string" && typ !== "chat.completion.chunk");
-  if (hasForeignObject) return false;
-
-  const choices = payload.choices;
-  if (!Array.isArray(choices) || choices.length === 0) return false;
-
-  return choices.some((choice) => {
-    if (!isRecord(choice)) return false;
-    const delta = choice.delta;
-    if (!isRecord(delta)) return false;
-
-    return (
-      hasNonEmptyString(delta.role) ||
-      hasChatCompletionToolCallStart(delta.tool_calls) ||
-      hasChatCompletionFunctionCallStart(delta.function_call)
-    );
-  });
-}
-
-function hasAcceptedStreamStartPayload(payload: unknown, eventType = ""): boolean {
-  if (!isRecord(payload)) return false;
-
-  // Anthropic/Claude streams can legitimately start with lifecycle frames and
-  // OpenAI Responses streams can do the same before the first text/tool delta
-  // arrives. Treating structurally valid lifecycle frames as readiness prevents
-  // false 504s while ping-only/generic-empty zombie streams still fail below.
-  const type = typeof payload.type === "string" ? payload.type : eventType;
-  if (type === "message_start" && isRecord(payload.message)) return true;
-  if (type === "content_block_start" && isRecord(payload.content_block)) return true;
-  if (hasOpenAIResponseLifecyclePayload(payload, type)) return true;
-  if (hasChatCompletionChunkStartPayload(payload)) return true;
-
-  return false;
+function hasNonPingStructuredPayload(payload: unknown, eventType = ""): boolean {
+  const type = getPayloadType(payload, eventType);
+  if (isPingEventType(eventType) || isPingEventType(type)) return false;
+  if (Array.isArray(payload)) return payload.length > 0;
+  if (isRecord(payload)) return Object.keys(payload).length > 0;
+  return payload !== null && payload !== undefined;
 }

 export function hasUsefulStreamContent(text: string): boolean {
@@ -184,13 +113,33 @@ export function hasUsefulStreamContent(text: string): boolean {

 type StreamReadinessSignalState = {
  currentEvent: string;
+  dataLines: string[];
  pendingLine: string;
 };

+function resetCurrentEvent(state: StreamReadinessSignalState): void {
+  state.currentEvent = "";
+  state.dataLines = [];
+}
+
+function processStreamReadinessEvent(state: StreamReadinessSignalState): boolean {
+  const eventType = state.currentEvent;
+  const data = state.dataLines.join("\n").trim();
+  resetCurrentEvent(state);
+
+  if (isPingEventType(eventType) || !data || data === "[DONE]") return false;
+
+  try {
+    return hasNonPingStructuredPayload(JSON.parse(data), eventType);
+  } catch {
+    return data.length > 0;
+  }
+}
+
 function processStreamReadinessLine(state: StreamReadinessSignalState, line: string): boolean {
  const trimmed = line.trim();
  if (!trimmed || trimmed.startsWith(":")) {
-    if (!trimmed) state.currentEvent = "";
+    if (!trimmed) return processStreamReadinessEvent(state);
    return false;
  }

@@ -199,20 +148,10 @@ function processStreamReadinessLine(state: StreamReadinessSignalState, line: str
    return false;
  }

-  if (/^(?:ping|keepalive)$/i.test(state.currentEvent)) return false;
-  if (!trimmed.startsWith("data:")) return false;
-
-  const data = trimmed.slice(5).trim();
-  if (!data || data === "[DONE]") return false;
-
-  try {
-    const parsed = JSON.parse(data);
-    return (
-      hasUsefulJsonPayload(parsed) || hasAcceptedStreamStartPayload(parsed, state.currentEvent)
-    );
-  } catch {
-    return data.length > 0;
+  if (trimmed.startsWith("data:")) {
+    state.dataLines.push(trimmed.slice(5).trimStart());
  }
+  return false;
 }

 function appendStreamReadinessSignal(state: StreamReadinessSignalState, chunk: string): boolean {
@@ -226,14 +165,20 @@ function appendStreamReadinessSignal(state: StreamReadinessSignalState, chunk: s
  return false;
 }

+function finishStreamReadinessSignal(state: StreamReadinessSignalState): boolean {
+  if (state.pendingLine && processStreamReadinessLine(state, state.pendingLine)) return true;
+  state.pendingLine = "";
+  return processStreamReadinessEvent(state);
+}
+
 export function hasStreamReadinessSignal(text: string): boolean {
  const state: StreamReadinessSignalState = {
    currentEvent: "",
+    dataLines: [],
    pendingLine: "",
  };
  if (appendStreamReadinessSignal(state, text)) return true;
-  if (state.pendingLine) return processStreamReadinessLine(state, state.pendingLine);
-  return false;
+  return finishStreamReadinessSignal(state);
 }

 function createErrorResponse(
@@ -320,17 +265,29 @@ export async function ensureStreamReadiness(
  const decoder = new TextDecoder();
  const readinessState: StreamReadinessSignalState = {
    currentEvent: "",
+    dataLines: [],
    pendingLine: "",
  };
  const startedAt = Date.now();
-  const deadline = startedAt + options.timeoutMs;
+  const effectiveTimeoutMs = Math.max(0, Math.floor(options.timeoutMs));
+  const deadline = startedAt + effectiveTimeoutMs;
  let handedOffReader = false;

+  const buildReadyResponse = () =>
+    new Response(prependBufferedChunks(chunks, reader), {
+      status: response.status,
+      statusText: response.statusText,
+      headers: response.headers,
+    });
+
+  const timeoutReason = () =>
+    `Stream produced no non-ping SSE event within ${effectiveTimeoutMs}ms`;
+
  try {
    while (true) {
      const remainingMs = deadline - Date.now();
      if (remainingMs <= 0) {
-        const reason = `Stream produced no useful content within ${options.timeoutMs}ms`;
+        const reason = timeoutReason();
        options.log?.warn?.(
          "STREAM",
          `${reason} (${options.provider || "provider"}/${options.model || "unknown"})`
@@ -354,7 +311,7 @@ export async function ensureStreamReadiness(
      try {
        readResult = await readWithTimeout(reader, remainingMs);
      } catch {
-        const reason = `Stream produced no useful content within ${options.timeoutMs}ms`;
+        const reason = timeoutReason();
        options.log?.warn?.(
          "STREAM",
          `${reason} (${options.provider || "provider"}/${options.model || "unknown"})`
@@ -375,7 +332,17 @@ export async function ensureStreamReadiness(
      }

      if (readResult.done) {
-        const reason = "Stream ended before producing useful content";
+        const tail = decoder.decode(undefined, { stream: false });
+        if (tail && appendStreamReadinessSignal(readinessState, tail)) {
+          handedOffReader = true;
+          return { ok: true, response: buildReadyResponse() };
+        }
+        if (finishStreamReadinessSignal(readinessState)) {
+          handedOffReader = true;
+          return { ok: true, response: buildReadyResponse() };
+        }
+
+        const reason = "Stream ended before producing a non-ping SSE event";
        options.log?.warn?.(
          "STREAM",
          `${reason} (${options.provider || "provider"}/${options.model || "unknown"})`
@@ -406,11 +373,7 @@ export async function ensureStreamReadiness(
        handedOffReader = true;
        return {
          ok: true,
-          response: new Response(prependBufferedChunks(chunks, reader), {
-            status: response.status,
-            statusText: response.statusText,
-            headers: response.headers,
-          }),
+          response: buildReadyResponse(),
        };
      }
    }
--- a/src/shared/utils/runtimeTimeouts.ts
+++ b/src/shared/utils/runtimeTimeouts.ts
@@ -97,7 +97,7 @@ export function getUpstreamTimeoutConfig(
  const streamReadinessTimeoutMs = readTimeoutMs(
    env,
    "STREAM_READINESS_TIMEOUT_MS",
-    DEFAULT_STREAM_READINESS_TIMEOUT_MS,
+    sharedRequestTimeoutMs ?? DEFAULT_STREAM_READINESS_TIMEOUT_MS,
    {
      allowZero: true,
      logger,
--- a/tests/unit/combo-stream-readiness-fallback.test.ts
+++ b/tests/unit/combo-stream-readiness-fallback.test.ts
@@ -1,7 +1,7 @@
 import test from "node:test";
 import assert from "node:assert/strict";

-import { handleComboChat } from "../../open-sse/services/combo.ts";
+import { handleComboChat, validateResponseQuality } from "../../open-sse/services/combo.ts";
 import { ensureStreamReadiness } from "../../open-sse/utils/streamReadiness.ts";

 const textEncoder = new TextEncoder();
@@ -77,6 +77,372 @@ function errorResponse(status: number, message: string): Response {
  });
 }

+async function readWithTimeout(response: Response, timeoutMs = 100): Promise<string> {
+  let timer: ReturnType<typeof setTimeout> | null = null;
+  try {
+    return await Promise.race([
+      response.text(),
+      new Promise<never>((_, reject) => {
+        timer = setTimeout(
+          () => reject(new Error(`Timed out reading response after ${timeoutMs}ms`)),
+          timeoutMs
+        );
+      }),
+    ]);
+  } finally {
+    if (timer) clearTimeout(timer);
+  }
+}
+
+test("streaming quality peek releases OpenAI-compatible reasoning-only SSE immediately", async () => {
+  let streamController: ReadableStreamDefaultController<Uint8Array> | null = null;
+
+  const firstChunk = `data: ${JSON.stringify({
+    id: "chatcmpl-test",
+    object: "chat.completion.chunk",
+    choices: [
+      {
+        delta: { role: "assistant", content: "", reasoning_content: "thinking" },
+        finish_reason: null,
+      },
+    ],
+  })}\n\n`;
+
+  const body = new ReadableStream<Uint8Array>({
+    start(controller) {
+      streamController = controller;
+      controller.enqueue(textEncoder.encode(firstChunk));
+    },
+  });
+
+  const result = await Promise.race([
+    validateResponseQuality(
+      new Response(body, {
+        status: 200,
+        headers: { "Content-Type": "text/event-stream" },
+      }),
+      true,
+      createLog()
+    ),
+    new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 50)),
+  ]);
+
+  assert.notEqual(result, "timeout", "quality peek should not wait for the full OpenAI stream");
+  assert.equal(result.valid, true);
+  assert.ok(result.clonedResponse, "quality peek should replay the already-read prefix");
+
+  streamController?.enqueue(textEncoder.encode("data: [DONE]\n\n"));
+  streamController?.close();
+
+  const text = await readWithTimeout(result.clonedResponse);
+  assert.match(text, /reasoning_content/);
+  assert.match(text, /\[DONE\]/);
+});
+
+test("streaming quality peek waits past OpenAI-compatible empty header chunks", async () => {
+  let streamController: ReadableStreamDefaultController<Uint8Array> | null = null;
+
+  const emptyHeaderChunk = `data: ${JSON.stringify({
+    id: "chatcmpl-test",
+    object: "chat.completion.chunk",
+    choices: [
+      {
+        delta: { role: "assistant", content: "" },
+        finish_reason: null,
+      },
+    ],
+  })}\n\n`;
+
+  const reasoningChunk = `data: ${JSON.stringify({
+    id: "chatcmpl-test",
+    object: "chat.completion.chunk",
+    choices: [
+      {
+        delta: { content: "", reasoning_content: "thinking" },
+        finish_reason: null,
+      },
+    ],
+  })}\n\n`;
+
+  const body = new ReadableStream<Uint8Array>({
+    start(controller) {
+      streamController = controller;
+      controller.enqueue(textEncoder.encode(emptyHeaderChunk));
+    },
+  });
+
+  const qualityPromise = validateResponseQuality(
+    new Response(body, {
+      status: 200,
+      headers: { "Content-Type": "text/event-stream" },
+    }),
+    true,
+    createLog()
+  );
+
+  const earlyResult = await Promise.race([
+    qualityPromise.then(() => "released"),
+    new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 50)),
+  ]);
+
+  assert.equal(earlyResult, "pending", "empty OpenAI-compatible wrapper should not release");
+
+  streamController?.enqueue(textEncoder.encode(reasoningChunk));
+
+  const result = await Promise.race([
+    qualityPromise,
+    new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 100)),
+  ]);
+
+  assert.notEqual(result, "timeout", "quality peek should release when reasoning starts");
+  assert.equal(result.valid, true);
+  assert.ok(result.clonedResponse, "quality peek should replay both buffered chunks");
+
+  streamController?.enqueue(textEncoder.encode("data: [DONE]\n\n"));
+  streamController?.close();
+
+  const text = await readWithTimeout(result.clonedResponse);
+  assert.match(text, /"role":"assistant"/);
+  assert.match(text, /reasoning_content/);
+  assert.match(text, /\[DONE\]/);
+});
+
+test("streaming quality peek waits past OpenAI-compatible finish-only chunks", async () => {
+  let streamController: ReadableStreamDefaultController<Uint8Array> | null = null;
+
+  const finishOnlyChunk = `data: ${JSON.stringify({
+    id: "chatcmpl-test",
+    object: "chat.completion.chunk",
+    choices: [{ delta: {}, finish_reason: "stop" }],
+  })}\n\n`;
+
+  const contentChunk = `data: ${JSON.stringify({
+    id: "chatcmpl-test",
+    object: "chat.completion.chunk",
+    choices: [{ delta: { content: "answer" }, finish_reason: null }],
+  })}\n\n`;
+
+  const body = new ReadableStream<Uint8Array>({
+    start(controller) {
+      streamController = controller;
+      controller.enqueue(textEncoder.encode(finishOnlyChunk));
+    },
+  });
+
+  const qualityPromise = validateResponseQuality(
+    new Response(body, {
+      status: 200,
+      headers: { "Content-Type": "text/event-stream" },
+    }),
+    true,
+    createLog()
+  );
+
+  const earlyResult = await Promise.race([
+    qualityPromise.then(() => "released"),
+    new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 50)),
+  ]);
+
+  assert.equal(earlyResult, "pending", "finish-only OpenAI chunk should not release");
+
+  streamController?.enqueue(textEncoder.encode(contentChunk));
+
+  const result = await Promise.race([
+    qualityPromise,
+    new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 100)),
+  ]);
+
+  assert.notEqual(result, "timeout", "quality peek should release when content starts");
+  assert.equal(result.valid, true);
+  assert.ok(result.clonedResponse, "quality peek should replay buffered finish and content chunks");
+
+  streamController?.enqueue(textEncoder.encode("data: [DONE]\n\n"));
+  streamController?.close();
+
+  const text = await readWithTimeout(result.clonedResponse);
+  assert.match(text, /finish_reason/);
+  assert.match(text, /"content":"answer"/);
+});
+
+test("streaming quality peek waits past Responses lifecycle-only events", async () => {
+  let streamController: ReadableStreamDefaultController<Uint8Array> | null = null;
+
+  const lifecycleEvent = [
+    "event: response.created",
+    `data: ${JSON.stringify({ response: { id: "resp_test" } })}`,
+    "",
+    "",
+  ].join("\n");
+
+  const textDeltaEvent = [
+    "event: response.output_text.delta",
+    `data: ${JSON.stringify({ delta: "visible" })}`,
+    "",
+    "",
+  ].join("\n");
+
+  const body = new ReadableStream<Uint8Array>({
+    start(controller) {
+      streamController = controller;
+      controller.enqueue(textEncoder.encode(lifecycleEvent));
+    },
+  });
+
+  const qualityPromise = validateResponseQuality(
+    new Response(body, {
+      status: 200,
+      headers: { "Content-Type": "text/event-stream" },
+    }),
+    true,
+    createLog()
+  );
+
+  const earlyResult = await Promise.race([
+    qualityPromise.then(() => "released"),
+    new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 50)),
+  ]);
+
+  assert.equal(earlyResult, "pending", "response.created should not release quality peek");
+
+  streamController?.enqueue(textEncoder.encode(textDeltaEvent));
+
+  const result = await Promise.race([
+    qualityPromise,
+    new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 100)),
+  ]);
+
+  assert.notEqual(result, "timeout", "quality peek should release on Responses text delta");
+  assert.equal(result.valid, true);
+  assert.ok(result.clonedResponse);
+
+  streamController?.close();
+  const text = await readWithTimeout(result.clonedResponse);
+  assert.match(text, /response.created/);
+  assert.match(text, /response.output_text.delta/);
+});
+
+test("streaming quality peek waits past Gemini finish-only candidates", async () => {
+  let streamController: ReadableStreamDefaultController<Uint8Array> | null = null;
+
+  const finishOnlyChunk = `data: ${JSON.stringify({
+    candidates: [{ finishReason: "STOP", content: { parts: [] } }],
+  })}\n\n`;
+
+  const textChunk = `data: ${JSON.stringify({
+    candidates: [{ content: { parts: [{ text: "gemini text" }] } }],
+  })}\n\n`;
+
+  const body = new ReadableStream<Uint8Array>({
+    start(controller) {
+      streamController = controller;
+      controller.enqueue(textEncoder.encode(finishOnlyChunk));
+    },
+  });
+
+  const qualityPromise = validateResponseQuality(
+    new Response(body, {
+      status: 200,
+      headers: { "Content-Type": "text/event-stream" },
+    }),
+    true,
+    createLog()
+  );
+
+  const earlyResult = await Promise.race([
+    qualityPromise.then(() => "released"),
+    new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 50)),
+  ]);
+
+  assert.equal(earlyResult, "pending", "finish-only Gemini candidate should not release");
+
+  streamController?.enqueue(textEncoder.encode(textChunk));
+
+  const result = await Promise.race([
+    qualityPromise,
+    new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 100)),
+  ]);
+
+  assert.notEqual(result, "timeout", "quality peek should release on Gemini text parts");
+  assert.equal(result.valid, true);
+  assert.ok(result.clonedResponse);
+
+  streamController?.close();
+  const text = await readWithTimeout(result.clonedResponse);
+  assert.match(text, /gemini text/);
+});
+
+test("streaming quality peek parses legal multi-line SSE data before releasing", async () => {
+  const payload = JSON.stringify({
+    id: "chatcmpl-test",
+    object: "chat.completion.chunk",
+    choices: [{ delta: { reasoning_content: "split thinking" }, finish_reason: null }],
+  });
+  const splitAt = payload.indexOf('"choices"') - 1;
+  assert.ok(splitAt > 0);
+
+  const body = new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(
+        textEncoder.encode(
+          `data: ${payload.slice(0, splitAt)}\ndata: ${payload.slice(splitAt)}\n\n`
+        )
+      );
+    },
+  });
+
+  const result = await Promise.race([
+    validateResponseQuality(
+      new Response(body, {
+        status: 200,
+        headers: { "Content-Type": "text/event-stream" },
+      }),
+      true,
+      createLog()
+    ),
+    new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 100)),
+  ]);
+
+  assert.notEqual(result, "timeout", "quality peek should release on multi-line SSE reasoning");
+  assert.equal(result.valid, true);
+  assert.ok(result.clonedResponse);
+});
+
+test("streaming quality peek still rejects complete Claude lifecycle without content blocks", async () => {
+  const body = new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(
+        textEncoder.encode(
+          [
+            `event: message_start`,
+            `data: ${JSON.stringify({ type: "message_start", message: { id: "msg_1" } })}`,
+            "",
+            `event: message_delta`,
+            `data: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "content_filter" } })}`,
+            "",
+            `event: message_stop`,
+            `data: ${JSON.stringify({ type: "message_stop" })}`,
+            "",
+            "",
+          ].join("\n")
+        )
+      );
+      controller.close();
+    },
+  });
+
+  const result = await validateResponseQuality(
+    new Response(body, {
+      status: 200,
+      headers: { "Content-Type": "text/event-stream" },
+    }),
+    true,
+    createLog()
+  );
+
+  assert.equal(result.valid, false);
+  assert.equal(result.reason, "streaming empty content block");
+});
+
 test("combo falls back when first model returns HTTP 200 zombie SSE stream", async () => {
  const calls: string[] = [];
  const log = createLog();
@@ -133,7 +499,7 @@ test("combo fails when all models return 504", async () => {
      strategy: "priority",
      models: [
        { model: "glm/zombie-a", weight: 0 },
-        { model: "glm/zombie-b", weight: 0 },
+        { model: "openai/gpt-5.4-mini", weight: 0 },
      ],
      config: { maxRetries: 0, retryDelayMs: 0 },
    },
@@ -150,7 +516,7 @@ test("combo fails when all models return 504", async () => {

  assert.ok(!result.ok, "combo should fail when all models return 504");
  assert.equal(result.status, 504);
-  assert.equal(calls.length, 2, "combo should try both models");
+  assert.deepEqual(calls, ["glm/zombie-a", "openai/gpt-5.4-mini"]);
 });

 test("combo retries 504 on same model before falling through (transient retry)", async () => {
--- a/tests/unit/runtime-timeouts.test.ts
+++ b/tests/unit/runtime-timeouts.test.ts
@@ -31,6 +31,7 @@ test("REQUEST_TIMEOUT_MS becomes the common timeout baseline when specific overr

  assert.equal(upstreamConfig.fetchTimeoutMs, 600000);
  assert.equal(upstreamConfig.streamIdleTimeoutMs, 600000);
+  assert.equal(upstreamConfig.streamReadinessTimeoutMs, 600000);
  assert.equal(upstreamConfig.fetchHeadersTimeoutMs, 600000);
  assert.equal(upstreamConfig.fetchBodyTimeoutMs, 600000);
  assert.equal(apiBridgeConfig.proxyTimeoutMs, 600000);
@@ -42,12 +43,14 @@ test("upstream timeout config honors explicit overrides and falls back on invali
    REQUEST_TIMEOUT_MS: "550000",
    FETCH_TIMEOUT_MS: "600000",
    STREAM_IDLE_TIMEOUT_MS: "600000",
+    STREAM_READINESS_TIMEOUT_MS: "90000",
    FETCH_HEADERS_TIMEOUT_MS: "610000",
    FETCH_BODY_TIMEOUT_MS: "0",
    FETCH_CONNECT_TIMEOUT_MS: "45000",
    FETCH_KEEPALIVE_TIMEOUT_MS: "-1",
  });

+  assert.equal(config.streamReadinessTimeoutMs, 90000);
  assert.equal(config.fetchHeadersTimeoutMs, 610000);
  assert.equal(config.fetchBodyTimeoutMs, 0);
  assert.equal(config.fetchConnectTimeoutMs, 45000);
--- a/tests/unit/stream-readiness.test.ts
+++ b/tests/unit/stream-readiness.test.ts
@@ -38,6 +38,7 @@ function delayedClaudeStartStream(): ReadableStream<Uint8Array> {
              },
            })}`,
            "",
+            "",
          ].join("\n")
        )
      );
@@ -53,6 +54,7 @@ function delayedClaudeStartStream(): ReadableStream<Uint8Array> {
              delta: { type: "text_delta", text: "slow hello" },
            })}`,
            "",
+            "",
          ].join("\n")
        )
      );
@@ -78,6 +80,7 @@ function delayedOpenAIResponsesStartStream(): ReadableStream<Uint8Array> {
              },
            })}`,
            "",
+            "",
          ].join("\n")
        )
      );
@@ -89,6 +92,7 @@ function delayedOpenAIResponsesStartStream(): ReadableStream<Uint8Array> {
            "event: response.output_text.delta",
            `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "slow hello" })}`,
            "",
+            "",
          ].join("\n")
        )
      );
@@ -125,6 +129,46 @@ function delayedChatCompletionStartStream(): ReadableStream<Uint8Array> {
  });
 }

+function delayedGeminiStartStream(): ReadableStream<Uint8Array> {
+  return new ReadableStream<Uint8Array>({
+    async start(controller) {
+      controller.enqueue(
+        encoder.encode(
+          `data: ${JSON.stringify({
+            candidates: [{ content: { parts: [] } }],
+          })}\n\n`
+        )
+      );
+
+      await new Promise((resolve) => setTimeout(resolve, 30));
+      controller.enqueue(
+        encoder.encode(
+          `data: ${JSON.stringify({
+            candidates: [{ content: { parts: [{ text: "slow gemini hello" }] } }],
+          })}\n\n`
+        )
+      );
+      controller.close();
+    },
+  });
+}
+
+function delayedUnknownStructuredStartStream(): ReadableStream<Uint8Array> {
+  return new ReadableStream<Uint8Array>({
+    async start(controller) {
+      controller.enqueue(
+        encoder.encode('event: provider.lifecycle\ndata: {"phase":"started"}\n\n')
+      );
+
+      await new Promise((resolve) => setTimeout(resolve, 30));
+      controller.enqueue(
+        encoder.encode('event: provider.delta\ndata: {"text":"slow unknown hello"}\n\n')
+      );
+      controller.close();
+    },
+  });
+}
+
 function zombieReadinessStream(): ReadableStream<Uint8Array> {
  return new ReadableStream<Uint8Array>({
    async start(controller) {
@@ -198,12 +242,12 @@ test("hasUsefulStreamContent detects text, reasoning, and tool deltas", () => {
  );
 });

-test("hasStreamReadinessSignal accepts Claude stream start events", () => {
+test("hasStreamReadinessSignal accepts any non-ping structured SSE event", () => {
  assert.equal(hasStreamReadinessSignal(": keepalive\n\n"), false);
  assert.equal(hasStreamReadinessSignal("event: ping\ndata: {}\n\n"), false);
  assert.equal(
    hasStreamReadinessSignal(`data: ${JSON.stringify({ type: "response.created" })}\n\n`),
-    false
+    true
  );
  assert.equal(
    hasStreamReadinessSignal(
@@ -241,13 +285,22 @@ test("hasStreamReadinessSignal accepts Claude stream start events", () => {
    ),
    true
  );
+  assert.equal(
+    hasStreamReadinessSignal('event: provider.lifecycle\ndata: {"phase":"started"}\n\n'),
+    true
+  );
+  assert.equal(hasStreamReadinessSignal('event: ping\ndata: {"phase":"started"}\n\n'), false);
+  assert.equal(
+    hasStreamReadinessSignal('event: ping\ndata: {"type":"response.created"}\n\n'),
+    false
+  );
 });

-test("hasStreamReadinessSignal accepts valid OpenAI Responses lifecycle events", () => {
+test("hasStreamReadinessSignal accepts Responses lifecycle events without schema gating", () => {
  assert.equal(hasStreamReadinessSignal(`data: ${JSON.stringify({})}\n\n`), false);
  assert.equal(
    hasStreamReadinessSignal(`data: ${JSON.stringify({ type: "response.created" })}\n\n`),
-    false
+    true
  );
  assert.equal(
    hasStreamReadinessSignal(
@@ -281,7 +334,7 @@ test("hasStreamReadinessSignal accepts valid OpenAI Responses lifecycle events",
  );
 });

-test("hasStreamReadinessSignal accepts structural chat completion chunk starts", () => {
+test("hasStreamReadinessSignal accepts chat completion structural chunks without content gating", () => {
  assert.equal(
    hasStreamReadinessSignal(
      `data: ${JSON.stringify({
@@ -316,7 +369,7 @@ test("hasStreamReadinessSignal accepts structural chat completion chunk starts",
    hasStreamReadinessSignal(
      `data: ${JSON.stringify({ object: "chat.completion.chunk", choices: [] })}\n\n`
    ),
-    false
+    true
  );
  assert.equal(
    hasStreamReadinessSignal(
@@ -325,7 +378,7 @@ test("hasStreamReadinessSignal accepts structural chat completion chunk starts",
        choices: [{ index: 0, delta: {} }],
      })}\n\n`
    ),
-    false
+    true
  );
  // #3612: index-only tool_call chunk (first chunk in OpenAI streaming — no id yet)
  // MUST be treated as a readiness signal (tool-call has started)
@@ -345,7 +398,7 @@ test("hasStreamReadinessSignal accepts structural chat completion chunk starts",
        choices: [{ index: 0, delta: { function_call: {} } }],
      })}\n\n`
    ),
-    false
+    true
  );
  // #3612: chunk with valid choices but NO object/type field (some OA-compatible backends
  // omit object) — must be accepted as a readiness signal when delta.role is present
@@ -358,7 +411,7 @@ test("hasStreamReadinessSignal accepts structural chat completion chunk starts",
    ),
    true
  );
-  // object present but a different (non-chat-chunk) type must still be rejected
+  // Stream readiness is a zombie filter now, not a provider-specific schema gate.
  assert.equal(
    hasStreamReadinessSignal(
      `data: ${JSON.stringify({
@@ -366,7 +419,7 @@ test("hasStreamReadinessSignal accepts structural chat completion chunk starts",
        choices: [{ index: 0, delta: { role: "assistant" } }],
      })}\n\n`
    ),
-    false
+    true
  );
 });

@@ -396,6 +449,26 @@ test("ensureStreamReadiness preserves buffered chunks when stream starts", async
  assert.match(text, / world/);
 });

+test("ensureStreamReadiness honors configured timeouts above 2000ms", async () => {
+  const response = new Response(
+    streamFromChunks(
+      [
+        `data: ${JSON.stringify({
+          type: "provider.started",
+          text: "slow first byte",
+        })}\n\n`,
+      ],
+      2_100
+    ),
+    { status: 200, headers: { "Content-Type": "text/event-stream" } }
+  );
+
+  const result = await ensureStreamReadiness(response, { timeoutMs: 3_000 });
+  assert.equal(result.ok, true);
+  const text = await result.response.text();
+  assert.match(text, /slow first byte/);
+});
+
 test("ensureStreamReadiness hands off long Claude streams after message_start", async () => {
  const response = new Response(delayedClaudeStartStream(), {
    status: 200,
@@ -447,7 +520,41 @@ test("ensureStreamReadiness hands off chat completion streams after role-only st
  assert.match(text, /slow chat hello/);
 });

-test("ensureStreamReadiness returns 504 when no useful content arrives before timeout", async () => {
+test("ensureStreamReadiness hands off Gemini streams after structural candidate start", async () => {
+  const response = new Response(delayedGeminiStartStream(), {
+    status: 200,
+    headers: { "Content-Type": "text/event-stream" },
+  });
+
+  const result = await ensureStreamReadiness(response, {
+    timeoutMs: 10,
+    provider: "gemini",
+    model: "gemini-3.0-pro",
+  });
+  assert.equal(result.ok, true);
+  const text = await result.response.text();
+  assert.match(text, /candidates/);
+  assert.match(text, /slow gemini hello/);
+});
+
+test("ensureStreamReadiness hands off unknown structured provider events promptly", async () => {
+  const response = new Response(delayedUnknownStructuredStartStream(), {
+    status: 200,
+    headers: { "Content-Type": "text/event-stream" },
+  });
+
+  const result = await ensureStreamReadiness(response, {
+    timeoutMs: 10,
+    provider: "provider-specific",
+    model: "custom-stream-model",
+  });
+  assert.equal(result.ok, true);
+  const text = await result.response.text();
+  assert.match(text, /provider\.lifecycle/);
+  assert.match(text, /slow unknown hello/);
+});
+
+test("ensureStreamReadiness returns 504 when no non-ping SSE event arrives before timeout", async () => {
  const response = new Response(zombieReadinessStream(), {
    status: 200,
    headers: { "Content-Type": "text/event-stream" },
@@ -456,10 +563,12 @@ test("ensureStreamReadiness returns 504 when no useful content arrives before ti
  const result = await ensureStreamReadiness(response, { timeoutMs: 10 });
  assert.equal(result.ok, false);
  assert.equal(result.response.status, 504);
-  assert.match(await result.response.text(), /STREAM_READINESS_TIMEOUT/);
+  const body = await result.response.json();
+  assert.equal(body.error.code, "STREAM_READINESS_TIMEOUT");
+  assert.match(body.error.message, /non-ping SSE event/);
 });

-test("ensureStreamReadiness returns 502 when stream ends without useful content", async () => {
+test("ensureStreamReadiness returns 502 when stream ends without a non-ping SSE event", async () => {
  const response = new Response(streamFromChunks([": keepalive\n\n"]), {
    status: 200,
    headers: { "Content-Type": "text/event-stream" },
@@ -470,6 +579,20 @@ test("ensureStreamReadiness returns 502 when stream ends without useful content"
  assert.equal(result.response.status, 502);
 });

+test("ensureStreamReadiness accepts a final event without a trailing blank line", async () => {
+  const response = new Response(
+    streamFromChunks(['event: provider.lifecycle\ndata: {"ok":true}']),
+    {
+      status: 200,
+      headers: { "Content-Type": "text/event-stream" },
+    }
+  );
+
+  const result = await ensureStreamReadiness(response, { timeoutMs: 100 });
+  assert.equal(result.ok, true);
+  assert.match(await result.response.text(), /provider\.lifecycle/);
+});
+
 // Regression for #2520: a reasoning-only stream (Mistral `thinking` array / StepFun
 // `reasoning_details`) is real output and must NOT be classified as "no useful content"
 // (which produced a spurious 502).
--- a/tests/unit/stream-utils.test.ts
+++ b/tests/unit/stream-utils.test.ts
@@ -49,6 +49,22 @@ async function readWithTransform(chunks, transformStream) {
  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));
+}
+
 test.after(() => {
  core.resetDbInstance();
  if (fs.existsSync(TEST_DATA_DIR)) {
@@ -564,6 +580,101 @@ test("createSSEStream passthrough flushes a buffered final line without a traili
  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(
@@ -619,6 +730,198 @@ test("createSSEStream translate mode converts Claude SSE into OpenAI chunks and
  assert.equal(onCompletePayload.responseBody.usage.total_tokens, 4);
 });

+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: "<thinking>\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),
+    ["<thinking>\n[metacognition", "]\n\nVisible answer"]
+  );
+  assert.equal(
+    deltas.some((delta) => delta.reasoning_content !== undefined),
+    false
+  );
+  assert.equal(
+    onCompletePayload.responseBody.choices[0].message.content,
+    "<thinking>\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]
@@ -939,7 +1242,88 @@ test("createSSEStream passthrough fixes generic ids and normalizes reasoning ali

  assert.match(text, /"id":"chatcmpl-/);
  assert.match(text, /"reasoning_content":"Let me think first"/);
-  assert.equal(text.includes('"reasoning":"Let me think first"'), false);
+});
+
+test("createSSEStream passthrough reserializes 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: "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"/);
+  assert.doesNotMatch(text, /"reasoning":"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: "<thinking>\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 === "<thinking>\nVisible prompt tag"));
+  assert.equal(
+    deltas.some((delta) => delta.reasoning_content !== undefined),
+    false
+  );
+  assert.equal(
+    onCompletePayload.responseBody.choices[0].message.content,
+    "<thinking>\nVisible prompt tag"
+  );
 });

 test("createSSEStream passthrough splits mixed reasoning and content deltas and estimates usage", async () => {
@@ -994,6 +1378,55 @@ test("createSSEStream passthrough splits mixed reasoning and content deltas and
  assert.ok(onCompletePayload.responseBody.usage.total_tokens > 0);
 });

+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(