From cc79237af725816b9244c15bd6c9835f24536b82 Mon Sep 17 00:00:00 2001 From: diegosouzapw Date: Sun, 10 May 2026 09:44:11 -0300 Subject: [PATCH] Fix CC-compatible streaming bridge --- open-sse/handlers/chatCore.ts | 154 +++++++++++++++++++++- open-sse/handlers/sseParser.ts | 23 +++- open-sse/utils/streamReadiness.ts | 77 ++++++++++- tests/unit/cc-compatible-provider.test.ts | 85 ++++++++++++ tests/unit/sse-parser.test.ts | 20 +++ tests/unit/stream-readiness.test.ts | 103 +++++++++++++++ 6 files changed, 455 insertions(+), 7 deletions(-) diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 74b9d7622e..5e60a86260 100644 --- a/open-sse/handlers/chatCore.ts +++ b/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, + timeoutMs: number +): Promise> { + 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 { + if ( + !upstreamStream || + !response.body || + (!contentType.includes("text/event-stream") && !contentType.includes("application/x-ndjson")) + ) { + return withBodyTimeout(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 | 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(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(providerResponse.text()); + const rawBody = await readNonStreamingResponseBody( + providerResponse, + contentType, + upstreamStream + ); const normalizedProviderPayload = normalizePayloadForLog(rawBody); const looksLikeSSE = contentType.includes("text/event-stream") || diff --git a/open-sse/handlers/sseParser.ts b/open-sse/handlers/sseParser.ts index 44beadf0da..7432590a4f 100644 --- a/open-sse/handlers/sseParser.ts +++ b/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); } } diff --git a/open-sse/utils/streamReadiness.ts b/open-sse/utils/streamReadiness.ts index 5899d27106..ef6cb71918 100644 --- a/open-sse/utils/streamReadiness.ts +++ b/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"})` diff --git a/tests/unit/cc-compatible-provider.test.ts b/tests/unit/cc-compatible-provider.test.ts index 0f03139ddf..5eac9493c8 100644 --- a/tests/unit/cc-compatible-provider.test.ts +++ b/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({ + 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 = {}) => { diff --git a/tests/unit/sse-parser.test.ts b/tests/unit/sse-parser.test.ts index 82b4fd655f..3d8bbfdf7f 100644 --- a/tests/unit/sse-parser.test.ts +++ b/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"), diff --git a/tests/unit/stream-readiness.test.ts b/tests/unit/stream-readiness.test.ts index a86c37dd31..15eb22654d 100644 --- a/tests/unit/stream-readiness.test.ts +++ b/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 { + return new ReadableStream({ + 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(