From 5eebaf9e5c09cef238394c8c8811b8240712f9d5 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 12 Aug 2026 19:31:00 -0700 Subject: [PATCH] refactor(ai): internalize ChatGPT SSE protocol (#122930) --- .../openai-chatgpt-responses-limits.test.ts | 164 ++++++++++++++++++ .../openai-chatgpt-responses-protocol.test.ts | 135 ++++++++++++++ .../openai-chatgpt-responses-protocol.ts | 94 ++++++++++ ...openai-chatgpt-responses-streaming.test.ts | 134 -------------- ...-chatgpt-responses.sse-parse-error.test.ts | 46 +---- .../openai-chatgpt-responses.test.ts | 131 -------------- .../src/providers/openai-chatgpt-responses.ts | 109 +----------- 7 files changed, 399 insertions(+), 414 deletions(-) create mode 100644 packages/ai/src/providers/openai-chatgpt-responses-limits.test.ts create mode 100644 packages/ai/src/providers/openai-chatgpt-responses-protocol.test.ts create mode 100644 packages/ai/src/providers/openai-chatgpt-responses-protocol.ts diff --git a/packages/ai/src/providers/openai-chatgpt-responses-limits.test.ts b/packages/ai/src/providers/openai-chatgpt-responses-limits.test.ts new file mode 100644 index 000000000000..710445a7569f --- /dev/null +++ b/packages/ai/src/providers/openai-chatgpt-responses-limits.test.ts @@ -0,0 +1,164 @@ +import { MAX_TIMER_TIMEOUT_MS } from "@openclaw/normalization-core/number-coercion"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { configureAiTransportHost } from "../host.js"; +import type { Context, Model } from "../types.js"; +import { + closeOpenAICodexWebSocketSessions, + resetOpenAICodexWebSocketStateForTest, + streamOpenAICodexResponses, +} from "./openai-chatgpt-responses.js"; + +function createJwt(payload: Record): string { + const header = Buffer.from(JSON.stringify({ alg: "none", typ: "JWT" })).toString("base64url"); + const body = Buffer.from(JSON.stringify(payload)).toString("base64url"); + return `${header}.${body}.signature`; +} + +const model = { + id: "gpt-5.5", + name: "GPT-5.5", + api: "openai-chatgpt-responses", + provider: "openai", + baseUrl: "https://chatgpt.test/backend-api", + reasoning: true, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 128_000, + maxTokens: 16_000, +} satisfies Model<"openai-chatgpt-responses">; + +const context = { + messages: [{ role: "user", content: "hi", timestamp: 1 }], +} satisfies Context; + +afterEach(() => { + closeOpenAICodexWebSocketSessions(); + vi.restoreAllMocks(); + vi.unstubAllGlobals(); + resetOpenAICodexWebSocketStateForTest(); + configureAiTransportHost({}); +}); + +describe("OpenAI ChatGPT Responses resource limits", () => { + it("bounds non-OK response bodies before formatting API errors", async () => { + const byteLimit = 16 * 1024; + const totalChunks = 32; + const prefix = "usage limit "; + const chunk = new TextEncoder().encode( + `${prefix}${"x".repeat(byteLimit - prefix.length - 2)}😀tail`, + ); + let pullCount = 0; + let canceled = false; + const overflowing = new ReadableStream({ + pull(controller) { + pullCount += 1; + if (pullCount > totalChunks) { + controller.close(); + return; + } + controller.enqueue(chunk); + }, + cancel() { + canceled = true; + }, + }); + const fetchMock = vi + .fn() + .mockResolvedValueOnce(new Response(overflowing, { status: 400, statusText: "Bad Request" })); + vi.stubGlobal("fetch", fetchMock); + + const result = await streamOpenAICodexResponses(model, context, { + apiKey: createJwt({ + "https://api.openai.com/auth": { chatgpt_account_id: "acct-1" }, + }), + transport: "sse", + }).result(); + + expect(result.stopReason).toBe("error"); + expect(result.errorMessage).toContain("usage limit"); + expect(result.errorMessage).not.toContain("�"); + expect(result.errorMessage).not.toContain("tail"); + expect(result.errorMessage?.length).toBeLessThanOrEqual(byteLimit); + expect(canceled).toBe(true); + expect(pullCount).toBeGreaterThanOrEqual(1); + expect(pullCount).toBeLessThanOrEqual(3); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it("bounds streamed success bodies without content-length", async () => { + // 1 MiB chunks; cap is 16 MiB so the bounded reader cancels well before + // draining the full 32 MiB advertised body. + const chunkBytes = 1024 * 1024; + const totalChunks = 32; + let pullCount = 0; + let cancelReason: unknown; + const overflowing = new ReadableStream({ + pull(controller) { + pullCount += 1; + if (pullCount > totalChunks) { + controller.close(); + return; + } + controller.enqueue(new Uint8Array(chunkBytes)); + }, + cancel(reason) { + cancelReason = reason; + }, + }); + const fetchMock = vi.fn().mockResolvedValueOnce( + new Response(overflowing, { + status: 200, + headers: { "content-type": "text/event-stream" }, + }), + ); + vi.stubGlobal("fetch", fetchMock); + + const result = await streamOpenAICodexResponses(model, context, { + apiKey: createJwt({ + "https://api.openai.com/auth": { chatgpt_account_id: "acct-1" }, + }), + transport: "sse", + }).result(); + + expect(result.stopReason).toBe("error"); + expect(result.errorMessage).toMatch( + /OpenAI ChatGPT Responses success body exceeded 16777216 bytes/, + ); + expect(cancelReason).toBeInstanceOf(Error); + expect(pullCount).toBeGreaterThanOrEqual(17); + expect(pullCount).toBeLessThanOrEqual(20); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it("caps oversized Retry-After delays before sleeping", async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce( + new Response("rate limited", { + status: 429, + headers: { "retry-after": String(Number.MAX_SAFE_INTEGER) }, + }), + ) + .mockRejectedValueOnce(new Error("usage limit: stop after retry delay")); + vi.stubGlobal("fetch", fetchMock); + const setTimeoutSpy = vi + .spyOn(globalThis, "setTimeout") + .mockImplementation((callback: TimerHandler) => { + if (typeof callback === "function") { + callback(); + } + return 0 as unknown as ReturnType; + }); + + const result = await streamOpenAICodexResponses(model, context, { + apiKey: createJwt({ + "https://api.openai.com/auth": { chatgpt_account_id: "acct-1" }, + }), + transport: "sse", + }).result(); + + expect(result.stopReason).toBe("error"); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(setTimeoutSpy).toHaveBeenCalledWith(expect.any(Function), MAX_TIMER_TIMEOUT_MS); + }); +}); diff --git a/packages/ai/src/providers/openai-chatgpt-responses-protocol.test.ts b/packages/ai/src/providers/openai-chatgpt-responses-protocol.test.ts new file mode 100644 index 000000000000..71f018a66319 --- /dev/null +++ b/packages/ai/src/providers/openai-chatgpt-responses-protocol.test.ts @@ -0,0 +1,135 @@ +import { describe, expect, it } from "vitest"; +import { parseOpenAIChatGptResponsesSse } from "./openai-chatgpt-responses-protocol.js"; + +const completedEvent = { + type: "response.completed", + response: { + id: "resp_parser", + status: "completed", + output: [], + usage: { input_tokens: 5, output_tokens: 3, total_tokens: 8 }, + }, +}; +const serializedCompletedEvent = JSON.stringify(completedEvent); +const multilineDataLines = JSON.stringify(completedEvent, null, 2) + .split("\n") + .map((line) => `data: ${line}`); + +describe("ChatGPT Responses SSE frame boundaries", () => { + it.each([ + { label: "LF", chunks: [`data: ${serializedCompletedEvent}\n\n`] }, + { label: "CRLF", chunks: [`data: ${serializedCompletedEvent}\r\n\r\n`] }, + { label: "lone CR", chunks: [`data: ${serializedCompletedEvent}\r\r`] }, + { + label: "mixed line endings", + chunks: [`event: response.completed\r\ndata: ${serializedCompletedEvent}\n\r\n`], + }, + { + label: "chunk-split CRLF", + chunks: [ + `event: response.completed\r`, + `\ndata: ${serializedCompletedEvent}\r`, + "\n\r", + "\n", + ], + }, + { + label: "chunk-split lone CR", + chunks: ["event: response.completed\r", `data: ${serializedCompletedEvent}\r`, "\r"], + }, + { label: "multiline LF", chunks: [`${multilineDataLines.join("\n")}\n\n`] }, + { label: "multiline CRLF", chunks: [`${multilineDataLines.join("\r\n")}\r\n\r\n`] }, + { label: "multiline lone CR", chunks: [`${multilineDataLines.join("\r")}\r\r`] }, + { + label: "multiline mixed line endings", + chunks: [ + `event: response.completed\r\n${multilineDataLines + .map( + (line, index) => `${line}${index % 3 === 0 ? "\r\n" : index % 3 === 1 ? "\r" : "\n"}`, + ) + .join("")}\r\n`, + ], + }, + { + label: "multiline chunk-split CRLF", + chunks: [...multilineDataLines.flatMap((line) => [`${line}\r`, "\n"]), "\r", "\n"], + }, + { + label: "multiline chunk-split lone CR", + chunks: [...multilineDataLines.flatMap((line) => [line, "\r"]), "\r"], + }, + ])("parses $label SSE frame boundaries", async ({ chunks }) => { + let chunkIndex = 0; + const body = new ReadableStream({ + pull(controller) { + const chunk = chunks[chunkIndex++]; + if (chunk === undefined) { + controller.close(); + return; + } + controller.enqueue(new TextEncoder().encode(chunk)); + }, + }); + const events = []; + + for await (const event of parseOpenAIChatGptResponsesSse(new Response(body))) { + events.push(event); + } + + expect(events).toEqual([completedEvent]); + }); + + it.each([ + { label: "lone CR", chunks: [`data: ${serializedCompletedEvent}\r\r`] }, + { label: "mixed LF and lone CR", chunks: [`data: ${serializedCompletedEvent}\n\r`] }, + { label: "mixed CRLF and lone CR", chunks: [`data: ${serializedCompletedEvent}\r\n\r`] }, + { label: "chunk-split lone CR", chunks: [`data: ${serializedCompletedEvent}\r`, "\r"] }, + { + label: "chunk-split mixed LF and lone CR", + chunks: [`data: ${serializedCompletedEvent}\n`, "\r"], + }, + ])("dispatches a $label SSE frame before an open response closes", async ({ chunks }) => { + const cleanup = new AbortController(); + let canceled = false; + const body = new ReadableStream({ + start(controller) { + cleanup.signal.addEventListener("abort", () => controller.close(), { once: true }); + for (const chunk of chunks) { + controller.enqueue(new TextEncoder().encode(chunk)); + } + }, + cancel() { + canceled = true; + }, + }); + const iterator = parseOpenAIChatGptResponsesSse(new Response(body))[Symbol.asyncIterator](); + let timeout: ReturnType | undefined; + let receivedEvent = false; + + try { + const result = await Promise.race([ + iterator.next(), + new Promise((_resolve, reject) => { + timeout = setTimeout(() => { + reject(new Error("SSE frame was not dispatched while the response remained open")); + }, 1_000); + }), + ]); + receivedEvent = true; + + expect(result).toEqual({ done: false, value: completedEvent }); + expect(cleanup.signal.aborted).toBe(false); + expect(canceled).toBe(false); + } finally { + if (timeout) { + clearTimeout(timeout); + } + if (!receivedEvent) { + cleanup.abort(); + } + await iterator.return(undefined); + } + + expect(canceled).toBe(true); + }); +}); diff --git a/packages/ai/src/providers/openai-chatgpt-responses-protocol.ts b/packages/ai/src/providers/openai-chatgpt-responses-protocol.ts new file mode 100644 index 000000000000..4b799fa94fc9 --- /dev/null +++ b/packages/ai/src/providers/openai-chatgpt-responses-protocol.ts @@ -0,0 +1,94 @@ +import { MALFORMED_STREAMING_FRAGMENT_ERROR_MESSAGE } from "../transports/transport-utils.js"; +import { createSseByteGuard } from "../utils/streaming-byte-guard.js"; + +const OPENAI_CHATGPT_RESPONSES_SUCCESS_BODY_MAX_BYTES = 16 * 1024 * 1024; + +export class CodexProtocolError extends Error { + readonly payload?: unknown; + + constructor(message: string, options?: { payload?: unknown; cause?: unknown }) { + super(message); + this.name = "CodexProtocolError"; + this.payload = options?.payload; + this.cause = options?.cause; + } +} + +export async function* parseOpenAIChatGptResponsesSse( + response: Response, +): AsyncGenerator> { + if (!response.body) { + return; + } + + const reader = response.body.getReader(); + // Cap the streaming 200 success-body read at 16 MiB, mirroring the + // non-streaming response cap so a hostile endpoint cannot exhaust memory. + const guard = createSseByteGuard(reader, { + maxBytes: OPENAI_CHATGPT_RESPONSES_SUCCESS_BODY_MAX_BYTES, + onOverflow: ({ size, maxBytes }) => + new Error( + `OpenAI ChatGPT Responses success body exceeded ${maxBytes} bytes (received ${size})`, + ), + }); + const decoder = new TextDecoder(); + let buffer = ""; + + try { + while (true) { + const { done, value } = await guard.read(); + if (value) { + buffer += decoder.decode(value, { stream: true }); + } + if (done) { + buffer += decoder.decode(); + } + + while (true) { + // Defer a possible CRLF only when CR does not already complete a blank line. + const deferTrailingCr = + !done && buffer.endsWith("\r") && !buffer.endsWith("\r\r") && !buffer.endsWith("\n\r"); + const searchable = deferTrailingCr ? buffer.slice(0, -1) : buffer; + // A CRLF is one line ending: never backtrack its CR into a false blank line. + const boundary = /(?:\r\n|\r(?!\n)|\n)(?:\r\n|\r(?!\n)|\n)/.exec(searchable); + if (!boundary) { + break; + } + const chunk = buffer.slice(0, boundary.index); + buffer = buffer.slice(boundary.index + boundary[0].length); + + const dataLines = chunk + .split(/\r\n|\r|\n/) + .filter((line) => line.startsWith("data:")) + .map((line) => line.slice(5).trim()); + if (dataLines.length > 0) { + const data = dataLines.join("\n").trim(); + if (data && data !== "[DONE]") { + let event: Record; + try { + event = JSON.parse(data) as Record; + } catch (cause) { + if (!(cause instanceof SyntaxError)) { + throw cause; + } + throw new CodexProtocolError(MALFORMED_STREAMING_FRAGMENT_ERROR_MESSAGE, { cause }); + } + // Keep suspension outside the parse catch so consumer failures stay consumer-owned. + yield event; + } + } + } + + if (done) { + break; + } + } + } finally { + try { + await guard.cancel(); + } catch {} + try { + reader.releaseLock(); + } catch {} + } +} diff --git a/packages/ai/src/providers/openai-chatgpt-responses-streaming.test.ts b/packages/ai/src/providers/openai-chatgpt-responses-streaming.test.ts index 2ea0a1eaf01a..4ad10c424edf 100644 --- a/packages/ai/src/providers/openai-chatgpt-responses-streaming.test.ts +++ b/packages/ai/src/providers/openai-chatgpt-responses-streaming.test.ts @@ -3,7 +3,6 @@ import { configureAiTransportHost } from "../host.js"; import type { Context, Model } from "../types.js"; import { closeOpenAICodexWebSocketSessions, - parseSSEForTest, resetOpenAICodexWebSocketStateForTest, streamOpenAICodexResponses, } from "./openai-chatgpt-responses.js"; @@ -313,136 +312,3 @@ describe("OpenAI ChatGPT Responses inference streaming", () => { }); }); }); - -describe("ChatGPT Responses SSE frame boundaries", () => { - const completedEvent = { - type: "response.completed", - response: { - id: "resp_parser", - status: "completed", - output: [], - usage: { input_tokens: 5, output_tokens: 3, total_tokens: 8 }, - }, - }; - const serializedCompletedEvent = JSON.stringify(completedEvent); - const multilineDataLines = JSON.stringify(completedEvent, null, 2) - .split("\n") - .map((line) => `data: ${line}`); - - it.each([ - { label: "LF", chunks: [`data: ${serializedCompletedEvent}\n\n`] }, - { label: "CRLF", chunks: [`data: ${serializedCompletedEvent}\r\n\r\n`] }, - { label: "lone CR", chunks: [`data: ${serializedCompletedEvent}\r\r`] }, - { - label: "mixed line endings", - chunks: [`event: response.completed\r\ndata: ${serializedCompletedEvent}\n\r\n`], - }, - { - label: "chunk-split CRLF", - chunks: [ - `event: response.completed\r`, - `\ndata: ${serializedCompletedEvent}\r`, - "\n\r", - "\n", - ], - }, - { - label: "chunk-split lone CR", - chunks: ["event: response.completed\r", `data: ${serializedCompletedEvent}\r`, "\r"], - }, - { label: "multiline LF", chunks: [`${multilineDataLines.join("\n")}\n\n`] }, - { label: "multiline CRLF", chunks: [`${multilineDataLines.join("\r\n")}\r\n\r\n`] }, - { label: "multiline lone CR", chunks: [`${multilineDataLines.join("\r")}\r\r`] }, - { - label: "multiline mixed line endings", - chunks: [ - `event: response.completed\r\n${multilineDataLines - .map( - (line, index) => `${line}${index % 3 === 0 ? "\r\n" : index % 3 === 1 ? "\r" : "\n"}`, - ) - .join("")}\r\n`, - ], - }, - { - label: "multiline chunk-split CRLF", - chunks: [...multilineDataLines.flatMap((line) => [`${line}\r`, "\n"]), "\r", "\n"], - }, - { - label: "multiline chunk-split lone CR", - chunks: [...multilineDataLines.flatMap((line) => [line, "\r"]), "\r"], - }, - ])("parses $label SSE frame boundaries", async ({ chunks }) => { - let chunkIndex = 0; - const body = new ReadableStream({ - pull(controller) { - const chunk = chunks[chunkIndex++]; - if (chunk === undefined) { - controller.close(); - return; - } - controller.enqueue(new TextEncoder().encode(chunk)); - }, - }); - const events = []; - - for await (const event of parseSSEForTest(new Response(body))) { - events.push(event); - } - - expect(events).toEqual([completedEvent]); - }); - - it.each([ - { label: "lone CR", chunks: [`data: ${serializedCompletedEvent}\r\r`] }, - { label: "mixed LF and lone CR", chunks: [`data: ${serializedCompletedEvent}\n\r`] }, - { label: "mixed CRLF and lone CR", chunks: [`data: ${serializedCompletedEvent}\r\n\r`] }, - { label: "chunk-split lone CR", chunks: [`data: ${serializedCompletedEvent}\r`, "\r"] }, - { - label: "chunk-split mixed LF and lone CR", - chunks: [`data: ${serializedCompletedEvent}\n`, "\r"], - }, - ])("dispatches a $label SSE frame before an open response closes", async ({ chunks }) => { - const cleanup = new AbortController(); - let canceled = false; - const body = new ReadableStream({ - start(controller) { - cleanup.signal.addEventListener("abort", () => controller.close(), { once: true }); - for (const chunk of chunks) { - controller.enqueue(new TextEncoder().encode(chunk)); - } - }, - cancel() { - canceled = true; - }, - }); - const iterator = parseSSEForTest(new Response(body))[Symbol.asyncIterator](); - let timeout: ReturnType | undefined; - let receivedEvent = false; - - try { - const result = await Promise.race([ - iterator.next(), - new Promise((_resolve, reject) => { - timeout = setTimeout(() => { - reject(new Error("SSE frame was not dispatched while the response remained open")); - }, 1_000); - }), - ]); - receivedEvent = true; - - expect(result).toEqual({ done: false, value: completedEvent }); - expect(cleanup.signal.aborted).toBe(false); - expect(canceled).toBe(false); - } finally { - if (timeout) { - clearTimeout(timeout); - } - if (!receivedEvent) { - cleanup.abort(); - } - await iterator.return(undefined); - } - - expect(canceled).toBe(true); - }); -}); diff --git a/packages/ai/src/providers/openai-chatgpt-responses.sse-parse-error.test.ts b/packages/ai/src/providers/openai-chatgpt-responses.sse-parse-error.test.ts index 86717b36fc4a..e892c100b9d9 100644 --- a/packages/ai/src/providers/openai-chatgpt-responses.sse-parse-error.test.ts +++ b/packages/ai/src/providers/openai-chatgpt-responses.sse-parse-error.test.ts @@ -3,7 +3,7 @@ import type { AddressInfo } from "node:net"; import { describe, expect, it } from "vitest"; import { MALFORMED_STREAMING_FRAGMENT_ERROR_MESSAGE } from "../transports/transport-utils.js"; import type { Context, Model } from "../types.js"; -import { parseSSEForTest, streamOpenAICodexResponses } from "./openai-chatgpt-responses.js"; +import { streamOpenAICodexResponses } from "./openai-chatgpt-responses.js"; // Stands in for the payload class this path exposes: text that reached the SSE // frame as ordinary stream content rather than as a provider error envelope. @@ -94,50 +94,6 @@ async function streamCodexSseFrames( } describe("Codex malformed SSE frames", () => { - it("classifies only parser-owned SyntaxErrors as malformed frames", async () => { - const iterator = parseSSEForTest( - new Response(`data: ${MALFORMED_FRAME}\n\n`, { - headers: { "content-type": "text/event-stream" }, - }), - ); - - let caught: unknown; - try { - await iterator.next(); - } catch (error) { - caught = error; - } - - expect(caught).toMatchObject({ - name: "CodexProtocolError", - message: MALFORMED_STREAMING_FRAGMENT_ERROR_MESSAGE, - cause: expect.any(SyntaxError), - }); - }); - - it("preserves a consumer-thrown SyntaxError unchanged", async () => { - const iterator = parseSSEForTest( - new Response(`data: ${COMPLETED_FRAME}\n\n`, { - headers: { "content-type": "text/event-stream" }, - }), - ); - const first = await iterator.next(); - expect(first).toMatchObject({ - done: false, - value: { type: "response.completed" }, - }); - - const consumerError = new SyntaxError("consumer failed after receiving an event"); - let caught: unknown; - try { - await iterator.throw(consumerError); - } catch (error) { - caught = error; - } - - expect(caught).toBe(consumerError); - }); - it("reports the shared malformed-fragment error without echoing parser text", async () => { const result = await streamCodexSseFrames([MALFORMED_FRAME]); diff --git a/packages/ai/src/providers/openai-chatgpt-responses.test.ts b/packages/ai/src/providers/openai-chatgpt-responses.test.ts index c80bfb36a463..3f5404ebea19 100644 --- a/packages/ai/src/providers/openai-chatgpt-responses.test.ts +++ b/packages/ai/src/providers/openai-chatgpt-responses.test.ts @@ -9,7 +9,6 @@ import { SYSTEM_PROMPT_CACHE_BOUNDARY } from "../utils/system-prompt-cache-bound import { closeOpenAICodexWebSocketSessions, extractOpenAICodexAccountId, - parseSSEForTest, resetOpenAICodexWebSocketStateForTest, streamSimpleOpenAICodexResponses, streamOpenAICodexResponses, @@ -981,134 +980,4 @@ describe("streamOpenAICodexResponses transport", () => { expect(fetchMock).toHaveBeenCalledTimes(2); expect(setTimeoutSpy).toHaveBeenCalledWith(expect.any(Function), MAX_TIMER_TIMEOUT_MS); }); - - it("caps oversized Retry-After delays before sleeping", async () => { - const fetchMock = vi - .fn() - .mockResolvedValueOnce( - new Response("rate limited", { - status: 429, - headers: { "retry-after": String(Number.MAX_SAFE_INTEGER) }, - }), - ) - .mockRejectedValueOnce(new Error("usage limit: stop after retry delay")); - vi.stubGlobal("fetch", fetchMock); - const setTimeoutSpy = vi - .spyOn(globalThis, "setTimeout") - .mockImplementation((callback: TimerHandler) => { - if (typeof callback === "function") { - callback(); - } - return 0 as unknown as ReturnType; - }); - - const stream = streamOpenAICodexResponses(model, context, { - apiKey: createJwt({ - "https://api.openai.com/auth": { - chatgpt_account_id: "acct-1", - }, - }), - transport: "sse", - }); - - const result = await stream.result(); - - expect(result.stopReason).toBe("error"); - expect(fetchMock).toHaveBeenCalledTimes(2); - expect(setTimeoutSpy).toHaveBeenCalledWith(expect.any(Function), MAX_TIMER_TIMEOUT_MS); - }); - - it("bounds non-OK ChatGPT response bodies before formatting API errors", async () => { - const byteLimit = 16 * 1024; - const totalChunks = 32; - const prefix = "usage limit "; - const chunk = new TextEncoder().encode( - `${prefix}${"x".repeat(byteLimit - prefix.length - 2)}😀tail`, - ); - let pullCount = 0; - let canceled = false; - const overflowing = new ReadableStream({ - pull(controller) { - pullCount += 1; - if (pullCount > totalChunks) { - controller.close(); - return; - } - controller.enqueue(chunk); - }, - cancel() { - canceled = true; - }, - }); - const fetchMock = vi.fn().mockResolvedValueOnce( - new Response(overflowing, { - status: 400, - statusText: "Bad Request", - }), - ); - vi.stubGlobal("fetch", fetchMock); - - const stream = streamOpenAICodexResponses(model, context, { - apiKey: createJwt({ - "https://api.openai.com/auth": { - chatgpt_account_id: "acct-1", - }, - }), - transport: "sse", - }); - - const result = await stream.result(); - - expect(result.stopReason).toBe("error"); - expect(result.errorMessage).toContain("usage limit"); - expect(result.errorMessage).not.toContain("�"); - expect(result.errorMessage).not.toContain("tail"); - expect(result.errorMessage?.length).toBeLessThanOrEqual(16 * 1024); - expect(canceled).toBe(true); - expect(pullCount).toBeGreaterThanOrEqual(1); - expect(pullCount).toBeLessThanOrEqual(3); - expect(fetchMock).toHaveBeenCalledTimes(1); - }); -}); - -describe("parseSSEForTest", () => { - it("bounds streamed OpenAI ChatGPT Responses success bodies without content-length", async () => { - // 1 MiB chunks; cap is 16 MiB so the bounded reader cancels well before - // draining the full 32 MiB advertised body. - const CHUNK = 1024 * 1024; - const TOTAL = 32; - let pullCount = 0; - let cancelReason: unknown; - const overflowing = new ReadableStream({ - pull(controller) { - pullCount += 1; - if (pullCount > TOTAL) { - controller.close(); - return; - } - controller.enqueue(new Uint8Array(CHUNK)); - }, - cancel(reason) { - cancelReason = reason; - }, - }); - let caught: Error | null = null; - try { - // parseSSE expects a Response-like; pass the streaming body directly - // through a minimal Response shim that only exposes .body. - const response = { body: overflowing } as unknown as Response; - for await (const event of parseSSEForTest(response)) { - expect(event).toBeDefined(); - } - } catch (err) { - caught = err as Error; - } - expect(caught?.message).toMatch( - /OpenAI ChatGPT Responses success body exceeded 16777216 bytes/, - ); - expect(cancelReason).toBeInstanceOf(Error); - // 16 MiB + a couple of overshoot pulls, well under 32. - expect(pullCount).toBeGreaterThanOrEqual(17); - expect(pullCount).toBeLessThanOrEqual(20); - }); }); diff --git a/packages/ai/src/providers/openai-chatgpt-responses.ts b/packages/ai/src/providers/openai-chatgpt-responses.ts index c5dc29855a29..39f1b1d59dba 100644 --- a/packages/ai/src/providers/openai-chatgpt-responses.ts +++ b/packages/ai/src/providers/openai-chatgpt-responses.ts @@ -79,9 +79,12 @@ import { getFirstStreamEventTimeoutMs, withFirstStreamEventTimeout, } from "../utils/stream-first-event-timeout.js"; -import { createSseByteGuard } from "../utils/streaming-byte-guard.js"; import { stripSystemPromptCacheBoundary } from "../utils/system-prompt-cache-boundary.js"; import { inspectTlsCertificateError } from "../utils/tls-certificate-errors.js"; +import { + CodexProtocolError, + parseOpenAIChatGptResponsesSse, +} from "./openai-chatgpt-responses-protocol.js"; import { clampOpenAIPromptCacheKey } from "./openai-prompt-cache.js"; import { supportsOpenAITemperature } from "./openai-reasoning-effort.js"; import { @@ -105,7 +108,6 @@ const CODEX_TOOL_CALL_PROVIDERS = new Set(["openai", "opencode"]); const WEBSOCKET_MESSAGE_TOO_BIG_CLOSE_CODE = 1009; const WEBSOCKET_CONNECTION_LIMIT_REACHED_CODE = "websocket_connection_limit_reached"; const OPENAI_CHATGPT_RESPONSES_ERROR_BODY_MAX_BYTES = 16 * 1024; -const OPENAI_CHATGPT_RESPONSES_SUCCESS_BODY_MAX_BYTES = 16 * 1024 * 1024; const CODEX_RESPONSE_STATUSES = new Set([ "completed", @@ -583,7 +585,7 @@ export const streamOpenAICodexResponses: StreamFunction< } const hookedResponseStream = withProviderResponseHook({ - stream: mapCodexEvents(parseSSE(response)), + stream: mapCodexEvents(parseOpenAIChatGptResponsesSse(response)), signal: firstEventAbort.signal, abort: firstEventAbort.abort, hook: createOpenAIResponseHook(options?.onResponse, response, model), @@ -781,17 +783,6 @@ class CodexApiError extends Error { } } -class CodexProtocolError extends Error { - readonly payload?: unknown; - - constructor(message: string, options?: { payload?: unknown; cause?: unknown }) { - super(message); - this.name = "CodexProtocolError"; - this.payload = options?.payload; - this.cause = options?.cause; - } -} - function isCodexNonTransportError(error: unknown): boolean { return ( error instanceof CodexApiError || @@ -875,96 +866,6 @@ function normalizeCodexStatus(status: unknown): CodexResponseStatus | undefined : undefined; } -// ============================================================================ -// SSE Parsing -// ============================================================================ - -async function* parseSSE(response: Response): AsyncGenerator> { - if (!response.body) { - return; - } - - const reader = response.body.getReader(); - // Cap the streaming 200 success-body read at 16 MiB, mirroring the - // non-streaming `readProviderJsonResponse` cap so a hostile or - // malfunctioning ChatGPT Responses endpoint cannot exhaust memory by - // streaming an unbounded SSE body. - const guard = createSseByteGuard(reader, { - maxBytes: OPENAI_CHATGPT_RESPONSES_SUCCESS_BODY_MAX_BYTES, - onOverflow: ({ size, maxBytes }) => - new Error( - `OpenAI ChatGPT Responses success body exceeded ${maxBytes} bytes (received ${size})`, - ), - }); - const decoder = new TextDecoder(); - let buffer = ""; - - try { - while (true) { - const { done, value } = await guard.read(); - if (value) { - buffer += decoder.decode(value, { stream: true }); - } - if (done) { - buffer += decoder.decode(); - } - - while (true) { - // Defer a possible CRLF only when CR does not already complete a blank line. - const deferTrailingCr = - !done && buffer.endsWith("\r") && !buffer.endsWith("\r\r") && !buffer.endsWith("\n\r"); - const searchable = deferTrailingCr ? buffer.slice(0, -1) : buffer; - // A CRLF is one line ending: never backtrack its CR into a false blank line. - const boundary = /(?:\r\n|\r(?!\n)|\n)(?:\r\n|\r(?!\n)|\n)/.exec(searchable); - if (!boundary) { - break; - } - const chunk = buffer.slice(0, boundary.index); - buffer = buffer.slice(boundary.index + boundary[0].length); - - const dataLines = chunk - .split(/\r\n|\r|\n/) - .filter((l) => l.startsWith("data:")) - .map((l) => l.slice(5).trim()); - if (dataLines.length > 0) { - const data = dataLines.join("\n").trim(); - if (data && data !== "[DONE]") { - let event: Record; - try { - event = JSON.parse(data) as Record; - } catch (cause) { - if (!(cause instanceof SyntaxError)) { - throw cause; - } - // Align with the canonical transport contract: the shared marker is what - // assistant error formatting maps to the malformed-fragment retry copy. - throw new CodexProtocolError(MALFORMED_STREAMING_FRAGMENT_ERROR_MESSAGE, { cause }); - } - // Keep suspension outside the parse catch so iterator.throw() cannot relabel a - // consumer failure as malformed provider input. - yield event; - } - } - } - - if (done) { - break; - } - } - } finally { - try { - await guard.cancel(); - } catch {} - try { - reader.releaseLock(); - } catch {} - } -} - -// Test-only re-export of the bounded SSE parser. Mirrors -// `parseAnthropicSseBodyForTest` / `iterateSseMessagesForTest` patterns. -export const parseSSEForTest = parseSSE; - // ============================================================================ // WebSocket Parsing // ============================================================================