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

Merge remote-tracking branch 'upstream/master'

Browse Source 
# Conflicts:
#	open-sse/handlers/chatCore.js
#	open-sse/providers/registry/index.js
#	open-sse/services/usage.js
This commit is contained in:
MUH. IQRAM BAHRING
2026-08-30 23:23:48 +08:00
parent 2d6673db61 90b52e06ff
commit eaf204b83c
114 changed files with 5587 additions and 2184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
open-sse/utils/stream.js
+54 -34
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;
}
@@ -353,16 +389,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
@@ -375,18 +401,26 @@ export function createSSEStream(options = {}) {
controller.enqueue(sharedEncoder.encode(doneOutput));
}
if (onStreamComplete) {
onStreamComplete({
content: accumulatedContent,
thinking: accumulatedThinking
}, usage, ttftAt);
}
finalizeStream();
return;
}
if (buffer.trim()) {
const parsed = parseSSELine(buffer.trim());
if (parsed && !parsed.done) {
// 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) {
@@ -442,24 +476,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();
}
}
});