fix(responses): bound body inactivity without premature shared abort

src/lib/response-body-inactivity.ts

@@ -0,0 +1,193 @@
+/** A response-body guard owns one reader, never the request-wide abort controller. */
+export interface ResponseBodyInactivityGuard {
+  response: Response;
+  /** Stop an abandoned parser without waiting for a broken source's cancel promise. */
+  dispose: () => void;
+}
+
+export class ResponseBodyInactivityError extends DOMException {
+  constructor(timeoutMs: number) {
+    super(`Upstream response body stalled for ${timeoutMs}ms`, "TimeoutError");
+  }
+}
+
+/**
+ * Bound silence only while the consumer is waiting for upstream bytes.
+ *
+ * A zero high-water mark prevents this wrapper from speculatively reading while
+ * the consumer is paused. Empty chunks neither count as progress nor reset the
+ * deadline. Cancellation is local to this body: aborting a shared request signal
+ * here would suppress the failure terminal in the enclosing Responses bridge.
+ */
+export function guardResponseBodyInactivity(
+  response: Response,
+  signal: AbortSignal | undefined,
+  timeoutMs: number,
+): ResponseBodyInactivityGuard {
+  const source = response.body;
+  if (!source) return { response, dispose: () => {} };
+
+  const reader = source.getReader();
+  const timeoutEnabled = Number.isFinite(timeoutMs) && timeoutMs > 0;
+  const timeoutError = new ResponseBodyInactivityError(timeoutMs);
+  let output!: ReadableStreamDefaultController<Uint8Array>;
+  let settled = false;
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  let deadline: number | undefined;
+
+  const pause = (): void => {
+    if (timer !== undefined) clearTimeout(timer);
+    timer = undefined;
+    deadline = undefined;
+  };
+  const release = (): void => {
+    try { reader.releaseLock(); } catch { /* a source may already have torn down */ }
+  };
+  const settle = (): boolean => {
+    if (settled) return false;
+    settled = true;
+    pause();
+    signal?.removeEventListener("abort", onAbort);
+    return true;
+  };
+  const cancelSource = (reason?: unknown): void => {
+    // reader.cancel settles outstanding reads before its source cleanup completes.
+    // Never await that cleanup; custom transports can leave it pending forever.
+    try { void reader.cancel(reason).catch(() => {}); } catch { /* already released */ }
+    release();
+  };
+  const fail = (reason: unknown): void => {
+    if (!settle()) return;
+    // Publish the failure before cancellation can synchronously tear down relays.
+    try { output.error(reason); } catch { /* downstream is already closed */ }
+    cancelSource(reason);
+  };
+  const onAbort = (): void => {
+    fail(signal?.reason ?? new DOMException("The operation was aborted", "AbortError"));
+  };
+  const arm = (): void => {
+    if (settled || deadline === undefined) return;
+    // Avoid the platform's signed-32-bit setTimeout overflow for large settings.
+    const remaining = deadline - performance.now();
+    if (remaining <= 0) {
+      fail(timeoutError);
+      return;
+    }
+    timer = setTimeout(() => {
+      timer = undefined;
+      arm();
+    }, Math.min(2_147_483_647, Math.max(1, remaining)));
+  };
+
+  const body = new ReadableStream<Uint8Array>({
+    start(controller) {
+      output = controller;
+      signal?.addEventListener("abort", onAbort, { once: true });
+      if (signal?.aborted) onAbort();
+    },
+    async pull(controller) {
+      if (settled) return;
+      if (timeoutEnabled) {
+        deadline = performance.now() + timeoutMs;
+        arm();
+      }
+      try {
+        let emptyReads = 0;
+        for (;;) {
+          // A producer of immediately resolved empty chunks can starve timers.
+          // The monotonic deadline check also bounds that microtask-only loop.
+          if (deadline !== undefined && performance.now() >= deadline) {
+            fail(timeoutError);
+            return;
+          }
+          const { done, value } = await reader.read();
+          if (settled) return;
+          if (done) {
+            settle();
+            release();
+            controller.close();
+            return;
+          }
+          if (value.byteLength === 0) {
+            // Empty chunks are not progress, so the deadline keeps running. Hand the
+            // macrotask queue a turn periodically: otherwise a long configured timeout
+            // lets this microtask chain hold timers and unrelated requests.
+            emptyReads += 1;
+            if (emptyReads % 64 === 0) {
+              await new Promise<void>(resolve => { setTimeout(resolve, 0); });
+              if (settled) return;
+            }
+            continue;
+          }
+          emptyReads = 0;
+          pause();
+          controller.enqueue(value);
+          return;
+        }
+      } catch (error) {
+        fail(error);
+      }
+    },
+    cancel(reason) {
+      if (!settle()) return;
+      cancelSource(reason);
+    },
+  }, { highWaterMark: 0 });
+
+  return {
+    response: new Response(body, {
+      status: response.status,
+      statusText: response.statusText,
+      headers: response.headers,
+    }),
+    dispose: () => {
+      if (!settle()) return;
+      try { output.close(); } catch { /* parser already settled */ }
+      cancelSource();
+    },
+  };
+}
+
+/** Native SSE and bounded JSON/error readers retain their existing ownership. */
+export function guardDirectPassthroughBodyInactivity(
+  response: Response,
+  signal: AbortSignal | undefined,
+  timeoutMs: number,
+): Response {
+  const redirect = response.status >= 300 && response.status < 400;
+  const contentType = response.headers.get("content-type")?.toLowerCase();
+  if (!redirect && (!response.ok
+    || contentType?.includes("text/event-stream")
+    || contentType?.includes("application/json"))) return response;
+  return guardResponseBodyInactivity(response, signal, timeoutMs).response;
+}
+
+/** Scope a buffered adapter parser, including throws and early successful returns. */
+export async function readResponseBodyWithInactivity<T>(
+  response: Response,
+  signal: AbortSignal | undefined,
+  timeoutMs: number,
+  parse: (response: Response) => Promise<T>,
+): Promise<T> {
+  const guarded = guardResponseBodyInactivity(response, signal, timeoutMs);
+  try {
+    return await parse(guarded.response);
+  } finally {
+    guarded.dispose();
+  }
+}
+
+/** Scope an adapter stream, including continuation legs and iterator abandonment. */
+export async function* readResponseStreamWithInactivity<T>(
+  response: Response,
+  signal: AbortSignal | undefined,
+  timeoutMs: number,
+  parse: (response: Response) => AsyncIterable<T>,
+): AsyncGenerator<T> {
+  const guarded = guardResponseBodyInactivity(response, signal, timeoutMs);
+  try {
+    yield* parse(guarded.response);
+  } finally {
+    guarded.dispose();
+  }
+}

