fix(adapter-utils): serialize duplex gateway stdout writes and fix reserve release order

The Stage-2 security re-review found two remaining defects in the
in-sandbox duplex gateway.

First, the gateway wrote every frame to process.stdout and ignored the
boolean write() returns. A writable pipe returns false at its
high-water mark, and Node then queues every later write in the
stream's own buffer. A slow host reader let one admitted read retain
its whole body's worth of encoded frames, not the one-frame margin the
retained-peak comment claimed. Fix: route every outbound frame -- the
heartbeat, the READY frame, the request envelope, and each body_chunk
-- through one serialized writer. The writer issues at most one write
at a time and waits for the stream to drain (or to error) before it
starts the next queued line, so a slow reader can no longer make the
gateway queue more than about one frame per active send.

Second, readBody held the chunks array and the Buffer.concat result at
the same time, so the transient peak was 2x the raw body while
admission charged 1x. Fix: readBody now preallocates one buffer sized
to the admission charge and copies each chunk into it in place, then
returns a view over the written bytes. This removes the double
retention; the tradeoff (a small body now retains the full
charge-sized buffer for the life of the request) is documented in the
function comment.

Third, the admission reserve released inside a finally block on the
response promise, so a response timeout could release the reserve
while the outbound send still held the body. Fix: the reserve now
releases only after both the response settled and the send finished
(drained fully or ended by a stream error), tracked independently and
released exactly once through an idempotent release function. The
send now starts after the pending entry and its timer are registered,
so a response timeout no longer stops or races the send.

The retained-peak comment is rewritten to describe what the new code
actually holds: one body buffer, plus at most one in-flight frame
bounded by the frame size limit, times the admitted read count.

Tests: a pipe-backpressure regression test proves the writer never
queues more than about one bounded frame per active send under a
stalled pipe; two isolated writer tests (extracted the same way the
frame codec already is) prove the writer serializes writes and ends a
stalled write on a stream error; an ordering test proves one request's
chunks stay in sequence while a concurrent request interleaves its
own; two reserve tests prove the reserve stays held through a stalled
send and releases only once the send and the response both finish,
for both a resumed pipe and a broken one.

Kept unchanged: the 10 MiB + 1 body cap, the 64 MiB inflight bound,
the host spill behavior, and the observable 413 / 503 / 502 statuses.

Every existing test on the three touched files stays green (189 tests
total). tsc is clean.

Co-authored-by: Paperclip <noreply@paperclip.ing>
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts

