From 9b0fd01ac06b8f15302a334b032e2d60a3f3942d Mon Sep 17 00:00:00 2001 From: Epinephrine Date: Fri, 2 Oct 2026 07:10:40 -0700 Subject: [PATCH 1/2] fix(sse): align event framing before tool validation Use consistent CR, LF, and CRLF event and field boundaries across payload rewrites, terminal inspection, Chat projection, and WebSocket delivery. Preserve split-CRLF wire ownership, event-field repairs, live terminals, and bounded accounting. Add offline full-handler and framing/repair regressions. --- .../content/docs/reference/proxy-formats.md | 4 + scripts/test-layout/layout.json | 2 + src/server/github-copilot-responses-repair.ts | 9 +- src/server/grok-responses-control-frame.ts | 4 +- src/server/grok-responses-snapshot-repair.ts | 9 +- src/server/relay.ts | 58 ++----- src/server/responses-custom-tool-repair.ts | 4 +- src/server/sse-frame-buffer.ts | 74 ++++---- src/server/sse-payload-rewrite.ts | 119 ++++++++----- src/server/ws-bridge.ts | 12 +- structure/data-planes/inbound-compat.md | 4 +- structure/transports/byte-accounting.md | 22 ++- structure/transports/responses-wire-shapes.md | 8 +- tests/fixtures/test-layout-expected.json | 2 + .../responses/chat-sse-framing-guard.test.ts | 124 +++++++++++++ .../sse-rewrite-line-endings.test.ts | 163 ++++++++++++++++++ 16 files changed, 452 insertions(+), 166 deletions(-) create mode 100644 tests/responses/chat-sse-framing-guard.test.ts create mode 100644 tests/responses/sse-rewrite-line-endings.test.ts diff --git a/docs-site/src/content/docs/reference/proxy-formats.md b/docs-site/src/content/docs/reference/proxy-formats.md index bad29c7e258..818d6a8fa47 100644 --- a/docs-site/src/content/docs/reference/proxy-formats.md +++ b/docs-site/src/content/docs/reference/proxy-formats.md @@ -248,6 +248,10 @@ buffer budget, the relay reports the failure without waiting for upstream inspec to finish. It cancels the upstream work and emits `response.failed` followed by `data: [DONE]`; a budget overflow uses the `translation_buffer_limit` error code. +Responses SSE accepts CR, LF and CRLF line endings, including mixed endings and CRLF split across +network chunks. Tool validation, Chat translation, terminal inspection and WebSocket projection +use the same event boundaries. A CR-only completion does not wait for the upstream connection to close. + 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 diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index fde377d7131..7b6080dffe9 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1686,6 +1686,8 @@ "sse-inspector-bounds.test.ts": "responses", "sse-null-data-frame.test.ts": "responses", "sse-payload-rewrite.test.ts": "responses", + "sse-rewrite-line-endings.test.ts": "responses", + "chat-sse-framing-guard.test.ts": "responses", "sse-unspaced-data-fields.test.ts": "responses", "stale-state-purge.test.ts": "service", "stall-subprocess-exit.test.ts": "lib", diff --git a/src/server/github-copilot-responses-repair.ts b/src/server/github-copilot-responses-repair.ts index 449c2080688..b77d515ea0c 100644 --- a/src/server/github-copilot-responses-repair.ts +++ b/src/server/github-copilot-responses-repair.ts @@ -14,6 +14,8 @@ import { } from "../lib/translator-budget"; import { replaceSseDataPayload, + splitSseBlock, + sseLineEnding as blockNewline, sseDataPayload, type SseBlockRewrite, } from "./sse-payload-rewrite"; @@ -46,14 +48,9 @@ function byteLength(value: unknown): number { return Buffer.byteLength(JSON.stringify(value), "utf8"); } -function blockNewline(block: string): "\r\n" | "\n" { - return block.includes("\r\n") ? "\r\n" : "\n"; -} - /** Keep a named SSE block's `event:` field aligned with its JSON `type`. */ function withEventName(block: string, eventName: string): string { - const newline = blockNewline(block); - const lines = block.split(/\r?\n/); + const { newline, lines } = splitSseBlock(block); let replaced = false; const rewritten = lines.flatMap((line) => { if (!line.startsWith("event:")) return [line]; diff --git a/src/server/grok-responses-control-frame.ts b/src/server/grok-responses-control-frame.ts index 4d10340bfc9..938c49be2c0 100644 --- a/src/server/grok-responses-control-frame.ts +++ b/src/server/grok-responses-control-frame.ts @@ -1,4 +1,4 @@ -import { replaceSseDataPayload, sseDataPayload, type SseBlockRewrite } from "./sse-payload-rewrite"; +import { replaceSseDataPayload, splitSseBlock, sseDataPayload, type SseBlockRewrite } from "./sse-payload-rewrite"; const GROK_CONTROL_FRAME_TYPES: Record = { "codex.rate_limits": true, @@ -144,7 +144,7 @@ export function createGrokResponsesControlFrameBlockRewrite(): SseBlockRewrite { let eventName = ""; // SSE overwrites the event type on every event field, including empty resets. // Like sseDataPayload, remove only one optional ASCII space after the colon. - for (const line of block.split(/\r?\n/)) { + for (const line of splitSseBlock(block).lines) { if (line === "event") eventName = ""; else if (line.startsWith("event:")) { const value = line.slice("event:".length); diff --git a/src/server/grok-responses-snapshot-repair.ts b/src/server/grok-responses-snapshot-repair.ts index 57d95ed4273..30647a34cbf 100644 --- a/src/server/grok-responses-snapshot-repair.ts +++ b/src/server/grok-responses-snapshot-repair.ts @@ -1,7 +1,7 @@ /** Strict terminal reconstruction selected by the Grok compatibility marker. */ import type { TranslatorBudget } from "../lib/translator-budget"; import { MAX_COMPLETED_OUTPUT_ITEMS, MAX_COMPLETED_OUTPUT_ITEM_SOURCE_BYTES } from "./relay"; -import { replaceSseDataPayload, sseDataPayload, type SseBlockRewrite } from "./sse-payload-rewrite"; +import { replaceSseDataPayload, splitSseBlock, sseDataPayload, type SseBlockRewrite } from "./sse-payload-rewrite"; import { isPlainObject, jsonBlock, type RetainedOutputItem } from "./responses-snapshot-codec"; import { requestToolScope, @@ -215,14 +215,13 @@ function refusedTerminalBlock( }, }); const rewritten = replaceSseDataPayload(block, payload); - const newline = block.includes("\r\n") ? "\r\n" : "\n"; + const { newline, lines } = splitSseBlock(rewritten); let eventRewritten = false; - const lines = rewritten.split(/\r?\n/).map(line => { + return lines.map(line => { if (eventRewritten || !line.startsWith("event:")) return line; eventRewritten = true; return `event: ${GROK_REFUSED_TERMINAL_EVENT_TYPE}`; - }); - return lines.join(newline); + }).join(newline); } /** diff --git a/src/server/relay.ts b/src/server/relay.ts index 3b12a1ea2f5..0c6d10218f3 100644 --- a/src/server/relay.ts +++ b/src/server/relay.ts @@ -30,8 +30,9 @@ import { joinSseFrameBytes, MAX_CLIENT_SSE_FRAME_BYTES, SseFrameCountLimitError, + sseDelimiterLengthAt, } from "./sse-frame-buffer"; -import { replaceSseDataPayload, sseDataPayload } from "./sse-payload-rewrite"; +import { replaceSseDataPayload, splitSseBlock, sseDataPayload } from "./sse-payload-rewrite"; import { createBoundedResponseLogBody } from "./response-log-body"; import { clientWireLogOf } from "./inference/client-wire"; import { recordClientWireRequestLog } from "./inference/client-wire-log"; @@ -355,10 +356,11 @@ export function createSseTerminalOutputBoundary( frames: ReturnType, ): Uint8Array => { if (disposed || terminal || frames.length === 0) return EMPTY_BYTES; - if (maxFrames !== undefined && frames.length > maxFrames - framesSeen) { + const completedFrames = frames.reduce((count, frame) => count + (frame.continuation ? 0 : 1), 0); + if (maxFrames !== undefined && completedFrames > maxFrames - framesSeen) { throw new SseFrameCountLimitError(maxFrames); } - framesSeen += frames.length; + framesSeen += completedFrames; const output: Uint8Array[] = []; const appendOutput = (...parts: Uint8Array[]): void => { for (const part of parts) { @@ -596,17 +598,7 @@ export function relaySseWithFailedTail( }); } -export function nextSseBlock(buffer: string): { block: string; delimiter: 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), - delimiter: match[0], - rest: buffer.slice(match.index + match[0].length), - }; -} - -export { sseDataPayload } from "./sse-payload-rewrite"; +export { nextSseBlock, sseDataPayload } from "./sse-payload-rewrite"; type JsonRecord = Record; @@ -698,9 +690,8 @@ function stripCodexSafetyBufferingField(block: string, parsed: unknown): string } function rewritePolicyTerminalBlock(block: string, payload: string): string { - const newline = block.includes("\r\n") ? "\r\n" : "\n"; const rewritten = replaceSseDataPayload(block, payload); - const lines = rewritten.split(/\r?\n/); + const { newline, lines } = splitSseBlock(rewritten); let eventRewritten = false; const withEvent = lines.map(line => { if (!eventRewritten && line.startsWith("event:")) { @@ -1156,31 +1147,6 @@ function outputBearingSseEvent(event: { type?: unknown; output_index?: unknown } .test(event.type); } -function delimiterLengthAt( - index: number, - length: number, - byteAt: (index: number) => number, -): number | 0 | 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; -} - /** * Per-chunk SSE inspection state machine shared by consumeForInspection, * consumeForResponseLogMetadata, and the eager bounded relay (relay-eager.ts). @@ -1202,6 +1168,7 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector let sawTerminal = false; let disposed = false; let delimiterTail: Uint8Array = EMPTY_BYTES; + let pendingLineFeed = false; let candidate: Uint8Array = EMPTY_BYTES; let candidateBytes = 0; let discardingOversizedFrame = false; @@ -1228,6 +1195,7 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector const clearFrameState = (): void => { delimiterTail = EMPTY_BYTES; + pendingLineFeed = false; candidate = EMPTY_BYTES; candidateBytes = 0; discardingOversizedFrame = false; @@ -1504,15 +1472,17 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector retainCandidateSlice(chunk.subarray(Math.max(0, start - tailLength), end - tailLength)); } }; - let index = 0; - let retainedThrough = 0; + let index = pendingLineFeed && byteAt(0) === 10 ? 1 : 0; + if (totalLength > 0) pendingLineFeed = false; + let retainedThrough = index; while (index < totalLength) { - const delimiterLength = delimiterLengthAt(index, totalLength, byteAt); + const delimiterLength = sseDelimiterLengthAt(index, totalLength, byteAt); if (delimiterLength === undefined) break; if (delimiterLength > 0) { retainRange(retainedThrough, index); completeCandidate(); index += delimiterLength; + pendingLineFeed = index === totalLength && byteAt(index - 1) === 13; retainedThrough = index; continue; } diff --git a/src/server/responses-custom-tool-repair.ts b/src/server/responses-custom-tool-repair.ts index b0ba7f637ab..137a28f9b59 100644 --- a/src/server/responses-custom-tool-repair.ts +++ b/src/server/responses-custom-tool-repair.ts @@ -13,6 +13,7 @@ import { } from "../responses/custom-tool-compat"; import { replaceSseDataPayload, + splitSseBlock, sseDataPayload, type SseBlockRewrite, } from "./sse-payload-rewrite"; @@ -22,8 +23,7 @@ function isPlainObject(value: unknown): value is Record { } function replaceSseEventName(block: string, type: string): string { - const newline = block.includes("\r\n") ? "\r\n" : "\n"; - const lines = block.split(/\r?\n/); + const { newline, lines } = splitSseBlock(block); let replaced = false; const next = lines.map(line => { if (!replaced && line.startsWith("event:")) { diff --git a/src/server/sse-frame-buffer.ts b/src/server/sse-frame-buffer.ts index e4302d60e72..8309d376c50 100644 --- a/src/server/sse-frame-buffer.ts +++ b/src/server/sse-frame-buffer.ts @@ -1,13 +1,14 @@ import { isCyberPolicyCode, isCyberPolicyMessage } from "../lib/errors"; +import { sseDataPayload } from "./sse-payload-rewrite"; export const MAX_CLIENT_SSE_FRAME_BYTES = 4 * 1024 * 1024; export const EMPTY_BYTES = new Uint8Array(0); -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); +// Intern every legal pair; a split CRLF may leave a framing-only LF continuation. +const DELIMITERS = new Map([ + "\n", "\n\n", "\n\r", "\n\r\n", "\r\r", "\r\r\n", "\r\n\n", "\r\n\r", "\r\n\r\n", +].map(text => [text, Uint8Array.from(text, char => char.charCodeAt(0))])); export class SseFrameTooLargeError extends Error { readonly maxBytes: number; @@ -32,6 +33,8 @@ export class SseFrameCountLimitError extends Error { export type BoundedSseFrame = { block: Uint8Array; delimiter: Uint8Array; + /** A late LF extending a previously returned CR; no new event. */ + continuation?: true; }; /** @@ -40,29 +43,22 @@ export type BoundedSseFrame = { * 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( +export function sseDelimiterLengthAt( 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; + if (first !== 10 && first !== 13) return 0; + let next = index + 1; + if (next >= length) return undefined; + if (first === 13 && byteAt(next) === 10) next += 1; + if (next >= length) return undefined; + const second = byteAt(next); + if (second !== 10 && second !== 13) return 0; + next += 1; + if (second === 13 && next < length && byteAt(next) === 10) next += 1; + return next - index; } function delimiterBytesAt( @@ -70,9 +66,9 @@ function delimiterBytesAt( 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; + let text = ""; + for (let offset = 0; offset < delimiterLength; offset++) text += String.fromCharCode(byteAt(index + offset)); + return DELIMITERS.get(text)!; } function copyRange( @@ -100,14 +96,8 @@ function copyRange( * 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"); + const payload = sseDataPayload(new TextDecoder().decode(block)); + if (payload === null) return false; if (payload === "[DONE]") return false; try { const parsed = JSON.parse(payload) as { @@ -161,6 +151,7 @@ export class BoundedSseFrameBuffer { private candidate: Uint8Array = EMPTY_BYTES; private candidateBytes = 0; private disposed = false; + private pendingLineFeed = false; constructor(maxFrameBytes = MAX_CLIENT_SSE_FRAME_BYTES) { if (!Number.isSafeInteger(maxFrameBytes) || maxFrameBytes <= 0) { @@ -177,6 +168,7 @@ export class BoundedSseFrameBuffer { this.delimiterTail = EMPTY_BYTES; this.candidate = EMPTY_BYTES; this.candidateBytes = 0; + this.pendingLineFeed = false; } private ensureCapacity(requiredBytes: number): void { @@ -226,6 +218,7 @@ export class BoundedSseFrameBuffer { if (chunk.byteLength === 0) return []; const frames: BoundedSseFrame[] = []; + let completedFrames = 0; const previousTail = this.delimiterTail; this.delimiterTail = EMPTY_BYTES; const tailLength = previousTail.byteLength; @@ -245,12 +238,19 @@ export class BoundedSseFrameBuffer { try { let index = 0; - let retainedThrough = 0; + if (this.pendingLineFeed && byteAt(0) === 10) { + // The prior event was already dispatched on CR. Preserve its later LF without + // charging it to the next event or letting it pair with another line ending. + frames.push({ block: EMPTY_BYTES, delimiter: DELIMITERS.get("\n")!, continuation: true }); + index = 1; + } + this.pendingLineFeed = false; + let retainedThrough = index; while (index < totalLength) { - const delimiterLength = delimiterLengthAt(index, totalLength, byteAt); + const delimiterLength = sseDelimiterLengthAt(index, totalLength, byteAt); if (delimiterLength === undefined) break; if (delimiterLength > 0) { - if (frames.length >= this.maxFramesPerFeed) { + if (completedFrames >= this.maxFramesPerFeed) { this.clear(); this.disposed = true; throw new SseFrameCountLimitError(this.maxFramesPerFeed); @@ -259,7 +259,9 @@ export class BoundedSseFrameBuffer { const block = this.takeCandidate(); const delimiter = delimiterBytesAt(index, delimiterLength, byteAt); frames.push({ block, delimiter }); + completedFrames += 1; index += delimiterLength; + this.pendingLineFeed = index === totalLength && byteAt(index - 1) === 13; retainedThrough = index; continue; } diff --git a/src/server/sse-payload-rewrite.ts b/src/server/sse-payload-rewrite.ts index 5f3a1ad0aa3..a00dcf9c811 100644 --- a/src/server/sse-payload-rewrite.ts +++ b/src/server/sse-payload-rewrite.ts @@ -65,7 +65,8 @@ export function composeSseBlockRewrites(...rewrites: SseBlockRewrite[]): SseBloc /** Split one complete SSE event block while retaining its original blank-line delimiter. */ export function nextSseBlock(buffer: string): { block: string; delimiter: string; rest: string } | null { - const match = buffer.match(/\r?\n\r?\n/); + // The negative lookahead prevents backtracking from splitting one CRLF into two endings. + const match = buffer.match(/(?:\r\n|\r(?!\n)|\n){2}/); if (!match || match.index === undefined) return null; return { block: buffer.slice(0, match.index), @@ -95,6 +96,7 @@ export function createSseBlockBuffer( let offset = 0; let scanOffset = 0; let bufferBytes = 0; + const delimiterPattern = /(?:\r\n|\r(?!\n)|\n){2}/g; const compact = (): void => { if (offset === 0) return; @@ -128,41 +130,30 @@ export function createSseBlockBuffer( } }, next() { - for (;;) { - const newline = buffer.indexOf("\n", scanOffset); - if (newline < 0) { - scanOffset = buffer.length; - return null; - } - let end = newline + 1; - if (buffer[end] === "\r") end += 1; - if (end === buffer.length) { - // Keep the candidate first newline until its possible blank-line delimiter arrives. - scanOffset = newline; - return null; - } - if (buffer[end] !== "\n") { - scanOffset = newline + 1; - continue; - } - end += 1; - const start = newline > offset && buffer[newline - 1] === "\r" ? newline - 1 : newline; - const block = buffer.slice(offset, start); - const delimiter = buffer.slice(start, end); - const nextBytes = bufferBytes - Buffer.byteLength(block, "utf8") - delimiter.length; - const reservation = budget.reserveTransient(nextBytes, scope); - reservation.commitRetained(); - budget.releaseRetained(bufferBytes, scope); - bufferBytes = nextBytes; - offset = end; - scanOffset = end; - if (offset === buffer.length) { - buffer = ""; - offset = 0; - scanOffset = 0; - } - return { block, delimiter }; + delimiterPattern.lastIndex = scanOffset; + const match = delimiterPattern.exec(buffer); + if (!match) { + // A delimiter is at most four code units. Revisit only its possible split prefix. + scanOffset = Math.max(offset, buffer.length - 3); + return null; + } + const start = match.index; + const end = start + match[0].length; + const block = buffer.slice(offset, start); + const delimiter = buffer.slice(start, end); + const nextBytes = bufferBytes - Buffer.byteLength(block, "utf8") - delimiter.length; + const reservation = budget.reserveTransient(nextBytes, scope); + reservation.commitRetained(); + budget.releaseRetained(bufferBytes, scope); + bufferBytes = nextBytes; + offset = end; + scanOffset = end; + if (offset === buffer.length) { + buffer = ""; + offset = 0; + scanOffset = 0; } + return { block, delimiter }; }, tail: () => buffer.slice(offset), compact, @@ -184,11 +175,11 @@ export function sseDataPayload(block: string): string | null { const len = block.length; while (lineStart < len) { - const nextNewline = block.indexOf("\n", lineStart); - let lineEnd = nextNewline === -1 ? len : nextNewline; - const nextStart = nextNewline === -1 ? len : nextNewline + 1; - if (lineEnd > lineStart && block.charCodeAt(lineEnd - 1) === 13) { - lineEnd -= 1; + let lineEnd = lineStart; + while (lineEnd < len && block.charCodeAt(lineEnd) !== 13 && block.charCodeAt(lineEnd) !== 10) lineEnd += 1; + let nextStart = lineEnd; + if (nextStart < len) { + nextStart += block.charCodeAt(lineEnd) === 13 && block.charCodeAt(lineEnd + 1) === 10 ? 2 : 1; } const lineLen = lineEnd - lineStart; @@ -218,10 +209,19 @@ export function sseDataPayload(block: string): string | null { return found ? result : null; } +/** Shared line rules for field-mutating repairs; identity paths keep the original block. */ +export function sseLineEnding(block: string): "\r\n" | "\r" | "\n" { + return block.match(/\r\n|\r|\n/)?.[0] as "\r\n" | "\r" | "\n" ?? "\n"; +} + +export function splitSseBlock(block: string): { newline: string; lines: string[] } { + return { newline: sseLineEnding(block), lines: block.split(/\r\n|\r|\n/) }; +} + /** Replace an SSE event's data field while preserving non-data fields and newline style. */ export function replaceSseDataPayload(block: string, payload: string): string { - const newline = block.includes("\r\n") ? "\r\n" : "\n"; - const lines = block.split(/\r?\n/); + if (sseDataPayload(block) === payload) return block; + const { newline, lines } = splitSseBlock(block); const rewritten: string[] = []; let replaced = false; for (const line of lines) { @@ -230,7 +230,7 @@ export function replaceSseDataPayload(block: string, payload: string): string { continue; } if (!replaced) { - rewritten.push(...payload.split(/\r?\n/).map(line => `data: ${line}`)); + rewritten.push(...payload.split(/\r\n|\r|\n/).map(line => `data: ${line}`)); replaced = true; } } @@ -278,6 +278,8 @@ export function relaySseWithBlockRewrite( // Relays have several independent teardown paths; disposal is exactly once. let disposed = false; let cancelled = false; + let pendingLineFeed = false; + let pendingLineFeedVisible = false; const disposeRewrite = (): void => { if (disposed) return; disposed = true; @@ -308,15 +310,38 @@ export function relaySseWithBlockRewrite( let emitted = 0; let next: { block: string; delimiter: string } | null; while (!cancelled && (next = buffer.next())) { - const outBlocks = rewrite(next.block); + // A final CR dispatches immediately. A later LF extends that same delimiter and + // must not become a field, or enter the next block's rewrite/drop/injection. + let { block, delimiter } = next; + if (pendingLineFeed && (block || delimiter).startsWith("\n")) { + if (pendingLineFeedVisible) { + enqueueText(controller, "\n"); + emitted += 1; + } + if (block) block = block.slice(1); + else delimiter = delimiter.slice(1); + } + pendingLineFeed = delimiter.endsWith("\r"); + const outBlocks = rewrite(block); + pendingLineFeedVisible = outBlocks.length > 0; if (cancelled) return emitted; - for (const outBlock of outBlocks) { - enqueueText(controller, outBlock + next.delimiter); + for (let index = 0; index < outBlocks.length; index++) { + // Synthetic earlier blocks need a settled delimiter; only the final one can + // inherit an upstream LF in a later chunk. Completing CR as CRLF is equivalent. + const settled = index < outBlocks.length - 1 && pendingLineFeed ? delimiter + "\n" : delimiter; + enqueueText(controller, outBlocks[index]! + settled); emitted += 1; } } buffer.compact(); - const tail = flushFinal ? buffer.tail() : ""; + let tail = flushFinal ? buffer.tail() : ""; + if (pendingLineFeed && tail.startsWith("\n")) { + if (pendingLineFeedVisible) { + enqueueText(controller, "\n"); + emitted += 1; + } + tail = tail.slice(1); + } if (tail.length > 0) { const tailBlocks = rewrite(tail); if (cancelled) return emitted; diff --git a/src/server/ws-bridge.ts b/src/server/ws-bridge.ts index 4a5c28c567d..da2d2c50883 100644 --- a/src/server/ws-bridge.ts +++ b/src/server/ws-bridge.ts @@ -8,6 +8,7 @@ import type { ResponsesTerminalStatus } from "../bridge"; import type { DataPlaneAdmission } from "./auth-cors"; import type { AdmissionLease, AdmissionReservation } from "../lib/admission"; import { BoundedSseFrameBuffer } from "./sse-frame-buffer"; +import { sseDataPayload as parseSseBlock } from "./sse-payload-rewrite"; import { safeResponseHeaders } from "./safe-response-headers"; import type { AudioSocketTarget } from "./audio-dictation"; @@ -183,17 +184,6 @@ export function buildWsErrorFrame( }; } -function parseSseBlock(block: string): string | null { - const data: string[] = []; - for (const line of block.split(/\r?\n/)) { - if (line.startsWith("data:")) { - const value = line.slice(5); - data.push(value.startsWith(" ") ? value.slice(1) : value); - } - } - return data.length > 0 ? data.join("\n") : null; -} - function payloadType(payload: string): string | null { try { const json = JSON.parse(payload) as { type?: unknown }; diff --git a/structure/data-planes/inbound-compat.md b/structure/data-planes/inbound-compat.md index 8e33465978b..3867d67e41f 100644 --- a/structure/data-planes/inbound-compat.md +++ b/structure/data-planes/inbound-compat.md @@ -146,7 +146,7 @@ which also decides which role that slot carries. Regression coverage is in `tests/responses/chat-inbound-developer-position.test.ts`, which compares the final upstream body on the native Chat route, a combo route and the Responses endpoint. -The direct SSE relay accepts CRLF and arbitrary transport chunk boundaries while retaining at most +The direct SSE relay accepts CR, LF and CRLF across arbitrary transport chunk boundaries while retaining at most one bounded event. EOF with an unterminated event and an event above the translator limit are typed upstream failures, never successful partial completions. Provider-controlled structured error messages are redacted before either JSON or SSE reaches the client. The native path uses the same @@ -165,7 +165,7 @@ allowance; comments, role-only frames, empty deltas, and usage alone do not. Dow pauses this wait budget. A stall emits a Chat error with `upstream_stall_timeout` and logs 502; the non-streaming endpoint returns HTTP 502 rather than a successful partial result. -`src/chat/outbound.ts` collects LF/CRLF, multiline data, and split UTF-8 through the shared SSE +`src/chat/outbound.ts` collects CR/LF/CRLF, multiline data, and split UTF-8 through the shared SSE block buffer and tracks appended output bytes incrementally. A caller cancellation before a native terminal returns 499 / `client_cancelled`; an already accepted terminal keeps its result. Reader, timer, turn, and translator ownership are released through the existing lifecycle. diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md index 160b25b1426..1e163aedc46 100644 --- a/structure/transports/byte-accounting.md +++ b/structure/transports/byte-accounting.md @@ -65,18 +65,26 @@ the request translator budget. Each owner releases its own reservations on compl cancellation or consumer closure; a buffer-limit failure terminates without another search. `src/server/sse-payload-rewrite.ts` shares an incremental block buffer with native Chat. It scans -only new input, counts consumed blocks rather than remaining suffixes, and preserves LF/CRLF, -partial-event, injection/drop, and EOF behavior. Output admission precedes its single UTF-8 encoding; +only new input plus a three-character delimiter prefix and counts consumed blocks rather than +remaining suffixes. CR, LF and CRLF share partial-event, injection/drop and EOF behavior. +Output admission precedes its single UTF-8 encoding; failed enqueue and cancellation release the reservation without re-entering a disposed rewrite. Old/new buffer overlap remains charged against the same translator cap. Complete SSE blocks extract `data` fields with one indexed pass over the block rather than a regular-expression split and intermediate line array. Colonless `data` fields, one optional ASCII -space after the colon, multiline joining, UTF-8 text, LF/CRLF input, and a trailing lone CR retain -their event-stream semantics. `src/server/relay.ts` re-exports this canonical extractor instead of -maintaining a second implementation. Empty byte results across the relay and -`src/server/sse-frame-buffer.ts` reuse one immutable zero-length view; non-empty frame ownership, -frame limits, cancellation, terminal detection, and wire bytes are unchanged. +space after the colon, multiline joining, UTF-8 text and mixed CR/LF/CRLF input retain their +event-stream semantics. Payload and event-field repairs use the shared line rules, including +policy failures, custom tools, Copilot and Grok events; identity rewrites preserve +original wire bytes. `src/server/relay.ts` re-exports the canonical extractor and block splitter. +Empty byte results across the relay and `src/server/sse-frame-buffer.ts` reuse one immutable +zero-length view; non-empty frame ownership, +frame limits, cancellation and wire ownership remain bounded. The byte framer and inspection share +one CR/LF/CRLF delimiter scanner; WebSocket projection uses the canonical data extractor. A terminal +CR dispatches without awaiting EOF. A later LF extends that delimiter: byte delivery returns a +framing-only continuation, excluded from event-count limits and the next event's byte cap; +inspection consumes it without creating an event. The text rewrite relay keeps that continuation +outside rewrite/drop callbacks and settles injected nonfinal CR delimiters as CRLF. `src/server/responses-custom-tool-repair.ts` continues to own retained routed argument bytes and their charge/release lifecycle while it asks the pure progressive decoder in diff --git a/structure/transports/responses-wire-shapes.md b/structure/transports/responses-wire-shapes.md index f2844b32528..7f1dfc00bb2 100644 --- a/structure/transports/responses-wire-shapes.md +++ b/structure/transports/responses-wire-shapes.md @@ -235,7 +235,6 @@ The restorable map is narrowed by `tool_choice` the same way `authorizedAliases` namespace layer: upstream still receives the whole aliased catalog, but a tool the caller disabled for the turn cannot be restored back into an executable client name. Arguments, user text, and schema property names are never rewritten. - > Decision record: [ADR-0042](../decisions/ADR-0042-responses-http-sse.md) > Decision record: [ADR-0043](../decisions/ADR-0043-responses-http-sse.md) @@ -248,7 +247,9 @@ Google wire compiler observes and reports compatibility narrowing; the Responses derive nor consume that report. `declaredToolNames` carries the request's tool catalog into both bridges, and it does two separate -jobs that are separately controlled. +jobs that are separately controlled. The passthrough guard reads the same CR/LF/CRLF event and +field boundaries as Responses-to-Chat projection, including mixed and split line endings; the +[shared SSE framing contract](byte-accounting.md#stream-buffer-accounting) owns that behavior. Normalization runs on every inbound wire. `normalizeDeclaredToolName` and `declaresCodeModeExec` in `src/types/tools.ts` read the same set to map a provider-invented `default.` namespace back to the @@ -256,8 +257,7 @@ declared bare tool and to rewrite code-mode helper names into the declared `exec input unchanged when the set is absent, so the set reaches the bridge on every wire and enforcement is expressed by a separate flag rather than by withholding it. -The passthrough guard resolves an emitted name through that same `normalizeDeclaredToolName`, so -whatever it admits it must also EMIT under the resolved name. The two halves disagreed once: +The passthrough guard resolves an emitted name through that same `normalizeDeclaredToolName`, so whatever it admits it must also EMIT under the resolved name. The two halves disagreed once: `normalizeDefaultNamespaceInItem` implemented only the bare-tool case (#4176), so a `default.`-prefixed code-mode helper was admitted as `exec` (#4412) and then relayed verbatim. The bounded helper vocabulary includes the goal lifecycle calls that Codex advertises inside its diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 46c16b4174a..b60cdc4294c 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1700,6 +1700,8 @@ "sse-inspector-bounds.test.ts": "responses", "sse-null-data-frame.test.ts": "responses", "sse-payload-rewrite.test.ts": "responses", + "sse-rewrite-line-endings.test.ts": "responses", + "chat-sse-framing-guard.test.ts": "responses", "sse-unspaced-data-fields.test.ts": "responses", "stale-state-purge.test.ts": "service", "stall-subprocess-exit.test.ts": "lib", diff --git a/tests/responses/chat-sse-framing-guard.test.ts b/tests/responses/chat-sse-framing-guard.test.ts new file mode 100644 index 00000000000..68731a4484f --- /dev/null +++ b/tests/responses/chat-sse-framing-guard.test.ts @@ -0,0 +1,124 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { handleChatCompletions } from "../../src/server/chat-completions"; +import type { OcxConfig, OcxProviderConfig } from "../../src/types"; +import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; + +let previousHome: string | undefined; +let home: string; +let releaseSpendHome: () => void; +beforeEach(() => { + previousHome = process.env.OPENCODEX_HOME; + home = mkdtempSync(join(tmpdir(), "ocx-chat-sse-framing-")); + process.env.OPENCODEX_HOME = home; + releaseSpendHome = acquireOwnedSpendHome(); +}); +afterEach(() => { + releaseSpendHome(); + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + removeTreeWithRetry(home); +}); + +const encoder = new TextEncoder(); +function event(name: string, newline: string, delimiter = newline + newline): string { + const item = { type: "function_call", id: "fc_fixture", call_id: "call_fixture", name, arguments: "{}", status: "completed" }; + return `: ignored${newline}event: response.output_item.done${newline}data: ${JSON.stringify({ type: "response.output_item.done", output_index: 0, item })}${delimiter}`; +} +function terminal(newline: string): string { + return `event: response.completed${newline}data: ${JSON.stringify({ type: "response.completed", response: { id: "resp_fixture", status: "completed", output: [], usage: { input_tokens: 1, output_tokens: 1 } } })}${newline}${newline}`; +} + +async function request(chunks: string[], stream: boolean, live = false) { + let requests = 0; + let cancelled = false; + const provider: OcxProviderConfig & { fetch: typeof fetch } = { + adapter: "openai-responses", baseUrl: "https://sse-framing.invalid/v1", authMode: "key", + apiKey: "fixture-key", models: ["model"], + // Own the executor: DNS-pinned egress does not necessarily use globalThis.fetch. + fetch: (async (url: string | URL | Request, init?: RequestInit) => { + expect(String(url)).toBe("https://sse-framing.invalid/v1/responses"); + requests += 1; + let index = 0; + return new Response(new ReadableStream({ + start(controller) { + init?.signal?.addEventListener("abort", () => { + cancelled = true; + try { controller.close(); } catch { /* already closed or cancelled */ } + }, { once: true }); + }, + pull(controller) { + if (index < chunks.length) controller.enqueue(encoder.encode(chunks[index++]!)); + else if (!live) controller.close(); + }, + cancel() { cancelled = true; }, + }), { headers: { "content-type": "text/event-stream" } }); + }) as typeof fetch, + }; + const config: OcxConfig = { port: 0, defaultProvider: "fixture", providers: { fixture: provider } }; + const response = await handleChatCompletions(new Request("http://localhost/v1/chat/completions", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "fixture/model", stream, messages: [{ role: "user", content: "fixture" }], + tools: [{ type: "function", function: { name: "declared_tool", parameters: { type: "object" } } }] }), + }), config, { model: "", provider: "" }); + expect(requests).toBe(1); + return { response, cancelled: () => cancelled }; +} + +async function readBounded(response: Response): Promise { + const reader = response.body!.getReader(); + const decoder = new TextDecoder(); + let timer: ReturnType | undefined; + try { + return await Promise.race([(async () => { + let text = ""; + for (;;) { + const { done, value } = await reader.read(); + if (done) return text; + text += decoder.decode(value, { stream: true }); + } + })(), new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error("SSE completion waited for EOF")), 2_000); + })]); + } finally { + clearTimeout(timer); + await reader.cancel().catch(() => {}); + } +} + +describe("Responses-to-Chat shared SSE security framing", () => { + test.each([[true, false], [false, false], [true, true], [false, true]])( + "rejects CR-hidden undeclared calls (stream=%s, fragmented=%s)", async (stream, fragmented) => { + const wire = event("unlisted_tool", "\r", "\r\r\n\n") + terminal("\n"); + const { response } = await request(fragmented ? [...wire] : [wire], stream); + const text = await readBounded(response); + expect(text).toContain("routed provider emitted undeclared client tool"); + expect(text).toContain('"error"'); + expect(text).not.toContain('"tool_calls"'); + }); + test.each(["\n", "\r\n", "\r"])("keeps declared calls with %j framing", async newline => { + const { response } = await request([...event("declared_tool", newline), ...terminal(newline)], true); + const text = await readBounded(response); + expect(text).toContain('"name":"declared_tool"'); + expect(text).toContain('"finish_reason":"tool_calls"'); + expect(text).not.toContain('"error"'); + }); + test("delivers a CR-only declared completion before upstream EOF", async () => { + const result = await request([event("declared_tool", "\r") + terminal("\r")], true, true); + const text = await readBounded(result.response); + expect(text).toContain('"name":"declared_tool"'); + expect(text).toContain("[DONE]"); + expect(result.cancelled()).toBe(true); + }); + test("rejects consecutive CR-only events before upstream EOF", async () => { + const result = await request([event("unlisted_tool", "\r") + terminal("\r")], true, true); + const text = await readBounded(result.response); + expect(text).toContain("routed provider emitted undeclared client tool"); + expect(text).toContain('"error"'); + expect(text).not.toContain('"tool_calls"'); + expect(result.cancelled()).toBe(true); + }); +}); diff --git a/tests/responses/sse-rewrite-line-endings.test.ts b/tests/responses/sse-rewrite-line-endings.test.ts new file mode 100644 index 00000000000..d10b5431c74 --- /dev/null +++ b/tests/responses/sse-rewrite-line-endings.test.ts @@ -0,0 +1,163 @@ +import { describe, expect, spyOn, test } from "bun:test"; +import { Buffer } from "node:buffer"; +import type { ServerWebSocket } from "bun"; +import { createSseInspector, createSseTerminalOutputBoundary } from "../../src/server/relay"; +import { createRoutedCustomToolRestoreBlockRewrite } from "../../src/server/responses-custom-tool-repair"; +import { createGithubCopilotResponsesBlockRewrite } from "../../src/server/github-copilot-responses-repair"; +import { createGrokResponsesSparseTerminalBlockRewrite, GROK_REFUSED_TERMINAL_EVENT_TYPE } from "../../src/server/grok-responses-snapshot-repair"; +import { createGrokResponsesControlFrameBlockRewrite } from "../../src/server/grok-responses-control-frame"; +import { pumpResponsesSseToWebSocket, type WsData } from "../../src/server/ws-bridge"; +import { decodeServerSentEvents } from "../../src/lib/sse-decoder"; +import { BoundedSseFrameBuffer, joinSseFrameBytes, SseFrameTooLargeError } from "../../src/server/sse-frame-buffer"; +import { createSseBlockBuffer, nextSseBlock, relaySseWithBlockRewrite, replaceSseDataPayload, sseDataPayload } from "../../src/server/sse-payload-rewrite"; +import { createTestTranslatorBudget } from "../helpers/translator-budget"; +import { readAll } from "../helpers/sse-stream"; + +const encoder = new TextEncoder(); +const decoder = new TextDecoder(); +function source(chunks: string[]) { + return new ReadableStream({ start(controller) { + for (const chunk of chunks) controller.enqueue(encoder.encode(chunk)); + controller.close(); + } }); +} +const delimiters = ["\n\n", "\n\r", "\n\r\n", "\r\r", "\r\r\n", "\r\n\n", "\r\n\r", "\r\n\r\n"]; + +describe("shared SSE line ending contract", () => { + test.each(delimiters)("frames %j without splitting a single CRLF", delimiter => { + expect(nextSseBlock(`event: x\r\ndata: one${delimiter}next`)).toEqual({ + block: "event: x\r\ndata: one", delimiter, rest: "next", + }); + expect(nextSseBlock("data: one\r\n")).toBeNull(); + }); + test.each(["\n", "\r", "\r\n"])("reads and rewrites fields with %j lines", newline => { + const block = [": ignored", "event: update", "data", "data: first", "data:\tsecond", "data", "database: ignored"].join(newline); + expect(sseDataPayload(block)).toBe("\n first\n\tsecond\n"); + expect(replaceSseDataPayload(block, "one\rtwo\r\nthree\nfour")).toBe( + [": ignored", "event: update", "data: one", "data: two", "data: three", "data: four", "database: ignored"].join(newline), + ); + expect(replaceSseDataPayload(block, sseDataPayload(block)!)).toBe(block); + }); + test("identity replacement keeps mixed field wire bytes", () => { + const block = ": comment\revent: update\r\ndata\ndata:one\rdata: two\r\nid: fixture"; + expect(sseDataPayload(block)).toBe("\none\ntwo"); + expect(replaceSseDataPayload(block, "\none\ntwo")).toBe(block); + }); + test.each(delimiters)("preserves %j identity bytes and decoder events at every split", async delimiter => { + const wire = `: ignored\revent: first\r\ndata: one${delimiter}data: two\rdata: three${delimiter}data: tail`; + const expected = [{ event: "first", data: "one" }, { data: "two\nthree" }, { data: "tail" }]; + for (let split = 0; split <= wire.length; split++) { + const chunks = [wire.slice(0, split), wire.slice(split)]; + const budget = createTestTranslatorBudget(); + const output = await readAll(relaySseWithBlockRewrite(source(chunks), block => [block], budget)); + expect(output).toBe(wire); + expect(budget.snapshot().currentBytes).toBe(0); + const decoded = []; + for await (const record of decodeServerSentEvents(source([output]), { translatorBudget: budget })) decoded.push(record); + expect(decoded).toEqual(expected); + expect(budget.snapshot().currentBytes).toBe(0); + } + }); + test("does not dispatch a split CRLF as a blank line", () => { + const budget = createTestTranslatorBudget(); + const buffer = createSseBlockBuffer(budget); + buffer.append("data: first\r"); + expect(buffer.next()).toBeNull(); + buffer.append("\ndata: second\r\n"); + expect(buffer.next()).toBeNull(); + buffer.append("\r"); + expect(sseDataPayload(buffer.next()!.block)).toBe("first\nsecond"); + buffer.append("\n"); + expect(buffer.next()).toBeNull(); + buffer.clear(); + expect(budget.snapshot().currentBytes).toBe(0); + }); + test("incremental delimiter search and byte counts stay linear for CR fragments", () => { + const budget = createTestTranslatorBudget(); + const buffer = createSseBlockBuffer(budget); + const fragments = ["data: ", ...Array.from({ length: 256 }, () => "x".repeat(128)), "\r", "\r"]; + const originalExec = RegExp.prototype.exec; + const originalByteLength = Buffer.byteLength; + let scanned = 0; + let counted = 0; + const exec = spyOn(RegExp.prototype, "exec").mockImplementation(function (this: RegExp, text) { + if (this.source === "(?:\\r\\n|\\r(?!\\n)|\\n){2}") scanned += text.length - this.lastIndex; + return originalExec.call(this, text); + }); + const byteLength = spyOn(Buffer, "byteLength").mockImplementation((value, encoding) => { + if (typeof value === "string") counted += value.length; + return originalByteLength(value, encoding); + }); + try { + for (const fragment of fragments) { buffer.append(fragment); buffer.next(); } + expect(scanned).toBeLessThan(fragments.join("").length * 2); + expect(counted).toBeLessThan(fragments.join("").length * 3); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { exec.mockRestore(); byteLength.mockRestore(); buffer.clear(); } + }); + test.each(["\r\r", "\n\r", "\r\n\r"])("byte framing preserves %j and split CRLF at the exact cap", delimiter => { + const framer = new BoundedSseFrameBuffer(8); + const chunks = ["12345678" + delimiter, "\nabcdefgh\r\r", "\n"]; + const frames = chunks.flatMap(chunk => framer.feed(encoder.encode(chunk))); + const tail = framer.finish(); + expect(decoder.decode(joinSseFrameBytes([...frames.flatMap(frame => [frame.block, frame.delimiter]), tail]))).toBe(chunks.join("")); + expect(frames.filter(frame => frame.block.length).map(frame => decoder.decode(frame.block))).toEqual(["12345678", "abcdefgh"]); + const oversized = new BoundedSseFrameBuffer(8); + expect(() => oversized.feed(encoder.encode("123456789" + delimiter))).toThrow(SseFrameTooLargeError); + }); + test.each(["\r", "\n", "\r\n"])("inspection and WebSocket delivery agree on %j fields", async newline => { + const terminal = { type: "response.completed", response: { id: "resp_fixture", status: "completed", output: [] } }; + const wire = `: ignored${newline}event: response.completed${newline}data: ${JSON.stringify(terminal)}${newline}${newline}`; + const observed: string[] = []; + const inspector = createSseInspector({ onTerminal: status => observed.push(status) }); + for (const char of wire) inspector.feed(encoder.encode(char)); + expect(inspector.terminalSeen()).toBe(true); + expect(observed).toEqual(["completed"]); + inspector.dispose(); + const sent: string[] = []; + const ws = { readyState: 1, data: {} as WsData, send(message: string) { sent.push(message); return 1; } } as unknown as ServerWebSocket; + await pumpResponsesSseToWebSocket(ws, source([...wire])); + expect(sent.map(text => JSON.parse(text))).toEqual([terminal]); + }); + + test.each(["\r", "\n", "\r\n"])("event-field repairs preserve %j payloads with or without a leading comment", newline => { + for (const prefix of ["", `: ignored${newline}`]) { + const block = (type: string, fields: Record) => + `${prefix}event: ${type}${newline}data: ${JSON.stringify({ type, ...fields })}`; + const policy = createSseTerminalOutputBoundary(); + const failed = decoder.decode(policy.feed(encoder.encode(block("error", { + code: "cyber_policy", message: "fixture rejection", + }) + newline + newline))); + expect(JSON.parse(sseDataPayload(nextSseBlock(failed)!.block)!)).toMatchObject({ + type: "response.failed", response: { error: { code: "cyber_policy" } }, + }); + expect(policy.terminalSeen()).toBe(true); + policy.dispose(); + + const custom = createRoutedCustomToolRestoreBlockRewrite(new Set(["exec"]), undefined, new Set(), new Set(["exec"])); + const copilot = createGithubCopilotResponsesBlockRewrite(createTestTranslatorBudget()); + const grok = createGrokResponsesSparseTerminalBlockRewrite(createTestTranslatorBudget(), { tools: [], tool_choice: "none" }); + try { + custom(block("response.output_item.added", { output_index: 0, + item: { type: "function_call", id: "fc_fixture", call_id: "call_fixture", name: "exec", arguments: "" } })); + const done = custom(block("response.function_call_arguments.done", { + output_index: 0, item_id: "fc_fixture", arguments: '{"input":"text(1)"}', + }))[0]!; + expect(JSON.parse(sseDataPayload(done)!)).toMatchObject({ type: "response.custom_tool_call_input.done", input: "text(1)" }); + expect(done.split(/\r\n|\r|\n/)).toContain("event: response.custom_tool_call_input.done"); + + const repaired = copilot(block("response.created", { obfuscation: "padding", response: { id: "resp_fixture" } }))[0]!; + expect(JSON.parse(sseDataPayload(repaired)!)).toMatchObject({ type: "response.created", response: { id: "resp_fixture" } }); + expect(sseDataPayload(repaired)).not.toContain("obfuscation"); + + grok(block("response.output_item.done", { output_index: 0, + item: { type: "function_call", id: "fc_other", call_id: "call_other", name: "other", arguments: "{}" } })); + const refused = grok(block("response.completed", { response: { id: "resp_fixture", status: "completed", output: [] } }))[0]!; + expect(JSON.parse(sseDataPayload(refused)!)).toMatchObject({ type: GROK_REFUSED_TERMINAL_EVENT_TYPE, response: { status: "incomplete" } }); + expect(refused.split(/\r\n|\r|\n/)).toContain(`event: ${GROK_REFUSED_TERMINAL_EVENT_TYPE}`); + expect(createGrokResponsesControlFrameBlockRewrite()(`${prefix}event: codex.rate_limits${newline}data: {}`)).toEqual([]); + } finally { custom.dispose?.(); copilot.dispose?.(); grok.dispose?.(); } + } + }); + +}); From 5a85cf31cc1813a9790d9f8755c945ab25d2fbfe Mon Sep 17 00:00:00 2001 From: Epinephrine Date: Fri, 2 Oct 2026 07:28:56 -0700 Subject: [PATCH 2/2] fix(sse): reconcile split CRLF before block scanning Consume a late LF before delimiter scanning so it cannot create an empty rewrite callback. Retain subsequent line endings for the next event and preserve pending-prefix byte accounting and cancellation. --- src/server/sse-payload-rewrite.ts | 52 ++++++++++--------- .../sse-rewrite-line-endings.test.ts | 22 ++++++++ 2 files changed, 50 insertions(+), 24 deletions(-) diff --git a/src/server/sse-payload-rewrite.ts b/src/server/sse-payload-rewrite.ts index a00dcf9c811..99c67056eaa 100644 --- a/src/server/sse-payload-rewrite.ts +++ b/src/server/sse-payload-rewrite.ts @@ -88,6 +88,7 @@ export function createSseBlockBuffer( append(fragment: string): void; next(): { block: string; delimiter: string } | null; tail(): string; + isEmpty(): boolean; compact(): void; clear(): void; } { @@ -156,6 +157,7 @@ export function createSseBlockBuffer( return { block, delimiter }; }, tail: () => buffer.slice(offset), + isEmpty: () => offset === buffer.length, compact, clear() { budget.releaseRetained(bufferBytes, scope); @@ -310,38 +312,23 @@ export function relaySseWithBlockRewrite( let emitted = 0; let next: { block: string; delimiter: string } | null; while (!cancelled && (next = buffer.next())) { - // A final CR dispatches immediately. A later LF extends that same delimiter and - // must not become a field, or enter the next block's rewrite/drop/injection. - let { block, delimiter } = next; - if (pendingLineFeed && (block || delimiter).startsWith("\n")) { - if (pendingLineFeedVisible) { - enqueueText(controller, "\n"); - emitted += 1; - } - if (block) block = block.slice(1); - else delimiter = delimiter.slice(1); - } - pendingLineFeed = delimiter.endsWith("\r"); + const { block, delimiter } = next; + // A CR followed by buffered text is already settled; only an end-of-buffer CR + // can still receive the LF that extends its delimiter in the next fragment. + pendingLineFeed = delimiter.endsWith("\r") && buffer.isEmpty(); const outBlocks = rewrite(block); pendingLineFeedVisible = outBlocks.length > 0; if (cancelled) return emitted; for (let index = 0; index < outBlocks.length; index++) { // Synthetic earlier blocks need a settled delimiter; only the final one can // inherit an upstream LF in a later chunk. Completing CR as CRLF is equivalent. - const settled = index < outBlocks.length - 1 && pendingLineFeed ? delimiter + "\n" : delimiter; + const settled = index < outBlocks.length - 1 && delimiter.endsWith("\r") ? delimiter + "\n" : delimiter; enqueueText(controller, outBlocks[index]! + settled); emitted += 1; } } buffer.compact(); - let tail = flushFinal ? buffer.tail() : ""; - if (pendingLineFeed && tail.startsWith("\n")) { - if (pendingLineFeedVisible) { - enqueueText(controller, "\n"); - emitted += 1; - } - tail = tail.slice(1); - } + const tail = flushFinal ? buffer.tail() : ""; if (tail.length > 0) { const tailBlocks = rewrite(tail); if (cancelled) return emitted; @@ -357,6 +344,23 @@ export function relaySseWithBlockRewrite( return emitted; }; + const appendFragment = (controller: ReadableStreamDefaultController, fragment: string): number => { + let continuation = false; + if (pendingLineFeed && fragment.length > 0) { + continuation = fragment.startsWith("\n"); + pendingLineFeed = false; + if (continuation) fragment = fragment.slice(1); + } + // Consume a late LF before scanning, so it cannot pair with a new LF to create + // a phantom empty callback. Any following line ending stays in the input buffer. + buffer.append(fragment); + if (continuation && pendingLineFeedVisible) { + enqueueText(controller, "\n"); + return 1; + } + return 0; + }; + return new ReadableStream({ async pull(controller) { try { @@ -371,7 +375,7 @@ export function relaySseWithBlockRewrite( // after its disposal (#893 review). if (cancelled) return; if (done) { - buffer.append(decoder.decode()); + appendFragment(controller, decoder.decode()); emitProcessedBlocks(controller, true); if (cancelled) return; buffer.clear(); @@ -379,8 +383,8 @@ export function relaySseWithBlockRewrite( controller.close(); return; } - buffer.append(decoder.decode(value, { stream: true })); - const emitted = emitProcessedBlocks(controller); + const continuation = appendFragment(controller, decoder.decode(value, { stream: true })); + const emitted = continuation + emitProcessedBlocks(controller); if (cancelled || emitted > 0) return; } } catch (error) { diff --git a/tests/responses/sse-rewrite-line-endings.test.ts b/tests/responses/sse-rewrite-line-endings.test.ts index d10b5431c74..739807f8790 100644 --- a/tests/responses/sse-rewrite-line-endings.test.ts +++ b/tests/responses/sse-rewrite-line-endings.test.ts @@ -72,6 +72,28 @@ describe("shared SSE line ending contract", () => { buffer.clear(); expect(budget.snapshot().currentBytes).toBe(0); }); + test("a late LF is reconciled before delimiter scanning without an empty callback", async () => { + const budget = createTestTranslatorBudget(); + let controller!: ReadableStreamDefaultController; + const upstream = new ReadableStream({ start(value) { controller = value; } }); + const seen: string[] = []; + const reader = relaySseWithBlockRewrite(upstream, block => { seen.push(block); return [block]; }, budget).getReader(); + try { + controller.enqueue(encoder.encode("data: one\r\r")); + expect(decoder.decode((await reader.read()).value)).toBe("data: one\r\r"); + controller.enqueue(encoder.encode("\n\n")); + expect(decoder.decode((await reader.read()).value)).toBe("\n"); + expect(seen).toEqual(["data: one"]); + expect(budget.snapshot().currentBytes).toBe(1); // The second LF still belongs to the input buffer. + controller.enqueue(encoder.encode("data: two\n\n")); + expect(decoder.decode((await reader.read()).value)).toBe("\ndata: two\n\n"); + expect(seen).toEqual(["data: one", "\ndata: two"]); + } finally { + await reader.cancel(); + expect(budget.snapshot().currentBytes).toBe(0); + } + }); + test("incremental delimiter search and byte counts stay linear for CR fragments", () => { const budget = createTestTranslatorBudget(); const buffer = createSseBlockBuffer(budget);