fix(sse): align event framing before tool validation

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

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",

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];

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<string, true> = {
   "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);

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);
 }
 
 /**

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<BoundedSseFrameBuffer["feed"]>,
   ): 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<string, unknown>;
 
@@ -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;
       }

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<string, unknown> {
 }
 
 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:")) {

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,39 +43,32 @@ 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(
   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;
+  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;
         }