fix(claude): bound unfinished picker uploads and refusal drains

src/claude/intercept/picker-listener.ts

@@ -34,6 +34,10 @@ export interface PickerListenerOptions {
   maxEncodedBytes?: number;
   /** Test seam: concurrent upstream requests; production uses PICKER_MAX_ACTIVE_UPSTREAMS. */
   maxActiveUpstreams?: number;
+  /** Test seam: maximum time for a client upload to finish. */
+  uploadDeadlineMs?: number;
+  /** Test seam: maximum time without upload progress. */
+  uploadIdleTimeoutMs?: number;
   log?: (line: string) => void;
 }
 export interface PickerListenerHandle { port: number; close(): Promise<void> }
@@ -51,6 +55,9 @@ export const PICKER_MAX_ACTIVE_UPSTREAMS = 256;
 export const PICKER_MAX_CONCURRENT_STREAMS = 100;
 /** A connection that has not sent a complete ClientHello by then is dropped. */
 const CLIENT_HELLO_TIMEOUT_MS = 10_000;
+const PICKER_UPLOAD_DEADLINE_MS = 60_000;
+const PICKER_UPLOAD_IDLE_TIMEOUT_MS = 15_000;
+const MAX_TIMER_DELAY_MS = 2_147_483_647;
 
 const HOP_HEADERS = new Set([
   "connection", "keep-alive", "proxy-connection", "proxy-authenticate", "proxy-authorization",
@@ -152,6 +159,13 @@ export async function startPickerListener(options: PickerListenerOptions): Promi
   const upstream = options.upstream ?? { host: "claude.ai", port: 443, servername: "claude.ai" };
   const cap = options.maxEncodedBytes ?? BOOTSTRAP_MAX_ENCODED_BYTES;
   const maxActiveUpstreams = options.maxActiveUpstreams ?? PICKER_MAX_ACTIVE_UPSTREAMS;
+  const uploadDeadlineMs = options.uploadDeadlineMs ?? PICKER_UPLOAD_DEADLINE_MS;
+  const uploadIdleTimeoutMs = options.uploadIdleTimeoutMs ?? PICKER_UPLOAD_IDLE_TIMEOUT_MS;
+  for (const [name, value] of [["uploadDeadlineMs", uploadDeadlineMs], ["uploadIdleTimeoutMs", uploadIdleTimeoutMs]] as const) {
+    if (!Number.isFinite(value) || value <= 0 || value > MAX_TIMER_DELAY_MS) {
+      throw new RangeError(`${name} must be a positive finite timer duration`);
+    }
+  }
   let activeUpstreams = 0;
   const upgrades = new Set<Duplex>();
   const spliced = new Set<Socket>();
@@ -209,10 +223,63 @@ export async function startPickerListener(options: PickerListenerOptions): Promi
     const refuse = (status: 400 | 503) => {
       // Do not leave a refused, still-uploading stream behind after delivering its empty reply.
       if (uploading && !req.complete) {
-        req.once("error", () => res.destroy());
-        responseWritable.once("finish", () => closeInput(true));
+        if (h2) {
+          responseWritable.once("finish", () => closeInput(true));
+        } else {
+          let settled = false;
+          let replyFinished = false;
+          let inputFinished = false;
+          let deadlineTimer: NodeJS.Timeout | undefined;
+          let idleTimer: NodeJS.Timeout | undefined;
+          const cleanup = () => {
+            if (settled) return;
+            settled = true;
+            if (deadlineTimer) clearTimeout(deadlineTimer);
+            if (idleTimer) clearTimeout(idleTimer);
+            req.off("data", onProgress);
+            req.off("end", onBodyEnd);
+            req.off("close", onBodyGone);
+            req.off("error", onBodyGone);
+            responseWritable.off("finish", onReplyFinished);
+          };
+          const closeOnTimeout = () => {
+            cleanup();
+            closeInput(true);
+          };
+          const onProgress = () => {
+            if (settled || !deadlineTimer) return;
+            if (idleTimer) clearTimeout(idleTimer);
+            idleTimer = setTimeout(closeOnTimeout, uploadIdleTimeoutMs);
+            idleTimer.unref?.();
+          };
+          // A body that finishes cleanly leaves a keep-alive socket the client may
+          // reuse; only a deadline or a premature close ends it.
+          const onBodyEnd = () => {
+            inputFinished = true;
+            if (replyFinished) cleanup();
+          };
+          const onBodyGone = () => {
+            if (inputFinished) return;
+            inputFinished = true;
+            if (replyFinished) { cleanup(); closeInput(true); }
+          };
+          const onReplyFinished = () => {
+            if (settled) return;
+            replyFinished = true;
+            if (inputFinished || req.complete) { cleanup(); return; }
+            deadlineTimer = setTimeout(closeOnTimeout, uploadDeadlineMs);
+            deadlineTimer.unref?.();
+            onProgress();
+          };
+          req.on("data", onProgress);
+          req.once("end", onBodyEnd);
+          req.once("close", onBodyGone);
+          req.once("error", onBodyGone);
+          responseWritable.once("finish", onReplyFinished);
+          req.resume();
+        }
       }
-      res.writeHead(status, { "Content-Length": "0", ...(!h2 && uploading && !req.complete ? { Connection: "close" } : {}) });
+      res.writeHead(status, { "Content-Length": "0" });
       res.end();
       options.log?.(`picker request refused ${status}`);
     };
@@ -329,9 +396,33 @@ export async function startPickerListener(options: PickerListenerOptions): Promi
     if (h2) (req as Http2ServerRequest).stream.once("close", onClientClose);
     if (uploading) {
       let uploadFinished = false;
+      let deadlineTimer: NodeJS.Timeout | undefined;
+      let idleTimer: NodeJS.Timeout | undefined;
+      let expireUpload: () => void = () => {};
+      const clearUploadTimers = () => {
+        if (deadlineTimer) clearTimeout(deadlineTimer);
+        if (idleTimer) clearTimeout(idleTimer);
+        deadlineTimer = undefined;
+        idleTimer = undefined;
+      };
+      const onUploadIdle = () => {
+        idleTimer = undefined;
+        // A full upstream buffer pauses the pipe: 'data' stops even while the client
+        // keeps sending, so backpressure is not client idleness. Recheck each interval
+        // until the upstream drains; the absolute deadline still bounds the upload.
+        if (upReq.writableNeedDrain) { onUploadProgress(); return; }
+        expireUpload();
+      };
+      const onUploadProgress = () => {
+        if (idleTimer) clearTimeout(idleTimer);
+        idleTimer = setTimeout(onUploadIdle, uploadIdleTimeoutMs);
+        idleTimer.unref?.();
+      };
       const finishUpload = () => {
         if (uploadFinished) return;
         uploadFinished = true;
+        clearUploadTimers();
+        req.off("data", onUploadProgress);
         req.off("end", finishUpload);
         req.off("close", finishUpload);
         upReq.off("close", onUpstreamClose);
@@ -349,16 +440,26 @@ export async function startPickerListener(options: PickerListenerOptions): Promi
           upReq.destroy();
         }
       };
+      expireUpload = () => {
+        if (uploadFinished) return;
+        clientGone = true;
+        stopUnfinishedUpload();
+        release();
+      };
       const onResponseFinish = () => stopUnfinishedUpload(true);
       const onUpstreamClose = () => {
         if (!upstreamResponded && !res.writableEnded) stopUnfinishedUpload();
       };
       req.once("end", finishUpload);
       req.once("close", finishUpload);
+      req.on("data", onUploadProgress);
       upReq.once("close", onUpstreamClose);
       responseWritable.once("finish", onResponseFinish);
       res.once("close", stopUnfinishedUpload);
       if (h2) (req as Http2ServerRequest).stream.once("close", finishUpload);
+      deadlineTimer = setTimeout(expireUpload, uploadDeadlineMs);
+      deadlineTimer.unref?.();
+      onUploadProgress();
     }
     req.pipe(upReq);
   };

