fix(responses): bound synthesized SSE expansion

src/server/responses-json-events.ts

@@ -7,6 +7,8 @@
 
 export type ResponsesJsonEventFrame = Record<string, unknown>;
 
+export const MAX_SYNTHESIZED_OUTPUT_ITEMS = 10_000;
+
 /**
  * The canonical minimal sequence Codex commits: response.created (empty
  * output, in_progress) → one response.output_item.done per output item → a
@@ -16,26 +18,38 @@ export function responsesJsonEventSequence(
   response: Record<string, unknown>,
   rewritePayload?: (payload: Record<string, unknown>) => Record<string, unknown>,
 ): ResponsesJsonEventFrame[] {
+  return [...iterateResponsesJsonEvents(response, rewritePayload)];
+}
+
+function* iterateResponsesJsonEvents(
+  response: Record<string, unknown>,
+  rewritePayload?: (payload: Record<string, unknown>) => Record<string, unknown>,
+): Generator<ResponsesJsonEventFrame> {
   const rewrite = rewritePayload ?? ((payload: Record<string, unknown>) => payload);
   const output = Array.isArray(response.output) ? response.output : [];
+  if (output.length > MAX_SYNTHESIZED_OUTPUT_ITEMS) {
+    throw new RangeError(
+      `Responses JSON output contains ${output.length} items; maximum is ${MAX_SYNTHESIZED_OUTPUT_ITEMS}`,
+    );
+  }
   const finalStatus = response.status === "failed" || response.status === "incomplete"
     ? response.status
     : "completed";
-  return [
-    rewrite({
-      type: "response.created",
-      response: { ...response, status: "in_progress", output: [] },
-    }),
-    ...output.map((item, outputIndex) => rewrite({
+  yield rewrite({
+    type: "response.created",
+    response: { ...response, status: "in_progress", output: [] },
+  });
+  for (const [outputIndex, item] of output.entries()) {
+    yield rewrite({
       type: "response.output_item.done",
       output_index: outputIndex,
       item,
-    })),
-    rewrite({
-      type: `response.${finalStatus}`,
-      response: { ...response, status: finalStatus },
-    }),
-  ];
+    });
+  }
+  yield rewrite({
+    type: `response.${finalStatus}`,
+    response: { ...response, status: finalStatus },
+  });
 }
 
 /**
@@ -50,3 +64,27 @@ export function responsesJsonToSseBody(
     .map(frame => `data: ${JSON.stringify(frame)}\n\n`);
   return `${frames.join("")}data: [DONE]\n\n`;
 }
+
+/** Stream synthesized SSE frames without retaining the expanded body in memory. */
+export function responsesJsonToSseStream(
+  response: Record<string, unknown>,
+  rewritePayload?: (payload: Record<string, unknown>) => Record<string, unknown>,
+): ReadableStream<Uint8Array> {
+  const output = Array.isArray(response.output) ? response.output : [];
+  if (output.length > MAX_SYNTHESIZED_OUTPUT_ITEMS) {
+    throw new RangeError(
+      `Responses JSON output contains ${output.length} items; maximum is ${MAX_SYNTHESIZED_OUTPUT_ITEMS}`,
+    );
+  }
+  const frames = iterateResponsesJsonEvents(response, rewritePayload);
+  const encoder = new TextEncoder();
+  return new ReadableStream({
+    pull(controller) {
+      const next = frames.next();
+      controller.enqueue(encoder.encode(
+        next.done ? "data: [DONE]\n\n" : `data: ${JSON.stringify(next.value)}\n\n`,
+      ));
+      if (next.done) controller.close();
+    },
+  });
+}

src/server/responses/core.ts

@@ -191,7 +191,7 @@ import {
   relaySseWithBlockRewrite,
 } from "../sse-payload-rewrite";
 import { createGithubCopilotResponsesBlockRewrite } from "../github-copilot-responses-repair";
