From 66b9cc793161f6ba821c629286474671b3e874b8 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 11 Jul 2026 14:15:53 -0400 Subject: [PATCH] fix(client): accept larger SSE events (#36442) --- .../src/promise/generated/client-error.ts | 7 +++- .../client/src/promise/generated/client.ts | 4 +- packages/client/test/promise.test.ts | 37 +++++++++++++++++++ packages/httpapi-codegen/src/index.ts | 4 +- 4 files changed, 48 insertions(+), 4 deletions(-) diff --git a/packages/client/src/promise/generated/client-error.ts b/packages/client/src/promise/generated/client-error.ts index c278f0ddc8..930b612383 100644 --- a/packages/client/src/promise/generated/client-error.ts +++ b/packages/client/src/promise/generated/client-error.ts @@ -1,4 +1,9 @@ -export type ClientErrorReason = "Transport" | "UnexpectedStatus" | "UnsupportedContentType" | "MalformedResponse" +export type ClientErrorReason = + | "Transport" + | "UnexpectedStatus" + | "UnsupportedContentType" + | "MalformedResponse" + | "SseEventTooLarge" export class ClientError extends Error { override readonly name = "ClientError" diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index c030946153..bdfd5958fc 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -213,6 +213,8 @@ interface RequestDescriptor { readonly binary?: true } +const maxSseEventBytes = 16 * 1024 * 1024 + export function make(options: ClientOptions) { const fetch = options.fetch ?? globalThis.fetch @@ -289,7 +291,7 @@ export function make(options: ClientOptions) { throw new ClientError("Transport", { cause }) } buffer += decoder.decode(next.value, { stream: !next.done }) - if (buffer.length > 1_048_576) throw new ClientError("MalformedResponse") + if (buffer.length > maxSseEventBytes) throw new ClientError("SseEventTooLarge") const trailingCarriageReturn = !next.done && buffer.endsWith("\r") if (trailingCarriageReturn) buffer = buffer.slice(0, -1) buffer = buffer.replaceAll("\r\n", "\n").replaceAll("\r", "\n") diff --git a/packages/client/test/promise.test.ts b/packages/client/test/promise.test.ts index e29522c226..ba822f28e5 100644 --- a/packages/client/test/promise.test.ts +++ b/packages/client/test/promise.test.ts @@ -284,6 +284,43 @@ test("event.subscribe terminates on malformed Promise SSE data", async () => { }) }) +test("event.subscribe accepts a fragmented SSE event below the size limit", async () => { + const event = { id: "evt_large", type: "test.large", data: { output: "x".repeat(12 * 1024 * 1024) } } + const encoded = new TextEncoder().encode(`data: ${JSON.stringify(event)}\n\n`) + const client = OpenCode.make({ + baseUrl: "http://localhost:3000", + fetch: async () => + new Response( + new ReadableStream({ + start(controller) { + for (let offset = 0; offset < encoded.length; offset += 64 * 1024) { + controller.enqueue(encoded.slice(offset, offset + 64 * 1024)) + } + controller.close() + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ), + }) + + await expect(client.event.subscribe()[Symbol.asyncIterator]().next()).resolves.toEqual({ done: false, value: event }) +}) + +test("event.subscribe rejects an SSE event above the size limit", async () => { + const client = OpenCode.make({ + baseUrl: "http://localhost:3000", + fetch: async () => + new Response(`data: ${JSON.stringify({ output: "x".repeat(16 * 1024 * 1024) })}`, { + headers: { "content-type": "text/event-stream" }, + }), + }) + + await expect(client.event.subscribe()[Symbol.asyncIterator]().next()).rejects.toMatchObject({ + name: "ClientError", + reason: "SseEventTooLarge", + }) +}) + test("session methods use the public HTTP contract", async () => { const requests: Array<{ url: string; init?: RequestInit }> = [] const client = OpenCode.make({ diff --git a/packages/httpapi-codegen/src/index.ts b/packages/httpapi-codegen/src/index.ts index 069da2c1e7..3cb1a69633 100644 --- a/packages/httpapi-codegen/src/index.ts +++ b/packages/httpapi-codegen/src/index.ts @@ -309,7 +309,7 @@ export function emitPromise( { path: "types.ts", content: renderPromiseTypes(groups, options?.outputTypes, options?.mutableOutputs ?? false) }, { path: "client-error.ts", - content: `export type ClientErrorReason = "Transport" | "UnexpectedStatus" | "UnsupportedContentType" | "MalformedResponse"\n\nexport class ClientError extends Error {\n override readonly name = "ClientError"\n constructor(readonly reason: ClientErrorReason, options?: ErrorOptions) {\n super(reason, options)\n }\n}\n`, + content: `export type ClientErrorReason = "Transport" | "UnexpectedStatus" | "UnsupportedContentType" | "MalformedResponse" | "SseEventTooLarge"\n\nexport class ClientError extends Error {\n override readonly name = "ClientError"\n constructor(readonly reason: ClientErrorReason, options?: ErrorOptions) {\n super(reason, options)\n }\n}\n`, }, { path: "client.ts", @@ -722,7 +722,7 @@ function renderPromiseClient(groups: ReadonlyArray) { if (group.endpoints[0]?.topLevel) return fields return `${JSON.stringify(group.identifier)}: { ${fields} }` }) - return `import type { ${imports.join(", ")} } from "./types"\nimport { ClientError } from "./client-error"\n\nexport interface ClientOptions {\n readonly baseUrl: string\n readonly fetch?: typeof globalThis.fetch\n readonly headers?: RequestInit["headers"]\n}\n\nexport interface RequestOptions {\n readonly signal?: AbortSignal\n readonly headers?: RequestInit["headers"]\n}\n\ninterface RequestDescriptor {\n readonly method: string\n readonly path: string\n readonly query?: Record\n readonly headers?: Record\n readonly body?: unknown\n readonly successStatus: number\n readonly declaredStatuses: ReadonlyArray\n readonly empty: boolean\n}\n\nexport function make(options: ClientOptions) {\n const fetch = options.fetch ?? globalThis.fetch\n\n const prepare = (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n const url = new URL(descriptor.path, options.baseUrl)\n for (const [key, value] of Object.entries(descriptor.query ?? {})) appendQuery(url.searchParams, key, value)\n const headers = new Headers(options.headers)\n for (const [key, value] of Object.entries(descriptor.headers ?? {})) {\n if (value !== undefined && value !== null) headers.set(key, String(value))\n }\n for (const [key, value] of new Headers(requestOptions?.headers)) headers.set(key, value)\n if (descriptor.body !== undefined && !headers.has("content-type")) headers.set("content-type", "application/json")\n return {\n url,\n init: {\n method: descriptor.method,\n signal: requestOptions?.signal,\n headers,\n body: descriptor.body === undefined ? undefined : JSON.stringify(descriptor.body),\n } satisfies RequestInit,\n }\n }\n\n const execute = async (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n try {\n const prepared = prepare(descriptor, requestOptions)\n return await fetch(prepared.url, prepared.init)\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n }\n\n const responseError = async (response: Response, descriptor: RequestDescriptor): Promise => {\n if (descriptor.declaredStatuses.includes(response.status)) throw await json(response)\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnexpectedStatus", { cause: { status: response.status } })\n }\n\n const request = async (descriptor: RequestDescriptor, requestOptions?: RequestOptions): Promise => {\n const response = await execute(descriptor, requestOptions)\n if (response.status !== descriptor.successStatus) return responseError(response, descriptor)\n if (descriptor.empty) {\n try {\n await response.body?.cancel()\n } catch {}\n return undefined as A\n }\n return await json(response) as A\n }\n\n const sse = (descriptor: RequestDescriptor, requestOptions?: RequestOptions): AsyncIterable => ({\n async *[Symbol.asyncIterator]() {\n const response = await execute(descriptor, requestOptions)\n if (response.status !== descriptor.successStatus) await responseError(response, descriptor)\n if (!isContentType(response, "text/event-stream")) {\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnsupportedContentType")\n }\n if (response.body === null) throw new ClientError("MalformedResponse")\n const reader = response.body.getReader()\n const decoder = new TextDecoder()\n let buffer = ""\n try {\n while (true) {\n let next: ReadableStreamReadResult\n try {\n next = await reader.read()\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n buffer += decoder.decode(next.value, { stream: !next.done })\n if (buffer.length > 1_048_576) throw new ClientError("MalformedResponse")\n const trailingCarriageReturn = !next.done && buffer.endsWith("\\r")\n if (trailingCarriageReturn) buffer = buffer.slice(0, -1)\n buffer = buffer.replaceAll("\\r\\n", "\\n").replaceAll("\\r", "\\n")\n if (trailingCarriageReturn) buffer += "\\r"\n if (next.done && buffer !== "") buffer += "\\n\\n"\n let boundary = buffer.indexOf("\\n\\n")\n while (boundary >= 0) {\n const block = buffer.slice(0, boundary)\n buffer = buffer.slice(boundary + 2)\n const data = block.split("\\n").flatMap((line) => line.startsWith("data:") ? [line.slice(5).trimStart()] : []).join("\\n")\n if (data !== "") {\n try {\n yield JSON.parse(data) as A\n } catch (cause) {\n throw new ClientError("MalformedResponse", { cause })\n }\n }\n boundary = buffer.indexOf("\\n\\n")\n }\n if (next.done) return\n }\n } finally {\n try {\n await reader.cancel()\n } catch {}\n reader.releaseLock()\n }\n },\n })\n\n return { ${fields.join(", ")} }\n}\n\nfunction appendQuery(params: URLSearchParams, key: string, value: unknown): void {\n if (value === undefined) return\n if (value === null) {\n params.append(key, "null")\n return\n }\n if (Array.isArray(value)) {\n for (const item of value) appendQuery(params, key, item)\n return\n }\n if (typeof value === "object") {\n for (const [child, item] of Object.entries(value)) appendQuery(params, \`\${key}[\${child}]\`, item)\n return\n }\n params.append(key, String(value))\n}\n\nasync function json(response: Response): Promise {\n if (!isContentType(response, "application/json") && !response.headers.get("content-type")?.includes("+json")) {\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnsupportedContentType")\n }\n let text: string\n try {\n text = await response.text()\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n if (text === "") throw new ClientError("MalformedResponse")\n try {\n return JSON.parse(text)\n } catch (cause) {\n throw new ClientError("MalformedResponse", { cause })\n }\n}\n\nfunction isContentType(response: Response, expected: string) {\n return response.headers.get("content-type")?.split(";", 1)[0]?.trim().toLowerCase() === expected\n}\n` + return `import type { ${imports.join(", ")} } from "./types"\nimport { ClientError } from "./client-error"\n\nexport interface ClientOptions {\n readonly baseUrl: string\n readonly fetch?: typeof globalThis.fetch\n readonly headers?: RequestInit["headers"]\n}\n\nexport interface RequestOptions {\n readonly signal?: AbortSignal\n readonly headers?: RequestInit["headers"]\n}\n\ninterface RequestDescriptor {\n readonly method: string\n readonly path: string\n readonly query?: Record\n readonly headers?: Record\n readonly body?: unknown\n readonly successStatus: number\n readonly declaredStatuses: ReadonlyArray\n readonly empty: boolean\n}\n\nconst maxSseEventBytes = 16 * 1024 * 1024\n\nexport function make(options: ClientOptions) {\n const fetch = options.fetch ?? globalThis.fetch\n\n const prepare = (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n const url = new URL(descriptor.path, options.baseUrl)\n for (const [key, value] of Object.entries(descriptor.query ?? {})) appendQuery(url.searchParams, key, value)\n const headers = new Headers(options.headers)\n for (const [key, value] of Object.entries(descriptor.headers ?? {})) {\n if (value !== undefined && value !== null) headers.set(key, String(value))\n }\n for (const [key, value] of new Headers(requestOptions?.headers)) headers.set(key, value)\n if (descriptor.body !== undefined && !headers.has("content-type")) headers.set("content-type", "application/json")\n return {\n url,\n init: {\n method: descriptor.method,\n signal: requestOptions?.signal,\n headers,\n body: descriptor.body === undefined ? undefined : JSON.stringify(descriptor.body),\n } satisfies RequestInit,\n }\n }\n\n const execute = async (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n try {\n const prepared = prepare(descriptor, requestOptions)\n return await fetch(prepared.url, prepared.init)\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n }\n\n const responseError = async (response: Response, descriptor: RequestDescriptor): Promise => {\n if (descriptor.declaredStatuses.includes(response.status)) throw await json(response)\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnexpectedStatus", { cause: { status: response.status } })\n }\n\n const request = async (descriptor: RequestDescriptor, requestOptions?: RequestOptions): Promise => {\n const response = await execute(descriptor, requestOptions)\n if (response.status !== descriptor.successStatus) return responseError(response, descriptor)\n if (descriptor.empty) {\n try {\n await response.body?.cancel()\n } catch {}\n return undefined as A\n }\n return await json(response) as A\n }\n\n const sse = (descriptor: RequestDescriptor, requestOptions?: RequestOptions): AsyncIterable => ({\n async *[Symbol.asyncIterator]() {\n const response = await execute(descriptor, requestOptions)\n if (response.status !== descriptor.successStatus) await responseError(response, descriptor)\n if (!isContentType(response, "text/event-stream")) {\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnsupportedContentType")\n }\n if (response.body === null) throw new ClientError("MalformedResponse")\n const reader = response.body.getReader()\n const decoder = new TextDecoder()\n let buffer = ""\n try {\n while (true) {\n let next: ReadableStreamReadResult\n try {\n next = await reader.read()\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n buffer += decoder.decode(next.value, { stream: !next.done })\n if (buffer.length > maxSseEventBytes) throw new ClientError("SseEventTooLarge")\n const trailingCarriageReturn = !next.done && buffer.endsWith("\\r")\n if (trailingCarriageReturn) buffer = buffer.slice(0, -1)\n buffer = buffer.replaceAll("\\r\\n", "\\n").replaceAll("\\r", "\\n")\n if (trailingCarriageReturn) buffer += "\\r"\n if (next.done && buffer !== "") buffer += "\\n\\n"\n let boundary = buffer.indexOf("\\n\\n")\n while (boundary >= 0) {\n const block = buffer.slice(0, boundary)\n buffer = buffer.slice(boundary + 2)\n const data = block.split("\\n").flatMap((line) => line.startsWith("data:") ? [line.slice(5).trimStart()] : []).join("\\n")\n if (data !== "") {\n try {\n yield JSON.parse(data) as A\n } catch (cause) {\n throw new ClientError("MalformedResponse", { cause })\n }\n }\n boundary = buffer.indexOf("\\n\\n")\n }\n if (next.done) return\n }\n } finally {\n try {\n await reader.cancel()\n } catch {}\n reader.releaseLock()\n }\n },\n })\n\n return { ${fields.join(", ")} }\n}\n\nfunction appendQuery(params: URLSearchParams, key: string, value: unknown): void {\n if (value === undefined) return\n if (value === null) {\n params.append(key, "null")\n return\n }\n if (Array.isArray(value)) {\n for (const item of value) appendQuery(params, key, item)\n return\n }\n if (typeof value === "object") {\n for (const [child, item] of Object.entries(value)) appendQuery(params, \`\${key}[\${child}]\`, item)\n return\n }\n params.append(key, String(value))\n}\n\nasync function json(response: Response): Promise {\n if (!isContentType(response, "application/json") && !response.headers.get("content-type")?.includes("+json")) {\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnsupportedContentType")\n }\n let text: string\n try {\n text = await response.text()\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n if (text === "") throw new ClientError("MalformedResponse")\n try {\n return JSON.parse(text)\n } catch (cause) {\n throw new ClientError("MalformedResponse", { cause })\n }\n}\n\nfunction isContentType(response: Response, expected: string) {\n return response.headers.get("content-type")?.split(";", 1)[0]?.trim().toLowerCase() === expected\n}\n` } function promiseTypePrefix(group: string, path: ReadonlyArray) {