diff --git a/changelog.d/fixes/13431-responses-post-keepalive-error-type.md b/changelog.d/fixes/13431-responses-post-keepalive-error-type.md new file mode 100644 index 0000000000..6bf0440598 --- /dev/null +++ b/changelog.d/fixes/13431-responses-post-keepalive-error-type.md @@ -0,0 +1 @@ +- **fix(sse):** frame post-keepalive `/v1/responses` stream errors with a top-level `type` field so Responses clients (Codex) surface the real upstream error instead of reporting "stream disconnected before completion" (#13431) — thanks @andrea-kingautomation diff --git a/open-sse/utils/earlyStreamKeepalive.ts b/open-sse/utils/earlyStreamKeepalive.ts index 0b7cdaab36..67b81e6f63 100644 --- a/open-sse/utils/earlyStreamKeepalive.ts +++ b/open-sse/utils/earlyStreamKeepalive.ts @@ -89,6 +89,44 @@ export const OPENAI_RESPONSES_ERROR_FRAME = ENCODER.encode( })}\n\n` ); +/** + * Reshapes an already-sanitized upstream error body into the Responses API + * convention (`{"type":"error",...}`) for the dynamic real-upstream-body branch + * of the slow path (#13431). The body reaching here is Chat-Completions-shaped + * (`{"error":{message,type,code}}`, the combo/handler failure convention) most of + * the time, but may also be a bare `{message}` or unparseable text — every shape + * must still produce a non-empty `message` so the client never sees an opaque + * frame (never crash the stream on a malformed body). + */ +function buildResponsesErrorDataLine(text: string): string { + const trimmed = text.trim(); + let parsed: Record | null = null; + if (trimmed) { + try { + const candidate = JSON.parse(trimmed); + if (candidate && typeof candidate === "object") parsed = candidate as Record; + } catch { + parsed = null; + } + } + const errorObj = + parsed && typeof parsed.error === "object" && parsed.error !== null + ? (parsed.error as Record) + : null; + const message = + (typeof errorObj?.message === "string" && errorObj.message) || + (typeof parsed?.message === "string" && parsed.message) || + trimmed || + "Upstream stream failed before completion."; + const code = (typeof errorObj?.code === "string" && errorObj.code) || null; + const param = (typeof errorObj?.param === "string" && errorObj.param) || null; + const extras = + parsed && typeof parsed.diagnostics === "object" && parsed.diagnostics !== null + ? { diagnostics: parsed.diagnostics } + : {}; + return JSON.stringify({ type: "error", code, message, param, ...extras }); +} + export type EarlyStreamKeepaliveOptions = { /** Wait this long for the handler before committing to a keepalive stream. */ thresholdMs?: number; @@ -168,11 +206,29 @@ export async function withEarlyStreamKeepalive( : null; const extraHeaders = options.extraHeaders ?? {}; const errorFrame = options.errorFrame ?? ERROR_FRAME; - // Single source of truth for whether THIS route's error framing uses a named SSE - // `event: error` line (Anthropic) or a plain `data:` line (OpenAI Chat Completions / - // Responses) — derived from errorFrame itself so the dynamic real-upstream-body case - // below stays consistent with the static default-message case without a second option. - const errorFrameUsesNamedEvent = new TextDecoder().decode(errorFrame).startsWith("event:"); + // Single source of truth for THIS route's error-framing convention, derived from + // errorFrame itself so the dynamic real-upstream-body case below stays consistent + // with the static default-message case without a second option. Three shapes exist: + // - "anthropic": named SSE `event: error` line (Anthropic /v1/messages). + // - "responses": plain `data:` line, discriminated by a top-level `type` field + // inside the JSON payload (OpenAI Responses API convention). + // - "chat": plain `data:` line, discriminated by a top-level `error` key + // (OpenAI Chat Completions convention) — the default/fallback. + const decodedErrorFrame = new TextDecoder().decode(errorFrame); + const errorFrameFormat: "anthropic" | "responses" | "chat" = decodedErrorFrame.startsWith( + "event:" + ) + ? "anthropic" + : (() => { + const dataLine = decodedErrorFrame.match(/^data: (.+)\n\n$/); + if (!dataLine) return "chat"; + try { + const parsed = JSON.parse(dataLine[1]); + return parsed && typeof parsed === "object" && "type" in parsed ? "responses" : "chat"; + } catch { + return "chat"; + } + })(); const correlationId = options.correlationId; const frameDecoder = correlationId ? new TextDecoder() : null; // Records every direct-to-client write EXCEPT the forwarded real response @@ -321,11 +377,14 @@ export async function withEarlyStreamKeepalive( // instead of forwarding raw JSON, which would be malformed SSE. const text = response.body ? await response.text().catch(() => "") : ""; const dataLine = - text.trim() || - JSON.stringify({ error: { message: "stream_error", type: "stream_error" } }); - const framed = errorFrameUsesNamedEvent - ? `event: error\ndata: ${dataLine}\n\n` - : `data: ${dataLine}\n\n`; + errorFrameFormat === "responses" + ? buildResponsesErrorDataLine(text) + : text.trim() || + JSON.stringify({ error: { message: "stream_error", type: "stream_error" } }); + const framed = + errorFrameFormat === "anthropic" + ? `event: error\ndata: ${dataLine}\n\n` + : `data: ${dataLine}\n\n`; const framedBytes = ENCODER.encode(framed); controller.enqueue(framedBytes); recordClientBytes(framedBytes); diff --git a/tests/unit/issue-13431-responses-post-keepalive-error-frame.test.ts b/tests/unit/issue-13431-responses-post-keepalive-error-frame.test.ts new file mode 100644 index 0000000000..50d2e4e07b --- /dev/null +++ b/tests/unit/issue-13431-responses-post-keepalive-error-frame.test.ts @@ -0,0 +1,172 @@ +/** + * Regression test for #13431. + * + * `withEarlyStreamKeepalive`'s dynamic real-upstream-body branch + * (`open-sse/utils/earlyStreamKeepalive.ts`) only distinguished Anthropic's named + * `event: error` framing from a plain `data:` line. It did not distinguish Chat + * Completions' `data: {"error":...}` shape from Responses' `data: {"type":"error",...}` + * shape, so on `/v1/responses` the raw upstream body (Chat-Completions-shaped) went out + * untouched, with no top-level `type` field. Responses clients (openai-python's Responses + * stream iterator, Codex's own SSE parser) dispatch on `type` and silently drop a frame + * without it, so the stream ends with no `response.completed`/`response.failed` and the + * client reports "stream disconnected before completion" instead of the real upstream + * error. + */ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { + withEarlyStreamKeepalive, + OPENAI_RESPONSES_ERROR_FRAME, + OPENAI_CHAT_ERROR_FRAME, + ANTHROPIC_PING_FRAME, +} from "../../open-sse/utils/earlyStreamKeepalive.ts"; + +async function readAll(response: Response): Promise { + const reader = response.body!.getReader(); + const decoder = new TextDecoder(); + let out = ""; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + out += decoder.decode(value); + } + return out; +} + +function lastDataPayload(body: string): Record { + const dataLines = [...body.matchAll(/^data: (.+)$/gm)].map((m) => m[1]); + return JSON.parse(dataLines[dataLines.length - 1]); +} + +test("Responses route: post-keepalive JSON error body must carry a `type` field (#13431)", async () => { + // Shape actually produced by combo failure (Chat-Completions-shaped: top-level + // `error` key, no `type` discriminator) — this is the real body from the issue. + const upstreamErrorBody = JSON.stringify({ + error: { + message: 'Unknown name "encrypted" ... Cannot find field.', + type: "invalid_request_error", + code: "bad_request", + }, + diagnostics: { attempted: 9, terminalReason: "[400]: ..." }, + }); + + const slowFail = new Promise((resolve) => { + setTimeout( + () => + resolve( + new Response(upstreamErrorBody, { + status: 400, + headers: { "Content-Type": "application/json" }, + }) + ), + 80 + ); + }); + + const result = await withEarlyStreamKeepalive(slowFail, { + thresholdMs: 20, + intervalMs: 20, + errorFrame: OPENAI_RESPONSES_ERROR_FRAME, // exactly what src/app/api/v1/responses/route.ts passes + }); + + assert.equal(result.status, 200, "already committed to 200 SSE before the error surfaced"); + + const lastPayload = lastDataPayload(await readAll(result)); + + assert.ok( + typeof lastPayload.type === "string" && lastPayload.type.length > 0, + `Responses API events must be discriminated by a top-level \`type\` field; ` + + `got ${JSON.stringify(lastPayload)} — a Responses client (Codex) drops any ` + + `frame without \`type\` and reports "stream disconnected before completion" ` + + `instead of surfacing the real upstream error.` + ); + assert.equal(lastPayload.type, "error"); + assert.equal(lastPayload.message, 'Unknown name "encrypted" ... Cannot find field.'); + assert.equal(lastPayload.code, "bad_request"); +}); + +test("Responses route: non-JSON/empty post-keepalive error body falls back to a safe `type:error` frame (#13431)", async () => { + const slowFail = new Promise((resolve) => { + setTimeout( + () => + resolve( + new Response("not json at all", { + status: 502, + headers: { "Content-Type": "text/plain" }, + }) + ), + 80 + ); + }); + + const result = await withEarlyStreamKeepalive(slowFail, { + thresholdMs: 20, + intervalMs: 20, + errorFrame: OPENAI_RESPONSES_ERROR_FRAME, + }); + + const lastPayload = lastDataPayload(await readAll(result)); + + assert.equal(lastPayload.type, "error"); + assert.ok( + typeof lastPayload.message === "string" && lastPayload.message.length > 0, + `fallback frame must never be opaque/empty; got ${JSON.stringify(lastPayload)}` + ); +}); + +test("Chat Completions route: post-keepalive JSON error body stays verbatim pass-through (regression guard) (#13431)", async () => { + const upstreamErrorBody = JSON.stringify({ + error: { message: "boom", type: "invalid_request_error", code: "bad_request" }, + }); + + const slowFail = new Promise((resolve) => { + setTimeout( + () => + resolve( + new Response(upstreamErrorBody, { + status: 400, + headers: { "Content-Type": "application/json" }, + }) + ), + 80 + ); + }); + + const result = await withEarlyStreamKeepalive(slowFail, { + thresholdMs: 20, + intervalMs: 20, + errorFrame: OPENAI_CHAT_ERROR_FRAME, + }); + + const lastPayload = lastDataPayload(await readAll(result)); + + // Unchanged: verbatim pass-through, top-level `error` key, no reshaping. + assert.equal(lastPayload.type, undefined); + assert.equal((lastPayload as { error: { message: string } }).error.message, "boom"); +}); + +test("Anthropic /v1/messages route: post-keepalive named event: error framing stays unaffected (regression guard) (#13431)", async () => { + const slowFail = new Promise((resolve) => { + setTimeout( + () => + resolve( + new Response(JSON.stringify({ type: "error", error: { message: "boom" } }), { + status: 400, + headers: { "Content-Type": "application/json" }, + }) + ), + 80 + ); + }); + + const result = await withEarlyStreamKeepalive(slowFail, { + thresholdMs: 20, + intervalMs: 20, + keepaliveFrame: ANTHROPIC_PING_FRAME, + // default errorFrame (Anthropic `event: error`) is used when omitted. + }); + + const body = await readAll(result); + assert.match(body, /^event: error\n/m, "Anthropic path must keep its named SSE event line"); +});