structure/clients/claude-desktop.md

@@ -218,20 +218,23 @@ natively, counting name + value + 32 bytes per field and rejecting an oversized
 The relay retains its 256-request aggregate ceiling across both HTTP versions and all sessions.
 Upload framing, not the method, determines whether input cleanup is needed: HTTP/2 headers
 without END_STREAM, or HTTP/1.1 transfer encoding or a positive Content-Length.
-Rejected unfinished uploads close after their empty reply's writable finishes; HTTP/2 uses the
+Rejected unfinished uploads preserve the empty reply before cleanup; HTTP/2 uses the
 underlying stream's finish event rather than the compatibility response's finish event, which
-waits for both stream halves to close. HTTP/1.1 refusals advertise `Connection: close` only while input is unfinished.
+waits for both stream halves to close.
 Early replies stop relaying unfinished input only after the downstream writable finishes or the
 response closes; upstream request closure alone does not prove a buffered reply was delivered.
 After a complete response, HTTP/2 closes only that stream with `RST_STREAM NO_ERROR` (RFC 9113
 §8.1), including a complete generated 502. Without a complete downstream response, cleanup uses
 `CANCEL` for a premature downstream close or an upstream close without a reply.
-HTTP/1.1 refusals, generated 502s and early replies advertise `Connection: close` only while input is unfinished; after the writable finishes,
-the listener unpipes the upstream upload, reads and discards remaining input, and ends the socket
-gracefully so queued response bytes flush. Errors and premature downstream closes still destroy
-the request. Draining adds no upload deadline or byte cap and does not wait for request-body
-completion before ending the response side. Body completion removes the input-cleanup listeners,
-leaving long-lived responses and SSE subscriptions independent of upload cleanup. Completed HTTP/1.1 uploads retain keep-alive on generated 502s and can reuse the same client socket.
+An early HTTP/1.1 refusal preserves its reply, then drains unfinished input: a body that
+completes leaves the keep-alive socket reusable, while only a body that stalls past the absolute
+or idle deadline, or a premature close, ends it. Errors and premature downstream
+closes still destroy the request. The 60-second absolute and 15-second idle-progress deadlines
+close client input and upstream work and release the aggregate slot; upstream backpressure only
+suspends the idle clock, never the absolute deadline. Body
+completion removes the input-cleanup listeners and both timers, leaving long-lived responses and
+SSE subscriptions independent of upload cleanup. Completed HTTP/1.1 uploads retain keep-alive on
+generated 502s and can reuse the same client socket.
 `tests/claude-integration/claude-picker-upload.test.ts` covers headers-only refusals on both
 protocols, byte-exact early replies under backpressure, completed uploads with SSE, cancellation,
 failure and shutdown cleanup, and healthy sibling streams.

