refactor(ai): internalize ChatGPT SSE protocol (#122930)

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, unknown>): 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<Uint8Array>({
+      pull(controller) {
+        pullCount += 1;
+        if (pullCount > totalChunks) {
+          controller.close();
+          return;
+        }
+        controller.enqueue(chunk);
+      },
+      cancel() {
+        canceled = true;
+      },
+    });
+    const fetchMock = vi
+      .fn<typeof fetch>()
+      .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<Uint8Array>({
+      pull(controller) {
+        pullCount += 1;
+        if (pullCount > totalChunks) {
+          controller.close();
+          return;
+        }
+        controller.enqueue(new Uint8Array(chunkBytes));
+      },
+      cancel(reason) {
+        cancelReason = reason;
+      },
+    });
+    const fetchMock = vi.fn<typeof fetch>().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<typeof fetch>()
+      .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<typeof setTimeout>;
+      });
+
+    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);
+  });
+});

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<Uint8Array>({
+      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<Uint8Array>({
+      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<typeof setTimeout> | undefined;
+    let receivedEvent = false;
+
+    try {
+      const result = await Promise.race([
+        iterator.next(),
+        new Promise<never>((_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);
+  });
+});

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<Record<string, unknown>> {
+  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<string, unknown>;
+            try {
+              event = JSON.parse(data) as Record<string, unknown>;
+            } 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 {}
+  }
+}

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<Uint8Array>({
-      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<Uint8Array>({
-      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<typeof setTimeout> | undefined;
-    let receivedEvent = false;
-
-    try {
-      const result = await Promise.race([
-        iterator.next(),
-        new Promise<never>((_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);
-  });
-});

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]);
 

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<typeof fetch>()
-      .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<typeof setTimeout>;
-      });
-
-    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<Uint8Array>({
-      pull(controller) {
-        pullCount += 1;
-        if (pullCount > totalChunks) {
-          controller.close();
-          return;
-        }
-        controller.enqueue(chunk);
-      },
-      cancel() {
-        canceled = true;
-      },
-    });
-    const fetchMock = vi.fn<typeof fetch>().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<Uint8Array>({
-      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);
-  });
 });

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<CodexResponseStatus>([
   "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<Record<string, unknown>> {
-  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<string, unknown>;
-            try {
-              event = JSON.parse(data) as Record<string, unknown>;
-            } 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
 // ============================================================================