diff --git a/changelog.d/fixes/14632-responses-error-frame-status.md b/changelog.d/fixes/14632-responses-error-frame-status.md new file mode 100644 index 000000000000..7806831f4845 --- /dev/null +++ b/changelog.d/fixes/14632-responses-error-frame-status.md @@ -0,0 +1 @@ +- **fix(sse):** `/v1/responses` error frames sent after the early keepalive committed HTTP 200 now carry `status_code`, `error_type` and `retry_after_seconds`, so clients can tell permanent from transient failures ([#14632](https://github.com/diegosouzapw/OmniRoute/pull/14632)) — thanks @QuangBlue diff --git a/open-sse/utils/earlyStreamKeepalive.ts b/open-sse/utils/earlyStreamKeepalive.ts index 3007ad232e86..684aa1281aed 100644 --- a/open-sse/utils/earlyStreamKeepalive.ts +++ b/open-sse/utils/earlyStreamKeepalive.ts @@ -94,6 +94,29 @@ export const OPENAI_RESPONSES_ERROR_FRAME = ENCODER.encode( })}\n\n` ); +const MAX_RETRY_AFTER_SECONDS = 3600; + +/** + * Seconds a client should wait before retrying, from the handler's error response: + * `Retry-After` (delta-seconds or HTTP-date) first, then OmniRoute's + * `x-omniroute-retry-after-seconds`. Clamped to 0..3600; null when neither is usable. + */ +function readRetryAfterSeconds(headers: Headers): number | null { + for (const name of ["retry-after", "x-omniroute-retry-after-seconds"]) { + const raw = headers.get(name)?.trim(); + if (!raw) continue; + let seconds: number | null = null; + if (/^\d{1,10}$/.test(raw)) { + seconds = Number(raw); + } else { + const at = Date.parse(raw); + if (Number.isFinite(at)) seconds = Math.ceil((at - Date.now()) / 1000); + } + if (seconds !== null) return Math.min(Math.max(seconds, 0), MAX_RETRY_AFTER_SECONDS); + } + return null; +} + /** * Reshapes an already-sanitized upstream error body into the Responses API * convention (`{"type":"error",...}`) for the dynamic real-upstream-body branch @@ -103,7 +126,10 @@ export const OPENAI_RESPONSES_ERROR_FRAME = ENCODER.encode( * 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 { +function buildResponsesErrorDataLine( + text: string, + meta: { status: number; retryAfterSeconds: number | null } +): string { const trimmed = text.trim(); let parsed: Record | null = null; if (trimmed) { @@ -125,6 +151,15 @@ function buildResponsesErrorDataLine(text: string): string { "Upstream stream failed before completion."; const code = (typeof errorObj?.code === "string" && errorObj.code) || null; const param = (typeof errorObj?.param === "string" && errorObj.param) || null; + // The HTTP status and retry hint are already lost once the stream committed to 200; + // carry them in-band so clients can still tell permanent from transient failures. + // `error_type` (not `type`): top-level `type` is the Responses event discriminator. + const errorType = typeof errorObj?.type === "string" && errorObj.type ? errorObj.type : null; + const statusFields = { + status_code: meta.status, + ...(errorType ? { error_type: errorType } : {}), + ...(meta.retryAfterSeconds !== null ? { retry_after_seconds: meta.retryAfterSeconds } : {}), + }; const extras = parsed && typeof parsed.diagnostics === "object" && parsed.diagnostics !== null ? { diagnostics: parsed.diagnostics } @@ -136,6 +171,7 @@ function buildResponsesErrorDataLine(text: string): string { param, // #14330: was hardcoded to 0 — see OPENAI_RESPONSES_ERROR_FRAME above. sequence_number: SYNTHETIC_RESPONSES_SEQUENCE_NUMBER, + ...statusFields, ...extras, }); } @@ -396,7 +432,10 @@ export async function withEarlyStreamKeepalive( const text = response.body ? await response.text().catch(() => "") : ""; const dataLine = errorFrameFormat === "responses" - ? buildResponsesErrorDataLine(text) + ? buildResponsesErrorDataLine(text, { + status: response.status, + retryAfterSeconds: readRetryAfterSeconds(response.headers), + }) : text.trim() || JSON.stringify({ error: { message: "stream_error", type: "stream_error" } }); const framed = diff --git a/tests/unit/responses-post-keepalive-error-frame-status.test.ts b/tests/unit/responses-post-keepalive-error-frame-status.test.ts new file mode 100644 index 000000000000..4cba29ba7365 --- /dev/null +++ b/tests/unit/responses-post-keepalive-error-frame-status.test.ts @@ -0,0 +1,138 @@ +/** + * Follow-up of #13431. After the early keepalive commits `/v1/responses` to HTTP 200, + * a handler error response became a `type:error` frame that kept only code/message/param. + * Clients lost the HTTP status and Retry-After, so they could not tell a permanent error + * (403 model not allowed) from a transient one (429/503) and retried permanent errors. + * The frame now also carries `status_code`, `error_type` and `retry_after_seconds`. + */ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { + withEarlyStreamKeepalive, + OPENAI_RESPONSES_ERROR_FRAME, + OPENAI_CHAT_ERROR_FRAME, +} from "../../open-sse/utils/earlyStreamKeepalive.ts"; +import { SYNTHETIC_RESPONSES_SEQUENCE_NUMBER } from "../../open-sse/utils/responsesSequence.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]); +} + +function slow(body: string, status: number, headers: Record = {}) { + return new Promise((resolve) => { + setTimeout( + () => + resolve( + new Response(body, { + status, + headers: { "Content-Type": "application/json", ...headers }, + }) + ), + 80 + ); + }); +} + +async function responsesFrame(response: Promise) { + const result = await withEarlyStreamKeepalive(response, { + thresholdMs: 20, + intervalMs: 20, + errorFrame: OPENAI_RESPONSES_ERROR_FRAME, + }); + assert.equal(result.status, 200, "committed to 200 before the error surfaced"); + return lastDataPayload(await readAll(result)); +} + +const errorBody = (type: string, code: string, message = "boom") => + JSON.stringify({ error: { message, type, code } }); + +test("429 keeps its status, error type and Retry-After seconds", async () => { + const frame = await responsesFrame( + slow(errorBody("rate_limit_error", "rate_limit_exceeded"), 429, { "Retry-After": "30" }) + ); + assert.equal(frame.type, "error"); + assert.equal(frame.code, "rate_limit_exceeded"); + assert.equal(frame.status_code, 429); + assert.equal(frame.error_type, "rate_limit_error"); + assert.equal(frame.retry_after_seconds, 30); + assert.equal(frame.sequence_number, SYNTHETIC_RESPONSES_SEQUENCE_NUMBER); +}); + +test("403 permission error has a status and no retry hint", async () => { + const frame = await responsesFrame(slow(errorBody("permission_error", "permission_denied"), 403)); + assert.equal(frame.status_code, 403); + assert.equal(frame.error_type, "permission_error"); + assert.equal("retry_after_seconds" in frame, false); +}); + +test("an HTTP-date Retry-After becomes seconds from now, clamped to 3600", async () => { + const soon = new Date(Date.now() + 90_000).toUTCString(); + const near = await responsesFrame( + slow(errorBody("server_error", "service_unavailable"), 503, { "Retry-After": soon }) + ); + assert.equal(near.status_code, 503); + assert.ok( + typeof near.retry_after_seconds === "number" && + near.retry_after_seconds >= 85 && + near.retry_after_seconds <= 90, + `got ${near.retry_after_seconds}` + ); + + const far = await responsesFrame( + slow(errorBody("server_error", "service_unavailable"), 503, { "Retry-After": "999999" }) + ); + assert.equal(far.retry_after_seconds, 3600); +}); + +test("falls back to x-omniroute-retry-after-seconds when Retry-After is absent", async () => { + const frame = await responsesFrame( + slow(errorBody("rate_limit_error", "rate_limit_exceeded"), 429, { + "x-omniroute-retry-after-seconds": "12", + }) + ); + assert.equal(frame.retry_after_seconds, 12); +}); + +test("a body without error.type still carries status_code but no error_type", async () => { + const frame = await responsesFrame(slow(JSON.stringify({ message: "plain" }), 502)); + assert.equal(frame.status_code, 502); + assert.equal(frame.message, "plain"); + assert.equal("error_type" in frame, false); +}); + +test("fast path is unchanged: real status and headers, no frame", async () => { + const result = await withEarlyStreamKeepalive( + Promise.resolve( + new Response(errorBody("rate_limit_error", "rate_limit_exceeded"), { + status: 429, + headers: { "Content-Type": "application/json", "Retry-After": "30" }, + }) + ), + { thresholdMs: 200, intervalMs: 20, errorFrame: OPENAI_RESPONSES_ERROR_FRAME } + ); + assert.equal(result.status, 429); + assert.equal(result.headers.get("retry-after"), "30"); +}); + +test("chat-format frames are untouched (no status_code added)", async () => { + const result = await withEarlyStreamKeepalive( + slow(errorBody("rate_limit_error", "rate_limit_exceeded"), 429, { "Retry-After": "30" }), + { thresholdMs: 20, intervalMs: 20, errorFrame: OPENAI_CHAT_ERROR_FRAME } + ); + const payload = lastDataPayload(await readAll(result)); + assert.equal("status_code" in payload, false); +});