From 7d138d4650996570ffd13b8d7f078ddc0120034f Mon Sep 17 00:00:00 2001 From: 0disoft Date: Wed, 28 Oct 2026 23:49:03 +0900 Subject: [PATCH 1/2] perf(anthropic): scan SSE inspection fragments incrementally Normalize and scan only newly decoded text, retain split CR/LF boundaries, and coalesce tiny fragments before joining completed frames. Add deterministic work-budget and framing regressions while preserving passthrough bytes and terminal accounting. --- scripts/test-layout/layout.json | 2 +- src/server/messages-shared.ts | 63 ++++++++-- structure/clients/claude-desktop.md | 2 +- .../messages-shared-sse.test.ts | 118 ++++++++++++++++++ tests/fixtures/test-layout-expected.json | 1 + 5 files changed, 171 insertions(+), 15 deletions(-) create mode 100644 tests/claude-integration/messages-shared-sse.test.ts diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index eab132c488c..cb02e32428c 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -450,7 +450,7 @@ "claude-manual-env.test.ts": "gui", "claude-messages-endpoint.test.ts": "claude-integration", "claude-messages-thread.test.ts": "claude-integration", - "messages-native.test.ts": "claude-integration", + "messages-native.test.ts": "claude-integration", "messages-shared-sse.test.ts": "claude-integration", "messages-native-decline-trace.test.ts": "claude-integration", "messages-native-oauth.test.ts": "claude-integration", "messages-native-opaque-state.test.ts": "claude-integration", diff --git a/src/server/messages-shared.ts b/src/server/messages-shared.ts index 8444fd73737..4fe22574ab7 100644 --- a/src/server/messages-shared.ts +++ b/src/server/messages-shared.ts @@ -82,7 +82,31 @@ export function tapAnthropicSseForLog( ): ReadableStream { const decoder = new TextDecoder(); const encoder = new TextEncoder(); - let buffer = ""; + let frameParts: string[] = []; + let pendingParts: string[] = []; + let pendingChars = 0; + let heldCr = false; + let heldLf = false; + const appendPart = (part: string) => { + if (!part) return; + pendingParts.push(part); + pendingChars += part.length; + // Coalesce tiny transport fragments: a one-byte feed must not retain one array + // entry per byte for the entire event. Each character is joined at most twice. + if (pendingChars >= 4096) { + frameParts.push(pendingParts.join("")); + pendingParts = []; + pendingChars = 0; + } + }; + const takeFrame = () => { + if (pendingParts.length) frameParts.push(pendingParts.join("")); + const frame = frameParts.join(""); + frameParts = []; + pendingParts = []; + pendingChars = 0; + return frame; + }; let usageAcc: Rec = {}; // message_stop or an upstream error event: the turn's own terminal already went through. let terminalSeen = false; @@ -106,17 +130,29 @@ export function tapAnthropicSseForLog( } }; const inspect = (chunk: Uint8Array) => { - buffer += decoder.decode(chunk, { stream: true }); - // SSE lines may end in CRLF, LF or CR. Normalize the inspection copy to LF (the forwarded - // bytes are untouched), holding a trailing CR until the next chunk shows whether an LF - // follows it, so a CRLF split across chunks stays one line ending. - const heldCr = buffer.endsWith("\r"); - buffer = (heldCr ? buffer.slice(0, -1) : buffer).replace(/\r\n?/g, "\n") + (heldCr ? "\r" : ""); + // Normalize and scan only new text. Keep an unfinished frame in fragments so a long + // SSE event arriving in small chunks is never flattened or rescanned on each read. + let text = (heldCr ? "\r" : "") + decoder.decode(chunk, { stream: true }); + heldCr = text.endsWith("\r"); + text = (heldCr ? text.slice(0, -1) : text).replace(/\r\n?/g, "\n"); + let start = 0; + if (text && heldLf) { + heldLf = false; + if (text[0] === "\n") { + inspectFrame(takeFrame()); + start = 1; + } else appendPart("\n"); + } let sep: number; - while ((sep = buffer.indexOf("\n\n")) !== -1) { - const frame = buffer.slice(0, sep); - buffer = buffer.slice(sep + 2); - inspectFrame(frame); + while ((sep = text.indexOf("\n\n", start)) !== -1) { + appendPart(text.slice(start, sep)); + inspectFrame(takeFrame()); + start = sep + 2; + } + if (start < text.length) { + heldLf = text.endsWith("\n"); + const end = text.length - (heldLf ? 1 : 0); + appendPart(text.slice(start, end)); } }; // The last block can sit in the buffer without its blank line: an upstream that stopped after @@ -125,8 +161,9 @@ export function tapAnthropicSseForLog( // client has not seen a blank line after, so the caller restores one. const flushTail = (): boolean => { const terminalBefore = terminalSeen; - const tail = (buffer + decoder.decode()).replace(/\r\n?/g, "\n"); - buffer = ""; + const tail = (takeFrame() + (heldLf ? "\n" : "") + (heldCr ? "\r" : "") + decoder.decode()).replace(/\r\n?/g, "\n"); + heldCr = false; + heldLf = false; if (tail) inspectFrame(tail); return terminalSeen && !terminalBefore && !tail.endsWith("\n\n"); }; diff --git a/structure/clients/claude-desktop.md b/structure/clients/claude-desktop.md index 4c259199390..ecb8eba67bc 100644 --- a/structure/clients/claude-desktop.md +++ b/structure/clients/claude-desktop.md @@ -593,7 +593,7 @@ Native Anthropic passthrough in `src/server/claude-messages.ts` forwards the cal Native passthrough retains upstream `anthropic-ratelimit-*` response headers for Claude Code quota/statusLine consumers on SSE, JSON and upstream errors. `src/server/anthropic-rate-limit-headers.ts` selects only that family instead of copying all upstream headers, so cookies and unrelated metadata are not relayed. Missing rate-limit headers are not fabricated; body, status, content type and existing non-stream `Retry-After` behavior stay unchanged. `tests/claude-integration/claude-native-rate-limit-headers.test.ts` exercises the production ingress against a synthetic upstream for all four response shapes (SSE, JSON, upstream error and count_tokens). -`tapAnthropicSseForLog` in `src/server/claude-messages.ts` relays the streamed body of both the native passthrough and the managed native Messages lane (`src/server/messages-native.ts`). The response headers are already sent, so a stall, a byte-cap overflow, or an upstream read failure ends the body with an Anthropic `event: error` frame after a blank-line boundary and a clean close: `timeout_error` for an idle stall, `api_error` for the byte cap, and `api_error` when an upstream read fails mid-stream (a socket reset). The mid-stream reset is logged like the Responses relay's read error: status 502, `terminalStatus: "failed"`, `closeReason: "terminal"`, `transportPhase: "mid_stream"`, a synthetic terminal source, the attempt marked `streamAborted`, the redacted reason in `upstreamError`, and the usage seen before the reset. The non-streaming fold in the managed lane closes its row with the tap's meta for a reset, a stall or the byte cap, so its row matches the streaming lane's. The request is not replayed. Every one of these frames logs status 502. A stall is `terminalStatus: "incomplete"` with `closeReason: "body_stall"`, the status the Responses relay gives a stall-timeout incomplete. The byte cap is `incomplete` with `closeReason: "body_overflow"`, the same 502 the non-stream passthrough answers for it. Both carry the proxy's message in `upstreamError`, and a reset is `terminalStatus: "failed"`. Either way the row keeps its failure diagnostics in usage.jsonl. A stall or overflow after the turn's own terminal (`message_stop` or an upstream `error` event) is a finished turn: it logs 200 and closes without an error frame, as the read-error branch does. The tap finds frames in a copy normalized to LF, because SSE lines may end in CRLF, LF or CR; it holds a trailing CR until the next chunk so a split CRLF stays one line ending, and forwards the original bytes unchanged. Some read failures are not upstream failures. When the cancel signal is already aborted, the rejection is a `499` client cancel, because Bun can reject the read before it dispatches the abort listener. The managed lane passes its upstream controller's signal, so shutdown and turn release count as cancels too. When `message_stop` or an upstream `error` event has been seen, including one still in the buffer without its blank-line delimiter, the turn is complete: it logs 200 and closes with no error frame. A terminal found only in that unterminated tail gets its blank line restored, because an SSE parser drops an event that EOF cuts off. A translator budget overflow is a local cap, so it still errors the stream, and the non-streaming fold answers it with 413. `tests/claude-integration/claude-native-passthrough.test.ts` and `tests/claude-integration/messages-native.test.ts` cover both lanes against an upstream that resets after a partial or a complete body, plus both cancel paths. +`tapAnthropicSseForLog` in `src/server/claude-messages.ts` relays the streamed body of both the native passthrough and the managed native Messages lane (`src/server/messages-native.ts`). The response headers are already sent, so a stall, a byte-cap overflow, or an upstream read failure ends the body with an Anthropic `event: error` frame after a blank-line boundary and a clean close: `timeout_error` for an idle stall, `api_error` for the byte cap, and `api_error` when an upstream read fails mid-stream (a socket reset). The mid-stream reset is logged like the Responses relay's read error: status 502, `terminalStatus: "failed"`, `closeReason: "terminal"`, `transportPhase: "mid_stream"`, a synthetic terminal source, the attempt marked `streamAborted`, the redacted reason in `upstreamError`, and the usage seen before the reset. The non-streaming fold in the managed lane closes its row with the tap's meta for a reset, a stall or the byte cap, so its row matches the streaming lane's. The request is not replayed. Every one of these frames logs status 502. A stall is `terminalStatus: "incomplete"` with `closeReason: "body_stall"`, the status the Responses relay gives a stall-timeout incomplete. The byte cap is `incomplete` with `closeReason: "body_overflow"`, the same 502 the non-stream passthrough answers for it. Both carry the proxy's message in `upstreamError`, and a reset is `terminalStatus: "failed"`. Either way the row keeps its failure diagnostics in usage.jsonl. A stall or overflow after the turn's own terminal (`message_stop` or an upstream `error` event) is a finished turn: it logs 200 and closes without an error frame, as the read-error branch does. The tap normalizes and scans only newly decoded text, holding a trailing CR and LF across reads so split CRLF and blank-line delimiters stay intact; it forwards the original bytes unchanged. Unfinished frames remain in fragments and are joined only at a frame boundary or tail flush. Tiny fragments are coalesced after at least 4096 code units, avoiding one retained array entry per byte on one-byte feeds. Each code unit is normalized once and joined at most twice; `tests/claude-integration/messages-shared-sse.test.ts` checks the linear work budget, tiny-fragment coalescing, split UTF-8 and line endings, usage, terminal-tail repair, byte caps and cancellation. Some read failures are not upstream failures. When the cancel signal is already aborted, the rejection is a `499` client cancel, because Bun can reject the read before it dispatches the abort listener. The managed lane passes its upstream controller's signal, so shutdown and turn release count as cancels too. When `message_stop` or an upstream `error` event has been seen, including one still in the buffer without its blank-line delimiter, the turn is complete: it logs 200 and closes with no error frame. A terminal found only in that unterminated tail gets its blank line restored, because an SSE parser drops an event that EOF cuts off. A translator budget overflow is a local cap, so it still errors the stream, and the non-streaming fold answers it with 413. `tests/claude-integration/claude-native-passthrough.test.ts` and `tests/claude-integration/messages-native.test.ts` cover both lanes against an upstream that resets after a partial or a complete body, plus both cancel paths. A native passthrough answered without a stream records its reason in `upstreamError`, so the row and the failure diagnostics in usage.jsonl name the cause. An upstream error response (status 400 or above) is relayed verbatim; the stored diagnostic is `Provider error : ` only for a valid Anthropic error envelope with one of the closed types `invalid_request_error`, `authentication_error`, `permission_error`, `not_found_error`, `rate_limit_error`, `api_error`, `overloaded_error` or `request_too_large`. Upstream messages and arbitrary type strings never enter this diagnostic. Unknown or malformed envelopes, non-JSON bodies and bodies over 64 Ki characters log `Provider error `. Local header timeout, body stall, byte-cap overflow and cancel diagnostics contain fixed text plus validated guard limits. Fetch failures log the fixed reason `anthropic passthrough failed: upstream connection error`; the existing redacted client response is preserved. Classification uses these stored reasons and HTTP status, so a `permission_error` at 403 is `permission_denied` regardless of upstream message wording. `tests/claude-integration/claude-native-passthrough.test.ts` covers these cases and checks that echoed account identifiers and request content are absent from both history sinks. diff --git a/tests/claude-integration/messages-shared-sse.test.ts b/tests/claude-integration/messages-shared-sse.test.ts new file mode 100644 index 00000000000..365b88eb46e --- /dev/null +++ b/tests/claude-integration/messages-shared-sse.test.ts @@ -0,0 +1,118 @@ +import { expect, spyOn, test } from "bun:test"; +import { tapAnthropicSseForLog, type PassthroughBodyGuard, type PassthroughFinalizeMeta } from "../../src/server/messages-shared"; +import type { RequestLogContext } from "../../src/server/request-log"; + +const encoder = new TextEncoder(); +const start = 'data: {"type":"message_start","message":{"usage":{"input_tokens":12,"output_tokens":0}}}'; +const delta = 'data: {"type":"message_delta","usage":{"output_tokens":5}}'; +const stop = 'data: {"type":"message_stop"}'; + +function tap(bytes: Uint8Array, chunkSize: number, ending: "close" | "error" | "stall" = "close", guard?: PassthroughBodyGuard) { + let offset = 0; + let cancelled = false; + const source = new ReadableStream({ + pull(controller) { + if (offset < bytes.length) { + const end = Math.min(bytes.length, offset + chunkSize); + controller.enqueue(bytes.subarray(offset, end)); + offset = end; + } else if (ending === "close") controller.close(); + else if (ending === "error") controller.error(new Error("fixture reset")); + }, + cancel() { cancelled = true; }, + }); + const log: RequestLogContext = { model: "fixture", provider: "fixture" }; + const calls: Array<{ status: number; meta: PassthroughFinalizeMeta }> = []; + const stream = tapAnthropicSseForLog(source, log, (status, meta) => calls.push({ status, meta }), guard); + return { stream, log, calls, cancelled: () => cancelled }; +} + +test("fragmented large Anthropic frames have a linear normalization budget", async () => { + const content = `data: ${JSON.stringify({ type: "content_block_delta", delta: { type: "text_delta", text: "x".repeat(256 * 1024) } })}`; + const bytes = encoder.encode([start, content, delta, stop, ""].join("\n\n")); + const original = String.prototype.replace; + let normalizedChars = 0; + const replace = spyOn(String.prototype, "replace"); + replace.mockImplementation(function (this: string, ...args: Parameters) { + if (args[0] instanceof RegExp && args[0].source === "\\r\\n?") normalizedChars += this.length; + return Reflect.apply(original, this, args); + }); + try { + const result = tap(bytes, 256); + expect(new Uint8Array(await new Response(result.stream).arrayBuffer())).toEqual(bytes); + expect(result.log.usage).toEqual({ inputTokens: 12, outputTokens: 5 }); + expect(result.calls).toEqual([{ status: 200, meta: { closeReason: "terminal" } }]); + // Count actual normalization input rather than asserting a machine-dependent time limit. + expect(normalizedChars).toBeLessThanOrEqual(bytes.length * 2); + } finally { + replace.mockRestore(); + } +}); + +test("one-byte fragments are coalesced instead of retaining one slot per byte", async () => { + const bytes = encoder.encode(`${start}\n\ndata: ${JSON.stringify({ padding: "x".repeat(32 * 1024) })}\n\n${stop}\n\n`); + const original = Array.prototype.join; + let maxParts = 0; + const join = spyOn(Array.prototype, "join"); + join.mockImplementation(function (this: unknown[], separator?: string) { + if (separator === "") maxParts = Math.max(maxParts, this.length); + return Reflect.apply(original, this, [separator]); + }); + try { + const result = tap(bytes, 1); + const output = new Uint8Array(await new Response(result.stream).arrayBuffer()); + join.mockRestore(); + expect(output).toEqual(bytes); + expect(maxParts).toBeLessThanOrEqual(8192); + } finally { + join.mockRestore(); + } +}); + +for (const eol of ["\n", "\r\n", "\r", "\n\r", "\r\n\n"]) { + test(`Anthropic inspection preserves fragmented ${JSON.stringify(eol)} framing and UTF-8`, async () => { + const content = `data: ${JSON.stringify({ type: "content_block_delta", delta: { type: "text_delta", text: "한글 🌸" } })}`; + const bytes = encoder.encode([start, content, delta, stop, ""].join("\n\n").replace(/\n/g, eol)); + const result = tap(bytes, 1, "stall", { stallMs: 30, maxBytes: 0 }); + expect(new Uint8Array(await new Response(result.stream).arrayBuffer())).toEqual(bytes); + expect(result.log.usage).toEqual({ inputTokens: 12, outputTokens: 5 }); + expect(result.log.generationStartedAt).toBeNumber(); + expect(result.calls).toEqual([{ status: 200, meta: { closeReason: "terminal" } }]); + expect(result.cancelled()).toBe(true); + }); +} + +for (const suffix of ["", "\n", "\r"]) { + test(`a reset after an unterminated terminal ${JSON.stringify(suffix)} restores its delimiter`, async () => { + const text = `${start}\n\n${delta}\n\n${stop}${suffix}`; + const result = tap(encoder.encode(text), 1, "error"); + expect(await new Response(result.stream).text()).toBe(`${text}\n\n`); + expect(result.log.usage).toEqual({ inputTokens: 12, outputTokens: 5 }); + expect(result.calls).toEqual([{ status: 200, meta: { closeReason: "terminal" } }]); + }); +} + +test("a reset before a terminal still emits the Anthropic error and records usage", async () => { + const result = tap(encoder.encode(`${start}\n\n${delta}\n\n`), 3, "error"); + expect(await new Response(result.stream).text()).toContain('"type":"error"'); + expect(result.log.usage).toEqual({ inputTokens: 12, outputTokens: 5 }); + expect(result.calls).toEqual([{ status: 502, meta: { terminalStatus: "failed", closeReason: "terminal" } }]); +}); + +test("a body cap after a held CR terminal preserves the completed turn", async () => { + const text = `${start}\n\n${delta}\n\n${stop}\r`; + const bytes = encoder.encode(`${text}${"x".repeat(64)}`); + const result = tap(bytes, encoder.encode(text).length, "close", { stallMs: 0, maxBytes: encoder.encode(text).length }); + expect(await new Response(result.stream).text()).toBe(`${text}\n\n`); + expect(result.calls).toEqual([{ status: 200, meta: { closeReason: "terminal" } }]); +}); + +test("client cancellation still settles once and cancels the source", async () => { + const abort = new AbortController(); + const result = tap(encoder.encode(`${start}\n\n`), 3, "stall", { stallMs: 0, maxBytes: 0, reqSignal: abort.signal }); + const text = new Response(result.stream).text(); + abort.abort(); + await text; + expect(result.calls).toEqual([{ status: 499, meta: { closeReason: "client_cancel" } }]); + expect(result.cancelled()).toBe(true); +}); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index e503df551af..6006a24361b 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -471,6 +471,7 @@ "messages-native-opaque-state.test.ts": "claude-integration", "messages-surface-matrix.test.ts": "claude-integration", "claude-model-info.test.ts": "claude-integration", "claude-models-discovery.test.ts": "claude-integration", "claude-native-affinity.test.ts": "claude-integration", "claude-native-passthrough.test.ts": "claude-integration", + "messages-shared-sse.test.ts": "claude-integration", "claude-native-rate-limit-headers.test.ts": "claude-integration", "claude-outbound.test.ts": "claude-integration", "claude-context-accounting.test.ts": "claude-integration", "claude-long-context-eligibility.test.ts": "claude-integration", From 156cb7ac9549e008d4b9b7a210a583199d5730bd Mon Sep 17 00:00:00 2001 From: JUN <243035832+lidge-jun@users.noreply.github.com> Date: Thu, 29 Oct 2026 06:50:02 +0900 Subject: [PATCH 2/2] test(anthropic): bound cumulative SSE fragment copying Measure cumulative joined characters across 128 KiB of one-byte fragments. The negative control rebuilding accumulated groups fails the linear bound. Co-authored-by: 0disoft <100320863+0disoft@users.noreply.github.com> --- tests/claude-integration/messages-shared-sse.test.ts | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/tests/claude-integration/messages-shared-sse.test.ts b/tests/claude-integration/messages-shared-sse.test.ts index 365b88eb46e..8f3599e05d8 100644 --- a/tests/claude-integration/messages-shared-sse.test.ts +++ b/tests/claude-integration/messages-shared-sse.test.ts @@ -50,12 +50,16 @@ test("fragmented large Anthropic frames have a linear normalization budget", asy }); test("one-byte fragments are coalesced instead of retaining one slot per byte", async () => { - const bytes = encoder.encode(`${start}\n\ndata: ${JSON.stringify({ padding: "x".repeat(32 * 1024) })}\n\n${stop}\n\n`); + const bytes = encoder.encode(`${start}\n\ndata: ${JSON.stringify({ padding: "x".repeat(128 * 1024) })}\n\n${stop}\n\n`); const original = Array.prototype.join; let maxParts = 0; + let joinedChars = 0; const join = spyOn(Array.prototype, "join"); join.mockImplementation(function (this: unknown[], separator?: string) { - if (separator === "") maxParts = Math.max(maxParts, this.length); + if (separator === "") { + maxParts = Math.max(maxParts, this.length); + for (const part of this) if (typeof part === "string") joinedChars += part.length; + } return Reflect.apply(original, this, [separator]); }); try { @@ -64,6 +68,8 @@ test("one-byte fragments are coalesced instead of retaining one slot per byte", join.mockRestore(); expect(output).toEqual(bytes); expect(maxParts).toBeLessThanOrEqual(8192); + expect(joinedChars).toBeGreaterThanOrEqual(bytes.length); + expect(joinedChars).toBeLessThanOrEqual(bytes.length * 4); } finally { join.mockRestore(); }