fix(stream): accept the buffer size glm.ts has been passing since #12179 (#12925)

Rebased onto the tip and completed, per the maintainer's call to finish the wiring rather than merge the capability alone.

What changed since your version:

The tip had already cleared the TS2554 by deleting the 16th argument, leaving a comment that the highWaterMark stays at the helper default. So the base-red you found is gone, but the 64 KB #12179 asked for was still not applied and your new parameter had no caller. glm.ts now passes it, which is what turns the capability into the fix.

Your test file also hung the runner: every stream createSSEStream builds arms a 10s idle watchdog via setInterval in start, and nothing cancelled them, so node:test waited on a non-empty event loop long after the assertions passed. Cancelling each readable in an after hook runs the cancel handler that clears the timer — the file now reports in about 7 seconds. Worth knowing for future stream tests.

Your five assertions are unchanged and all pass. Reading the writable's desiredSize to measure the queue budget the stream was actually built with, rather than standing in for it, is the detail that makes this testable at all — and the 0-budget case pinning `??` against `||` is the kind of thing that silently rots otherwise.

Thank you also for separating your own red checks from the base's and reporting what you found there. That is how #12919's identical failures got explained instead of chased.

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))

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");

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<string>;
+  /**
+   * 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<string>(),
     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<string> = new Set(),
-  requestToolIdentityMap: Map<string, { namespace: string; name: string }> | null = null
+  requestToolIdentityMap: Map<string, { namespace: string; name: string }> | 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,
   });
 }
 

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);
+  });
+});