From 6781a843f27b3be1d6b41a726bd3cd5d0e9ee18e Mon Sep 17 00:00:00 2001 From: Randi <55005611+rdself@users.noreply.github.com> Date: Sun, 14 Jun 2026 00:10:24 -0400 Subject: [PATCH] fix: stream routed SSE chunks promptly (#3759) Reworks stream readiness as a ping/zombie filter instead of a semantic content gate: downstream streaming is released as soon as any structured non-ping SSE event arrives, replaying the buffered prefix. Removes the fixed 2s first-byte cap (readiness now inherits REQUEST_TIMEOUT_MS unless STREAM_READINESS_TIMEOUT_MS is set) so slow first-byte reasoning providers no longer false-504. Combo stream quality stays strict: validateResponseQuality still requires an actual content_block / known non-Claude payload before accepting a routed target. Also normalizes multi-line data: framing, metadata-prefixed events, and final events that arrive without a trailing blank line. Integrated into release/v3.8.25. Co-authored-by: R.D. --- docs/reference/ENVIRONMENT.md | 2 + open-sse/config/constants.ts | 6 +- open-sse/handlers/chatCore.ts | 2 + open-sse/services/combo.ts | 32 +- open-sse/utils/passthroughTailProcessor.ts | 303 ++++++++++++ open-sse/utils/responsesStreamHelpers.ts | 166 +++++++ open-sse/utils/stream.ts | 426 ++++++----------- open-sse/utils/streamHelpers.ts | 295 +++++++++++- open-sse/utils/streamReadiness.ts | 179 +++---- src/shared/utils/runtimeTimeouts.ts | 2 +- .../combo-stream-readiness-fallback.test.ts | 372 ++++++++++++++- tests/unit/runtime-timeouts.test.ts | 3 + tests/unit/stream-readiness.test.ts | 149 +++++- tests/unit/stream-utils.test.ts | 435 +++++++++++++++++- 14 files changed, 1940 insertions(+), 432 deletions(-) create mode 100644 open-sse/utils/passthroughTailProcessor.ts create mode 100644 open-sse/utils/responsesStreamHelpers.ts diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index 9712fbc4e9..4f26b39539 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/docs/reference/ENVIRONMENT.md @@ -543,6 +543,7 @@ REQUEST_TIMEOUT_MS (global override) │ ├── FETCH_CONNECT_TIMEOUT_MS (independent, default: 30000) │ └── FETCH_KEEPALIVE_TIMEOUT_MS (independent, default: 4000) ├─→ STREAM_IDLE_TIMEOUT_MS (inherits from REQUEST_TIMEOUT_MS, default: 600000) +├─→ STREAM_READINESS_TIMEOUT_MS (inherits from REQUEST_TIMEOUT_MS, default: 80000) └─→ API_BRIDGE_PROXY_TIMEOUT_MS (inherits from REQUEST_TIMEOUT_MS, default: 30000) ├─→ API_BRIDGE_SERVER_REQUEST_TIMEOUT_MS (derived, default: 300000) ├── API_BRIDGE_SERVER_HEADERS_TIMEOUT_MS (default: 60000) @@ -555,6 +556,7 @@ REQUEST_TIMEOUT_MS (global override) | `REQUEST_TIMEOUT_MS` | _(unset)_ | Global shortcut — overrides both `FETCH_TIMEOUT_MS` and `STREAM_IDLE_TIMEOUT_MS` defaults. | | `FETCH_TIMEOUT_MS` | `600000` | Total HTTP request timeout for upstream provider calls. | | `STREAM_IDLE_TIMEOUT_MS` | `600000` | Max silence between SSE chunks before aborting. Extended-thinking models rarely pause >90s. | +| `STREAM_READINESS_TIMEOUT_MS` | `80000` | Time to receive the first non-ping SSE event. Inherits `REQUEST_TIMEOUT_MS` when set. | | `FETCH_HEADERS_TIMEOUT_MS` | = `FETCH_TIMEOUT_MS` | Time to receive response headers. | | `FETCH_BODY_TIMEOUT_MS` | = `FETCH_TIMEOUT_MS` | Time to receive the full response body. | | `FETCH_CONNECT_TIMEOUT_MS` | `30000` | TCP connection establishment timeout. | diff --git a/open-sse/config/constants.ts b/open-sse/config/constants.ts index aaa6e442a0..e1c4555e55 100644 --- a/open-sse/config/constants.ts +++ b/open-sse/config/constants.ts @@ -17,9 +17,9 @@ export const FETCH_TIMEOUT_MS = upstreamTimeouts.fetchTimeoutMs; // idle for this duration. Override with STREAM_IDLE_TIMEOUT_MS env var. export const STREAM_IDLE_TIMEOUT_MS = upstreamTimeouts.streamIdleTimeoutMs; -// Timeout for the first useful SSE event. Keep this much shorter than the -// post-start idle timeout so slow-thinking models can keep streaming after the -// first token, while dead 200 OK streams fail fast enough for combo fallback. +// Timeout for the first non-ping SSE event. Inherits REQUEST_TIMEOUT_MS when +// set, unless STREAM_READINESS_TIMEOUT_MS is specified directly. This must stay +// conservative for large prompts and slow first-byte reasoning providers. export const STREAM_READINESS_TIMEOUT_MS = upstreamTimeouts.streamReadinessTimeoutMs; // Error code used when an upstream Antigravity request stalls before response diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 713167afe5..6291b3b018 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -83,6 +83,7 @@ import { import { getCallLogPipelineCaptureStreamChunks, + getCallLogPipelineMaxSizeBytes, getChatLogTextLimit, getChatLogArrayTailItems, getChatLogMaxDepth, @@ -2302,6 +2303,7 @@ export async function handleChatCore({ const reqLogger = await createRequestLogger(sourceFormat, targetFormat, model, { enabled: detailedLoggingEnabled, captureStreamChunks: capturePipelineStreamChunks, + maxStreamChunkBytes: getCallLogPipelineMaxSizeBytes(), // Provide model/provider/connectionId so streamChunks can be attached to the // in-memory pending request record before final call-log persistence. model, diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 1dc770cd21..8bf2ac492b 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -22,6 +22,10 @@ import { import { FETCH_TIMEOUT_MS, RateLimitReason } from "../config/constants.ts"; import { errorResponse, unavailableResponse } from "../utils/error.ts"; import { clamp01 } from "../utils/number.ts"; +import { + createSSEDataLineNormalizer, + isKnownNonClaudeStreamPayload, +} from "../utils/streamHelpers.ts"; import { recordComboIntent, recordComboRequest, @@ -490,19 +494,10 @@ export async function validateResponseQuality( // detect the empty-content-block pattern (content_filter stop_reason with // no content_block_* events) WITHOUT de-streaming non-empty responses. // - // Strategy: - // - Read chunks from response.body one at a time, accumulating raw bytes. - // - Parse SSE events incrementally. - // - If a content_block_* event appears → stream HAS content. Stop buffering. - // Return a clonedResponse whose body replays buffered bytes then pipes the - // remainder of the original reader. Only the chunks up to the first content - // block were held in memory — the rest stream normally. - // - If the stream ends with a complete Claude lifecycle but NO content_block - // → return invalid (combo failover). The empty lifecycle is tiny so fully - // reading it is acceptable. - // - If the stream ends without a recognisable complete Claude lifecycle → - // return valid with a clonedResponse replaying all buffered bytes (don't - // misclassify non-Claude or partial streams as empty). + // Parse SSE events incrementally. Stop buffering once a content_block_* event + // or a known non-Claude SSE payload appears, replay the buffered prefix, then + // pipe the original reader so the rest of the stream keeps flowing normally. + // Only fail over when a complete Claude lifecycle ends without content_block. // // Non-text/event-stream streaming responses are not buffered at all. if (isStreaming) { @@ -530,7 +525,7 @@ export async function validateResponseQuality( let hasMessageStart = false; let hasContentBlock = false; let hasLifecycleEnd = false; - // `event:` type line seen before the next `data:` line in the same event. + const sseLineNormalizer = createSSEDataLineNormalizer(); let pendingEventType = ""; /** @@ -546,8 +541,8 @@ export async function validateResponseQuality( // Retain the potentially-incomplete trailing fragment. decodedSoFar = lines[lines.length - 1]; - for (let i = 0; i < lines.length - 1; i++) { - const trimmed = lines[i].trim(); + for (const line of sseLineNormalizer.normalize(lines.slice(0, -1))) { + const trimmed = line.trim(); if (trimmed.startsWith("event:")) { pendingEventType = trimmed.slice(6).trim(); @@ -573,6 +568,10 @@ export async function validateResponseQuality( (typeof parsed.type === "string" ? parsed.type : null) || pendingEventType || ""; pendingEventType = ""; + if (isKnownNonClaudeStreamPayload(parsed, eventType)) { + return true; + } + switch (eventType) { case "message_start": hasMessageStart = true; @@ -651,6 +650,7 @@ export async function validateResponseQuality( // Stream finished — flush the TextDecoder and parse any remaining text. const tail = decoder.decode(undefined, { stream: false }); if (tail) decodedSoFar += tail; + if (decodedSoFar.trim()) decodedSoFar += "\n\n"; parseAccumulatedSse(); if (hasMessageStart && hasLifecycleEnd && !hasContentBlock) { diff --git a/open-sse/utils/passthroughTailProcessor.ts b/open-sse/utils/passthroughTailProcessor.ts new file mode 100644 index 0000000000..cbc00f6987 --- /dev/null +++ b/open-sse/utils/passthroughTailProcessor.ts @@ -0,0 +1,303 @@ +import { extractUsage } from "./usageTracking.ts"; +import { parseSSEDataPayload } from "./streamHelpers.ts"; +import { + backfillResponsesCompletedOutput, + normalizeResponsesSseIds, + pushUniqueResponsesOutputItems, + stringifyIdValue, + stripResponsesLifecycleEcho, +} from "./responsesStreamHelpers.ts"; + +type JsonRecord = Record; + +type EventPrefixBuffer = { + eventType: () => string; + flush: () => string; + prefixData: (output: string, line: string) => string; + remember: (line: string) => void; +}; + +export type PassthroughTailProcessorContext = { + getSkipPassthroughEvent: () => boolean; + setSkipPassthroughEvent: (value: boolean) => void; + clearPendingPassthroughEvent: () => void; + shouldAbortOnClaudeLifecycle: (payload: unknown) => boolean; + emitClaudeEmptyStreamErrorAndAbort: () => void; + isClaudeEventPayload: (payload: unknown) => boolean; + updateClaudeEmptyResponseLifecycle: (payload: unknown) => void; + passthroughEventPrefix: EventPrefixBuffer; + emitConvertedOutput: (output: string) => void; + pushProviderPayload: (payload: unknown) => void; + pushClientPayload: (payload: unknown) => void; + setPassthroughResponsesId: (value: string) => void; + setUsage: (value: unknown) => void; + addTotalContentLength: (value: number) => void; + appendPassthroughContent: (value: string) => void; + appendPassthroughReasoning: (value: string) => void; + getResponsesReasoningKey: (payload: Record) => string | null; + markResponsesReasoningSummarySeen: (key: string) => void; + ensureVisibleResponsesReasoningSummary: (payload: Record) => boolean; + emitSyntheticResponsesReasoningSummary: (payload: Record) => void; + passthroughResponsesOutputItems: unknown[]; + passthroughResponsesPendingFunctionCalls: Map; + getPassthroughResponsesCurrentFunctionCallKey: () => string | null; + setPassthroughResponsesCurrentFunctionCallKey: (value: string | null) => void; + hasPassthroughToolCalls: () => boolean; + toResponsesCompletedWithToolCalls: (parsed: JsonRecord) => JsonRecord; +}; + +function asRecord(value: unknown): JsonRecord { + return value && typeof value === "object" && !Array.isArray(value) ? (value as JsonRecord) : {}; +} + +function getFunctionCallPendingKey(item: JsonRecord | null): string | null { + if (!item) return null; + if (typeof item.id === "string") return item.id; + if (typeof item.call_id === "string") return item.call_id; + return null; +} + +function handleResponsesTailPayload( + parsed: JsonRecord, + output: string, + context: PassthroughTailProcessorContext +): string { + const responsesIdsNormalized = normalizeResponsesSseIds(parsed); + const parsedResponse = asRecord(parsed.response); + const responseId = + (parsedResponse ? stringifyIdValue(parsedResponse.id) : null) || + stringifyIdValue(parsed.response_id); + if (responseId) { + context.setPassthroughResponsesId(responseId); + } + + const extracted = extractUsage(parsed); + if (extracted) { + context.setUsage(extracted); + } + + if (typeof parsed.delta === "string") { + context.addTotalContentLength(parsed.delta.length); + } + if (parsed.type === "response.output_text.delta" && typeof parsed.delta === "string") { + context.appendPassthroughContent(parsed.delta); + } + if ( + parsed.type === "response.reasoning_summary_text.delta" || + parsed.type === "response.reasoning_summary_text.done" || + parsed.type === "response.reasoning_summary_part.done" + ) { + const reasoningKey = context.getResponsesReasoningKey(parsed); + if (reasoningKey) { + context.markResponsesReasoningSummarySeen(reasoningKey); + } + } + if ( + parsed.type === "response.output_item.added" && + asRecord(parsed.item).type === "function_call" + ) { + const item = { ...(parsed.item as JsonRecord) }; + const pendingKey = getFunctionCallPendingKey(item); + if (pendingKey) { + if (typeof item.arguments !== "string") { + item.arguments = ""; + } + context.passthroughResponsesPendingFunctionCalls.set(pendingKey, item); + context.setPassthroughResponsesCurrentFunctionCallKey(pendingKey); + } + } + if (parsed.type === "response.function_call_arguments.delta") { + const pendingKey = + typeof parsed.item_id === "string" + ? parsed.item_id + : context.getPassthroughResponsesCurrentFunctionCallKey(); + const pending = pendingKey + ? context.passthroughResponsesPendingFunctionCalls.get(pendingKey) + : undefined; + if (pending && typeof parsed.delta === "string") { + const previousArgs = typeof pending.arguments === "string" ? pending.arguments : ""; + pending.arguments = previousArgs + parsed.delta; + } + } + if (parsed.type === "response.function_call_arguments.done") { + const pendingKey = + typeof parsed.item_id === "string" + ? parsed.item_id + : context.getPassthroughResponsesCurrentFunctionCallKey(); + const pending = pendingKey + ? context.passthroughResponsesPendingFunctionCalls.get(pendingKey) + : undefined; + if (pending) { + if (typeof parsed.arguments === "string") { + pending.arguments = parsed.arguments; + } + pushUniqueResponsesOutputItems(context.passthroughResponsesOutputItems, [pending]); + } + } + if (parsed.type === "response.output_item.done" && parsed.item) { + const reasoningSummaryInjected = context.ensureVisibleResponsesReasoningSummary(parsed); + context.emitSyntheticResponsesReasoningSummary(parsed); + pushUniqueResponsesOutputItems(context.passthroughResponsesOutputItems, [parsed.item]); + if (reasoningSummaryInjected) { + output = `data: ${JSON.stringify(parsed)}\n\n`; + } + const item = asRecord(parsed.item); + if (item.type === "function_call") { + const pendingKey = getFunctionCallPendingKey(item); + if (pendingKey) { + context.passthroughResponsesPendingFunctionCalls.delete(pendingKey); + if (context.getPassthroughResponsesCurrentFunctionCallKey() === pendingKey) { + context.setPassthroughResponsesCurrentFunctionCallKey(null); + } + } + } + } + if ( + parsed.type === "response.completed" && + Array.isArray(asRecord(parsed.response).output) && + (asRecord(parsed.response).output as unknown[]).length > 0 + ) { + pushUniqueResponsesOutputItems( + context.passthroughResponsesOutputItems, + asRecord(parsed.response).output as unknown[] + ); + } + if ( + parsed.type === "response.completed" && + context.passthroughResponsesPendingFunctionCalls.size > 0 + ) { + pushUniqueResponsesOutputItems(context.passthroughResponsesOutputItems, [ + ...context.passthroughResponsesPendingFunctionCalls.values(), + ]); + context.passthroughResponsesPendingFunctionCalls.clear(); + context.setPassthroughResponsesCurrentFunctionCallKey(null); + } + + const textualToolCallBackfilled = + parsed.type === "response.completed" && context.hasPassthroughToolCalls(); + const outputPayload = textualToolCallBackfilled + ? context.toResponsesCompletedWithToolCalls(parsed) + : parsed; + const stripped = stripResponsesLifecycleEcho(outputPayload); + const backfilled = backfillResponsesCompletedOutput( + outputPayload, + context.passthroughResponsesOutputItems + ); + + if (stripped || backfilled || textualToolCallBackfilled || responsesIdsNormalized) { + output = `data: ${JSON.stringify(outputPayload)}\n\n`; + } + + return output; +} + +function handleOpenAiTailPayload(parsed: JsonRecord, context: PassthroughTailProcessorContext) { + const firstChoice = Array.isArray(parsed.choices) + ? (parsed.choices[0] as JsonRecord | undefined) + : undefined; + const delta = asRecord(firstChoice?.delta); + if (typeof delta.content === "string") { + 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 + : ""; + if (reasoningDelta) { + context.appendPassthroughReasoning(reasoningDelta); + } +} + +export function processBufferedPassthroughLine( + line: string, + context: PassthroughTailProcessorContext +): boolean { + const trimmed = line.trim(); + + if (context.getSkipPassthroughEvent()) { + if (!trimmed) { + context.setSkipPassthroughEvent(false); + context.clearPendingPassthroughEvent(); + } + return false; + } + + if (/^event:\s*keepalive\b/i.test(trimmed)) { + context.setSkipPassthroughEvent(true); + context.clearPendingPassthroughEvent(); + return false; + } + + if (/^event:/i.test(trimmed)) { + const eventType = trimmed.replace(/^event:\s*/i, ""); + if (context.shouldAbortOnClaudeLifecycle({ type: eventType })) { + context.emitClaudeEmptyStreamErrorAndAbort(); + return true; + } + context.passthroughEventPrefix.remember(line); + return false; + } + + if (/^(?::|id:|retry:)/i.test(trimmed)) { + context.passthroughEventPrefix.remember(line); + return false; + } + + if (!trimmed) { + const pendingOutput = context.passthroughEventPrefix.flush(); + if (pendingOutput) { + context.emitConvertedOutput(pendingOutput); + } + context.clearPendingPassthroughEvent(); + return false; + } + + if (!trimmed.startsWith("data:")) { + context.passthroughEventPrefix.remember(line); + return false; + } + + const parsedPassthroughData = parseSSEDataPayload(trimmed.slice(5), { + eventType: context.passthroughEventPrefix.eventType(), + }); + if (parsedPassthroughData?.done === true) { + return false; + } + + let output = + line.startsWith("data:") && !line.startsWith("data: ") + ? `data: ${line.slice(5)}\n\n` + : `${line}\n\n`; + + if (parsedPassthroughData) { + context.pushProviderPayload(parsedPassthroughData); + if (context.shouldAbortOnClaudeLifecycle(parsedPassthroughData)) { + context.emitClaudeEmptyStreamErrorAndAbort(); + return true; + } + + if (context.isClaudeEventPayload(parsedPassthroughData)) { + context.updateClaudeEmptyResponseLifecycle(parsedPassthroughData); + } + + const parsed = parsedPassthroughData as JsonRecord; + const parsedType = typeof parsed.type === "string" ? parsed.type : ""; + const isResponses = parsedType.startsWith("response."); + const isClaude = context.isClaudeEventPayload(parsed); + + if (isResponses) { + output = handleResponsesTailPayload(parsed, output, context); + } else if (!isClaude) { + handleOpenAiTailPayload(parsed, context); + } + + context.pushClientPayload(parsed); + } + + output = context.passthroughEventPrefix.prefixData(output, line); + context.emitConvertedOutput(output); + return false; +} diff --git a/open-sse/utils/responsesStreamHelpers.ts b/open-sse/utils/responsesStreamHelpers.ts new file mode 100644 index 0000000000..bbc2b14911 --- /dev/null +++ b/open-sse/utils/responsesStreamHelpers.ts @@ -0,0 +1,166 @@ +type JsonRecord = Record; + +export function stringifyIdValue(value: unknown): string | null { + return value === null || value === undefined ? null : String(value); +} + +function normalizeResponsesOutputItemIds(item: unknown): unknown { + if (!item || typeof item !== "object" || Array.isArray(item)) { + return item; + } + + const record = item as JsonRecord; + let changed = false; + const normalized = { ...record }; + + const id = stringifyIdValue(record.id); + if (id !== null && record.id !== id) { + normalized.id = id; + changed = true; + } + + const callId = stringifyIdValue(record.call_id); + if (callId !== null && record.call_id !== callId) { + normalized.call_id = callId; + changed = true; + } + + return changed ? normalized : item; +} + +export function normalizeResponsesSseIds(payload: JsonRecord): boolean { + let changed = false; + + for (const key of ["response_id", "item_id", "call_id"] as const) { + const value = stringifyIdValue(payload[key]); + if (value !== null && payload[key] !== value) { + payload[key] = value; + changed = true; + } + } + + if (payload.item && typeof payload.item === "object" && !Array.isArray(payload.item)) { + const normalizedItem = normalizeResponsesOutputItemIds(payload.item); + if (normalizedItem !== payload.item) { + payload.item = normalizedItem; + changed = true; + } + } + + if ( + payload.response && + typeof payload.response === "object" && + !Array.isArray(payload.response) + ) { + const response = payload.response as JsonRecord; + let responseChanged = false; + const normalizedResponse = { ...response }; + + const responseId = stringifyIdValue(response.id); + if (responseId !== null && response.id !== responseId) { + normalizedResponse.id = responseId; + responseChanged = true; + } + + if (Array.isArray(response.output)) { + const normalizedOutput = response.output.map(normalizeResponsesOutputItemIds); + if (normalizedOutput.some((item, index) => item !== response.output[index])) { + normalizedResponse.output = normalizedOutput; + responseChanged = true; + } + } + + if (responseChanged) { + payload.response = normalizedResponse; + changed = true; + } + } + + return changed; +} + +function buildResponsesOutputItemKey(item: unknown): string | null { + if (!item || typeof item !== "object" || Array.isArray(item)) { + return null; + } + + const record = item as JsonRecord; + const type = typeof record.type === "string" ? record.type : ""; + const id = stringifyIdValue(record.id) ?? ""; + const callId = stringifyIdValue(record.call_id) ?? ""; + const outputIndex = typeof record.output_index === "number" ? record.output_index : ""; + const name = typeof record.name === "string" ? record.name : ""; + + if (!type && !id && !callId) { + return null; + } + + return `${type}:${id}:${callId}:${outputIndex}:${name}`; +} + +export function pushUniqueResponsesOutputItems(target: unknown[], items: readonly unknown[]) { + const seen = new Set(); + + for (const existingItem of target) { + const key = buildResponsesOutputItemKey(existingItem); + if (key) { + seen.add(key); + } + } + + for (const item of items) { + const key = buildResponsesOutputItemKey(item); + if (key && seen.has(key)) { + continue; + } + + target.push(item); + if (key) { + seen.add(key); + } + } +} + +export function backfillResponsesCompletedOutput( + parsed: unknown, + collectedItems: readonly unknown[] +): boolean { + if (!collectedItems.length) return false; + if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return false; + const obj = parsed as Record; + if (obj.type !== "response.completed") return false; + const resp = obj.response; + if (!resp || typeof resp !== "object" || Array.isArray(resp)) return false; + const r = resp as Record; + const existing = r.output; + if (Array.isArray(existing) && existing.length > 0) return false; + r.output = collectedItems.slice(); + return true; +} + +const RESPONSES_LIFECYCLE_EVENT_TYPES = new Set([ + "response.created", + "response.in_progress", + "response.completed", +]); + +export function stripResponsesLifecycleEcho(parsed: unknown): boolean { + if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return false; + const obj = parsed as Record; + if (typeof obj.type !== "string" || !RESPONSES_LIFECYCLE_EVENT_TYPES.has(obj.type)) { + return false; + } + const resp = obj.response; + if (!resp || typeof resp !== "object" || Array.isArray(resp)) return false; + const r = resp as Record; + let changed = false; + if ("instructions" in r) { + delete r.instructions; + changed = true; + } + if ("tools" in r) { + delete r.tools; + changed = true; + } + return changed; +} diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 48633fdc7c..a1347ddcce 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -12,6 +12,9 @@ import { } from "./usageTracking.ts"; import { parseSSELine, + parseSSEDataPayload, + createSSEDataLineNormalizer, + createSSEEventPrefixBuffer, hasValuableContent, fixInvalidId, formatSSE, @@ -27,7 +30,6 @@ import { STREAM_IDLE_TIMEOUT_MS, FETCH_BODY_TIMEOUT_MS, HTTP_STATUS } from "../c import { OMIT_STREAMING_CHUNK_MARKER, sanitizeStreamingChunk, - extractThinkingFromContent, } from "../handlers/responseSanitizer.ts"; import { buildErrorBody } from "./error.ts"; import { parseTextualToolCallCandidate, isValidToolCallHeaderPrefix } from "./textualToolCall.ts"; @@ -37,6 +39,14 @@ import { markToolFinish, consumeToolFinishTime, } from "../services/sessionManager.ts"; +import { + backfillResponsesCompletedOutput, + normalizeResponsesSseIds, + pushUniqueResponsesOutputItems, + stringifyIdValue, + stripResponsesLifecycleEcho, +} from "./responsesStreamHelpers.ts"; +import { processBufferedPassthroughLine } from "./passthroughTailProcessor.ts"; /** * Race a response body read against a timeout. @@ -59,88 +69,10 @@ export function withBodyTimeout( } export { COLORS, formatSSE }; +export { backfillResponsesCompletedOutput, stripResponsesLifecycleEcho }; type JsonRecord = Record; -function stringifyIdValue(value: unknown): string | null { - return value === null || value === undefined ? null : String(value); -} - -function normalizeResponsesOutputItemIds(item: unknown): unknown { - if (!item || typeof item !== "object" || Array.isArray(item)) { - return item; - } - - const record = item as JsonRecord; - let changed = false; - const normalized = { ...record }; - - const id = stringifyIdValue(record.id); - if (id !== null && record.id !== id) { - normalized.id = id; - changed = true; - } - - const callId = stringifyIdValue(record.call_id); - if (callId !== null && record.call_id !== callId) { - normalized.call_id = callId; - changed = true; - } - - return changed ? normalized : item; -} - -function normalizeResponsesSseIds(payload: JsonRecord): boolean { - let changed = false; - - for (const key of ["response_id", "item_id", "call_id"] as const) { - const value = stringifyIdValue(payload[key]); - if (value !== null && payload[key] !== value) { - payload[key] = value; - changed = true; - } - } - - if (payload.item && typeof payload.item === "object" && !Array.isArray(payload.item)) { - const normalizedItem = normalizeResponsesOutputItemIds(payload.item); - if (normalizedItem !== payload.item) { - payload.item = normalizedItem; - changed = true; - } - } - - if ( - payload.response && - typeof payload.response === "object" && - !Array.isArray(payload.response) - ) { - const response = payload.response as JsonRecord; - let responseChanged = false; - const normalizedResponse = { ...response }; - - const responseId = stringifyIdValue(response.id); - if (responseId !== null && response.id !== responseId) { - normalizedResponse.id = responseId; - responseChanged = true; - } - - if (Array.isArray(response.output)) { - const normalizedOutput = response.output.map(normalizeResponsesOutputItemIds); - if (normalizedOutput.some((item, index) => item !== response.output[index])) { - normalizedResponse.output = normalizedOutput; - responseChanged = true; - } - } - - if (responseChanged) { - payload.response = normalizedResponse; - changed = true; - } - } - - return changed; -} - export const PENDING_REQUEST_CLEARED_MARKER = "__omniroutePendingRequestCleared"; function markPendingRequestCleared(error: Error): Error { @@ -148,48 +80,6 @@ function markPendingRequestCleared(error: Error): Error { return error; } -function buildResponsesOutputItemKey(item: unknown): string | null { - if (!item || typeof item !== "object" || Array.isArray(item)) { - return null; - } - - const record = item as JsonRecord; - const type = typeof record.type === "string" ? record.type : ""; - const id = stringifyIdValue(record.id) ?? ""; - const callId = stringifyIdValue(record.call_id) ?? ""; - const outputIndex = typeof record.output_index === "number" ? record.output_index : ""; - const name = typeof record.name === "string" ? record.name : ""; - - if (!type && !id && !callId) { - return null; - } - - return `${type}:${id}:${callId}:${outputIndex}:${name}`; -} - -function pushUniqueResponsesOutputItems(target: unknown[], items: readonly unknown[]) { - const seen = new Set(); - - for (const existingItem of target) { - const key = buildResponsesOutputItemKey(existingItem); - if (key) { - seen.add(key); - } - } - - for (const item of items) { - const key = buildResponsesOutputItemKey(item); - if (key && seen.has(key)) { - continue; - } - - target.push(item); - if (key) { - seen.add(key); - } - } -} - type StreamLogger = { appendProviderChunk?: (value: string) => void; appendConvertedChunk?: (value: string) => void; @@ -394,7 +284,7 @@ function buildResponsesFunctionCallEvents(toolCall: ToolCall) { } function formatSSEDataEvents(events: unknown[]) { - return events.map((event) => `data: ${JSON.stringify(event)}\n`).join("\n"); + return events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join(""); } function toChatCompletionChunkWithToolCall(base: JsonRecord, toolCall: ToolCall) { @@ -684,80 +574,6 @@ const STREAM_MODE = { PASSTHROUGH: "passthrough", // No translation, normalize output, extract usage }; -/** - * Lifecycle event types in OpenAI Responses API streams whose `response` - * payload is a snapshot of the request (echoes back `instructions` + `tools`). - */ -const RESPONSES_LIFECYCLE_EVENT_TYPES = new Set([ - "response.created", - "response.in_progress", - "response.completed", -]); - -/** - * Backfill `parsed.response.output` on a `response.completed` event from the - * snapshots accumulated as the stream progressed (`response.output_item.done`). - * - * Why: when the upstream request runs with `store: false`, OpenAI's Responses - * API leaves `response.output` empty in the final `response.completed` - * snapshot — clients that rebuild assistant messages from that snapshot - * (notably the GitHub Copilot CLI 1.0.36) end up with `choices: []` and never - * trigger tool execution. Codex CLI and others that consume per-item events - * are unaffected; backfilling the array makes both styles work. - * - * Returns true when `parsed.response.output` was empty and got replaced, so - * the caller can re-serialize. - */ -export function backfillResponsesCompletedOutput( - parsed: unknown, - collectedItems: readonly unknown[] -): boolean { - if (!collectedItems.length) return false; - if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return false; - const obj = parsed as Record; - if (obj.type !== "response.completed") return false; - const resp = obj.response; - if (!resp || typeof resp !== "object" || Array.isArray(resp)) return false; - const r = resp as Record; - const existing = r.output; - if (Array.isArray(existing) && existing.length > 0) return false; - r.output = collectedItems.slice(); - return true; -} - -/** - * Strip the request echo (`instructions`, `tools`) from `parsed.response` - * on Responses API lifecycle events. - * - * Why: those fields can balloon the SSE message past 100 KB when the request - * carries large tool definitions / instructions. Some clients (notably the - * GitHub Copilot CLI) cannot process oversized SSE events and stop rendering - * mid-stream. The fields are pure echo of the original request — clients - * already hold the original locally — so removing them is observably safe. - * - * Returns true when the payload was modified and must be re-serialized. - */ -export function stripResponsesLifecycleEcho(parsed: unknown): boolean { - if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return false; - const obj = parsed as Record; - if (typeof obj.type !== "string" || !RESPONSES_LIFECYCLE_EVENT_TYPES.has(obj.type)) { - return false; - } - const resp = obj.response; - if (!resp || typeof resp !== "object" || Array.isArray(resp)) return false; - const r = resp as Record; - let changed = false; - if ("instructions" in r) { - delete r.instructions; - changed = true; - } - if ("tools" in r) { - delete r.tools; - changed = true; - } - return changed; -} - /** * Create unified SSE transform stream with idle timeout protection. * If the upstream provider stops sending data for STREAM_IDLE_TIMEOUT_MS, @@ -894,8 +710,8 @@ export function createSSEStream(options: StreamOptions = {}) { let idleTimer: ReturnType | null = null; let streamTimedOut = false; const claudeEmptyResponseLifecycle = createClaudeEmptyResponseLifecycle(); - let pendingPassthroughEventLine: string | null = null; - let pendingPassthroughEventEmitted = false; + const passthroughEventPrefix = createSSEEventPrefixBuffer(); + const multilineSseDataLineNormalizer = createSSEDataLineNormalizer(); const clearIdleTimer = () => { if (idleTimer) { @@ -905,19 +721,7 @@ export function createSSEStream(options: StreamOptions = {}) { }; const clearPendingPassthroughEvent = () => { - pendingPassthroughEventLine = null; - pendingPassthroughEventEmitted = false; - }; - - const maybePrefixPendingPassthroughEvent = (output: string, line: string) => { - if (!pendingPassthroughEventLine || !line.startsWith("data:")) { - return output; - } - if (!pendingPassthroughEventEmitted) { - pendingPassthroughEventEmitted = true; - return `${pendingPassthroughEventLine}\n${output}`; - } - return output; + passthroughEventPrefix.clear(); }; const applyTextualToolCallStreamingGuard = (parsed: Record) => { @@ -1054,15 +858,6 @@ export function createSSEStream(options: StreamOptions = {}) { const isResponsesEvent = typeof item?.event === "string" && item.event.startsWith("response."); if (sourceFormat === FORMATS.OPENAI && !isResponsesEvent) { itemSanitized = sanitizeStreamingChunk(itemSanitized) as Record; - - const delta = itemSanitized?.choices?.[0]?.delta; - if (delta?.content && typeof delta.content === "string") { - const { content, thinking } = extractThinkingFromContent(delta.content); - delta.content = content; - if (thinking && !delta.reasoning_content) { - delta.reasoning_content = thinking; - } - } } if (!hasValuableContent(itemSanitized, sourceFormat)) { @@ -1286,7 +1081,7 @@ export function createSSEStream(options: StreamOptions = {}) { const lines = buffer.split("\n"); buffer = lines.pop() || ""; - for (const line of lines) { + for (const line of multilineSseDataLineNormalizer.normalize(lines)) { const trimmed = line.trim(); // Passthrough mode: normalize and forward @@ -1313,12 +1108,6 @@ export function createSSEStream(options: StreamOptions = {}) { } if (/^event:/i.test(trimmed)) { - if (pendingPassthroughEventLine && !pendingPassthroughEventEmitted) { - const pendingOutput = `${pendingPassthroughEventLine}\n`; - reqLogger?.appendConvertedChunk?.(pendingOutput); - controller.enqueue(encoder.encode(pendingOutput)); - } - const eventType = trimmed.replace(/^event:\s*/i, ""); if ( shouldInjectClaudeEmptyResponseBeforeCurrentEvent(claudeEmptyResponseLifecycle, { @@ -1329,13 +1118,38 @@ export function createSSEStream(options: StreamOptions = {}) { return; } - pendingPassthroughEventLine = line; - pendingPassthroughEventEmitted = false; + passthroughEventPrefix.remember(line); continue; } + if (/^(?::|id:|retry:)/i.test(trimmed)) { + passthroughEventPrefix.remember(line); + continue; + } + + if (!trimmed) { + const pendingOutput = passthroughEventPrefix.flush(); + if (pendingOutput) { + reqLogger?.appendConvertedChunk?.(pendingOutput); + controller.enqueue(encoder.encode(pendingOutput)); + } + clearPendingPassthroughEvent(); + continue; + } + + if (!trimmed.startsWith("data:")) { + passthroughEventPrefix.remember(line); + continue; + } + + const parsedPassthroughData = trimmed.startsWith("data:") + ? parseSSEDataPayload(trimmed.slice(5), { + eventType: passthroughEventPrefix.eventType(), + }) + : null; + if (trimmed.startsWith("data:")) { - const providerPayload = parseSSELine(trimmed); + const providerPayload = parsedPassthroughData ?? parseSSELine(trimmed); if (providerPayload) { providerPayloadCollector.push(providerPayload); if ((providerPayload as { done?: unknown }).done === true) { @@ -1350,7 +1164,7 @@ export function createSSEStream(options: StreamOptions = {}) { if (trimmed.startsWith("data:") && trimmed.slice(5).trim() !== "[DONE]") { try { - let parsed = JSON.parse(trimmed.slice(5).trim()); + let parsed = parsedPassthroughData ?? JSON.parse(trimmed.slice(5).trim()); // Some upstream Responses-compatible providers leak an initial Chat Completions // bootstrap chunk (assistant role + empty content) before emitting proper @@ -1464,8 +1278,7 @@ export function createSSEStream(options: StreamOptions = {}) { controller.enqueue(encoder.encode(output)); injectedUsage = true; } else { - output = `data: ${JSON.stringify(parsed)} -`; + output = `data: ${JSON.stringify(parsed)}\n\n`; injectedUsage = true; } passthroughBufferedTextualToolCallContent = ""; @@ -1476,12 +1289,12 @@ export function createSSEStream(options: StreamOptions = {}) { incomingDelta ); parsed.delta = ""; - output = `data: ${JSON.stringify(parsed)}\n`; + output = `data: ${JSON.stringify(parsed)}\n\n`; injectedUsage = true; } else { if (passthroughBufferedTextualToolCallContent) { parsed.delta = passthroughBufferedTextualToolCallContent + incomingDelta; - output = `data: ${JSON.stringify(parsed)}\n`; + output = `data: ${JSON.stringify(parsed)}\n\n`; injectedUsage = true; } passthroughAccumulatedContent = appendBoundedText( @@ -1569,7 +1382,7 @@ export function createSSEStream(options: StreamOptions = {}) { emitSyntheticResponsesReasoningSummary(controller, parsed); pushUniqueResponsesOutputItems(passthroughResponsesOutputItems, [parsed.item]); if (reasoningSummaryInjected) { - output = `data: ${JSON.stringify(parsed)}\n`; + output = `data: ${JSON.stringify(parsed)}\n\n`; injectedUsage = true; } if (parsed.item?.type === "function_call") { @@ -1633,7 +1446,7 @@ export function createSSEStream(options: StreamOptions = {}) { textualToolCallBackfilled || responsesIdsNormalized ) { - output = `data: ${JSON.stringify(parsed)}\n`; + output = `data: ${JSON.stringify(parsed)}\n\n`; injectedUsage = true; } } else if (isClaudeSSE) { @@ -1680,7 +1493,7 @@ export function createSSEStream(options: StreamOptions = {}) { ); } if (restoredToolName) { - output = `data: ${JSON.stringify(parsed)}\n`; + output = `data: ${JSON.stringify(parsed)}\n\n`; injectedUsage = true; } } else { @@ -1708,7 +1521,7 @@ export function createSSEStream(options: StreamOptions = {}) { const emptyChoicesUsage = extractUsage(parsed) ?? parsed.usage; if (hasValidUsage(emptyChoicesUsage)) { usage = emptyChoicesUsage; - output = `data: ${JSON.stringify(parsed)}\n`; + output = `data: ${JSON.stringify(parsed)}\n\n`; injectedUsage = true; clientPayload = parsed; clientPayloadCollector.push(clientPayload); @@ -1723,12 +1536,6 @@ export function createSSEStream(options: StreamOptions = {}) { continue; } - // Detect reasoning alias before sanitization strips it - const hadReasoningAlias = !!( - parsed.choices?.[0]?.delta?.reasoning && - typeof parsed.choices[0].delta.reasoning === "string" && - !parsed.choices[0].delta.reasoning_content - ); const hadNonStringToolCallId = Array.isArray(parsed.choices) ? parsed.choices.some( (choice) => @@ -1740,6 +1547,11 @@ 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 + ); parsed = sanitizeStreamingChunk(parsed); if ( @@ -1761,15 +1573,6 @@ export function createSSEStream(options: StreamOptions = {}) { let textualToolCallConverted = false; let toolCallIdCoerced = false; - // Extract tags from streaming content - if (delta?.content && typeof delta.content === "string") { - const { content, thinking } = extractThinkingFromContent(delta.content); - delta.content = content; - if (thinking && !delta.reasoning_content) { - delta.reasoning_content = thinking; - } - } - // Split combined reasoning+content deltas into separate SSE events. // Standard OpenAI streaming never mixes both fields in one delta; // clients (e.g. LobeChat) may skip content when reasoning_content @@ -1780,7 +1583,7 @@ export function createSSEStream(options: StreamOptions = {}) { delete rDelta.content; reasoningChunk.choices[0].finish_reason = null; delete reasoningChunk.usage; - const rOutput = `data: ${JSON.stringify(reasoningChunk)}\n`; + const rOutput = `data: ${JSON.stringify(reasoningChunk)}\n\n`; passthroughAccumulatedReasoning = appendBoundedText( passthroughAccumulatedReasoning, delta.reasoning_content @@ -1789,7 +1592,6 @@ export function createSSEStream(options: StreamOptions = {}) { clientPayloadCollector.push(reasoningChunk); reqLogger?.appendConvertedChunk?.(rOutput); controller.enqueue(encoder.encode(rOutput)); - controller.enqueue(encoder.encode("\n")); delete delta.reasoning_content; } @@ -1898,7 +1700,7 @@ export function createSSEStream(options: StreamOptions = {}) { parsed.choices[0].finish_reason = "tool_calls"; // If we modify it, we must output the modified object if (!injectedUsage && hasValidUsage(parsed.usage)) { - output = `data: ${JSON.stringify(parsed)}\n`; + output = `data: ${JSON.stringify(parsed)}\n\n`; injectedUsage = true; } } @@ -1909,16 +1711,16 @@ export function createSSEStream(options: StreamOptions = {}) { ) { const estimated = estimateUsage(body, totalContentLength, FORMATS.OPENAI); parsed.usage = filterUsageForFormat(estimated, FORMATS.OPENAI); - output = `data: ${JSON.stringify(parsed)}\n`; + output = `data: ${JSON.stringify(parsed)}\n\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`; + output = `data: ${JSON.stringify(parsed)}\n\n`; injectedUsage = true; } else if (textualToolCallConverted) { - output = `data: ${JSON.stringify(parsed)}\n`; + output = `data: ${JSON.stringify(parsed)}\n\n`; injectedUsage = true; } else if ( idFixed || @@ -1927,7 +1729,7 @@ export function createSSEStream(options: StreamOptions = {}) { hadNonStringToolCallId || hadNonStringTopLevelId ) { - output = `data: ${JSON.stringify(parsed)}\n`; + output = `data: ${JSON.stringify(parsed)}\n\n`; injectedUsage = true; } } @@ -1938,18 +1740,13 @@ export function createSSEStream(options: StreamOptions = {}) { if (!injectedUsage) { if (line.startsWith("data:") && !line.startsWith("data: ")) { - output = "data: " + line.slice(5) + "\n"; + output = "data: " + line.slice(5) + "\n\n"; } else { - output = line + "\n"; + output = line + "\n\n"; } } - if (!trimmed && pendingPassthroughEventLine && !pendingPassthroughEventEmitted) { - output = `${pendingPassthroughEventLine}\n${output}`; - pendingPassthroughEventEmitted = true; - } - - output = maybePrefixPendingPassthroughEvent(output, line); + output = passthroughEventPrefix.prefixData(output, line); if (clientPayload) { clientPayloadCollector.push(clientPayload); @@ -2152,8 +1949,85 @@ export function createSSEStream(options: StreamOptions = {}) { try { const remaining = decoder.decode(); if (remaining) buffer += remaining; + let normalizedTailLines: string[] = []; + if (multilineSseDataLineNormalizer.hasPending()) { + const tailLines = buffer ? [buffer, ""] : [""]; + normalizedTailLines = multilineSseDataLineNormalizer.normalize(tailLines); + buffer = ""; + } if (mode === STREAM_MODE.PASSTHROUGH) { + const tailProcessorContext = { + getSkipPassthroughEvent: () => skipPassthroughEvent, + setSkipPassthroughEvent: (value: boolean) => { + skipPassthroughEvent = value; + }, + clearPendingPassthroughEvent, + shouldAbortOnClaudeLifecycle: (payload: unknown) => + shouldInjectClaudeEmptyResponseBeforeCurrentEvent( + claudeEmptyResponseLifecycle, + payload + ), + emitClaudeEmptyStreamErrorAndAbort: () => + emitClaudeEmptyStreamErrorAndAbort(controller, false), + isClaudeEventPayload, + updateClaudeEmptyResponseLifecycle: (payload: unknown) => + updateClaudeEmptyResponseLifecycle(claudeEmptyResponseLifecycle, payload), + passthroughEventPrefix, + emitConvertedOutput: (output: string) => { + reqLogger?.appendConvertedChunk?.(output); + controller.enqueue(encoder.encode(output)); + }, + pushProviderPayload: (payload: unknown) => providerPayloadCollector.push(payload), + pushClientPayload: (payload: unknown) => clientPayloadCollector.push(payload), + setPassthroughResponsesId: (value: string) => { + passthroughResponsesId = value; + }, + setUsage: (value: unknown) => { + usage = value as UsageTokenRecord; + }, + addTotalContentLength: (value: number) => { + totalContentLength += value; + }, + appendPassthroughContent: (value: string) => { + passthroughAccumulatedContent = appendBoundedText( + passthroughAccumulatedContent, + value + ); + }, + appendPassthroughReasoning: (value: string) => { + passthroughAccumulatedReasoning = appendBoundedText( + passthroughAccumulatedReasoning, + value + ); + }, + getResponsesReasoningKey, + markResponsesReasoningSummarySeen: (key: string) => { + passthroughResponsesReasoningSummarySeen.add(key); + }, + ensureVisibleResponsesReasoningSummary, + emitSyntheticResponsesReasoningSummary: (payload: Record) => + emitSyntheticResponsesReasoningSummary(controller, payload), + passthroughResponsesOutputItems, + passthroughResponsesPendingFunctionCalls, + getPassthroughResponsesCurrentFunctionCallKey: () => + passthroughResponsesCurrentFunctionCallKey, + setPassthroughResponsesCurrentFunctionCallKey: (value: string | null) => { + passthroughResponsesCurrentFunctionCallKey = value; + }, + hasPassthroughToolCalls: () => passthroughToolCalls.size > 0, + toResponsesCompletedWithToolCalls: (parsed: JsonRecord) => + toResponsesCompletedWithToolCalls(parsed, [ + ...passthroughToolCalls.values(), + ]) as JsonRecord, + }; + + for (const line of normalizedTailLines) { + if (processBufferedPassthroughLine(line, tailProcessorContext)) { + return; + } + } + const bufferedLine = buffer.trim(); if (skipPassthroughEvent || /^event:\s*keepalive\b/i.test(bufferedLine)) { skipPassthroughEvent = false; @@ -2189,7 +2063,7 @@ export function createSSEStream(options: StreamOptions = {}) { const isClaude = isClaudeEventPayload(flushedParsed); if (isResponses) { if (normalizeResponsesSseIds(flushedParsed)) { - output = `data: ${JSON.stringify(flushedParsed)}\n`; + output = `data: ${JSON.stringify(flushedParsed)}\n\n`; } } else if (!isClaude) { let flushChanged = false; @@ -2215,16 +2089,16 @@ export function createSSEStream(options: StreamOptions = {}) { } } if (flushChanged) { - output = `data: ${JSON.stringify(flushedParsed)}\n`; + output = `data: ${JSON.stringify(flushedParsed)}\n\n`; } } } } - if (!bufferedLine && pendingPassthroughEventLine && !pendingPassthroughEventEmitted) { - output = `${pendingPassthroughEventLine}\n${output}`; - pendingPassthroughEventEmitted = true; + if (!bufferedLine) output = passthroughEventPrefix.flush() || output; + output = passthroughEventPrefix.prefixData(output, buffer); + if (output && !output.endsWith("\n\n")) { + output = output.endsWith("\n") ? `${output}\n` : `${output}\n\n`; } - output = maybePrefixPendingPassthroughEvent(output, buffer); reqLogger?.appendConvertedChunk?.(output); controller.enqueue(encoder.encode(output)); } diff --git a/open-sse/utils/streamHelpers.ts b/open-sse/utils/streamHelpers.ts index 2146f8cccd..d209d9dae6 100644 --- a/open-sse/utils/streamHelpers.ts +++ b/open-sse/utils/streamHelpers.ts @@ -13,29 +13,302 @@ import { FORMATS } from "../translator/formats.ts"; -// Parse SSE data line -export function parseSSELine(line) { - if (!line) return null; +type SSEPayloadOptions = { + eventType?: string; + logWarning?: boolean; +}; - // Trim leading whitespace before checking prefix character - const trimmed = line.trimStart(); - if (!trimmed || trimmed.charCodeAt(0) !== 100) return null; // 'd' = 100 +type SSEChoicePayload = { + delta?: Record & { tool_calls?: unknown }; + finish_reason?: unknown; + [key: string]: unknown; +}; - const data = trimmed.slice(5).trim(); - if (data === "[DONE]") return { done: true }; +type SSEJsonPayload = Record & { + done?: boolean; + choices?: SSEChoicePayload[]; +}; +type SSEDataLineNormalizer = { + hasPending: () => boolean; + normalize: (lines: string[]) => string[]; +}; + +type SSEEventPrefixBuffer = { + clear: () => void; + eventType: () => string; + flush: () => string; + prefixData: (output: string, line: string) => string; + remember: (line: string) => void; +}; + +function isRecord(value: unknown): value is Record { + return !!value && typeof value === "object" && !Array.isArray(value); +} + +export function parseSSEDataPayload( + data: unknown, + options: SSEPayloadOptions = {} +): SSEJsonPayload | null { + const payload = String(data ?? "").trim(); + if (!payload) return null; + if (payload === "[DONE]") return { done: true }; try { - return JSON.parse(data); + const parsed = JSON.parse(payload) as unknown; + const eventType = options.eventType; + if (eventType && isRecord(parsed) && typeof parsed.type !== "string") { + return { ...parsed, type: eventType } as SSEJsonPayload; + } + return parsed as SSEJsonPayload; } catch (error) { - if (data.length > 0) { + if (options.logWarning !== false && payload.length > 0) { console.log( - `[WARN] Failed to parse SSE line (${data.length} chars): ${data.substring(0, 200)}...` + `[WARN] Failed to parse SSE payload (${payload.length} chars): ${payload.substring(0, 200)}...` ); } return null; } } +export function parseSSEDataLines( + dataLines: string[], + options: SSEPayloadOptions = {} +): SSEJsonPayload | null { + return parseSSEDataPayload(dataLines.join("\n"), options); +} + +// Parse SSE data line +export function parseSSELine(line: string): SSEJsonPayload | null { + if (!line) return null; + + // Trim leading whitespace before checking field name. + const trimmed = line.trimStart(); + if (!trimmed.startsWith("data:")) return null; + + return parseSSEDataPayload(trimmed.slice(5)); +} + +function extractSseDataLine(line: string): string | null { + const trimmed = line.trimStart().replace(/\r$/, ""); + if (!trimmed.startsWith("data:")) return null; + return trimmed.slice(5).trimStart(); +} + +export function createSSEDataLineNormalizer(): SSEDataLineNormalizer { + let pendingEventLines: string[] = []; + + const getPendingDataLines = () => + pendingEventLines + .map((line) => extractSseDataLine(line)) + .filter((line): line is string => line !== null); + + const hasSelfDescribingPendingDataPayload = () => { + const dataLines = getPendingDataLines(); + const parsed = + dataLines.length > 0 ? parseSSEDataLines(dataLines, { logWarning: false }) : null; + if (!parsed) return false; + return ( + parsed.done === true || + typeof parsed.type === "string" || + typeof parsed.object === "string" || + Array.isArray(parsed.choices) || + Array.isArray(parsed.candidates) || + isRecord(parsed.response) + ); + }; + + const flush = (output: string[]) => { + if (pendingEventLines.length === 0) return; + + const eventLines = pendingEventLines.filter((line) => line.trim().length > 0); + const dataLines: string[] = []; + const passthroughLines: string[] = []; + + for (const line of eventLines) { + const dataLine = extractSseDataLine(line); + if (dataLine !== null) { + dataLines.push(dataLine); + } else { + passthroughLines.push(line); + } + } + + output.push(...passthroughLines); + if (dataLines.length > 0) { + const parsed = parseSSEDataLines(dataLines, { logWarning: false }); + if (parsed) { + output.push(parsed.done === true ? "data: [DONE]" : `data: ${JSON.stringify(parsed)}`); + } else { + output.push(...eventLines.filter((line) => extractSseDataLine(line) !== null)); + } + } else { + output.push(...eventLines.filter((line) => extractSseDataLine(line) !== null)); + } + + output.push(""); + pendingEventLines = []; + }; + + return { + hasPending() { + return pendingEventLines.length > 0; + }, + normalize(lines: string[]) { + const output: string[] = []; + for (const line of lines) { + const normalizedLine = line.replace(/\r$/, ""); + const trimmed = normalizedLine.trim(); + + if ( + trimmed && + /^(?:event:|id:|retry:|:)/i.test(trimmed) && + hasSelfDescribingPendingDataPayload() + ) { + flush(output); + } + + pendingEventLines.push(normalizedLine); + if (!trimmed) { + flush(output); + } + } + return output; + }, + }; +} + +export function createSSEEventPrefixBuffer(): SSEEventPrefixBuffer { + let lines: string[] = []; + let emitted = false; + const hasUnemitted = () => lines.length > 0 && !emitted; + const prefix = (output: string) => { + if (!hasUnemitted()) return output; + emitted = true; + return `${lines.join("\n")}\n${output}`; + }; + return { + clear() { + lines = []; + emitted = false; + }, + eventType() { + for (let i = lines.length - 1; i >= 0; i--) { + const match = lines[i].trim().match(/^event:\s*(.+)$/i); + if (match) return match[1].trim(); + } + return ""; + }, + flush() { + return hasUnemitted() ? prefix("\n") : ""; + }, + prefixData(output, line) { + return line.startsWith("data:") ? prefix(output) : output; + }, + remember(line) { + lines.push(line); + emitted = false; + }, + }; +} + +function hasOpenAICompatibleStreamValue(parsed: Record): boolean { + if (!Array.isArray(parsed.choices)) return false; + + return parsed.choices.some((choice) => { + if (!isRecord(choice)) return false; + + const delta = isRecord(choice.delta) ? choice.delta : null; + if (!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; + } + return Array.isArray(delta.tool_calls) && delta.tool_calls.length > 0; + }); +} + +function hasResponsesStreamValue(parsed: Record, eventType = ""): boolean { + const type = typeof parsed.type === "string" ? parsed.type : eventType; + if (!type.startsWith("response.")) return false; + + if ( + type === "response.output_text.delta" || + type === "response.reasoning_text.delta" || + type === "response.reasoning_summary_text.delta" || + type === "response.function_call_arguments.delta" + ) { + return ( + (typeof parsed.delta === "string" && parsed.delta.length > 0) || + (typeof parsed.text === "string" && parsed.text.length > 0) || + (typeof parsed.arguments === "string" && parsed.arguments.length > 0) + ); + } + + if (type === "response.output_item.added" || type === "response.output_item.done") { + return isRecord(parsed.item); + } + + if (type === "response.content_part.added") { + return isRecord(parsed.part); + } + + if (type === "response.completed" && isRecord(parsed.response)) { + const output = parsed.response.output; + return Array.isArray(output) && output.length > 0; + } + + return false; +} + +function hasGeminiCandidateStreamValue(parsed: Record): boolean { + const candidates = Array.isArray(parsed.candidates) + ? parsed.candidates + : isRecord(parsed.response) && Array.isArray(parsed.response.candidates) + ? parsed.response.candidates + : []; + + return candidates.some((candidate) => { + if (!isRecord(candidate)) return false; + const content = isRecord(candidate.content) ? candidate.content : null; + const parts = Array.isArray(content?.parts) ? content.parts : []; + return parts.some((part) => { + if (!isRecord(part)) return false; + if (typeof part.text === "string" && part.text.length > 0) return true; + return isRecord(part.functionCall) || isRecord(part.executableCode); + }); + }); +} + +export function isKnownNonClaudeStreamPayload( + parsed: Record, + eventType = "" +): boolean { + if (Array.isArray(parsed.choices)) { + return hasOpenAICompatibleStreamValue(parsed); + } + + const objectType = typeof parsed.object === "string" ? parsed.object : ""; + if ( + objectType === "chat.completion.chunk" || + objectType === "text_completion" || + objectType.endsWith(".completion.chunk") + ) { + return hasOpenAICompatibleStreamValue(parsed); + } + + const type = typeof parsed.type === "string" ? parsed.type : eventType; + if (type.startsWith("response.")) return hasResponsesStreamValue(parsed, eventType); + if (Array.isArray(parsed.candidates)) return hasGeminiCandidateStreamValue(parsed); + + const response = parsed.response; + return isRecord(response) && Array.isArray(response.candidates) + ? hasGeminiCandidateStreamValue(parsed) + : false; +} + // Check if chunk has valuable content (not empty) export function hasValuableContent(chunk, format) { // OpenAI format diff --git a/open-sse/utils/streamReadiness.ts b/open-sse/utils/streamReadiness.ts index 22cae4ce35..0ee108eb6b 100644 --- a/open-sse/utils/streamReadiness.ts +++ b/open-sse/utils/streamReadiness.ts @@ -71,93 +71,22 @@ function hasUsefulJsonPayload(payload: unknown): boolean { return hasUsefulValue(payload); } -function hasOpenAIResponseLifecyclePayload( - payload: Record, - type: string -): boolean { - if (type === "response.created" || type === "response.in_progress") { - const response = payload.response; - if (!isRecord(response)) return false; - - return ( - hasNonEmptyString(response.id) || - hasNonEmptyString(response.object) || - hasNonEmptyString(response.status) || - typeof response.created_at === "number" - ); - } - - if (type === "response.output_item.added") { - const item = payload.item; - if (!isRecord(item)) return false; - - return ( - hasNonEmptyString(item.id) || - hasNonEmptyString(item.type) || - hasNonEmptyString(item.status) || - Array.isArray(item.content) || - isRecord(item.content) - ); - } - - return false; +function isPingEventType(type: string): boolean { + return /^(?:ping|keepalive|heartbeat)$/i.test(type); } -function hasChatCompletionToolCallStart(value: unknown): boolean { - // Accept a tool_call item if it has a non-empty id (fully-formed chunk) OR if it - // carries a numeric index (first OpenAI streaming chunk — id/name arrive in later - // chunks). Either form signals that a tool call has started. (#3612) - const hasToolCallStart = (item: unknown) => - isRecord(item) && (hasNonEmptyString(item.id) || typeof item.index === "number"); - if (Array.isArray(value)) return value.some(hasToolCallStart); - return hasToolCallStart(value); +function getPayloadType(payload: unknown, eventType = ""): string { + if (!isRecord(payload)) return eventType; + const type = payload.type ?? payload.event ?? payload.object; + return typeof type === "string" ? type : eventType; } -function hasChatCompletionFunctionCallStart(value: unknown): boolean { - return isRecord(value) && hasNonEmptyString(value.name); -} - -function hasChatCompletionChunkStartPayload(payload: Record): boolean { - // If object or type is PRESENT and is clearly a non-chat-chunk type, reject early. - // If both are absent (many OA-compatible backends omit object), we fall through and - // let the choices structure decide. (#3612) - const obj = payload.object; - const typ = payload.type; - const hasForeignObject = - (typeof obj === "string" && obj !== "chat.completion.chunk") || - (typeof typ === "string" && typ !== "chat.completion.chunk"); - if (hasForeignObject) return false; - - const choices = payload.choices; - if (!Array.isArray(choices) || choices.length === 0) return false; - - return choices.some((choice) => { - if (!isRecord(choice)) return false; - const delta = choice.delta; - if (!isRecord(delta)) return false; - - return ( - hasNonEmptyString(delta.role) || - hasChatCompletionToolCallStart(delta.tool_calls) || - hasChatCompletionFunctionCallStart(delta.function_call) - ); - }); -} - -function hasAcceptedStreamStartPayload(payload: unknown, eventType = ""): boolean { - if (!isRecord(payload)) return false; - - // Anthropic/Claude streams can legitimately start with lifecycle frames and - // OpenAI Responses streams can do the same before the first text/tool delta - // arrives. Treating structurally valid lifecycle frames as readiness prevents - // false 504s while ping-only/generic-empty zombie streams still fail below. - const type = typeof payload.type === "string" ? payload.type : eventType; - if (type === "message_start" && isRecord(payload.message)) return true; - if (type === "content_block_start" && isRecord(payload.content_block)) return true; - if (hasOpenAIResponseLifecyclePayload(payload, type)) return true; - if (hasChatCompletionChunkStartPayload(payload)) return true; - - return false; +function hasNonPingStructuredPayload(payload: unknown, eventType = ""): boolean { + const type = getPayloadType(payload, eventType); + if (isPingEventType(eventType) || isPingEventType(type)) return false; + if (Array.isArray(payload)) return payload.length > 0; + if (isRecord(payload)) return Object.keys(payload).length > 0; + return payload !== null && payload !== undefined; } export function hasUsefulStreamContent(text: string): boolean { @@ -184,13 +113,33 @@ export function hasUsefulStreamContent(text: string): boolean { type StreamReadinessSignalState = { currentEvent: string; + dataLines: string[]; pendingLine: string; }; +function resetCurrentEvent(state: StreamReadinessSignalState): void { + state.currentEvent = ""; + state.dataLines = []; +} + +function processStreamReadinessEvent(state: StreamReadinessSignalState): boolean { + const eventType = state.currentEvent; + const data = state.dataLines.join("\n").trim(); + resetCurrentEvent(state); + + if (isPingEventType(eventType) || !data || data === "[DONE]") return false; + + try { + return hasNonPingStructuredPayload(JSON.parse(data), eventType); + } catch { + return data.length > 0; + } +} + function processStreamReadinessLine(state: StreamReadinessSignalState, line: string): boolean { const trimmed = line.trim(); if (!trimmed || trimmed.startsWith(":")) { - if (!trimmed) state.currentEvent = ""; + if (!trimmed) return processStreamReadinessEvent(state); return false; } @@ -199,20 +148,10 @@ function processStreamReadinessLine(state: StreamReadinessSignalState, line: str return false; } - if (/^(?:ping|keepalive)$/i.test(state.currentEvent)) return false; - if (!trimmed.startsWith("data:")) return false; - - const data = trimmed.slice(5).trim(); - if (!data || data === "[DONE]") return false; - - try { - const parsed = JSON.parse(data); - return ( - hasUsefulJsonPayload(parsed) || hasAcceptedStreamStartPayload(parsed, state.currentEvent) - ); - } catch { - return data.length > 0; + if (trimmed.startsWith("data:")) { + state.dataLines.push(trimmed.slice(5).trimStart()); } + return false; } function appendStreamReadinessSignal(state: StreamReadinessSignalState, chunk: string): boolean { @@ -226,14 +165,20 @@ function appendStreamReadinessSignal(state: StreamReadinessSignalState, chunk: s return false; } +function finishStreamReadinessSignal(state: StreamReadinessSignalState): boolean { + if (state.pendingLine && processStreamReadinessLine(state, state.pendingLine)) return true; + state.pendingLine = ""; + return processStreamReadinessEvent(state); +} + export function hasStreamReadinessSignal(text: string): boolean { const state: StreamReadinessSignalState = { currentEvent: "", + dataLines: [], pendingLine: "", }; if (appendStreamReadinessSignal(state, text)) return true; - if (state.pendingLine) return processStreamReadinessLine(state, state.pendingLine); - return false; + return finishStreamReadinessSignal(state); } function createErrorResponse( @@ -320,17 +265,29 @@ export async function ensureStreamReadiness( const decoder = new TextDecoder(); const readinessState: StreamReadinessSignalState = { currentEvent: "", + dataLines: [], pendingLine: "", }; const startedAt = Date.now(); - const deadline = startedAt + options.timeoutMs; + const effectiveTimeoutMs = Math.max(0, Math.floor(options.timeoutMs)); + const deadline = startedAt + effectiveTimeoutMs; let handedOffReader = false; + const buildReadyResponse = () => + new Response(prependBufferedChunks(chunks, reader), { + status: response.status, + statusText: response.statusText, + headers: response.headers, + }); + + const timeoutReason = () => + `Stream produced no non-ping SSE event within ${effectiveTimeoutMs}ms`; + try { while (true) { const remainingMs = deadline - Date.now(); if (remainingMs <= 0) { - const reason = `Stream produced no useful content within ${options.timeoutMs}ms`; + const reason = timeoutReason(); options.log?.warn?.( "STREAM", `${reason} (${options.provider || "provider"}/${options.model || "unknown"})` @@ -354,7 +311,7 @@ export async function ensureStreamReadiness( try { readResult = await readWithTimeout(reader, remainingMs); } catch { - const reason = `Stream produced no useful content within ${options.timeoutMs}ms`; + const reason = timeoutReason(); options.log?.warn?.( "STREAM", `${reason} (${options.provider || "provider"}/${options.model || "unknown"})` @@ -375,7 +332,17 @@ export async function ensureStreamReadiness( } if (readResult.done) { - const reason = "Stream ended before producing useful content"; + const tail = decoder.decode(undefined, { stream: false }); + if (tail && appendStreamReadinessSignal(readinessState, tail)) { + handedOffReader = true; + return { ok: true, response: buildReadyResponse() }; + } + if (finishStreamReadinessSignal(readinessState)) { + handedOffReader = true; + return { ok: true, response: buildReadyResponse() }; + } + + const reason = "Stream ended before producing a non-ping SSE event"; options.log?.warn?.( "STREAM", `${reason} (${options.provider || "provider"}/${options.model || "unknown"})` @@ -406,11 +373,7 @@ export async function ensureStreamReadiness( handedOffReader = true; return { ok: true, - response: new Response(prependBufferedChunks(chunks, reader), { - status: response.status, - statusText: response.statusText, - headers: response.headers, - }), + response: buildReadyResponse(), }; } } diff --git a/src/shared/utils/runtimeTimeouts.ts b/src/shared/utils/runtimeTimeouts.ts index 446f151c4b..233208f706 100644 --- a/src/shared/utils/runtimeTimeouts.ts +++ b/src/shared/utils/runtimeTimeouts.ts @@ -97,7 +97,7 @@ export function getUpstreamTimeoutConfig( const streamReadinessTimeoutMs = readTimeoutMs( env, "STREAM_READINESS_TIMEOUT_MS", - DEFAULT_STREAM_READINESS_TIMEOUT_MS, + sharedRequestTimeoutMs ?? DEFAULT_STREAM_READINESS_TIMEOUT_MS, { allowZero: true, logger, diff --git a/tests/unit/combo-stream-readiness-fallback.test.ts b/tests/unit/combo-stream-readiness-fallback.test.ts index 7f1122a24e..4cd5f3d371 100644 --- a/tests/unit/combo-stream-readiness-fallback.test.ts +++ b/tests/unit/combo-stream-readiness-fallback.test.ts @@ -1,7 +1,7 @@ import test from "node:test"; import assert from "node:assert/strict"; -import { handleComboChat } from "../../open-sse/services/combo.ts"; +import { handleComboChat, validateResponseQuality } from "../../open-sse/services/combo.ts"; import { ensureStreamReadiness } from "../../open-sse/utils/streamReadiness.ts"; const textEncoder = new TextEncoder(); @@ -77,6 +77,372 @@ function errorResponse(status: number, message: string): Response { }); } +async function readWithTimeout(response: Response, timeoutMs = 100): Promise { + let timer: ReturnType | null = null; + try { + return await Promise.race([ + response.text(), + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error(`Timed out reading response after ${timeoutMs}ms`)), + timeoutMs + ); + }), + ]); + } finally { + if (timer) clearTimeout(timer); + } +} + +test("streaming quality peek releases OpenAI-compatible reasoning-only SSE immediately", async () => { + let streamController: ReadableStreamDefaultController | null = null; + + const firstChunk = `data: ${JSON.stringify({ + id: "chatcmpl-test", + object: "chat.completion.chunk", + choices: [ + { + delta: { role: "assistant", content: "", reasoning_content: "thinking" }, + finish_reason: null, + }, + ], + })}\n\n`; + + const body = new ReadableStream({ + start(controller) { + streamController = controller; + controller.enqueue(textEncoder.encode(firstChunk)); + }, + }); + + const result = await Promise.race([ + validateResponseQuality( + new Response(body, { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }), + true, + createLog() + ), + new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 50)), + ]); + + assert.notEqual(result, "timeout", "quality peek should not wait for the full OpenAI stream"); + assert.equal(result.valid, true); + assert.ok(result.clonedResponse, "quality peek should replay the already-read prefix"); + + streamController?.enqueue(textEncoder.encode("data: [DONE]\n\n")); + streamController?.close(); + + const text = await readWithTimeout(result.clonedResponse); + assert.match(text, /reasoning_content/); + assert.match(text, /\[DONE\]/); +}); + +test("streaming quality peek waits past OpenAI-compatible empty header chunks", async () => { + let streamController: ReadableStreamDefaultController | null = null; + + const emptyHeaderChunk = `data: ${JSON.stringify({ + id: "chatcmpl-test", + object: "chat.completion.chunk", + choices: [ + { + delta: { role: "assistant", content: "" }, + finish_reason: null, + }, + ], + })}\n\n`; + + const reasoningChunk = `data: ${JSON.stringify({ + id: "chatcmpl-test", + object: "chat.completion.chunk", + choices: [ + { + delta: { content: "", reasoning_content: "thinking" }, + finish_reason: null, + }, + ], + })}\n\n`; + + const body = new ReadableStream({ + start(controller) { + streamController = controller; + controller.enqueue(textEncoder.encode(emptyHeaderChunk)); + }, + }); + + const qualityPromise = validateResponseQuality( + new Response(body, { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }), + true, + createLog() + ); + + const earlyResult = await Promise.race([ + qualityPromise.then(() => "released"), + new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 50)), + ]); + + assert.equal(earlyResult, "pending", "empty OpenAI-compatible wrapper should not release"); + + streamController?.enqueue(textEncoder.encode(reasoningChunk)); + + const result = await Promise.race([ + qualityPromise, + new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 100)), + ]); + + assert.notEqual(result, "timeout", "quality peek should release when reasoning starts"); + assert.equal(result.valid, true); + assert.ok(result.clonedResponse, "quality peek should replay both buffered chunks"); + + streamController?.enqueue(textEncoder.encode("data: [DONE]\n\n")); + streamController?.close(); + + const text = await readWithTimeout(result.clonedResponse); + assert.match(text, /"role":"assistant"/); + assert.match(text, /reasoning_content/); + assert.match(text, /\[DONE\]/); +}); + +test("streaming quality peek waits past OpenAI-compatible finish-only chunks", async () => { + let streamController: ReadableStreamDefaultController | null = null; + + const finishOnlyChunk = `data: ${JSON.stringify({ + id: "chatcmpl-test", + object: "chat.completion.chunk", + choices: [{ delta: {}, finish_reason: "stop" }], + })}\n\n`; + + const contentChunk = `data: ${JSON.stringify({ + id: "chatcmpl-test", + object: "chat.completion.chunk", + choices: [{ delta: { content: "answer" }, finish_reason: null }], + })}\n\n`; + + const body = new ReadableStream({ + start(controller) { + streamController = controller; + controller.enqueue(textEncoder.encode(finishOnlyChunk)); + }, + }); + + const qualityPromise = validateResponseQuality( + new Response(body, { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }), + true, + createLog() + ); + + const earlyResult = await Promise.race([ + qualityPromise.then(() => "released"), + new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 50)), + ]); + + assert.equal(earlyResult, "pending", "finish-only OpenAI chunk should not release"); + + streamController?.enqueue(textEncoder.encode(contentChunk)); + + const result = await Promise.race([ + qualityPromise, + new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 100)), + ]); + + assert.notEqual(result, "timeout", "quality peek should release when content starts"); + assert.equal(result.valid, true); + assert.ok(result.clonedResponse, "quality peek should replay buffered finish and content chunks"); + + streamController?.enqueue(textEncoder.encode("data: [DONE]\n\n")); + streamController?.close(); + + const text = await readWithTimeout(result.clonedResponse); + assert.match(text, /finish_reason/); + assert.match(text, /"content":"answer"/); +}); + +test("streaming quality peek waits past Responses lifecycle-only events", async () => { + let streamController: ReadableStreamDefaultController | null = null; + + const lifecycleEvent = [ + "event: response.created", + `data: ${JSON.stringify({ response: { id: "resp_test" } })}`, + "", + "", + ].join("\n"); + + const textDeltaEvent = [ + "event: response.output_text.delta", + `data: ${JSON.stringify({ delta: "visible" })}`, + "", + "", + ].join("\n"); + + const body = new ReadableStream({ + start(controller) { + streamController = controller; + controller.enqueue(textEncoder.encode(lifecycleEvent)); + }, + }); + + const qualityPromise = validateResponseQuality( + new Response(body, { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }), + true, + createLog() + ); + + const earlyResult = await Promise.race([ + qualityPromise.then(() => "released"), + new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 50)), + ]); + + assert.equal(earlyResult, "pending", "response.created should not release quality peek"); + + streamController?.enqueue(textEncoder.encode(textDeltaEvent)); + + const result = await Promise.race([ + qualityPromise, + new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 100)), + ]); + + assert.notEqual(result, "timeout", "quality peek should release on Responses text delta"); + assert.equal(result.valid, true); + assert.ok(result.clonedResponse); + + streamController?.close(); + const text = await readWithTimeout(result.clonedResponse); + assert.match(text, /response.created/); + assert.match(text, /response.output_text.delta/); +}); + +test("streaming quality peek waits past Gemini finish-only candidates", async () => { + let streamController: ReadableStreamDefaultController | null = null; + + const finishOnlyChunk = `data: ${JSON.stringify({ + candidates: [{ finishReason: "STOP", content: { parts: [] } }], + })}\n\n`; + + const textChunk = `data: ${JSON.stringify({ + candidates: [{ content: { parts: [{ text: "gemini text" }] } }], + })}\n\n`; + + const body = new ReadableStream({ + start(controller) { + streamController = controller; + controller.enqueue(textEncoder.encode(finishOnlyChunk)); + }, + }); + + const qualityPromise = validateResponseQuality( + new Response(body, { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }), + true, + createLog() + ); + + const earlyResult = await Promise.race([ + qualityPromise.then(() => "released"), + new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 50)), + ]); + + assert.equal(earlyResult, "pending", "finish-only Gemini candidate should not release"); + + streamController?.enqueue(textEncoder.encode(textChunk)); + + const result = await Promise.race([ + qualityPromise, + new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 100)), + ]); + + assert.notEqual(result, "timeout", "quality peek should release on Gemini text parts"); + assert.equal(result.valid, true); + assert.ok(result.clonedResponse); + + streamController?.close(); + const text = await readWithTimeout(result.clonedResponse); + assert.match(text, /gemini text/); +}); + +test("streaming quality peek parses legal multi-line SSE data before releasing", async () => { + const payload = JSON.stringify({ + id: "chatcmpl-test", + object: "chat.completion.chunk", + choices: [{ delta: { reasoning_content: "split thinking" }, finish_reason: null }], + }); + const splitAt = payload.indexOf('"choices"') - 1; + assert.ok(splitAt > 0); + + const body = new ReadableStream({ + start(controller) { + controller.enqueue( + textEncoder.encode( + `data: ${payload.slice(0, splitAt)}\ndata: ${payload.slice(splitAt)}\n\n` + ) + ); + }, + }); + + const result = await Promise.race([ + validateResponseQuality( + new Response(body, { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }), + true, + createLog() + ), + new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 100)), + ]); + + assert.notEqual(result, "timeout", "quality peek should release on multi-line SSE reasoning"); + assert.equal(result.valid, true); + assert.ok(result.clonedResponse); +}); + +test("streaming quality peek still rejects complete Claude lifecycle without content blocks", async () => { + const body = new ReadableStream({ + start(controller) { + controller.enqueue( + textEncoder.encode( + [ + `event: message_start`, + `data: ${JSON.stringify({ type: "message_start", message: { id: "msg_1" } })}`, + "", + `event: message_delta`, + `data: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "content_filter" } })}`, + "", + `event: message_stop`, + `data: ${JSON.stringify({ type: "message_stop" })}`, + "", + "", + ].join("\n") + ) + ); + controller.close(); + }, + }); + + const result = await validateResponseQuality( + new Response(body, { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }), + true, + createLog() + ); + + assert.equal(result.valid, false); + assert.equal(result.reason, "streaming empty content block"); +}); + test("combo falls back when first model returns HTTP 200 zombie SSE stream", async () => { const calls: string[] = []; const log = createLog(); @@ -133,7 +499,7 @@ test("combo fails when all models return 504", async () => { strategy: "priority", models: [ { model: "glm/zombie-a", weight: 0 }, - { model: "glm/zombie-b", weight: 0 }, + { model: "openai/gpt-5.4-mini", weight: 0 }, ], config: { maxRetries: 0, retryDelayMs: 0 }, }, @@ -150,7 +516,7 @@ test("combo fails when all models return 504", async () => { assert.ok(!result.ok, "combo should fail when all models return 504"); assert.equal(result.status, 504); - assert.equal(calls.length, 2, "combo should try both models"); + assert.deepEqual(calls, ["glm/zombie-a", "openai/gpt-5.4-mini"]); }); test("combo retries 504 on same model before falling through (transient retry)", async () => { diff --git a/tests/unit/runtime-timeouts.test.ts b/tests/unit/runtime-timeouts.test.ts index 07053c5eb8..1947411b78 100644 --- a/tests/unit/runtime-timeouts.test.ts +++ b/tests/unit/runtime-timeouts.test.ts @@ -31,6 +31,7 @@ test("REQUEST_TIMEOUT_MS becomes the common timeout baseline when specific overr assert.equal(upstreamConfig.fetchTimeoutMs, 600000); assert.equal(upstreamConfig.streamIdleTimeoutMs, 600000); + assert.equal(upstreamConfig.streamReadinessTimeoutMs, 600000); assert.equal(upstreamConfig.fetchHeadersTimeoutMs, 600000); assert.equal(upstreamConfig.fetchBodyTimeoutMs, 600000); assert.equal(apiBridgeConfig.proxyTimeoutMs, 600000); @@ -42,12 +43,14 @@ test("upstream timeout config honors explicit overrides and falls back on invali REQUEST_TIMEOUT_MS: "550000", FETCH_TIMEOUT_MS: "600000", STREAM_IDLE_TIMEOUT_MS: "600000", + STREAM_READINESS_TIMEOUT_MS: "90000", FETCH_HEADERS_TIMEOUT_MS: "610000", FETCH_BODY_TIMEOUT_MS: "0", FETCH_CONNECT_TIMEOUT_MS: "45000", FETCH_KEEPALIVE_TIMEOUT_MS: "-1", }); + assert.equal(config.streamReadinessTimeoutMs, 90000); assert.equal(config.fetchHeadersTimeoutMs, 610000); assert.equal(config.fetchBodyTimeoutMs, 0); assert.equal(config.fetchConnectTimeoutMs, 45000); diff --git a/tests/unit/stream-readiness.test.ts b/tests/unit/stream-readiness.test.ts index ebfedf1c11..723aed3ee5 100644 --- a/tests/unit/stream-readiness.test.ts +++ b/tests/unit/stream-readiness.test.ts @@ -38,6 +38,7 @@ function delayedClaudeStartStream(): ReadableStream { }, })}`, "", + "", ].join("\n") ) ); @@ -53,6 +54,7 @@ function delayedClaudeStartStream(): ReadableStream { delta: { type: "text_delta", text: "slow hello" }, })}`, "", + "", ].join("\n") ) ); @@ -78,6 +80,7 @@ function delayedOpenAIResponsesStartStream(): ReadableStream { }, })}`, "", + "", ].join("\n") ) ); @@ -89,6 +92,7 @@ function delayedOpenAIResponsesStartStream(): ReadableStream { "event: response.output_text.delta", `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "slow hello" })}`, "", + "", ].join("\n") ) ); @@ -125,6 +129,46 @@ function delayedChatCompletionStartStream(): ReadableStream { }); } +function delayedGeminiStartStream(): ReadableStream { + return new ReadableStream({ + async start(controller) { + controller.enqueue( + encoder.encode( + `data: ${JSON.stringify({ + candidates: [{ content: { parts: [] } }], + })}\n\n` + ) + ); + + await new Promise((resolve) => setTimeout(resolve, 30)); + controller.enqueue( + encoder.encode( + `data: ${JSON.stringify({ + candidates: [{ content: { parts: [{ text: "slow gemini hello" }] } }], + })}\n\n` + ) + ); + controller.close(); + }, + }); +} + +function delayedUnknownStructuredStartStream(): ReadableStream { + return new ReadableStream({ + async start(controller) { + controller.enqueue( + encoder.encode('event: provider.lifecycle\ndata: {"phase":"started"}\n\n') + ); + + await new Promise((resolve) => setTimeout(resolve, 30)); + controller.enqueue( + encoder.encode('event: provider.delta\ndata: {"text":"slow unknown hello"}\n\n') + ); + controller.close(); + }, + }); +} + function zombieReadinessStream(): ReadableStream { return new ReadableStream({ async start(controller) { @@ -198,12 +242,12 @@ test("hasUsefulStreamContent detects text, reasoning, and tool deltas", () => { ); }); -test("hasStreamReadinessSignal accepts Claude stream start events", () => { +test("hasStreamReadinessSignal accepts any non-ping structured SSE event", () => { assert.equal(hasStreamReadinessSignal(": keepalive\n\n"), false); assert.equal(hasStreamReadinessSignal("event: ping\ndata: {}\n\n"), false); assert.equal( hasStreamReadinessSignal(`data: ${JSON.stringify({ type: "response.created" })}\n\n`), - false + true ); assert.equal( hasStreamReadinessSignal( @@ -241,13 +285,22 @@ test("hasStreamReadinessSignal accepts Claude stream start events", () => { ), true ); + assert.equal( + hasStreamReadinessSignal('event: provider.lifecycle\ndata: {"phase":"started"}\n\n'), + true + ); + assert.equal(hasStreamReadinessSignal('event: ping\ndata: {"phase":"started"}\n\n'), false); + assert.equal( + hasStreamReadinessSignal('event: ping\ndata: {"type":"response.created"}\n\n'), + false + ); }); -test("hasStreamReadinessSignal accepts valid OpenAI Responses lifecycle events", () => { +test("hasStreamReadinessSignal accepts Responses lifecycle events without schema gating", () => { assert.equal(hasStreamReadinessSignal(`data: ${JSON.stringify({})}\n\n`), false); assert.equal( hasStreamReadinessSignal(`data: ${JSON.stringify({ type: "response.created" })}\n\n`), - false + true ); assert.equal( hasStreamReadinessSignal( @@ -281,7 +334,7 @@ test("hasStreamReadinessSignal accepts valid OpenAI Responses lifecycle events", ); }); -test("hasStreamReadinessSignal accepts structural chat completion chunk starts", () => { +test("hasStreamReadinessSignal accepts chat completion structural chunks without content gating", () => { assert.equal( hasStreamReadinessSignal( `data: ${JSON.stringify({ @@ -316,7 +369,7 @@ test("hasStreamReadinessSignal accepts structural chat completion chunk starts", hasStreamReadinessSignal( `data: ${JSON.stringify({ object: "chat.completion.chunk", choices: [] })}\n\n` ), - false + true ); assert.equal( hasStreamReadinessSignal( @@ -325,7 +378,7 @@ test("hasStreamReadinessSignal accepts structural chat completion chunk starts", choices: [{ index: 0, delta: {} }], })}\n\n` ), - false + true ); // #3612: index-only tool_call chunk (first chunk in OpenAI streaming — no id yet) // MUST be treated as a readiness signal (tool-call has started) @@ -345,7 +398,7 @@ test("hasStreamReadinessSignal accepts structural chat completion chunk starts", choices: [{ index: 0, delta: { function_call: {} } }], })}\n\n` ), - false + true ); // #3612: chunk with valid choices but NO object/type field (some OA-compatible backends // omit object) — must be accepted as a readiness signal when delta.role is present @@ -358,7 +411,7 @@ test("hasStreamReadinessSignal accepts structural chat completion chunk starts", ), true ); - // object present but a different (non-chat-chunk) type must still be rejected + // Stream readiness is a zombie filter now, not a provider-specific schema gate. assert.equal( hasStreamReadinessSignal( `data: ${JSON.stringify({ @@ -366,7 +419,7 @@ test("hasStreamReadinessSignal accepts structural chat completion chunk starts", choices: [{ index: 0, delta: { role: "assistant" } }], })}\n\n` ), - false + true ); }); @@ -396,6 +449,26 @@ test("ensureStreamReadiness preserves buffered chunks when stream starts", async assert.match(text, / world/); }); +test("ensureStreamReadiness honors configured timeouts above 2000ms", async () => { + const response = new Response( + streamFromChunks( + [ + `data: ${JSON.stringify({ + type: "provider.started", + text: "slow first byte", + })}\n\n`, + ], + 2_100 + ), + { status: 200, headers: { "Content-Type": "text/event-stream" } } + ); + + const result = await ensureStreamReadiness(response, { timeoutMs: 3_000 }); + assert.equal(result.ok, true); + const text = await result.response.text(); + assert.match(text, /slow first byte/); +}); + test("ensureStreamReadiness hands off long Claude streams after message_start", async () => { const response = new Response(delayedClaudeStartStream(), { status: 200, @@ -447,7 +520,41 @@ test("ensureStreamReadiness hands off chat completion streams after role-only st assert.match(text, /slow chat hello/); }); -test("ensureStreamReadiness returns 504 when no useful content arrives before timeout", async () => { +test("ensureStreamReadiness hands off Gemini streams after structural candidate start", async () => { + const response = new Response(delayedGeminiStartStream(), { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }); + + const result = await ensureStreamReadiness(response, { + timeoutMs: 10, + provider: "gemini", + model: "gemini-3.0-pro", + }); + assert.equal(result.ok, true); + const text = await result.response.text(); + assert.match(text, /candidates/); + assert.match(text, /slow gemini hello/); +}); + +test("ensureStreamReadiness hands off unknown structured provider events promptly", async () => { + const response = new Response(delayedUnknownStructuredStartStream(), { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }); + + const result = await ensureStreamReadiness(response, { + timeoutMs: 10, + provider: "provider-specific", + model: "custom-stream-model", + }); + assert.equal(result.ok, true); + const text = await result.response.text(); + assert.match(text, /provider\.lifecycle/); + assert.match(text, /slow unknown hello/); +}); + +test("ensureStreamReadiness returns 504 when no non-ping SSE event arrives before timeout", async () => { const response = new Response(zombieReadinessStream(), { status: 200, headers: { "Content-Type": "text/event-stream" }, @@ -456,10 +563,12 @@ test("ensureStreamReadiness returns 504 when no useful content arrives before ti const result = await ensureStreamReadiness(response, { timeoutMs: 10 }); assert.equal(result.ok, false); assert.equal(result.response.status, 504); - assert.match(await result.response.text(), /STREAM_READINESS_TIMEOUT/); + const body = await result.response.json(); + assert.equal(body.error.code, "STREAM_READINESS_TIMEOUT"); + assert.match(body.error.message, /non-ping SSE event/); }); -test("ensureStreamReadiness returns 502 when stream ends without useful content", async () => { +test("ensureStreamReadiness returns 502 when stream ends without a non-ping SSE event", async () => { const response = new Response(streamFromChunks([": keepalive\n\n"]), { status: 200, headers: { "Content-Type": "text/event-stream" }, @@ -470,6 +579,20 @@ test("ensureStreamReadiness returns 502 when stream ends without useful content" assert.equal(result.response.status, 502); }); +test("ensureStreamReadiness accepts a final event without a trailing blank line", async () => { + const response = new Response( + streamFromChunks(['event: provider.lifecycle\ndata: {"ok":true}']), + { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + } + ); + + const result = await ensureStreamReadiness(response, { timeoutMs: 100 }); + assert.equal(result.ok, true); + assert.match(await result.response.text(), /provider\.lifecycle/); +}); + // Regression for #2520: a reasoning-only stream (Mistral `thinking` array / StepFun // `reasoning_details`) is real output and must NOT be classified as "no useful content" // (which produced a spurious 502). diff --git a/tests/unit/stream-utils.test.ts b/tests/unit/stream-utils.test.ts index a955759ec1..bc6c7405e6 100644 --- a/tests/unit/stream-utils.test.ts +++ b/tests/unit/stream-utils.test.ts @@ -49,6 +49,22 @@ async function readWithTransform(chunks, transformStream) { return new Response(source.pipeThrough(transformStream)).text(); } +function multilineDataEvent(payload, splitBeforeKey) { + const json = JSON.stringify(payload); + const splitAt = json.indexOf(`"${splitBeforeKey}"`) - 1; + assert.ok(splitAt > 0, `split key ${splitBeforeKey} must exist in payload`); + return `data: ${json.slice(0, splitAt)}\ndata: ${json.slice(splitAt)}\n\n`; +} + +function parseJsonDataPayloads(text) { + return text + .split("\n") + .filter((line) => line.startsWith("data: ")) + .map((line) => line.slice(6)) + .filter((data) => data && data !== "[DONE]") + .map((data) => JSON.parse(data)); +} + test.after(() => { core.resetDbInstance(); if (fs.existsSync(TEST_DATA_DIR)) { @@ -564,6 +580,101 @@ test("createSSEStream passthrough flushes a buffered final line without a traili assert.equal(text.includes("data: "), true); }); +test("createSSEStream passthrough merges multi-line SSE data before forwarding", async () => { + const text = await readTransformed( + [ + multilineDataEvent( + { + id: "chatcmpl_multiline", + object: "chat.completion.chunk", + created: 1, + model: "gpt-4.1-mini", + choices: [{ index: 0, delta: { content: "hello" }, finish_reason: null }], + }, + "choices" + ), + `data: [DONE]\n\n`, + ], + { + mode: "passthrough", + sourceFormat: FORMATS.OPENAI, + provider: "openai", + model: "gpt-4.1-mini", + body: { + messages: [{ role: "user", content: "hello" }], + }, + } + ); + + assert.match(text, /"content":"hello"/); + assert.doesNotMatch(text, /\ndata:\s*,/); + const firstPayload = text + .split("\n") + .find((line) => line.startsWith("data: ") && !line.includes("[DONE]")); + assert.ok(firstPayload, "merged SSE payload should be forwarded as a single JSON data line"); + assert.equal(JSON.parse(firstPayload.slice(6)).choices[0].delta.content, "hello"); +}); + +test("createSSEStream passthrough forwards data only after the complete SSE event boundary", async () => { + const text = await readTransformed( + [ + [ + `data: ${JSON.stringify({ delta: "hello" })}`, + "event: response.output_text.delta", + "", + "", + ].join("\n"), + ], + { + mode: "passthrough", + sourceFormat: FORMATS.OPENAI, + provider: "openai", + model: "responses-model", + body: { + messages: [{ role: "user", content: "hello" }], + }, + } + ); + + assert.match(text, /^event: response\.output_text\.delta\ndata: /); + assert.doesNotMatch(text, /^data: .*?\n\nevent:/s); +}); + +test("createSSEStream passthrough preserves event metadata in a single SSE event", async () => { + const text = await readTransformed( + [ + [ + ": upstream-note", + "id: 42", + "trace: upstream-abc", + `data: ${JSON.stringify({ + id: "chatcmpl_meta", + object: "chat.completion.chunk", + created: 1, + model: "gpt-4.1-mini", + choices: [{ index: 0, delta: { content: "metadata content" } }], + })}`, + "", + "", + ].join("\n"), + ], + { + mode: "passthrough", + sourceFormat: FORMATS.OPENAI, + provider: "openai", + model: "gpt-4.1-mini", + body: { + messages: [{ role: "user", content: "hello" }], + }, + } + ); + + assert.match(text, /^: upstream-note\nid: 42\ntrace: upstream-abc\ndata: /); + assert.doesNotMatch(text, /^: upstream-note\n\nid: 42/s); + assert.doesNotMatch(text, /\ntrace: upstream-abc\n\n/s); + assert.match(text, /metadata content/); +}); + test("createSSEStream translate mode converts Claude SSE into OpenAI chunks and completion payload", async () => { let onCompletePayload = null; const text = await readTransformed( @@ -619,6 +730,198 @@ test("createSSEStream translate mode converts Claude SSE into OpenAI chunks and assert.equal(onCompletePayload.responseBody.usage.total_tokens, 4); }); +test("createSSEStream translate mode preserves Claude text_delta thinking tags as content", async () => { + let onCompletePayload = null; + const text = await readTransformed( + [ + `data: ${JSON.stringify({ + type: "message_start", + message: { + id: "msg_text_thinking_tag", + model: "claude-opus-4-6", + role: "assistant", + usage: { input_tokens: 3 }, + }, + })}\n\n`, + `data: ${JSON.stringify({ + type: "content_block_start", + index: 0, + content_block: { type: "text", text: "" }, + })}\n\n`, + `data: ${JSON.stringify({ + type: "content_block_delta", + index: 0, + delta: { type: "text_delta", text: "\n[metacognition" }, + })}\n\n`, + `data: ${JSON.stringify({ + type: "content_block_delta", + index: 0, + delta: { type: "text_delta", text: "]\n\nVisible answer" }, + })}\n\n`, + `data: ${JSON.stringify({ + type: "message_delta", + delta: { stop_reason: "end_turn" }, + usage: { output_tokens: 4 }, + })}\n\n`, + `data: ${JSON.stringify({ + type: "message_stop", + })}\n\n`, + ], + { + mode: "translate", + targetFormat: FORMATS.CLAUDE, + sourceFormat: FORMATS.OPENAI, + provider: "claude", + model: "claude-opus-4-6", + body: { + messages: [{ role: "user", content: "hello" }], + }, + onComplete(payload) { + onCompletePayload = payload; + }, + } + ); + + const deltas = parseJsonDataPayloads(text) + .map((payload) => payload.choices?.[0]?.delta) + .filter(Boolean); + + assert.deepEqual( + deltas.filter((delta) => typeof delta.content === "string").map((delta) => delta.content), + ["\n[metacognition", "]\n\nVisible answer"] + ); + assert.equal( + deltas.some((delta) => delta.reasoning_content !== undefined), + false + ); + assert.equal( + onCompletePayload.responseBody.choices[0].message.content, + "\n[metacognition]\n\nVisible answer" + ); +}); + +test("createSSEStream translate mode keeps native Claude thinking_delta as reasoning_content", async () => { + const text = await readTransformed( + [ + `data: ${JSON.stringify({ + type: "message_start", + message: { + id: "msg_native_thinking", + model: "claude-opus-4-6", + role: "assistant", + usage: { input_tokens: 3 }, + }, + })}\n\n`, + `data: ${JSON.stringify({ + type: "content_block_start", + index: 0, + content_block: { type: "thinking", thinking: "" }, + })}\n\n`, + `data: ${JSON.stringify({ + type: "content_block_delta", + index: 0, + delta: { type: "thinking_delta", thinking: "Native plan" }, + })}\n\n`, + `data: ${JSON.stringify({ + type: "content_block_stop", + index: 0, + })}\n\n`, + `data: ${JSON.stringify({ + type: "content_block_start", + index: 1, + content_block: { type: "text", text: "" }, + })}\n\n`, + `data: ${JSON.stringify({ + type: "content_block_delta", + index: 1, + delta: { type: "text_delta", text: "Final answer" }, + })}\n\n`, + `data: ${JSON.stringify({ + type: "message_delta", + delta: { stop_reason: "end_turn" }, + usage: { output_tokens: 4 }, + })}\n\n`, + `data: ${JSON.stringify({ + type: "message_stop", + })}\n\n`, + ], + { + mode: "translate", + targetFormat: FORMATS.CLAUDE, + sourceFormat: FORMATS.OPENAI, + provider: "claude", + model: "claude-opus-4-6", + body: { + messages: [{ role: "user", content: "hello" }], + }, + } + ); + + const deltas = parseJsonDataPayloads(text) + .map((payload) => payload.choices?.[0]?.delta) + .filter(Boolean); + + assert.ok(deltas.some((delta) => delta.reasoning_content === "Native plan")); + assert.ok(deltas.some((delta) => delta.content === "Final answer")); +}); + +test("createSSEStream translate mode parses multi-line SSE data events", async () => { + let onCompletePayload = null; + const text = await readTransformed( + [ + `data: ${JSON.stringify({ + type: "message_start", + message: { + id: "msg_multiline", + model: "claude-sonnet-4", + role: "assistant", + usage: { input_tokens: 3 }, + }, + })}\n\n`, + `data: ${JSON.stringify({ + type: "content_block_start", + index: 0, + content_block: { type: "text", text: "" }, + })}\n\n`, + multilineDataEvent( + { + type: "content_block_delta", + index: 0, + delta: { type: "text_delta", text: "Hello from multiline SSE" }, + }, + "delta" + ), + `data: ${JSON.stringify({ + type: "message_delta", + delta: { stop_reason: "end_turn" }, + usage: { output_tokens: 5 }, + })}\n\n`, + `data: ${JSON.stringify({ + type: "message_stop", + })}\n\n`, + ], + { + mode: "translate", + targetFormat: FORMATS.CLAUDE, + sourceFormat: FORMATS.OPENAI, + provider: "claude", + model: "claude-sonnet-4", + body: { + messages: [{ role: "user", content: "hello" }], + }, + onComplete(payload) { + onCompletePayload = payload; + }, + } + ); + + assert.match(text, /"content":"Hello from multiline SSE"/); + assert.equal( + onCompletePayload.responseBody.choices[0].message.content, + "Hello from multiline SSE" + ); +}); + test("createSSEStream Responses passthrough converts textual tool-call deltas before streaming", async () => { let onCompletePayload = null; const toolText = `[Tool call: terminal] @@ -939,7 +1242,88 @@ 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.equal(text.includes('"reasoning":"Let me think first"'), false); +}); + +test("createSSEStream passthrough reserializes reasoning aliases with valid ids", async () => { + const text = await readTransformed( + [ + `data: ${JSON.stringify({ + id: "chatcmpl_reasoning_alias", + object: "chat.completion.chunk", + created: 1, + model: "kimi-k2.5", + choices: [ + { + index: 0, + delta: { + reasoning: "Alias-only reasoning", + }, + }, + ], + })}\n\n`, + ], + { + mode: "passthrough", + sourceFormat: FORMATS.OPENAI, + provider: "openai", + model: "kimi-k2.5", + body: { messages: [{ role: "user", content: "hello" }] }, + } + ); + + 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 () => { + let onCompletePayload = null; + const text = await readTransformed( + [ + `data: ${JSON.stringify({ + id: "chatcmpl_text_tag", + object: "chat.completion.chunk", + created: 1, + model: "openai-compatible-model", + choices: [ + { + index: 0, + delta: { role: "assistant", content: "\nVisible prompt tag" }, + }, + ], + })}\n\n`, + `data: ${JSON.stringify({ + id: "chatcmpl_text_tag", + object: "chat.completion.chunk", + created: 1, + model: "openai-compatible-model", + choices: [{ index: 0, delta: {}, finish_reason: "stop" }], + })}\n\n`, + ], + { + mode: "passthrough", + sourceFormat: FORMATS.OPENAI, + provider: "openai-compatible", + model: "openai-compatible-model", + body: { messages: [{ role: "user", content: "hello" }] }, + onComplete(payload) { + onCompletePayload = payload; + }, + } + ); + + const deltas = parseJsonDataPayloads(text) + .map((payload) => payload.choices?.[0]?.delta) + .filter(Boolean); + + assert.ok(deltas.some((delta) => delta.content === "\nVisible prompt tag")); + assert.equal( + deltas.some((delta) => delta.reasoning_content !== undefined), + false + ); + assert.equal( + onCompletePayload.responseBody.choices[0].message.content, + "\nVisible prompt tag" + ); }); test("createSSEStream passthrough splits mixed reasoning and content deltas and estimates usage", async () => { @@ -994,6 +1378,55 @@ test("createSSEStream passthrough splits mixed reasoning and content deltas and assert.ok(onCompletePayload.responseBody.usage.total_tokens > 0); }); +test("createSSEStream passthrough writes complete SSE events per converted chunk", async () => { + const convertedChunks = []; + await readTransformed( + [ + `data: ${JSON.stringify({ + id: "chatcmpl_reasoning", + object: "chat.completion.chunk", + created: 1, + model: "gpt-4.1-mini", + choices: [ + { + index: 0, + delta: { + reasoning_content: "First think", + content: "Then answer", + }, + }, + ], + })}\n\n`, + `data: ${JSON.stringify({ + id: "chatcmpl_reasoning", + object: "chat.completion.chunk", + created: 1, + model: "gpt-4.1-mini", + choices: [{ index: 0, delta: {}, finish_reason: "stop" }], + })}\n\n`, + ], + { + mode: "passthrough", + sourceFormat: FORMATS.OPENAI, + provider: "openai", + model: "gpt-4.1-mini", + body: { + messages: [{ role: "user", content: "hello world" }], + }, + reqLogger: { + appendConvertedChunk(value) { + convertedChunks.push(value); + }, + }, + } + ); + + assert.equal(convertedChunks.includes("\n"), false); + for (const chunk of convertedChunks.filter((value) => value.startsWith("data:"))) { + assert.equal(chunk.endsWith("\n\n"), true); + } +}); + test("createSSEStream passthrough merges Claude usage chunks and restores mapped tool names", async () => { let onCompletePayload = null; const text = await readTransformed(