src/server/responses/adapter-continuation.ts

@@ -46,7 +46,12 @@ import {
 import { shouldAttemptImageTierRetry } from "../image-retry";
 import { readDisplaySafeErrorText, normalizeUpstreamErrorText } from "./core-errors";
 import { isCyberPolicyCode, CYBER_POLICY_FALLBACK_MESSAGE, CYBER_POLICY_ERROR_CODE } from "../../lib/errors";
-import { cancelBodyOnAbort } from "../../lib/abort";
+import {
+  readResponseBodyWithInactivity,
+  readResponseStreamWithInactivity,
+  ResponseBodyInactivityError,
+} from "../../lib/response-body-inactivity";
+import { resolveStallTimeoutSec } from "../../stall-timeout";
 import { guardTerminalEventStream } from "./terminal-guard";
 
 /** One responsibility of the Responses request pipeline; state owners are explicit. */
@@ -113,6 +118,7 @@ export function createAdapterContinuations(
   const { route, translatorBudget, inboundWire, parsed } = requestState;
   const { routedCompaction } = sidecarState;
   const { upstream, connectMs, rateLimitPolicy, stallTimeoutMs } = adapterExchange;
+  const bodyInactivityMs = resolveStallTimeoutSec(config.stallTimeoutSec) * 1000;
   const {
     adapterDispatchBudget,
     noteAdapterPhysicalSend,
@@ -512,23 +518,35 @@ export function createAdapterContinuations(
     }
 
     try {
-      // Protect the continuation body against a client abort landing between fetch resolution and
-      // reader attach, exactly as the initial response is guarded above (#390/366e3053). Without
-      // this, a client cancel during the continuation reopens the Bun fetch-to-reader abort race.
-      const detachContinuationBodyGuard = cancelBodyOnAbort(response.body, upstream.signal);
-      try {
-        if (nextParsed.stream) {
-          yield* transportState.activeAdapter.parseStream(response, translatorBudget, logCtx.activeTierMetadata);
-        } else if (transportState.activeAdapter.parseResponse) {
-          yield* await transportState.activeAdapter.parseResponse(response, translatorBudget, logCtx.activeTierMetadata);
-        } else {
-          yield { type: "error", message: "Provider continuation does not support response parsing" };
-        }
-      } finally {
-        detachContinuationBodyGuard();
+      // Each successful continuation owns a fresh pending-read deadline and an abort
+      // listener that can cancel its reader even while the parser holds the body lock.
+      // The shared signal is not aborted on timeout: this generator must still emit
+      // the error that the already-live Responses bridge turns into a terminal.
+      if (nextParsed.stream) {
+        yield* readResponseStreamWithInactivity(
+          response,
+          upstream.signal,
+          bodyInactivityMs,
+          guarded => transportState.activeAdapter.parseStream(guarded, translatorBudget, logCtx.activeTierMetadata),
+        );
+      } else if (transportState.activeAdapter.parseResponse) {
+        yield* await readResponseBodyWithInactivity(
+          response,
+          upstream.signal,
+          bodyInactivityMs,
+          guarded => transportState.activeAdapter.parseResponse!(guarded, translatorBudget, logCtx.activeTierMetadata),
+        );
+      } else {
+        try { void response.body?.cancel().catch(() => {}); } catch { /* already closed */ }
+        yield { type: "error", message: "Provider continuation does not support response parsing" };
       }
     } catch (error) {
-      if (options.abortSignal?.aborted) {
+      // Classify the thrown error before reading signal state: cancelling the stalled
+      // source can synchronously abort the shared signal, which would otherwise report
+      // this timeout as a client cancellation.
+      if (error instanceof ResponseBodyInactivityError) {
+        yield { type: "error", message: "Provider continuation response body stalled before completing", status: 504 };
+      } else if (options.abortSignal?.aborted || upstream.signal.aborted) {
         yield { type: "error", message: "client closed request during terminal continuation", status: 499 };
       } else {
         yield { type: "error", message: `Provider continuation parse failed: ${redactSecretString(error instanceof Error ? error.message : String(error))}` };

src/server/responses/adapter-delivery.ts

@@ -14,6 +14,12 @@ import { rememberResponseState } from "../../responses/state";
 import { trackStreamLifetime } from "../lifecycle";
 import { awaitThoughtSignatureDurability } from "../../responses/thought-signature-replay";
 import { adapterResponseReachedServingTerminal } from "./core-replay";
+import {
+  readResponseBodyWithInactivity,
+  readResponseStreamWithInactivity,
+  ResponseBodyInactivityError,
+} from "../../lib/response-body-inactivity";
+import { resolveStallTimeoutSec } from "../../stall-timeout";
 
 /** One responsibility of the Responses request pipeline; state owners are explicit. */
 export async function deliverAdapterResponse(
@@ -61,14 +67,33 @@ export async function deliverAdapterResponse(
     notifyResponseComplete,
   } = responseEffects;
   const { routedCompaction } = sidecarState;
+  const bodyInactivityMs = resolveStallTimeoutSec(config.stallTimeoutSec) * 1000;
 
 
   if (parsed.stream) {
-    const initialEventStream = transportState.activeAdapter.parseStream(
-      upstreamResponse,
-      translatorBudget,
-      logCtx.activeTierMetadata,
-    );
+    // The continuation legs classify a stalled body themselves; the initial stream needs the
+    // same mapping or the bridge catch reports this upstream timeout as a 500 proxy_error.
+    const initialEventStream = (async function* (): AsyncGenerator<AdapterEvent> {
+      try {
+        yield* readResponseStreamWithInactivity(
+          upstreamResponse,
+          upstream.signal,
+          bodyInactivityMs,
+          response => transportState.activeAdapter.parseStream(response, translatorBudget, logCtx.activeTierMetadata),
+        );
+      } catch (error) {
+        if (error instanceof ResponseBodyInactivityError) {
+          yield {
+            type: "error",
+            message: "Upstream response body stalled before completing",
+            status: 504,
+            errorType: "upstream_error",
+          };
+          return;
+        }
+        throw error;
+      }
+    })();
     const eventStream = terminalGuardEnabled
       ? guardTerminalEventStream({
           parsed,
@@ -137,10 +162,11 @@ export async function deliverAdapterResponse(
   if (transportState.activeAdapter.parseResponse) {
     let events: AdapterEvent[];
     try {
-      const initialEvents = await transportState.activeAdapter.parseResponse(
+      const initialEvents = await readResponseBodyWithInactivity(
         upstreamResponse,
-        translatorBudget,
-        logCtx.activeTierMetadata,
+        upstream.signal,
+        bodyInactivityMs,
+        response => transportState.activeAdapter.parseResponse!(response, translatorBudget, logCtx.activeTierMetadata),
       );
       let guardedEvents: AdapterEvent[];
       if (terminalGuardEnabled) {
@@ -164,6 +190,11 @@ export async function deliverAdapterResponse(
       } else {
         events = guardedEvents;
       }
+    } catch (error) {
+      if (error instanceof ResponseBodyInactivityError) {
+        return formatErrorResponse(504, "upstream_error", "Upstream response body stalled before completing");
+      }
+      throw error;
     } finally {
       cleanupUpstreamAbort();
     }

src/server/responses/passthrough-execution.ts

@@ -10,6 +10,8 @@ import type { ResponsesEffects } from "./response-effects";
 import type { ResponsesSendBudget } from "./request-send-budget";
 import { preparePassthroughExchange } from "./passthrough-dispatch";
 import { deliverPassthroughResponse } from "./passthrough-delivery";
+import { guardDirectPassthroughBodyInactivity } from "../../lib/response-body-inactivity";
+import { resolveStallTimeoutSec } from "../../stall-timeout";
 import { releaseUpstreamHostAdmission } from "../../codex/upstream-host-health";
 import { releaseCodexAuthContextProbeLease } from "../../codex/auth-context";
 
@@ -36,7 +38,7 @@ export async function executePassthroughResponse(
       sendBudgetState,
     );
     if (nativeExchange instanceof Response) return nativeExchange;
-    return await deliverPassthroughResponse(
+    const response = await deliverPassthroughResponse(
       requestContext,
       admissionState,
       requestState,
@@ -45,6 +47,14 @@ export async function executePassthroughResponse(
       responseEffects,
       nativeExchange,
     );
+    // Delivery has already classified SSE (including a missing upstream content type)
+    // and consumed bounded JSON/errors. Guard only its remaining direct body, outside
+    // the relay/lifetime wrappers so their prefetch cannot arm our inactivity clock.
+    return guardDirectPassthroughBodyInactivity(
+      response,
+      nativeExchange.upstream.signal,
+      resolveStallTimeoutSec(requestContext.config.stallTimeoutSec) * 1000,
+    );
   } finally {
     if (nativeHostState.lease) {
       releaseUpstreamHostAdmission(nativeHostState.lease);