f9d82c6575
createSSEStream splits on "\n" and keeps the remainder, which only flush() parses. That call omitted targetFormat, so parseSSELine required a "data: " prefix and dropped whatever an NDJSON provider left without a closing newline. The !parsed.done guard compounded it: the SSE sentinel and an Ollama final chunk both carry done:true, but the latter is the real last chunk holding done_reason and the token counts. Pass targetFormat and scope the sentinel check to formats that emit one, so the tail reaches the translator. Accumulate its usage into state the same way the transform loop does, so finalizeStream logs those tokens instead of null.
519 lines
21 KiB
JavaScript
519 lines
21 KiB
JavaScript
import { translateResponse, initState } from "../translator/index.js";
|
|
import { FORMATS } from "../translator/formats.js";
|
|
import { trackPendingRequest, appendRequestLog } from "@/lib/usageDb.js";
|
|
import { extractUsage, mergeUsage, hasValidUsage, estimateUsage, logUsage, addBufferToUsage, filterUsageForFormat, COLORS } from "./usageTracking.js";
|
|
import { parseSSELine, hasValuableContent, fixInvalidId, formatSSE } from "./streamHelpers.js";
|
|
import { getOpenAIResponsesEventName, isOpenAIResponsesTerminalEvent, formatIncompleteOpenAIResponsesStreamFailure } from "./responsesStreamHelpers.js";
|
|
import { dbg, isDebugEnabled } from "./debugLog.js";
|
|
|
|
import { SSE_DONE, SSE_HEADERS, SSE_HEADERS_NO_BUFFER } from "./sseConstants.js";
|
|
|
|
export { COLORS, formatSSE };
|
|
export { SSE_DONE, SSE_HEADERS, SSE_HEADERS_NO_BUFFER };
|
|
|
|
// sharedEncoder is stateless — safe to share across streams
|
|
const sharedEncoder = new TextEncoder();
|
|
|
|
/**
|
|
* Stream modes
|
|
*/
|
|
const STREAM_MODE = {
|
|
TRANSLATE: "translate", // Full translation between formats
|
|
PASSTHROUGH: "passthrough" // No translation, normalize output, extract usage
|
|
};
|
|
|
|
/**
|
|
* Create unified SSE transform stream
|
|
* @param {object} options
|
|
* @param {string} options.mode - Stream mode: translate, passthrough
|
|
* @param {string} options.targetFormat - Provider format (for translate mode)
|
|
* @param {string} options.sourceFormat - Client format (for translate mode)
|
|
* @param {string} options.provider - Provider name
|
|
* @param {object} options.reqLogger - Request logger instance
|
|
* @param {string} options.model - Model name
|
|
* @param {string} options.connectionId - Connection ID for usage tracking
|
|
* @param {object} options.body - Request body (for input token estimation)
|
|
* @param {function} options.onStreamComplete - Callback when stream completes (content, usage)
|
|
* @param {string} options.apiKey - API key for usage tracking
|
|
*/
|
|
export function createSSEStream(options = {}) {
|
|
const {
|
|
mode = STREAM_MODE.TRANSLATE,
|
|
targetFormat,
|
|
sourceFormat,
|
|
provider = null,
|
|
reqLogger = null,
|
|
toolNameMap = null,
|
|
customToolNames = null,
|
|
model = null,
|
|
connectionId = null,
|
|
body = null,
|
|
onStreamComplete = null,
|
|
apiKey = null
|
|
} = options;
|
|
|
|
let buffer = "";
|
|
let usage = null;
|
|
|
|
// Per-stream decoder with stream:true to correctly handle multi-byte chars split across chunks
|
|
const decoder = new TextDecoder("utf-8", { fatal: false });
|
|
|
|
const state = mode === STREAM_MODE.TRANSLATE
|
|
? { ...initState(sourceFormat), provider, toolNameMap, customToolNames: new Set(customToolNames || []), model }
|
|
: null;
|
|
|
|
let totalContentLength = 0;
|
|
let accumulatedContent = "";
|
|
let accumulatedThinking = "";
|
|
let ttftAt = null;
|
|
let sseLineCount = 0;
|
|
let sseEmittedCount = 0;
|
|
const eventTypeCounts = {};
|
|
|
|
// Track Responses API event framing for same-format passthrough (codex)
|
|
let currentOpenAIResponsesEvent = null;
|
|
let openAIResponsesTerminalSeen = false;
|
|
let openAIResponsesDoneSent = false;
|
|
let streamDoneSent = false; // track duplicate [DONE] across transform + flush
|
|
let finalized = false;
|
|
|
|
// Usage/logging tail, callable from transform() as well as flush(): a client that
|
|
// closes right after the terminal event cancels the reader, and flush() never runs.
|
|
const finalizeStream = () => {
|
|
if (finalized) return;
|
|
finalized = true;
|
|
|
|
const isPassthrough = mode === STREAM_MODE.PASSTHROUGH;
|
|
let finalUsage = isPassthrough ? usage : state?.usage;
|
|
|
|
if (!hasValidUsage(finalUsage) && totalContentLength > 0) {
|
|
finalUsage = estimateUsage(body, totalContentLength, isPassthrough ? FORMATS.OPENAI : sourceFormat);
|
|
if (isPassthrough) usage = finalUsage; else state.usage = finalUsage;
|
|
}
|
|
|
|
if (hasValidUsage(finalUsage)) {
|
|
logUsage(isPassthrough ? provider : (state?.provider || targetFormat), finalUsage, model, connectionId, apiKey);
|
|
} else {
|
|
appendRequestLog({ model, provider, connectionId, tokens: null, status: "200 OK" }).catch(() => { });
|
|
}
|
|
|
|
if (onStreamComplete) {
|
|
onStreamComplete({
|
|
content: accumulatedContent,
|
|
thinking: accumulatedThinking
|
|
}, finalUsage, ttftAt);
|
|
}
|
|
};
|
|
|
|
return new TransformStream({
|
|
transform(chunk, controller) {
|
|
if (!ttftAt) ttftAt = Date.now();
|
|
const text = decoder.decode(chunk, { stream: true });
|
|
buffer += text;
|
|
reqLogger?.appendProviderChunk?.(text);
|
|
|
|
const lines = buffer.split("\n");
|
|
buffer = lines.pop() || "";
|
|
|
|
for (const line of lines) {
|
|
const trimmed = line.trim();
|
|
if (isDebugEnabled && trimmed) {
|
|
sseLineCount++;
|
|
if (trimmed.startsWith("event:")) {
|
|
const evt = trimmed.slice(6).trim();
|
|
eventTypeCounts[evt] = (eventTypeCounts[evt] || 0) + 1;
|
|
}
|
|
}
|
|
|
|
// Capture Responses API event name to preserve framing in same-format passthrough
|
|
if (mode === STREAM_MODE.TRANSLATE && targetFormat === FORMATS.OPENAI_RESPONSES && trimmed.startsWith("event:")) {
|
|
currentOpenAIResponsesEvent = trimmed.slice(6).trim();
|
|
}
|
|
|
|
// Passthrough mode: normalize and forward
|
|
if (mode === STREAM_MODE.PASSTHROUGH) {
|
|
let output;
|
|
let injectedUsage = false;
|
|
let responsesTerminal = false;
|
|
|
|
if (trimmed.startsWith("data:") && trimmed.slice(5).trim() !== "[DONE]") {
|
|
try {
|
|
const parsed = JSON.parse(trimmed.slice(5).trim());
|
|
|
|
const idFixed = fixInvalidId(parsed);
|
|
|
|
// Ensure OpenAI-required fields are present on streaming chunks (Letta compat)
|
|
let fieldsInjected = false;
|
|
if (parsed.choices !== undefined) {
|
|
if (!parsed.object) { parsed.object = "chat.completion.chunk"; fieldsInjected = true; }
|
|
if (!parsed.created) { parsed.created = Math.floor(Date.now() / 1000); fieldsInjected = true; }
|
|
}
|
|
|
|
// Strip Azure-specific non-standard fields from streaming chunks
|
|
if (parsed.prompt_filter_results !== undefined) {
|
|
delete parsed.prompt_filter_results;
|
|
fieldsInjected = true;
|
|
}
|
|
if (parsed?.choices) {
|
|
for (const choice of parsed.choices) {
|
|
if (choice.content_filter_results !== undefined) {
|
|
delete choice.content_filter_results;
|
|
fieldsInjected = true;
|
|
}
|
|
}
|
|
}
|
|
|
|
// Strip empty tool_calls arrays that break AI SDK reasoning tracking.
|
|
// Some providers (e.g. CodeBuddy CN) include `"tool_calls": []` in
|
|
// every streaming delta. @ai-sdk/openai-compatible checks
|
|
// `delta.tool_calls != null` — an empty array passes this check,
|
|
// causing premature `reasoning-end` on every chunk.
|
|
if (parsed?.choices) {
|
|
for (const choice of parsed.choices) {
|
|
if (choice.delta?.tool_calls && Array.isArray(choice.delta.tool_calls) && choice.delta.tool_calls.length === 0) {
|
|
delete choice.delta.tool_calls;
|
|
fieldsInjected = true;
|
|
}
|
|
}
|
|
}
|
|
|
|
if (!hasValuableContent(parsed, FORMATS.OPENAI)) {
|
|
continue;
|
|
}
|
|
|
|
const delta = parsed.choices?.[0]?.delta;
|
|
const content = delta?.content;
|
|
const reasoning = delta?.reasoning_content;
|
|
if (content && typeof content === "string") {
|
|
totalContentLength += content.length;
|
|
accumulatedContent += content;
|
|
}
|
|
if (reasoning && typeof reasoning === "string") {
|
|
totalContentLength += reasoning.length;
|
|
accumulatedThinking += reasoning;
|
|
}
|
|
|
|
const extracted = extractUsage(parsed);
|
|
if (extracted) {
|
|
usage = mergeUsage(usage, extracted);
|
|
}
|
|
|
|
responsesTerminal = isOpenAIResponsesTerminalEvent(currentOpenAIResponsesEvent, parsed);
|
|
|
|
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 || fieldsInjected) {
|
|
output = `data: ${JSON.stringify(parsed)}\n`;
|
|
injectedUsage = true;
|
|
}
|
|
} 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 (line.startsWith("data:") && !line.startsWith("data: ")) {
|
|
output = "data: " + line.slice(5) + "\n";
|
|
} else {
|
|
output = line + "\n";
|
|
}
|
|
}
|
|
|
|
reqLogger?.appendConvertedChunk?.(output);
|
|
controller.enqueue(sharedEncoder.encode(output));
|
|
// Responses clients (codex CLI) close on response.completed instead of [DONE]
|
|
if (responsesTerminal) finalizeStream();
|
|
continue;
|
|
}
|
|
|
|
// Translate mode
|
|
if (!trimmed) continue;
|
|
|
|
const parsed = parseSSELine(trimmed, targetFormat);
|
|
if (!parsed) continue;
|
|
|
|
// Responses API same-format passthrough: preserve event framing + track terminal state
|
|
const isOpenAIResponsesStream = targetFormat === FORMATS.OPENAI_RESPONSES;
|
|
const keepsOpenAIResponsesFormat = isOpenAIResponsesStream && sourceFormat === FORMATS.OPENAI_RESPONSES;
|
|
const openAIResponsesEventName = isOpenAIResponsesStream
|
|
? getOpenAIResponsesEventName(currentOpenAIResponsesEvent, parsed)
|
|
: null;
|
|
|
|
if (isOpenAIResponsesStream && isOpenAIResponsesTerminalEvent(openAIResponsesEventName, parsed)) {
|
|
openAIResponsesTerminalSeen = true;
|
|
}
|
|
|
|
// For Ollama: done=true is the final chunk with finish_reason/usage, must translate
|
|
// For other formats: done=true is the [DONE] sentinel, skip
|
|
if (parsed && parsed.done && targetFormat !== FORMATS.OLLAMA) {
|
|
// Synthesize response.failed if the Responses stream never sent a terminal event
|
|
if (keepsOpenAIResponsesFormat && !openAIResponsesTerminalSeen) {
|
|
const failedOutput = formatIncompleteOpenAIResponsesStreamFailure();
|
|
reqLogger?.appendConvertedChunk?.(failedOutput);
|
|
controller.enqueue(sharedEncoder.encode(failedOutput));
|
|
openAIResponsesTerminalSeen = true;
|
|
sseEmittedCount++;
|
|
}
|
|
|
|
if (keepsOpenAIResponsesFormat && !streamDoneSent) {
|
|
const doneOutput = "data: [DONE]\n\n";
|
|
reqLogger?.appendConvertedChunk?.(doneOutput);
|
|
controller.enqueue(sharedEncoder.encode(doneOutput));
|
|
}
|
|
streamDoneSent = true;
|
|
if (keepsOpenAIResponsesFormat) openAIResponsesDoneSent = true;
|
|
continue;
|
|
}
|
|
|
|
// Claude format - content
|
|
if (parsed.delta?.text) {
|
|
totalContentLength += parsed.delta.text.length;
|
|
accumulatedContent += parsed.delta.text;
|
|
}
|
|
// Claude format - thinking
|
|
if (parsed.delta?.thinking) {
|
|
totalContentLength += parsed.delta.thinking.length;
|
|
accumulatedThinking += parsed.delta.thinking;
|
|
}
|
|
|
|
// OpenAI format - content
|
|
if (parsed.choices?.[0]?.delta?.content) {
|
|
totalContentLength += parsed.choices[0].delta.content.length;
|
|
accumulatedContent += parsed.choices[0].delta.content;
|
|
}
|
|
// OpenAI format - reasoning
|
|
if (parsed.choices?.[0]?.delta?.reasoning_content) {
|
|
totalContentLength += parsed.choices[0].delta.reasoning_content.length;
|
|
accumulatedThinking += parsed.choices[0].delta.reasoning_content;
|
|
}
|
|
|
|
// Gemini format
|
|
if (parsed.candidates?.[0]?.content?.parts) {
|
|
for (const part of parsed.candidates[0].content.parts) {
|
|
if (part.text && typeof part.text === "string") {
|
|
totalContentLength += part.text.length;
|
|
// Check if this is thinking content
|
|
if (part.thought === true) {
|
|
accumulatedThinking += part.text;
|
|
} else {
|
|
accumulatedContent += part.text;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Extract usage
|
|
const extracted = extractUsage(parsed);
|
|
if (extracted) state.usage = mergeUsage(state.usage, extracted); // Keep original usage for logging
|
|
|
|
// Responses same-format passthrough: re-emit with original event framing
|
|
if (keepsOpenAIResponsesFormat && openAIResponsesEventName) {
|
|
const output = formatSSE({ event: openAIResponsesEventName, data: parsed }, sourceFormat);
|
|
reqLogger?.appendConvertedChunk?.(output);
|
|
controller.enqueue(sharedEncoder.encode(output));
|
|
currentOpenAIResponsesEvent = null;
|
|
sseEmittedCount++;
|
|
// Responses clients (codex) close on response.completed instead of [DONE]
|
|
if (openAIResponsesTerminalSeen) finalizeStream();
|
|
continue;
|
|
}
|
|
|
|
currentOpenAIResponsesEvent = null;
|
|
|
|
// Translate: targetFormat -> openai -> sourceFormat
|
|
const translated = translateResponse(targetFormat, sourceFormat, parsed, state);
|
|
|
|
// Log OpenAI intermediate chunks (if available)
|
|
if (translated?._openaiIntermediate) {
|
|
for (const item of translated._openaiIntermediate) {
|
|
const openaiOutput = formatSSE(item, FORMATS.OPENAI);
|
|
reqLogger?.appendOpenAIChunk?.(openaiOutput);
|
|
}
|
|
}
|
|
|
|
if (translated?.length > 0) {
|
|
for (const item of translated) {
|
|
if (item === null || item === undefined) continue;
|
|
// Filter empty chunks
|
|
if (!hasValuableContent(item, sourceFormat)) {
|
|
continue; // Skip this empty chunk
|
|
}
|
|
|
|
// Inject estimated usage if finish chunk has no valid usage
|
|
const isFinishChunk = item.type === "message_delta" || item.choices?.[0]?.finish_reason;
|
|
if (state.finishReason && isFinishChunk && !hasValidUsage(item.usage) && totalContentLength > 0) {
|
|
const estimated = estimateUsage(body, totalContentLength, sourceFormat);
|
|
item.usage = filterUsageForFormat(estimated, sourceFormat); // Filter + already has buffer
|
|
state.usage = estimated;
|
|
} else if (state.finishReason && isFinishChunk && state.usage) {
|
|
// Add buffer and filter usage for client (but keep original in state.usage for logging)
|
|
const buffered = addBufferToUsage(state.usage);
|
|
item.usage = filterUsageForFormat(buffered, sourceFormat);
|
|
}
|
|
|
|
const output = formatSSE(item, sourceFormat);
|
|
reqLogger?.appendConvertedChunk?.(output);
|
|
controller.enqueue(sharedEncoder.encode(output));
|
|
sseEmittedCount++;
|
|
}
|
|
}
|
|
}
|
|
},
|
|
|
|
flush(controller) {
|
|
const evtSummary = Object.entries(eventTypeCounts).map(([k, v]) => `${k}=${v}`).join(",") || "none";
|
|
dbg("SSE", `flush | provider=${provider} | model=${model} | recvLines=${sseLineCount} | emitted=${sseEmittedCount} | events=[${evtSummary}]`);
|
|
trackPendingRequest(model, provider, connectionId, false);
|
|
try {
|
|
const remaining = decoder.decode();
|
|
if (remaining) buffer += remaining;
|
|
|
|
if (mode === STREAM_MODE.PASSTHROUGH) {
|
|
if (buffer) {
|
|
let output = buffer;
|
|
if (buffer.startsWith("data:") && !buffer.startsWith("data: ")) {
|
|
output = "data: " + buffer.slice(5);
|
|
}
|
|
reqLogger?.appendConvertedChunk?.(output);
|
|
controller.enqueue(sharedEncoder.encode(output));
|
|
}
|
|
|
|
// IMPORTANT: In passthrough mode we still must terminate the SSE stream.
|
|
// Some clients (e.g. OpenClaw) expect the OpenAI-style sentinel:
|
|
// data: [DONE]\n\n
|
|
// Without it they can hang until timeout and trigger failover.
|
|
// Gemini-family clients (Antigravity, Vertex, Gemini) reject this sentinel with 400 syntax errors.
|
|
const isGeminiFamily = provider === "antigravity" || provider === "gemini" || provider === "vertex";
|
|
if (!streamDoneSent && !isGeminiFamily) {
|
|
const doneOutput = "data: [DONE]\n\n";
|
|
reqLogger?.appendConvertedChunk?.(doneOutput);
|
|
controller.enqueue(sharedEncoder.encode(doneOutput));
|
|
}
|
|
|
|
finalizeStream();
|
|
return;
|
|
}
|
|
|
|
if (buffer.trim()) {
|
|
// Same parse as the transform loop: without targetFormat this only
|
|
// accepts "data: " lines, so an NDJSON provider (Ollama) lost whatever
|
|
// arrived without its closing newline.
|
|
const parsed = parseSSELine(buffer.trim(), targetFormat);
|
|
// parseSSELine turns the SSE sentinel "data: [DONE]" into { done: true },
|
|
// which must not be translated. An Ollama chunk also carries done:true,
|
|
// but it is the real final chunk — it holds finish_reason and the token
|
|
// counts — so it has to go through.
|
|
const isDoneSentinel = parsed?.done && targetFormat !== FORMATS.OLLAMA;
|
|
if (parsed && !isDoneSentinel) {
|
|
// Same accumulation the transform loop does, so finalizeStream() can
|
|
// log a tail chunk's tokens instead of falling back to null.
|
|
const extracted = extractUsage(parsed);
|
|
if (extracted) state.usage = mergeUsage(state.usage, extracted);
|
|
|
|
const translated = translateResponse(targetFormat, sourceFormat, parsed, state);
|
|
|
|
if (translated?._openaiIntermediate) {
|
|
for (const item of translated._openaiIntermediate) {
|
|
const openaiOutput = formatSSE(item, FORMATS.OPENAI);
|
|
reqLogger?.appendOpenAIChunk?.(openaiOutput);
|
|
}
|
|
}
|
|
|
|
if (translated?.length > 0) {
|
|
for (const item of translated) {
|
|
if (item === null || item === undefined) continue;
|
|
const output = formatSSE(item, sourceFormat);
|
|
reqLogger?.appendConvertedChunk?.(output);
|
|
controller.enqueue(sharedEncoder.encode(output));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
const flushed = translateResponse(targetFormat, sourceFormat, null, state);
|
|
|
|
if (flushed?._openaiIntermediate) {
|
|
for (const item of flushed._openaiIntermediate) {
|
|
const openaiOutput = formatSSE(item, FORMATS.OPENAI);
|
|
reqLogger?.appendOpenAIChunk?.(openaiOutput);
|
|
}
|
|
}
|
|
|
|
if (flushed?.length > 0) {
|
|
for (const item of flushed) {
|
|
if (item === null || item === undefined) continue;
|
|
const output = formatSSE(item, sourceFormat);
|
|
reqLogger?.appendConvertedChunk?.(output);
|
|
controller.enqueue(sharedEncoder.encode(output));
|
|
}
|
|
}
|
|
|
|
// Synthesize response.failed if a Responses passthrough stream never reached a terminal event
|
|
const keepsOpenAIResponsesFormat = targetFormat === FORMATS.OPENAI_RESPONSES && sourceFormat === FORMATS.OPENAI_RESPONSES;
|
|
if (keepsOpenAIResponsesFormat && !openAIResponsesTerminalSeen) {
|
|
const failedOutput = formatIncompleteOpenAIResponsesStreamFailure();
|
|
reqLogger?.appendConvertedChunk?.(failedOutput);
|
|
controller.enqueue(sharedEncoder.encode(failedOutput));
|
|
openAIResponsesTerminalSeen = true;
|
|
}
|
|
|
|
if (keepsOpenAIResponsesFormat && !openAIResponsesDoneSent && !streamDoneSent) {
|
|
const doneOutput = "data: [DONE]\n\n";
|
|
reqLogger?.appendConvertedChunk?.(doneOutput);
|
|
controller.enqueue(sharedEncoder.encode(doneOutput));
|
|
openAIResponsesDoneSent = true;
|
|
streamDoneSent = true;
|
|
}
|
|
|
|
finalizeStream();
|
|
} catch (error) {
|
|
console.log("Error in flush:", error);
|
|
finalizeStream();
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
export function createSSETransformStreamWithLogger(targetFormat, sourceFormat, provider = null, reqLogger = null, toolNameMap = null, model = null, connectionId = null, body = null, onStreamComplete = null, apiKey = null, customToolNames = null) {
|
|
return createSSEStream({
|
|
mode: STREAM_MODE.TRANSLATE,
|
|
targetFormat,
|
|
sourceFormat,
|
|
provider,
|
|
reqLogger,
|
|
toolNameMap,
|
|
customToolNames,
|
|
model,
|
|
connectionId,
|
|
body,
|
|
onStreamComplete,
|
|
apiKey
|
|
});
|
|
}
|
|
|
|
export function createPassthroughStreamWithLogger(provider = null, reqLogger = null, model = null, connectionId = null, body = null, onStreamComplete = null, apiKey = null) {
|
|
return createSSEStream({
|
|
mode: STREAM_MODE.PASSTHROUGH,
|
|
provider,
|
|
reqLogger,
|
|
model,
|
|
connectionId,
|
|
body,
|
|
onStreamComplete,
|
|
apiKey
|
|
});
|
|
}
|