fix(responses): keep caller cancellations out of upstream failure logs (carry of #3515)

docs-site/src/content/docs/reference/proxy-formats.md

@@ -70,6 +70,10 @@ With `stream: true`, the response is `text/event-stream`. The bridge emits Respo
 With `stream: false` or no `stream`, the same adapter events are collected into one Responses JSON
 object. Both forms preserve the selected model, output items, terminal status, and usage.
 
+For native HTTP/SSE passthrough, a client cancellation without an observed upstream terminal is
+logged as `499` with `closeReason: "client_cancel"` and does not penalize the account pool.
+A terminal captured during the bounded post-disconnect drain retains its actual outcome.
+
 Client-facing Responses SSE frames are limited to 4 MiB per frame, measured in raw bytes before the
 SSE block delimiter. On HTTP, an unterminated upstream frame that exceeds the limit fails closed
 with a synthetic `response.failed` event followed by `data: [DONE]`. On the Responses WebSocket

src/server/relay.ts

@@ -1275,6 +1275,7 @@ function startBoundedInspectionPump(options: InspectionPumpOptions): void {
     try {
       for (;;) {
         const { done, value } = await reader.read();
+        if (clientGoneSignal?.aborted) markClientGone();
         if (drainStopped) {
           // stopDrain() cancelled the reader; the settled read is the wake-up.
           clientGoneWithoutTerminal = !inspector.terminalSeen();
@@ -1312,6 +1313,9 @@ function startBoundedInspectionPump(options: InspectionPumpOptions): void {
         }
       }
     } catch {
+      // Bun can settle a fetch body read before dispatching all abort listeners.
+      // Observe the signal itself before classifying that rejection as upstream.
+      if (clientGoneSignal?.aborted) markClientGone();
       // A read error can follow a final SSE block without its blank-line
       // delimiter. Flush that candidate before classifying the transport as a
       // synthetic reset; otherwise a real completed/failed/policy terminal is

src/server/responses/core.ts

@@ -4922,7 +4922,10 @@ async function handleResponsesInner(
       linkAbortSignal(upstream, turnAc.signal);
       registerTurn(turnAc, options.turnAdmissionLease);
       const inspectionConsumerOptions = {
-        clientGoneSignal: clientGone.signal,
+        // Request abort can reject the fetch body before the response cancel hook runs.
+        clientGoneSignal: options.abortSignal
+          ? AbortSignal.any([clientGone.signal, options.abortSignal])
+          : clientGone.signal,
         drainBounds: { ms: 15_000, bytes: 32 * 1024 * 1024 },
         upstream,
         pinCompletedResponseIdToFirstSeen: githubCopilotRepairEnabled,

tests/server/server-auth.test.ts

@@ -4115,6 +4115,81 @@
     }
   });
 
+  test("native passthrough caller abort logs cancellation without penalizing the pool", async () => {
+    const enc = new TextEncoder();
+    const harness = await startPoolRetryHarness(() => new Response(
+      new ReadableStream<Uint8Array>({
+        start(controller) {
+          controller.enqueue(enc.encode('data: {"type":"response.output_text.delta","delta":"hello"}\n\n'));
+        },
+      }),
+      { headers: { "content-type": "text/event-stream" } },
+    ), { secondAccount: false, streamMode: "legacy-tee" });
+    const caller = new AbortController();
+    try {
+      const response = await harness.request({ stream: true, signal: caller.signal });
+      expect(response.status).toBe(200);
+      const requestId = response.headers.get("x-opencodex-request-id");
+      const reader = response.body!.getReader();
+      expect((await reader.read()).done).toBe(false);
+      // Abort the HTTP request, without calling response.body.cancel(): the
+      // incoming request signal may reach upstream before the body cancel hook.
+      caller.abort();
+      await expect(reader.read()).rejects.toThrow();
+      const deadline = Date.now() + INTERNAL_DEADLINE_MS;
+      while (!getRequestLogEntries().some(entry => entry.requestId === requestId) && Date.now() < deadline) {
+        await Bun.sleep(5);
+      }
+      const logs = getRequestLogEntries().filter(entry => entry.requestId === requestId);
+      expect(logs).toHaveLength(1);
+      expect(logs[0]).toMatchObject({ status: 499, closeReason: "client_cancel" });
+      expect(logs[0]?.attempts?.[0]).toMatchObject({ status: 499 });
+      expect(logs[0]?.attempts?.[0]?.streamAborted).not.toBe(true);
+      expect(readUsageEntries().filter(entry => entry.requestId === requestId)).toMatchObject([
+        { status: 499, closeReason: "client_cancel" },
+      ]);
+      expect(getCodexUpstreamHealth("pool-a")?.consecutiveFailures ?? 0).toBe(0);
+      expect(harness.dispatches).toEqual(["acct-pool-a"]);
+    } finally {
+      caller.abort();
+      await stopPoolRetryHarness(harness);
+    }
+  }, { timeout: SERVER_BUDGET_MS });
+
+  test("native passthrough upstream reset still logs 502 and penalizes the pool", async () => {
+    const enc = new TextEncoder();
+    const source = Promise.withResolvers<ReadableStreamDefaultController<Uint8Array>>();
+    const harness = await startPoolRetryHarness(() => new Response(
+      new ReadableStream<Uint8Array>({
+        start(controller) {
+          source.resolve(controller);
+          controller.enqueue(enc.encode('data: {"type":"response.output_text.delta","delta":"hello"}\n\n'));
+        },
+      }),
+      { headers: { "content-type": "text/event-stream" } },
+    ), { secondAccount: false, streamMode: "legacy-tee" });
+    try {
+      const response = await harness.request({ stream: true });
+      const requestId = response.headers.get("x-opencodex-request-id");
+      const reader = response.body!.getReader();
+      expect((await reader.read()).done).toBe(false);
+      (await source.promise).error(new Error("fixture upstream connection reset"));
+      while (!(await reader.read()).done) { /* drain the synthetic failed terminal */ }
+      const deadline = Date.now() + INTERNAL_DEADLINE_MS;
+      while (!getRequestLogEntries().some(entry => entry.requestId === requestId) && Date.now() < deadline) {
+        await Bun.sleep(5);
+      }
+      const logs = getRequestLogEntries().filter(entry => entry.requestId === requestId);
+      expect(logs).toHaveLength(1);
+      expect(logs[0]).toMatchObject({ status: 502, closeReason: "terminal", terminalStatus: "failed" });
+      expect(logs[0]?.attempts?.[0]).toMatchObject({ status: 502, streamAborted: true });
+      expect(getCodexUpstreamHealth("pool-a")?.consecutiveFailures).toBe(1);
+      expect(harness.dispatches).toEqual(["acct-pool-a"]);
+    } finally {
+      await stopPoolRetryHarness(harness);
+    }
+  }, { timeout: SERVER_BUDGET_MS });
+
   test("non-forward generated stream does not mutate active pool health", async () => {
     if (existsSync(TEST_DIR)) removeTreeWithRetry(TEST_DIR);
     mkdirSync(TEST_DIR, { recursive: true });