From 9caea8852801dc0cdbb6f67aace3b4f95ca9fe13 Mon Sep 17 00:00:00 2001 From: decolua Date: Sat, 6 Jun 2026 16:07:43 +0700 Subject: [PATCH] fix(codex): harden streaming timeouts + Responses terminal events MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Raise stall/connect timeouts to 60s (configurable per-provider), accept codex response.done, and always emit a terminal response.failed + [DONE] for Responses passthrough when a stream closes, stalls, or aborts before a terminal event — preventing codex clients from hanging. Co-authored-by: jonathanli12 Co-authored-by: rifuki Co-authored-by: nguyenha935 Co-authored-by: trananhtung Co-authored-by: Cursor --- open-sse/config/providers.js | 4 +- open-sse/config/runtimeConfig.js | 4 +- open-sse/executors/base.js | 7 +- .../handlers/chatCore/sseToJsonHandler.js | 3 +- .../handlers/chatCore/streamingHandler.js | 7 +- open-sse/handlers/imageProviders/codex.js | 4 +- open-sse/transformer/streamToJsonConverter.js | 2 +- .../translator/response/openai-responses.js | 2 +- open-sse/utils/responsesStreamHelpers.js | 49 +++++++++++ open-sse/utils/stream.js | 64 +++++++++++++- open-sse/utils/streamHandler.js | 36 +++++--- src/app/api/providers/[id]/test/testUtils.js | 2 +- tests/unit/image-generation.test.js | 2 +- .../openai-responses-terminal-event.test.js | 83 +++++++++++++++++++ tests/unit/responses-abort-terminal.test.js | 72 ++++++++++++++++ 15 files changed, 311 insertions(+), 30 deletions(-) create mode 100644 open-sse/utils/responsesStreamHelpers.js create mode 100644 tests/unit/openai-responses-terminal-event.test.js create mode 100644 tests/unit/responses-abort-terminal.test.js diff --git a/open-sse/config/providers.js b/open-sse/config/providers.js index 8658aba9..14e73f29 100644 --- a/open-sse/config/providers.js +++ b/open-sse/config/providers.js @@ -71,8 +71,8 @@ export const PROVIDERS = { baseUrl: "https://chatgpt.com/backend-api/codex/responses", format: "openai-responses", headers: { - "originator": "codex-cli", - "User-Agent": "codex-cli/1.0.18 (macOS; arm64)" + "originator": "codex_cli_rs", + "User-Agent": "codex_cli_rs/0.136.0" }, clientId: "app_EMoamEEZ73f0CkXaXp7hrann", tokenUrl: "https://auth.openai.com/oauth/token" diff --git a/open-sse/config/runtimeConfig.js b/open-sse/config/runtimeConfig.js index 1bdce2b1..37a5f84e 100644 --- a/open-sse/config/runtimeConfig.js +++ b/open-sse/config/runtimeConfig.js @@ -32,10 +32,10 @@ export const MEMORY_CONFIG = { }; // Stream stall timeout: abort if no chunk received within this duration -export const STREAM_STALL_TIMEOUT_MS = 30 * 1000; +export const STREAM_STALL_TIMEOUT_MS = 60 * 1000; // Fetch connect timeout: abort if upstream doesn't return response headers within this duration -export const FETCH_CONNECT_TIMEOUT_MS = 20 * 1000; +export const FETCH_CONNECT_TIMEOUT_MS = 60 * 1000; // Default token limits export const DEFAULT_MAX_TOKENS = 64000; diff --git a/open-sse/executors/base.js b/open-sse/executors/base.js index 4ae2e8c4..aaf6ede8 100644 --- a/open-sse/executors/base.js +++ b/open-sse/executors/base.js @@ -121,15 +121,16 @@ export class BaseExecutor { if (!retryAttemptsByUrl[urlIndex]) retryAttemptsByUrl[urlIndex] = 0; - // Abort if upstream doesn't return response headers within FETCH_CONNECT_TIMEOUT_MS + // Abort if upstream doesn't return response headers within connection timeout const connectCtrl = new AbortController(); - const connectTimer = setTimeout(() => connectCtrl.abort(new Error("fetch connect timeout")), FETCH_CONNECT_TIMEOUT_MS); + const timeoutMs = this.config?.timeoutMs || FETCH_CONNECT_TIMEOUT_MS; + const connectTimer = setTimeout(() => connectCtrl.abort(new Error("fetch connect timeout")), timeoutMs); const mergedSignal = signal ? AbortSignal.any([signal, connectCtrl.signal]) : connectCtrl.signal; try { const bodyStr = JSON.stringify(transformedBody); const fetchT0 = Date.now(); - dbg("FETCH", `${this.provider.toUpperCase()} → ${url} | body=${bodyStr.length}B | connectTimeout=${FETCH_CONNECT_TIMEOUT_MS}ms`); + dbg("FETCH", `${this.provider.toUpperCase()} → ${url} | body=${bodyStr.length}B | connectTimeout=${timeoutMs}ms`); const response = await proxyAwareFetch(url, { method: "POST", headers, diff --git a/open-sse/handlers/chatCore/sseToJsonHandler.js b/open-sse/handlers/chatCore/sseToJsonHandler.js index b4d68ae5..9919173e 100644 --- a/open-sse/handlers/chatCore/sseToJsonHandler.js +++ b/open-sse/handlers/chatCore/sseToJsonHandler.js @@ -167,7 +167,8 @@ export async function handleForcedSSEToJson({ providerResponse, sourceFormat, pr } else { const message = { role: "assistant", content: textContent || (hasToolCalls ? null : "") }; if (hasToolCalls) message.tool_calls = toolCalls; - const finishReason = hasToolCalls ? "tool_calls" : (jsonResponse.status === "completed" ? "stop" : (jsonResponse.status || "stop")); + const responseDone = jsonResponse.status === "completed" || jsonResponse.status === "done"; + const finishReason = hasToolCalls ? "tool_calls" : (responseDone ? "stop" : (jsonResponse.status || "stop")); finalResp = { id: jsonResponse.id || `chatcmpl-${Date.now()}`, object: "chat.completion", diff --git a/open-sse/handlers/chatCore/streamingHandler.js b/open-sse/handlers/chatCore/streamingHandler.js index 03d12c13..89339587 100644 --- a/open-sse/handlers/chatCore/streamingHandler.js +++ b/open-sse/handlers/chatCore/streamingHandler.js @@ -2,6 +2,7 @@ import { FORMATS } from "../../translator/formats.js"; import { needsTranslation } from "../../translator/index.js"; import { createSSETransformStreamWithLogger, createPassthroughStreamWithLogger } from "../../utils/stream.js"; import { pipeWithDisconnect } from "../../utils/streamHandler.js"; +import { buildAbortedResponsesTerminalBytes } from "../../utils/responsesStreamHelpers.js"; import { buildRequestDetail, extractRequestConfig, saveUsageStats } from "./requestDetail.js"; import { saveRequestDetail } from "@/lib/usageDb.js"; @@ -43,7 +44,11 @@ export function handleStreamingResponse({ providerResponse, provider, model, sou if (onRequestSuccess) onRequestSuccess(); const transformStream = buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey }); - const transformedBody = pipeWithDisconnect(providerResponse, transformStream, streamController); + + // Responses passthrough: synthesize response.failed + [DONE] if the stream aborts/stalls before a terminal event + const isResponsesPassthrough = sourceFormat === FORMATS.OPENAI_RESPONSES && targetFormat === FORMATS.OPENAI_RESPONSES; + const onAbortTerminal = isResponsesPassthrough ? buildAbortedResponsesTerminalBytes : null; + const transformedBody = pipeWithDisconnect(providerResponse, transformStream, streamController, onAbortTerminal); const streamDetailId = `${Date.now()}-${Math.random().toString(36).slice(2, 11)}`; saveRequestDetail(buildRequestDetail({ diff --git a/open-sse/handlers/imageProviders/codex.js b/open-sse/handlers/imageProviders/codex.js index 2f8edc55..591c7262 100644 --- a/open-sse/handlers/imageProviders/codex.js +++ b/open-sse/handlers/imageProviders/codex.js @@ -3,8 +3,8 @@ import { randomUUID } from "node:crypto"; import { nowSec } from "./_base.js"; const CODEX_RESPONSES_URL = "https://chatgpt.com/backend-api/codex/responses"; -const CODEX_USER_AGENT = "codex-imagen/0.2.6"; -const CODEX_VERSION = "0.129.0"; +const CODEX_USER_AGENT = "codex_cli_rs/0.136.0"; +const CODEX_VERSION = "0.136.0"; const CODEX_ORIGINATOR = "codex_cli_rs"; const CODEX_MODEL_SUFFIX = "-image"; const CODEX_REF_DETAIL = "high"; diff --git a/open-sse/transformer/streamToJsonConverter.js b/open-sse/transformer/streamToJsonConverter.js index 92cb723b..c08acb25 100644 --- a/open-sse/transformer/streamToJsonConverter.js +++ b/open-sse/transformer/streamToJsonConverter.js @@ -27,7 +27,7 @@ function processSSEMessage(msg, state) { state.created = parsed.response?.created_at || state.created; } else if (eventType === "response.output_item.done") { state.items.set(parsed.output_index ?? 0, parsed.item); - } else if (eventType === "response.completed") { + } else if (eventType === "response.completed" || eventType === "response.done") { state.status = "completed"; if (parsed.response?.usage) { state.usage.input_tokens = parsed.response.usage.input_tokens || 0; diff --git a/open-sse/translator/response/openai-responses.js b/open-sse/translator/response/openai-responses.js index 0cf79cbd..e500e4c7 100644 --- a/open-sse/translator/response/openai-responses.js +++ b/open-sse/translator/response/openai-responses.js @@ -490,7 +490,7 @@ export function openaiResponsesToOpenAIResponse(chunk, state) { } // Response completed - if (eventType === "response.completed") { + if (eventType === "response.completed" || eventType === "response.done") { // Extract usage from response.completed event const responseUsage = data.response?.usage; if (responseUsage && typeof responseUsage === "object") { diff --git a/open-sse/utils/responsesStreamHelpers.js b/open-sse/utils/responsesStreamHelpers.js new file mode 100644 index 00000000..6f90a0c1 --- /dev/null +++ b/open-sse/utils/responsesStreamHelpers.js @@ -0,0 +1,49 @@ +// Helpers for OpenAI Responses API streaming termination + event framing +import { FORMATS } from "../translator/formats.js"; +import { formatSSE } from "./streamHelpers.js"; + +// Responses API events that signal the stream has reached a terminal state +const OPENAI_RESPONSES_TERMINAL_EVENTS = new Set([ + "response.completed", + "response.failed", + "error" +]); + +export function getOpenAIResponsesEventName(eventName, chunk) { + if (eventName) return eventName; + if (chunk && typeof chunk.type === "string") return chunk.type; + return null; +} + +export function isOpenAIResponsesTerminalEvent(eventName, chunk) { + const type = getOpenAIResponsesEventName(eventName, chunk); + if (OPENAI_RESPONSES_TERMINAL_EVENTS.has(type)) return true; + const status = chunk?.response?.status; + return status === "completed" || status === "failed"; +} + +const sharedEncoder = new TextEncoder(); + +// Encoded response.failed + [DONE] payload for aborted/stalled Responses passthrough streams +export function buildAbortedResponsesTerminalBytes() { + return sharedEncoder.encode(`${formatIncompleteOpenAIResponsesStreamFailure()}data: [DONE]\n\n`); +} + +// Synthesize a response.failed event for streams that close without a terminal event +export function formatIncompleteOpenAIResponsesStreamFailure() { + return formatSSE({ + event: "response.failed", + data: { + type: "response.failed", + response: { + id: `resp_${Date.now()}`, + status: "failed", + error: { + type: "stream_error", + code: "stream_disconnected", + message: "stream closed before response.completed" + } + } + } + }, FORMATS.OPENAI_RESPONSES); +} diff --git a/open-sse/utils/stream.js b/open-sse/utils/stream.js index 582c0f9c..4886efe0 100644 --- a/open-sse/utils/stream.js +++ b/open-sse/utils/stream.js @@ -3,6 +3,7 @@ import { FORMATS } from "../translator/formats.js"; import { trackPendingRequest, appendRequestLog } from "@/lib/usageDb.js"; import { extractUsage, hasValidUsage, estimateUsage, logUsage, addBufferToUsage, filterUsageForFormat, COLORS } from "./usageTracking.js"; import { parseSSELine, hasValuableContent, fixInvalidId, formatSSE } from "./streamHelpers.js"; +import { getOpenAIResponsesEventName, isOpenAIResponsesTerminalEvent, formatIncompleteOpenAIResponsesStreamFailure } from "./responsesStreamHelpers.js"; import { dbg, isDebugEnabled } from "./debugLog.js"; export { COLORS, formatSSE }; @@ -63,6 +64,11 @@ export function createSSEStream(options = {}) { let sseEmittedCount = 0; const eventTypeCounts = {}; + // Track Responses API event framing for same-format passthrough (codex) + let currentOpenAIResponsesEvent = null; + let openAIResponsesTerminalSeen = false; + let openAIResponsesDoneSent = false; + return new TransformStream({ transform(chunk, controller) { if (!ttftAt) ttftAt = Date.now(); @@ -83,6 +89,11 @@ export function createSSEStream(options = {}) { } } + // Capture Responses API event name to preserve framing in same-format passthrough + if (mode === STREAM_MODE.TRANSLATE && targetFormat === FORMATS.OPENAI_RESPONSES && trimmed.startsWith("event:")) { + currentOpenAIResponsesEvent = trimmed.slice(6).trim(); + } + // Passthrough mode: normalize and forward if (mode === STREAM_MODE.PASSTHROUGH) { let output; @@ -174,12 +185,33 @@ export function createSSEStream(options = {}) { const parsed = parseSSELine(trimmed, targetFormat); if (!parsed) continue; + // Responses API same-format passthrough: preserve event framing + track terminal state + const isOpenAIResponsesStream = targetFormat === FORMATS.OPENAI_RESPONSES; + const keepsOpenAIResponsesFormat = isOpenAIResponsesStream && sourceFormat === FORMATS.OPENAI_RESPONSES; + const openAIResponsesEventName = isOpenAIResponsesStream + ? getOpenAIResponsesEventName(currentOpenAIResponsesEvent, parsed) + : null; + + if (isOpenAIResponsesStream && isOpenAIResponsesTerminalEvent(openAIResponsesEventName, parsed)) { + openAIResponsesTerminalSeen = true; + } + // For Ollama: done=true is the final chunk with finish_reason/usage, must translate // For other formats: done=true is the [DONE] sentinel, skip if (parsed && parsed.done && targetFormat !== FORMATS.OLLAMA) { + // Synthesize response.failed if the Responses stream never sent a terminal event + if (keepsOpenAIResponsesFormat && !openAIResponsesTerminalSeen) { + const failedOutput = formatIncompleteOpenAIResponsesStreamFailure(); + reqLogger?.appendConvertedChunk?.(failedOutput); + controller.enqueue(sharedEncoder.encode(failedOutput)); + openAIResponsesTerminalSeen = true; + sseEmittedCount++; + } + const output = "data: [DONE]\n\n"; reqLogger?.appendConvertedChunk?.(output); controller.enqueue(sharedEncoder.encode(output)); + if (keepsOpenAIResponsesFormat) openAIResponsesDoneSent = true; continue; } @@ -224,6 +256,18 @@ export function createSSEStream(options = {}) { const extracted = extractUsage(parsed); if (extracted) state.usage = extracted; // Keep original usage for logging + // Responses same-format passthrough: re-emit with original event framing + if (keepsOpenAIResponsesFormat && openAIResponsesEventName) { + const output = formatSSE({ event: openAIResponsesEventName, data: parsed }, sourceFormat); + reqLogger?.appendConvertedChunk?.(output); + controller.enqueue(sharedEncoder.encode(output)); + currentOpenAIResponsesEvent = null; + sseEmittedCount++; + continue; + } + + currentOpenAIResponsesEvent = null; + // Translate: targetFormat -> openai -> sourceFormat const translated = translateResponse(targetFormat, sourceFormat, parsed, state); @@ -237,6 +281,7 @@ export function createSSEStream(options = {}) { if (translated?.length > 0) { for (const item of translated) { + if (item === null || item === undefined) continue; // Filter empty chunks if (!hasValuableContent(item, sourceFormat)) { continue; // Skip this empty chunk @@ -322,6 +367,7 @@ export function createSSEStream(options = {}) { if (translated?.length > 0) { for (const item of translated) { + if (item === null || item === undefined) continue; const output = formatSSE(item, sourceFormat); reqLogger?.appendConvertedChunk?.(output); controller.enqueue(sharedEncoder.encode(output)); @@ -341,15 +387,27 @@ export function createSSEStream(options = {}) { if (flushed?.length > 0) { for (const item of flushed) { + if (item === null || item === undefined) continue; const output = formatSSE(item, sourceFormat); reqLogger?.appendConvertedChunk?.(output); controller.enqueue(sharedEncoder.encode(output)); } } - const doneOutput = "data: [DONE]\n\n"; - reqLogger?.appendConvertedChunk?.(doneOutput); - controller.enqueue(sharedEncoder.encode(doneOutput)); + // Synthesize response.failed if a Responses passthrough stream never reached a terminal event + const keepsOpenAIResponsesFormat = targetFormat === FORMATS.OPENAI_RESPONSES && sourceFormat === FORMATS.OPENAI_RESPONSES; + if (keepsOpenAIResponsesFormat && !openAIResponsesTerminalSeen) { + const failedOutput = formatIncompleteOpenAIResponsesStreamFailure(); + reqLogger?.appendConvertedChunk?.(failedOutput); + controller.enqueue(sharedEncoder.encode(failedOutput)); + openAIResponsesTerminalSeen = true; + } + + if (!keepsOpenAIResponsesFormat || !openAIResponsesDoneSent) { + const doneOutput = "data: [DONE]\n\n"; + reqLogger?.appendConvertedChunk?.(doneOutput); + controller.enqueue(sharedEncoder.encode(doneOutput)); + } if (!hasValidUsage(state?.usage) && totalContentLength > 0) { state.usage = estimateUsage(body, totalContentLength, sourceFormat); diff --git a/open-sse/utils/streamHandler.js b/open-sse/utils/streamHandler.js index 567b59ac..35ef5e4a 100644 --- a/open-sse/utils/streamHandler.js +++ b/open-sse/utils/streamHandler.js @@ -94,13 +94,25 @@ export function createStreamController({ onDisconnect, onError, log, provider, m * for long periods while raw bytes still flow (e.g. Kiro EventStream * binary frames buffering, Claude reasoning streams). */ -export function createDisconnectAwareStream(transformStream, streamController) { +export function createDisconnectAwareStream(transformStream, streamController, onAbortTerminal = null) { const reader = transformStream.readable.getReader(); const writer = transformStream.writable.getWriter(); + let terminalEmitted = false; + + // Emit a synthesized terminal payload (e.g. Responses response.failed + [DONE]) once + const emitTerminal = (controller) => { + if (terminalEmitted || !onAbortTerminal) return; + terminalEmitted = true; + try { + const bytes = onAbortTerminal(); + if (bytes) controller.enqueue(bytes); + } catch { /* best-effort terminal */ } + }; return new ReadableStream({ async pull(controller) { if (!streamController.isConnected()) { + emitTerminal(controller); controller.close(); return; } @@ -135,17 +147,16 @@ export function createDisconnectAwareStream(transformStream, streamController) { code === "EPIPE" || code === "UND_ERR_SOCKET"; - if (!wasConnected || isNetworkClose) { - try { + // Graceful close on network/abort, or when a structured terminal is available + // (Responses passthrough prefers response.failed + [DONE] over a raw transport error) + try { + if (!wasConnected || isNetworkClose || onAbortTerminal) { + emitTerminal(controller); controller.close(); - } catch (e) { - // Stream might already be closed or cancelled - } - } else { - try { + } else { controller.error(error); - } catch (e) { /* already closed */ } - } + } + } catch (e) { /* already closed or cancelled */ } } }, @@ -173,7 +184,7 @@ export function createDisconnectAwareStream(transformStream, streamController) { * @param {TransformStream} transformStream - Transform stream for SSE * @param {object} streamController - Stream controller from createStreamController */ -export function pipeWithDisconnect(providerResponse, transformStream, streamController) { +export function pipeWithDisconnect(providerResponse, transformStream, streamController, onAbortTerminal = null) { let stallTimer = null; let chunkCount = 0; let totalBytes = 0; @@ -232,7 +243,8 @@ export function pipeWithDisconnect(providerResponse, transformStream, streamCont return createDisconnectAwareStream( { readable: transformedBody, writable: { getWriter: () => ({ abort: () => Promise.resolve() }) } }, - wrappedController + wrappedController, + onAbortTerminal ); } diff --git a/src/app/api/providers/[id]/test/testUtils.js b/src/app/api/providers/[id]/test/testUtils.js index bbc7b020..d0725d83 100644 --- a/src/app/api/providers/[id]/test/testUtils.js +++ b/src/app/api/providers/[id]/test/testUtils.js @@ -28,7 +28,7 @@ const OAUTH_TEST_CONFIG = { method: "POST", authHeader: "Authorization", authPrefix: "Bearer ", - extraHeaders: { "Content-Type": "application/json", "originator": "codex-cli", "User-Agent": "codex-cli/1.0.18 (macOS; arm64)" }, + extraHeaders: { "Content-Type": "application/json", "originator": "codex_cli_rs", "User-Agent": "codex_cli_rs/0.136.0" }, // Minimal invalid body — triggers fast 400 without consuming quota body: JSON.stringify({ model: "gpt-5.3-codex", input: [], stream: false, store: false }), // 400 (bad request) means auth succeeded; only 401/403 means token is bad diff --git a/tests/unit/image-generation.test.js b/tests/unit/image-generation.test.js index 4aca1de1..d74fea97 100644 --- a/tests/unit/image-generation.test.js +++ b/tests/unit/image-generation.test.js @@ -319,7 +319,7 @@ describe("handleImageGenerationCore", () => { headers: expect.objectContaining({ authorization: "Bearer codex-token", "chatgpt-account-id": "account-123", - version: "0.129.0", + version: "0.136.0", }), }) ); diff --git a/tests/unit/openai-responses-terminal-event.test.js b/tests/unit/openai-responses-terminal-event.test.js new file mode 100644 index 00000000..db2e3ad8 --- /dev/null +++ b/tests/unit/openai-responses-terminal-event.test.js @@ -0,0 +1,83 @@ +import { describe, expect, it } from "vitest"; + +import { FORMATS } from "../../open-sse/translator/formats.js"; +import { createSSETransformStreamWithLogger } from "../../open-sse/utils/stream.js"; + +async function runTransform(input) { + const encoder = new TextEncoder(); + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(input)); + controller.close(); + }, + }); + + const output = stream.pipeThrough( + createSSETransformStreamWithLogger( + FORMATS.OPENAI_RESPONSES, + FORMATS.OPENAI_RESPONSES, + "codex", + null, + null, + "gpt-5.5", + ), + ); + + const reader = output.getReader(); + const decoder = new TextDecoder(); + let text = ""; + + while (true) { + const { value, done } = await reader.read(); + if (done) break; + text += decoder.decode(value, { stream: true }); + } + + text += decoder.decode(); + return text; +} + +describe("OpenAI Responses streaming termination", () => { + it("emits a response.failed event when a Responses stream closes before a terminal event", async () => { + const output = await runTransform([ + `event: response.created`, + `data: ${JSON.stringify({ type: "response.created", response: { id: "resp_test", status: "in_progress" } })}`, + "", + `event: response.output_text.delta`, + `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "partial" })}`, + "", + ].join("\n")); + + expect(output).toContain("event: response.failed"); + expect(output).toContain('"type":"response.failed"'); + expect(output).not.toContain("data: null"); + expect(output).toContain("data: [DONE]"); + }); + + it("does not add response.failed when a Responses stream already completed", async () => { + const output = await runTransform([ + `event: response.completed`, + `data: ${JSON.stringify({ type: "response.completed", response: { id: "resp_test", status: "completed" } })}`, + "", + ].join("\n")); + + expect(output).toContain("event: response.completed"); + expect(output).not.toContain("event: response.failed"); + expect(output).not.toContain("data: null"); + expect(output).toContain("data: [DONE]"); + }); + + it("emits response.failed before DONE when a Responses stream sends DONE without a terminal event", async () => { + const output = await runTransform([ + `event: response.created`, + `data: ${JSON.stringify({ type: "response.created", response: { id: "resp_test", status: "in_progress" } })}`, + "", + "data: [DONE]", + "", + ].join("\n")); + + expect(output.indexOf("event: response.failed")).toBeLessThan(output.indexOf("data: [DONE]")); + expect(output.match(/data: \[DONE\]/g)).toHaveLength(1); + expect(output).not.toContain("data: null"); + }); +}); diff --git a/tests/unit/responses-abort-terminal.test.js b/tests/unit/responses-abort-terminal.test.js new file mode 100644 index 00000000..10af5848 --- /dev/null +++ b/tests/unit/responses-abort-terminal.test.js @@ -0,0 +1,72 @@ +import { describe, expect, it } from "vitest"; + +import { createDisconnectAwareStream } from "../../open-sse/utils/streamHandler.js"; +import { buildAbortedResponsesTerminalBytes } from "../../open-sse/utils/responsesStreamHelpers.js"; + +// Minimal stream controller stub +function makeController() { + let connected = true; + return { + signal: new AbortController().signal, + startTime: Date.now(), + isConnected: () => connected, + handleComplete: () => { connected = false; }, + handleError: () => { connected = false; }, + handleDisconnect: () => { connected = false; }, + abort: () => { connected = false; }, + }; +} + +async function readAll(stream) { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let text = ""; + while (true) { + const { value, done } = await reader.read(); + if (done) break; + text += decoder.decode(value, { stream: true }); + } + text += decoder.decode(); + return text; +} + +describe("Responses abort terminal synthesis", () => { + it("emits response.failed + [DONE] when upstream errors (abort/stall)", async () => { + // Upstream readable that errors mid-stream (simulates fetch abort on stall) + const upstream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("event: response.created\ndata: {}\n\n")); + controller.error(new Error("stream stall timeout")); + }, + }); + + const out = createDisconnectAwareStream( + { readable: upstream, writable: { getWriter: () => ({ abort: () => Promise.resolve() }) } }, + makeController(), + buildAbortedResponsesTerminalBytes + ); + + const text = await readAll(out); + expect(text).toContain("event: response.failed"); + expect(text).toContain("data: [DONE]"); + }); + + it("does not synthesize terminal for non-Responses streams (callback null)", async () => { + const upstream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("data: hi\n\n")); + controller.error(new Error("socket hang up")); + }, + }); + + const out = createDisconnectAwareStream( + { readable: upstream, writable: { getWriter: () => ({ abort: () => Promise.resolve() }) } }, + makeController(), + null + ); + + const text = await readAll(out); + expect(text).not.toContain("response.failed"); + expect(text).not.toContain("[DONE]"); + }); +});