diff --git a/changelog.d/fixes/12925-glm-stream-buffer-arity.md b/changelog.d/fixes/12925-glm-stream-buffer-arity.md new file mode 100644 index 0000000000..3267f4f7cb --- /dev/null +++ b/changelog.d/fixes/12925-glm-stream-buffer-arity.md @@ -0,0 +1 @@ +- **fix(stream):** the 64 KB stream buffer GLM asks for is honoured instead of dropped, and the type error it caused no longer fails the API Route Typecheck gate on every open PR ([#12925](https://github.com/diegosouzapw/OmniRoute/pull/12925)) diff --git a/open-sse/executors/glm.ts b/open-sse/executors/glm.ts index c275e6f290..329c0b9da9 100644 --- a/open-sse/executors/glm.ts +++ b/open-sse/executors/glm.ts @@ -216,6 +216,9 @@ function translateAnthropicJsonError(parsed: unknown): JsonRecord { }; } +/** 64 KB queue budget for GLM streaming (#12179, wired through in #12925). */ +const GLM_STREAM_BUFFER_BYTES = 65536; + export function translateSseResponse( response: Response, provider: string, @@ -223,8 +226,11 @@ export function translateSseResponse( suppressThinkClose: boolean = false ): Response { if (!response.body) return response; - // Helper has 15 parameters; a 16th positional (65536) was a TS2554 and - // never reached TransformStream. highWaterMark stays at the helper default. + // GLM is a high-throughput provider: a 64 KB queue budget keeps provider -> + // client pacing ahead of the model's emission rate. #12179 asked for this by + // passing a 16th positional the helper did not take (a TS2554 that never + // reached the TransformStream); the helper now accepts it as its last + // parameter, so the request finally takes effect (#12925). const transform = createSSETransformStreamWithLogger( FORMATS.CLAUDE, FORMATS.OPENAI, @@ -238,7 +244,10 @@ export function translateSseResponse( null, null, false, - suppressThinkClose + suppressThinkClose, + undefined, + undefined, + GLM_STREAM_BUFFER_BYTES ); const headers = cloneHeaders(response.headers); headers.set("content-type", "text/event-stream"); diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index a5d761063c..4051bb647a 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -145,6 +145,9 @@ type StreamCompletePayload = { interrupted?: boolean; }; +/** Queue budget every provider used before `streamBufferBytes` existed. */ +const DEFAULT_STREAM_BUFFER_BYTES = 16384; + type StreamOptions = { mode?: string; targetFormat?: string; @@ -160,6 +163,14 @@ type StreamOptions = { */ dropResponsesCommentary?: boolean; customToolNames?: ReadonlySet; + /** + * Byte budget for the transform's readable and writable queues. + * + * Defaults to the 16 KB every provider used before this was configurable. A + * high-throughput provider can raise it so provider -> client pacing stays + * ahead of the model's emission rate; nothing else should need to. + */ + streamBufferBytes?: number; provider?: string | null; reqLogger?: StreamLogger | null; toolNameMap?: unknown; @@ -655,6 +666,7 @@ export function createSSEStream(options: StreamOptions = {}) { dropResponsesCommentary, customToolNames = new Set(), requestToolIdentityMap = null, + streamBufferBytes = DEFAULT_STREAM_BUFFER_BYTES, } = options; const signatureNamespace = connectionId; // Request-body-size metric (for monitoring payload size distribution & correlation with TTFT). @@ -1103,7 +1115,8 @@ export function createSSEStream(options: StreamOptions = {}) { cacheHit: false, latencyMs: Date.now() - streamStartedAt, usage: timing.withTps(finalUsage), - costUsd, ttftMs: timing.ttftMs(), + costUsd, + ttftMs: timing.ttftMs(), }); if (!comment) return; reqLogger?.appendConvertedChunk?.(comment); @@ -2069,7 +2082,9 @@ export function createSSEStream(options: StreamOptions = {}) { // estimate is now emitted in flush(), only when the upstream stayed silent. if (isFinishChunk && hasValidUsage(usage) && !passthroughForwardedUsage) { const buffered = addBufferToUsage(usage); - parsed.usage = timing.withTps(filterUsageForFormat(buffered, sourceFormat || FORMATS.OPENAI)); + parsed.usage = timing.withTps( + filterUsageForFormat(buffered, sourceFormat || FORMATS.OPENAI) + ); output = `data: ${JSON.stringify(parsed)}\n\n`; passthroughForwardedUsage = true; injectedUsage = true; @@ -3020,8 +3035,8 @@ export function createSSEStream(options: StreamOptions = {}) { clearIdleTimer(); }, }, - { highWaterMark: 16384 }, - { highWaterMark: 16384 } + { highWaterMark: streamBufferBytes }, + { highWaterMark: streamBufferBytes } ); } @@ -3043,7 +3058,8 @@ export function createSSETransformStreamWithLogger( copilotCompatibleReasoning = false, suppressThinkClose = false, customToolNames: ReadonlySet = new Set(), - requestToolIdentityMap: Map | null = null + requestToolIdentityMap: Map | null = null, + streamBufferBytes: number = DEFAULT_STREAM_BUFFER_BYTES ) { return createSSEStream({ mode: STREAM_MODE.TRANSLATE, @@ -3062,6 +3078,7 @@ export function createSSETransformStreamWithLogger( suppressThinkClose, customToolNames, requestToolIdentityMap, + streamBufferBytes, }); } diff --git a/tests/unit/sse-stream-buffer-bytes.test.ts b/tests/unit/sse-stream-buffer-bytes.test.ts new file mode 100644 index 0000000000..be54b24f86 --- /dev/null +++ b/tests/unit/sse-stream-buffer-bytes.test.ts @@ -0,0 +1,96 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { + createSSEStream, + createSSETransformStreamWithLogger, +} from "../../open-sse/utils/stream.ts"; +import { FORMATS } from "../../open-sse/translator/formats.ts"; + +// A TransformStream's writable queue starts with `desiredSize === highWaterMark`, +// so reading it off a fresh writer measures the queue budget the stream was +// actually built with rather than standing in for it. +// Each stream arms a 10s idle watchdog (setInterval in createSSEStream's start). +// Cancelling the readable runs the TransformStream's cancel handler, which clears +// it — without this the node:test runner never sees an empty event loop and the +// file hangs after the assertions have already passed. +const openStreams: TransformStream[] = []; + +const writableBudget = (transform: TransformStream) => { + openStreams.push(transform); + return transform.writable.getWriter().desiredSize; +}; + +test.after(async () => { + for (const transform of openStreams) { + await transform.readable.cancel().catch(() => {}); + } +}); + +const DEFAULT = 16384; + +test.describe("SSE stream buffer budget", () => { + test("defaults to the 16 KB every provider used before it was configurable", () => { + const transform = createSSEStream({ + targetFormat: FORMATS.CLAUDE, + sourceFormat: FORMATS.OPENAI, + }); + + assert.equal(writableBudget(transform), DEFAULT); + }); + + test("createSSEStream honours an explicit budget", () => { + const transform = createSSEStream({ + targetFormat: FORMATS.CLAUDE, + sourceFormat: FORMATS.OPENAI, + streamBufferBytes: 65536, + }); + + assert.equal(writableBudget(transform), 65536); + }); + + // The defect this pins: glm.ts has passed a 16th positional argument since + // #12179, and the signature stopped at 15. It was a type error, and the value + // was dropped — the 64 KB that call site asks for never reached the queue. + // These are the exact 16 arguments glm.ts passes. + test("the convenience wrapper carries a 16th positional budget through", () => { + const transform = createSSETransformStreamWithLogger( + FORMATS.CLAUDE, + FORMATS.OPENAI, + "zai", + null, + null, + "glm-4.6", + null, + null, + null, + null, + null, + false, + false, + undefined, + undefined, + 65536 + ); + + assert.equal(writableBudget(transform), 65536); + }); + + test("the wrapper still defaults when no budget is given", () => { + const transform = createSSETransformStreamWithLogger(FORMATS.CLAUDE, FORMATS.OPENAI); + + assert.equal(writableBudget(transform), DEFAULT); + }); + + test("a budget of 0 is honoured rather than treated as absent", () => { + // `?? DEFAULT` and `|| DEFAULT` differ here, and 0 is a legitimate + // highWaterMark: it makes the queue apply backpressure immediately. + const transform = createSSEStream({ + targetFormat: FORMATS.CLAUDE, + sourceFormat: FORMATS.OPENAI, + streamBufferBytes: 0, + }); + + assert.equal(writableBudget(transform), 0); + }); +});