luulam/9router
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(stream): record usage when a client closes on the terminal event

Browse Source 
The Responses API has no [DONE] sentinel, so codex closes the socket
as soon as response.completed arrives. That cancels the reader before
flush() runs — and flush() held every usage side effect, so a fully
successful request logged nothing: no 📊 done line, no token stats,
no request detail.

Extract that tail into a once-guarded finalizeStream() and also call
it right after the terminal event is forwarded, in both passthrough
and translate mode. flush() still calls it; the guard makes the
second call a no-op. Streams that end normally are unaffected, and a
terminal event carrying no usage falls through to the existing
estimate/null path rather than blocking.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
decolua
2026-08-27 18:52:42 +07:00
parent 9c45b27cd7
commit d7f7d70dd5
1 changed files with 39 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
open-sse/utils/stream.js
View File

@@ -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();
}
}
});