@@ -15,6 +15,7 @@ import {
 } from "./sandbox-callback-bridge.js";
 import {
   DuplexFrameDecoder,
+  DEFAULT_MAX_DUPLEX_FRAME_BYTES,
   DUPLEX_BODY_CHUNK_RAW_BYTES,
   DUPLEX_FRAME_VERSION,
   encodeDuplexFrame,
@@ -978,4 +979,252 @@ describe("duplex bridge local end-to-end harness", () => {
     const later = await post();
     expect(later.status).toBe(200);
   }, 30000);
+
+  it("bounds the outbound writer to about one frame per active send while the pipe backpressures", async () => {
+    // Regression test for the backpressure block. The gateway used to call
+    // stdout.write() for every frame in a tight loop and ignore the return
+    // value, so a slow reader let one admitted read queue its whole body's
+    // worth of encoded frames. Pause the pipe read side, so every later write
+    // meets a full pipe, then post several concurrent cap-sized bodies at
+    // once and prove the writer never reports more than about one bounded
+    // frame stuck behind the full pipe.
+    const maxBodyBytes = 4 * 1024 * 1024;
+    const concurrency = 4;
+    const harness = await createHarness({
+      maxBodyBytes,
+      maxInflightBodyBytes: maxBodyBytes * concurrency,
+      responseTimeoutMs: 20000,
+    });
+    harnesses.push(harness);
+    await harness.waitFor(readyPredicate(harness), "the gateway to become ready");
+
+    // Stop reading the child's stdout, so the OS pipe fills and the gateway's
+    // writes meet real backpressure, the same as a slow host reader would
+    // cause in production.
+    harness.child.stdout.pause();
+
+    const body = "a".repeat(maxBodyBytes);
+    const posts = Array.from({ length: concurrency }, (_unused, index) =>
+      fetch(`${harness.baseUrl}/api/issues/issue-${index}/comments`, {
+        method: "POST",
+        headers: {
+          authorization: `Bearer ${harness.bridgeToken}`,
+          "content-type": "application/json",
+        },
+        body,
+      }).catch(() => undefined),
+    );
+
+    // Give the gateway time to read every body and drive its sends into
+    // backpressure.
+    await new Promise((resolve) => setTimeout(resolve, 1500));
+
+    const backpressureLines = harness
+      .stderr()
+      .split("\n")
+      .filter((line) => line.includes("stdout frame write backpressured"));
+    expect(backpressureLines.length).toBeGreaterThan(0);
+    const reportedLengths = backpressureLines.map((line) => {
+      const match = /writableLength=(\d+)/.exec(line);
+      if (!match) throw new Error("backpressure diagnostic is missing writableLength: " + line);
+      return Number(match[1]);
+    });
+    // One buffered frame is at most DEFAULT_MAX_DUPLEX_FRAME_BYTES (1 MB). An
+    // unbounded queue would report a writableLength close to one whole body's
+    // worth of encoded frames -- several megabytes for a 4 MiB body. A
+    // generous multiple of one frame proves the writer never queued anywhere
+    // near that, regardless of how long the pipe stays full.
+    const oneFrameBound = DEFAULT_MAX_DUPLEX_FRAME_BYTES * 2;
+    for (const length of reportedLengths) {
+      expect(length).toBeLessThan(oneFrameBound);
+    }
+
+    // Resume so the harness can read the outstanding frames and tear down
+    // cleanly.
+    harness.child.stdout.resume();
+    await Promise.allSettled(posts);
+  }, 20000);
+
+  it("keeps one request's frames in sequence while a concurrent request interleaves its own", async () => {
+    const harness = await createHarness({
+      maxBodyBytes: 2 * 1024 * 1024,
+      maxInflightBodyBytes: 8 * 1024 * 1024,
+    });
+    harnesses.push(harness);
+    await harness.waitFor(readyPredicate(harness), "the gateway to become ready");
+
+    harness.api.setResponder((_req, res) => {
+      res.writeHead(200, { "content-type": "application/json" });
+      res.end(JSON.stringify({ ok: true }));
+    });
+
+    // Each body is large enough to split into several body_chunk frames, so a
+    // correct sequence is not a coincidence of a one-frame body.
+    const bodyA = "a".repeat(1_500_000);
+    const bodyB = "b".repeat(1_500_000);
+    const post = (body: string) =>
+      fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, {
+        method: "POST",
+        headers: {
+          authorization: `Bearer ${harness.bridgeToken}`,
+          "content-type": "application/json",
+        },
+        body,
+      });
+
+    const [responseA, responseB] = await Promise.all([post(bodyA), post(bodyB)]);
+    expect(responseA.status).toBe(200);
+    expect(responseB.status).toBe(200);
+
+    const requestFrames = harness.observedFrames.filter(
+      (frame): frame is DuplexRequestFrame => frame.type === "request",
+    );
+    expect(requestFrames).toHaveLength(2);
+    for (const requestFrame of requestFrames) {
+      const chunkSeqs = harness.observedFrames
+        .filter(
+          (frame): frame is DuplexBodyChunkFrame =>
+            frame.type === "body_chunk" && frame.id === requestFrame.id,
+        )
+        .map((frame) => frame.seq);
+      expect(chunkSeqs.length).toBeGreaterThan(1);
+      // This request's own chunks arrive in increasing sequence order, even
+      // though the other request's frames may interleave between them.
+      expect(chunkSeqs).toEqual(Array.from({ length: chunkSeqs.length }, (_unused, index) => index));
+    }
+  }, 20000);
+
+  it("holds the admission reserve while a send is stalled behind a full pipe, then releases it once the send drains", async () => {
+    const maxBodyBytes = 2 * 1024 * 1024;
+    const harness = await createHarness({
+      maxBodyBytes,
+      // Admits exactly one concurrent body read.
+      maxInflightBodyBytes: maxBodyBytes,
+      responseTimeoutMs: 15000,
+    });
+    harnesses.push(harness);
+    await harness.waitFor(readyPredicate(harness), "the gateway to become ready");
+    harness.setForwardMode("hang");
+
+    // Stop reading the child's stdout, so the first request's send stalls
+    // mid-flight, well before it ever reaches the host.
+    harness.child.stdout.pause();
+
+    const body = "a".repeat(maxBodyBytes);
+    const first = fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, {
+      method: "POST",
+      headers: {
+        authorization: `Bearer ${harness.bridgeToken}`,
+        "content-type": "application/json",
+      },
+      body,
+    });
+    void first.catch(() => undefined);
+
+    // Give the gateway time to read the body and drive its send into
+    // backpressure.
+    await new Promise((resolve) => setTimeout(resolve, 500));
+
+    // The reserve is still held: a second request cannot reserve, so the
+    // gateway sheds it with a clean 503.
+    const shed = await fetch(`${harness.baseUrl}/api/issues/issue-2/comments`, {
+      method: "POST",
+      headers: {
+        authorization: `Bearer ${harness.bridgeToken}`,
+        "content-type": "application/json",
+      },
+      body: "small",
+    });
+    expect(shed.status).toBe(503);
+    await expect(shed.json()).resolves.toEqual({ error: "bridge_busy" });
+
+    // Resume the pipe and let the host answer. The stalled send drains and
+    // finishes, and the first request's response settles.
+    harness.child.stdout.resume();
+    harness.setForwardMode("proxy");
+    harness.api.setResponder((_req, res) => {
+      res.writeHead(200, { "content-type": "application/json" });
+      res.end(JSON.stringify({ ok: true }));
+    });
+    const firstResponse = await first;
+    expect(firstResponse.status).toBe(200);
+
+    // The reserve released once the send finished. A later request is
+    // admitted again.
+    const later = await fetch(`${harness.baseUrl}/api/issues/issue-3/comments`, {
+      method: "POST",
+      headers: {
+        authorization: `Bearer ${harness.bridgeToken}`,
+        "content-type": "application/json",
+      },
+      body: "small",
+    });
+    expect(later.status).toBe(200);
+  }, 20000);
+
+  it("ends a stalled send when the outbound pipe breaks, and releases the reserve for a later request", async () => {
+    const maxBodyBytes = 2 * 1024 * 1024;
+    const harness = await createHarness({
+      maxBodyBytes,
+      // Admits exactly one concurrent body read.
+      maxInflightBodyBytes: maxBodyBytes,
+      responseTimeoutMs: 1200,
+    });
+    harnesses.push(harness);
+    await harness.waitFor(readyPredicate(harness), "the gateway to become ready");
+
+    // Stop reading the child's stdout, so the first request's send is
+    // genuinely in flight -- stalled behind a full pipe -- when the pipe
+    // breaks.
+    harness.child.stdout.pause();
+
+    const body = "a".repeat(maxBodyBytes);
+    const first = fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, {
+      method: "POST",
+      headers: {
+        authorization: `Bearer ${harness.bridgeToken}`,
+        "content-type": "application/json",
+      },
+      body,
+    });
+    void first.catch(() => undefined);
+
+    // Give the gateway time to read the body and start writing frames into
+    // the full pipe.
+    await new Promise((resolve) => setTimeout(resolve, 400));
+
+    // Confirm the reserve is held before the pipe breaks: a second request
+    // cannot reserve, so the gateway sheds it with a clean 503.
+    const shedBeforeBreak = await fetch(`${harness.baseUrl}/api/issues/issue-2/comments`, {
+      method: "POST",
+      headers: {
+        authorization: `Bearer ${harness.bridgeToken}`,
+        "content-type": "application/json",
+      },
+      body: "small",
+    });
+    expect(shedBeforeBreak.status).toBe(503);
+
+    // Break the pipe. The stalled write meets a stream error, which ends the
+    // wait for drain -- the only other end condition the writer accepts.
+    harness.child.stdout.destroy();
+
+    const firstResponse = await first;
+    // No response frame can ever arrive over a broken pipe. The gateway's own
+    // wait budget ends the request with its local 502 timeout.
+    expect(firstResponse.status).toBe(502);
+
+    // The reserve released once the failed send and the timed-out response
+    // both finished. A later request is admitted again -- it is not shed with
+    // 503, even though it can never complete over the broken pipe either.
+    const later = await fetch(`${harness.baseUrl}/api/issues/issue-3/comments`, {
+      method: "POST",
+      headers: {
+        authorization: `Bearer ${harness.bridgeToken}`,
+        "content-type": "application/json",
+      },
+      body: "small",
+    });
+    expect(later.status).toBe(502);
+  }, 20000);
 });

