fix(adapter-utils): stop duplex bridge body-read amplification and bound the retained peak

The in-sandbox gateway charged request-body admission on the raw source
bytes, but it retained and forwarded more bytes than it charged. Two
defects allowed a body to exceed the charged bytes.

First, readBody returned a UTF-8 string. The duplex gateway re-encoded
that string to a Buffer. Each invalid byte became the 3-byte U+FFFD
replacement, so a cap-sized malformed body expanded to about 3x the
charged bytes after admission. Preserve the raw request bytes end to end:
readBody now returns the raw Buffer and the duplex gateway uses it
directly. The base64 frames and the envelope byte count now equal the
wire bytes, so a malformed body cannot expand.

Second, the duplex path held the raw body plus the full base64 frame
array plus the full encoded-line array at once, about 3.7x the raw body.
Stream the body as body_chunk frames one at a time, so the gateway holds
the raw body plus at most one frame. The retained peak per read is now
about the charged bytes plus 1 MB, so the inflight bound bounds the true
retained bytes.

The file gateway writes the body as a JSON string that the host reader
consumes. It rejects an invalid-UTF-8 body with a clean 400 before it
converts the body, so it does not amplify a malformed body and keeps the
on-disk payload contract unchanged.

Tests: a cap-sized malformed-UTF-8 body forwards byte-for-byte with no
expanded frame; the admission bound caps concurrent reads at the
retained-byte model; the file gateway answers 400 for an invalid body
and forwards nothing.

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

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

@@ -18,6 +18,7 @@ import {
   DUPLEX_BODY_CHUNK_RAW_BYTES,
   DUPLEX_FRAME_VERSION,
   encodeDuplexFrame,
+  type DuplexBodyChunkFrame,
   type DuplexFrame,
   type DuplexReadyFrame,
   type DuplexRequestFrame,
@@ -859,4 +860,122 @@ describe("duplex bridge local end-to-end harness", () => {
     const later = await post();
     expect(later.status).toBe(200);
   }, 20000);
+
+  it("forwards a cap-sized malformed-UTF-8 body byte-for-byte and emits no expanded frame", async () => {
+    // Regression test for the admission-bypass block. The gateway used to decode
+    // the raw body to a UTF-8 string, then re-encode it. Each invalid byte
+    // became the three-byte U+FFFD replacement, so a cap-sized malformed body
+    // expanded to about three times the bytes that admission charged. The gateway
+    // now preserves the raw bytes, so the body rides base64 frames unchanged and
+    // no frame carries more than the charged bytes.
+    const maxBodyBytes = 700 * 1024;
+    const harness = await createHarness({ maxBodyBytes });
+    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 }));
+    });
+
+    // 0xFF is never a valid UTF-8 byte. A cap-sized body of 0xFF bytes is the
+    // worst case for the old expansion path.
+    const malformed = Buffer.alloc(maxBodyBytes, 0xff);
+    const response = await fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, {
+      method: "POST",
+      headers: {
+        authorization: `Bearer ${harness.bridgeToken}`,
+        "content-type": "application/json",
+      },
+      body: new Uint8Array(malformed),
+    });
+    expect(response.status).toBe(200);
+
+    // The observed frames are the exact bytes the gateway sent to the host. The
+    // envelope reports the raw byte count, not an expanded count.
+    const requestFrame = harness.observedFrames.find(
+      (frame): frame is DuplexRequestFrame => frame.type === "request",
+    );
+    expect(requestFrame?.bodyByteCount).toBe(maxBodyBytes);
+
+    // The body_chunk frames decode to the exact source bytes. No frame carried an
+    // expanded body, so no body larger than the charged bytes reached the host.
+    const bodyChunks = harness.observedFrames.filter(
+      (frame): frame is DuplexBodyChunkFrame => frame.type === "body_chunk",
+    );
+    const reassembled = Buffer.concat(
+      bodyChunks.map((frame) => Buffer.from(frame.data, "base64")),
+    );
+    expect(reassembled.length).toBe(maxBodyBytes);
+    expect(reassembled.equals(malformed)).toBe(true);
+
+    // The route stays usable after the malformed body.
+    const later = await fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, {
+      method: "POST",
+      headers: {
+        authorization: `Bearer ${harness.bridgeToken}`,
+        "content-type": "application/json",
+      },
+      body: JSON.stringify({ ok: true }),
+    });
+    expect(later.status).toBe(200);
+  }, 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.
+    const maxBodyBytes = 10 * 1024 * 1024 + 1;
+    const maxInflightBodyBytes = 64 * 1024 * 1024;
+    const admitCount = Math.floor(maxInflightBodyBytes / maxBodyBytes);
+    expect(admitCount).toBe(6);
+    const harness = await createHarness({
+      maxBodyBytes,
+      maxInflightBodyBytes,
+      // A short response deadline releases each hung read fast.
+      responseTimeoutMs: 800,
+    });
+    harnesses.push(harness);
+    await harness.waitFor(readyPredicate(harness), "the gateway to become ready");
+
+    // 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 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: admitCount + 1 }, () => post()),
+    );
+    const statuses = responses.map((response) => response.status);
+    // The model admits exactly admitCount reads. Each admitted read times out
+    // with a 502. The one read past the model is shed with a 503 before it
+    // buffers a body.
+    const shedCount = statuses.filter((status) => status === 503).length;
+    const admittedTimedOut = statuses.filter((status) => status === 502).length;
+    expect(shedCount).toBe(1);
+    expect(admittedTimedOut).toBe(admitCount);
+    const shed = responses.find((response) => response.status === 503);
+    await expect(shed?.json()).resolves.toEqual({ error: "bridge_busy" });
+
+    // The admitted reads released their reserve on timeout. A later request is
+    // admitted again and reaches the API.
+    harness.setForwardMode("proxy");
+    harness.api.setResponder((_req, res) => {
+      res.writeHead(200, { "content-type": "application/json" });
+      res.end(JSON.stringify({ ok: true }));
+    });
+    const later = await post();
+    expect(later.status).toBe(200);
+  }, 30000);
 });

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

