diff --git a/open-sse/utils/stream.js b/open-sse/utils/stream.js index 33e7fb04..9573d4c3 100644 --- a/open-sse/utils/stream.js +++ b/open-sse/utils/stream.js @@ -75,6 +75,35 @@ export function createSSEStream(options = {}) { 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) { @@ -105,6 +134,7 @@ export function createSSEStream(options = {}) { if (mode === STREAM_MODE.PASSTHROUGH) { let output; let injectedUsage = false; + let responsesTerminal = false; if (trimmed.startsWith("data:") && trimmed.slice(5).trim() !== "[DONE]") { try { @@ -168,6 +198,8 @@ export function createSSEStream(options = {}) { 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); @@ -202,6 +234,8 @@ export function createSSEStream(options = {}) { reqLogger?.appendConvertedChunk?.(output); controller.enqueue(sharedEncoder.encode(output)); + // Responses clients (codex CLI) close on response.completed instead of [DONE] + if (responsesTerminal) finalizeStream(); continue; } @@ -292,6 +326,8 @@ export function createSSEStream(options = {}) { controller.enqueue(sharedEncoder.encode(output)); currentOpenAIResponsesEvent = null; sseEmittedCount++; + // Responses clients (codex) close on response.completed instead of [DONE] + if (openAIResponsesTerminalSeen) finalizeStream(); continue; } @@ -355,16 +391,6 @@ export function createSSEStream(options = {}) { controller.enqueue(sharedEncoder.encode(output)); } - if (!hasValidUsage(usage) && totalContentLength > 0) { - usage = estimateUsage(body, totalContentLength, FORMATS.OPENAI); - } - - if (hasValidUsage(usage)) { - logUsage(provider, usage, model, connectionId, apiKey); - } else { - appendRequestLog({ model, provider, connectionId, tokens: null, status: "200 OK" }).catch(() => { }); - } - // 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 @@ -377,12 +403,7 @@ export function createSSEStream(options = {}) { controller.enqueue(sharedEncoder.encode(doneOutput)); } - if (onStreamComplete) { - onStreamComplete({ - content: accumulatedContent, - thinking: accumulatedThinking - }, usage, ttftAt); - } + finalizeStream(); return; } @@ -444,24 +465,10 @@ export function createSSEStream(options = {}) { streamDoneSent = true; } - if (!hasValidUsage(state?.usage) && totalContentLength > 0) { - state.usage = estimateUsage(body, totalContentLength, sourceFormat); - } - - if (hasValidUsage(state?.usage)) { - logUsage(state.provider || targetFormat, state.usage, model, connectionId, apiKey); - } else { - appendRequestLog({ model, provider, connectionId, tokens: null, status: "200 OK" }).catch(() => { }); - } - - if (onStreamComplete) { - onStreamComplete({ - content: accumulatedContent, - thinking: accumulatedThinking - }, state?.usage, ttftAt); - } + finalizeStream(); } catch (error) { console.log("Error in flush:", error); + finalizeStream(); } } });