Fix CC-compatible streaming bridge

open-sse/handlers/chatCore.ts

@@ -32,6 +32,7 @@ import {
 import {
   COOLDOWN_MS,
   HTTP_STATUS,
+  FETCH_BODY_TIMEOUT_MS,
   MAX_TOOLS_LIMIT,
   PROVIDER_MAX_TOKENS,
   STREAM_IDLE_TIMEOUT_MS,
@@ -574,6 +575,146 @@ function normalizeNonStreamingEventPayload(rawBody: string, contentType: string)
   return rawBody;
 }
 
+const NON_STREAMING_SSE_TERMINAL_TYPES = new Set([
+  "message_stop",
+  "response.completed",
+  "response.done",
+  "response.cancelled",
+  "response.canceled",
+  "response.failed",
+  "response.incomplete",
+]);
+
+type NonStreamingSseTerminalState = {
+  currentEvent: string;
+  pendingLine: string;
+};
+
+function processNonStreamingSseTerminalLine(
+  state: NonStreamingSseTerminalState,
+  rawLine: string
+): boolean {
+  const trimmed = rawLine.trim();
+  if (!trimmed || trimmed.startsWith(":")) {
+    if (!trimmed) state.currentEvent = "";
+    return false;
+  }
+
+  if (trimmed.startsWith("event:")) {
+    state.currentEvent = trimmed.slice(6).trim();
+    return false;
+  }
+
+  if (!trimmed.startsWith("data:")) return false;
+  const data = trimmed.slice(5).trim();
+  if (data === "[DONE]") return true;
+  if (!data) return false;
+
+  try {
+    const parsed = JSON.parse(data);
+    const eventType =
+      parsed && typeof parsed === "object" && typeof parsed.type === "string"
+        ? parsed.type
+        : state.currentEvent;
+    return NON_STREAMING_SSE_TERMINAL_TYPES.has(eventType);
+  } catch {
+    // Keep reading malformed data so the parser can report a useful upstream error.
+    return false;
+  }
+}
+
+function appendNonStreamingSseTerminalSignal(
+  state: NonStreamingSseTerminalState,
+  chunk: string
+): boolean {
+  const lines = `${state.pendingLine}${chunk}`.split(/\r?\n/);
+  state.pendingLine = lines.pop() ?? "";
+
+  for (const rawLine of lines) {
+    if (processNonStreamingSseTerminalLine(state, rawLine)) return true;
+  }
+
+  return false;
+}
+
+function createBodyTimeoutError(timeoutMs: number): Error {
+  const err = new Error(`Response body read timeout after ${timeoutMs}ms`);
+  err.name = "BodyTimeoutError";
+  return err;
+}
+
+function readStreamChunkWithTimeout(
+  reader: ReadableStreamDefaultReader<Uint8Array>,
+  timeoutMs: number
+): Promise<ReadableStreamReadResult<Uint8Array>> {
+  if (timeoutMs <= 0) return reader.read();
+
+  return new Promise((resolve, reject) => {
+    const timeout = setTimeout(() => reject(createBodyTimeoutError(timeoutMs)), timeoutMs);
+    reader.read().then(
+      (value) => {
+        clearTimeout(timeout);
+        resolve(value);
+      },
+      (error) => {
+        clearTimeout(timeout);
+        reject(error);
+      }
+    );
+  });
+}
+
+async function readNonStreamingResponseBody(
+  response: Response,
+  contentType: string,
+  upstreamStream: boolean
+): Promise<string> {
+  if (
+    !upstreamStream ||
+    !response.body ||
+    (!contentType.includes("text/event-stream") && !contentType.includes("application/x-ndjson"))
+  ) {
+    return withBodyTimeout<string>(response.text());
+  }
+
+  const reader = response.body.getReader();
+  const decoder = new TextDecoder();
+  const terminalState: NonStreamingSseTerminalState = {
+    currentEvent: "",
+    pendingLine: "",
+  };
+  let rawBody = "";
+  const deadline = FETCH_BODY_TIMEOUT_MS > 0 ? Date.now() + FETCH_BODY_TIMEOUT_MS : 0;
+
+  try {
+    while (true) {
+      const timeoutMs = deadline > 0 ? deadline - Date.now() : 0;
+      if (deadline > 0 && timeoutMs <= 0) {
+        throw createBodyTimeoutError(FETCH_BODY_TIMEOUT_MS);
+      }
+
+      const { done, value } = await readStreamChunkWithTimeout(reader, timeoutMs);
+      if (done) break;
+      if (!value) continue;
+
+      const decodedChunk = decoder.decode(value, { stream: true });
+      rawBody += decodedChunk;
+      if (appendNonStreamingSseTerminalSignal(terminalState, decodedChunk)) {
+        await reader.cancel("non-streaming bridge consumed terminal SSE event").catch(() => {});
+        break;
+      }
+    }
+  } catch (error) {
+    await reader.cancel(error).catch(() => {});
+    throw error;
+  } finally {
+    rawBody += decoder.decode();
+    reader.releaseLock();
+  }
+
+  return rawBody;
+}
+
 function getHeaderValueCaseInsensitive(
   headers: Record<string, unknown> | Headers | null | undefined,
   targetName: string
@@ -2984,7 +3125,12 @@ export async function handleChatCore({
         const status = rawResult.response.status;
         const statusText = rawResult.response.statusText;
         const headers = new Headers(rawResult.response.headers);
-        const payload = await withBodyTimeout<string>(rawResult.response.text());
+        const contentType = (headers.get("content-type") || "").toLowerCase();
+        const payload = await readNonStreamingResponseBody(
+          rawResult.response,
+          contentType,
+          upstreamStream
+        );
         acquireAccountSemaphoreRelease();
 
         return {
@@ -3657,7 +3803,11 @@ export async function handleChatCore({
     const contentType = (providerResponse.headers.get("content-type") || "").toLowerCase();
     let responseBody;
     let responsePayloadFormat = targetFormat;
-    const rawBody = await withBodyTimeout<string>(providerResponse.text());
+    const rawBody = await readNonStreamingResponseBody(
+      providerResponse,
+      contentType,
+      upstreamStream
+    );
     const normalizedProviderPayload = normalizePayloadForLog(rawBody);
     const looksLikeSSE =
       contentType.includes("text/event-stream") ||

open-sse/handlers/sseParser.ts

@@ -22,9 +22,19 @@ function readSSEEvents(rawSSE) {
     }
 
     try {
+      const data = JSON.parse(payload);
+      if (
+        currentEvent &&
+        data &&
+        typeof data === "object" &&
+        !Array.isArray(data) &&
+        typeof data.type !== "string"
+      ) {
+        data.type = currentEvent;
+      }
       events.push({
         event: currentEvent || undefined,
-        data: JSON.parse(payload),
+        data,
       });
     } catch {
       // Ignore malformed SSE events and continue best-effort parsing.
@@ -41,12 +51,21 @@ function readSSEEvents(rawSSE) {
     }
 
     if (line.startsWith("event:")) {
+      // Some relays omit the blank separator between Claude events. Flush the
+      // previous event before accepting the next event name.
+      if (currentData.length > 0) flush();
       currentEvent = line.slice(6).trim();
       continue;
     }
 
     if (line.startsWith("data:")) {
-      currentData.push(line.slice(5).trimStart());
+      const dataLine = line.slice(5).trimStart();
+      if (dataLine.trim() === "[DONE]") {
+        flush();
+        currentEvent = "";
+        continue;
+      }
+      currentData.push(dataLine);
     }
   }
 

open-sse/utils/streamReadiness.ts

@@ -66,6 +66,20 @@ function hasUsefulJsonPayload(payload: unknown): boolean {
   return hasUsefulValue(payload);
 }
 
+function hasAcceptedStreamStartPayload(payload: unknown, eventType = ""): boolean {
+  if (!isRecord(payload)) return false;
+
+  // Anthropic/Claude streams can legitimately start with lifecycle frames and
+  // then spend a long time thinking before the first text/tool delta arrives.
+  // Treating the start frame as readiness prevents false 504s while ping-only
+  // 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;
+
+  return false;
+}
+
 export function hasUsefulStreamContent(text: string): boolean {
   const lines = text.split(/\r?\n/);
 
@@ -88,6 +102,60 @@ export function hasUsefulStreamContent(text: string): boolean {
   return false;
 }
 
+type StreamReadinessSignalState = {
+  currentEvent: string;
+  pendingLine: string;
+};
+
+function processStreamReadinessLine(state: StreamReadinessSignalState, line: string): boolean {
+  const trimmed = line.trim();
+  if (!trimmed || trimmed.startsWith(":")) {
+    if (!trimmed) state.currentEvent = "";
+    return false;
+  }
+
+  if (trimmed.startsWith("event:")) {
+    state.currentEvent = trimmed.slice(6).trim();
+    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;
+  }
+}
+
+function appendStreamReadinessSignal(state: StreamReadinessSignalState, chunk: string): boolean {
+  const lines = `${state.pendingLine}${chunk}`.split(/\r?\n/);
+  state.pendingLine = lines.pop() ?? "";
+
+  for (const line of lines) {
+    if (processStreamReadinessLine(state, line)) return true;
+  }
+
+  return false;
+}
+
+export function hasStreamReadinessSignal(text: string): boolean {
+  const state: StreamReadinessSignalState = {
+    currentEvent: "",
+    pendingLine: "",
+  };
+  if (appendStreamReadinessSignal(state, text)) return true;
+  if (state.pendingLine) return processStreamReadinessLine(state, state.pendingLine);
+  return false;
+}
+
 function createErrorResponse(
   status: number,
   message: string,
@@ -170,7 +238,10 @@ export async function ensureStreamReadiness(
   const reader = response.body.getReader();
   const chunks: Uint8Array[] = [];
   const decoder = new TextDecoder();
-  let bufferedText = "";
+  const readinessState: StreamReadinessSignalState = {
+    currentEvent: "",
+    pendingLine: "",
+  };
   const startedAt = Date.now();
   const deadline = startedAt + options.timeoutMs;
   let handedOffReader = false;
@@ -245,9 +316,9 @@ export async function ensureStreamReadiness(
 
       if (!readResult.value) continue;
       chunks.push(readResult.value);
-      bufferedText += decoder.decode(readResult.value, { stream: true });
+      const decodedChunk = decoder.decode(readResult.value, { stream: true });
 
-      if (hasUsefulStreamContent(bufferedText)) {
+      if (appendStreamReadinessSignal(readinessState, decodedChunk)) {
         options.log?.debug?.(
           "STREAM",
           `Stream readiness confirmed in ${Date.now() - startedAt}ms (${options.provider || "provider"}/${options.model || "unknown"})`

tests/unit/cc-compatible-provider.test.ts

@@ -571,6 +571,91 @@ test("handleChatCore forces SSE upstream for CC compatible providers while retur
   assert.equal(payload.usage.completion_tokens, 5);
 });
 
+test("handleChatCore stops buffering CC-compatible SSE once a non-stream response completes", async () => {
+  const encoder = new TextEncoder();
+  let upstreamCancelled = false;
+  const upstreamChunks = [
+    "data:\n\n",
+    [
+      "event: message_start",
+      'data: {"type":"message_start","message":{"id":"msg_3","type":"message","role":"assistant","model":"claude-sonnet-4-6","usage":{"input_tokens":4,"output_tokens":0}}}',
+      "",
+      "event: content_block_delta",
+      'data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Finished but connection stayed open"}}',
+      "",
+      "event: message_delta",
+      'data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":6}}',
+      "",
+      "event: message_stop",
+      'data: {"type":"message_stop"}',
+      "",
+    ].join("\n"),
+  ];
+  let chunkIndex = 0;
+
+  globalThis.fetch = async () =>
+    new Response(
+      new ReadableStream<Uint8Array>({
+        pull(controller) {
+          if (chunkIndex < upstreamChunks.length) {
+            controller.enqueue(encoder.encode(upstreamChunks[chunkIndex++]));
+          }
+        },
+        cancel() {
+          upstreamCancelled = true;
+        },
+      }),
+      {
+        status: 200,
+        headers: {
+          "content-type": "text/event-stream",
+        },
+      }
+    );
+
+  const result = await handleChatCore({
+    body: {
+      model: "claude-sonnet-4-6",
+      messages: [{ role: "user", content: "Ping" }],
+      stream: false,
+    },
+    modelInfo: {
+      provider: "anthropic-compatible-cc-test",
+      model: "claude-sonnet-4-6",
+      extendedContext: false,
+    },
+    credentials: {
+      apiKey: "sk-test",
+      providerSpecificData: {
+        baseUrl: "https://proxy.example.com",
+        chatPath: CLAUDE_CODE_COMPATIBLE_DEFAULT_CHAT_PATH,
+      },
+    },
+    clientRawRequest: {
+      endpoint: "/v1/chat/completions",
+      body: {
+        model: "claude-sonnet-4-6",
+        messages: [{ role: "user", content: "Ping" }],
+        stream: false,
+      },
+      headers: new Headers({ accept: "application/json" }),
+    },
+    userAgent: "unit-test",
+    log: {
+      debug() {},
+      info() {},
+      warn() {},
+      error() {},
+    },
+  });
+
+  assert.equal(result.success, true);
+  assert.equal(upstreamCancelled, true);
+  const payload = (await result.response.json()) as any;
+  assert.equal(payload.choices[0].message.content, "Finished but connection stayed open");
+  assert.equal(payload.usage.completion_tokens, 6);
+});
+
 test("handleChatCore preserves client cache markers for Claude Code requests to CC-compatible providers", async () => {
   const calls = [];
   globalThis.fetch = async (url, init = {}) => {

tests/unit/sse-parser.test.ts

@@ -114,6 +114,26 @@ test("parseSSEToClaudeResponse parses text, thinking, tool_use, and usage events
   assert.deepEqual(parsed.usage, { input_tokens: 10, output_tokens: 4 });
 });
 
+test("parseSSEToClaudeResponse tolerates event-only types and missing blank separators", () => {
+  const rawSSE = [
+    "event: message_start",
+    'data: {"message":{"id":"msg_event_fallback","model":"claude-sonnet-4-6","role":"assistant","usage":{"input_tokens":3}}}',
+    "event: content_block_delta",
+    'data: {"index":0,"delta":{"text":"event fallback ok"}}',
+    "event: message_delta",
+    'data: {"delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":2}}',
+    "event: message_stop",
+    "data: {}",
+  ].join("\n");
+
+  const parsed = parseSSEToClaudeResponse(rawSSE, "fallback-model");
+
+  assert.equal(parsed.id, "msg_event_fallback");
+  assert.equal(parsed.model, "claude-sonnet-4-6");
+  assert.equal((parsed.content[0] as any).text, "event fallback ok");
+  assert.deepEqual(parsed.usage, { input_tokens: 3, output_tokens: 2 });
+});
+
 test("parseSSEToClaudeResponse ignores malformed payloads and returns null when nothing valid remains", () => {
   const parsed = parseSSEToClaudeResponse(
     ["event: content_block_delta", "data: not-json", "", "data: [DONE]"].join("\n"),

tests/unit/stream-readiness.test.ts

@@ -3,6 +3,7 @@ import assert from "node:assert/strict";
 
 import {
   ensureStreamReadiness,
+  hasStreamReadinessSignal,
   hasUsefulStreamContent,
 } from "../../open-sse/utils/streamReadiness.ts";
 
@@ -20,6 +21,46 @@ function streamFromChunks(chunks: string[], delayMs = 0): ReadableStream<Uint8Ar
   });
 }
 
+function delayedClaudeStartStream(): ReadableStream<Uint8Array> {
+  return new ReadableStream<Uint8Array>({
+    async start(controller) {
+      controller.enqueue(
+        encoder.encode(
+          [
+            "event: message_start",
+            `data: ${JSON.stringify({
+              message: {
+                id: "msg_1",
+                type: "message",
+                role: "assistant",
+                model: "claude-sonnet-4-6",
+                usage: { input_tokens: 10, output_tokens: 0 },
+              },
+            })}`,
+            "",
+          ].join("\n")
+        )
+      );
+
+      await new Promise((resolve) => setTimeout(resolve, 30));
+      controller.enqueue(
+        encoder.encode(
+          [
+            "event: content_block_delta",
+            `data: ${JSON.stringify({
+              type: "content_block_delta",
+              index: 0,
+              delta: { type: "text_delta", text: "slow hello" },
+            })}`,
+            "",
+          ].join("\n")
+        )
+      );
+      controller.close();
+    },
+  });
+}
+
 test("hasUsefulStreamContent ignores keepalives and lifecycle-only chunks", () => {
   assert.equal(hasUsefulStreamContent(": keepalive\n\n"), false);
   assert.equal(hasUsefulStreamContent("event: ping\ndata: {}\n\n"), false);
@@ -80,6 +121,51 @@ test("hasUsefulStreamContent detects text, reasoning, and tool deltas", () => {
   );
 });
 
+test("hasStreamReadinessSignal accepts Claude stream start events", () => {
+  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
+  );
+  assert.equal(
+    hasStreamReadinessSignal(
+      `event: message_start\ndata: ${JSON.stringify({
+        type: "message_start",
+        message: {
+          id: "msg_1",
+          type: "message",
+          role: "assistant",
+          model: "claude-sonnet-4-6",
+        },
+      })}\n\n`
+    ),
+    true
+  );
+  assert.equal(
+    hasStreamReadinessSignal(
+      `event: message_start\ndata: ${JSON.stringify({
+        message: {
+          id: "msg_2",
+          type: "message",
+          role: "assistant",
+          model: "claude-sonnet-4-6",
+        },
+      })}\n\n`
+    ),
+    true
+  );
+  assert.equal(
+    hasStreamReadinessSignal(
+      `event: content_block_start\ndata: ${JSON.stringify({
+        index: 0,
+        content_block: { type: "text", text: "" },
+      })}\n\n`
+    ),
+    true
+  );
+});
+
 test("ensureStreamReadiness preserves buffered chunks when stream starts", async () => {
   const response = new Response(
     streamFromChunks([
@@ -98,6 +184,23 @@ test("ensureStreamReadiness preserves buffered chunks when stream starts", async
   assert.match(text, / world/);
 });
 
+test("ensureStreamReadiness hands off long Claude streams after message_start", async () => {
+  const response = new Response(delayedClaudeStartStream(), {
+    status: 200,
+    headers: { "Content-Type": "text/event-stream" },
+  });
+
+  const result = await ensureStreamReadiness(response, {
+    timeoutMs: 10,
+    provider: "anthropic-compatible-cc-test",
+    model: "claude-sonnet-4-6",
+  });
+  assert.equal(result.ok, true);
+  const text = await result.response.text();
+  assert.match(text, /message_start/);
+  assert.match(text, /slow hello/);
+});
+
 test("ensureStreamReadiness returns 504 when no useful content arrives before timeout", async () => {
   const response = new Response(
     streamFromChunks(