fix(adapter-utils): charge duplex bridge body admission from the declared length

Round 4 of the bridge body admission fix. reserveBodyRead and readBody
took a flat maxBodyBytes charge for every active read. That charge was
too large for a small control-plane call and too small for the file
gateway, whose real retained peak reaches several times the raw body
size once it decodes, re-serializes, and re-encodes the body.

Read the Content-Length header before the gateway reserves anything.
Reject a body that declares more than the cap with a clean 413, before
the gateway reserves or reads one byte. When the header is absent (a
chunked request), fall back to the cap as the worst case, the same as
before.

Charge the reserve and the read buffer from the declared length times a
named per-gateway factor:
- The duplex gateway adds a fixed frame allowance to the declared
  length, because it holds one body buffer plus at most one in-flight
  frame that does not grow with the body.
- The file gateway multiplies the declared length by a factor of 15,
  the sum of the four representations it holds: the raw buffer, the
  decoded string, the JSON-escaped string, and the write buffer. The
  comment on the constant names each one and its worst-case size.

Fix the outbound stdout writer so a stream error settles every queued
write, not only the one at the head of the queue. The old top-of-loop
check discarded queued entries with no reject call, and the resume path
after a stalled write could remove and orphan a line a caller queued
after the stream had already failed. Both left a promise that never
settled, which stalled the reserve release and shed every later request
until the process restarted.

Add unit tests for the writer fix and end-to-end tests on both gateways
for the declared-length admission model: the pre-reserve 413, the
worst-case fallback for a request with no declared length, many small
concurrent requests the old flat charge would have shed, and the
admission-bound math at the new per-gateway reserve.

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

@@ -781,13 +781,20 @@ describe("duplex bridge local end-to-end harness", () => {
     expect(receivedBodyLength).toBe(apiLimitBytes);
   }, 30000);
 
