diff --git a/src/lib/response-body-inactivity.ts b/src/lib/response-body-inactivity.ts
new file mode 100644
index 00000000000..7c9a64836c7
--- /dev/null
+++ b/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();
+  }
+}
diff --git a/src/server/responses/adapter-continuation.ts b/src/server/responses/adapter-continuation.ts
index 79879e2f4ac..425e933c02e 100644
--- a/src/server/responses/adapter-continuation.ts
+++ b/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))}` };
diff --git a/src/server/responses/adapter-delivery.ts b/src/server/responses/adapter-delivery.ts
index e982e9dfdeb..e0351128bd6 100644
--- a/src/server/responses/adapter-delivery.ts
+++ b/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();
     }
diff --git a/src/server/responses/passthrough-execution.ts b/src/server/responses/passthrough-execution.ts
index 4fcdf84eeb5..425f2c16e5a 100644
--- a/src/server/responses/passthrough-execution.ts
+++ b/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);
diff --git a/structure/transports/streaming-health.md b/structure/transports/streaming-health.md
index 5658981201d..ca5ce00455f 100644
--- a/structure/transports/streaming-health.md
+++ b/structure/transports/streaming-health.md
@@ -63,6 +63,43 @@ response headers/status and any 429 key rotations are handled eagerly. A failure
 SSE starts returns non-2xx JSON; once headers have started the final response, a generation failure
 is emitted as `response.failed` SSE.
 
+### Pending response-body reads
+
+`src/lib/response-body-inactivity.ts` bounds pending byte reads using the resolved
+`stallTimeoutSec` (the 300-second default and configuration schema are unchanged).
+Connect/stall budgets and this body-silence budget therefore read the same setting in
+different phases: one bounds event stalls on the bridge, the other bounds pending raw
+byte reads. There is no separate body-inactivity setting.
+The guard has no read-ahead queue: it starts a monotonic deadline only when its
+consumer asks for bytes, pauses on a non-empty chunk, and does not reset on empty
+chunks. Discarded empty chunks yield to the macrotask queue periodically, so a large
+configured timeout cannot starve timers or unrelated requests, and yielding is not
+progress. A slow downstream consumer is not an upstream stall. EOF, errors, caller
+abort and parser abandonment remove the timer/listener and release the source
+reader without awaiting a potentially broken cancellation promise.
+
+`src/server/responses/passthrough-execution.ts` applies the guard after native
+response classification, outside the direct relay and lifetime wrappers. Native
+SSE, including the missing-Content-Type fallback, retains its original Response
+identity and existing watchdog/terminal ownership. Bounded JSON and error readers
+are not wrapped again. Direct bodies and returned redirect bodies are guarded;
+bytes, response headers and status are preserved.
+
+`src/server/responses/adapter-delivery.ts` and
+`src/server/responses/adapter-continuation.ts` scope initial and continuation
+parsers independently, for both streaming and buffered HTTP adapters. Retried
+responses are classified before parsing, so unread retry-body cancellation does
+not abort the shared request or start a generation deadline during backoff.
+A body timeout publishes a typed read failure and cancels only that body's reader;
+it does not abort the request-wide signal before the enclosing bridge can emit
+its failure terminal. A caught body failure is classified before signal state, so a
+source cancellation that synchronously aborts the shared signal still reports the
+stall. Buffered initial timeouts return HTTP 504; continuation
+timeouts become an in-stream error with status 504, while caller cancellation
+retains status 499. Normal completion does not abort a shared request.
+
+Regression coverage lives in `tests/lib/abort-idle-deadline.test.ts`.
+
 ### Pre-stream provider input overflow
 
 A provider HTTP 413 received before streaming starts is unambiguous request-size refusal, but raw
diff --git a/tests/lib/abort-idle-deadline.test.ts b/tests/lib/abort-idle-deadline.test.ts
index d72f09488e5..a4c855f3d79 100644
--- a/tests/lib/abort-idle-deadline.test.ts
+++ b/tests/lib/abort-idle-deadline.test.ts
@@ -1,3 +1,12 @@
+import { strict as assert } from "node:assert";
+import {
+  guardResponseBodyInactivity,
+  guardDirectPassthroughBodyInactivity,
+  readResponseBodyWithInactivity,
+  readResponseStreamWithInactivity,
+  ResponseBodyInactivityError,
+} from "../../src/lib/response-body-inactivity";
+
 import { expect, spyOn, test } from "bun:test";
 import { idleDeadline } from "../../src/lib/abort";
 
@@ -112,3 +121,501 @@ test("idleDeadline starts disarmed: constructing without reset never fires", asy
   await sleep(60);
   expect(fired).toBe(0);
 });
+
+// Deterministic clock: advancing it never depends on a loaded runner's real sleeps.
+function bodyClock() {
+  let now = 0;
+  let next = 0;
+  const timers = new Map<number, { at: number; fire: () => void }>();
+  const disposers: Array<() => void> = [];
+  const originalSet = globalThis.setTimeout;
+  const originalClear = globalThis.clearTimeout;
+  const originalNow = Object.getOwnPropertyDescriptor(performance, "now");
+  globalThis.setTimeout = ((fn: (...args: unknown[]) => void, ms = 0, ...args: unknown[]) => {
+    const id = ++next;
+    timers.set(id, { at: now + ms, fire: () => fn(...args) });
+    return id;
+  }) as unknown as typeof setTimeout;
+  globalThis.clearTimeout = ((id: unknown) => { timers.delete(id as number); }) as typeof clearTimeout;
+  Object.defineProperty(performance, "now", { configurable: true, value: () => now });
+  return {
+    defer(fn: () => void) { disposers.push(fn); },
+    get pending() { return timers.size; },
+    elapse(ms: number) { now += ms; },
+    advance(ms: number) {
+      const target = now + ms;
+      for (;;) {
+        const due = [...timers].filter(([, t]) => t.at <= target).sort((a, b) => a[1].at - b[1].at)[0];
+        if (!due) break;
+        timers.delete(due[0]);
+        now = due[1].at;
+        due[1].fire();
+      }
+      now = target;
+    },
+    restore() {
+      try { for (const dispose of disposers.reverse()) dispose(); }
+      finally {
+        globalThis.setTimeout = originalSet;
+        globalThis.clearTimeout = originalClear;
+        if (originalNow) Object.defineProperty(performance, "now", originalNow);
+        else Reflect.deleteProperty(performance, "now");
+      }
+    },
+  };
+}
+
+type BodyClock = ReturnType<typeof bodyClock>;
+function bodyCase(name: string, run: (clock: BodyClock) => Promise<void>) {
+  test(`response body inactivity: ${name}`, async () => {
+    const clock = bodyClock();
+    try { await run(clock); } finally { clock.restore(); }
+  });
+}
+
+function bodyFixture(clock: BodyClock, onCancel?: (reason: unknown) => void | Promise<void>) {
+  let controller!: ReadableStreamDefaultController<Uint8Array>;
+  let pulls = 0;
+  const waiters: Array<{ count: number; resolve: () => void }> = [];
+  const cancelled: unknown[] = [];
+  const upstream = new AbortController();
+  clock.defer(() => upstream.abort());
+  const source = new ReadableStream<Uint8Array>({
+    start(value) { controller = value; },
+    pull() {
+      pulls += 1;
+      for (const waiter of waiters) if (pulls >= waiter.count) waiter.resolve();
+    },
+    cancel(reason) { cancelled.push(reason); return onCancel?.(reason); },
+  }, { highWaterMark: 0 });
+  return {
+    source, upstream, cancelled,
+    get controller() { return controller; },
+    get pulls() { return pulls; },
+    waitForPull(count = 1): Promise<void> {
+      return pulls >= count ? Promise.resolve() : new Promise(resolve => waiters.push({ count, resolve }));
+    },
+    guard(timeoutMs = 100, init?: ResponseInit) {
+      const guarded = guardResponseBodyInactivity(new Response(source, init), upstream.signal, timeoutMs);
+      clock.defer(guarded.dispose);
+      return guarded;
+    },
+  };
+}
+
+const bodyBytes = (text: string) => new TextEncoder().encode(text);
+async function* bodyChunks(response: Response): AsyncGenerator<Uint8Array> {
+  const reader = response.body!.getReader();
+  try {
+    for (;;) {
+      const result = await reader.read();
+      if (result.done) return;
+      yield result.value;
+    }
+  } finally { reader.releaseLock(); }
+}
+
+bodyCase("bounds headers-only silence without aborting the shared signal", async clock => {
+  const f = bodyFixture(clock);
+  const failure = assert.rejects(f.guard().response.text(), ResponseBodyInactivityError);
+  await f.waitForPull();
+  clock.advance(100);
+  await failure;
+  assert.equal(f.cancelled.length, 1);
+  assert.equal((f.cancelled[0] as Error).name, "TimeoutError");
+  assert.equal(f.source.locked, false);
+  assert.equal(f.upstream.signal.aborted, false);
+  assert.equal(clock.pending, 0);
+});
+
+bodyCase("does not prefetch or arm before a downstream read", async clock => {
+  const f = bodyFixture(clock);
+  const guard = f.guard();
+  await Promise.resolve();
+  clock.advance(10_000);
+  assert.equal(f.pulls, 0);
+  assert.equal(clock.pending, 0);
+  assert.equal(f.cancelled.length, 0);
+  guard.dispose();
+});
+
+bodyCase("preserves response metadata and exact chunk identity on clean EOF", async clock => {
+  const f = bodyFixture(clock);
+  const response = f.guard(100, { status: 206, statusText: "Partial Content", headers: { "x-test": "kept" } }).response;
+  assert.equal(response.status, 206);
+  assert.equal(response.statusText, "Partial Content");
+  assert.equal(response.headers.get("x-test"), "kept");
+  const reader = response.body!.getReader();
+  const bytes = new Uint8Array([0, 1, 255, 10]);
+  f.controller.enqueue(bytes);
+  assert.equal((await reader.read()).value, bytes);
+  f.controller.close();
+  assert.equal((await reader.read()).done, true);
+  assert.equal(f.cancelled.length, 0);
+  assert.equal(f.source.locked, false);
+  assert.equal(f.upstream.signal.aborted, false);
+  assert.equal(clock.pending, 0);
+});
+
+bodyCase("does not time out a slow consumer after a single chunk", async clock => {
+  const f = bodyFixture(clock);
+  const reader = f.guard().response.body!.getReader();
+  f.controller.enqueue(bodyBytes("first"));
+  assert.equal(new TextDecoder().decode((await reader.read()).value), "first");
+  clock.advance(10_000);
+  assert.equal(clock.pending, 0);
+  assert.equal(f.cancelled.length, 0);
+  f.controller.enqueue(bodyBytes("second"));
+  assert.equal(new TextDecoder().decode((await reader.read()).value), "second");
+  f.controller.close();
+  assert.equal((await reader.read()).done, true);
+});
+
+bodyCase("starts a fresh window for a mid-body pending read", async clock => {
+  const f = bodyFixture(clock);
+  const reader = f.guard().response.body!.getReader();
+  f.controller.enqueue(bodyBytes("first"));
+  await reader.read();
+  clock.advance(1_000);
+  const failed = assert.rejects(reader.read(), ResponseBodyInactivityError);
+  await f.waitForPull();
+  clock.advance(99);
+  assert.equal(f.cancelled.length, 0);
+  clock.advance(1);
+  await failed;
+  assert.equal(f.cancelled.length, 1);
+});
+
+bodyCase("empty chunks do not reset the pending-read window", async clock => {
+  const f = bodyFixture(clock);
+  const failure = assert.rejects(f.guard().response.text(), ResponseBodyInactivityError);
+  await f.waitForPull();
+  clock.advance(90);
+  f.controller.enqueue(new Uint8Array());
+  await f.waitForPull(2);
+  clock.advance(10);
+  await failure;
+  assert.equal(f.cancelled.length, 1);
+});
+
+bodyCase("real bytes after empty chunks pause and then reset the window", async clock => {
+  const f = bodyFixture(clock);
+  const reader = f.guard().response.body!.getReader();
+  const first = reader.read();
+  await f.waitForPull();
+  clock.advance(80);
+  f.controller.enqueue(new Uint8Array());
+  await f.waitForPull(2);
+  f.controller.enqueue(bodyBytes("progress"));
+  assert.equal(new TextDecoder().decode((await first).value), "progress");
+  clock.advance(500);
+  assert.equal(f.cancelled.length, 0);
+  const failure = assert.rejects(reader.read(), ResponseBodyInactivityError);
+  await f.waitForPull(3);
+  clock.advance(100);
+  await failure;
+});
+
+bodyCase("bounds a microtask-only producer of empty chunks", async clock => {
+  let pulls = 0;
+  let cancelled = false;
+  const source = new ReadableStream<Uint8Array>({
+    pull(controller) { pulls += 1; clock.elapse(1); controller.enqueue(new Uint8Array()); },
+    cancel() { cancelled = true; },
+  }, { highWaterMark: 0 });
+  const guard = guardResponseBodyInactivity(new Response(source), undefined, 10);
+  clock.defer(guard.dispose);
+  await assert.rejects(guard.response.text(), ResponseBodyInactivityError);
+  assert.equal(pulls, 10);
+  assert.equal(cancelled, true);
+  assert.equal(source.locked, false);
+  assert.equal(clock.pending, 0);
+});
+
+// Real timers on purpose: the point is that queued macrotasks still run while the
+// guard discards empty chunks, which a faked clock cannot observe.
+test("a microtask-only empty producer still lets queued tasks run before the deadline", async () => {
+  const source = new ReadableStream<Uint8Array>({
+    pull(controller) { controller.enqueue(new Uint8Array()); },
+  }, { highWaterMark: 0 });
+  const guard = guardResponseBodyInactivity(new Response(source), undefined, 5_000);
+  const started = performance.now();
+  let queuedTaskAt: number | undefined;
+  setTimeout(() => { queuedTaskAt = performance.now() - started; }, 0);
+  const failure = guard.response.text().then(() => undefined, (error: unknown) => error);
+  await sleep(50);
+  expect(queuedTaskAt).toBeDefined();
+  expect(queuedTaskAt!).toBeLessThan(1_000);
+  guard.dispose();
+  await failure;
+});
+
+bodyCase("consumer cancellation retains its reason and releases the reader", async clock => {
+  const f = bodyFixture(clock);
+  const guard = f.guard();
+  const reason = new Error("client cancelled");
+  await guard.response.body!.cancel(reason);
+  guard.dispose();
+  assert.deepEqual(f.cancelled, [reason]);
+  assert.equal(f.source.locked, false);
+  assert.equal(clock.pending, 0);
+});
+
+bodyCase("timeout does not await a never-settling cancellation hook", async clock => {
+  const f = bodyFixture(clock, () => new Promise<void>(() => {}));
+  const failure = assert.rejects(f.guard().response.text(), ResponseBodyInactivityError);
+  await f.waitForPull();
+  clock.advance(100);
+  await failure;
+  assert.equal(f.source.locked, false);
+  assert.equal(clock.pending, 0);
+});
+
+bodyCase("consumer cancellation does not await a never-settling cancellation hook", async clock => {
+  const f = bodyFixture(clock, () => new Promise<void>(() => {}));
+  await f.guard().response.body!.cancel("stop");
+  assert.deepEqual(f.cancelled, ["stop"]);
+  assert.equal(f.source.locked, false);
+});
+
+bodyCase("absorbs cancellation rejection without replacing the timeout", async clock => {
+  const f = bodyFixture(clock, () => Promise.reject(new Error("cleanup failed")));
+  const failure = assert.rejects(f.guard().response.text(), ResponseBodyInactivityError);
+  await f.waitForPull();
+  clock.advance(100);
+  await failure;
+  assert.equal(f.source.locked, false);
+});
+
+bodyCase("preserves upstream read errors rather than recasting them as timeouts", async clock => {
+  const f = bodyFixture(clock);
+  const error = new Error("upstream reset");
+  const failed = assert.rejects(f.guard().response.text(), candidate => candidate === error);
+  await f.waitForPull();
+  f.controller.error(error);
+  await failed;
+  assert.equal(f.source.locked, false);
+  assert.equal(clock.pending, 0);
+});
+
+bodyCase("observes an already-aborted signal before attaching a parser", async clock => {
+  const f = bodyFixture(clock);
+  const reason = new DOMException("client gone", "AbortError");
+  f.upstream.abort(reason);
+  await assert.rejects(f.guard().response.text(), candidate => candidate === reason);
+  assert.deepEqual(f.cancelled, [reason]);
+  assert.equal(f.pulls, 0);
+  assert.equal(clock.pending, 0);
+});
+
+bodyCase("external abort settles a locked pending read immediately", async clock => {
+  const f = bodyFixture(clock);
+  const reason = new Error("client gone");
+  const failed = assert.rejects(f.guard().response.text(), candidate => candidate === reason);
+  await f.waitForPull();
+  f.upstream.abort(reason);
+  await failed;
+  clock.advance(500);
+  assert.deepEqual(f.cancelled, [reason]);
+  assert.equal(f.source.locked, false);
+  assert.equal(clock.pending, 0);
+});
+
+bodyCase("external abort also works while downstream is paused", async clock => {
+  const f = bodyFixture(clock);
+  const reader = f.guard().response.body!.getReader();
+  f.controller.enqueue(bodyBytes("one"));
+  await reader.read();
+  const reason = new Error("cancel between reads");
+  f.upstream.abort(reason);
+  await assert.rejects(reader.read(), candidate => candidate === reason);
+  assert.deepEqual(f.cancelled, [reason]);
+});
+
+bodyCase("normal completion detaches the abort listener", async clock => {
+  const f = bodyFixture(clock);
+  const completed = f.guard().response.text();
+  await f.waitForPull();
+  f.controller.close();
+  assert.equal(await completed, "");
+  f.upstream.abort();
+  clock.advance(1_000);
+  assert.equal(f.cancelled.length, 0);
+  assert.equal(clock.pending, 0);
+});
+
+bodyCase("explicit disposal is idempotent and releases an unread body", async clock => {
+  const f = bodyFixture(clock, () => new Promise<void>(() => {}));
+  const guard = f.guard();
+  guard.dispose();
+  guard.dispose();
+  f.upstream.abort();
+  assert.equal(await guard.response.text(), "");
+  assert.equal(f.cancelled.length, 1);
+  assert.equal(f.source.locked, false);
+  assert.equal(clock.pending, 0);
+});
+
+bodyCase("buffered parser failure cleans up its body", async clock => {
+  const f = bodyFixture(clock);
+  const error = new Error("parser failed before reading");
+  await assert.rejects(readResponseBodyWithInactivity(
+    new Response(f.source), f.upstream.signal, 100, async () => { throw error; },
+  ), candidate => candidate === error);
+  assert.equal(f.cancelled.length, 1);
+  assert.equal(f.source.locked, false);
+});
+
+bodyCase("buffered parser early return cancels its unread remainder", async clock => {
+  const f = bodyFixture(clock);
+  assert.equal(await readResponseBodyWithInactivity(
+    new Response(f.source), f.upstream.signal, 100, async () => "early answer",
+  ), "early answer");
+  assert.equal(f.cancelled.length, 1);
+  assert.equal(f.source.locked, false);
+  assert.equal(f.upstream.signal.aborted, false);
+});
+
+bodyCase("stream parser failure cleans up even before its first read", async clock => {
+  const f = bodyFixture(clock);
+  const error = new Error("stream parser failed");
+  const stream = readResponseStreamWithInactivity(
+    new Response(f.source), f.upstream.signal, 100, async function* () { throw error; },
+  );
+  await assert.rejects(stream.next(), candidate => candidate === error);
+  assert.equal(f.cancelled.length, 1);
+  assert.equal(f.source.locked, false);
+});
+
+bodyCase("abandoning a parsed stream cleans up the original body", async clock => {
+  const f = bodyFixture(clock);
+  f.controller.enqueue(bodyBytes("one"));
+  const events = readResponseStreamWithInactivity(new Response(f.source), f.upstream.signal, 100, bodyChunks);
+  assert.equal(new TextDecoder().decode((await events.next()).value), "one");
+  await events.return(undefined);
+  assert.equal(f.cancelled.length, 1);
+  assert.equal(f.source.locked, false);
+  assert.equal(clock.pending, 0);
+});
+
+bodyCase("a headers-only continuation has its own window and leaves terminal delivery possible", async clock => {
+  const upstream = new AbortController();
+  clock.defer(() => upstream.abort());
+  assert.equal(await readResponseBodyWithInactivity(new Response("initial"), upstream.signal, 100, r => r.text()), "initial");
+  clock.advance(5_000);
+  const f = bodyFixture(clock);
+  const events = readResponseStreamWithInactivity(new Response(f.source), upstream.signal, 100, bodyChunks);
+  const order: string[] = [];
+  // A minimal enclosing relay consumes the real guarded iterator and drains a
+  // terminal before request teardown. This is not the native SSE bridge fixture.
+  const wire = new ReadableStream<Uint8Array>({
+    async start(controller) {
+      try {
+        for await (const chunk of events) controller.enqueue(chunk);
+      } catch (error) {
+        assert.ok(error instanceof ResponseBodyInactivityError);
+        assert.equal(upstream.signal.aborted, false);
+        controller.enqueue(bodyBytes("event: response.failed\ndata: {}\n\n"));
+        order.push("terminal");
+      } finally {
+        controller.close();
+        upstream.abort();
+        order.push("teardown");
+      }
+    },
+  });
+  const received = new Response(wire).text();
+  await f.waitForPull();
+  clock.advance(100);
+  assert.match(await received, /response.failed/);
+  assert.deepEqual(order, ["terminal", "teardown"]);
+  assert.equal(f.source.locked, false);
+});
+
+bodyCase("native SSE, bounded JSON and error responses retain identity without another body access", async () => {
+  for (const [status, contentType] of [[200, "Text/Event-Stream"], [200, "application/json"], [429, "text/plain"]] as const) {
+    const response = new Response("already owned", { status, headers: { "content-type": contentType } });
+    const tagged = new WeakSet([response]);
+    Object.defineProperty(response, "body", { get() { throw new Error("duplicate body access"); } });
+    const guarded = guardDirectPassthroughBodyInactivity(response, undefined, 100);
+    assert.equal(guarded, response);
+    assert.equal(tagged.has(guarded), true);
+  }
+});
+
+bodyCase("redirect bodies are bounded even with an SSE content type", async clock => {
+  const f = bodyFixture(clock);
+  const response = guardDirectPassthroughBodyInactivity(new Response(f.source, {
+    status: 302, headers: { "content-type": "text/event-stream", location: "/next" },
+  }), f.upstream.signal, 100);
+  assert.equal(response.status, 302);
+  assert.equal(response.headers.get("location"), "/next");
+  const failed = assert.rejects(response.text(), ResponseBodyInactivityError);
+  await f.waitForPull();
+  clock.advance(100);
+  await failed;
+  assert.equal(f.source.locked, false);
+});
+
+bodyCase("a plain successful passthrough body is inactivity-bounded", async clock => {
+  const f = bodyFixture(clock);
+  const response = guardDirectPassthroughBodyInactivity(new Response(f.source, {
+    status: 200, headers: { "content-type": "text/plain" },
+  }), f.upstream.signal, 100);
+  assert.notEqual(response.body, null);
+  const failed = assert.rejects(response.text(), ResponseBodyInactivityError);
+  await f.waitForPull();
+  clock.advance(100);
+  await failed;
+  assert.equal(f.cancelled.length, 1);
+  assert.equal(f.source.locked, false);
+  assert.equal(clock.pending, 0);
+});
+
+bodyCase("large deadlines do not overflow setTimeout into an immediate failure", async clock => {
+  const f = bodyFixture(clock);
+  const failed = assert.rejects(f.guard(3_000_000_000).response.text(), ResponseBodyInactivityError);
+  await f.waitForPull();
+  clock.advance(2_147_483_647);
+  assert.equal(f.cancelled.length, 0);
+  assert.equal(clock.pending, 1);
+  clock.advance(3_000_000_000 - 2_147_483_647);
+  await failed;
+});
+
+bodyCase("disabled and non-finite deadlines still preserve cancellation", async clock => {
+  for (const ms of [0, -1, Number.NaN, Number.POSITIVE_INFINITY]) {
+    const f = bodyFixture(clock);
+    const reason = new Error("cancel disabled deadline");
+    const failed = assert.rejects(f.guard(ms).response.text(), candidate => candidate === reason);
+    await f.waitForPull();
+    clock.advance(1_000);
+    assert.equal(clock.pending, 0);
+    assert.equal(f.cancelled.length, 0);
+    f.upstream.abort(reason);
+    await failed;
+  }
+});
+
+bodyCase("a bodyless response is an identity-preserving no-op", async () => {
+  const response = new Response(null, { status: 204 });
+  const guard = guardResponseBodyInactivity(response, undefined, 100);
+  assert.equal(guard.response, response);
+  guard.dispose();
+  guard.dispose();
+});
+
+bodyCase("failure is published before a source cancellation can abort other readers", async clock => {
+  const upstream = new AbortController();
+  const source = new ReadableStream<Uint8Array>({
+    cancel() { upstream.abort(new Error("synchronous teardown")); },
+  });
+  const guard = guardResponseBodyInactivity(new Response(source), upstream.signal, 100);
+  clock.defer(guard.dispose);
+  const failure = assert.rejects(guard.response.text(), ResponseBodyInactivityError);
+  // Starting the guard's pull requires one microtask; the deadline itself is deterministic.
+  await Promise.resolve();
+  clock.advance(100);
+  await failure;
+  assert.equal(source.locked, false);
+});
diff --git a/tests/responses/responses-core-modules.test.ts b/tests/responses/responses-core-modules.test.ts
index 589a6176cd2..6d92a3ab551 100644
--- a/tests/responses/responses-core-modules.test.ts
+++ b/tests/responses/responses-core-modules.test.ts
@@ -80,7 +80,15 @@ describe("Responses core module boundaries", () => {
     const ingress = readResponsesCoreModule("core.ts");
     const native = readResponsesCoreModule("passthrough-execution.ts");
     expect(ingress).toContain("return await executePassthroughResponse(");
-    expect(native).toContain("return await deliverPassthroughResponse(");
+    // Delivery is awaited inside the try, and its direct body is wrapped before
+    // the return. What matters is that both awaits stay inside the lease owner,
+    // not that the delivery call is itself the return expression.
+    expect(native).toContain("const response = await deliverPassthroughResponse(");
+    expect(native).toContain("return guardDirectPassthroughBodyInactivity(");
+    expect(native.indexOf("await deliverPassthroughResponse("))
+      .toBeLessThan(native.indexOf("return guardDirectPassthroughBodyInactivity("));
+    expect(native.indexOf("return guardDirectPassthroughBodyInactivity("))
+      .toBeLessThan(native.indexOf("} finally {"));
     expect(native.indexOf("admissionState.pendingHostAdmissionLease = null;"))
       .toBeLessThan(native.indexOf("await preparePassthroughExchange("));
     expect(native).toMatch(/finally\s*\{\s*if \(nativeHostState\.lease\)\s*\{\s*releaseUpstreamHostAdmission\(nativeHostState\.lease\);\s*releaseCodexAuthContextProbeLease\(admissionState\.authCtx\);/);
diff --git a/tests/server/terminal-guard-server.test.ts b/tests/server/terminal-guard-server.test.ts
index c2c889c3d88..e95a380ef45 100644
--- a/tests/server/terminal-guard-server.test.ts
+++ b/tests/server/terminal-guard-server.test.ts
@@ -363,6 +363,66 @@ describe("server terminal guard integration", () => {
     expect(sends).toBe(2);
   });
 
+  test("a stalled continuation body reports 504 even when cancelling it aborts the client signal", async () => {
+    // Cancelling the stalled source can disconnect the client in the same tick. The
+    // classifier has to read the thrown error first, or this timeout is reported as a
+    // client cancellation and the caller loses the upstream stall signal.
+    const stallConfig = { ...config, stallTimeoutSec: 1 } as unknown as OcxConfig;
+    const abort = new AbortController();
+    let sends = 0;
+    globalThis.fetch = (async () => {
+      sends += 1;
+      if (sends === 1) return anthropicSse(firstTurn);
+      const stalled = new ReadableStream<Uint8Array>({
+        cancel() { abort.abort(new DOMException("client disconnected", "AbortError")); },
+      });
+      return new Response(stalled, { status: 200, headers: { "content-type": "text/event-stream" } });
+    }) as typeof fetch;
+
+    const response = await handleResponses(new Request("http://localhost/v1/responses", {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({
+        model: "se-claude-opus-4.8",
+        input: "请检查这个问题并修复代码",
+        stream: true,
+        tools: [{ type: "function", name: "exec_command", description: "run a command", parameters: { type: "object" } }],
+      }),
+    }), stallConfig, { model: "", provider: "" }, { abortSignal: abort.signal });
+
+    const text = await response.text();
+    expect(sends).toBe(2);
+    expect(text).toContain("Provider continuation response body stalled before completing");
+    expect(text).not.toContain("client closed request during terminal continuation");
+  });
+
+  test("a stalled initial body fails with a 504 upstream error instead of a proxy error", async () => {
+    // The initial stream has no continuation classifier: without one the bridge catch
+    // reports this upstream timeout as a 500 proxy_error.
+    const stallConfig = { ...config, stallTimeoutSec: 1 } as unknown as OcxConfig;
+    globalThis.fetch = (async () => {
+      const stalled = new ReadableStream<Uint8Array>({});
+      return new Response(stalled, { status: 200, headers: { "content-type": "text/event-stream" } });
+    }) as typeof fetch;
+
+    const response = await handleResponses(new Request("http://localhost/v1/responses", {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({
+        model: "se-claude-opus-4.8",
+        input: "请检查这个问题并修复代码",
+        stream: true,
+        tools: [{ type: "function", name: "exec_command", description: "run a command", parameters: { type: "object" } }],
+      }),
+    }), stallConfig, { model: "", provider: "" });
+
+    const text = await response.text();
+    expect(text.match(/event: response\.failed/g)?.length).toBe(1);
+    expect(text).toContain("Upstream response body stalled before completing");
+    expect(text).toContain('"type":"upstream_error"');
+    expect(text).not.toContain("proxy_error");
+  });
+
   test("terminal-guard 429 wait longer than the stall budget still succeeds (heartbeats)", async () => {
     const stallConfig = {
       ...config,