diff --git a/changelog.d/fixes/14330-responses-keepalive-frames-sequence-number.md b/changelog.d/fixes/14330-responses-keepalive-frames-sequence-number.md new file mode 100644 index 000000000000..13ac6fbaa4de --- /dev/null +++ b/changelog.d/fixes/14330-responses-keepalive-frames-sequence-number.md @@ -0,0 +1 @@ +- fix(sse): give synthesized Responses API keepalive and failure frames a real `sequence_number` (#14330) diff --git a/open-sse/executors/codex.ts b/open-sse/executors/codex.ts index 5e6c4175ac87..aef3d330a864 100644 --- a/open-sse/executors/codex.ts +++ b/open-sse/executors/codex.ts @@ -44,6 +44,7 @@ import { getThinkingBudgetConfig, ThinkingMode } from "../services/thinkingBudge import { CORS_HEADERS } from "../utils/cors.ts"; import { projectCodexPublicError } from "../utils/codexPublicError.ts"; import { errorResponse } from "../utils/error.ts"; +import { buildSyntheticResponsesFailedEvent } from "../utils/responsesSequence.ts"; import { normalizeCodexResponsesInput } from "../utils/responsesInputNormalization.ts"; import * as prl from "../utils/providerRequestLogging.ts"; import { createRequire } from "module"; @@ -511,14 +512,11 @@ function toCodexResponseFailedEvent(parsed: Record): Record +): Record { + return { + type: "response.failed", + sequence_number: SYNTHETIC_RESPONSES_SEQUENCE_NUMBER, + response, + }; +} diff --git a/open-sse/utils/sseHeartbeat.ts b/open-sse/utils/sseHeartbeat.ts index 62574bd2e2d0..2763ba0e5e83 100644 --- a/open-sse/utils/sseHeartbeat.ts +++ b/open-sse/utils/sseHeartbeat.ts @@ -5,8 +5,17 @@ * @changes * - [2026-07-28] [Cursor Grok 4.5] - Brand-neutral default OpenAI keepalive id/model */ +import { SYNTHETIC_RESPONSES_SEQUENCE_NUMBER } from "./responsesSequence.ts"; + const HEARTBEAT_ENCODER = new TextEncoder(); -const OPENAI_RESPONSES_IN_PROGRESS_PAYLOAD = 'data: {"type":"response.in_progress"}\n\n'; +// #14330: a bare {"type":"response.in_progress"} frame has no `sequence_number` or +// `response` object, so a strict Responses decoder (openai-python, Codex/OpenCode/Grok +// CLIs) aborts on it. Every typed Responses event requires both fields. +const OPENAI_RESPONSES_IN_PROGRESS_PAYLOAD = `data: ${JSON.stringify({ + type: "response.in_progress", + sequence_number: SYNTHETIC_RESPONSES_SEQUENCE_NUMBER, + response: { id: null, status: "in_progress" }, +})}\n\n`; export const DEFAULT_SSE_HEARTBEAT_INTERVAL_MS = 15_000; diff --git a/open-sse/utils/streamErrorFormat.ts b/open-sse/utils/streamErrorFormat.ts index 05a065a864b5..47979fcfcdbb 100644 --- a/open-sse/utils/streamErrorFormat.ts +++ b/open-sse/utils/streamErrorFormat.ts @@ -1,6 +1,7 @@ import { FORMATS } from "../translator/formats.ts"; import { buildErrorBody, sanitizeErrorMessage } from "./error.ts"; import { projectResponsesFailureOutput } from "./responsesFailureOutput.ts"; +import { SYNTHETIC_RESPONSES_SEQUENCE_NUMBER } from "./responsesSequence.ts"; /** * Upstream stream-failure normalization + client-format error framing. @@ -303,7 +304,9 @@ export function formatTranslatedStreamError(payload: unknown, sourceFormat?: str error: errorBody.error, output: [], }, - sequence_number: 0, + // #14330: was hardcoded to 0 — see OPENAI_RESPONSES_ERROR_FRAME's comment in + // earlyStreamKeepalive.ts for why this collided with the real emitter's first event. + sequence_number: SYNTHETIC_RESPONSES_SEQUENCE_NUMBER, }; return `event: response.failed\ndata: ${JSON.stringify(failed)}\n\n`; } diff --git a/tests/unit/early-stream-keepalive.test.ts b/tests/unit/early-stream-keepalive.test.ts index df7450430b63..5e0668ca9bd4 100644 --- a/tests/unit/early-stream-keepalive.test.ts +++ b/tests/unit/early-stream-keepalive.test.ts @@ -196,7 +196,13 @@ test("slow Responses handler uses comments plus sparse in_progress events", asyn const body = await readAll(result); const frames = body.split("\n\n").filter(Boolean); const earlyFrames = frames.slice(0, -1); - assert.equal(earlyFrames[0], 'data: {"type":"response.in_progress"}'); + // #14330: the frame now carries a required `sequence_number` and `response` + // object so a strict Responses decoder does not abort on it. + assert.deepEqual(JSON.parse(earlyFrames[0].slice("data: ".length)), { + type: "response.in_progress", + sequence_number: 1, + response: { id: null, status: "in_progress" }, + }); assert.ok( earlyFrames.some((frame) => frame === ": keepalive"), "transport ticks must remain lightweight SSE comments" @@ -206,6 +212,8 @@ test("slow Responses handler uses comments plus sparse in_progress events", asyn for (const frame of applicationFrames) { assert.deepEqual(JSON.parse(frame.slice("data: ".length)), { type: "response.in_progress", + sequence_number: 1, + response: { id: null, status: "in_progress" }, }); assert.doesNotMatch(frame, /output_item|reasoning|✨/); } @@ -240,7 +248,7 @@ test("a correlationId records the startup frame and keepalive ticks, but not the const recorded = takeEarlyKeepaliveBytes(correlationId).join(""); assert.match( recorded, - /data: {"type":"response\.in_progress"}/, + /data: {"type":"response\.in_progress","sequence_number":1,"response":/, "startup frame must be recorded" ); assert.match(recorded, /: keepalive/, "transport heartbeat must be recorded"); diff --git a/tests/unit/responses-sse-frame-schema-14330.test.ts b/tests/unit/responses-sse-frame-schema-14330.test.ts new file mode 100644 index 000000000000..9d5423f20266 --- /dev/null +++ b/tests/unit/responses-sse-frame-schema-14330.test.ts @@ -0,0 +1,80 @@ +// Regression test for issue #14330: synthesized keepalive and failure frames on +// /v1/responses are not valid Responses API events (missing or duplicated +// `sequence_number`). Each SSE `data:` payload on /v1/responses must satisfy +// the documented Responses event union, which requires a numeric +// `sequence_number` on every typed event. +import { test } from "node:test"; +import assert from "node:assert/strict"; + +import { OPENAI_RESPONSES_IN_PROGRESS_FRAME } from "../../open-sse/utils/sseHeartbeat.ts"; +import { synthResponsesFailure } from "../../open-sse/utils/diagnostics.ts"; +import { OPENAI_RESPONSES_ERROR_FRAME } from "../../open-sse/utils/earlyStreamKeepalive.ts"; +import { formatTranslatedStreamError } from "../../open-sse/utils/streamErrorFormat.ts"; + +function extractDataLine(sseText: string): Record { + const dataLine = sseText.split("\n").find((line) => line.startsWith("data: ")); + assert.ok(dataLine, `expected a data: line in frame, got: ${JSON.stringify(sseText)}`); + return JSON.parse(dataLine!.slice("data: ".length)); +} + +test("#14330: OPENAI_RESPONSES_IN_PROGRESS_FRAME (startup + mid-stream keepalive) is a valid response.in_progress event", () => { + const text = new TextDecoder().decode(OPENAI_RESPONSES_IN_PROGRESS_FRAME); + const payload = extractDataLine(text); + + assert.equal(payload.type, "response.in_progress"); + assert.equal( + typeof payload.sequence_number, + "number", + "response.in_progress frame is missing sequence_number — a strict client " + + "(openai-python, etc.) fails to decode this frame and aborts the stream" + ); + assert.equal( + typeof payload.response, + "object", + "response.in_progress frame is missing the required `response` object" + ); +}); + +test("#14330: synthResponsesFailure() produces a response.failed event with a numeric sequence_number", () => { + const text = synthResponsesFailure("timeout"); + const payload = extractDataLine(text); + + assert.equal(payload.type, "response.failed"); + assert.equal( + typeof payload.sequence_number, + "number", + "synthResponsesFailure() omits sequence_number — a strict client fails to " + + "decode the failure frame and never sees the real upstream error" + ); +}); + +test("#14330: OPENAI_RESPONSES_ERROR_FRAME does not hardcode a duplicate sequence_number: 0", () => { + const text = new TextDecoder().decode(OPENAI_RESPONSES_ERROR_FRAME); + const payload = extractDataLine(text); + + assert.equal(payload.type, "error"); + assert.equal(typeof payload.sequence_number, "number"); + assert.notEqual( + payload.sequence_number, + 0, + "OPENAI_RESPONSES_ERROR_FRAME hardcodes sequence_number: 0 instead of " + + "continuing the real per-stream sequence" + ); +}); + +test("#14330: formatTranslatedStreamError() (Responses format) does not hardcode a duplicate sequence_number: 0", () => { + const text = formatTranslatedStreamError( + { status: 502, message: "boom", code: "stream_error", type: "server_error" }, + "openai-responses" + ); + const payload = extractDataLine(text); + + assert.equal(payload.type, "response.failed"); + assert.equal(typeof payload.sequence_number, "number"); + assert.notEqual( + payload.sequence_number, + 0, + "formatTranslatedStreamError() hardcodes sequence_number: 0 for the " + + "response.failed frame instead of continuing the real per-stream sequence" + ); +}); diff --git a/tests/unit/sse-heartbeat.test.ts b/tests/unit/sse-heartbeat.test.ts index 2f26b295b72b..420154025df7 100644 --- a/tests/unit/sse-heartbeat.test.ts +++ b/tests/unit/sse-heartbeat.test.ts @@ -51,50 +51,54 @@ async function withSseCommentsOn(fn) { } test("createSseHeartbeatTransform emits SSE comments while preserving stream output", async () => { - await withSseCommentsOn(() => withFakeIntervals(async (intervals) => { - const transform = createSseHeartbeatTransform({ intervalMs: 250 }); - const writer = transform.writable.getWriter(); - const reader = transform.readable.getReader(); - const emitted = []; - const pump = (async () => { - while (true) { - const { value, done } = await reader.read(); - if (done) break; - emitted.push(decodeChunk(value)); - } - })(); + await withSseCommentsOn(() => + withFakeIntervals(async (intervals) => { + const transform = createSseHeartbeatTransform({ intervalMs: 250 }); + const writer = transform.writable.getWriter(); + const reader = transform.readable.getReader(); + const emitted = []; + const pump = (async () => { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + emitted.push(decodeChunk(value)); + } + })(); - await writer.write(new TextEncoder().encode('data: {"chunk":"one"}\n\n')); + await writer.write(new TextEncoder().encode('data: {"chunk":"one"}\n\n')); - assert.equal(intervals.length, 1); - assert.equal(intervals[0].delay, 250); + assert.equal(intervals.length, 1); + assert.equal(intervals[0].delay, 250); - await intervals[0].callback(...intervals[0].args); - await writer.close(); - await pump; + await intervals[0].callback(...intervals[0].args); + await writer.close(); + await pump; - assert.equal(emitted[0], 'data: {"chunk":"one"}\n\n'); - assert.match(emitted[1], /^: keepalive /); - assert.equal(intervals[0].cleared, true); - })); + assert.equal(emitted[0], 'data: {"chunk":"one"}\n\n'); + assert.match(emitted[1], /^: keepalive /); + assert.equal(intervals[0].cleared, true); + }) + ); }); test("createSseHeartbeatTransform clears the interval when aborted", async () => { - await withSseCommentsOn(() => withFakeIntervals(async (intervals) => { - const controller = new AbortController(); - const transform = createSseHeartbeatTransform({ signal: controller.signal }); - const reader = transform.readable.getReader(); - const writer = transform.writable.getWriter(); + await withSseCommentsOn(() => + withFakeIntervals(async (intervals) => { + const controller = new AbortController(); + const transform = createSseHeartbeatTransform({ signal: controller.signal }); + const reader = transform.readable.getReader(); + const writer = transform.writable.getWriter(); - assert.equal(intervals.length, 1); - assert.equal(intervals[0].cleared, false); + assert.equal(intervals.length, 1); + assert.equal(intervals[0].cleared, false); - controller.abort(); - assert.equal(intervals[0].cleared, true); + controller.abort(); + assert.equal(intervals[0].cleared, true); - await writer.close(); - await reader.cancel(); - })); + await writer.close(); + await reader.cancel(); + }) + ); }); const { shapeForClientFormat } = await import("../../open-sse/utils/sseHeartbeat.ts"); @@ -172,30 +176,37 @@ test("shape: openai-responses-in-progress emits response.in_progress data event" await writer.close(); await pump; - assert.equal(emitted[0], 'data: {"type":"response.in_progress"}\n\n'); + // #14330: the frame now carries a required `sequence_number` and `response` + // object so a strict Responses decoder does not abort on it. + assert.equal( + emitted[0], + 'data: {"type":"response.in_progress","sequence_number":1,"response":{"id":null,"status":"in_progress"}}\n\n' + ); }); }); test("shape default is comment (back-compat)", async () => { - await withSseCommentsOn(() => withFakeIntervals(async (intervals) => { - const transform = createSseHeartbeatTransform({ intervalMs: 100 }); - const writer = transform.writable.getWriter(); - const reader = transform.readable.getReader(); - const emitted = []; - const pump = (async () => { - while (true) { - const { value, done } = await reader.read(); - if (done) break; - emitted.push(decodeChunk(value)); - } - })(); + await withSseCommentsOn(() => + withFakeIntervals(async (intervals) => { + const transform = createSseHeartbeatTransform({ intervalMs: 100 }); + const writer = transform.writable.getWriter(); + const reader = transform.readable.getReader(); + const emitted = []; + const pump = (async () => { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + emitted.push(decodeChunk(value)); + } + })(); - await intervals[0].callback(...intervals[0].args); - await writer.close(); - await pump; + await intervals[0].callback(...intervals[0].args); + await writer.close(); + await pump; - assert.match(emitted[0], /^: keepalive /); - })); + assert.match(emitted[0], /^: keepalive /); + }) + ); }); test("intervalMs <= 0 returns passthrough (no setInterval, no heartbeat)", async () => { @@ -235,34 +246,36 @@ test("shapeForClientFormat maps formats correctly", () => { test("no shape collides with stream.ts event: keepalive strip regex", async () => { const shapes = ["comment", "anthropic-ping", "openai-chunk", "openai-responses-in-progress"]; for (const shape of shapes) { - await withSseCommentsOn(() => withFakeIntervals(async (intervals) => { - const transform = createSseHeartbeatTransform({ - intervalMs: 100, - shape, - }); - const writer = transform.writable.getWriter(); - const reader = transform.readable.getReader(); - const emitted = []; - const pump = (async () => { - while (true) { - const { value, done } = await reader.read(); - if (done) break; - emitted.push(decodeChunk(value)); + await withSseCommentsOn(() => + withFakeIntervals(async (intervals) => { + const transform = createSseHeartbeatTransform({ + intervalMs: 100, + shape, + }); + const writer = transform.writable.getWriter(); + const reader = transform.readable.getReader(); + const emitted = []; + const pump = (async () => { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + emitted.push(decodeChunk(value)); + } + })(); + + await intervals[0].callback(...intervals[0].args); + await writer.close(); + await pump; + + const lines = emitted[0].split("\n"); + for (const line of lines) { + assert.doesNotMatch( + line.trim(), + /^event:\s*keepalive\b/i, + `shape ${shape} produced forbidden line: ${line}` + ); } - })(); - - await intervals[0].callback(...intervals[0].args); - await writer.close(); - await pump; - - const lines = emitted[0].split("\n"); - for (const line of lines) { - assert.doesNotMatch( - line.trim(), - /^event:\s*keepalive\b/i, - `shape ${shape} produced forbidden line: ${line}` - ); - } - })); + }) + ); } }); diff --git a/tests/unit/sseHeartbeat.test.ts b/tests/unit/sseHeartbeat.test.ts index d7dc19f3d97b..097a4ecf14c2 100644 --- a/tests/unit/sseHeartbeat.test.ts +++ b/tests/unit/sseHeartbeat.test.ts @@ -91,6 +91,10 @@ test("comment opt-out suppresses only COMMENT heartbeats, not data-event heartbe assert.match(anthropic, /event: ping\ndata: \{"type":"ping"\}/); const responses = await collectHeartbeatOutput(HEARTBEAT_SHAPES.OPENAI_RESPONSES_IN_PROGRESS); - assert.match(responses, /data: \{"type":"response\.in_progress"\}/); + // #14330: the frame now carries a required `sequence_number` and `response` object. + assert.match( + responses, + /data: \{"type":"response\.in_progress","sequence_number":1,"response":/ + ); }); });