-  it("answers 413 request_too_large, not 502, when the body passes the bridge cap", async () => {
+  it("answers 413 from the declared length alone, before it reserves or reads any body byte", async () => {
     // Regression test: the body-size overflow path threw a generic Error that
     // unwound to the 502 catch. A 502 is retryable and leaked the internal
     // message. The overflow must answer a clean 413 request_too_large. Assert the
     // status code specifically, because the wrong 502 stood for a long time.
     const maxBodyBytes = 4096;
-    const harness = await createHarness({ maxBodyBytes });
+    const laterBodySize = "small".length;
+    // Just enough admission budget for the later small read below (its
+    // declared size plus the fixed in-flight frame allowance), with no
+    // margin. If the gateway reserved the oversize request's declared length
+    // before it rejected the request, this budget would stay exhausted and
+    // the later request would be shed with a 503 instead of admitted.
+    const maxInflightBodyBytes = laterBodySize + DEFAULT_MAX_DUPLEX_FRAME_BYTES;
+    const harness = await createHarness({ maxBodyBytes, maxInflightBodyBytes });
     harnesses.push(harness);
     await harness.waitFor(readyPredicate(harness), "the gateway to become ready");
 
@@ -812,18 +819,126 @@ describe("duplex bridge local end-to-end harness", () => {
     // The oversize body never reached the API, and the response leaked no
     // internal message.
     expect(apiWasCalled).toBe(false);
+
+    // The rejected request reserved nothing, so a small request right after
+    // is still admitted -- proving the gateway answered the 413 from the
+    // declared length header alone, before it ever reserved or read a byte.
+    const later = await fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, {
+      method: "POST",
+      headers: {
+        authorization: `Bearer ${harness.bridgeToken}`,
+        "content-type": "application/json",
+      },
+      body: "small",
+    });
+    expect(later.status).toBe(200);
+    expect(apiWasCalled).toBe(true);
+  }, 20000);
+
+  it("reserves the worst case and answers 413 for a chunked request with no declared length that overruns the cap", async () => {
+    // A request with no Content-Length header -- for example a chunked
+    // request -- gives the gateway no declared size ahead of time. The
+    // gateway must fall back to the worst case (maxBodyBytes) for its
+    // reserve, and it must still answer a clean 413 once the body it
+    // actually reads exceeds the cap.
+    const maxBodyBytes = 4096;
+    const harness = await createHarness({ maxBodyBytes });
+    harnesses.push(harness);
+    await harness.waitFor(readyPredicate(harness), "the gateway to become ready");
+
+    let apiWasCalled = false;
+    harness.api.setResponder((_req, res) => {
+      apiWasCalled = true;
+      res.writeHead(200, { "content-type": "application/json" });
+      res.end(JSON.stringify({ ok: true }));
+    });
+
+    const oversizeBody = "a".repeat(maxBodyBytes + 1024);
+    // A ReadableStream body sends with chunked transfer-encoding and no
+    // Content-Length header, so the gateway cannot know the real size ahead
+    // of time.
+    const stream = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode(oversizeBody));
+        controller.close();
+      },
+    });
+    const response = await fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, {
+      method: "POST",
+      headers: {
+        authorization: `Bearer ${harness.bridgeToken}`,
+        "content-type": "application/json",
+      },
+      body: stream,
+      duplex: "half",
+    } as RequestInit & { duplex: "half" });
+    expect(response.status).toBe(413);
+    await expect(response.json()).resolves.toEqual({ error: "request_too_large" });
+    expect(apiWasCalled).toBe(false);
+
+    // The rejected read released its reserve, so a later request is admitted
+    // again and reaches the API.
+    const later = await fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, {
+      method: "POST",
+      headers: {
+        authorization: `Bearer ${harness.bridgeToken}`,
+        "content-type": "application/json",
+      },
+      body: "small",
+    });
+    expect(later.status).toBe(200);
+  }, 20000);
+
+  it("admits many small concurrent requests that the old flat 10 MiB charge would have shed with 503", async () => {
+    // Production constants. Under the old flat charge every active read
+    // reserved maxBodyBytes (about 10 MiB) regardless of the real body size,
+    // so the 64 MiB inflight bound admitted at most
+    // floor(64 MiB / (10 MiB + 1)) = 6 concurrent reads. Charging the
+    // declared body size instead means a control-plane-sized body reserves
+    // only its own few hundred bytes plus the fixed frame allowance, so the
+    // same bound admits far more than 6 concurrent small reads at once.
+    const maxBodyBytes = 10 * 1024 * 1024 + 1;
+    const maxInflightBodyBytes = 64 * 1024 * 1024;
+    const concurrency = 20;
+    const harness = await createHarness({ maxBodyBytes, maxInflightBodyBytes });
+    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 }));
+    });
+
+    const body = JSON.stringify({ body: "a small status comment" });
+    const post = () =>
+      fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, {
+        method: "POST",
+        headers: {
+          authorization: `Bearer ${harness.bridgeToken}`,
+          "content-type": "application/json",
+        },
+        body,
+      });
+    const responses = await Promise.all(Array.from({ length: concurrency }, () => post()));
+    expect(responses.map((response) => response.status)).toEqual(
+      Array.from({ length: concurrency }, () => 200),
+    );
   }, 20000);
 
   it("sheds a concurrent body read past the admission bound, then releases the reserve", async () => {
-    // The gateway reserves one worst-case body per active read against the local
-    // admission bound. With the bound set to one body, the first read is admitted
-    // and the second read is shed with a 503. The gateway releases the reserve on
-    // the response timeout, so a later request is admitted again.
+    // The gateway reserves the declared body size plus the fixed in-flight
+    // frame allowance for each active read against the local admission
+    // bound. With the bound set to exactly one such reserve, the first read
+    // is admitted and the second read is shed with a 503. The gateway
+    // releases the reserve on the response timeout, so a later request is
+    // admitted again.
     const maxBodyBytes = 4096;
+    const bodySize = 1024;
+    const reservePerRead = bodySize + DEFAULT_MAX_DUPLEX_FRAME_BYTES;
     const harness = await createHarness({
       maxBodyBytes,
-      // The bound admits exactly one concurrent body read.
-      maxInflightBodyBytes: maxBodyBytes,
+      // The bound admits exactly one concurrent body read at bodySize.
+      maxInflightBodyBytes: reservePerRead,
       // A short response deadline releases the first (hung) read fast.
       responseTimeoutMs: 700,
     });
@@ -833,7 +948,7 @@ describe("duplex bridge local end-to-end harness", () => {
     // The first admitted request hangs on the forward until it times out. The
     // second request cannot reserve, so the gateway sheds it with a 503.
     harness.setForwardMode("hang");
-    const body = "a".repeat(1024);
+    const body = "a".repeat(bodySize);
     const post = () =>
       fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, {
         method: "POST",
@@ -923,16 +1038,21 @@ describe("duplex bridge local end-to-end harness", () => {
   }, 30000);
 
   it("caps concurrent body reads at the retained-byte model, then releases the reserve", async () => {
-    // Production-size accounting. Admission charges maxBodyBytes per read against
-    // the inflight bound, so the bound admits floor(bound / maxBodyBytes) reads.
-    // With the production constants that is floor(64 MiB / (10 MiB + 1)) = 6. The
-    // reservation is the worst-case body per read, so a small body still charges
-    // the full maxBodyBytes. So the test drives the model with small bodies and
-    // asserts the admitted count matches the model.
+    // Production-size accounting. The gateway charges each active read the
+    // declared body size plus the fixed in-flight frame allowance, not the
+    // flat body cap, so the bound admits
+    // floor(inflightBound / (declaredBytes + frameAllowance)) reads. The
+    // declared body here is well under the cap, so this model admits far
+    // more than the old flat-charge model's 6 -- but the test still keeps
+    // the count under the gateway's separate maxQueueDepth bound (64), so
+    // that unrelated queue-depth gate does not also shed a request and
+    // confound the assertions below.
     const maxBodyBytes = 10 * 1024 * 1024 + 1;
     const maxInflightBodyBytes = 64 * 1024 * 1024;
-    const admitCount = Math.floor(maxInflightBodyBytes / maxBodyBytes);
-    expect(admitCount).toBe(6);
+    const bodySize = 1024 * 1024;
+    const reservePerRead = bodySize + DEFAULT_MAX_DUPLEX_FRAME_BYTES;
+    const admitCount = Math.floor(maxInflightBodyBytes / reservePerRead);
+    expect(admitCount).toBe(32);
     const harness = await createHarness({
       maxBodyBytes,
       maxInflightBodyBytes,
@@ -945,7 +1065,7 @@ describe("duplex bridge local end-to-end harness", () => {
     // Each admitted read hangs on the forward until it times out, so all admitted
     // reads hold their reserve at once and the next read cannot reserve.
     harness.setForwardMode("hang");
-    const body = "a".repeat(1024);
+    const body = "a".repeat(bodySize);
     const post = () =>
       fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, {
         method: "POST",
@@ -992,7 +1112,9 @@ describe("duplex bridge local end-to-end harness", () => {
     const concurrency = 4;
     const harness = await createHarness({
       maxBodyBytes,
-      maxInflightBodyBytes: maxBodyBytes * concurrency,
+      // Each concurrent read charges its declared body size plus the fixed
+      // in-flight frame allowance, so the bound must admit all four at once.
+      maxInflightBodyBytes: (maxBodyBytes + DEFAULT_MAX_DUPLEX_FRAME_BYTES) * concurrency,
       responseTimeoutMs: 20000,
     });
     harnesses.push(harness);
@@ -1098,8 +1220,9 @@ describe("duplex bridge local end-to-end harness", () => {
     const maxBodyBytes = 2 * 1024 * 1024;
     const harness = await createHarness({
       maxBodyBytes,
-      // Admits exactly one concurrent body read.
-      maxInflightBodyBytes: maxBodyBytes,
+      // Admits exactly one concurrent body read at maxBodyBytes: the declared
+      // body size plus the fixed in-flight frame allowance.
+      maxInflightBodyBytes: maxBodyBytes + DEFAULT_MAX_DUPLEX_FRAME_BYTES,
       responseTimeoutMs: 15000,
     });
     harnesses.push(harness);
@@ -1166,8 +1289,9 @@ describe("duplex bridge local end-to-end harness", () => {
     const maxBodyBytes = 2 * 1024 * 1024;
     const harness = await createHarness({
       maxBodyBytes,
-      // Admits exactly one concurrent body read.
-      maxInflightBodyBytes: maxBodyBytes,
+      // Admits exactly one concurrent body read at maxBodyBytes: the declared
+      // body size plus the fixed in-flight frame allowance.
+      maxInflightBodyBytes: maxBodyBytes + DEFAULT_MAX_DUPLEX_FRAME_BYTES,
       responseTimeoutMs: 1200,
     });
     harnesses.push(harness);

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

@@ -5854,6 +5854,37 @@ describe("sandbox duplex gateway", () => {
     expect(stream.writes).toEqual(["frame-1\n"]);
   });
 
+  it("settles every write queued in the same tick as a stream error, not only the queue head", async () => {
+    // Regression test. The writer's top-of-loop error check rejected only the
+    // queue head and discarded the rest of the queue with no reject call, so
+    // a second write queued in the same synchronous tick as the error never
+    // settled. Separately, the code path that resumed after a stalled write
+    // shifted the queue head unconditionally, so a write queued during that
+    // stall -- after the stream had already errored, but before the writer's
+    // pending wait had a chance to resume -- was removed and never settled
+    // either. Queue two more writes in the same tick as the error, so both
+    // failure modes are on the table, and prove every one of the three
+    // pending promises rejects.
+    const { createStdoutFrameWriter } = loadEmbeddedStdoutFrameWriter();
+    const stream = new FakeOutboundStream({ accepting: false });
+    const writeLine = createStdoutFrameWriter(stream);
+
+    const first = writeLine("frame-1\n");
+    await Promise.resolve();
+    expect(stream.writes).toEqual(["frame-1\n"]);
+
+    const boom = new Error("EPIPE");
+    // Queue two more lines in the same synchronous tick as the error, before
+    // the writer's drain-or-error wait for frame-1 resumes.
+    stream.errorOut(boom);
+    const second = writeLine("frame-2\n");
+    const third = writeLine("frame-3\n");
+
+    await expect(first).rejects.toBe(boom);
+    await expect(second).rejects.toBe(boom);
+    await expect(third).rejects.toBe(boom);
+  });
+
   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.test.ts

@@ -885,6 +885,166 @@ describe("sandbox callback bridge", () => {
     });
   });
 
+  it("reserves the declared body size, not the flat body cap, against the file gateway's admission bound", async () => {
+    const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-bridge-body-reserve-"));
+    cleanupDirs.push(rootDir);
+
+    const localWorkspaceDir = path.join(rootDir, "local-workspace");
+    const remoteWorkspaceDir = path.join(rootDir, "remote-workspace");
+    await mkdir(localWorkspaceDir, { recursive: true });
+    await mkdir(remoteWorkspaceDir, { recursive: true });
+    await writeFile(path.join(localWorkspaceDir, "README.md"), "bridge body reserve test\n", "utf8");
+
+    const runner = createExecRunner();
+    const bridgeAsset = await createSandboxCallbackBridgeAsset();
+    cleanupFns.push(bridgeAsset.cleanup);
+    const prepared = await prepareCommandManagedRuntime({
+      runner,
+      spec: {
+        remoteCwd: remoteWorkspaceDir,
+        timeoutMs: 30_000,
+      },
+      adapterKey: "codex",
+      workspaceLocalDir: localWorkspaceDir,
+      assets: [{ key: "bridge", localDir: bridgeAsset.localDir }],
+    });
+
+    const queueDir = path.posix.join(prepared.runtimeRootDir, "paperclip-bridge");
+    const bridgeToken = createSandboxCallbackBridgeToken();
+
+    const maxBodyBytes = 4096;
+    const bodySize = 1024;
+    // The file gateway charges 15x the declared body size for its reserve:
+    // 1x for the raw buffer, up to 2x for the decoded string, and up to 6x
+    // each for the JSON-escaped string and its UTF-8 write buffer. See the
+    // comment on FILE_GATEWAY_BODY_RESERVE_FACTOR in sandbox-callback-bridge.ts.
+    const reservePerRead = bodySize * 15;
+    const bridge = await startSandboxCallbackBridgeServer({
+      runner,
+      remoteCwd: remoteWorkspaceDir,
+      assetRemoteDir: prepared.assetDirs.bridge,
+      queueDir,
+      bridgeToken,
+      timeoutMs: 30_000,
+      pollIntervalMs: 10,
+      // No worker answers in this test, so an admitted read hangs until this
+      // deadline and times out with a 502. A shed read never reaches that
+      // far; it answers a 503 at once.
+      responseTimeoutMs: 300,
+      maxBodyBytes,
+      // Admits exactly one concurrent body read at bodySize.
+      maxInflightBodyBytes: reservePerRead,
+    });
+    cleanupFns.push(async () => {
+      await bridge.stop();
+    });
+
+    const body = "a".repeat(bodySize);
+    const post = () =>
+      fetch(`${bridge.baseUrl}/api/issues/issue-1/comments`, {
+        method: "POST",
+        headers: {
+          authorization: `Bearer ${bridgeToken}`,
+          "content-type": "application/json",
+        },
+        body,
+      });
+    const [first, second] = await Promise.all([post(), post()]);
+    const statuses = [first.status, second.status].sort((a, b) => a - b);
+    // One request was admitted (it later times out with a 502, because no
+    // worker answers). One request was shed with a 503 before it could
+    // buffer a second body.
+    expect(statuses).toEqual([502, 503]);
+    const shed = first.status === 503 ? first : second;
+    await expect(shed.json()).resolves.toEqual({ error: "bridge_busy" });
+
+    // The timed-out read released its reserve, so a later request is
+    // admitted again -- it also times out with a 502, not a 503.
+    const later = await post();
+    expect(later.status).toBe(502);
+  }, 20000);
+
+  it("admits many small concurrent file-gateway requests that the old flat 10 MiB charge would have shed with 503", async () => {
+    // Production constants. Under the old flat charge every active read
+    // reserved maxBodyBytes (about 10 MiB) regardless of the real body size,
+    // so the 64 MiB inflight bound admitted at most
+    // floor(64 MiB / (10 MiB + 1)) = 6 concurrent reads. Charging the
+    // declared body size instead (times the file gateway's factor) means a
+    // control-plane-sized body reserves only a small multiple of its own
+    // size, so the same bound admits far more than 6 concurrent small reads
+    // at once.
+    const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-bridge-small-concurrent-"));
+    cleanupDirs.push(rootDir);
+
+    const localWorkspaceDir = path.join(rootDir, "local-workspace");
+    const remoteWorkspaceDir = path.join(rootDir, "remote-workspace");
+    await mkdir(localWorkspaceDir, { recursive: true });
+    await mkdir(remoteWorkspaceDir, { recursive: true });
+    await writeFile(path.join(localWorkspaceDir, "README.md"), "bridge small concurrent test\n", "utf8");
+
+    const runner = createExecRunner();
+    const bridgeAsset = await createSandboxCallbackBridgeAsset();
+    cleanupFns.push(bridgeAsset.cleanup);
+    const prepared = await prepareCommandManagedRuntime({
+      runner,
+      spec: {
+        remoteCwd: remoteWorkspaceDir,
+        timeoutMs: 30_000,
+      },
+      adapterKey: "codex",
+      workspaceLocalDir: localWorkspaceDir,
+      assets: [{ key: "bridge", localDir: bridgeAsset.localDir }],
+    });
+
+    const queueDir = path.posix.join(prepared.runtimeRootDir, "paperclip-bridge");
+    const bridgeToken = createSandboxCallbackBridgeToken();
+
+    const worker = await startSandboxCallbackBridgeWorker({
+      client: createFileSystemSandboxCallbackBridgeQueueClient(),
+      queueDir,
+      authorizeRequest: async () => null,
+      handleRequest: async (request) => ({
+        status: 200,
+        headers: { "content-type": "application/json" },
+        body: JSON.stringify({ ok: true, id: request.id }),
+      }),
+    });
+    cleanupFns.push(async () => {
+      await worker.stop();
+    });
+
+    const bridge = await startSandboxCallbackBridgeServer({
+      runner,
+      remoteCwd: remoteWorkspaceDir,
+      assetRemoteDir: prepared.assetDirs.bridge,
+      queueDir,
+      bridgeToken,
+      timeoutMs: 30_000,
+      pollIntervalMs: 10,
+      maxBodyBytes: 10 * 1024 * 1024 + 1,
+      // maxInflightBodyBytes stays at the production default (64 MiB).
+    });
+    cleanupFns.push(async () => {
+      await bridge.stop();
+    });
+
+    const concurrency = 20;
+    const body = JSON.stringify({ body: "a small status comment" });
+    const post = () =>
+      fetch(`${bridge.baseUrl}/api/issues/issue-1/comments`, {
+        method: "POST",
+        headers: {
+          authorization: `Bearer ${bridgeToken}`,
+          "content-type": "application/json",
+        },
+        body,
+      });
+    const responses = await Promise.all(Array.from({ length: concurrency }, () => post()));
+    expect(responses.map((response) => response.status)).toEqual(
+      Array.from({ length: concurrency }, () => 200),
+    );
+  }, 20000);
+
   it("returns a 502 when the host response times out", async () => {
     const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-bridge-timeout-"));
     cleanupDirs.push(rootDir);

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

@@ -44,11 +44,12 @@ const DEFAULT_BRIDGE_MAX_DUPLEX_DECODER_BYTES = 8 * 1024 * 1024;
 // operating-system process inside the sandbox, so it cannot share the host
 // aggregate byte ledger. At a 10 MiB body cap, the 64-deep pending queue could
 // let many concurrent reads each retain a full body and exhaust the sandbox
-// process memory before any host limit applies. The gateway reserves one
-// worst-case body per active read against this bound, and sheds a new read with
-// a 503 when the bound is full. The host passes the value through
-// PAPERCLIP_BRIDGE_MAX_INFLIGHT_BODY_BYTES. This bound protects the sandbox
-// process; it is separate from the host aggregate byte ledger.
+// process memory before any host limit applies. The gateway reserves bytes for
+// each active read against this bound -- the declared body size times a
+// per-gateway representation factor, not a flat per-request charge -- and sheds
+// a new read with a 503 when the bound is full. The host passes the value
+// through PAPERCLIP_BRIDGE_MAX_INFLIGHT_BODY_BYTES. This bound protects the
+// sandbox process; it is separate from the host aggregate byte ledger.
 const DEFAULT_BRIDGE_MAX_INFLIGHT_BODY_BYTES = 64 * 1024 * 1024;
 // Per-iteration timeout for one poll-loop client call. A healthy control-plane
 // round trip finishes in well under one second, so 10s is far above a normal
@@ -476,6 +477,7 @@ export function buildSandboxCallbackBridgeEnv(input: {
   responseTimeoutMs?: number | null;
   maxQueueDepth?: number | null;
   maxBodyBytes?: number | null;
+  maxInflightBodyBytes?: number | null;
 }): Record<string, string> {
   return {
     PAPERCLIP_API_BRIDGE_MODE: SANDBOX_CALLBACK_BRIDGE_FILE_MODE,
@@ -495,6 +497,9 @@ export function buildSandboxCallbackBridgeEnv(input: {
     PAPERCLIP_BRIDGE_MAX_BODY_BYTES: String(
       normalizeTimeoutMs(input.maxBodyBytes, DEFAULT_BRIDGE_MAX_BODY_BYTES),
     ),
+    PAPERCLIP_BRIDGE_MAX_INFLIGHT_BODY_BYTES: String(
+      normalizeTimeoutMs(input.maxInflightBodyBytes, DEFAULT_BRIDGE_MAX_INFLIGHT_BODY_BYTES),
+    ),
   };
 }
 
@@ -1643,6 +1648,7 @@ export async function startSandboxCallbackBridgeServer(input: {
   shellCommand?: "bash" | "sh" | null;
   maxQueueDepth?: number | null;
   maxBodyBytes?: number | null;
+  maxInflightBodyBytes?: number | null;
 }): Promise<StartedSandboxCallbackBridgeServer> {
   const timeoutMs = normalizeTimeoutMs(input.timeoutMs, DEFAULT_BRIDGE_RESPONSE_TIMEOUT_MS);
   const shellCommand = preferredShellForSandbox(input.shellCommand);
@@ -1668,6 +1674,7 @@ export async function startSandboxCallbackBridgeServer(input: {
     responseTimeoutMs: input.responseTimeoutMs,
     maxQueueDepth: input.maxQueueDepth,
     maxBodyBytes: input.maxBodyBytes,
+    maxInflightBodyBytes: input.maxInflightBodyBytes,
   });
   const nodeCommand = input.nodeCommand?.trim() || "node";
   const startResult = await input.runner.execute({
@@ -2023,22 +2030,29 @@ const DUPLEX_GATEWAY_STDOUT_WRITER_SOURCE = `function createStdoutFrameWriter(st
   let pumping = false;
   let streamError = null;
 
+  // Reject every entry still queued and empty the queue. Both the "error"
+  // listener below and the top-of-loop check inside pump call this, so a
+  // stream error always settles every entry that was queued when it struck,
+  // not only the one pump happened to be holding.
+  function strandQueue(error) {
+    const stranded = queue.splice(0, queue.length);
+    for (const entry of stranded) entry.reject(error);
+  }
+
   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);
+    strandQueue(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);
+        strandQueue(streamError);
         break;
       }
+      const entry = queue[0];
       let wroteOk = false;
       try {
         wroteOk = stream.write(entry.line);
@@ -2064,12 +2078,18 @@ const DUPLEX_GATEWAY_STDOUT_WRITER_SOURCE = `function createStdoutFrameWriter(st
           stream.once("error", onStreamErrorWhileWaiting);
         });
       }
-      queue.shift();
       if (streamError) {
-        entry.reject(streamError);
-      } else {
-        entry.resolve();
+        // The "error" listener already stranded and rejected every entry
+        // that was in the queue when the stream failed, including this one
+        // if it was still queued at that moment. Do not shift here: the
+        // queue head may now be a line a caller pushed after the stream
+        // failed but before this wait resumed, and shifting it here would
+        // discard that entry without ever settling its promise. The
+        // top-of-loop check above strands it on the next turn instead.
+        continue;
       }
+      queue.shift();
+      entry.resolve();
     }
     pumping = false;
   }
@@ -2140,9 +2160,10 @@ const maxDuplexDecoderBytes = Number(
   process.env.PAPERCLIP_BRIDGE_MAX_DUPLEX_DECODER_BYTES || "${DEFAULT_BRIDGE_MAX_DUPLEX_DECODER_BYTES}",
 );
 // The host passes the local in-sandbox body-read admission bound here. The
-// gateway reserves one worst-case body (maxBodyBytes) per active read and sheds
-// a new read with a 503 when the reserve is full. This bound protects the
-// sandbox process memory; it never shares the host aggregate byte ledger.
+// gateway reserves declaredBytes times a per-gateway factor for each active
+// read and sheds a new read with a 503 when the reserve is full. This bound
+// protects the sandbox process memory; it never shares the host aggregate
+// byte ledger.
 const maxInflightBodyBytes = Number(
   process.env.PAPERCLIP_BRIDGE_MAX_INFLIGHT_BODY_BYTES || "${DEFAULT_BRIDGE_MAX_INFLIGHT_BODY_BYTES}",
 );
@@ -2193,9 +2214,10 @@ function normalizeHeaders(headers) {
   return out;
 }
 
-// Typed marker for a request body that exceeds maxBodyBytes. readBody throws it,
-// so each gateway answers a clean 413 request_too_large instead of the generic
-// 502. A 502 is retryable by convention and would leak the internal message.
+// Typed marker for a request body that exceeds its declared length or the
+// bridge cap. readBody throws it, so each gateway answers a clean 413
+// request_too_large instead of the generic 502. A 502 is retryable by
+// convention and would leak the internal message.
 class BridgeBodyTooLargeError extends Error {
   constructor() {
     super("request_too_large");
@@ -2203,50 +2225,98 @@ class BridgeBodyTooLargeError extends Error {
   }
 }
 
+// Read the declared body size from the Content-Length request header. Node's
+// HTTP server validates the header's syntax before a request reaches a
+// handler, so this only needs to guard the two cases that leave the gateway
+// with no declared size: the header is absent, or its value is not a plain
+// non-negative integer (a chunked request carries no Content-Length header
+// at all). Both cases return maxBodyBytes, the prior worst-case charge, so a
+// request that declares nothing keeps the old admission behavior.
+function declaredBodyBytes(req) {
+  const raw = req.headers["content-length"];
+  if (typeof raw !== "string" || !/^[0-9]+$/.test(raw)) {
+    return maxBodyBytes;
+  }
+  const parsed = Number(raw);
+  return Number.isSafeInteger(parsed) ? parsed : maxBodyBytes;
+}
+
+// The file gateway holds four representations of one request body at once
+// after it reads the raw bytes -- see the comment at the writeFile call in
+// runFileGateway. Each one's size relative to the declared body size D:
+//   raw body buffer        1x D  (readBody allocates exactly D)
+//   decodeUtf8Body string  2x D  (worst case: V8 stores the string two bytes
+//                                 per character)
+//   JSON.stringify string  6x D  (worst case: every raw byte is a JSON
+//                                 control character, so it escapes to
+//                                 "\\u00XX", six characters, for one input
+//                                 byte)
+//   fs.writeFile Buffer    6x D  (the escaped string above is pure ASCII, so
+//                                 its UTF-8 encoding is one byte per
+//                                 character, matching the escaped string's
+//                                 six-to-one ratio)
+// 1 + 2 + 6 + 6 = 15. The factor charges the sum of every representation's
+// own worst case, not the true instantaneous peak, so it stays a safe upper
+// bound even if a future change makes two representations overlap that do
+// not overlap today.
+const FILE_GATEWAY_BODY_RESERVE_FACTOR = 15;
+
+// The duplex gateway holds one body buffer plus at most one in-flight
+// base64-encoded frame while it sends that buffer to the host -- see the
+// retained-peak note in runDuplexGateway. The frame never exceeds
+// DEFAULT_MAX_DUPLEX_FRAME_BYTES, and that bound does not grow with the body
+// (the gateway always sends the body as a series of fixed-size frames), so
+// the reserve adds it as a fixed term instead of folding it into a
+// multiplier.
+const DUPLEX_GATEWAY_BODY_READ_RESERVE_EXTRA_BYTES = DEFAULT_MAX_DUPLEX_FRAME_BYTES;
+
 // The running total of reserved body-read bytes across all active reads.
 let reservedBodyBytes = 0;
 const noopRelease = () => {};
 
-// Reserve capacity for one body read against the local admission bound. Return a
-// release function on success, or null when the bound is full. Each caller
-// releases on every terminal path: success, size rejection, other error,
-// timeout, and connection close.
-function reserveBodyRead() {
-  if (reservedBodyBytes + maxBodyBytes > maxInflightBodyBytes) {
+// Reserve reserveBytes against the local admission bound for one active read.
+// Return a release function on success, or null when the bound is full. Each
+// caller releases on every terminal path: success, size rejection, other
+// error, timeout, and connection close.
+//
+// A caller passes the bytes the read will actually retain -- the declared
+// body size times the calling gateway's representation factor -- not a flat
+// per-request charge.
+function reserveBodyRead(reserveBytes) {
+  if (reservedBodyBytes + reserveBytes > maxInflightBodyBytes) {
     return null;
   }
-  reservedBodyBytes += maxBodyBytes;
+  reservedBodyBytes += reserveBytes;
   let released = false;
   return () => {
     if (released) return;
     released = true;
-    reservedBodyBytes -= maxBodyBytes;
+    reservedBodyBytes -= reserveBytes;
   };
 }
 
-// Read one request body into a single buffer sized to the admission charge.
+// Read one request body into a single buffer sized to the declared body
+// length.
 //
 // 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.
+// full body at once, so the transient peak was 2x the raw body -- a
+// dishonest bound. This version preallocates one buffer of declaredBytes 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 for the raw body is exactly one declaredBytes
+// buffer, matching the reserve for it.
 //
-// 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 out = Buffer.allocUnsafe(maxBodyBytes);
+// Buffer.allocUnsafe is safe here because the returned view never exposes a
+// byte the loop did not write: out.subarray(0, totalBytes) stops at
+// totalBytes, and totalBytes only ever advances by the byte count a copy
+// call actually wrote.
+async function readBody(req, declaredBytes) {
+  const out = Buffer.allocUnsafe(declaredBytes);
   let totalBytes = 0;
   for await (const chunk of req) {
     const nextChunk = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
-    if (totalBytes + nextChunk.byteLength > maxBodyBytes) {
+    if (totalBytes + nextChunk.byteLength > declaredBytes) {
       throw new BridgeBodyTooLargeError();
     }
     nextChunk.copy(out, totalBytes);
@@ -2347,13 +2417,22 @@ async function runFileGateway() {
       }
       const requestId = randomUUID();
       const hasRequestBody = req.method && req.method !== "GET" && req.method !== "HEAD";
-      const releaseBodyRead = hasRequestBody ? reserveBodyRead() : noopRelease;
+      const declaredBytes = hasRequestBody ? declaredBodyBytes(req) : 0;
+      if (hasRequestBody && declaredBytes > maxBodyBytes) {
+        // Reject an oversize body from its declared length alone, before the
+        // gateway reserves or reads one byte of it.
+        writeJsonResponse(res, 413, { error: "request_too_large" });
+        return;
+      }
+      const releaseBodyRead = hasRequestBody
+        ? reserveBodyRead(declaredBytes * FILE_GATEWAY_BODY_RESERVE_FACTOR)
+        : noopRelease;
       if (!releaseBodyRead) {
         writeJsonResponse(res, 503, { error: "bridge_busy" });
         return;
       }
       try {
-      const requestBody = decodeUtf8Body(await readBody(req));
+      const requestBody = decodeUtf8Body(await readBody(req, declaredBytes));
       const payload = {
         id: requestId,
         method: req.method || "GET",
@@ -2365,6 +2444,11 @@ async function runFileGateway() {
       };
       const requestPath = path.posix.join(requestsDir, \`\${requestId}.json\`);
       const tempPath = \`\${requestPath}.tmp\`;
+      // JSON.stringify builds a third representation of the body (the escaped
+      // copy inside payload), and fs.writeFile's "utf8" encoding builds a
+      // fourth (the write Buffer). FILE_GATEWAY_BODY_RESERVE_FACTOR charges
+      // for all four representations at once; see its comment for the
+      // worst-case size of each.
       await fs.writeFile(tempPath, \`\${JSON.stringify(payload)}\\n\`, "utf8");
       await fs.rename(tempPath, requestPath);
 
@@ -2679,7 +2763,16 @@ function runDuplexGateway() {
       }
       const requestId = randomUUID();
       const hasRequestBody = req.method && req.method !== "GET" && req.method !== "HEAD";
-      const releaseBodyRead = hasRequestBody ? reserveBodyRead() : noopRelease;
+      const declaredBytes = hasRequestBody ? declaredBodyBytes(req) : 0;
+      if (hasRequestBody && declaredBytes > maxBodyBytes) {
+        // Reject an oversize body from its declared length alone, before the
+        // gateway reserves or reads one byte of it.
+        writeJsonResponse(res, 413, { error: "request_too_large" });
+        return;
+      }
+      const releaseBodyRead = hasRequestBody
+        ? reserveBodyRead(declaredBytes + DUPLEX_GATEWAY_BODY_READ_RESERVE_EXTRA_BYTES)
+        : noopRelease;
       if (!releaseBodyRead) {
         writeJsonResponse(res, 503, { error: "bridge_busy" });
         return;
@@ -2690,7 +2783,7 @@ function runDuplexGateway() {
       // every path that never commits to a send.
       let handedOffToSend = false;
       try {
-      const requestBodyBuffer = await readBody(req);
+      const requestBodyBuffer = await readBody(req, declaredBytes);
       if (unavailable) {
         writeJsonResponse(res, 503, { error: "bridge_unavailable" });
         return;
@@ -2730,7 +2823,7 @@ function runDuplexGateway() {
         return;
       }
       // Retained-peak note. readBody preallocates one buffer sized to
-      // maxBodyBytes and returns a view over it, so no second (concatenated)
+      // declaredBytes 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
@@ -2742,20 +2835,18 @@ function runDuplexGateway() {
       // 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 body buffer     <= declaredBytes
       //   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.
+      // and admission charges exactly that sum
+      // (declaredBytes + DUPLEX_GATEWAY_BODY_READ_RESERVE_EXTRA_BYTES) per
+      // read against the 64 MiB inflight bound, so a small request no longer
+      // charges the full 10 MiB body cap and a large request charges no less
+      // than its true retained peak. 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;