fix(adapter-utils): pause the http2 bridge request body until admission

A queued stream held its full request body in host memory during the
admission wait, because the body capture put the stream in flowing
mode as soon as it attached its listeners. This broke the plan's rule
that a queued stream holds no body bytes and adds no new buffer.

Pause the stream in the same turn as the listener attach, and resume
it only after the admission gate grants a slot. The peer's own
flow-control window now stalls a queued stream at the wire instead of
the JS layer buffering the body early.

Co-authored-by: Paperclip <noreply@paperclip.ing>

packages/adapter-utils/src/http2-bridge-server.test.ts

@@ -1210,6 +1210,162 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => {
       }
     });
 
+    it("a queued stream reads no request body bytes until the gate admits it", async () => {
+      const admissionGate = new Http2BridgeAdmissionGate({ maxParallel: 1, maxSessions: 4 });
+      let releaseFirstStream: (() => void) | undefined;
+      const firstStreamHeld = new Promise<void>((resolve) => {
+        releaseFirstStream = resolve;
+      });
+      const secondForwardTracker = createForwarderCallTracker();
+      // Count the `data` events the second stream's underlying transport
+      // delivers. A session-level listener sees the same stream object the
+      // request handler holds, so this counts real transport delivery, not
+      // a private field inside the handler.
+      let secondStreamDataEvents = 0;
+      const { handle, bridgeToken, clientSide } = bindTestServer({
+        admissionGate,
+        forwardRequest: async (request) => {
+          if (request.pathname === "/api/agents/me") {
+            // Hold the only slot until the test explicitly releases it.
+            await firstStreamHeld;
+            return { status: 200 };
+          }
+          secondForwardTracker.markCalled();
+          return { status: 200, body: JSON.stringify({ bodyLength: request.body.byteLength }) };
+        },
+        onSession: (session) => {
+          session.on("stream", (stream, headers) => {
+            if (headers[":path"] === "/api/issues/abc/comments") {
+              stream.on("data", () => {
+                secondStreamDataEvents += 1;
+              });
+            }
+          });
+        },
+      });
+      const rawClient = connectRawClient(clientSide);
+      try {
+        const firstReq = rawClient.request({
+          ":method": "GET",
+          ":path": "/api/agents/me",
+          authorization: `Bearer ${bridgeToken}`,
+        });
+        firstReq.resume();
+        firstReq.end();
+        await new Promise((resolve) => setTimeout(resolve, 20));
+        expect(admissionGate.activeCount).toBe(1);
+        expect(admissionGate.queuedCount).toBe(0);
+
+        const secondReq = rawClient.request({
+          ":method": "POST",
+          ":path": "/api/issues/abc/comments",
+          authorization: `Bearer ${bridgeToken}`,
+        });
+        let status = 0;
+        let responseBody = "";
+        secondReq.setEncoding("utf8");
+        secondReq.on("response", (headers) => {
+          status = Number(headers[":status"]) || 0;
+        });
+        secondReq.on("data", (chunk) => (responseBody += chunk));
+        const secondDone = new Promise<void>((resolve, reject) => {
+          secondReq.on("end", () => resolve());
+          secondReq.on("error", reject);
+        });
+        // Send the full body and end the stream while it still queues
+        // behind the first stream's held slot.
+        secondReq.write(Buffer.alloc(4096, "b"));
+        secondReq.end();
+
+        await new Promise((resolve) => setTimeout(resolve, 30));
+        expect(admissionGate.queuedCount).toBe(1);
+        // The second stream's body reached the transport, but the gate has
+        // not admitted it: the handler must hold none of those bytes yet.
+        expect(secondForwardTracker.called).toBe(false);
+        expect(secondStreamDataEvents).toBe(0);
+
+        releaseFirstStream?.();
+        await secondDone;
+
+        expect(status).toBe(200);
+        expect(JSON.parse(responseBody)).toEqual({ bodyLength: 4096 });
+        expect(secondForwardTracker.called).toBe(true);
+        expect(secondStreamDataEvents).toBeGreaterThan(0);
+      } finally {
+        rawClient.close();
+        await handle.close();
+      }
+    });
+
+    it("a body that arrives during the queue wait completes after admission", async () => {
+      const admissionGate = new Http2BridgeAdmissionGate({ maxParallel: 1, maxSessions: 4 });
+      let releaseFirstStream: (() => void) | undefined;
+      const firstStreamHeld = new Promise<void>((resolve) => {
+        releaseFirstStream = resolve;
+      });
+      const { handle, bridgeToken, clientSide } = bindTestServer({
+        admissionGate,
+        forwardRequest: async (request) => {
+          if (request.pathname === "/api/agents/me") {
+            await firstStreamHeld;
+            return { status: 200 };
+          }
+          return { status: 200, body: JSON.stringify({ bodyLength: request.body.byteLength }) };
+        },
+      });
+      const rawClient = connectRawClient(clientSide);
+      try {
+        const firstReq = rawClient.request({
+          ":method": "GET",
+          ":path": "/api/agents/me",
+          authorization: `Bearer ${bridgeToken}`,
+        });
+        firstReq.resume();
+        firstReq.end();
+        await new Promise((resolve) => setTimeout(resolve, 20));
+        expect(admissionGate.activeCount).toBe(1);
+
+        const secondReq = rawClient.request({
+          ":method": "POST",
+          ":path": "/api/issues/abc/comments",
+          authorization: `Bearer ${bridgeToken}`,
+        });
+        let status = 0;
+        let responseBody = "";
+        secondReq.setEncoding("utf8");
+        secondReq.on("response", (headers) => {
+          status = Number(headers[":status"]) || 0;
+        });
+        secondReq.on("data", (chunk) => (responseBody += chunk));
+        const secondDone = new Promise<void>((resolve, reject) => {
+          secondReq.on("end", () => resolve());
+          secondReq.on("error", reject);
+        });
+
+        // Stream the body over several writes while the stream still
+        // queues, and end it before the gate admits it.
+        const bodyParts = ["first-chunk-", "second-chunk-", "third-chunk"];
+        for (const part of bodyParts) {
+          secondReq.write(part);
+          await new Promise((resolve) => setTimeout(resolve, 5));
+        }
+        secondReq.end();
+
+        await new Promise((resolve) => setTimeout(resolve, 20));
+        expect(admissionGate.queuedCount).toBe(1);
+
+        releaseFirstStream?.();
+        await secondDone;
+
+        const expectedBody = bodyParts.join("");
+        expect(status).toBe(200);
+        expect(JSON.parse(responseBody)).toEqual({ bodyLength: expectedBody.length });
+      } finally {
+        rawClient.close();
+        await handle.close();
+      }
+    });
+
     it("a token, path, or route denial takes no gate slot", async () => {
       const admissionGate = new Http2BridgeAdmissionGate({ maxParallel: 1, maxSessions: 4 });
       const { handle, bridgeToken, clientSide } = bindTestServer({ admissionGate });

packages/adapter-utils/src/http2-bridge-server.ts

@@ -727,11 +727,22 @@ type Http2BridgeBodyCaptureOutcome = { ok: true; body: Buffer } | { ok: false; e
  * peer's `end` (Node does not replay a body the JS layer was not yet
  * listening for), stalling a request that already fully arrived.
  *
- * `awaitBody` arms the idle and lifetime timers and returns the promise the
- * caller awaits. Call it once, when the caller is ready to enforce the
- * bounds (after the admission wait, in this server). If the body already
- * finished, or the stream already errored, closed, or aborted, before
- * `awaitBody` runs, it delivers that outcome at once and arms no timer.
+ * The function pauses the stream right after it attaches the listeners, in
+ * that same synchronous turn. A queued stream must hold no request body
+ * bytes in host memory while it waits for admission: the pause stops the
+ * peer's flow-control window from advancing, so the peer itself stalls at
+ * the wire and the queue applies real back pressure instead of buffering
+ * the body early. `awaitBody` resumes the stream once the caller is ready
+ * to read it. Both halves are needed together: the synchronous listener
+ * attach keeps the `end` event, and the pause keeps the body out of memory
+ * until admission.
+ *
+ * `awaitBody` resumes the stream, then arms the idle and lifetime timers,
+ * and returns the promise the caller awaits. Call it once, when the caller
+ * is ready to enforce the bounds (after the admission wait, in this
+ * server). If the body already finished, or the stream already errored,
+ * closed, or aborted, before `awaitBody` runs, it delivers that outcome at
+ * once, resumes no stream, and arms no timer.
  */
 function beginHttp2BridgeStreamBodyCapture(
   stream: http2.ServerHttp2Stream,
@@ -786,6 +797,11 @@ function beginHttp2BridgeStreamBodyCapture(
   stream.once("close", () =>
     settle({ ok: false, error: new Error("Bridge request stream closed before it completed.") }),
   );
+  // Pause at once, in the same synchronous turn as the listener attach
+  // above: a queued stream then holds no request body bytes in host
+  // memory, because the peer's own flow-control window stalls it at the
+  // wire instead of the JS layer buffering the body early.
+  stream.pause();
 
   return {
     awaitBody(idleTimeoutMs: number, lifetimeCeilingMs: number): Promise<Buffer> {
@@ -803,6 +819,10 @@ function beginHttp2BridgeStreamBodyCapture(
         }
         onSettled = deliver;
         armedIdleTimeoutMs = idleTimeoutMs;
+        // The stream stayed paused for the whole admission wait: resume it
+        // here, now that the caller is ready to enforce the idle and
+        // lifetime bounds below.
+        stream.resume();
         armIdleTimer();
         // Arms one time, from here, and never rearms on a chunk: see
         // DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_LIFETIME_CEILING_MS for why the