fix(server): settle upstream bodies on the failure paths too

devlog/_plan/260812_five_bug_fix_campaign/004_audit_wp5_synthesis.md

@@ -0,0 +1,79 @@
+# 004 — WP5 audit: my fetch-site audit was wrong
+
+Independent reviewer verdict on the `051` audit deliverable: **FAIL**. The claim
+it falsified was mine, and the falsification is correct.
+
+## What I claimed, and why it was wrong
+
+I enumerated the request-path fetch sites, found `cancelBodyOnAbort` at eight of
+them, and concluded there was no unguarded site. Two errors:
+
+**1. I checked whether a guard exists in the function, not whether it covers the
+branch.** At `src/server/responses/core.ts` the guard is installed at `:3436`,
+but the non-2xx branch runs at `:3387-3394` and does
+`await upstreamResponse.text()` **before** it. Same shape in
+`src/web-search/executor.ts:85-90` and `anthropic-executor.ts:169-175`: the
+failure branch reads the body, then the guard is attached for the success path.
+So the error path reopens exactly the fetch-resolution-to-reader-attach race the
+guard exists to close. Verified both by reading the code.
+
+**2. I enumerated from a grep of files that already imported the helper.** That
+is a survivorship filter: it can only ever find sites that already have a guard.
+`/v1/live` (`src/server/live.ts:554-570`) has no guard at all, passes no signal
+to `readBodyCapped`, and releases its reader lock without cancelling on read
+failure. It never appeared in my table because it never imported the helper.
+
+The reviewer also found request-time OAuth refresh (`src/oauth/index.ts` dispatch
+reached from `core.ts:1799`), the CCA image fallback, the MiMo JWT bootstrap, and
+the xAI image/video clients — all model-turn paths, none in my table.
+
+## Consequence for the disposition
+
+`051` acceptance criterion 1 is **unmet**, and the disposition must not say "no
+unguarded request-path site found". That sentence would have been a false
+all-clear on the exact question the issue is about.
+
+What stays true and was independently confirmed:
+
+- `installCrashGuards` does install both handlers, redacts, persists best-effort,
+  and keeps the process alive for JS-level failures — with bounded gaps
+  (best-effort persistence, a five-minute fold for known native teardown
+  rejections, installation after listener startup). None of that explains a
+  native `SIGTRAP`.
+- Bun's fetch/TLS teardown is the only in-tree mechanism that fits the reported
+  sequence. Keyring N-API, `bun:sqlite`, Workers, and subprocesses exist but have
+  no temporal or causal link to TLS verification failure; the FFI sites are
+  Windows-gated and cannot explain a macOS crash.
+- Supervision exists (launchd `KeepAlive`, systemd `Restart=on-failure`, WinSW
+  restart-on-failure) but **does not cover an `ocx gui`-started process**:
+  `src/cli/dispatch.ts:240-248` calls a detached spawn directly, and the child at
+  `src/cli/index.ts:924-931` is not adopted by any supervisor. That is precisely
+  why the reporter saw the dashboard die and stay dead.
+- #1419 stays open pending the `.ips` frames.
+
+## Amendment: hardening lands, framed honestly
+
+The unguarded sites are real defects worth fixing on their own merits. They are
+**not** a fix for the reported crash, and the PR and issue comment must say so:
+this is hardening discovered while investigating #1419, not proof the trap is
+resolved.
+
+Scope for this work-phase, ordered by how directly the path serves a user turn:
+
+1. `src/server/responses/core.ts` — guard the non-2xx branches (initial and
+   continuation) before reading the error body.
+2. `src/server/live.ts` — guard the call-create response and cancel on read
+   failure.
+3. `src/web-search/executor.ts`, `anthropic-executor.ts` — move the guard above
+   the failure branch.
+
+Deferred, with reasons recorded rather than silently dropped: request-time OAuth
+refresh (`oauth/*` token endpoints) is a broad surface touching credential
+handling and wants its own change with security review; CCA image fallback, MiMo
+bootstrap, and the xAI clients have a narrower pre-attach interval; quota and
+model discovery are not turn-body paths. Each is named in the follow-up so the
+list is not lost.
+
+Disclosure ordering (`AGENTS.md`): the working table stays in `.tmp/`. The public
+comment names a finding only once its fix is merged and therefore already
+disclosed by the diff.