-import { responsesJsonToSseBody } from "../responses-json-events";
+import { responsesJsonToSseStream } from "../responses-json-events";
 import { guardTerminalEventStream } from "./terminal-guard";
 
 /**
@@ -2323,25 +2323,46 @@ async function handleResponsesInner(
         && options.inboundTransport !== "websocket"
         && providerModelResponsesUpstreamStreaming(route.providerName, route.provider, route.modelId) === false
         && route.provider.adapter === "openai-responses") {
+        let completed: Record<string, unknown> | undefined;
         try {
-          let completed = JSON.parse(clientJson) as Record<string, unknown>;
+          const parsedCompleted = JSON.parse(clientJson) as unknown;
+          if (!parsedCompleted || typeof parsedCompleted !== "object" || Array.isArray(parsedCompleted)) {
+            throw new TypeError("bounded Responses JSON is not an object");
+          }
+          let candidate = parsedCompleted as Record<string, unknown>;
           // The bounded-JSON answer bypasses the SSE relay, so it also bypasses
           // the SSE item-id rewrite. Apply the same client-facing normalization
           // here or this policy would silently disable id repair for the very
           // providers that need it (raw record already happened above).
           if (hasResponsesItemIdRepair(route.provider.responsesItemIdRepair)) {
-            completed = repairResponsesJsonItemIds(completed, route.provider.responsesItemIdRepair!, translatorBudget);
+            candidate = repairResponsesJsonItemIds(candidate, route.provider.responsesItemIdRepair!, translatorBudget);
+          }
+          completed = candidate;
+        } catch {
+          // Non-JSON despite content-type: fall through to the plain relay.
+        }
+        if (completed) {
+          let stream: ReadableStream<Uint8Array>;
+          try {
+            stream = responsesJsonToSseStream(completed);
+          } catch (error) {
+            if (error instanceof RangeError) {
+              return formatErrorResponse(
+                502,
+                "upstream_error",
+                "upstream JSON response exceeded the synthesized SSE item limit",
+              );
+            }
+            throw error;
           }
           const sseHeaders = sanitizePassthroughHeaders(headers);
           sseHeaders.set("content-type", "text/event-stream");
           sseHeaders.set("cache-control", "no-store");
-          return new Response(responsesJsonToSseBody(completed), {
+          return new Response(stream, {
             status: upstreamResponse.status,
             statusText: upstreamResponse.statusText,
             headers: sseHeaders,
           });
-        } catch {
-          // Non-JSON despite content-type: fall through to the plain relay.
         }
       }
       // WS turns reframe this JSON into events in the bridge, which is the

structure/04_transports-and-sidecars.md

@@ -218,6 +218,8 @@ HTTP/SSE. The bridge reframes that JSON into the same Responses event sequence
 HTTP clients that requested streaming receive a synthesized terminal SSE body (created →
 output_item.done → terminal → `[DONE]`). DeepSeek V4 Flash uses this path because its Codex
 streaming response can deliver output without closing on a terminal event.
+Synthesized output is capped at 10,000 items across HTTP and WebSocket reframing. HTTP frames are
+encoded incrementally, so bounded upstream JSON cannot expand into an unbounded event array or SSE string.
 
 `ws-bridge.ts` preserves upstream `failed` and `incomplete` status values in the final WebSocket
 frame rather than always emitting `response.completed`. If the response status is `failed`, a

tests/deepseek-responses-item-id-repair.test.ts

@@ -6,6 +6,7 @@ import {
   hasResponsesItemIdRepair,
   repairResponsesJsonItemIds,
 } from "../src/server/responses-item-id-repair";
+import { MAX_SYNTHESIZED_OUTPUT_ITEMS } from "../src/server/responses-json-events";
 import { handleResponses } from "../src/server/responses/core";
 import type { OcxConfig, OcxProviderConfig } from "../src/types";
 import { createTestTranslatorBudget } from "./helpers/translator-budget";
@@ -158,4 +159,33 @@ describe("bounded-JSON HTTP path carries canonical ids (#938 + #875)", () => {
     expect(text).toContain("call_keep");
     expect(text).toContain("data: [DONE]");
   });
+
+  test("fails closed when bounded JSON would synthesize too many SSE frames", async () => {
+    const plainSeed = { ...providerConfigSeed(getProviderRegistryEntry("deepseek")!), apiKey: "sk-test" };
+    globalThis.fetch = (async () => Response.json({
+      id: "resp_deepseek",
+      object: "response",
+      status: "completed",
+      output: Array.from({ length: MAX_SYNTHESIZED_OUTPUT_ITEMS + 1 }, () => null),
+    })) as typeof fetch;
+
+    const config = { providers: { deepseek: plainSeed } } as unknown as OcxConfig;
+    const response = await handleResponses(
+      new Request("http://localhost/v1/responses", {
+        method: "POST",
+        headers: { "content-type": "application/json" },
+        body: JSON.stringify({ model: "deepseek-v4-flash", input: "ping", stream: true }),
+      }),
+      config,
+      { model: "", provider: "" },
+      {},
+    );
+    expect(response.status).toBe(502);
+    expect(response.headers.get("content-type")).toContain("application/json");
+    const text = await response.text();
+    const body = JSON.parse(text) as { error?: { type?: string; message?: string } };
+    expect(body.error?.type).toBe("server_error");
+    expect(body.error?.message).toContain("synthesized SSE item limit");
+    expect(text).not.toContain("data: [DONE]");
+  });
 });

tests/responses-json-events.test.ts

@@ -1,7 +1,9 @@
 import { describe, expect, test } from "bun:test";
 import {
+  MAX_SYNTHESIZED_OUTPUT_ITEMS,
   responsesJsonEventSequence,
   responsesJsonToSseBody,
+  responsesJsonToSseStream,
 } from "../src/server/responses-json-events";
 
 describe("responsesJsonEventSequence", () => {
@@ -60,4 +62,28 @@ describe("responsesJsonToSseBody", () => {
     expect(frames[3]).toBe("data: [DONE]");
     expect(body.endsWith("data: [DONE]\n\n")).toBe(true);
   });
+
+  test("rejects output arrays that could amplify synthesized frames", () => {
+    const output = Array.from({ length: MAX_SYNTHESIZED_OUTPUT_ITEMS + 1 }, () => null);
+    expect(() => responsesJsonToSseBody({ id: "r", output })).toThrow(RangeError);
+    expect(() => responsesJsonToSseStream({ id: "r", output })).toThrow(RangeError);
+  });
+
+  test("streams one SSE frame per pull and ends with [DONE]", async () => {
+    const stream = responsesJsonToSseStream({
+      id: "r",
+      status: "completed",
+      output: [{ type: "message", id: "m" }],
+    });
+    const reader = stream.getReader();
+    const decoder = new TextDecoder();
+    const chunks: string[] = [];
+    while (true) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      chunks.push(decoder.decode(value));
+    }
+    expect(chunks).toHaveLength(4);
+    expect(chunks.at(-1)).toBe("data: [DONE]\n\n");
+  });
 });