Repository navigation
fix: make passthrough stream format-aware for Responses SSE (#186) #188
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -136,51 +136,71 @@ export function createSSEStream(options: any = {}) { | |
| try { | ||
| let parsed = JSON.parse(trimmed.slice(5).trim()); | ||
|
|
||
| // Sanitize: strip non-standard fields for OpenAI SDK compatibility | ||
| parsed = sanitizeStreamingChunk(parsed); | ||
| // Detect Responses SSE payloads (have a `type` field like "response.created", | ||
| // "response.output_item.added", etc.) and skip Chat Completions-specific | ||
| // sanitization to avoid corrupting the stream for Responses-native clients. | ||
| const isResponsesSSE = | ||
| parsed.type && | ||
| typeof parsed.type === "string" && | ||
| parsed.type.startsWith("response."); | ||
|
|
||
| if (isResponsesSSE) { | ||
| // Responses SSE: only extract usage, forward payload as-is | ||
| const extracted = extractUsage(parsed); | ||
| if (extracted) { | ||
| usage = extracted; | ||
| } | ||
| // Track content length from Responses format | ||
| if (parsed.delta && typeof parsed.delta === "string") { | ||
| totalContentLength += parsed.delta.length; | ||
| } | ||
| } else { | ||
| // Chat Completions: full sanitization pipeline | ||
| parsed = sanitizeStreamingChunk(parsed); | ||
|
|
||
| const idFixed = fixInvalidId(parsed); | ||
| const idFixed = fixInvalidId(parsed); | ||
|
|
||
| if (!hasValuableContent(parsed, FORMATS.OPENAI)) { | ||
| continue; | ||
| } | ||
| if (!hasValuableContent(parsed, FORMATS.OPENAI)) { | ||
| continue; | ||
| } | ||
|
Comment on lines
+157
to
+165
|
||
|
|
||
| const delta = parsed.choices?.[0]?.delta; | ||
| const delta = parsed.choices?.[0]?.delta; | ||
|
|
||
| // 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; | ||
| // 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; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| const content = delta?.content || delta?.reasoning_content; | ||
| if (content && typeof content === "string") { | ||
| totalContentLength += content.length; | ||
| } | ||
| const content = delta?.content || delta?.reasoning_content; | ||
| if (content && typeof content === "string") { | ||
| totalContentLength += content.length; | ||
| } | ||
|
|
||
| const extracted = extractUsage(parsed); | ||
| if (extracted) { | ||
| usage = extracted; | ||
| } | ||
| const extracted = extractUsage(parsed); | ||
| if (extracted) { | ||
| usage = extracted; | ||
| } | ||
|
|
||
| const isFinishChunk = parsed.choices?.[0]?.finish_reason; | ||
| if (isFinishChunk && !hasValidUsage(parsed.usage)) { | ||
| const estimated = estimateUsage(body, totalContentLength, FORMATS.OPENAI); | ||
| parsed.usage = filterUsageForFormat(estimated, FORMATS.OPENAI); | ||
| output = `data: ${JSON.stringify(parsed)}\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`; | ||
| injectedUsage = true; | ||
| } else if (idFixed) { | ||
| output = `data: ${JSON.stringify(parsed)}\n`; | ||
| injectedUsage = true; | ||
| const isFinishChunk = parsed.choices?.[0]?.finish_reason; | ||
| if (isFinishChunk && !hasValidUsage(parsed.usage)) { | ||
| const estimated = estimateUsage(body, totalContentLength, FORMATS.OPENAI); | ||
| parsed.usage = filterUsageForFormat(estimated, FORMATS.OPENAI); | ||
| output = `data: ${JSON.stringify(parsed)}\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`; | ||
| injectedUsage = true; | ||
| } else if (idFixed) { | ||
| output = `data: ${JSON.stringify(parsed)}\n`; | ||
| injectedUsage = true; | ||
| } | ||
| } | ||
|
Comment on lines
+142
to
204
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The logic for extracting usage is duplicated in both the const isResponsesSSE =
parsed.type &&
typeof parsed.type === "string" &&
parsed.type.startsWith("response.");
const extracted = extractUsage(parsed);
if (extracted) {
usage = extracted;
}
if (isResponsesSSE) {
// Responses SSE: only track content length, forward payload as-is
if (parsed.delta && typeof parsed.delta === "string") {
totalContentLength += parsed.delta.length;
}
} else {
// Chat Completions: full sanitization pipeline
parsed = sanitizeStreamingChunk(parsed);
const idFixed = fixInvalidId(parsed);
if (!hasValuableContent(parsed, FORMATS.OPENAI)) {
continue;
}
const delta = parsed.choices?.[0]?.delta;
// 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;
}
}
const content = delta?.content || delta?.reasoning_content;
if (content && typeof content === "string") {
totalContentLength += content.length;
}
const isFinishChunk = parsed.choices?.[0]?.finish_reason;
if (isFinishChunk && !hasValidUsage(parsed.usage)) {
const estimated = estimateUsage(body, totalContentLength, FORMATS.OPENAI);
parsed.usage = filterUsageForFormat(estimated, FORMATS.OPENAI);
output = `data: ${JSON.stringify(parsed)}\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`;
injectedUsage = true;
} else if (idFixed) {
output = `data: ${JSON.stringify(parsed)}\n`;
injectedUsage = true;
}
} |
||
| } catch {} | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. WARNING: Silent exception swallowing. The |
||
| } | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
WARNING: Incomplete content length tracking for Responses SSE format.
The code only tracks content length when
parsed.deltaexists (line 154-156), but Responses SSE events can contain content in other ways:response.output_item.addedwithitem.content[].textresponse.content_part.addedwithpart.textCompare to Chat Completions handling at lines 178-180 which checks both
delta.contentanddelta.reasoning_content. This could lead to underestimation of usage for Responses SSE format when content arrives in non-delta events.