fix(stream): prevent non-JSON SSE lines and duplicate [DONE] from breaking clients
- Passthrough: skip non-JSON data lines instead of forwarding raw garbage - Translate: stop emitting redundant [DONE] sentinel (message_stop terminates) - Add streamDoneSent flag to prevent duplicate [DONE] across transform + flush - Warn on unexpected upstream Content-Type for streaming responses PR #2046 by @qianze0628 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -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 });
|
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
|
// Responses passthrough: synthesize response.failed + [DONE] if the stream aborts/stalls before a terminal event
|
||||||
|
|||||||
@@ -71,6 +71,7 @@ export function createSSEStream(options = {}) {
|
|||||||
let currentOpenAIResponsesEvent = null;
|
let currentOpenAIResponsesEvent = null;
|
||||||
let openAIResponsesTerminalSeen = false;
|
let openAIResponsesTerminalSeen = false;
|
||||||
let openAIResponsesDoneSent = false;
|
let openAIResponsesDoneSent = false;
|
||||||
|
let streamDoneSent = false; // track duplicate [DONE] across transform + flush
|
||||||
|
|
||||||
return new TransformStream({
|
return new TransformStream({
|
||||||
transform(chunk, controller) {
|
transform(chunk, controller) {
|
||||||
@@ -166,7 +167,12 @@ export function createSSEStream(options = {}) {
|
|||||||
output = `data: ${JSON.stringify(parsed)}\n`;
|
output = `data: ${JSON.stringify(parsed)}\n`;
|
||||||
injectedUsage = true;
|
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) {
|
if (!injectedUsage) {
|
||||||
@@ -211,9 +217,10 @@ export function createSSEStream(options = {}) {
|
|||||||
sseEmittedCount++;
|
sseEmittedCount++;
|
||||||
}
|
}
|
||||||
|
|
||||||
const output = "data: [DONE]\n\n";
|
// [DONE] not emitted in translate mode — some clients' SSE decoders
|
||||||
reqLogger?.appendConvertedChunk?.(output);
|
// fail to parse the OpenAI sentinel on Claude-format translated streams.
|
||||||
controller.enqueue(sharedEncoder.encode(output));
|
// message_stop already signals end-of-response; stream close handles it.
|
||||||
|
streamDoneSent = true;
|
||||||
if (keepsOpenAIResponsesFormat) openAIResponsesDoneSent = true;
|
if (keepsOpenAIResponsesFormat) openAIResponsesDoneSent = true;
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
@@ -343,9 +350,11 @@ export function createSSEStream(options = {}) {
|
|||||||
// Some clients (e.g. OpenClaw) expect the OpenAI-style sentinel:
|
// Some clients (e.g. OpenClaw) expect the OpenAI-style sentinel:
|
||||||
// data: [DONE]\n\n
|
// data: [DONE]\n\n
|
||||||
// Without it they can hang until timeout and trigger failover.
|
// Without it they can hang until timeout and trigger failover.
|
||||||
|
if (!streamDoneSent) {
|
||||||
const doneOutput = "data: [DONE]\n\n";
|
const doneOutput = "data: [DONE]\n\n";
|
||||||
reqLogger?.appendConvertedChunk?.(doneOutput);
|
reqLogger?.appendConvertedChunk?.(doneOutput);
|
||||||
controller.enqueue(sharedEncoder.encode(doneOutput));
|
controller.enqueue(sharedEncoder.encode(doneOutput));
|
||||||
|
}
|
||||||
|
|
||||||
if (onStreamComplete) {
|
if (onStreamComplete) {
|
||||||
onStreamComplete({
|
onStreamComplete({
|
||||||
@@ -406,11 +415,8 @@ export function createSSEStream(options = {}) {
|
|||||||
openAIResponsesTerminalSeen = true;
|
openAIResponsesTerminalSeen = true;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (!keepsOpenAIResponsesFormat || !openAIResponsesDoneSent) {
|
// [DONE] not emitted in translate mode — see comment above.
|
||||||
const doneOutput = "data: [DONE]\n\n";
|
// Passthrough mode still emits it for standard OpenAI clients.
|
||||||
reqLogger?.appendConvertedChunk?.(doneOutput);
|
|
||||||
controller.enqueue(sharedEncoder.encode(doneOutput));
|
|
||||||
}
|
|
||||||
|
|
||||||
if (!hasValidUsage(state?.usage) && totalContentLength > 0) {
|
if (!hasValidUsage(state?.usage) && totalContentLength > 0) {
|
||||||
state.usage = estimateUsage(body, totalContentLength, sourceFormat);
|
state.usage = estimateUsage(body, totalContentLength, sourceFormat);
|
||||||
|
|||||||
Reference in New Issue
Block a user