packages/adapter-utils/src/execution-target-sandbox.test.ts

@@ -1,3 +1,4 @@
+import { EventEmitter } from "node:events";
 import { createServer } from "node:http";
 import net from "node:net";
 import { execFile, spawn } from "node:child_process";
@@ -11,6 +12,7 @@ import { afterEach, describe, expect, it, vi } from "vitest";
 import {
   getSandboxCallbackBridgeServerSource,
   getSandboxDuplexGatewayCodecSource,
+  getSandboxDuplexGatewayStdoutWriterSource,
 } from "./sandbox-callback-bridge.js";
 
 import {
@@ -5746,6 +5748,112 @@ describe("sandbox duplex gateway", () => {
     expect(ledger.liveTokenCount).toBe(0);
   });
 
+  /**
+   * A minimal fake outbound stream for the embedded stdout-writer tests below.
+   * It lets a test control exactly when one write "drains" and when the
+   * stream "errors", with no real pipe and no child process.
+   */
+  class FakeOutboundStream extends EventEmitter {
+    writes: string[] = [];
+    writableLength = 0;
+    private accepting: boolean;
+
+    constructor(options: { accepting?: boolean } = {}) {
+      super();
+      this.accepting = options.accepting ?? true;
+    }
+
+    write(line: string): boolean {
+      this.writes.push(line);
+      if (this.accepting) return true;
+      this.writableLength += Buffer.byteLength(line, "utf8");
+      return false;
+    }
+
+    drain(): void {
+      this.writableLength = 0;
+      this.emit("drain");
+    }
+
+    errorOut(error: Error): void {
+      this.emit("error", error);
+    }
+  }
+
+  interface EmbeddedStdoutWriter {
+    createStdoutFrameWriter: (
+      stream: FakeOutboundStream,
+      onBackpressure?: (writableLength: number) => void,
+    ) => (line: string) => Promise<void>;
+  }
+
+  function loadEmbeddedStdoutFrameWriter(): EmbeddedStdoutWriter {
+    const factory = new Function(
+      `${getSandboxDuplexGatewayStdoutWriterSource()}\nreturn { createStdoutFrameWriter };`,
+    ) as () => EmbeddedStdoutWriter;
+    return factory();
+  }
+
+  it("issues one outbound write at a time and waits for drain before the next", async () => {
+    // Regression test for the backpressure block. The old gateway called
+    // `stream.write()` for every frame in a tight loop and ignored the return
+    // value, so a slow reader let it queue a whole request's frames at once.
+    // The writer must issue at most one write, then wait for the stream to
+    // drain before it starts the next queued line.
+    const { createStdoutFrameWriter } = loadEmbeddedStdoutFrameWriter();
+    const stream = new FakeOutboundStream({ accepting: false });
+    const backpressureLengths: number[] = [];
+    const writeLine = createStdoutFrameWriter(stream, (length) => backpressureLengths.push(length));
+
+    const first = writeLine("frame-1\n");
+    const second = writeLine("frame-2\n");
+    const third = writeLine("frame-3\n");
+
+    // Only the first line reached the stream. The writer holds the rest back
+    // in its own queue; it never hands every queued line to the stream at
+    // once.
+    await Promise.resolve();
+    expect(stream.writes).toEqual(["frame-1\n"]);
+    expect(backpressureLengths).toEqual([Buffer.byteLength("frame-1\n", "utf8")]);
+
+    stream.drain();
+    await first;
+    expect(stream.writes).toEqual(["frame-1\n", "frame-2\n"]);
+
+    stream.drain();
+    await second;
+    expect(stream.writes).toEqual(["frame-1\n", "frame-2\n", "frame-3\n"]);
+
+    stream.drain();
+    await third;
+  });
+
+  it("ends a stalled outbound write and fails every queued line when the stream errors", async () => {
+    // The writer accepts exactly two end conditions for a stalled write: the
+    // stream drains, or the stream errors. This proves the second one: a
+    // stream error ends the wait at once and fails every line still queued,
+    // because a broken pipe can never drain.
+    const { createStdoutFrameWriter } = loadEmbeddedStdoutFrameWriter();
+    const stream = new FakeOutboundStream({ accepting: false });
+    const writeLine = createStdoutFrameWriter(stream);
+
+    const first = writeLine("frame-1\n");
+    const second = writeLine("frame-2\n");
+    await Promise.resolve();
+    expect(stream.writes).toEqual(["frame-1\n"]);
+
+    const boom = new Error("EPIPE");
+    stream.errorOut(boom);
+
+    await expect(first).rejects.toBe(boom);
+    await expect(second).rejects.toBe(boom);
+
+    // The writer never issues another write once the stream has errored.
+    const third = writeLine("frame-3\n");
+    await expect(third).rejects.toBe(boom);
+    expect(stream.writes).toEqual(["frame-1\n"]);
+  });
+
   it("returns the same HTTP response as the file gateway for a forwarded request", async () => {
     const token = "duplex-token-forward";
     const gateway = await startDuplexGateway({ PAPERCLIP_BRIDGE_TOKEN: token });

packages/adapter-utils/src/sandbox-callback-bridge.ts

@@ -2009,6 +2009,79 @@ class DuplexFrameDecoder {
   }
 }`;
 
+// The embedded zero-dependency outbound frame writer. The duplex gateway
+// routes every stdout frame write through the one function this factory
+// returns: the heartbeat, the READY frame, one request envelope, and each
+// body_chunk frame. The writer issues at most one `stream.write()` call at a
+// time and waits for the stream to drain before it starts the next queued
+// line, so a slow reader never lets the gateway queue a whole request's
+// frames at once. A stream "error" event ends an in-flight wait the same way
+// "drain" does, and it fails every line still queued, because a broken pipe
+// can never drain.
+const DUPLEX_GATEWAY_STDOUT_WRITER_SOURCE = `function createStdoutFrameWriter(stream, onBackpressure) {
+  const queue = [];
+  let pumping = false;
+  let streamError = null;
+
+  stream.on("error", (error) => {
+    streamError = error instanceof Error ? error : new Error(String(error));
+    const stranded = queue.splice(0, queue.length);
+    for (const entry of stranded) entry.reject(streamError);
+  });
+
+  async function pump() {
+    if (pumping) return;
+    pumping = true;
+    while (queue.length > 0) {
+      const entry = queue[0];
+      if (streamError) {
+        queue.length = 0;
+        entry.reject(streamError);
+        break;
+      }
+      let wroteOk = false;
+      try {
+        wroteOk = stream.write(entry.line);
+      } catch (error) {
+        queue.shift();
+        entry.reject(error instanceof Error ? error : new Error(String(error)));
+        continue;
+      }
+      if (!wroteOk && !streamError) {
+        if (typeof onBackpressure === "function") {
+          onBackpressure(stream.writableLength);
+        }
+        await new Promise((resolve) => {
+          const onDrain = () => {
+            stream.off("error", onStreamErrorWhileWaiting);
+            resolve();
+          };
+          const onStreamErrorWhileWaiting = () => {
+            stream.off("drain", onDrain);
+            resolve();
+          };
+          stream.once("drain", onDrain);
+          stream.once("error", onStreamErrorWhileWaiting);
+        });
+      }
+      queue.shift();
+      if (streamError) {
+        entry.reject(streamError);
+      } else {
+        entry.resolve();
+      }
+    }
+    pumping = false;
+  }
+
+  return function writeStdoutFrameLine(line) {
+    return new Promise((resolve, reject) => {
+      queue.push({ line: line, resolve: resolve, reject: reject });
+      void pump();
+    });
+  };
+}`;
+
 /**
  * Return the exact zero-dependency codec source the generated duplex gateway
  * embeds. A test runs every fixture vector against this source, so it proves the
@@ -2022,6 +2095,18 @@ export function getSandboxDuplexGatewayCodecSource(): string {
   return DUPLEX_GATEWAY_CODEC_SOURCE;
 }
 
+/**
+ * Return the exact zero-dependency outbound-writer source the generated duplex
+ * gateway embeds. A test wraps this source with a fake stream, so it proves
+ * the writer serializes every write, waits for drain, and ends a stalled write
+ * on a stream error, all without a real child process. The source declares
+ * `createStdoutFrameWriter` but exports it from no module; a caller wraps it
+ * to read the name.
+ */
+export function getSandboxDuplexGatewayStdoutWriterSource(): string {
+  return DUPLEX_GATEWAY_STDOUT_WRITER_SOURCE;
+}
+
 export function getSandboxCallbackBridgeServerSource(): string {
   return `import { randomUUID, timingSafeEqual } from "node:crypto";
 import { createServer } from "node:http";
@@ -2087,6 +2172,10 @@ if (bridgeMode !== "${SANDBOX_CALLBACK_BRIDGE_DUPLEX_MODE}" && !queueDir) {
 // gateway ignores it.
 ${DUPLEX_GATEWAY_CODEC_SOURCE}
 
+// The embedded zero-dependency outbound frame writer. The duplex gateway uses
+// it; the file gateway ignores it.
+${DUPLEX_GATEWAY_STDOUT_WRITER_SOURCE}
+
 function sleep(ms) {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }
@@ -2135,22 +2224,40 @@ function reserveBodyRead() {
   };
 }
 
+// Read one request body into a single buffer sized to the admission charge.
+//
+// Accounting choice: the old code pushed each chunk onto a chunks array, then
+// called Buffer.concat on it. Both the array and the concat result held the
+// full body at once, so the transient peak was 2x the raw body while
+// admission charged 1x -- a dishonest bound. This version preallocates one
+// buffer of maxBodyBytes and copies each incoming chunk into it in place, then
+// returns a view over the written bytes. A view shares the same backing
+// memory; it adds no second copy. So the retained peak per active read is one
+// maxBodyBytes buffer, matching the charge exactly.
+//
+// The tradeoff: a small body now retains the full maxBodyBytes allocation for
+// the life of the request, where the old code retained only the small body's
+// own bytes. reserveBodyRead already charges maxBodyBytes for every active
+// read regardless of the real size, so this tradeoff raises no read past what
+// admission already assumed was possible; it only makes the real memory shape
+// match the existing worst-case charge instead of usually running under it.
 async function readBody(req) {
-  const chunks = [];
+  const out = Buffer.allocUnsafe(maxBodyBytes);
   let totalBytes = 0;
   for await (const chunk of req) {
     const nextChunk = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
-    chunks.push(nextChunk);
-    totalBytes += nextChunk.byteLength;
-    if (totalBytes > maxBodyBytes) {
+    if (totalBytes + nextChunk.byteLength > maxBodyBytes) {
       throw new BridgeBodyTooLargeError();
     }
+    nextChunk.copy(out, totalBytes);
+    totalBytes += nextChunk.byteLength;
   }
-  // Return the raw request bytes. The duplex gateway base64-encodes these bytes
-  // unchanged, so a malformed body cannot expand during transport and cannot
-  // exceed the bytes that admission charged. The file gateway decodes a valid
-  // UTF-8 body with decodeUtf8Body and rejects an invalid one.
-  return Buffer.concat(chunks);
+  // Return the raw request bytes as a view, not a copy. The duplex gateway
+  // base64-encodes these bytes unchanged, so a malformed body cannot expand
+  // during transport and cannot exceed the bytes that admission charged. The
+  // file gateway decodes a valid UTF-8 body with decodeUtf8Body and rejects an
+  // invalid one.
+  return out.subarray(0, totalBytes);
 }
 
 // Strict UTF-8 decoder for the file gateway. It throws on any invalid byte, so
@@ -2368,8 +2475,27 @@ function runDuplexGateway() {
     process.stderr.write("[paperclip-bridge] " + message + "\\n");
   }
 
+  // The one serialized, backpressure-aware writer for this process. Every
+  // outbound frame -- the heartbeat, the READY frame, one request envelope,
+  // and each body_chunk -- writes through it, in call order.
+  const writeStdoutFrameLine = createStdoutFrameWriter(process.stdout, (writableLength) => {
+    diag("stdout frame write backpressured: writableLength=" + writableLength);
+  });
+
+  // Write a control frame (heartbeat or READY). This is fire-and-forget: a
+  // dropped heartbeat is not fatal, because the heartbeat repeats on its own
+  // interval, and a lost READY frame surfaces as a broker wait timeout. A
+  // request send awaits its own writes instead -- see the request handler
+  // below -- because it must run to completion once it starts.
   function writeFrame(frame) {
-    process.stdout.write(encodeDuplexFrame(frame));
+    writeStdoutFrameLine(encodeDuplexFrame(frame)).catch((error) => {
+      diag(
+        "failed to write a " +
+          frame.type +
+          " frame: " +
+          (error && error.message ? error.message : String(error)),
+      );
+    });
   }
 
   function stopTimers() {
@@ -2558,6 +2684,11 @@ function runDuplexGateway() {
         writeJsonResponse(res, 503, { error: "bridge_busy" });
         return;
       }
+      // Once the send below starts, releaseReserveWhenBothDone owns the
+      // reserve release. This flag stays false only while a return above the
+      // send can still happen, so the outer finally releases directly for
+      // every path that never commits to a send.
+      let handedOffToSend = false;
       try {
       const requestBodyBuffer = await readBody(req);
       if (unavailable) {
@@ -2598,37 +2729,100 @@ function runDuplexGateway() {
         writeJsonResponse(res, 413, { error: "request_too_large" });
         return;
       }
-      // Retained-peak note. The gateway preserves the raw request bytes end to
-      // end and streams the body as body_chunk frames, one frame at a time. It
-      // does not hold the raw body plus the full base64 frame array plus the
-      // full encoded-line array (about 3.7x the raw body). The retained peak per
-      // read is now the raw body plus one encoded frame:
-      //   raw body       <= maxBodyBytes (about 10 MiB)
-      //   one body_chunk <= DEFAULT_MAX_DUPLEX_FRAME_BYTES (1 MB)
-      // so the peak is about maxBodyBytes + 1 MB per read. Admission charges
-      // maxBodyBytes per read against the 64 MiB inflight bound. The bound admits
-      // at most floor(64 MiB / maxBodyBytes) = 6 reads, so the true retained peak
-      // stays near 6 * (10 MiB + 1 MB) ~= 66 MiB. That is a fixed small margin
-      // over the admission threshold, not a multiple of the body. A malformed
-      // body cannot expand, because its raw bytes ride base64 frames unchanged.
+      // Retained-peak note. readBody preallocates one buffer sized to
+      // maxBodyBytes and returns a view over it, so no second (concatenated)
+      // copy of the body ever coexists with it -- see the comment on readBody.
+      // The send below builds and encodes one body_chunk frame at a time from
+      // that same view; it holds no per-request buffer of its own. Every
+      // frame -- this envelope, each body_chunk, and every heartbeat and READY
+      // frame from the rest of the process -- writes through the one
+      // serialized writer above. That writer issues at most one
+      // written-but-not-yet-drained line at a time, bounded by
+      // DEFAULT_MAX_DUPLEX_FRAME_BYTES (1 MB), plus whatever the stream itself
+      // still buffers up to its own high-water mark; it never queues a whole
+      // request's frames at once behind a slow reader. So the retained peak
+      // per active read is about:
+      //   one body buffer     <= maxBodyBytes (about 10 MiB)
+      //   one in-flight frame <= DEFAULT_MAX_DUPLEX_FRAME_BYTES (1 MB)
+      // and admission charges maxBodyBytes per read against the 64 MiB
+      // inflight bound, so the bound admits at most
+      // floor(64 MiB / maxBodyBytes) = 6 reads. The true retained peak across
+      // every admitted read stays near 6 * (10 MiB + 1 MB) ~= 66 MiB, the same
+      // fixed small margin over the admission threshold as before -- not a
+      // multiple of the body, and not made worse by a slow host. A malformed
+      // body cannot expand, because its raw bytes ride base64 frames
+      // unchanged. The reserve for one read releases only once both its send
+      // finished (drained fully or ended by a stream error) and its response
+      // settled -- see releaseReserveWhenBothDone below -- so a slow host
+      // cannot make the gateway release a reserve while it still retains the
+      // bytes.
+      handedOffToSend = true;
+      let sendDone = false;
+      let responseDone = false;
+      function releaseReserveWhenBothDone() {
+        if (sendDone && responseDone) {
+          // releaseBodyRead is itself idempotent, so a second call here (for
+          // example if both flags flip true in the same tick) is harmless.
+          releaseBodyRead();
+        }
+      }
+
       const response = await new Promise((resolve) => {
         const timer = setTimeout(() => {
           responseAssembly.delete(requestId);
           if (pending.delete(requestId)) {
-            resolve({
+            settleResponse({
               status: 502,
               headers: { "content-type": "application/json" },
               body: JSON.stringify({ error: "Timed out waiting for host bridge response." }),
             });
           }
         }, responseTimeoutMs);
-        pending.set(requestId, { resolve: resolve, timer: timer });
-        process.stdout.write(encodedRequest.line);
-        for (const frame of duplexBodyChunkFrames(requestId, requestBodyBuffer)) {
-          const encoded = encodeDuplexFrameChecked(frame);
-          // Pass 1 proved every frame fits, so encoded.ok is always true here.
-          process.stdout.write(encoded.line);
+
+        // Every path that settles this request's response -- a real response,
+        // this timeout, a local 502 from failRequest, or the 409 triggerLoss
+        // sends on channel loss -- resolves through entry.resolve, which is
+        // this function. So responseDone flips true exactly once, on
+        // whichever of those settles first.
+        function settleResponse(value) {
+          clearTimeout(timer);
+          responseDone = true;
+          releaseReserveWhenBothDone();
+          resolve(value);
         }
+        pending.set(requestId, { resolve: settleResponse, timer: timer });
+
+        // Run the serialized send after the pending entry and the timer are
+        // already registered, so the response side and the send side start
+        // from the same point and settle independently. There is no cancel
+        // frame in the protocol: once this send writes the envelope, it must
+        // run to completion or the stdout stream must error, because a
+        // stopped send would leave a partial body only channel teardown
+        // reclaims. A response timeout above does not stop this send; it only
+        // resolves the HTTP caller early while the send keeps running.
+        void (async () => {
+          try {
+            await writeStdoutFrameLine(encodedRequest.line);
+            for (const frame of duplexBodyChunkFrames(requestId, requestBodyBuffer)) {
+              const encoded = encodeDuplexFrameChecked(frame);
+              // Pass 1 proved every frame fits, so encoded.ok is always true here.
+              await writeStdoutFrameLine(encoded.line);
+            }
+          } catch (error) {
+            // The writer's only two end conditions for a stalled write are a
+            // drain and a stream error. This catches the stream-error case: it
+            // ends the send here instead of leaving it stalled forever.
+            diag(
+              "the outbound send for request " +
+                requestId +
+                " ended: " +
+                (error && error.message ? error.message : String(error)),
+            );
+          } finally {
+            sendDone = true;
+            releaseReserveWhenBothDone();
+          }
+        })();
       });
       res.statusCode = typeof response.status === "number" ? response.status : 200;
       for (const [key, value] of Object.entries(response.headers || {})) {
@@ -2637,7 +2831,9 @@ function runDuplexGateway() {
       }
       res.end(typeof response.body === "string" ? response.body : "");
       } finally {
-        releaseBodyRead();
+        if (!handedOffToSend) {
+          releaseBodyRead();
+        }
       }
     } catch (error) {
       if (error && error.code === "request_too_large") {