Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- fix(sse): give synthesized Responses API keepalive and failure frames a real `sequence_number` (#14330)
23 changes: 10 additions & 13 deletions open-sse/executors/codex.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -511,14 +512,11 @@ function toCodexResponseFailedEvent(parsed: Record<string, unknown>): Record<str

if (statusCode !== null) error.status_code = statusCode;

return {
type: "response.failed",
response: {
id: typeof response?.id === "string" ? response.id : null,
status: "failed",
error,
},
};
return buildSyntheticResponsesFailedEvent({
id: typeof response?.id === "string" ? response.id : null,
status: "failed",
error,
});
}

// Drop non-standard `codex.*` SSE events (notably `codex.rate_limits`) from
Expand Down Expand Up @@ -979,14 +977,13 @@ export class CodexExecutor extends BaseExecutor {
if (closed) return;
nextInput.log?.warn?.("CODEX", `WebSocket stream failed (${code}): ${message}`);
const controller = streamController;
const payload = JSON.stringify({
type: "response.failed",
response: {
const payload = JSON.stringify(
buildSyntheticResponsesFailedEvent({
id: null,
status: "failed",
error: projectCodexPublicError({ status: 502, code, type: "provider_error" }),
},
});
})
);
try {
controller?.enqueue(encoder.encode(`event: response.failed\ndata: ${payload}\n\n`));
} catch {
Expand Down
5 changes: 5 additions & 0 deletions open-sse/utils/diagnostics.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@

import { sanitizeErrorMessage } from "./error.ts";
import { classifyFakeSuccessBody } from "../services/errorClassifier.ts";
import { SYNTHETIC_RESPONSES_SEQUENCE_NUMBER } from "./responsesSequence.ts";

// ── Types ────────────────────────────────────────────────────────────────────

Expand Down Expand Up @@ -146,6 +147,10 @@ export function synthResponsesFailure(reason?: MalformedReason): string {
);
const event = {
type: "response.failed",
// #14330: this frame is synthesized outside the real per-stream sequence
// counter, so it uses the shared synthetic seed instead of omitting the
// required field — a strict Responses decoder aborts without it.
sequence_number: SYNTHETIC_RESPONSES_SEQUENCE_NUMBER,
response: {
id: null,
status: "failed",
Expand Down
16 changes: 14 additions & 2 deletions open-sse/utils/earlyStreamKeepalive.ts
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
*/

import { recordEarlyKeepaliveBytes } from "./earlyKeepaliveByteBuffer.ts";
import { SYNTHETIC_RESPONSES_SEQUENCE_NUMBER } from "./responsesSequence.ts";

const ENCODER = new TextEncoder();
const KEEPALIVE_FRAME = ENCODER.encode(": keepalive\n\n");
Expand Down Expand Up @@ -86,7 +87,10 @@ export const OPENAI_RESPONSES_ERROR_FRAME = ENCODER.encode(
code: null,
message: "Upstream stream failed before completion.",
param: null,
sequence_number: 0,
// #14330: was hardcoded to 0, colliding with the real per-stream emitter's
// first event (also numbered 1 from its own `state.seq` base of 0) — this
// frame is synthesized outside that counter, so it uses the shared seed.
sequence_number: SYNTHETIC_RESPONSES_SEQUENCE_NUMBER,
})}\n\n`
);

Expand Down Expand Up @@ -125,7 +129,15 @@ function buildResponsesErrorDataLine(text: string): string {
parsed && typeof parsed.diagnostics === "object" && parsed.diagnostics !== null
? { diagnostics: parsed.diagnostics }
: {};
return JSON.stringify({ type: "error", code, message, param, sequence_number: 0, ...extras });
return JSON.stringify({
type: "error",
code,
message,
param,
// #14330: was hardcoded to 0 — see OPENAI_RESPONSES_ERROR_FRAME above.
sequence_number: SYNTHETIC_RESPONSES_SEQUENCE_NUMBER,
...extras,
});
}

export type EarlyStreamKeepaliveOptions = {
Expand Down
27 changes: 27 additions & 0 deletions open-sse/utils/responsesSequence.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
/**
* @file responsesSequence.ts
* @description Shared `sequence_number` convention for Responses API frames that are
* synthesized OUTSIDE the real per-stream event emitter (open-sse/translator/response/
* openai-responses/eventEmitter.ts, whose `state.seq` starts at 0 so the first real event
* is numbered 1).
*
* These synthesized sites (startup/mid-stream keepalive, early-stream and translated-error
* failure frames, Codex WebSocket failure frames) have no access to that live counter —
* threading a shared counter across four independent modules would risk colliding with a
* real event's own sequence_number (#14330 risk notes). Per the Responses event schema, a
* strict decoder only requires the field to be a monotonic-looking number starting at 1 (it
* never validates it against a prior stream), so every synthesized site uses this same
* seed: 1, never the previous default of 0, which collided with the real first event.
*/
export const SYNTHETIC_RESPONSES_SEQUENCE_NUMBER = 1;

/** Wraps a `response.failed` payload's `response` object with the synthetic seed. */
export function buildSyntheticResponsesFailedEvent(
response: Record<string, unknown>
): Record<string, unknown> {
return {
type: "response.failed",
sequence_number: SYNTHETIC_RESPONSES_SEQUENCE_NUMBER,
response,
};
}
11 changes: 10 additions & 1 deletion open-sse/utils/sseHeartbeat.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down
5 changes: 4 additions & 1 deletion open-sse/utils/streamErrorFormat.ts
Original file line number Diff line number Diff line change
@@ -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.
Expand Down Expand Up @@ -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`;
}
Expand Down
12 changes: 10 additions & 2 deletions tests/unit/early-stream-keepalive.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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|✨/);
}
Expand Down Expand Up @@ -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");
Expand Down
80 changes: 80 additions & 0 deletions tests/unit/responses-sse-frame-schema-14330.test.ts
Original file line number Diff line number Diff line change
@@ -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<string, unknown> {
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"
);
});
Loading
Loading