tests/claude-integration/claude-picker-upload.test.ts

@@ -70,13 +70,15 @@ function h2Result(stream: ClientHttp2Stream) {
 }
 function h1Result(req: ClientRequest) {
   const result = Promise.withResolvers<{ status: number; text: string }>();
+  let responseStarted = false;
   req.once("response", res => {
+    responseStarted = true;
     const chunks: Buffer[] = [];
     res.on("data", chunk => chunks.push(Buffer.from(chunk)));
     res.once("end", () => result.resolve({ status: res.statusCode!, text: Buffer.concat(chunks).toString() }));
     res.once("error", result.reject);
   });
-  req.once("error", result.reject);
+  req.once("error", error => { if (!responseStarted) result.reject(error); });
   void result.promise.catch(() => {});
   return result.promise;
 }
@@ -89,7 +91,7 @@ function observeClose(stream: ClientRequest | ClientHttp2Stream | ClientHttp2Ses
 function observeSocketClose(req: ClientRequest) {
   const closed = Promise.withResolvers<void>();
   req.once("socket", socket => { socket.once("close", () => closed.resolve()); });
-  req.on("error", () => req.destroy());
+  req.on("error", () => {});
   return closed.promise;
 }
 function normalH2(session: ClientHttp2Session, path = "/ok") {
@@ -109,7 +111,7 @@ for (const protocol of ["h1", "h2"] as const) {
           if (req.url === "/held") {
             req.once("close", () => left.resolve()); held.resolve();
           } else { relayed++; res.end("ordinary"); }
-        }, { maxActiveUpstreams: 1 });
+        }, { maxActiveUpstreams: 1, uploadIdleTimeoutMs: 250 });
         try {
           const session = f.h2();
           let blocker: ClientHttp2Stream | undefined;
@@ -131,7 +133,7 @@ for (const protocol of ["h1", "h2"] as const) {
           }
           if (input === "buffered") upload.write(Buffer.alloc(4 * 1024 * 1024, 0x61));
           expect(await wait(result)).toEqual({ status, text: "" });
-          if (protocol === "h1") expect(connection).toBe("close");
+          if (protocol === "h1") expect(connection).not.toBe("close");
           await wait(closed);
           if (protocol === "h2") expect((upload as ClientHttp2Stream).rstCode).toBe(constants.NGHTTP2_NO_ERROR);
           expect(relayed).toBe(0);
@@ -145,6 +147,29 @@ for (const protocol of ["h1", "h2"] as const) {
   }
 }
 
+bounded("h1 refusal keeps the same connection alive once the upload drains", async wait => {
+  const body = "complete";
+  const f = await fixture((req, res) => {
+    req.resume(); res.end("ordinary");
+  });
+  try {
+    // A refused upload that finishes its body must not have its socket closed:
+    // the keep-alive advertisement in the empty reply stays honest and the next
+    // request on that socket is relayed instead of dropped.
+    const upload = f.h1("//[", "POST", { "Content-Length": String(body.length) });
+    const firstSocket = Promise.withResolvers<Duplex>();
+    upload.once("socket", firstSocket.resolve);
+    const result = h1Result(upload); upload.end(body);
+    expect(await wait(result)).toEqual({ status: 400, text: "" });
+    const next = f.h1("/next");
+    const nextSocket = Promise.withResolvers<Duplex>();
+    next.once("socket", nextSocket.resolve);
+    const nextResult = h1Result(next); next.end();
+    expect(await wait(nextResult)).toEqual({ status: 200, text: "ordinary" });
+    expect(await wait(nextSocket.promise)).toBe(await wait(firstSocket.promise));
+  } finally { await wait(f.close()); }
+});
+
 bounded("h1 completed upload keeps the same connection alive after an upstream failure", async wait => {
   const body = "complete";
   let received = "";
@@ -251,6 +276,90 @@ for (const protocol of ["h1", "h2"] as const) {
   });
 }
 
