From b50f2394351f9954732d3441332054cf6ab24060 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Fri, 7 Aug 2026 16:19:47 +0900 Subject: [PATCH] fix(responses): bound synthesized SSE expansion --- src/server/responses-json-events.ts | 62 +++++++++++++++---- src/server/responses/core.ts | 33 ++++++++-- structure/04_transports-and-sidecars.md | 2 + .../deepseek-responses-item-id-repair.test.ts | 30 +++++++++ tests/responses-json-events.test.ts | 26 ++++++++ 5 files changed, 135 insertions(+), 18 deletions(-) diff --git a/src/server/responses-json-events.ts b/src/server/responses-json-events.ts index 9910d17d54..6606efb72f 100644 --- a/src/server/responses-json-events.ts +++ b/src/server/responses-json-events.ts @@ -7,6 +7,8 @@ export type ResponsesJsonEventFrame = Record; +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, rewritePayload?: (payload: Record) => Record, ): ResponsesJsonEventFrame[] { + return [...iterateResponsesJsonEvents(response, rewritePayload)]; +} + +function* iterateResponsesJsonEvents( + response: Record, + rewritePayload?: (payload: Record) => Record, +): Generator { const rewrite = rewritePayload ?? ((payload: Record) => 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, + rewritePayload?: (payload: Record) => Record, +): ReadableStream { + 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(); + }, + }); +} diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 5396172082..6dd480ffe7 100644 --- a/src/server/responses/core.ts +++ b/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 | undefined; try { - let completed = JSON.parse(clientJson) as Record; + 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; // 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; + 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 diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 24535af60f..4b2736cd5d 100644 --- a/structure/04_transports-and-sidecars.md +++ b/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 diff --git a/tests/deepseek-responses-item-id-repair.test.ts b/tests/deepseek-responses-item-id-repair.test.ts index 44fe98a59c..ca021defca 100644 --- a/tests/deepseek-responses-item-id-repair.test.ts +++ b/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]"); + }); }); diff --git a/tests/responses-json-events.test.ts b/tests/responses-json-events.test.ts index e477f02497..9e27ffc5cc 100644 --- a/tests/responses-json-events.test.ts +++ b/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"); + }); });