diff --git a/docs-site/src/content/docs/ja/reference/proxy-formats.md b/docs-site/src/content/docs/ja/reference/proxy-formats.md index e01011047..e5cb35132 100644 --- a/docs-site/src/content/docs/ja/reference/proxy-formats.md +++ b/docs-site/src/content/docs/ja/reference/proxy-formats.md @@ -53,6 +53,8 @@ provider events → internal adapter events → client dialect `stream: false` を指定するか、`stream` を指定しないと、同じアダプター イベントが 1 つの Responses JSON オブジェクトに収集されます。どちらの形式でも、選択したモデル、出力項目、端末の状態、使用状況が保存されます。 +クライアント向け Responses SSE フレームは、SSE ブロック区切りの前の生バイトで測って 1 フレームあたり 4 MiB に制限されます。HTTP では、区切りなしでこの上限を超えたアップストリーム フレームは、合成 `response.failed` イベントと続く `data: [DONE]` でフェイルクローズします。Responses WebSocket ブリッジでは、同じ条件で 502 `websocket_protocol_error` を送信し、アップストリーム リーダーをキャンセルします。完全な Responses 終端フレームがすでに到着している場合はそれが優先され、その後のサイズ超過または不正なバイトは、完了したターンをトランスポート障害に置き換えず破棄されます。 + すべての端末応答使用状況オブジェクトには、プロバイダーが詳細を報告しなかった場合でも、両方の詳細オブジェクトが含まれます。 ```json @@ -211,4 +213,4 @@ Anthropic オリジンの失敗は Anthropic のエラー エンベロープで プロキシは、本物のバックエンド暗号文を不透明なものとして扱います。構造的に有効な暗号文はバイト単位で保存されます。opencodex は暗号文を復号したり、その内容を変換したり、別のプロバイダー用に再暗号化したりしません。 -一部のエージェント フックはこれまで、プレーンテキストの制御テキストを `encrypted_content` スロットに配置していました。互換性を確保するために、プロキシは、構造的に有効な Fernet の実行を変更せずに保持しながら、プレーンテキストをテキスト部分に分割します。 `agent_message` が修復中にすべての暗号化された部分を失った場合、それは通常のユーザー メッセージになります。現在の v2 タスクが完全に暗号化されたままであるが、選択したルーティングされたターゲットがネイティブ ChatGPT 暗号文を読み取ることができない場合、opencodex は読み取り不能なバイトをそのプロバイダーに送信する代わりに `unreadable_encrypted_agent_task` で失敗します。ワーカー タスクに関するクライアントの動作については、[サブエージェントサーフェス](/guides/sub-agent-surface/) を参照してください。 +一部のエージェント フックはこれまで、プレーンテキストの制御テキストを `encrypted_content` スロットに配置していました。互換性を確保するために、プロキシは、構造的に有効な Fernet の実行を変更せずに保持しながら、プレーンテキストをテキスト部分に分割します。 `agent_message` が修復中にすべての暗号化された部分を失った場合、それは通常のユーザー メッセージになります。現在の v2 タスクが完全に暗号化されたままであるが、選択したルーティングされたターゲットがネイティブ ChatGPT 暗号文を読み取ることができない場合、opencodex は読み取り不能なバイトをそのプロバイダーに送信する代わりに `unreadable_encrypted_agent_task` で失敗します。ワーカー タスクに関するクライアントの動作については、[サブエージェントサーフェス](/guides/sub-agent-surface/) を参照してください。 \ No newline at end of file diff --git a/docs-site/src/content/docs/ko/reference/proxy-formats.md b/docs-site/src/content/docs/ko/reference/proxy-formats.md index de1f19bda..066dc99b4 100644 --- a/docs-site/src/content/docs/ko/reference/proxy-formats.md +++ b/docs-site/src/content/docs/ko/reference/proxy-formats.md @@ -63,6 +63,8 @@ deltas, 그리고 정확히 하나의 종료 `response.completed`, `response.fai `stream: false`이거나 `stream`이 없으면, 같은 adapter 이벤트가 하나의 Responses JSON 객체로 수집됩니다. 두 형식 모두 선택한 모델, output item, 종료 상태, usage를 보존합니다. +클라이언트로 전달되는 Responses SSE 프레임은 SSE 블록 구분자 앞의 원시 바이트 기준으로 프레임당 4 MiB로 제한됩니다. HTTP에서는 구분자 없이 이 한도를 초과한 업스트림 프레임을 합성 `response.failed` 이벤트와 이어지는 `data: [DONE]`으로 fail closed 처리합니다. Responses WebSocket 브리지에서는 같은 조건에서 502 `websocket_protocol_error`를 보내고 업스트림 reader를 취소합니다. 완전한 Responses 종료 프레임이 이미 수신된 경우에는 그 종료가 우선하며, 이후의 과도한 크기 또는 잘못된 바이트는 완료된 턴을 전송 오류로 바꾸지 않고 버립니다. + 모든 종료 Responses usage 객체에는 제공자가 해당 세부 정보를 보고하지 않았더라도 두 상세 객체가 모두 포함됩니다. @@ -263,4 +265,4 @@ OpenAI 스타일 `origin_rejected` body가 아니라 403 `permission_error`입 작업이 실제로 암호화된 상태이고 선택된 라우팅 대상이 네이티브 ChatGPT 암호문을 읽을 수 없다면, opencodex는 읽을 수 없는 바이트를 프로바이더에 보내는 대신 `unreadable_encrypted_agent_task`로 실패합니다. worker task와 관련된 클라이언트 동작은 [서브에이전트 표면](/guides/sub-agent-surface/)을 -참조하세요. +참조하세요. \ No newline at end of file diff --git a/docs-site/src/content/docs/reference/proxy-formats.md b/docs-site/src/content/docs/reference/proxy-formats.md index 5e0996ef2..0ae53ad02 100644 --- a/docs-site/src/content/docs/reference/proxy-formats.md +++ b/docs-site/src/content/docs/reference/proxy-formats.md @@ -65,6 +65,13 @@ With `stream: true`, the response is `text/event-stream`. The bridge emits Respo With `stream: false` or no `stream`, the same adapter events are collected into one Responses JSON object. Both forms preserve the selected model, output items, terminal status, and usage. +Client-facing Responses SSE frames are limited to 4 MiB per frame, measured in raw bytes before the +SSE block delimiter. On HTTP, an unterminated upstream frame that exceeds the limit fails closed +with a synthetic `response.failed` event followed by `data: [DONE]`. On the Responses WebSocket +bridge, the same condition emits a 502 `websocket_protocol_error` and cancels the upstream reader. +A complete Responses terminal frame is authoritative: oversized or malformed trailing bytes after +that terminal are dropped rather than replacing the completed turn with a transport failure. + Every terminal Responses usage object includes both detail objects, even when the provider did not report those details: @@ -280,4 +287,4 @@ structurally valid Fernet runs unchanged. If an `agent_message` loses all encryp that repair, it becomes a normal user message. If a current v2 task remains genuinely encrypted but the selected routed target cannot read native ChatGPT ciphertext, opencodex fails with `unreadable_encrypted_agent_task` instead of sending unreadable bytes to that provider. See -[Sub-agent Surface](/guides/sub-agent-surface/) for the client behavior around worker tasks. +[Sub-agent Surface](/guides/sub-agent-surface/) for the client behavior around worker tasks. \ No newline at end of file diff --git a/docs-site/src/content/docs/ru/reference/proxy-formats.md b/docs-site/src/content/docs/ru/reference/proxy-formats.md index 8b7dc537d..e2d4d3fa3 100644 --- a/docs-site/src/content/docs/ru/reference/proxy-formats.md +++ b/docs-site/src/content/docs/ru/reference/proxy-formats.md @@ -67,6 +67,8 @@ Translated-adapter'ы обрабатывают только известные При `stream: false` или при отсутствии `stream` те же события адаптера собираются в один JSON Responses. Обе формы сохраняют выбранную модель, output item'ы, terminal status и usage. +Клиентские frame'ы Responses SSE ограничены 4 MiB на frame, считая сырые байты до разделителя SSE-блока. В HTTP незавершённый upstream-frame, превысивший этот предел, завершается fail-closed синтетическим событием `response.failed`, после которого идёт `data: [DONE]`. В мосте Responses WebSocket то же условие даёт 502 `websocket_protocol_error` и отменяет upstream-reader. Если полноценный terminal-frame Responses уже получен, он остаётся авторитетным: слишком большие или некорректные байты после него отбрасываются и не заменяют завершённый ход транспортной ошибкой. + Каждый terminal usage-объект Responses всегда включает оба detail-объекта, даже если провайдер их не сообщил: @@ -281,4 +283,4 @@ ciphertext сохраняется байт в байт: opencodex его не р остаётся по-настоящему зашифрованной, а выбранная routed-цель не умеет читать ciphertext нативного ChatGPT, opencodex завершит запрос ошибкой `unreadable_encrypted_agent_task`, вместо того чтобы отправить нечитаемые байты этому провайдеру. О поведении клиента вокруг worker-task'ов см. -[Поверхность подагентов](/guides/sub-agent-surface/). +[Поверхность подагентов](/guides/sub-agent-surface/). \ No newline at end of file diff --git a/docs-site/src/content/docs/zh-cn/reference/proxy-formats.md b/docs-site/src/content/docs/zh-cn/reference/proxy-formats.md index 1f3c2cfba..9b9cbc8aa 100644 --- a/docs-site/src/content/docs/zh-cn/reference/proxy-formats.md +++ b/docs-site/src/content/docs/zh-cn/reference/proxy-formats.md @@ -62,6 +62,8 @@ Responses 表示是这座桥的中心。原生兼容的路由可以跳过部分 当 `stream: false` 或未提供 `stream` 时，同样的适配器事件会被收集为一个 Responses JSON 对象。两种形式都会保留所选模型、输出项、终止状态和 usage。 +面向客户端的 Responses SSE 帧按 SSE 块分隔符之前的原始字节计算，每帧限制为 4 MiB。对于 HTTP，未终止的上游帧一旦超过该限制，会以合成的 `response.failed` 事件并随后发送 `data: [DONE]` 的方式 fail closed。对于 Responses WebSocket 桥，相同情况会发送 502 `websocket_protocol_error` 并取消上游 reader。已经完整到达的 Responses 终止帧具有优先权；其后的超大或格式错误字节会被丢弃，而不会把已经完成的轮次替换为传输失败。 + 每个终止的 Responses usage 对象都包含两个 detail 对象，即使提供方没有报告这些细节： ```json @@ -235,4 +237,4 @@ Anthropic 来源的失败会以 Anthropic 的错误封装呈现，因此该方 代理把真正的后端密文视为不透明数据。结构有效的密文会逐字节保留：opencodex 不会对其解密、翻译其内容，或为另一个提供方重新加密。 某些 agent hook 历史上会把明文控制文本放进 `encrypted_content` 槽。为兼容起见，代理会把那部分明文拆分为文本片段，同时保持任何结构有效的 Fernet 片段不变。如果一个 `agent_message` 在该修复过程中失去了所有加密部分，它就会变成普通的 user message。如果当前的 v2 task 仍然真的是加密的，但所选路由目标无法读取原生 ChatGPT 密文，opencodex 会以 -`unreadable_encrypted_agent_task` 失败，而不是把不可读字节发送给该提供方。有关 worker task 周边的客户端行为，请参见 [Sub-agent Surface](/guides/sub-agent-surface/)。 +`unreadable_encrypted_agent_task` 失败，而不是把不可读字节发送给该提供方。有关 worker task 周边的客户端行为，请参见 [Sub-agent Surface](/guides/sub-agent-surface/)。 \ No newline at end of file diff --git a/src/server/relay.ts b/src/server/relay.ts index 2378c02c6..a456c6124 100644 --- a/src/server/relay.ts +++ b/src/server/relay.ts @@ -11,11 +11,16 @@ import { type RequestLogContext, type RequestLogEntry, } from "./request-log"; +import { + BoundedSseFrameBuffer, + joinSseFrameBytes, + MAX_CLIENT_SSE_FRAME_BYTES, +} from "./sse-frame-buffer"; const nativePassthroughSseResponses = new WeakSet(); const eagerRelaySseResponses = new WeakSet(); -export const MAX_INSPECTION_SSE_FRAME_BYTES = 4 * 1024 * 1024; +export const MAX_INSPECTION_SSE_FRAME_BYTES = MAX_CLIENT_SSE_FRAME_BYTES; export const MAX_COMPLETED_OUTPUT_ITEMS = 256; export const MAX_COMPLETED_OUTPUT_ITEM_SOURCE_BYTES = 8 * 1024 * 1024; export const MAX_TAIL_ERROR_MESSAGE_CHARS = 512; @@ -104,30 +109,32 @@ export type SseTerminalOutputBoundary = { /** * Frame-aware client output boundary shared by both native Responses relays. - * It buffers only the current incomplete SSE block, forwards complete blocks - * through the first Responses terminal, and drops every later block/byte. + * It buffers only the current incomplete SSE block under the same hard byte + * cap as inspection, forwards complete blocks through the first Responses + * terminal, and drops every later block/byte. */ export function createSseTerminalOutputBoundary(): SseTerminalOutputBoundary { - let decoder: TextDecoder | null = new TextDecoder(); - const encoder = new TextEncoder(); - let buffer = ""; + const decoder = new TextDecoder(); + const framer = new BoundedSseFrameBuffer(MAX_INSPECTION_SSE_FRAME_BYTES); let terminal = false; let done = false; let disposed = false; - const process = (flush: boolean): Uint8Array => { - if (disposed || terminal) return new Uint8Array(0); - let output = ""; + const processFrames = ( + frames: ReturnType, + ): Uint8Array => { + if (disposed || terminal || frames.length === 0) return new Uint8Array(0); + const output: Uint8Array[] = []; let responsesTerminal = false; - for (;;) { - const next = nextSseBlock(buffer); - if (!next) break; - buffer = next.rest; - const payload = sseDataPayload(next.block); - if (!responsesTerminal) output += next.block + next.delimiter; - if (payload === "[DONE]") { + for (const frame of frames) { + const payload = sseDataPayload(decoder.decode(frame.block)); + const isDone = payload === "[DONE]"; + // Preserve every frame through the first Responses terminal. A [DONE] + // frame is also preserved when it immediately follows that terminal in + // the same upstream chunk; every later non-DONE frame is dropped. + if (!responsesTerminal || isDone) output.push(frame.block, frame.delimiter); + if (isDone) { done = true; - if (responsesTerminal) output += next.block + next.delimiter; continue; } if (!responsesTerminal && payload && terminalStatusFromSsePayload(payload)) { @@ -136,33 +143,26 @@ export function createSseTerminalOutputBoundary(): SseTerminalOutputBoundary { } if (responsesTerminal) { terminal = true; - buffer = ""; + framer.dispose(); } - if (flush && !terminal && buffer.length > 0) { - output += buffer; - buffer = ""; - } - return encoder.encode(output); + return joinSseFrameBytes(output); }; return { feed(chunk) { if (disposed || terminal) return new Uint8Array(0); - buffer += decoder!.decode(chunk, { stream: true }); - return process(false); + return processFrames(framer.feed(chunk)); }, finish() { if (disposed || terminal) return new Uint8Array(0); - buffer += decoder!.decode(); - return process(true); + return framer.finish(); }, terminalSeen: () => terminal, doneSeen: () => done, dispose() { if (disposed) return; disposed = true; - decoder = null; - buffer = ""; + framer.dispose(); }, }; } @@ -227,7 +227,14 @@ export function relaySseWithFailedTail( if (result !== "buffered") return; } } catch (err) { - const partial = terminalBoundary.finish(); + let partial: Uint8Array = new Uint8Array(0); + try { + partial = terminalBoundary.finish(); + } catch { + // A near-cap ambiguous delimiter tail may itself overflow at EOF. + // Preserve the original read/framing failure and continue emitting + // the bounded failed tail instead of letting cleanup throw again. + } terminalBoundary.dispose(); if (closed) return; const payload = buildFailedTailPayload(err); @@ -622,17 +629,6 @@ function delimiterLengthAt( return byteAt(index + 3) === 10 ? 4 : 0; } -function joinedBytes(slices: readonly Uint8Array[], byteLength: number): Uint8Array { - if (slices.length === 1 && slices[0]!.byteLength === byteLength) return slices[0]!; - const joined = new Uint8Array(byteLength); - let offset = 0; - for (const slice of slices) { - joined.set(slice, offset); - offset += slice.byteLength; - } - return joined; -} - /** * Per-chunk SSE inspection state machine shared by consumeForInspection, * consumeForResponseLogMetadata, and the eager bounded relay (relay-eager.ts). @@ -653,8 +649,8 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector let reported = false; let sawTerminal = false; let disposed = false; - let delimiterTail = new Uint8Array(0); - let candidateSlices: Uint8Array[] = []; + let delimiterTail: Uint8Array = new Uint8Array(0); + let candidate: Uint8Array = new Uint8Array(0); let candidateBytes = 0; let discardingOversizedFrame = false; const reportFirstOutput = createFirstOutputReporter(handlers.onFirstOutput); @@ -668,7 +664,7 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector const clearFrameState = (): void => { delimiterTail = new Uint8Array(0); - candidateSlices = []; + candidate = new Uint8Array(0); candidateBytes = 0; discardingOversizedFrame = false; }; @@ -688,6 +684,30 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector firstResponseId = undefined; }; + const ensureCandidateCapacity = (requiredBytes: number): void => { + if (candidate.byteLength >= requiredBytes) return; + let capacity = candidate.byteLength === 0 + ? Math.min(MAX_INSPECTION_SSE_FRAME_BYTES, Math.max(requiredBytes, 4096)) + : candidate.byteLength; + while (capacity < requiredBytes) { + capacity = Math.min( + MAX_INSPECTION_SSE_FRAME_BYTES, + Math.max(requiredBytes, capacity * 2), + ); + } + const grown = new Uint8Array(capacity); + if (candidateBytes > 0) grown.set(candidate.subarray(0, candidateBytes)); + candidate = grown; + }; + + const takeCandidate = (): Uint8Array => { + if (candidateBytes === 0) return new Uint8Array(0); + const frame = candidate.slice(0, candidateBytes); + candidate = new Uint8Array(0); + candidateBytes = 0; + return frame; + }; + const retainCandidateSlice = (slice: Uint8Array): void => { if (slice.byteLength === 0 || discardingOversizedFrame) return; const nextBytes = candidateBytes + slice.byteLength; @@ -696,7 +716,7 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector Math.min(nextBytes, MAX_INSPECTION_SSE_FRAME_BYTES), ); if (nextBytes > MAX_INSPECTION_SSE_FRAME_BYTES) { - candidateSlices = []; + candidate = new Uint8Array(0); candidateBytes = 0; discardingOversizedFrame = true; inspectionCounters.frameCapOverflows += 1; @@ -706,10 +726,8 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector reconstructionTainted = true; return; } - // `subarray()` aliases the upstream chunk's backing buffer. Copy only the - // live candidate bytes so a tiny trailing frame cannot pin a multi-MiB - // chunk whose preceding frames have already been consumed. - candidateSlices.push(slice.slice()); + ensureCandidateCapacity(nextBytes); + candidate.set(slice, candidateBytes); candidateBytes = nextBytes; }; @@ -845,9 +863,7 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector return; } const sourceBytes = candidateBytes; - const frame = joinedBytes(candidateSlices, sourceBytes); - candidateSlices = []; - candidateBytes = 0; + const frame = takeCandidate(); if (reported && !handlers.onCompletedResponse) return; const decoded = decoder!.decode(frame); scanPayload(sseDataPayload(decoded), sourceBytes); @@ -904,7 +920,7 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector delimiterTail = new Uint8Array(0); if (!discardingOversizedFrame && candidateBytes > 0 && !reported) { const sourceBytes = candidateBytes; - const decoded = decoder!.decode(joinedBytes(candidateSlices, sourceBytes)); + const decoded = decoder!.decode(takeCandidate()); scanPayload(decoded.trim() ? sseDataPayload(decoded) : null, sourceBytes); } } finally { diff --git a/src/server/sse-frame-buffer.ts b/src/server/sse-frame-buffer.ts new file mode 100644 index 000000000..175e7bcab --- /dev/null +++ b/src/server/sse-frame-buffer.ts @@ -0,0 +1,292 @@ +export const MAX_CLIENT_SSE_FRAME_BYTES = 4 * 1024 * 1024; + +const LF_LF = Uint8Array.of(10, 10); +const LF_CR_LF = Uint8Array.of(10, 13, 10); +const CR_LF_LF = Uint8Array.of(13, 10, 10); +const CR_LF_CR_LF = Uint8Array.of(13, 10, 13, 10); + +export class SseFrameTooLargeError extends Error { + readonly maxBytes: number; + + constructor(maxBytes: number) { + super(`upstream SSE frame exceeded ${maxBytes} bytes`); + this.name = "SseFrameTooLargeError"; + this.maxBytes = maxBytes; + } +} + +export class SseFrameCountLimitError extends Error { + readonly maxFrames: number; + + constructor(maxFrames: number) { + super(`upstream SSE chunk exceeded ${maxFrames} frame limit`); + this.name = "SseFrameCountLimitError"; + this.maxFrames = maxFrames; + } +} + +export type BoundedSseFrame = { + block: Uint8Array; + delimiter: Uint8Array; +}; + +/** + * Classify the bytes at `index` as an SSE block delimiter. + * + * Returns the delimiter length in bytes, `0` when `index` does not start a + * delimiter, and `undefined` when more bytes are required to decide. + */ +function delimiterLengthAt( + index: number, + length: number, + byteAt: (index: number) => number, +): number | undefined { + const first = byteAt(index); + if (first === 10) { + if (index + 1 >= length) return undefined; + const second = byteAt(index + 1); + if (second === 10) return 2; + if (second !== 13) return 0; + if (index + 2 >= length) return undefined; + return byteAt(index + 2) === 10 ? 3 : 0; + } + if (first !== 13) return 0; + if (index + 1 >= length) return undefined; + if (byteAt(index + 1) !== 10) return 0; + if (index + 2 >= length) return undefined; + const third = byteAt(index + 2); + if (third === 10) return 3; + if (third !== 13) return 0; + if (index + 3 >= length) return undefined; + return byteAt(index + 3) === 10 ? 4 : 0; +} + +function delimiterBytesAt( + index: number, + delimiterLength: number, + byteAt: (index: number) => number, +): Uint8Array { + if (delimiterLength === 2) return LF_LF; + if (delimiterLength === 4) return CR_LF_CR_LF; + return byteAt(index) === 10 ? LF_CR_LF : CR_LF_LF; +} + +function copyRange( + start: number, + end: number, + tailLength: number, + previousTail: Uint8Array, + chunk: Uint8Array, +): Uint8Array { + const out = new Uint8Array(end - start); + for (let index = start; index < end; index += 1) { + out[index - start] = index < tailLength + ? previousTail[index]! + : chunk[index - tailLength]!; + } + return out; +} + +/** + * True when a complete SSE block already commits a Responses terminal event. + * + * Framing errors in bytes *after* such a block must not retroactively turn an + * already-completed/failed/incomplete model turn into a transport failure. This + * helper is used only on the exceptional path, so decoding/JSON parsing has no + * cost on ordinary framing. + */ +function isResponsesTerminalFrame(block: Uint8Array): boolean { + const data: string[] = []; + for (const line of new TextDecoder().decode(block).split(/\r?\n/)) { + if (!line.startsWith("data:")) continue; + const value = line.slice(5); + data.push(value.startsWith(" ") ? value.slice(1) : value); + } + if (data.length === 0) return false; + const payload = data.join("\n"); + if (payload === "[DONE]") return false; + try { + const parsed = JSON.parse(payload) as { type?: unknown }; + return parsed.type === "response.completed" + || parsed.type === "response.failed" + || parsed.type === "response.incomplete"; + } catch { + return false; + } +} + +/** + * Byte-bounded SSE block framer for client-facing protocol paths. + * + * The delimiter scanner works on raw bytes, so fragmented UTF-8 cannot change + * accounting and a hostile upstream cannot grow an unterminated JS string + * without limit. Candidate bytes live in one geometrically grown buffer rather + * than one allocation per upstream chunk, bounding both bytes and object count. + * Complete blocks are returned without their delimiter; the exact delimiter + * bytes are returned separately so callers can relay bytes unchanged. + */ +export class BoundedSseFrameBuffer { + private readonly maxFrameBytes: number; + private readonly maxFramesPerFeed: number; + private delimiterTail: Uint8Array = new Uint8Array(0); + private candidate: Uint8Array = new Uint8Array(0); + private candidateBytes = 0; + private disposed = false; + + constructor(maxFrameBytes = MAX_CLIENT_SSE_FRAME_BYTES) { + if (!Number.isSafeInteger(maxFrameBytes) || maxFrameBytes <= 0) { + throw new RangeError("maxFrameBytes must be a positive safe integer"); + } + this.maxFrameBytes = maxFrameBytes; + // Delimiter-only input otherwise creates an object-amplification path that + // is independent of candidate bytes. Keep frame count proportional to the + // configured byte budget while leaving ample room for real Responses events. + this.maxFramesPerFeed = Math.max(1, Math.ceil(maxFrameBytes / 1024)); + } + + private clear(): void { + this.delimiterTail = new Uint8Array(0); + this.candidate = new Uint8Array(0); + this.candidateBytes = 0; + } + + private ensureCapacity(requiredBytes: number): void { + if (this.candidate.byteLength >= requiredBytes) return; + if (requiredBytes > this.maxFrameBytes) { + throw new SseFrameTooLargeError(this.maxFrameBytes); + } + let capacity = this.candidate.byteLength === 0 + ? Math.min(this.maxFrameBytes, Math.max(requiredBytes, 4096)) + : this.candidate.byteLength; + while (capacity < requiredBytes) { + capacity = Math.min(this.maxFrameBytes, Math.max(requiredBytes, capacity * 2)); + } + const grown = new Uint8Array(capacity); + if (this.candidateBytes > 0) { + grown.set(this.candidate.subarray(0, this.candidateBytes)); + } + this.candidate = grown; + } + + private retain(slice: Uint8Array): void { + if (slice.byteLength === 0) return; + const nextBytes = this.candidateBytes + slice.byteLength; + if (nextBytes > this.maxFrameBytes) { + this.clear(); + this.disposed = true; + throw new SseFrameTooLargeError(this.maxFrameBytes); + } + this.ensureCapacity(nextBytes); + this.candidate.set(slice, this.candidateBytes); + this.candidateBytes = nextBytes; + } + + private takeCandidate(): Uint8Array { + if (this.candidateBytes === 0) return new Uint8Array(0); + const block = this.candidate.slice(0, this.candidateBytes); + // Release the working allocation after each complete frame. This avoids + // retaining a rare multi-MiB frame allocation for the rest of a long-lived + // stream; normal small-frame allocation remains bounded by feed's frame cap. + this.candidate = new Uint8Array(0); + this.candidateBytes = 0; + return block; + } + + feed(chunk: Uint8Array): BoundedSseFrame[] { + if (this.disposed) return []; + if (chunk.byteLength === 0) return []; + + const frames: BoundedSseFrame[] = []; + const previousTail = this.delimiterTail; + this.delimiterTail = new Uint8Array(0); + const tailLength = previousTail.byteLength; + const totalLength = tailLength + chunk.byteLength; + const byteAt = (index: number): number => index < tailLength + ? previousTail[index]! + : chunk[index - tailLength]!; + const retainRange = (start: number, end: number): void => { + if (end <= start) return; + if (start < tailLength) { + this.retain(previousTail.subarray(start, Math.min(end, tailLength))); + } + if (end > tailLength) { + this.retain(chunk.subarray(Math.max(0, start - tailLength), end - tailLength)); + } + }; + + try { + let index = 0; + let retainedThrough = 0; + while (index < totalLength) { + const delimiterLength = delimiterLengthAt(index, totalLength, byteAt); + if (delimiterLength === undefined) break; + if (delimiterLength > 0) { + if (frames.length >= this.maxFramesPerFeed) { + this.clear(); + this.disposed = true; + throw new SseFrameCountLimitError(this.maxFramesPerFeed); + } + retainRange(retainedThrough, index); + const block = this.takeCandidate(); + const delimiter = delimiterBytesAt(index, delimiterLength, byteAt); + frames.push({ block, delimiter }); + index += delimiterLength; + retainedThrough = index; + continue; + } + index += 1; + } + + retainRange(retainedThrough, index); + if (index < totalLength) { + this.delimiterTail = copyRange(index, totalLength, tailLength, previousTail, chunk); + } + return frames; + } catch (err) { + const framingError = err instanceof SseFrameTooLargeError + || err instanceof SseFrameCountLimitError; + if (framingError && frames.some(frame => isResponsesTerminalFrame(frame.block))) { + // A terminal frame is the Responses protocol boundary. Ignore malformed + // or oversized bytes that occur later in the same upstream chunk rather + // than retroactively replacing the committed terminal with a 502. + this.clear(); + this.disposed = true; + return frames; + } + throw err; + } + } + + /** Return the final unterminated block bytes and release all retained state. */ + finish(): Uint8Array { + if (this.disposed) return new Uint8Array(0); + try { + this.retain(this.delimiterTail); + this.delimiterTail = new Uint8Array(0); + return this.takeCandidate(); + } finally { + this.clear(); + this.disposed = true; + } + } + + dispose(): void { + if (this.disposed) return; + this.clear(); + this.disposed = true; + } +} + +export function joinSseFrameBytes(parts: readonly Uint8Array[]): Uint8Array { + let byteLength = 0; + for (const part of parts) byteLength += part.byteLength; + if (byteLength === 0) return new Uint8Array(0); + if (parts.length === 1 && parts[0]!.byteLength === byteLength) return parts[0]!; + const joined = new Uint8Array(byteLength); + let offset = 0; + for (const part of parts) { + joined.set(part, offset); + offset += part.byteLength; + } + return joined; +} \ No newline at end of file diff --git a/src/server/ws-bridge.ts b/src/server/ws-bridge.ts index 23631fd52..da052c1b8 100644 --- a/src/server/ws-bridge.ts +++ b/src/server/ws-bridge.ts @@ -6,6 +6,7 @@ import { headersForCodexAuthContext } from "../codex/auth-context"; import type { ResponsesTerminalStatus } from "../bridge"; import type { DataPlaneAdmission } from "./auth-cors"; import type { AdmissionLease, AdmissionReservation } from "../lib/admission"; +import { BoundedSseFrameBuffer } from "./sse-frame-buffer"; const OPEN = 1; type ResponsesTerminalReporter = (status: ResponsesTerminalStatus) => void; @@ -163,15 +164,6 @@ function parseSseBlock(block: string): string | null { return data.length > 0 ? data.join("\n") : null; } -function nextSseBlock(buffer: string): { block: string; rest: string } | null { - const match = buffer.match(/\r?\n\r?\n/); - if (!match || match.index === undefined) return null; - return { - block: buffer.slice(0, match.index), - rest: buffer.slice(match.index + match[0].length), - }; -} - function payloadType(payload: string): string | null { try { const json = JSON.parse(payload) as { type?: unknown }; @@ -231,7 +223,7 @@ export async function pumpResponsesSseToWebSocket( ws.data.cancel = cancel; const decoder = new TextDecoder(); - let buffer = ""; + const framer = new BoundedSseFrameBuffer(); let terminalSeen = false; const handlePayload = (payload: string): boolean => { @@ -266,17 +258,14 @@ export async function pumpResponsesSseToWebSocket( while (!terminalSeen) { const { done, value } = await reader.read(); if (done) break; - buffer += decoder.decode(value, { stream: true }); - let next: { block: string; rest: string } | null; - while ((next = nextSseBlock(buffer))) { - buffer = next.rest; - const payload = parseSseBlock(next.block); + for (const frame of framer.feed(value)) { + const payload = parseSseBlock(decoder.decode(frame.block)); if (payload && handlePayload(payload)) break; } } - buffer += decoder.decode(); - if (!terminalSeen && buffer.trim()) { - const payload = parseSseBlock(buffer); + const tail = framer.finish(); + if (!terminalSeen && tail.byteLength > 0) { + const payload = parseSseBlock(decoder.decode(tail)); if (payload) handlePayload(payload); } if (!terminalSeen && isCurrent() && !clientCancelled) { @@ -284,11 +273,27 @@ export async function pumpResponsesSseToWebSocket( sendProtocolError(ws, 502, "Upstream stream ended before response terminal event"); } } catch (err) { - if (!terminalSeen && isCurrent() && ws.readyState === OPEN) { - if (!(err instanceof WsSendDroppedError)) reportTerminal("incomplete"); - sendProtocolError(ws, 502, err instanceof Error ? err.message : String(err)); + framer.dispose(); + if (err instanceof WsSendDroppedError) throw err; + if (!terminalSeen + && isCurrent() + && ws.readyState === OPEN + && !(err instanceof WsSendDroppedError)) { + reportTerminal("incomplete"); + try { + sendProtocolError(ws, 502, err instanceof Error ? err.message : String(err)); + } catch (sendErr) { + // If delivery is already dropped, there is no useful error frame left + // to send. Swallow only that expected transport signal; other failures + // still surface to the caller after the upstream reader is released. + if (!(sendErr instanceof WsSendDroppedError)) throw sendErr; + } } } finally { + framer.dispose(); + // Framing errors can occur while the upstream body is still live. Always + // release the reader, even when terminal/send paths already cancelled it. + void reader.cancel().catch(() => {}); if (ws.data.cancel === cancel) ws.data.cancel = undefined; } } @@ -462,4 +467,4 @@ export async function readBoundedPrefix( export function looksLikeSse(prefix: Uint8Array): boolean { const text = new TextDecoder().decode(prefix); return /^\s*(event:|data:)/.test(text); -} +} \ No newline at end of file diff --git a/tests/sse-client-frame-bounds.test.ts b/tests/sse-client-frame-bounds.test.ts new file mode 100644 index 000000000..eaf1979c5 --- /dev/null +++ b/tests/sse-client-frame-bounds.test.ts @@ -0,0 +1,269 @@ +import { describe, expect, test } from "bun:test"; +import type { ServerWebSocket } from "bun"; +import { + BoundedSseFrameBuffer, + MAX_CLIENT_SSE_FRAME_BYTES, + SseFrameCountLimitError, + SseFrameTooLargeError, +} from "../src/server/sse-frame-buffer"; +import { relaySseWithFailedTail } from "../src/server/relay"; +import { pumpResponsesSseToWebSocket, type WsData } from "../src/server/ws-bridge"; + +const enc = new TextEncoder(); +const dec = new TextDecoder(); + +function streamFromChunks(chunks: Uint8Array[]): ReadableStream { + let index = 0; + return new ReadableStream({ + pull(controller) { + if (index < chunks.length) controller.enqueue(chunks[index++]!); + else controller.close(); + }, + }); +} + +function concatBytes(...chunks: Uint8Array[]): Uint8Array { + const total = chunks.reduce((sum, chunk) => sum + chunk.byteLength, 0); + const joined = new Uint8Array(total); + let offset = 0; + for (const chunk of chunks) { + joined.set(chunk, offset); + offset += chunk.byteLength; + } + return joined; +} + +describe("client-facing SSE frame bounds", () => { + test("the byte framer accepts the exact cap and preserves a split delimiter", () => { + const framer = new BoundedSseFrameBuffer(8); + + expect(framer.feed(enc.encode("1234"))).toEqual([]); + expect(framer.feed(enc.encode("5678\n"))).toEqual([]); + const frames = framer.feed(enc.encode("\n")); + + expect(frames).toHaveLength(1); + expect(dec.decode(frames[0]!.block)).toBe("12345678"); + expect(dec.decode(frames[0]!.delimiter)).toBe("\n\n"); + expect(framer.finish().byteLength).toBe(0); + }); + + test("the byte framer rejects cap + 1 without retaining an oversized tail", () => { + const framer = new BoundedSseFrameBuffer(8); + + expect(() => framer.feed(enc.encode("123456789"))).toThrow(SseFrameTooLargeError); + expect(framer.finish().byteLength).toBe(0); + }); + + test("the byte framer preserves a committed Responses terminal before trailing overflow", () => { + const terminal = enc.encode('data: {"type":"response.completed"}\n\n'); + const oversizedTail = new Uint8Array(65); + oversizedTail.fill(120); + const framer = new BoundedSseFrameBuffer(64); + + const frames = framer.feed(concatBytes(terminal, oversizedTail)); + + expect(frames).toHaveLength(1); + expect(dec.decode(frames[0]!.block)).toBe('data: {"type":"response.completed"}'); + expect(framer.finish().byteLength).toBe(0); + }); + + test("delimiter-only input cannot amplify one chunk into unbounded frame objects", () => { + const framer = new BoundedSseFrameBuffer(4096); + + expect(() => framer.feed(enc.encode("\n\n".repeat(5)))).toThrow(SseFrameCountLimitError); + expect(framer.finish().byteLength).toBe(0); + }); + + test("fragmented multibyte UTF-8 is decoded only after the complete frame arrives", () => { + const text = 'data: {"type":"response.created","label":"€"}'; + const bytes = enc.encode(text); + const euro = enc.encode("€"); + const euroStart = bytes.findIndex((value, index) => ( + value === euro[0] + && bytes[index + 1] === euro[1] + && bytes[index + 2] === euro[2] + )); + expect(euroStart).toBeGreaterThan(0); + + const framer = new BoundedSseFrameBuffer(1024); + expect(framer.feed(bytes.slice(0, euroStart + 1))).toEqual([]); + const remainder = bytes.slice(euroStart + 1); + const delimiter = enc.encode("\n\n"); + const secondChunk = new Uint8Array(remainder.byteLength + delimiter.byteLength); + secondChunk.set(remainder, 0); + secondChunk.set(delimiter, remainder.byteLength); + const frames = framer.feed(secondChunk); + + expect(frames).toHaveLength(1); + expect(dec.decode(frames[0]!.block)).toBe(text); + }); + + test("HTTP native relay fails closed instead of retaining an oversized unterminated frame", async () => { + const upstream = new AbortController(); + const oversized = new Uint8Array(MAX_CLIENT_SSE_FRAME_BYTES + 1); + oversized.fill(120); + const relayed = relaySseWithFailedTail(streamFromChunks([oversized]), upstream); + + const text = await new Response(relayed).text(); + + expect(text.length).toBeLessThan(2048); + expect(text).toContain("response.failed"); + expect(text).toContain(`upstream SSE frame exceeded ${MAX_CLIENT_SSE_FRAME_BYTES} bytes`); + expect(text).toContain("data: [DONE]"); + expect(upstream.signal.aborted).toBe(true); + }); + + test("HTTP relay honours a completed frame before oversized trailing bytes in the same chunk", async () => { + const terminal = enc.encode( + 'event: response.completed\ndata: {"type":"response.completed","response":{"id":"r1"}}\n\n', + ); + const oversizedTail = new Uint8Array(MAX_CLIENT_SSE_FRAME_BYTES + 1); + oversizedTail.fill(120); + const relayed = relaySseWithFailedTail( + streamFromChunks([concatBytes(terminal, oversizedTail)]), + new AbortController(), + ); + + const text = await new Response(relayed).text(); + + expect(text).toContain('"type":"response.completed"'); + expect(text).toContain("data: [DONE]"); + expect(text).not.toContain('"type":"response.failed"'); + expect(text).not.toContain("upstream SSE frame exceeded"); + }); + + test("HTTP failed-tail cleanup preserves the original failure when finish also overflows", async () => { + const upstream = new AbortController(); + const nearCapWithAmbiguousTail = new Uint8Array(MAX_CLIENT_SSE_FRAME_BYTES + 1); + nearCapWithAmbiguousTail.fill(120, 0, MAX_CLIENT_SSE_FRAME_BYTES); + nearCapWithAmbiguousTail[MAX_CLIENT_SSE_FRAME_BYTES] = 10; + let reads = 0; + const source = new ReadableStream({ + pull(controller) { + reads += 1; + if (reads === 1) { + controller.enqueue(nearCapWithAmbiguousTail); + return; + } + controller.error(new Error("socket reset after partial frame")); + }, + }); + const relayed = relaySseWithFailedTail(source, upstream); + + const text = await new Response(relayed).text(); + + expect(text.length).toBeLessThan(2048); + expect(text).toContain("response.failed"); + expect(text).toContain("socket reset after partial frame"); + expect(text).toContain("data: [DONE]"); + expect(upstream.signal.aborted).toBe(true); + }); + + test("WebSocket pump emits one bounded protocol error and cancels upstream on overflow", async () => { + const sent: string[] = []; + const terminals: string[] = []; + let sourceCancelled = false; + const ws = { + readyState: 1, + data: {} as WsData, + send(message: string) { + sent.push(message); + return 1; + }, + } as unknown as ServerWebSocket; + const oversized = new Uint8Array(MAX_CLIENT_SSE_FRAME_BYTES + 1); + oversized.fill(120); + const source = new ReadableStream({ + start(controller) { + controller.enqueue(oversized); + }, + cancel() { + sourceCancelled = true; + }, + }); + + await pumpResponsesSseToWebSocket(ws, source, { + onTerminal: status => terminals.push(status), + }); + + expect(terminals).toEqual(["incomplete"]); + expect(sent).toHaveLength(1); + const error = JSON.parse(sent[0]!) as { + type?: string; + status?: number; + error?: { code?: string; message?: string }; + }; + expect(error.type).toBe("error"); + expect(error.status).toBe(502); + expect(error.error?.code).toBe("websocket_protocol_error"); + expect(error.error?.message).toBe( + `upstream SSE frame exceeded ${MAX_CLIENT_SSE_FRAME_BYTES} bytes`, + ); + expect(sourceCancelled).toBe(true); + expect(ws.data.cancel).toBeUndefined(); + }); + + test("WebSocket pump honours a completed frame before oversized trailing bytes in the same chunk", async () => { + const sent: string[] = []; + const terminals: string[] = []; + let sourceCancelled = false; + const ws = { + readyState: 1, + data: {} as WsData, + send(message: string) { + sent.push(message); + return 1; + }, + } as unknown as ServerWebSocket; + const terminal = enc.encode( + 'event: response.completed\ndata: {"type":"response.completed","response":{"id":"r1"}}\n\n', + ); + const oversizedTail = new Uint8Array(MAX_CLIENT_SSE_FRAME_BYTES + 1); + oversizedTail.fill(120); + const source = new ReadableStream({ + start(controller) { + controller.enqueue(concatBytes(terminal, oversizedTail)); + }, + cancel() { + sourceCancelled = true; + }, + }); + + await pumpResponsesSseToWebSocket(ws, source, { + onTerminal: status => terminals.push(status), + }); + + expect(terminals).toEqual(["completed"]); + expect(sent).toHaveLength(1); + expect(JSON.parse(sent[0]!).type).toBe("response.completed"); + expect(sourceCancelled).toBe(true); + expect(ws.data.cancel).toBeUndefined(); + }); + + test("WebSocket send drops do not trigger a second failing protocol-error send", async () => { + let sourceCancelled = false; + let sendCalls = 0; + const ws = { + readyState: 1, + data: {} as WsData, + send() { + sendCalls += 1; + return 0; + }, + } as unknown as ServerWebSocket; + const source = new ReadableStream({ + start(controller) { + controller.enqueue(enc.encode('data: {"type":"response.created"}\n\n')); + }, + cancel() { + sourceCancelled = true; + }, + }); + + await expect(pumpResponsesSseToWebSocket(ws, source)).rejects.toThrow("websocket send dropped"); + + expect(sendCalls).toBe(1); + expect(sourceCancelled).toBe(true); + expect(ws.data.cancel).toBeUndefined(); + }); +}); \ No newline at end of file