From 7e95c02be53ba3edce9d19aaa60eb172d9200720 Mon Sep 17 00:00:00 2001 From: wangmiao0668000666 Date: Sat, 27 Jun 2026 08:54:09 +0800 Subject: [PATCH] fix(provider-transport-fetch): bound SSE buffer to prevent OOM (#96989) * fix(provider-transport-fetch): bound SSE buffer to prevent OOM * fix(provider-transport-fetch): appease oxlint curly rule in test * fix(provider-transport-fetch): drain events before cap + cancel reader on overflow * fix(provider-transport-fetch): remove unused encoder from coalesced chunk test Co-Authored-By: Claude * fix(transport): tighten SSE buffer guards --------- Co-authored-by: Claude Co-authored-by: Vincent Koc (cherry picked from commit 1bccd293043745dd0274f7537d4a730a7ba8d739) --- src/agents/provider-transport-fetch.test.ts | 127 ++++++++++++++++++++ src/agents/provider-transport-fetch.ts | 31 +++++ 2 files changed, 158 insertions(+) diff --git a/src/agents/provider-transport-fetch.test.ts b/src/agents/provider-transport-fetch.test.ts index 2665b2a1e002..78294feb1f30 100644 --- a/src/agents/provider-transport-fetch.test.ts +++ b/src/agents/provider-transport-fetch.test.ts @@ -1130,6 +1130,37 @@ describe("buildGuardedModelFetch", () => { expect(items).toEqual([{ ok: true }]); }); + it("handles a large transport chunk containing many valid small SSE events", async () => { + // Regression: one TCP read can deliver >64 KiB of already-delimited SSE + // events; the cap must apply only to the unterminated tail, not the full chunk. + const eventCount = 5_000; + const manyEvents = `data: ${JSON.stringify({ ok: true })}\n\n`.repeat(eventCount); + fetchWithSsrFGuardMock.mockResolvedValue({ + response: new Response(manyEvents, { + headers: { "content-type": "text/event-stream" }, + }), + finalUrl: "https://openrouter.ai/api/v1/chat/completions", + release: vi.fn(async () => undefined), + }); + const model = { + id: "gpt-5.4", + provider: "openrouter", + api: "openai-completions", + baseUrl: "https://openrouter.ai/api/v1", + } as unknown as Model<"openai-completions">; + + const response = await buildGuardedModelFetch(model)( + "https://openrouter.ai/api/v1/chat/completions", + { method: "POST" }, + ); + const items: unknown[] = []; + for await (const item of Stream.fromSSEResponse(response, new AbortController())) { + items.push(item); + } + expect(items.length).toBe(eventCount); + expect(items[0]).toEqual({ ok: true }); + }); + it("synthesizes SSE frames for JSON bodies returned to streaming OpenAI SDK requests", async () => { fetchWithSsrFGuardMock.mockResolvedValue({ response: new Response(' {"ok": true} ', { @@ -1338,6 +1369,102 @@ describe("buildGuardedModelFetch", () => { expect(refreshTimeout).toHaveBeenCalledTimes(2); }); + it("errors on oversized SSE body without event boundary in sanitizer", async () => { + const oversized = "x".repeat(65 * 1024); + const encoder = new TextEncoder(); + fetchWithSsrFGuardMock.mockResolvedValue({ + response: new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(oversized)); + controller.close(); + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ), + finalUrl: "https://openrouter.ai/api/v1/chat/completions", + release: vi.fn(async () => undefined), + }); + const model = { + id: "gpt-5.4", + provider: "openrouter", + api: "openai-completions", + baseUrl: "https://openrouter.ai/api/v1", + } as unknown as Model<"openai-completions">; + + const response = await buildGuardedModelFetch(model)( + "https://openrouter.ai/api/v1/chat/completions", + { method: "POST" }, + ); + + const reader = response.body?.getReader(); + let caught: unknown = null; + try { + while (true) { + const { done } = await reader!.read(); + if (done) { + break; + } + } + } catch (e) { + caught = e; + } + expect(caught).toBeTruthy(); + expect(String(caught)).toMatch(/exceeded max buffer size/i); + }); + + it("errors on oversized streaming JSON body without content-length in SSE synthesis", async () => { + const CHUNK = 1024 * 1024; + let sends = 0; + fetchWithSsrFGuardMock.mockResolvedValue({ + response: new Response( + new ReadableStream({ + pull(controller) { + if (sends < 17) { + sends++; + controller.enqueue(new Uint8Array(CHUNK)); + } else { + controller.close(); + } + }, + }), + { headers: { "content-type": "application/json" } }, + ), + finalUrl: "https://openrouter.ai/api/v1/chat/completions", + release: vi.fn(async () => undefined), + }); + const model = { + id: "moonshotai/kimi-k2.6", + provider: "openrouter", + api: "openai-completions", + baseUrl: "https://openrouter.ai/api/v1", + } as unknown as Model<"openai-completions">; + + const response = await buildGuardedModelFetch(model)( + "https://openrouter.ai/api/v1/chat/completions", + { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "moonshotai/kimi-k2.6", stream: true }), + }, + ); + + const reader = response.body?.getReader(); + let caught: unknown = null; + try { + while (true) { + const { done } = await reader!.read(); + if (done) { + break; + } + } + } catch (e) { + caught = e; + } + expect(caught).toBeTruthy(); + expect(String(caught)).toMatch(/exceeded.*bytes while synthesizing SSE/i); + }); + describe("long retry-after handling", () => { const anthropicModel = { id: "sonnet-4.6", diff --git a/src/agents/provider-transport-fetch.ts b/src/agents/provider-transport-fetch.ts index 775bb719343d..f8a22bb0752a 100644 --- a/src/agents/provider-transport-fetch.ts +++ b/src/agents/provider-transport-fetch.ts @@ -45,6 +45,17 @@ import { const DEFAULT_MAX_SDK_RETRY_WAIT_SECONDS = 60; const OPENAI_SDK_STREAM_CONTENT_SNIFF_BYTES = 2 * 1024; const log = createSubsystemLogger("provider-transport-fetch"); + +/** Max bytes for an entire JSON body synthesized into SSE frames. Prevents OOM + * when a hostile streaming endpoint returns a never-ending JSON response + * without Content-Length. */ +const SSE_SYNTHESIZE_JSON_MAX_BYTES = 16 * 1024 * 1024; + +/** Max bytes for the internal SSE sanitization buffer between event boundaries. + * A response that cannot find a \n\n boundary within this many characters is + * almost certainly hostile or broken — cap the buffer rather than let it grow. */ +const SSE_SANITIZE_BUFFER_MAX_BYTES = 64 * 1024; + const BLOCKED_EXACT_ORIGIN_TRUST_HOSTNAME_LABELS = new Set(["instance-data"]); const PLAIN_DECIMAL_NUMBER_RE = /^\d+(?:\.\d+)?$/; const RETRY_AFTER_HTTP_DATE_RE = @@ -102,6 +113,7 @@ function sanitizeOpenAISdkSseResponse( const encoder = new TextEncoder(); let reader: ReadableStreamDefaultReader | undefined; let buffer = ""; + let totalBytes = 0; const sseBody = new ReadableStream({ start() { reader = source.getReader(); @@ -120,9 +132,17 @@ function sanitizeOpenAISdkSseResponse( controller.close(); return; } + const nextTotalBytes = totalBytes + chunk.value.byteLength; + if (nextTotalBytes > SSE_SYNTHESIZE_JSON_MAX_BYTES) { + throw new Error( + `Streaming JSON body exceeded ${SSE_SYNTHESIZE_JSON_MAX_BYTES} bytes while synthesizing SSE frames`, + ); + } + totalBytes = nextTotalBytes; buffer += decoder.decode(chunk.value, { stream: true }); } } catch (error) { + await reader?.cancel(error).catch(() => {}); controller.error(error); } }, @@ -157,6 +177,11 @@ function sanitizeOpenAISdkSseResponse( for (;;) { const boundary = findSseEventBoundary(buffer); if (!boundary) { + if (buffer.length > SSE_SANITIZE_BUFFER_MAX_BYTES) { + throw new Error( + `SSE response exceeded max buffer size (${SSE_SANITIZE_BUFFER_MAX_BYTES} bytes) without event boundary`, + ); + } return enqueued; } const block = buffer.slice(0, boundary.index); @@ -167,6 +192,7 @@ function sanitizeOpenAISdkSseResponse( if (hasReadableSseData(block)) { controller.enqueue(encoder.encode(`${block}${separator}`)); enqueued += 1; + return enqueued; } } }; @@ -178,6 +204,10 @@ function sanitizeOpenAISdkSseResponse( async pull(controller) { try { for (;;) { + const pending = enqueueSanitized(controller, ""); + if (pending > 0) { + return; + } const chunk = await reader?.read(); if (!chunk || chunk.done) { const tail = decoder.decode(); @@ -200,6 +230,7 @@ function sanitizeOpenAISdkSseResponse( } } } catch (error) { + await reader?.cancel(error).catch(() => {}); controller.error(error); } },