+bounded("unfinished uploads share the aggregate budget across HTTP/2 sessions and recover on idle timeout", async wait => {
+  const arrived = Promise.withResolvers<void>();
+  let requests = 0;
+  const f = await fixture((req, res) => {
+    requests++;
+    if (req.url === "/stall") {
+      req.on("data", () => {});
+      req.once("end", () => res.end("uploaded"));
+      arrived.resolve();
+    } else res.end("ordinary");
+  }, { maxActiveUpstreams: 1, uploadDeadlineMs: 1_000, uploadIdleTimeoutMs: 350 });
+  try {
+    const firstSession = f.h2();
+    const secondSession = f.h2();
+    const upload = firstSession.request({ ":method": "POST", ":path": "/stall", ":authority": "claude.ai" });
+    const uploadClosed = observeClose(upload);
+    upload.write("partial");
+    await wait(arrived.promise);
+    const blocked = await wait(normalH2(secondSession, "/blocked"));
+    expect(blocked.status).toBe(503);
+    await wait(uploadClosed);
+    expect((await wait(normalH2(secondSession, "/ok"))).status).toBe(200);
+    expect(requests).toBe(2);
+  } finally { await wait(f.close()); }
+});
+
+bounded("absolute upload deadline expires despite continued progress", async wait => {
+  const arrived = Promise.withResolvers<void>();
+  const f = await fixture((req, res) => {
+    req.on("data", () => {});
+    arrived.resolve();
+    req.once("end", () => res.end("uploaded"));
+  }, { uploadDeadlineMs: 400, uploadIdleTimeoutMs: 250 });
+  let progress: NodeJS.Timeout | undefined;
+  try {
+    const session = f.h2();
+    const upload = session.request({ ":method": "POST", ":path": "/trickle", ":authority": "claude.ai" });
+    const closed = observeClose(upload);
+    progress = setInterval(() => upload.write("x"), 80);
+    upload.write("start");
+    await wait(arrived.promise);
+    await wait(closed);
+    expect(upload.rstCode).toBe(constants.NGHTTP2_CANCEL);
+  } finally { if (progress) clearInterval(progress); await wait(f.close()); }
+});
+
+bounded("upstream backpressure does not expire an active upload as idle", async wait => {
+  const arrived = Promise.withResolvers<void>();
+  const f = await fixture((req, res) => {
+    // Never consume the body: the upstream read buffer stays full, so the relay
+    // pipe pauses client input even though the client has more bytes to send.
+    if (req.url === "/stall") arrived.resolve(); else res.end("ordinary");
+  }, { uploadDeadlineMs: 1_000, uploadIdleTimeoutMs: 250 });
+  try {
+    const session = f.h2();
+    const upload = session.request({ ":method": "POST", ":path": "/stall", ":authority": "claude.ai" });
+    const closed = observeClose(upload);
+    let closedEarly = false;
+    upload.once("close", () => { closedEarly = true; });
+    upload.write(Buffer.alloc(4 * 1024 * 1024, 0x61));
+    await wait(arrived.promise);
+    // Past the idle interval the upload must still be open: the client is stalled
+    // by upstream backpressure, not idle. The absolute deadline still ends it.
+    await new Promise(resolve => setTimeout(resolve, 600));
+    expect(closedEarly).toBe(false);
+    await wait(closed);
+    expect(upload.rstCode).toBe(constants.NGHTTP2_CANCEL);
+  } finally { await wait(f.close()); }
+});
+
+bounded("completed uploads are not timed out while a long response is pending", async wait => {
+  const f = await fixture((req, res) => {
+    req.resume();
+    req.once("end", () => setTimeout(() => res.end("slow response"), 120));
+  }, { uploadDeadlineMs: 50, uploadIdleTimeoutMs: 25 });
+  try {
+    const session = f.h2();
+    const upload = session.request({ ":method": "POST", ":path": "/complete", ":authority": "claude.ai" });
+    const result = h2Result(upload);
+    upload.end("complete");
+    expect(await wait(result)).toEqual({ status: 200, text: "slow response" });
+  } finally { await wait(f.close()); }
+});
+
 for (const protocol of ["h1", "h2"] as const) {
   bounded(`${protocol} backpressure preserves every byte of a large early reply`, async wait => {
     // Exceed the h2 receive window and the h1 socket buffers, including a non-text tail.