luulam
59f17b3725
* Add /api/usage/request-details/raw endpoint serving a single stored request detail verbatim (raw payloads), with /raw doc clarifying it stays gated by the dashboard auth layer. * Add RawDetailModal opened from a new 'Raw' button in RequestDetailsTab. Modal loads /raw, exposes per-section copy buttons and a 'Copy all (JSON)' that bundles every section. * Capture the raw provider SSE text inside the streaming transform (cap 64KB) and forward it through onStreamComplete.rawProviderText so handler stores it as the providerResponse. response.content stays the extracted user text. Tool-call-only turns remain so the marker. * Accumulate from translated client-facing chunks instead of raw provider shapes so Responses, Claude delta types, and Gemini/Antigravity parts all contribute. * Drop redaction from the list endpoint; raw access is now via the dedicated /raw endpoint. Tests cover the new behavior.
561 lines
24 KiB
JavaScript
561 lines
24 KiB
JavaScript
import { translateResponse, initState } from "../translator/index.js";
|
|
import { FORMATS } from "../translator/formats.js";
|
|
import { trackPendingRequest } 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,
|
|
credentials = 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, sessionId: credentials?._clientSessionId || null }
|
|
: null;
|
|
|
|
let totalContentLength = 0;
|
|
let accumulatedContent = "";
|
|
let accumulatedThinking = "";
|
|
// Raw provider SSE text for "Copy all (JSON)" debugging. Kept separate
|
|
// from accumulatedContent (user-visible text) so Raw keeps the original
|
|
// upstream chunks even for formats we translate.
|
|
let rawProviderText = "";
|
|
const MAX_RAW_PROVIDER_CHARS = 64 * 1024;
|
|
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,
|
|
rawProviderText,
|
|
}, finalUsage, ttftAt);
|
|
}
|
|
};
|
|
// Accumulate user-visible text from a translated (client-facing) chunk.
|
|
// Translated chunks arrive in the client sourceFormat, which covers every
|
|
// provider path through its translator.
|
|
function accumulateStreamText(value, intoThinking) {
|
|
if (typeof value !== "string" || !value) return;
|
|
totalContentLength += value.length;
|
|
if (intoThinking) accumulatedThinking += value;
|
|
else accumulatedContent += value;
|
|
}
|
|
|
|
function accumulateTranslatedContent(item) {
|
|
if (!item || typeof item !== "object") return;
|
|
|
|
// OpenAI chat.completion.chunk shape
|
|
if (Array.isArray(item.choices)) {
|
|
for (const choice of item.choices) {
|
|
const delta = choice?.delta;
|
|
if (!delta || typeof delta !== "object") continue;
|
|
accumulateStreamText(delta.content, false);
|
|
accumulateStreamText(delta.reasoning_content ?? delta.reasoning, true);
|
|
}
|
|
return;
|
|
}
|
|
|
|
// Claude SSE shape: content_block_delta with text_delta/thinking_delta.
|
|
const claudeDelta = item.delta;
|
|
if (claudeDelta && typeof claudeDelta === "object") {
|
|
if (claudeDelta.type === "text_delta") accumulateStreamText(claudeDelta.text, false);
|
|
else if (claudeDelta.type === "thinking_delta") accumulateStreamText(claudeDelta.thinking, true);
|
|
else {
|
|
accumulateStreamText(claudeDelta.text, false);
|
|
accumulateStreamText(claudeDelta.thinking, true);
|
|
}
|
|
return;
|
|
}
|
|
|
|
// Gemini / Antigravity SSE shape.
|
|
const response = item.response || item;
|
|
const parts = response?.candidates?.[0]?.content?.parts;
|
|
if (Array.isArray(parts)) {
|
|
for (const part of parts) {
|
|
if (part?.thought === true) accumulateStreamText(part.text, true);
|
|
else accumulateStreamText(part?.text, false);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Accumulate straight from OpenAI Responses SSE events. Same-format
|
|
// passthrough skips translation, so it never reaches the helper above.
|
|
function accumulateResponsesEvent(eventName, parsed) {
|
|
if (!parsed || typeof parsed !== "object") return;
|
|
const data = parsed.data && typeof parsed.data === "object" ? parsed.data : parsed;
|
|
const type = eventName || parsed.type || data.type;
|
|
if (type === "response.output_text.delta") accumulateStreamText(data.delta, false);
|
|
else if (type === "response.reasoning_summary_text.delta") accumulateStreamText(data.delta, true);
|
|
else if (type === "response.output_text.done" && !accumulatedContent) accumulateStreamText(data.text, false);
|
|
}
|
|
|
|
function appendRawProviderText(current, text) {
|
|
if (!text) return current;
|
|
if (current.length >= MAX_RAW_PROVIDER_CHARS) return current;
|
|
const room = MAX_RAW_PROVIDER_CHARS - current.length;
|
|
return current + (text.length > room ? text.slice(0, room) : text);
|
|
}
|
|
|
|
|
|
return new TransformStream({
|
|
transform(chunk, controller) {
|
|
if (!ttftAt) ttftAt = Date.now();
|
|
const text = decoder.decode(chunk, { stream: true });
|
|
buffer += text;
|
|
rawProviderText = appendRawProviderText(rawProviderText, 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;
|
|
}
|
|
|
|
// Accumulate through the shared helper so OpenAI deltas and
|
|
// Gemini/Antigravity candidate parts are both covered.
|
|
accumulateTranslatedContent(parsed);
|
|
|
|
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;
|
|
}
|
|
|
|
// Text accumulation happens from the translated (client-facing) chunks
|
|
// below via accumulateTranslatedContent, plus accumulateResponsesEvent
|
|
// in the same-format Responses passthrough branch. Reading only the
|
|
// provider shape here missed formats like OpenAI Responses and left
|
|
// Raw request details as "[Empty streaming response]".
|
|
|
|
// 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) {
|
|
// Same-format Responses streams skip translation — accumulate here.
|
|
accumulateResponsesEvent(openAIResponsesEventName, parsed);
|
|
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;
|
|
// Accumulate what the client actually received — translated chunks
|
|
// cover every provider format via its translator.
|
|
accumulateTranslatedContent(item);
|
|
// 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;
|
|
// Buffer-remainder chunk may still carry text.
|
|
accumulateTranslatedContent(item);
|
|
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;
|
|
// Include flush-synthesized chunks (usually finish/usage only).
|
|
accumulateTranslatedContent(item);
|
|
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, credentials = null) {
|
|
return createSSEStream({
|
|
mode: STREAM_MODE.TRANSLATE,
|
|
targetFormat,
|
|
sourceFormat,
|
|
provider,
|
|
reqLogger,
|
|
toolNameMap,
|
|
customToolNames,
|
|
model,
|
|
connectionId,
|
|
body,
|
|
onStreamComplete,
|
|
apiKey,
|
|
credentials
|
|
});
|
|
}
|
|
|
|
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
|
|
});
|
|
}
|