pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-08-03 05:45:04 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(streaming): harden long OpenAI-compatible SSE streams (#5124)

Browse Source 
Integrated into release/v3.8.38 (rebased on tip; streamHandler conflict with #5007 disconnect-guard resolved — both coexist, stream-handler 22/22 green)
This commit is contained in:
Randi
2026-06-26 21:27:11 -04:00
committed by GitHub
parent 843680f6ca
commit 9e2c2817da
21 changed files with 585 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGELOG.md
View File

@@ -24,6 +24,7 @@ _In development — bullets added per PR; finalized at release._
### 🔧 Bug Fixes
- **fix(streaming): harden long OpenAI-compatible SSE streams** — a late pipeline-wind-down error can no longer overwrite an already-recorded successful stream (`streamCompletionRecorded` guard), client disconnects finalize as `499 client_disconnected` instead of poisoning provider/account failure state, JSON bodies that are actually SSE (wrong `application/json` content-type) are sniffed and re-streamed, and reasoning fields (`reasoning`/`reasoning_content` + OpenRouter/Gemini encrypted `reasoning_details`) are preserved through the JSON-as-SSE fallback. ([#5124](https://github.com/diegosouzapw/OmniRoute/pull/5124) — thanks @rdself)
- **fix(usage): dedupe request-usage logging and debounce stats events** — `saveRequestUsage` now guards against duplicate inserts (natural key: timestamp + provider + model + connection + api-key + token counts), back-fills a missing `endpoint`, and only emits `usageRecorded` when a row was actually inserted; stats `update`/`pending` event bursts are collapsed into a single debounced notification to reduce churn. ([#4940](https://github.com/diegosouzapw/OmniRoute/pull/4940) — thanks @nguyenxvotanminh3)
- **fix(sse): convert the native Gemini request body to OpenAI format in the Antigravity MITM handler** — `contents` / `systemInstruction` / `generationConfig` / `thinkingConfig` are now translated to OpenAI chat-completions format before forwarding to `/v1/chat/completions`, so thinking-capable models (e.g. `ag/claude-opus-4-6-thinking`) no longer fail with provider-side 400 "invalid argument" errors. ([#4845](https://github.com/diegosouzapw/OmniRoute/pull/4845) — thanks @anuragg-saxenaa)
- **fix(db): translate the two pt-BR SQLite driver-fallback log lines to English** — `[DB] Pré-inicializando sql.js WASM…` and `[DB] Drivers síncronos indisponíveis…` were the only non-English server log strings, mixing languages in the logs. Now `[DB] Pre-initializing sql.js WASM (synchronous drivers unavailable)…` / `[DB] Synchronous drivers unavailable — falling back to sql.js (WASM)`, guarded by a test that scans the driver path for accented log strings. ([#5103](https://github.com/diegosouzapw/OmniRoute/issues/5103))

3
config/quality/complexity-baseline.json
View File

@@ -1,6 +1,7 @@
{
"_comment": "Catraca de complexidade (check-complexity.mjs, ESLint core rules complexity>=15 e max-lines-per-function>80 sobre src+open-sse+electron+bin via eslint.complexity.config.mjs). Conta total de violacoes; so pode cair. --update ratcheta.",
"count": 1963,
"count": 1972,
"_rebaseline_2026_06_26_v3838_release_fast_gate": "1963->1972 (+9). Reconciles inherited release/v3.8.38 drift surfaced by PR #5124. Local origin/release/v3.8.38 measured 1971 and this PR also measured 1971 after refactoring the new JSON-body SSE sniffing path, while the GitHub Ubuntu fast gate measured 1972; use the CI-observed value so the release branch gate is deterministic. The streaming fix is complexity-net-zero relative to the local release base. The release fast-path does not consistently ratchet complexity between release-cycle merges; keep structural reductions as separate debt.",
"_rebaseline_2026_06_25_v3836_release": "Reconciliacao release-volatil 1920->1950 (+30) no fechamento do ciclo v3.8.36, surfada pelo CI da fix-PR #5029 (a catraca de complexidade NAO roda no fast-path PR->release nem foi medida no release PR #4854 — Quality Ratchet foi SKIPPED la — so PR->main, entao o ramo acumulou os 137 commits sem rebaselinar). O +30 e drift de condicionais NOVOS das features legitimas do ciclo: Quota-Share Fase 2/3 (estrategia dedicada DRR+P2C, multi-window buckets, concurrency control, headroom, saturacao proativa — #4885/#4907/#4908/#4927/#4928/#4929/#4939/#4965/#4967/#4970), task-aware + Fusion combo (#4945/#4652), e ramos de provider/translator de contribuidores. A god-file decomposition #3501 e PURA (move codigo p/ leaves, complexity-neutra). Verificado que esta fix-PR (#5029) toca SO scripts/build/pack-artifact-policy.ts (array de strings), tests/integration/resilience-http-e2e.test.ts (2 keys) e os 2 baselines json — contribui 0 ao gate que varre src+open-sse+electron+bin. Mesma familia dos rebaselines anteriores — crescimento de feature legitimo recem-TDD'd, nao regressao; reducao estrutural fica como debt (#3501).",
"_rebaseline_2026_06_23_v3835_release": "Reconciliacao release-volatil 1916->1920 (+4) no fechamento do ciclo v3.8.35, surfada pelo pre-flight check:release-green (a catraca de complexidade NAO roda no fast-path PR->release, so release->main, entao o ramo acumula sem rebaselinar). O +4 e drift de condicionais NOVOS dos merges de contribuidor/feature deste ciclo (Compression Phase 4 #4694/#4707/#4716/#4720, combos auto-promote #4774, tier no-auth #4753, deepseek-web tool-fold #4756, dedupe provider nodes #4768). Verificado que o trabalho de release-finalize desta sessao toca SO docs/*.md (THREAT_MODEL), CHANGELOG.md, baselines e 1 linha de string em scripts/check/check-fabricated-docs.mjs (fora do escopo src+open-sse+electron+bin que o gate varre) — contribui 0. Mesma familia dos rebaselines anteriores — crescimento de feature legitimo recem-TDD'd, nao regressao; reducao estrutural fica como debt (#3501).",
"_rebaseline_2026_06_23_v3834_release": "Reconciliacao release-volatil 1915->1916 (+1) no fechamento do ciclo v3.8.34. check:complexity NAO roda no fast-path PR->release (so release->main), entao o ramo acumula sem rebaselinar; surfou no full CI do release PR (run em c98e7ff6d). O +1 e drift de condicional NOVO de merge de contribuidor do ciclo (features quota/usage/opencode-go/M365). Verificado que o commit de release-finalize NAO adiciona complexity: toca CHANGELOG/baseline/mirrors/3 testes + 1 linha de regex em opencodeOllamaUsage.ts (sem novo ramo) + reorder de dados no reka registry — local mede 1916 com ou sem essa mudanca. Mesma familia dos rebaselines anteriores — crescimento de feature legitimo, nao regressao; reducao estrutural fica como debt (#3501).",

5
config/quality/quality-baseline.json
View File

@@ -102,13 +102,14 @@
"_rebaseline_2026_06_26_v3837_release": "343->345. v3.8.37 cycle drift surfaced by the release-green pre-flight (the Quality Ratchet does NOT run on PR->release fast-gates, so warnings/complexity accrued unmeasured across this cycle's 76 commits — provider adds DGrid/Pioneer/xAI, headroom proxy lifecycle #4649, ~50 SSE/translator fixes, Engine Combos #5062). Trust-but-verify: this release-finalize working tree touches ONLY CHANGELOG.md, docs/i18n/*/CHANGELOG.md mirrors, and these baselines — 0 production-code change, so all drift is inherited cycle drift (`any` warn-allowed in open-sse/ + tests/). Tighten via --require-tighten next cycle."
},
"cognitiveComplexity": {
"value": 826,
"value": 833,
"direction": "down",
"dedicatedGate": true,
"_rebaseline_2026_06_25_v3836_release": "801→816 (+15) — v3.8.36 cycle drift surfaced by the fix-PR #5029 CI (cognitive-complexity does NOT run on PR→release fast-gates, and the Quality Ratchet job was SKIPPED on the release PR #4854 itself, so the +15 from this cycle's 137 commits accrued unmeasured). Measured locally = 816 (identical to the CI Cognitive complexity ratchet). Drift from legit cycle features (Quota-Share Fase 2/3, task-aware/Fusion combo, contributor provider/translator branches); god-file decomposition #3501 is complexity-neutral. This fix-PR touches only pack-artifact-policy.ts + a test + the 2 baseline JSONs — contributes 0. Structural shrink tracked in #3501.",
"_rebaseline_2026_06_22_v3834_release": "797→801 (+4) — v3.8.34 cycle drift surfaced by the release-green pre-flight (cognitive-complexity does NOT run on PR→release fast-gates). Verified my release-finalize working tree touches ONLY CHANGELOG.md (git status: 0 code changes), so all +4 is inherited contributor drift from this cycle's parallel-session merges. Structural shrink tracked in #3501.",
"_rebaseline_2026_06_22_v3833_release": "793→797 (+4) — pre-existing cycle drift on origin/release/v3.8.33.",
"_rebaseline_2026_06_26_v3837_release": "816->826. v3.8.37 cycle drift surfaced by the release-green pre-flight (the Quality Ratchet does NOT run on PR->release fast-gates, so warnings/complexity accrued unmeasured across this cycle's 76 commits — provider adds DGrid/Pioneer/xAI, headroom proxy lifecycle #4649, ~50 SSE/translator fixes, Engine Combos #5062). Trust-but-verify: this release-finalize working tree touches ONLY CHANGELOG.md, docs/i18n/*/CHANGELOG.md mirrors, and these baselines — 0 production-code change, so all drift is inherited cycle drift (`any` warn-allowed in open-sse/ + tests/). Tighten via --require-tighten next cycle."
"_rebaseline_2026_06_26_v3837_release": "816->826. v3.8.37 cycle drift surfaced by the release-green pre-flight (the Quality Ratchet does NOT run on PR->release fast-gates, so warnings/complexity accrued unmeasured across this cycle's 76 commits — provider adds DGrid/Pioneer/xAI, headroom proxy lifecycle #4649, ~50 SSE/translator fixes, Engine Combos #5062). Trust-but-verify: this release-finalize working tree touches ONLY CHANGELOG.md, docs/i18n/*/CHANGELOG.md mirrors, and these baselines — 0 production-code change, so all drift is inherited cycle drift (`any` warn-allowed in open-sse/ + tests/). Tighten via --require-tighten next cycle.",
"_rebaseline_2026_06_26_v3838_release": "826->833. v3.8.38 release base measures 833 locally on origin/release/v3.8.38 (800b04ad6) while the committed baseline still says 826. This PR measures the same 833 after refactoring jsonToSse helpers back under the sonarjs/cognitive-complexity threshold, so it does not add a net cognitive-complexity violation. The baseline bump records inherited release-base drift only; structural shrink remains tracked by the existing chatCore decomposition work."
},
"typeCoveragePct": {
"value": 92.17,

46
open-sse/handlers/chatCore.ts
View File

@@ -405,7 +405,11 @@ export async function handleChatCore({
// only when this function is called outside the normal chat dispatch.
{
const _s = cachedSettings ?? (await getCachedSettings());
if (_s.customSystemPromptEnabled === true && typeof _s.customSystemPrompt === "string" && _s.customSystemPrompt) {
if (
_s.customSystemPromptEnabled === true &&
typeof _s.customSystemPrompt === "string" &&
_s.customSystemPrompt
) {
body = injectCustomSystemPrompt(body as Record<string, unknown>, _s.customSystemPrompt);
log?.debug?.("CUSTOMSP", "custom system prompt injected");
}
@@ -1804,8 +1808,8 @@ export async function handleChatCore({
// `_toolNameMap` so kiro-to-openai maps streamed tool-call names back (#1375).
if (targetFormat === FORMATS.KIRO) {
const kiroTools =
translatedBody?.conversationState?.currentMessage?.userInputMessage
?.userInputMessageContext?.tools;
translatedBody?.conversationState?.currentMessage?.userInputMessage?.userInputMessageContext
?.tools;
if (kiroTools) {
const { tools: sanitizedKiroTools, nameMap: kiroNameMap } = sanitizeKiroTools(kiroTools);
translatedBody.conversationState.currentMessage.userInputMessage.userInputMessageContext.tools =
@@ -2060,15 +2064,38 @@ export async function handleChatCore({
});
let onPipelineStreamError: streamFailure.PipelineStreamErrorHandler | null = null;
let onClientDisconnectFinalize:
| ((event: { reason: string; duration: number }) => boolean)
| null = null;
// Create stream controller for disconnect detection
const streamController = createStreamController({
onDisconnect,
onDisconnect: (event) => {
let finalized = false;
try {
finalized = onClientDisconnectFinalize?.(event) === true;
} catch {}
if (!finalized) {
try {
finalizePendingScope(pendingScope, {
status: 499,
error: `Client disconnected: ${event.reason}`,
errorCode: "client_disconnected",
});
finalized = true;
} catch {}
}
try {
onDisconnect?.(event);
} catch {}
return finalized;
},
onError: (event) => onPipelineStreamError?.(event),
provider,
model,
connectionId,
clientResponseFormat,
clientAbortSignal: clientRawRequest?.signal,
});
const dedupRequestBody = { ...translatedBody, model: `${provider}/${model}`, stream };
@@ -3893,6 +3920,7 @@ export async function handleChatCore({
(finalBody as Record<string, unknown> | null | undefined) ?? null
);
let streamCompletionRecorded = false;
let streamFailureCompletionRecorded = false;
// Callback to save call log when stream completes (include responseBody when provided by stream)
@@ -3907,6 +3935,8 @@ export async function handleChatCore({
ttft,
}) => {
const normalizedStreamStatus = streamStatus || 200;
if (streamCompletionRecorded) return;
streamCompletionRecorded = true;
if (normalizedStreamStatus !== 200) {
if (streamFailureCompletionRecorded) return;
streamFailureCompletionRecorded = true;
@@ -4046,12 +4076,20 @@ export async function handleChatCore({
const streamFailureFinalizers = streamFailure.createStreamFailureFinalizers({
isFailureCompletionRecorded: () => streamFailureCompletionRecorded,
isStreamCompletionRecorded: () => streamCompletionRecorded,
onStreamComplete,
persistFailureUsage,
onStreamFailure,
});
const handleStreamFailure = streamFailureFinalizers.handleStreamFailure;
onPipelineStreamError = streamFailureFinalizers.onPipelineStreamError;
onClientDisconnectFinalize = (event) =>
handleStreamFailure({
status: 499,
message: `Client disconnected: ${event.reason}`,
code: "client_disconnected",
type: "client_disconnected",
});
// For providers using Responses API format, translate stream back to openai (Chat Completions) format
// UNLESS client is Droid CLI which expects openai-responses format back

95
open-sse/handlers/chatCore/jsonBodyToSse.ts
View File

@@ -28,6 +28,95 @@ const DEFAULT_DEPS: JsonBodyToSseDeps = {
synthesizeOpenAiSseFromJson: defaultSynthesize,
};
function prependBufferedChunks(
chunks: Uint8Array[],
reader: ReadableStreamDefaultReader<Uint8Array>
): ReadableStream<Uint8Array> {
let index = 0;
return new ReadableStream<Uint8Array>({
async pull(controller) {
if (index < chunks.length) {
controller.enqueue(chunks[index++]);
return;
}
try {
const { done, value } = await reader.read();
if (done) {
controller.close();
} else {
controller.enqueue(value);
}
} catch (error) {
controller.error(error);
}
},
async cancel(reason) {
try {
await reader.cancel(reason);
} catch {}
},
});
}
function classifyBodyPrefix(text: string): "sse" | "non-sse" | "unknown" {
const trimmed = text.replace(/^\uFEFF/, "").trimStart();
if (!trimmed) return "unknown";
if (trimmed.startsWith(":")) return "sse";
if (/^(?:data|event|id|retry)\s*:/i.test(trimmed)) return "sse";
const lower = trimmed.toLowerCase();
for (const field of ["data", "event", "id", "retry"]) {
if (field.startsWith(lower)) return "unknown";
if (lower.startsWith(field)) {
const rest = lower.slice(field.length);
if (/^\s*$/.test(rest)) return "unknown";
if (/^\s*:/.test(rest)) return "sse";
}
}
return "non-sse";
}
async function sniffJsonBodyForSse(
providerResponse: Response,
ctx: { log?: LoggerLike; provider: string | null | undefined; model: string | null | undefined },
deps: JsonBodyToSseDeps
): Promise<{ sseResponse?: Response; jsonBody: Response }> {
const reader = providerResponse.body!.getReader();
const bufferedChunks: Uint8Array[] = [];
const decoder = new TextDecoder();
let sniffed = "";
let sniffedBytes = 0;
const maxSniffBytes = 4096;
while (sniffedBytes < maxSniffBytes) {
const chunk = await deps.withBodyTimeout<ReadableStreamReadResult<Uint8Array>>(reader.read());
if (chunk.done || !chunk.value) break;
bufferedChunks.push(chunk.value);
sniffedBytes += chunk.value.byteLength;
sniffed += decoder.decode(chunk.value, { stream: true });
if (classifyBodyPrefix(sniffed) === "sse") {
const rebuiltHeaders = new Headers(providerResponse.headers);
rebuiltHeaders.delete("content-length");
rebuiltHeaders.set("content-type", "text/event-stream");
ctx.log?.debug?.(
"STREAM",
`Upstream returned SSE bytes with application/json content-type — preserving streaming body (${ctx.provider}/${ctx.model})`
);
return {
sseResponse: new Response(prependBufferedChunks(bufferedChunks, reader), {
status: providerResponse.status,
statusText: providerResponse.statusText,
headers: rebuiltHeaders,
}),
jsonBody: new Response(null),
};
}
}
return { jsonBody: new Response(prependBufferedChunks(bufferedChunks, reader)) };
}
export async function maybeConvertJsonBodyToSse(
providerResponse: Response,
ctx: { log?: LoggerLike; provider: string | null | undefined; model: string | null | undefined },
@@ -42,7 +131,11 @@ export async function maybeConvertJsonBodyToSse(
if (!isNonSseJsonBody) {
return providerResponse;
}
const jsonText = await deps.withBodyTimeout<string>(providerResponse.text());
const { sseResponse, jsonBody } = await sniffJsonBodyForSse(providerResponse, ctx, deps);
if (sseResponse) return sseResponse;
const jsonText = await deps.withBodyTimeout<string>(jsonBody.text());
const synthesizedSse = deps.synthesizeOpenAiSseFromJson(jsonText);
const rebuiltHeaders = new Headers(providerResponse.headers);
rebuiltHeaders.delete("content-length");

29
open-sse/handlers/responseSanitizer.ts
View File

@@ -1,3 +1,5 @@
import { copyOpenAICompatibleReasoningFields } from "../utils/reasoningFields.ts";
/**
* Response Sanitizer — Normalizes LLM responses to strict OpenAI SDK format.
*
@@ -1072,32 +1074,7 @@ export function sanitizeStreamingChunk(parsed: unknown): unknown {
? collapseExcessiveNewlines(deltaRecord.content)
: deltaRecord.content;
}
if (deltaRecord.reasoning_content !== undefined) {
delta.reasoning_content = deltaRecord.reasoning_content;
}
if (deltaRecord.reasoning_text !== undefined) {
delta.reasoning_text = deltaRecord.reasoning_text;
} else if (typeof deltaRecord.reasoning === "string" && deltaRecord.reasoning) {
// Alias: some providers use 'reasoning' instead of 'reasoning_content'
delta.reasoning_content = deltaRecord.reasoning;
} else if (Array.isArray(deltaRecord.reasoning_details)) {
// StepFun/OpenRouter: reasoning_details[{type:"reasoning.text", text:"..."}]
const parts: string[] = [];
for (const detail of deltaRecord.reasoning_details) {
const d = detail && typeof detail === "object" ? (detail as JsonRecord) : null;
if (!d) continue;
const text =
typeof d.text === "string"
? d.text
: typeof d.content === "string"
? d.content
: "";
if (text) parts.push(text);
}
if (parts.length > 0) {
delta.reasoning_content = parts.join("");
}
}
copyOpenAICompatibleReasoningFields(deltaRecord, delta);
if (deltaRecord.tool_calls !== undefined) {
delta.tool_calls = Array.isArray(deltaRecord.tool_calls)
? deltaRecord.tool_calls.map((tc) => {

50
open-sse/utils/jsonToSse.ts
View File

@@ -13,6 +13,7 @@
* Returns "" when the text is not a parseable chat-completion object with at
* least one choice — callers then fall back to the original (error) handling.
*/
import { getUnsupportedReasoningValue } from "./reasoningFields.ts";
type JsonRecord = Record<string, unknown>;
@@ -20,6 +21,46 @@ function isRecord(value: unknown): value is JsonRecord {
return typeof value === "object" && value !== null && !Array.isArray(value);
}
function nonEmptyString(value: unknown): string {
return typeof value === "string" && value.length > 0 ? value : "";
}
function addReadableReasoning(message: JsonRecord, delta: JsonRecord): boolean {
const reasoningContent = nonEmptyString(message.reasoning_content);
if (reasoningContent) {
delta.reasoning_content = reasoningContent;
return true;
}
const reasoning = nonEmptyString(message.reasoning);
if (reasoning) {
delta.reasoning = reasoning;
return true;
}
return false;
}
function addUnsupportedReasoning(message: JsonRecord, delta: JsonRecord) {
const reasoningContent = getUnsupportedReasoningValue(message);
if (reasoningContent) {
delta.reasoning_content = reasoningContent;
}
}
function buildReasoningDelta(message: JsonRecord): JsonRecord | null {
const delta: JsonRecord = {};
if (Array.isArray(message.reasoning_details)) {
delta.reasoning_details = message.reasoning_details;
}
if (!addReadableReasoning(message, delta)) {
addUnsupportedReasoning(message, delta);
}
return Object.keys(delta).length > 0 ? delta : null;
}
function sseEvent(payload: JsonRecord): string {
return `data: ${JSON.stringify(payload)}\n\n`;
}
@@ -60,8 +101,9 @@ export function synthesizeOpenAiSseFromJson(jsonText: string): string {
};
emitDelta({ role });
if (typeof message.reasoning_content === "string" && message.reasoning_content.length > 0) {
emitDelta({ reasoning_content: message.reasoning_content });
const reasoningDelta = buildReasoningDelta(message);
if (reasoningDelta) {
emitDelta(reasoningDelta);
}
if (typeof message.content === "string" && message.content.length > 0) {
emitDelta({ content: message.content });
@@ -71,7 +113,9 @@ export function synthesizeOpenAiSseFromJson(jsonText: string): string {
}
const finishReason =
typeof choice.finish_reason === "string" && choice.finish_reason ? choice.finish_reason : "stop";
typeof choice.finish_reason === "string" && choice.finish_reason
? choice.finish_reason
: "stop";
const finalChoice: JsonRecord = { index, delta: {}, finish_reason: finishReason };
const finalChunk: JsonRecord = { ...base, choices: [finalChoice] };
if (isRecord(parsed.usage)) finalChunk.usage = parsed.usage;

8
open-sse/utils/passthroughTailProcessor.ts
View File

@@ -7,6 +7,7 @@ import {
stringifyIdValue,
stripResponsesLifecycleEcho,
} from "./responsesStreamHelpers.ts";
import { getAnyReasoningValue } from "./reasoningFields.ts";
type JsonRecord = Record<string, unknown>;
@@ -200,12 +201,7 @@ function handleOpenAiTailPayload(parsed: JsonRecord, context: PassthroughTailPro
context.appendPassthroughContent(delta.content);
context.addTotalContentLength(delta.content.length);
}
const reasoningDelta =
typeof delta.reasoning_content === "string"
? delta.reasoning_content
: typeof delta.reasoning === "string"
? delta.reasoning
: "";
const reasoningDelta = getAnyReasoningValue(delta);
if (reasoningDelta) {
context.appendPassthroughReasoning(reasoningDelta);
}

72
open-sse/utils/reasoningFields.ts Normal file
View File

@@ -0,0 +1,72 @@
type JsonRecord = Record<string, unknown>;
export function asReasoningRecord(value: unknown): JsonRecord {
return value && typeof value === "object" && !Array.isArray(value) ? (value as JsonRecord) : {};
}
function nonEmptyString(value: unknown): string {
return typeof value === "string" && value.length > 0 ? value : "";
}
export function extractReasoningDetailsText(value: unknown): string {
const record = asReasoningRecord(value);
if (!Array.isArray(record.reasoning_details)) return "";
return record.reasoning_details
.map((detail) => {
const item = asReasoningRecord(detail);
return nonEmptyString(item.text) || nonEmptyString(item.content);
})
.join("");
}
export function getReadableReasoningValue(value: unknown): string {
const record = asReasoningRecord(value);
return nonEmptyString(record.reasoning_content) || nonEmptyString(record.reasoning);
}
export function getUnsupportedReasoningValue(value: unknown): string {
const record = asReasoningRecord(value);
return (
nonEmptyString(record.reasoning_text) ||
nonEmptyString(record.thinking) ||
nonEmptyString(record.thought) ||
extractReasoningDetailsText(record)
);
}
export function getAnyReasoningValue(value: unknown): string {
return getReadableReasoningValue(value) || getUnsupportedReasoningValue(value);
}
export function hasUnsupportedReasoningSignal(value: unknown): boolean {
const record = asReasoningRecord(value);
return Boolean(
!getReadableReasoningValue(record) &&
(nonEmptyString(record.reasoning_text) ||
nonEmptyString(record.thinking) ||
nonEmptyString(record.thought) ||
(Array.isArray(record.reasoning_details) && record.reasoning_details.length > 0))
);
}
export function hasAnyReasoningSignal(value: unknown): boolean {
const record = asReasoningRecord(value);
return Boolean(
getReadableReasoningValue(record) ||
nonEmptyString(record.reasoning_text) ||
nonEmptyString(record.thinking) ||
nonEmptyString(record.thought) ||
(Array.isArray(record.reasoning_details) && record.reasoning_details.length > 0)
);
}
export function copyOpenAICompatibleReasoningFields(source: JsonRecord, target: JsonRecord) {
if (source.reasoning_content !== undefined) target.reasoning_content = source.reasoning_content;
if (source.reasoning !== undefined) target.reasoning = source.reasoning;
if (source.reasoning_text !== undefined) target.reasoning_text = source.reasoning_text;
if (Array.isArray(source.reasoning_details)) target.reasoning_details = source.reasoning_details;
if (!getReadableReasoningValue(target)) {
const mirrored = getUnsupportedReasoningValue(source);
if (mirrored) target.reasoning_content = mirrored;
}
}

56
open-sse/utils/stream.ts
View File

@@ -47,6 +47,12 @@ import {
stripResponsesLifecycleEcho,
} from "./responsesStreamHelpers.ts";
import { processBufferedPassthroughLine } from "./passthroughTailProcessor.ts";
import {
getAnyReasoningValue,
getReadableReasoningValue,
getUnsupportedReasoningValue,
hasUnsupportedReasoningSignal,
} from "./reasoningFields.ts";
/**
* Race a response body read against a timeout.
@@ -1592,11 +1598,8 @@ export function createSSEStream(options: StreamOptions = {}) {
: false;
const hadNonStringTopLevelId =
parsed?.id != null && typeof parsed.id !== "string";
const hadReasoningAlias = !!(
parsed.choices?.[0]?.delta?.reasoning &&
typeof parsed.choices[0].delta.reasoning === "string" &&
!parsed.choices[0].delta.reasoning_content
);
const rawDelta = parsed.choices?.[0]?.delta;
const hadReasoningAlias = hasUnsupportedReasoningSignal(rawDelta);
parsed = sanitizeStreamingChunk(parsed);
if (
@@ -1699,7 +1702,7 @@ export function createSSEStream(options: StreamOptions = {}) {
}
}
const content = delta?.content || delta?.reasoning_content;
const content = delta?.content;
if (typeof content === "string") {
totalContentLength += content.length;
@@ -1720,6 +1723,10 @@ export function createSSEStream(options: StreamOptions = {}) {
}
}
}
const reasoningDelta = getReadableReasoningValue(delta);
if (reasoningDelta) {
totalContentLength += reasoningDelta.length;
}
{
const guarded = applyTextualToolCallStreamingGuard(
parsed as Record<string, unknown>
@@ -1727,10 +1734,10 @@ export function createSSEStream(options: StreamOptions = {}) {
parsed = guarded.parsed as typeof parsed;
textualToolCallConverted = guarded.textualToolCallConverted;
}
if (typeof delta?.reasoning_content === "string")
if (reasoningDelta)
passthroughAccumulatedReasoning = appendBoundedText(
passthroughAccumulatedReasoning,
delta.reasoning_content
reasoningDelta
);
const extracted = extractUsage(parsed);
@@ -1899,23 +1906,24 @@ export function createSSEStream(options: StreamOptions = {}) {
}
}
}
if (parsed.choices?.[0]?.delta?.reasoning_content) {
const r = parsed.choices[0].delta.reasoning_content;
if (typeof r === "string") {
totalContentLength += r.length;
if (state?.accumulatedContent !== undefined)
state.accumulatedContent = appendBoundedText(state.accumulatedContent, r);
}
const openAiDelta = parsed.choices?.[0]?.delta;
const openAiReasoning = getReadableReasoningValue(openAiDelta);
if (openAiReasoning) {
totalContentLength += openAiReasoning.length;
if (state?.accumulatedContent !== undefined)
state.accumulatedContent = appendBoundedText(
state.accumulatedContent,
openAiReasoning
);
}
// Normalize `reasoning` alias → `reasoning_content` (NVIDIA kimi-k2.5 etc.)
if (
parsed.choices?.[0]?.delta?.reasoning &&
!parsed.choices?.[0]?.delta?.reasoning_content
) {
const r = parsed.choices[0].delta.reasoning;
if (typeof r === "string") {
// Mirror only client-unsupported reasoning aliases into `reasoning_content`.
if (!openAiReasoning) {
const delta = openAiDelta;
const r = getUnsupportedReasoningValue(delta);
if (typeof r === "string" && r.length > 0) {
parsed.choices[0].delta.reasoning_content = r;
delete parsed.choices[0].delta.reasoning;
delete parsed.choices[0].delta.thinking;
delete parsed.choices[0].delta.thought;
totalContentLength += r.length;
if (state?.accumulatedContent !== undefined)
state.accumulatedContent = appendBoundedText(state.accumulatedContent, r);
@@ -1962,7 +1970,7 @@ export function createSSEStream(options: StreamOptions = {}) {
const translateHasContent =
typeof parsed.delta?.text === "string" ||
typeof parsed.choices?.[0]?.delta?.content === "string" ||
typeof parsed.choices?.[0]?.delta?.reasoning_content === "string";
Boolean(getAnyReasoningValue(parsed.choices?.[0]?.delta));
if (translateHasContent && !contentAfterToolSeen) {
const toolTs = toolFinishTime || pendingToolFinishTime;
const lastChunkTs = lastToolCallChunkTime;

37
open-sse/utils/streamFailureFinalization.ts
View File

@@ -87,16 +87,22 @@ export function finalizeStreamRequestLog({
export function createStreamFailureFinalizers({
isFailureCompletionRecorded,
isStreamCompletionRecorded = () => false,
onStreamComplete,
persistFailureUsage,
onStreamFailure,
}: {
isFailureCompletionRecorded: () => boolean;
isStreamCompletionRecorded?: () => boolean;
onStreamComplete: (payload: StreamCompletionPayload) => void;
persistFailureUsage: (status: number, errorCode?: string) => void;
onStreamFailure?: ((failure: StreamFailurePayload) => void) | null;
}) {
const handleStreamFailure = (failure: StreamFailurePayload) => {
if (isStreamCompletionRecorded()) {
return true;
}
const status = failure.status || HTTP_STATUS.BAD_GATEWAY;
const message = failure.message || "Upstream stream error";
const code = failure.code || failure.type || String(status);
@@ -124,25 +130,42 @@ export function createStreamFailureFinalizers({
return true;
};
const isClientClosedPipelineError = (message: string, statusCode: number) => {
const normalized = message.toLowerCase();
return (
statusCode === 499 ||
normalized.includes("responseaborted") ||
normalized.includes("controller is already closed") ||
normalized.includes("readablestream is closed") ||
normalized.includes("writablestream is closed") ||
normalized.includes("aborterror")
);
};
let pipelineStreamFailureFinalized = false;
const onPipelineStreamError: PipelineStreamErrorHandler = ({ message, statusCode }) => {
if (pipelineStreamFailureFinalized) return true;
pipelineStreamFailureFinalized = true;
const status =
Number.isFinite(statusCode) && statusCode >= 400 && statusCode <= 599
const normalizedMessage = message || "Upstream stream error";
const clientClosed = isClientClosedPipelineError(normalizedMessage, statusCode);
const status = clientClosed
? 499
: Number.isFinite(statusCode) && statusCode >= 400 && statusCode <= 599
? statusCode
: HTTP_STATUS.BAD_GATEWAY;
const normalizedMessage = message || "Upstream stream error";
const code = normalizedMessage.toLowerCase().includes("terminated")
? "stream_terminated"
: "stream_pipeline_error";
const code = clientClosed
? "client_disconnected"
: normalizedMessage.toLowerCase().includes("terminated")
? "stream_terminated"
: "stream_pipeline_error";
const type = clientClosed ? "client_disconnected" : "stream_error";
handleStreamFailure({
status,
message: normalizedMessage,
code,
type: "stream_error",
type,
});
return true;
};

120
open-sse/utils/streamHandler.ts
View File

@@ -5,8 +5,6 @@ import { PENDING_REQUEST_CLEARED_MARKER } from "./stream.ts";
// Stream handler with disconnect detection - shared for all providers
const DISCONNECT_ABORT_DELAY_MS = 2_000;
// Default budget for the pipeWithDisconnect raw-upstream stall watchdog.
// Inherits STREAM_IDLE_TIMEOUT_MS so a single env knob still governs the
// max time we tolerate silence from upstream. Reasoning models (Claude
@@ -30,12 +28,13 @@ type StreamErrorEvent = {
};
type StreamControllerOptions = {
onDisconnect?: (event: StreamDisconnectEvent) => void;
onDisconnect?: (event: StreamDisconnectEvent) => boolean | void;
onError?: (event: StreamErrorEvent) => boolean | void;
provider?: string;
model?: string;
connectionId?: string | null;
clientResponseFormat?: string | null;
clientAbortSignal?: AbortSignal | null;
};
type StreamController = ReturnType<typeof createStreamController>;
@@ -177,6 +176,28 @@ function getErrorStatusCode(error: unknown): number {
return 502;
}
function hasClientTerminalSseMarker(text: string, clientResponseFormat?: string | null): boolean {
if (/(?:^|\r?\n)data:\s*\[DONE\]\s*(?:\r?\n|$)/.test(text)) {
return true;
}
if (isResponsesClientFormat(clientResponseFormat)) {
return (
/(?:^|\r?\n)event:\s*response\.completed\s*(?:\r?\n|$)/.test(text) ||
/"type"\s*:\s*"response\.completed"/.test(text)
);
}
if (clientResponseFormat === FORMATS.CLAUDE) {
return (
/(?:^|\r?\n)event:\s*message_stop\s*(?:\r?\n|$)/.test(text) ||
/"type"\s*:\s*"message_stop"/.test(text)
);
}
return false;
}
/**
* Create stream controller with abort and disconnect detection
* @param {object} options
@@ -193,12 +214,13 @@ export function createStreamController({
model,
connectionId,
clientResponseFormat,
clientAbortSignal,
}: StreamControllerOptions = {}) {
const abortController = new AbortController();
const startTime = Date.now();
let disconnected = false;
let abortTimeout: ReturnType<typeof setTimeout> | null = null;
let pendingRequestCleared = false;
let cleanupClientAbortSignal: (() => void) | null = null;
const logStream = (status) => {
const duration = Date.now() - startTime;
@@ -226,7 +248,24 @@ export function createStreamController({
} catch {}
};
return {
const cleanupClientAbortListener = () => {
if (!cleanupClientAbortSignal) return;
cleanupClientAbortSignal();
cleanupClientAbortSignal = null;
};
const getClientAbortReason = () => {
const reason = clientAbortSignal?.reason;
if (typeof reason === "string" && reason.trim().length > 0) {
return reason;
}
if (reason instanceof Error && reason.message) {
return reason.message;
}
return "request_signal_aborted";
};
const controller = {
signal: abortController.signal,
startTime,
@@ -236,6 +275,7 @@ export function createStreamController({
handleDisconnect: (reason = "client_closed") => {
if (disconnected) return;
disconnected = true;
cleanupClientAbortListener();
logStream(`disconnect: ${reason}`);
@@ -243,10 +283,7 @@ export function createStreamController({
// fire when the client aborts mid-stream, so we must clean up here.
clearPendingRequest();
// Delay abort to allow cleanup
abortTimeout = setTimeout(() => {
abortController.abort();
}, DISCONNECT_ABORT_DELAY_MS);
abortController.abort(reason);
onDisconnect?.({ reason, duration: Date.now() - startTime });
},
@@ -255,21 +292,14 @@ export function createStreamController({
handleComplete: () => {
if (disconnected) return;
disconnected = true;
cleanupClientAbortListener();
logStream("complete");
if (abortTimeout) {
clearTimeout(abortTimeout);
abortTimeout = null;
}
},
// Call on error
handleError: (error: unknown) => {
if (abortTimeout) {
clearTimeout(abortTimeout);
abortTimeout = null;
}
cleanupClientAbortListener();
// A client disconnect is not a provider failure. If the client already went away
// (disconnected) or the error is a client abort / "Controller is already closed",
@@ -313,9 +343,28 @@ export function createStreamController({
logStream("error: unknown");
},
abort: () => abortController.abort(),
abort: () => {
cleanupClientAbortListener();
abortController.abort();
},
clientResponseFormat,
};
if (clientAbortSignal && typeof clientAbortSignal.addEventListener === "function") {
const handleClientAbort = () => {
controller.handleDisconnect(getClientAbortReason());
};
if (clientAbortSignal.aborted) {
queueMicrotask(handleClientAbort);
} else {
clientAbortSignal.addEventListener("abort", handleClientAbort, { once: true });
cleanupClientAbortSignal = () => {
clientAbortSignal.removeEventListener("abort", handleClientAbort);
};
}
}
return controller;
}
function buildStreamErrorChunks(
@@ -396,6 +445,23 @@ export function createNoopAbortWritable(): {
export function createDisconnectAwareStream(transformStream, streamController) {
const reader = transformStream.readable.getReader();
const writer = transformStream.writable.getWriter();
const terminalDecoder = new TextDecoder();
let terminalTail = "";
let clientTerminalSeen = false;
const noteClientChunk = (chunk: unknown) => {
if (clientTerminalSeen) return;
if (!(chunk instanceof Uint8Array)) return;
terminalTail += terminalDecoder.decode(chunk, { stream: true });
if (terminalTail.length > 4096) {
terminalTail = terminalTail.slice(-4096);
}
clientTerminalSeen = hasClientTerminalSseMarker(
terminalTail,
streamController.clientResponseFormat
);
};
return new ReadableStream(
{
@@ -413,6 +479,7 @@ export function createDisconnectAwareStream(transformStream, streamController) {
return;
}
controller.enqueue(value);
noteClientChunk(value);
} catch (error) {
if (!streamController.isConnected()) {
try {
@@ -421,6 +488,14 @@ export function createDisconnectAwareStream(transformStream, streamController) {
return;
}
if (clientTerminalSeen) {
streamController.handleComplete();
try {
controller.close();
} catch {}
return;
}
streamController.handleError(error);
// T35: Encapsulate mid-stream errors as SSE events instead of abruptly aborting
@@ -447,12 +522,9 @@ export function createDisconnectAwareStream(transformStream, streamController) {
}
},
cancel(reason) {
async cancel(reason) {
streamController.handleDisconnect(reason || "cancelled");
reader.cancel();
setTimeout(() => {
writer.abort();
}, DISCONNECT_ABORT_DELAY_MS).unref?.();
await Promise.allSettled([reader.cancel(reason), writer.abort(reason)]);
},
},
{ highWaterMark: 16384 }

5
open-sse/utils/streamHelpers.ts
View File

@@ -12,6 +12,7 @@
*/
import { FORMATS } from "../translator/formats.ts";
import { hasAnyReasoningSignal } from "./reasoningFields.ts";
type SSEPayloadOptions = {
eventType?: string;
@@ -324,9 +325,7 @@ export function hasValuableContent(chunk: Record<string, unknown>, format: strin
const delta = isRecord(firstChoice?.delta) ? firstChoice.delta : null;
if (!firstChoice || !delta) return false;
if (typeof delta.content === "string" && delta.content.length > 0) return true;
if (typeof delta.reasoning_content === "string" && delta.reasoning_content.length > 0)
return true;
if (typeof delta.reasoning_text === "string" && delta.reasoning_text.length > 0) return true;
if (hasAnyReasoningSignal(delta)) return true;
if (Array.isArray(delta.tool_calls) && delta.tool_calls.length > 0) return true;
if (firstChoice.finish_reason) return true;
if (typeof delta.role === "string" && delta.role.length > 0) return true;

2
src/app/(dashboard)/dashboard/providers/[id]/components/modals/AddApiKeyModal.tsx
View File

@@ -1,5 +1,4 @@
"use client";
import { useState, useEffect, useRef } from "react";
import { useTranslations } from "next-intl";
import { Button, Badge, Input, Modal, Toggle } from "@/shared/components";
@@ -28,7 +27,6 @@ import WebSessionCredentialGuide from "../WebSessionCredentialGuide";
import CcCompatibleRequestDefaultsFields from "./CcCompatibleRequestDefaultsFields";
import { buildAddProviderSpecificData } from "./connectionProviderSpecificData";
import QuotaScrapingFields, { EMPTY_QUOTA_SCRAPING_FIELDS } from "./QuotaScrapingFields";
export interface AddApiKeyModalProps {
isOpen: boolean;
provider?: string;

16
src/sse/handlers/chat.ts
View File

@@ -783,6 +783,7 @@ export function buildClientRawRequest(request: Request, body: unknown) {
endpoint: url.pathname,
body: cloneLogPayload(body),
headers: Object.fromEntries(request.headers.entries()),
signal: request.signal ?? null,
};
}
@@ -1499,16 +1500,15 @@ async function handleSingleModelChat(
// A3 guard: if 401 and connection has extra keys, skip connection-level disable
// (key-level failure already recorded in chatCore.ts via T07)
// Check extra keys directly from credentials for reliability across restarts
const extraKeys =
(credentials.providerSpecificData?.extraApiKeys as string[] | undefined) ?? [];
const hasExtraKeys = extraKeys.length > 0 || connectionHasExtraKeys(credentials.connectionId);
const hasExtraKeys =
((credentials.providerSpecificData?.extraApiKeys as string[] | undefined) ?? []).length >
0 || connectionHasExtraKeys(credentials.connectionId);
const is401 = result.status === 401;
// Our own deadline timeout (fetch-start / body / combo-per-model, surfaced as a 504
// tagged "upstream_timeout") fired on a slow-but-not-failed upstream — the request
// was still being processed. The connection is healthy, so don't cool it down: a
// self-inflicted-timeout cooldown penalises a healthy account and, when a provider
// has a single connection, blocks every subsequent request.
// Our own timeout fired on a slow upstream; don't cool down a healthy account.
const skipConnectionDisable =
result.status === 499 ||
result.errorCode === "client_disconnected" ||
result.errorType === "client_disconnected" ||
(is401 && hasExtraKeys) ||
isSelfInflictedUpstreamTimeout(result.status, result.errorType, provider);

12
src/sse/handlers/chatHelpers.ts
View File

@@ -429,6 +429,13 @@ export async function executeChatWithBreaker({
onStreamFailure: async (failure: any) => {
if (isShadowTraffic) return;
if (!credentials.connectionId) return;
if (
Number(failure?.status) === 499 ||
failure?.code === "client_disconnected" ||
failure?.type === "client_disconnected"
) {
return;
}
// A3 guard: if 401 and connection has extra keys, skip connection-level disable
// (key-level failure already recorded in chatCore.ts via T07)
// Check extra keys directly from credentials for reliability across restarts
@@ -603,10 +610,7 @@ export function handleNoCredentials(
// all disabled. log level is `warn` rather than `error` because zero active
// credentials is an expected operator-driven state, not a server fault.
log.warn("AUTH", `No active credentials for provider: ${provider}`);
return errorResponse(
HTTP_STATUS.NOT_FOUND,
`No active credentials for provider: ${provider}`
);
return errorResponse(HTTP_STATUS.NOT_FOUND, `No active credentials for provider: ${provider}`);
}
log.warn("CHAT", "No more accounts available", { provider });
return errorResponse(

67
tests/unit/json-to-sse-3089.test.ts
View File

@@ -58,7 +58,11 @@ describe("synthesizeOpenAiSseFromJson (#3089)", () => {
const finishChunk = events[events.length - 1];
assert.equal(finishChunk.choices[0].finish_reason, "stop");
assert.deepEqual(finishChunk.usage, { prompt_tokens: 7, completion_tokens: 3, total_tokens: 10 });
assert.deepEqual(finishChunk.usage, {
prompt_tokens: 7,
completion_tokens: 3,
total_tokens: 10,
});
});
test("content-only completion converts without a reasoning_content delta", () => {
@@ -69,7 +73,62 @@ describe("synthesizeOpenAiSseFromJson (#3089)", () => {
.filter((c) => c !== "[DONE]")
.map((c) => JSON.parse(c).choices[0].delta);
assert.equal(deltas.filter((d) => d.content === "ok").length, 1);
assert.equal(deltas.some((d) => d.reasoning_content !== undefined), false);
assert.equal(
deltas.some((d) => d.reasoning_content !== undefined),
false
);
});
test("preserves client-readable reasoning alias", () => {
const sse = synthesizeOpenAiSseFromJson(
JSON.stringify({
choices: [
{
message: {
role: "assistant",
reasoning: "client-readable thinking",
content: "final text",
},
},
],
})
);
const deltas = parseDataChunks(sse)
.filter((c) => c !== "[DONE]")
.map((c) => JSON.parse(c).choices[0].delta);
assert.equal(
deltas.find((d) => d.reasoning !== undefined)?.reasoning,
"client-readable thinking"
);
assert.equal(
deltas.some((d) => d.reasoning_content !== undefined),
false
);
});
test("mirrors unsupported reasoning aliases to reasoning_content", () => {
const sse = synthesizeOpenAiSseFromJson(
JSON.stringify({
choices: [
{
message: {
role: "assistant",
reasoning_text: "alias thinking",
content: "final text",
},
},
],
})
);
const deltas = parseDataChunks(sse)
.filter((c) => c !== "[DONE]")
.map((c) => JSON.parse(c).choices[0].delta);
assert.equal(
deltas.find((d) => d.reasoning_content !== undefined)?.reasoning_content,
"alias thinking"
);
});
test("forwards tool_calls in the delta", () => {
@@ -79,7 +138,9 @@ describe("synthesizeOpenAiSseFromJson (#3089)", () => {
{
message: {
role: "assistant",
tool_calls: [{ id: "t1", type: "function", function: { name: "f", arguments: "{}" } }],
tool_calls: [
{ id: "t1", type: "function", function: { name: "f", arguments: "{}" } },
],
},
finish_reason: "tool_calls",
},

26
tests/unit/response-sanitizer.test.ts
View File

@@ -353,7 +353,7 @@ test("sanitizeResponsesApiResponse preserves native Responses payloads and usage
assert.equal((sanitized as any).usage.output_tokens_details.reasoning_tokens, 3);
});
test("sanitizeStreamingChunk keeps only safe chunk fields and maps reasoning aliases", () => {
test("sanitizeStreamingChunk keeps only safe chunk fields and preserves readable reasoning aliases", () => {
const sanitized = sanitizeStreamingChunk({
id: "chunk_1",
object: "chat.completion.chunk",
@@ -388,7 +388,7 @@ test("sanitizeStreamingChunk keeps only safe chunk fields and maps reasoning ali
delta: {
role: "assistant",
content: "Line 1\n\nLine 2",
reasoning_content: "stream reasoning",
reasoning: "stream reasoning",
tool_calls: [{ id: "call_1" }],
},
finish_reason: "stop",
@@ -416,9 +416,28 @@ test("sanitizeStreamingChunk converts reasoning_details arrays in deltas", () =>
});
assert.equal((sanitized as any).choices[0].delta.reasoning_content, "alphabeta");
assert.deepEqual((sanitized as any).choices[0].delta.reasoning_details, [
{ type: "reasoning.text", text: "alpha" },
{ content: "beta" },
]);
});
test("sanitizeStreamingChunk preserves Copilot reasoning_text deltas", () => {
test("sanitizeStreamingChunk preserves client-readable reasoning deltas", () => {
const sanitized = sanitizeStreamingChunk({
choices: [
{
delta: {
reasoning: "readable reasoning",
},
},
],
});
assert.equal((sanitized as any).choices[0].delta.reasoning, "readable reasoning");
assert.equal((sanitized as any).choices[0].delta.reasoning_content, undefined);
});
test("sanitizeStreamingChunk preserves and mirrors Copilot reasoning_text deltas", () => {
const sanitized = sanitizeStreamingChunk({
choices: [
{
@@ -430,6 +449,7 @@ test("sanitizeStreamingChunk preserves Copilot reasoning_text deltas", () => {
});
assert.equal((sanitized as any).choices[0].delta.reasoning_text, "copilot reasoning");
assert.equal((sanitized as any).choices[0].delta.reasoning_content, "copilot reasoning");
});
test("sanitizeStreamingChunk strips commentary content from Responses completed events", () => {

39
tests/unit/stream-handler.test.ts
View File

@@ -59,6 +59,45 @@ test("createDisconnectAwareStream converts upstream errors into SSE error chunks
assert.match(text, /\[DONE\]/);
});
test("createDisconnectAwareStream treats errors after OpenAI DONE as successful completion", async () => {
let pullCount = 0;
let errorHandled = false;
const transformStream = {
readable: new ReadableStream({
pull(controller) {
pullCount += 1;
if (pullCount === 1) {
controller.enqueue(encoder.encode("data: [DONE]\n\n"));
return;
}
controller.error(new Error("terminated"));
},
}),
writable: {
getWriter() {
return {
abort() {},
};
},
},
};
const stream = createDisconnectAwareStream(
transformStream,
createStreamController({
onError() {
errorHandled = true;
},
})
);
const text = await readStreamText(stream);
assert.equal(text, "data: [DONE]\n\n");
assert.equal(errorHandled, false);
assert.doesNotMatch(text, /finish_reason/);
assert.doesNotMatch(text, /terminated/);
});
test("createDisconnectAwareStream: Gemini 503 high-demand error becomes SSE error chunk with message preserved", async () => {
const geminiMsg =
"[503]: This model is currently experiencing high demand. Spikes in demand are usually temporary. Please try again later.";

10
tests/unit/stream-utils.test.ts
View File

@@ -1286,7 +1286,7 @@ test("createSSEStream passthrough restores Claude tool names from the mapping ta
assert.equal(text.includes("tool_alias"), false);
});
test("createSSEStream passthrough fixes generic ids and normalizes reasoning aliases", async () => {
test("createSSEStream passthrough fixes generic ids and preserves readable reasoning aliases", async () => {
const text = await readTransformed(
[
`data: ${JSON.stringify({
@@ -1314,10 +1314,11 @@ test("createSSEStream passthrough fixes generic ids and normalizes reasoning ali
);
assert.match(text, /"id":"chatcmpl-/);
assert.match(text, /"reasoning_content":"Let me think first"/);
assert.match(text, /"reasoning":"Let me think first"/);
assert.doesNotMatch(text, /"reasoning_content":"Let me think first"/);
});
test("createSSEStream passthrough reserializes reasoning aliases with valid ids", async () => {
test("createSSEStream passthrough mirrors unsupported reasoning aliases with valid ids", async () => {
const text = await readTransformed(
[
`data: ${JSON.stringify({
@@ -1329,7 +1330,7 @@ test("createSSEStream passthrough reserializes reasoning aliases with valid ids"
{
index: 0,
delta: {
reasoning: "Alias-only reasoning",
reasoning_text: "Alias-only reasoning",
},
},
],
@@ -1345,7 +1346,6 @@ test("createSSEStream passthrough reserializes reasoning aliases with valid ids"
);
assert.match(text, /"reasoning_content":"Alias-only reasoning"/);
assert.doesNotMatch(text, /"reasoning":"Alias-only reasoning"/);
});
test("createSSEStream passthrough preserves OpenAI content thinking tags as content", async () => {

12
tests/unit/streamHelpers.test.ts
View File

@@ -25,11 +25,23 @@ describe("hasValuableContent", () => {
assert.strictEqual(hasValuableContent(chunk, FORMATS.OPENAI), true);
});
it("returns true for client-readable reasoning", () => {
const chunk = { choices: [{ delta: { reasoning: "thinking" } }] };
assert.strictEqual(hasValuableContent(chunk, FORMATS.OPENAI), true);
});
it("returns true for Copilot reasoning_text", () => {
const chunk = { choices: [{ delta: { reasoning_text: "thinking" } }] };
assert.strictEqual(hasValuableContent(chunk, FORMATS.OPENAI), true);
});
it("returns true for OpenRouter reasoning_details", () => {
const chunk = {
choices: [{ delta: { reasoning_details: [{ type: "reasoning.text", text: "thinking" }] } }],
};
assert.strictEqual(hasValuableContent(chunk, FORMATS.OPENAI), true);
});
it("returns true for finish_reason", () => {
const chunk = { choices: [{ delta: {}, finish_reason: "stop" }] };
assert.strictEqual(hasValuableContent(chunk, FORMATS.OPENAI), true);