fix(sse): frame post-keepalive /v1/responses stream errors with a type field (#13431) (#13785)

Merged in the 2026-09-16 sweep of the maintainer's own open PRs, at the owner's explicit instruction. No push was made to the PR branch: the merge took the head as the owning session left it (verified OPEN, non-draft and MERGEABLE against the release tip immediately before merging).

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

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<string, unknown> | null = null;
+  if (trimmed) {
+    try {
+      const candidate = JSON.parse(trimmed);
+      if (candidate && typeof candidate === "object") parsed = candidate as Record<string, unknown>;
+    } catch {
+      parsed = null;
+    }
+  }
+  const errorObj =
+    parsed && typeof parsed.error === "object" && parsed.error !== null
+      ? (parsed.error as Record<string, unknown>)
+      : 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);

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<string> {
+  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<string, unknown> {
+  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<Response>((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<Response>((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<Response>((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<Response>((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");
+});