diff --git a/open-sse/handlers/chatCore/streamingHandler.js b/open-sse/handlers/chatCore/streamingHandler.js index d91bb3a2..41be3d7d 100644 --- a/open-sse/handlers/chatCore/streamingHandler.js +++ b/open-sse/handlers/chatCore/streamingHandler.js @@ -52,6 +52,14 @@ export function handleStreamingResponse({ providerResponse, provider, model, sou }); } + // Warn when upstream returns unexpected Content-Type for a streaming response. + // This often means the provider returned an HTML error page or plain-text error + // that the SSE transform stream would forward as garbage to the client. + const upstreamContentType = (providerResponse.headers.get('content-type') || '').toLowerCase(); + if (upstreamContentType && !upstreamContentType.includes('text/event-stream') && !upstreamContentType.includes('application/json')) { + console.warn('[STREAM] ' + provider + ' | ' + model + ' | unexpected Content-Type: ' + upstreamContentType); + } + const transformStream = buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey }); // Responses passthrough: synthesize response.failed + [DONE] if the stream aborts/stalls before a terminal event diff --git a/open-sse/utils/stream.js b/open-sse/utils/stream.js index 7f843734..387891aa 100644 --- a/open-sse/utils/stream.js +++ b/open-sse/utils/stream.js @@ -71,6 +71,7 @@ export function createSSEStream(options = {}) { let currentOpenAIResponsesEvent = null; let openAIResponsesTerminalSeen = false; let openAIResponsesDoneSent = false; + let streamDoneSent = false; // track duplicate [DONE] across transform + flush return new TransformStream({ transform(chunk, controller) { @@ -166,7 +167,12 @@ export function createSSEStream(options = {}) { output = `data: ${JSON.stringify(parsed)}\n`; injectedUsage = true; } - } catch { } + } catch { + // Skip non-JSON data lines silently — don't forward garbage to clients. + // Upstream providers sometimes return plain-text errors (HTML, rate-limit + // messages) in the SSE stream that would break downstream JSON decoders. + continue; + } } if (!injectedUsage) { @@ -211,9 +217,10 @@ export function createSSEStream(options = {}) { sseEmittedCount++; } - const output = "data: [DONE]\n\n"; - reqLogger?.appendConvertedChunk?.(output); - controller.enqueue(sharedEncoder.encode(output)); + // [DONE] not emitted in translate mode — some clients' SSE decoders + // fail to parse the OpenAI sentinel on Claude-format translated streams. + // message_stop already signals end-of-response; stream close handles it. + streamDoneSent = true; if (keepsOpenAIResponsesFormat) openAIResponsesDoneSent = true; continue; } @@ -343,9 +350,11 @@ export function createSSEStream(options = {}) { // Some clients (e.g. OpenClaw) expect the OpenAI-style sentinel: // data: [DONE]\n\n // Without it they can hang until timeout and trigger failover. - const doneOutput = "data: [DONE]\n\n"; - reqLogger?.appendConvertedChunk?.(doneOutput); - controller.enqueue(sharedEncoder.encode(doneOutput)); + if (!streamDoneSent) { + const doneOutput = "data: [DONE]\n\n"; + reqLogger?.appendConvertedChunk?.(doneOutput); + controller.enqueue(sharedEncoder.encode(doneOutput)); + } if (onStreamComplete) { onStreamComplete({ @@ -406,11 +415,8 @@ export function createSSEStream(options = {}) { openAIResponsesTerminalSeen = true; } - if (!keepsOpenAIResponsesFormat || !openAIResponsesDoneSent) { - const doneOutput = "data: [DONE]\n\n"; - reqLogger?.appendConvertedChunk?.(doneOutput); - controller.enqueue(sharedEncoder.encode(doneOutput)); - } + // [DONE] not emitted in translate mode — see comment above. + // Passthrough mode still emits it for standard OpenAI clients. if (!hasValidUsage(state?.usage) && totalContentLength > 0) { state.usage = estimateUsage(body, totalContentLength, sourceFormat);