@@ -2557,6 +2557,78 @@ describe("sandbox adapter execution targets", () => {
     }
   });
 
+  it("rejects an invalid-UTF-8 request body with a clean 400 and never forwards it", async () => {
+    // The file gateway writes the request body as a JSON string that the host
+    // reader consumes. A valid JSON body is always valid UTF-8. An invalid body
+    // would expand during a lossy UTF-8 round trip, so the gateway would retain
+    // and forward more bytes than admission charged. The gateway rejects an
+    // invalid-UTF-8 body with a clean 400 before it writes the body, so no
+    // expanded body reaches the host and the host runs no mutation.
+    const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-execution-target-bridge-invalid-utf8-"));
+    cleanupDirs.push(rootDir);
+    const remoteCwd = path.join(rootDir, "workspace");
+    const runtimeRootDir = path.join(remoteCwd, ".paperclip-runtime", "codex");
+    await mkdir(runtimeRootDir, { recursive: true });
+
+    const requests: Array<{ method: string; url: string }> = [];
+    const apiServer = createServer((req, res) => {
+      requests.push({ method: req.method ?? "GET", url: req.url ?? "/" });
+      res.writeHead(200, { "content-type": "application/json" });
+      res.end(JSON.stringify({ ok: true }));
+    });
+    await new Promise<void>((resolve, reject) => {
+      apiServer.once("error", reject);
+      apiServer.listen(0, "127.0.0.1", () => resolve());
+    });
+    const address = apiServer.address();
+    if (!address || typeof address === "string") {
+      throw new Error("Expected the bridge test API server to listen on a TCP port.");
+    }
+
+    const target: AdapterSandboxExecutionTarget = {
+      kind: "remote",
+      transport: "sandbox",
+      providerKey: "e2b",
+      environmentId: "env-1",
+      leaseId: "lease-1",
+      remoteCwd,
+      runner: createLocalSandboxRunner(),
+      timeoutMs: 30_000,
+    };
+
+    const bridge = await startAdapterExecutionTargetPaperclipBridge({
+      runId: "run-bridge-invalid-utf8",
+      target,
+      runtimeRootDir,
+      adapterKey: "codex",
+      hostApiToken: "real-run-jwt",
+      hostApiUrl: `http://127.0.0.1:${address.port}`,
+      maxBodyBytes: 4096,
+    });
+    try {
+      // 0xFF is never a valid UTF-8 byte, so this body is not valid UTF-8.
+      const malformed = new Uint8Array([0x7b, 0xff, 0xff, 0x7d]);
+      const response = await fetch(`${bridge!.env.PAPERCLIP_API_URL}/api/issues/issue-1/comments`, {
+        method: "POST",
+        headers: {
+          authorization: `Bearer ${bridge!.env.PAPERCLIP_API_KEY}`,
+          "content-type": "application/json",
+        },
+        body: malformed,
+      });
+      // The gateway answers a clean, non-retryable 400 and leaks no internal
+      // message.
+      expect(response.status).toBe(400);
+      await expect(response.json()).resolves.toEqual({ error: "invalid_utf8_body" });
+      // The gateway rejected the body before it wrote the request, so the host
+      // reader forwarded nothing and ran no mutation.
+      expect(requests).toEqual([]);
+    } finally {
+      await bridge?.stop();
+      await new Promise<void>((resolve) => apiServer.close(() => resolve()));
+    }
+  });
+
   it("fails an oversized host response with a non-retryable 409 so a committed mutation never repeats", async () => {
     // The host receives the request and commits the mutation, then sends a
     // response body over the size limit. The forward reads the body after the

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

@@ -2146,7 +2146,38 @@ async function readBody(req) {
       throw new BridgeBodyTooLargeError();
     }
   }
-  return Buffer.concat(chunks).toString("utf8");
+  // 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);
+}
+
+// Strict UTF-8 decoder for the file gateway. It throws on any invalid byte, so
+// the gateway rejects a malformed body instead of expanding it. A single decode
+// call does not stream, so the gateway reuses one decoder across requests.
+const bridgeUtf8Decoder = new TextDecoder("utf8", { fatal: true });
+
+// Typed marker for a request body that is not valid UTF-8. Only the file gateway
+// uses it. That gateway writes the body as a JSON string that the host reader
+// consumes. A valid JSON body is always valid UTF-8 and round-trips
+// byte-for-byte. An invalid body would expand, because each invalid byte becomes
+// the 3-byte U+FFFD replacement. The gateway must not retain or forward more
+// bytes than admission charged, so it answers a clean 400 instead.
+class BridgeInvalidBodyError extends Error {
+  constructor() {
+    super("invalid_utf8_body");
+    this.code = "invalid_utf8_body";
+  }
+}
+
+// Decode a request body Buffer to a UTF-8 string, or throw BridgeInvalidBodyError.
+function decodeUtf8Body(buffer) {
+  try {
+    return bridgeUtf8Decoder.decode(buffer);
+  } catch (error) {
+    throw new BridgeInvalidBodyError();
+  }
 }
 
 function tokensMatch(received) {
@@ -2215,7 +2246,7 @@ async function runFileGateway() {
         return;
       }
       try {
-      const requestBody = await readBody(req);
+      const requestBody = decodeUtf8Body(await readBody(req));
       const payload = {
         id: requestId,
         method: req.method || "GET",
@@ -2258,6 +2289,10 @@ async function runFileGateway() {
         writeJsonResponse(res, 413, { error: "request_too_large" });
         return;
       }
+      if (error && error.code === "invalid_utf8_body") {
+        writeJsonResponse(res, 400, { error: "invalid_utf8_body" });
+        return;
+      }
       writeJsonResponse(res, 502, { error: error instanceof Error ? error.message : String(error) });
     }
   });