src/server/live.ts

@@ -32,7 +32,7 @@ import {
   CodexThreadAffinityExpiredError,
 } from "../codex/auth-context";
 import { formatCodexProviderForLog } from "../codex/routing";
-import { signalWithTimeout } from "../lib/abort";
+import { cancelBodyOnAbort, signalWithTimeout } from "../lib/abort";
 import { sidecarEnter } from "../lib/sidecar-tracker";
 import type { OcxConfig } from "../types";
 import { resolveFirstUsableOpenAiSidecar, selectOpenAiImagesProvider } from "../providers/openai-sidecar";
@@ -364,8 +364,18 @@ export async function readBodyCapped(
       }
       chunks.push(value);
     }
+  } catch (err) {
+    // A read that throws leaves the stream neither drained nor cancelled, and releasing the
+    // lock alone hands back an unsettled body. Cancel first, then rethrow so the caller's
+    // existing classification (client abort / timeout / connect error) is unchanged. The
+    // cancel itself can reject with the stream's stored error — that is expected and must not
+    // mask the original failure, so it is swallowed here.
+    await reader.cancel(err).catch(() => {});
+    throw err;
   } finally {
     try {
+      // Always release: `reader.cancel()` does NOT drop the lock, and holding it would leave
+      // the stream permanently locked for any later consumer (audit R-WP5-2).
       reader.releaseLock();
     } catch {
       // already released / cancelled
@@ -563,11 +573,23 @@ export async function handleLive(
     // Record every completed upstream response before body size handling so account health /
     // cooldown still updates when we reject an oversized payload.
     relay.recordOutcome?.(upstreamResponse.status);
-    const payload = await readBodyCapped(
-      upstreamResponse.body,
-      LIVE_RESPONSE_MAX_BYTES,
-      total => `live response too large (${total} bytes)`,
-    );
+    // Settle the body on abort before the reader attaches. Without this, a client cancel or the
+    // linked timeout landing between fetch resolution and `readBodyCapped`'s `getReader()`
+    // leaves Bun's internal read rejection orphaned off the awaited path, where no caller
+    // try/catch can intercept it (src/lib/abort.ts). The guard covers the window BEFORE the
+    // reader exists; once a reader holds the lock only the reader can cancel, which is why
+    // readBodyCapped also cancels on a failed read. Found while investigating #1419.
+    const detachBodyGuard = cancelBodyOnAbort(upstreamResponse.body, linkedSignal.signal);
+    let payload: ArrayBuffer | Response;
+    try {
+      payload = await readBodyCapped(
+        upstreamResponse.body,
+        LIVE_RESPONSE_MAX_BYTES,
+        total => `live response too large (${total} bytes)`,
+      );
+    } finally {
+      detachBodyGuard();
+    }
     if (payload instanceof Response) return payload;
     const relayHeaders: Record<string, string> = {};
     for (const name of LIVE_RELAY_HEADERS) {

src/server/responses/core.ts

@@ -2356,15 +2356,29 @@ async function handleResponsesInner(
     }
     if (!upstreamResponse.ok) {
       if (options.comboAttempt) {
+        // No pre-read guard here: `consumeComboFailure` -> `readBoundedResponseBody` reads
+        // `response.body` itself and already threads the abort signal through its own read,
+        // and the combo contract is that this body's getter is touched exactly once (pinned by
+        // "captures passthrough failed usage from its original bounded body exactly once").
+        // Attaching a guard would be a second `.body` access and break that contract for no
+        // gain, since the bounded reader owns settlement on this path.
         const failure = await consumeComboFailure(upstreamResponse, options.abortSignal);
         options.onConsumedComboFailure?.(failure);
         return failure.response;
       }
-      const errorText = await upstreamResponse.text().catch(() => "");
-      return formatPassthroughUpstreamError(upstreamResponse.status, errorText, {
-        statusText: upstreamResponse.statusText,
-        headers,
-      });
+      // The plain passthrough error path has no bounded reader of its own: `.text()` attaches
+      // the reader only when it runs, so an abort landing between fetch resolution and that
+      // call orphans Bun's internal read rejection (src/lib/abort.ts).
+      const detachPassthroughErrorGuard = cancelBodyOnAbort(upstreamResponse.body, upstream.signal);
+      try {
+        const errorText = await upstreamResponse.text().catch(() => "");
+        return formatPassthroughUpstreamError(upstreamResponse.status, errorText, {
+          statusText: upstreamResponse.statusText,
+          headers,
+        });
+      } finally {
+        detachPassthroughErrorGuard();
+      }
     }
 
     // Bun#32111 workaround: passthrough SSE uses tee()+native relay to avoid the
@@ -3386,12 +3400,22 @@ async function handleResponsesInner(
     }
     if (!upstreamResponse.ok) {
       if (options.comboAttempt) {
+        // No pre-read guard: `consumeComboFailure` -> `readBoundedResponseBody` reads
+        // `response.body` itself with the abort signal threaded through, and the combo
+        // contract is that this body's getter is touched exactly once. A guard here would be
+        // a second `.body` access for no gain, since the bounded reader owns settlement.
         const failure = await consumeComboFailure(upstreamResponse, options.abortSignal)
           .finally(cleanupUpstreamAbort);
         options.onConsumedComboFailure?.(failure);
         return failure.response;
       }
+      // The plain error path has no bounded reader of its own: `.text()` attaches the reader
+      // only when it runs, so an abort landing between fetch resolution and that call orphans
+      // Bun's internal read rejection — the uncatchable teardown `cancelBodyOnAbort` absorbs
+      // (src/lib/abort.ts). Found while investigating #1419; not a fix for the native trap.
+      const detachErrorBodyGuard = cancelBodyOnAbort(upstreamResponse.body, upstream.signal);
       const errorText = await upstreamResponse.text().catch(() => "unknown error");
+      detachErrorBodyGuard();
       cleanupUpstreamAbort();
       if (!isFixedCodexAccount(authCtx)) {
         recordSubagentQuotaFailureForThreadSpawn(
@@ -3655,7 +3679,12 @@ async function handleResponsesInner(
     }
 
     if (!response.ok) {
+      // Same pre-read guard as the initial response's error branch: a non-2xx continuation body
+      // is still a Bun fetch body, and an abort landing before `.text()` attaches its reader
+      // orphans the internal rejection.
+      const detachContinuationErrorGuard = cancelBodyOnAbort(response.body, upstream.signal);
       const errorText = await response.text().catch(() => "unknown error");
+      detachContinuationErrorGuard();
       yield {
         type: "error",
         status: response.status,

src/web-search/anthropic-executor.ts

@@ -166,13 +166,17 @@ export async function runAnthropicWebSearch(
       () => fetch(url, { method: "POST", headers, body: JSON.stringify(body), signal: linkedSignal.signal }),
       { abortSignal: linkedSignal.signal, label: "web-search-sidecar-anthropic" },
     );
+    // Guard before any branch reads the body: the failure branch's `res.text()` ran ahead of
+    // the success-path guard, reopening the fetch-resolution-to-reader-attach race
+    // (found investigating #1419).
+    const detachBodyGuard = cancelBodyOnAbort(res.body, linkedSignal.signal);
     if (!res.ok) {
       const t = await res.text().catch(() => "");
+      detachBodyGuard();
       console.warn(`[web-search] anthropic sidecar HTTP ${res.status} for query "${query.slice(0, 80)}" (${Date.now() - t0}ms)`);
       // Redact before surfacing: the body can echo auth headers/tokens (#398 review).
       return { text: "", sources: [], error: `sidecar HTTP ${res.status}: ${redactSecretString(t.slice(0, 200))}` };
     }
-    const detachBodyGuard = cancelBodyOnAbort(res.body, linkedSignal.signal);
     try {
       return await parseAnthropicSidecarSSE(res);
     } finally {

src/web-search/executor.ts

@@ -82,12 +82,16 @@ export async function runWebSearch(
       { abortSignal: linkedSignal.signal, label: "web-search-sidecar" },
     );
     recordOutcome?.(res.status);
+    // Attach the body guard before ANY branch reads it. The success path guarded itself below,
+    // but the failure branch's `res.text()` runs first, so a cancel landing between fetch
+    // resolution and reader attach orphaned the internal rejection (found investigating #1419).
+    const detachBodyGuard = cancelBodyOnAbort(res.body, linkedSignal.signal);
     if (!res.ok) {
       const t = await res.text().catch(() => "");
+      detachBodyGuard();
       console.warn(`[web-search] sidecar HTTP ${res.status} for query "${query.slice(0, 80)}" (${Date.now() - t0}ms)`);
       return { text: "", sources: [], error: `sidecar HTTP ${res.status}: ${redactSecretString(t.slice(0, 200))}` };
     }
-    const detachBodyGuard = cancelBodyOnAbort(res.body, linkedSignal.signal);
     try {
       return await parseSidecarSSE(res);
     } finally {

tests/cancel-body-on-abort.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, test } from "bun:test";
 import { cancelBodyOnAbort } from "../src/lib/abort";
+import { readBodyCapped } from "../src/server/live";
 
 function bodyWithCancelSpy(): { body: ReadableStream<Uint8Array>; cancelled: () => boolean } {
   let cancelled = false;
@@ -10,6 +11,135 @@ function bodyWithCancelSpy(): { body: ReadableStream<Uint8Array>; cancelled: ()
   return { body, cancelled: () => cancelled };
 }
 
+describe("readBodyCapped settles the stream when a read throws", () => {
+  test("a rejected read propagates and leaves no live reader lock", async () => {
+    // Note on what this can and cannot assert: a source whose own `pull()` rejects is errored
+    // by the stream machinery itself, which by spec does NOT invoke the source's `cancel()`.
+    // So the observable contract here is that the failure propagates to the caller (whose
+    // existing classification turns it into 499/504/502) and that the stream is left settled
+    // rather than pending. The cancel path added alongside this covers the case where the
+    // source is still live — an abort delivered while a read is outstanding.
+    let cancelled = false;
+    const failing = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode("partial"));
+      },
+      pull() { return Promise.reject(new Error("upstream reset")); },
+      cancel() { cancelled = true; },
+    });
+
+    await expect(readBodyCapped(failing, 1024, total => `too large (${total})`)).rejects.toThrow("upstream reset");
+    // The stream errored itself, so `cancel()` is not expected to have run.
+    expect(cancelled).toBe(false);
+    // The lock is released even though the cancel rejected with the stored error: holding it
+    // would leave the stream permanently locked for any later consumer. A second reader is
+    // therefore obtainable, and it observes the original failure rather than hanging.
+    expect(failing.locked).toBe(false);
+    await expect(failing.getReader().read()).rejects.toThrow("upstream reset");
+  });
+
+  test("cancelBodyOnAbort cannot settle a body once a reader holds the lock", async () => {
+    // The reason readBodyCapped needed its own cancel path. `cancelBodyOnAbort` calls
+    // `body.cancel()`, which throws on a locked stream, so once `getReader()` has run the
+    // guard alone can no longer settle the body — only the code holding the reader can.
+    // This is why the guard is attached BEFORE the read (covering the pre-attach window) and
+    // the reader cancels on failure (covering the window after it).
+    let cancelled = false;
+    const pending = new ReadableStream<Uint8Array>({
+      pull() { return new Promise<never>(() => {}); },
+      cancel() { cancelled = true; },
+    });
+
+    const reader = pending.getReader();
+    const ac = new AbortController();
+    cancelBodyOnAbort(pending, ac.signal);
+    ac.abort(new DOMException("client closed request", "AbortError"));
+    await Promise.resolve();
+    await Promise.resolve();
+
+    expect(cancelled).toBe(false);
+
+    // The reader itself can still settle it, which is what the new failure path does.
+    await reader.cancel(new Error("client closed request"));
+    expect(cancelled).toBe(true);
+  });
+
+  // Wiring guard. The unit tests above exercise readBodyCapped and cancelBodyOnAbort
+  // directly, which means they ALL still pass when the /v1/live relay forgets to call the
+  // guard — an earlier revision of this change imported the helper and never invoked it, and
+  // no test noticed. Asserting the call site is crude but it is the thing that was actually
+  // missing.
+  test("the live relay attaches the body guard before consuming the upstream body", async () => {
+    const source = await Bun.file(new URL("../src/server/live.ts", import.meta.url)).text();
+
+    const guardAt = source.indexOf("cancelBodyOnAbort(upstreamResponse.body");
+    const readAt = source.indexOf("payload = await readBodyCapped(");
+    expect(guardAt).toBeGreaterThan(-1);
+    expect(readAt).toBeGreaterThan(-1);
+    // Guard first, read second.
+    expect(guardAt).toBeLessThan(readAt);
+    // And detached on the normal path.
+    expect(source).toContain("detachBodyGuard()");
+  });
+
+  test("the passthrough error branch attaches the body guard before consuming it", async () => {
+    const source = await Bun.file(new URL("../src/server/responses/core.ts", import.meta.url)).text();
+
+    const guardAt = source.indexOf("cancelBodyOnAbort(upstreamResponse.body, upstream.signal)");
+    expect(guardAt).toBeGreaterThan(-1);
+    expect(source).toContain("detachPassthroughErrorGuard");
+  });
+
+  // The combo branches are deliberately NOT guarded: consumeComboFailure ->
+  // readBoundedResponseBody reads `response.body` itself with the abort signal threaded
+  // through, and the combo contract is that the getter is touched exactly once (pinned by
+  // "captures passthrough failed usage from its original bounded body exactly once" in
+  // tests/server-combo-failover-e2e.test.ts). An earlier revision guarded them anyway and
+  // broke that test by adding a second `.body` read.
+  test("the combo failure branches do not add a second body read", async () => {
+    const source = await Bun.file(new URL("../src/server/responses/core.ts", import.meta.url)).text();
+
+    for (const marker of ["const failure = await consumeComboFailure("]) {
+      let from = 0;
+      for (;;) {
+        const at = source.indexOf(marker, from);
+        if (at === -1) break;
+        // Look back a short window: no body guard may be attached immediately before a
+        // combo consumption.
+        const preceding = source.slice(Math.max(0, at - 400), at);
+        expect(preceding).not.toContain("cancelBodyOnAbort(upstreamResponse.body");
+        from = at + marker.length;
+      }
+    }
+  });
+
+  test("a normal read still returns the buffered payload", async () => {
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode("hello"));
+        controller.close();
+      },
+    });
+
+    const payload = await readBodyCapped(body, 1024, total => `too large (${total})`);
+    expect(payload).toBeInstanceOf(ArrayBuffer);
+    expect(new TextDecoder().decode(payload as ArrayBuffer)).toBe("hello");
+  });
+
+  test("the byte cap still short-circuits with a 502 rather than throwing", async () => {
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new Uint8Array(64));
+        controller.close();
+      },
+    });
+
+    const payload = await readBodyCapped(body, 8, total => `too large (${total})`);
+    expect(payload).toBeInstanceOf(Response);
+    expect((payload as Response).status).toBe(502);
+  });
+});
+
 describe("cancelBodyOnAbort", () => {
   test("cancels the body when the signal aborts", async () => {
     const { body, cancelled } = bodyWithCancelSpy();