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

fix(codex): harden streaming timeouts + Responses terminal events

Browse Source 
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 <jonathanli12@users.noreply.github.com>
Co-authored-by: rifuki <rifuki@users.noreply.github.com>
Co-authored-by: nguyenha935 <nguyenha935@users.noreply.github.com>
Co-authored-by: trananhtung <trananhtung@users.noreply.github.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
decolua
2026-06-06 16:07:43 +07:00
parent f161b295a5
commit 9caea88528
15 changed files with 311 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
open-sse/config/providers.js
View File

@@ -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"

4
open-sse/config/runtimeConfig.js
View File

@@ -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;

7
open-sse/executors/base.js
View File

@@ -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,

3
open-sse/handlers/chatCore/sseToJsonHandler.js
View File

@@ -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",

7
open-sse/handlers/chatCore/streamingHandler.js
View File

@@ -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({

4
open-sse/handlers/imageProviders/codex.js
View File

@@ -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";

2
open-sse/transformer/streamToJsonConverter.js
View File

@@ -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;

2
open-sse/translator/response/openai-responses.js
View File

@@ -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") {

49
open-sse/utils/responsesStreamHelpers.js Normal file
View File

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

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

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

36
open-sse/utils/streamHandler.js
View File

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

2
src/app/api/providers/[id]/test/testUtils.js
View File

@@ -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

2
tests/unit/image-generation.test.js
View File

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

83
tests/unit/openai-responses-terminal-event.test.js Normal file
View File

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

72
tests/unit/responses-abort-terminal.test.js Normal file
View File

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