@@ -2293,26 +2328,26 @@ async function runFileGateway() {
   });
 }
 
-// Split one whole body buffer into body_chunk frames. The gateway buffers the
-// whole source body once, then splits it here. Each frame carries one fixed raw
-// slice as base64 text, except the final frame, which carries the remaining
-// bytes. A zero-length body yields no frame. Send-side true streaming is a
-// separate goal; this whole-body split is acceptable for this gateway.
-function splitDuplexBodyIntoChunks(id, body) {
-  const frames = [];
+// Yield the body_chunk frames for one request body, one frame at a time. The
+// gateway encodes and writes each frame, then drops it before it builds the next
+// frame. So the gateway holds the raw body plus at most one base64 frame, not
+// the raw body plus every base64 frame. Each frame carries one fixed raw slice
+// as base64 text, except the final frame, which carries the remaining bytes. A
+// zero-length body yields no frame. See the retained-peak note at the duplex
+// send site.
+function* duplexBodyChunkFrames(id, body) {
   let seq = 0;
   for (let offset = 0; offset < body.length; offset += DUPLEX_BODY_CHUNK_RAW_BYTES) {
     const slice = body.subarray(offset, offset + DUPLEX_BODY_CHUNK_RAW_BYTES);
-    frames.push({
+    yield {
       version: DUPLEX_FRAME_VERSION,
       type: "body_chunk",
       id: id,
       seq: seq,
       data: slice.toString("base64"),
-    });
+    };
     seq += 1;
   }
-  return frames;
 }
 
 function runDuplexGateway() {
@@ -2524,7 +2559,7 @@ function runDuplexGateway() {
         return;
       }
       try {
-      const requestBodyBuffer = Buffer.from(await readBody(req), "utf8");
+      const requestBodyBuffer = await readBody(req);
       if (unavailable) {
         writeJsonResponse(res, 503, { error: "bridge_unavailable" });
         return;
@@ -2547,26 +2582,35 @@ function runDuplexGateway() {
         writeJsonResponse(res, 413, { error: "request_too_large" });
         return;
       }
-      // Pre-encode every body_chunk frame and enforce the frame size bound on
-      // each. A fixed raw slice never exceeds the bound, so this guard is
-      // defensive. On a rejection, fail this one local request with a clean 413.
-      // The frames never leave the gateway, so no other in-flight request is
-      // affected and the channel stays open.
-      const chunkFrames = splitDuplexBodyIntoChunks(requestId, requestBodyBuffer);
-      const encodedChunks = [];
+      // Validate every body_chunk frame against the frame bound before the
+      // gateway emits the envelope. A fixed 256 KiB raw slice never exceeds the
+      // 1 MB frame bound, so this guard is defensive. On a rejection the gateway
+      // fails this one local request with a clean 413 and emits nothing, so no
+      // partial body reaches the host and the channel stays open.
       let chunkTooLarge = false;
-      for (const chunk of chunkFrames) {
-        const encodedChunk = encodeDuplexFrameChecked(chunk);
-        if (!encodedChunk.ok) {
+      for (const frame of duplexBodyChunkFrames(requestId, requestBodyBuffer)) {
+        if (!encodeDuplexFrameChecked(frame).ok) {
           chunkTooLarge = true;
           break;
         }
-        encodedChunks.push(encodedChunk.line);
       }
       if (chunkTooLarge) {
         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.
       const response = await new Promise((resolve) => {
         const timer = setTimeout(() => {
           responseAssembly.delete(requestId);
@@ -2580,7 +2624,11 @@ function runDuplexGateway() {
         }, responseTimeoutMs);
         pending.set(requestId, { resolve: resolve, timer: timer });
         process.stdout.write(encodedRequest.line);
-        for (const line of encodedChunks) process.stdout.write(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);
+        }
       });
       res.statusCode = typeof response.status === "number" ? response.status : 200;
       for (const [key, value] of Object.entries(response.headers || {})) {