fix(sse): bound active streams without terminal events (#12913)

* fix(sse): bound active streams without terminal events

* fix(sse): derive the active-stream ceiling from the largest registered model budget

The watchdog is a hard lifetime cap that never resets on bytes, so a flat
15-minute default killed models the registry already allows to run for 20
minutes (the Codex entries declare timeoutMs: 1_200_000). The default is now
that maximum plus a one-minute margin, and a new test re-derives the maximum
from the registry so a future larger budget fails the gate instead of silently
re-opening the bug.

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

---------

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

.env.example

@@ -1523,7 +1523,7 @@ CURSOR_USER_AGENT="Cursor/3.4"
 #
 # Hierarchy: REQUEST_TIMEOUT_MS acts as a global override.
 # If set, it becomes the default for FETCH_TIMEOUT_MS, STREAM_IDLE_TIMEOUT_MS,
-# and STREAM_READINESS_TIMEOUT_MS.
+# and STREAM_READINESS_TIMEOUT_MS. STREAM_ACTIVE_TIMEOUT_MS is independent.
 # The fine-grained variables below override their respective defaults only when set.
 
 # ── Global shortcut ──
@@ -1705,6 +1705,8 @@ CURSOR_USER_AGENT="Cursor/3.4"
 # ── Stream idle detection ──
 # STREAM_IDLE_TIMEOUT_MS=600000      # Max silence between SSE chunks (default: 600000)
 #                                     # Extended-thinking models rarely pause >90s.
+# STREAM_ACTIVE_TIMEOUT_MS=1260000    # Max total active SSE lifetime (default: 21 min = the largest registered model timeoutMs + 1 min; 0 disables)
+#                                     # Independent of REQUEST_TIMEOUT_MS and byte activity.
 # STREAM_READINESS_TIMEOUT_MS=80000  # Time to receive the first non-ping SSE event
 # STREAM_READINESS_MAX_TIMEOUT_MS=180000 # Cap for adaptive first-event extensions
 #                                         # (large/tool-heavy/high-reasoning requests).

changelog.d/fixes/12913-stream-active-timeout.md

@@ -0,0 +1 @@
+- fix(sse): bound streams that keep sending raw upstream bytes forever without ever emitting a terminal event — a new `STREAM_ACTIVE_TIMEOUT_MS` watchdog (default 1260000ms/21min — the largest per-model `timeoutMs` in the registry plus a one-minute margin, `0` disables) tracks the stream's total lifetime independently of the existing byte-stall watchdog, so a continuously-active non-terminal stream can no longer occupy a connection indefinitely (#12913)

docs/reference/ENVIRONMENT.md

@@ -748,6 +748,7 @@ REQUEST_TIMEOUT_MS (global override)
 │   ├── FETCH_CONNECT_TIMEOUT_MS (independent, default: 30000)
 │   └── FETCH_KEEPALIVE_TIMEOUT_MS (independent, default: 4000)
 ├─→ STREAM_IDLE_TIMEOUT_MS (inherits from REQUEST_TIMEOUT_MS, default: 600000)
+├─→ STREAM_ACTIVE_TIMEOUT_MS (independent, default: 1260000; 0 disables)
 ├─→ STREAM_READINESS_TIMEOUT_MS (inherits from REQUEST_TIMEOUT_MS, default: 80000)
 ├─→ STREAM_READINESS_MAX_TIMEOUT_MS (caps adaptive readiness extensions, default: 180000)
 └─→ API_BRIDGE_PROXY_TIMEOUT_MS (inherits from REQUEST_TIMEOUT_MS, default: 30000)
@@ -761,7 +762,8 @@ REQUEST_TIMEOUT_MS (global override)
 | ----------------------------------------- | -------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | `REQUEST_TIMEOUT_MS`                      | _(unset)_            | Global shortcut — overrides both `FETCH_TIMEOUT_MS` and `STREAM_IDLE_TIMEOUT_MS` defaults.                                                                      |
 | `FETCH_TIMEOUT_MS`                        | `600000`             | Total HTTP request timeout for upstream provider calls.                                                                                                         |
-| `STREAM_IDLE_TIMEOUT_MS`                  | `600000`             | Max silence between SSE chunks before aborting. Extended-thinking models rarely pause >90s.                                                                     |
+| `STREAM_IDLE_TIMEOUT_MS`                  | `600000`             | Max silence between raw upstream bytes before aborting. Extended-thinking models rarely pause >90s.                                                               |
+| `STREAM_ACTIVE_TIMEOUT_MS`                | `1260000`            | Maximum total active SSE stream lifetime; never resets on upstream bytes and is independent of `REQUEST_TIMEOUT_MS`. Derived from the largest per-model `timeoutMs` in the registry (1200000, Codex) plus a 60000 margin, so a model allowed to run its full budget is never killed mid-answer. Set to `0` to disable. |
 | `OMNIROUTE_SSE_COMMENTS`                  | _(disabled)_         | Whether OmniRoute may emit SSE `:` comment lines (e.g. the `: keepalive` heartbeat and `x-omniroute-*` metadata trailers). Disabled by default (#10524) since strict OpenAI-compatible clients JSON.parse every SSE line and crash on `:` comments; `data:` heartbeats are unaffected. Set `on`/`true`/`1`/`yes` to opt back in. Used by `open-sse/utils/sseHeartbeat.ts`.                |
 | `STREAM_READINESS_TIMEOUT_MS`             | `80000`              | Time to receive the first non-ping SSE event. Inherits `REQUEST_TIMEOUT_MS` when set.                                                                           |
 | `STREAM_READINESS_MAX_TIMEOUT_MS`         | `180000`             | Maximum adaptive first-event readiness window for large, tool-heavy, or high-reasoning streaming requests.                                                       |
@@ -855,6 +857,7 @@ Provider-level circuit breaker tuning. Defaults reflect the scaled values used s
 | Scenario                         | Configuration                                          |
 | -------------------------------- | ------------------------------------------------------ |
 | **Long-running code generation** | `REQUEST_TIMEOUT_MS=900000` (15 min)                   |
+| **Bound total stream lifetime** | `STREAM_ACTIVE_TIMEOUT_MS=1260000` (21 min)            |
 | **Fast-fail for production API** | `API_BRIDGE_PROXY_TIMEOUT_MS=10000`                    |
 | **Extended thinking models**     | `STREAM_IDLE_TIMEOUT_MS=300000` (5 min between chunks) |
 

open-sse/config/constants.ts

@@ -28,6 +28,11 @@ export const STREAM_IDLE_TIMEOUT_MS = upstreamTimeouts.streamIdleTimeoutMs;
 // immediate-fail behavior.
 export const STREAM_DISCONNECT_GRACE_PERIOD_MS = upstreamTimeouts.streamDisconnectGracePeriodMs;
 
+// Hard cap for a connected upstream stream. This timer never resets on
+// upstream byte activity and is independent of REQUEST_TIMEOUT_MS. Set
+// STREAM_ACTIVE_TIMEOUT_MS=0 to disable it.
+export const STREAM_ACTIVE_TIMEOUT_MS = upstreamTimeouts.streamActiveTimeoutMs;
+
 // Timeout for the first non-ping SSE event. Inherits REQUEST_TIMEOUT_MS when
 // set, unless STREAM_READINESS_TIMEOUT_MS is specified directly. This must stay
 // conservative for large prompts and slow first-byte reasoning providers.

open-sse/utils/streamHandler.ts

@@ -1,5 +1,5 @@
 import { trackPendingRequest } from "@/lib/usageDb";
-import { STREAM_IDLE_TIMEOUT_MS } from "../config/constants.ts";
+import { STREAM_ACTIVE_TIMEOUT_MS, STREAM_IDLE_TIMEOUT_MS } from "../config/constants.ts";
 import { FORMATS } from "../translator/formats.ts";
 import { buildErrorBody } from "./error.ts";
 import { PENDING_REQUEST_CLEARED_MARKER } from "./stream.ts";
@@ -842,7 +842,9 @@ export function createDisconnectAwareStream(
  * output for long stretches while partial EventStream frames keep arriving;
  * measuring stall on the transform output caused false stalls. Any upstream
  * chunk resets the timer. If no bytes arrive for `stallTimeoutMs`, the
- * stream surfaces a "stream stall timeout" error and aborts.
+ * stream surfaces a "stream stall timeout" error and aborts. A separate
+ * active lifetime budget never resets on bytes and surfaces
+ * "stream active timeout" when exceeded.
  *
  * Ported from decolua/9router#1243 by @zakirkun.
  *
@@ -851,15 +853,25 @@ export function createDisconnectAwareStream(
  * @param streamController - Stream controller from createStreamController
  * @param opts.stallTimeoutMs - Override the stall budget (defaults to
  *   STREAM_IDLE_TIMEOUT_MS / DEFAULT_STREAM_STALL_TIMEOUT_MS). `0` disables
- *   the watchdog.
+ *   the stall watchdog.
+ * @param opts.activeTimeoutMs - Override the total active-stream budget
+ *   (defaults to STREAM_ACTIVE_TIMEOUT_MS). `0` disables the active watchdog.
  */
 export function pipeWithDisconnect(
   providerResponse: Response,
   transformStream: TransformStream<Uint8Array, Uint8Array>,
   streamController: StreamController,
-  opts: { stallTimeoutMs?: number; contentStallTimeoutMs?: number; highWaterMark?: number } = {}
+  opts: {
+    stallTimeoutMs?: number;
+    activeTimeoutMs?: number;
+    contentStallTimeoutMs?: number;
+    highWaterMark?: number;
+  } = {}
 ) {
   const stallTimeoutMs = opts.stallTimeoutMs ?? DEFAULT_STREAM_STALL_TIMEOUT_MS;
+  const activeTimeoutMs = opts.activeTimeoutMs ?? STREAM_ACTIVE_TIMEOUT_MS;
+  const stallEnabled = Number.isFinite(stallTimeoutMs) && stallTimeoutMs > 0;
+  const activeEnabled = Number.isFinite(activeTimeoutMs) && activeTimeoutMs > 0;
   // Disabled unless a caller opts in with an explicit budget (chatCore wires
   // the adaptive streamReadinessPolicy.timeoutMs — see its own doc comment).
   // No blanket default here: an arbitrary constant picked at this layer,
@@ -869,7 +881,7 @@ export function pipeWithDisconnect(
   const contentStallTimeoutMs = opts.contentStallTimeoutMs ?? 0;
 
   // Watchdogs disabled — preserve legacy behavior verbatim.
-  if ((!stallTimeoutMs || stallTimeoutMs <= 0) && contentStallTimeoutMs <= 0) {
+  if (!stallEnabled && !activeEnabled && contentStallTimeoutMs <= 0) {
     const transformedBody = providerResponse.body.pipeThrough(transformStream);
     return createDisconnectAwareStream(
       { readable: transformedBody, writable: createNoopAbortWritable() },
@@ -879,55 +891,60 @@ export function pipeWithDisconnect(
   }
 
   let stallTimer: ReturnType<typeof setTimeout> | null = null;
-  // Captured on the upstream tap's `start`, used by the watchdog to error the
-  // pipeline so the downstream reader unblocks and emits a clean SSE error
-  // event. Without this, aborting the AbortController alone does not unblock
-  // a `reader.read()` already suspended on the transform pipe — the request
-  // would hang until the upstream finally closed the socket.
+  let activeTimer: ReturnType<typeof setTimeout> | null = null;
+  // Erroring the upstream tap unblocks a downstream reader suspended on the
+  // transform pipe; aborting the controller alone does not always do that.
   let upstreamTapController: TransformStreamDefaultController<Uint8Array> | null = null;
-  // Set when the watchdog fires so the downstream pull() catch (which sees
-  // the same error propagated through the pipeline) does not call
-  // handleError a second time — pending-cleanup is idempotent but onError
-  // callbacks should fire once per error.
   let stallFired = false;
+  let activeFired = false;
 
   const clearStall = () => {
-    if (stallTimer) {
-      clearTimeout(stallTimer);
-      stallTimer = null;
+    if (stallTimer) clearTimeout(stallTimer);
+    stallTimer = null;
+  };
+  const clearActive = () => {
+    if (activeTimer) clearTimeout(activeTimer);
+    activeTimer = null;
+  };
+  const clearWatchdogs = () => {
+    clearStall();
+    clearActive();
+  };
+  const triggerWatchdog = (kind: "stall" | "active") => {
+    if (stallFired || activeFired) return;
+    const message = kind === "active" ? "stream active timeout" : "stream stall timeout";
+    if (kind === "active") activeFired = true;
+    else stallFired = true;
+    clearWatchdogs();
+    const error = new Error(message);
+    try {
+      streamController.handleError?.(error);
+    } catch (e) {
+      console.debug(`[STREAM-HANDLER] ${kind} watchdog handleError failed:`, e);
+    }
+    try {
+      upstreamTapController?.error(error);
+    } catch (e) {
+      console.debug(`[STREAM-HANDLER] ${kind} watchdog upstream tap error failed:`, e);
+    }
+    try {
+      streamController.abort?.();
+    } catch (e) {
+      console.debug(`[STREAM-HANDLER] ${kind} watchdog abort failed:`, e);
     }
   };
   const armStall = () => {
-    if (!stallTimeoutMs || stallTimeoutMs <= 0) return;
+    if (!stallEnabled) return;
     clearStall();
-    stallTimer = setTimeout(() => {
-      stallTimer = null;
-      stallFired = true;
-      const stallError = new Error("stream stall timeout");
-      // Notify the controller (onError callback + pending-request cleanup).
-      try {
-        streamController.handleError?.(stallError);
-      } catch (e) {
-        console.debug(`[STREAM-HANDLER] stall watchdog handleError failed:`, e);
-      }
-      // Error the pipeline so the downstream reader unblocks. createDisconnect-
-      // AwareStream's catch block translates this into buildStreamErrorChunks
-      // (sanitized SSE error event with finish_reason:"error", per the format).
-      try {
-        upstreamTapController?.error(stallError);
-      } catch (e) {
-        console.debug(`[STREAM-HANDLER] stall watchdog upstream tap error failed:`, e);
-      }
-      // Abort the underlying fetch so upstream releases the connection.
-      try {
-        streamController.abort?.();
-      } catch (e) {
-        console.debug(`[STREAM-HANDLER] stall watchdog abort failed:`, e);
-      }
-    }, stallTimeoutMs);
+    stallTimer = setTimeout(() => triggerWatchdog("stall"), stallTimeoutMs);
+  };
+  const armActive = () => {
+    if (!activeEnabled) return;
+    clearActive();
+    activeTimer = setTimeout(() => triggerWatchdog("active"), activeTimeoutMs);
   };
 
-  // Second, independent watchdog: fires when the upstream keeps sending raw
+  // Third, independent watchdog: fires when the upstream keeps sending raw
   // bytes (so armStall() above keeps resetting and never fires) but none of
   // them ever carry real model output — only lifecycle/ping frames
   // (OpenAI Responses response.in_progress/response.created, bare
@@ -985,32 +1002,32 @@ export function pipeWithDisconnect(
     }, contentStallTimeoutMs);
   };
 
-  // Wrap controller so every termination path clears both stall timers.
+  // Wrap controller so every termination path clears all watchdog timers.
   // Without this, abort/complete/error/disconnect paths leave a timer armed
   // and a stale abort could fire after the request has already ended.
   const wrappedController: StreamController = {
     ...streamController,
     handleComplete: () => {
-      clearStall();
+      clearWatchdogs();
       clearContentStall();
       streamController.handleComplete();
     },
     handleError: (e: unknown) => {
-      clearStall();
+      clearWatchdogs();
       clearContentStall();
       // A watchdog already fired its own handleError — the inner pull()
       // catch sees the same error propagated through the pipeline; suppress
       // the duplicate to keep onError callbacks single-fire.
-      if (stallFired || contentStallFired) return;
+      if (stallFired || activeFired || contentStallFired) return;
       streamController.handleError(e);
     },
     handleDisconnect: (reason?: string) => {
-      clearStall();
+      clearWatchdogs();
       clearContentStall();
       streamController.handleDisconnect(reason);
     },
     abort: () => {
-      clearStall();
+      clearWatchdogs();
       clearContentStall();
       streamController.abort();
     },
@@ -1025,6 +1042,7 @@ export function pipeWithDisconnect(
     start(controller) {
       upstreamTapController = controller;
       armStall();
+      armActive();
       armContentStall();
     },
     transform(chunk, controller) {
@@ -1036,7 +1054,7 @@ export function pipeWithDisconnect(
       controller.enqueue(chunk);
     },
     flush() {
-      clearStall();
+      clearWatchdogs();
       clearContentStall();
     },
   });

src/shared/utils/runtimeTimeouts.ts

@@ -8,6 +8,17 @@ type ReadTimeoutOptions = {
 
 export const DEFAULT_FETCH_TIMEOUT_MS = 600_000;
 export const DEFAULT_STREAM_IDLE_TIMEOUT_MS = 600_000;
+// Hard cap on a connected stream's TOTAL lifetime (it never resets on bytes,
+// unlike STREAM_IDLE_TIMEOUT_MS). It must therefore stay ABOVE the largest
+// per-model `timeoutMs` any provider registers, or a model that is allowed to
+// run for its full budget would be killed mid-answer by this watchdog. The
+// current maximum registered budget is 1_200_000ms (20 min, the Codex models
+// in open-sse/config/providers/registry/codex), so the default is that value
+// plus a one-minute margin. tests/unit/stream-active-timeout-covers-model-budgets.test.ts
+// re-derives the maximum from the registry and fails if this constant ever
+// falls below it again (#12913).
+export const MAX_REGISTERED_MODEL_TIMEOUT_MARGIN_MS = 60_000;
+export const DEFAULT_STREAM_ACTIVE_TIMEOUT_MS = 1_260_000;
 export const MAX_TIMER_TIMEOUT_MS = 2_147_483_647;
 export const DEFAULT_SSE_HEARTBEAT_INTERVAL_MS = 15_000;
 export const DEFAULT_STREAM_READINESS_TIMEOUT_MS = 80_000;
@@ -57,6 +68,7 @@ function hasEnvValue(env: EnvSource, name: string): boolean {
 export type UpstreamTimeoutConfig = {
   fetchTimeoutMs: number;
   streamIdleTimeoutMs: number;
+  streamActiveTimeoutMs: number;
   sseHeartbeatIntervalMs: number;
   streamReadinessTimeoutMs: number;
   streamReadinessMaxTimeoutMs: number;
@@ -131,6 +143,15 @@ export function getUpstreamTimeoutConfig(
       logger,
     }
   );
+  const streamActiveTimeoutMs = readTimeoutMs(
+    env,
+    "STREAM_ACTIVE_TIMEOUT_MS",
+    DEFAULT_STREAM_ACTIVE_TIMEOUT_MS,
+    {
+      allowZero: true,
+      logger,
+    }
+  );
   const streamReadinessTimeoutMs = readTimeoutMs(
     env,
     "STREAM_READINESS_TIMEOUT_MS",
@@ -171,6 +192,7 @@ export function getUpstreamTimeoutConfig(
   return {
     fetchTimeoutMs,
     streamIdleTimeoutMs,
+    streamActiveTimeoutMs,
     streamReadinessTimeoutMs,
     streamReadinessMaxTimeoutMs,
     sseHeartbeatIntervalMs,

tests/unit/runtime-timeouts.test.ts

@@ -12,6 +12,7 @@ test("upstream timeout config derives hidden fetch timeouts from FETCH_TIMEOUT_M
   assert.deepEqual(config, {
     fetchTimeoutMs: 600000,
     streamIdleTimeoutMs: 600000,
+    streamActiveTimeoutMs: 1260000,
     sseHeartbeatIntervalMs: 15000,
     streamReadinessTimeoutMs: 80000,
     streamReadinessMaxTimeoutMs: 180000,
@@ -104,6 +105,33 @@ test("API bridge timeouts align request timeout with long proxy timeout by defau
   });
 });
 
+test("active stream timeout is independent from REQUEST_TIMEOUT_MS and validates its own setting", () => {
+  assert.equal(runtimeTimeouts.DEFAULT_STREAM_ACTIVE_TIMEOUT_MS, 1_260_000);
+  assert.equal(runtimeTimeouts.getUpstreamTimeoutConfig({}).streamActiveTimeoutMs, 1_260_000);
+  assert.equal(
+    runtimeTimeouts.getUpstreamTimeoutConfig({ REQUEST_TIMEOUT_MS: "25000" }).streamActiveTimeoutMs,
+    1_260_000
+  );
+  assert.equal(
+    runtimeTimeouts.getUpstreamTimeoutConfig({ STREAM_ACTIVE_TIMEOUT_MS: "120000" })
+      .streamActiveTimeoutMs,
+    120_000
+  );
+  assert.equal(
+    runtimeTimeouts.getUpstreamTimeoutConfig({ STREAM_ACTIVE_TIMEOUT_MS: "0" })
+      .streamActiveTimeoutMs,
+    0
+  );
+
+  for (const value of ["-1", "NaN", "Infinity"]) {
+    assert.equal(
+      runtimeTimeouts.getUpstreamTimeoutConfig({ STREAM_ACTIVE_TIMEOUT_MS: value })
+        .streamActiveTimeoutMs,
+      1_260_000
+    );
+  }
+});
+
 test("idle timeout default stays at 10min (600_000) for slow-thinking model safety", () => {
   // NOTE: PR #2233 originally lowered this to 300_000, but the reviewer asked to keep
   // the legacy default (slow thinking models, long Anthropic extended-thinking runs).

tests/unit/stream-active-timeout-covers-model-budgets.test.ts

@@ -0,0 +1,60 @@
+/**
+ * #12913 — the active-stream watchdog is a HARD lifetime cap: it never resets on
+ * upstream bytes. If its default ever drops below the largest per-model
+ * `timeoutMs` the provider registry declares, a model that is allowed to run for
+ * its full budget gets killed mid-answer by the watchdog instead of finishing.
+ *
+ * This test re-derives the maximum registered budget from the registry source and
+ * fails when the default stops covering it, so a future `timeoutMs: 1_800_000`
+ * entry cannot silently re-open the bug.
+ */
+import test from "node:test";
+import assert from "node:assert/strict";
+import { readdirSync, readFileSync, statSync } from "node:fs";
+import { join } from "node:path";
+
+import {
+  DEFAULT_STREAM_ACTIVE_TIMEOUT_MS,
+  MAX_REGISTERED_MODEL_TIMEOUT_MARGIN_MS,
+} from "../../src/shared/utils/runtimeTimeouts.ts";
+
+const REGISTRY_ROOT = join(process.cwd(), "open-sse/config/providers/registry");
+const TIMEOUT_LITERAL = /timeoutMs:\s*([0-9_]+)/g;
+
+function collectTsFiles(dir: string, out: string[] = []): string[] {
+  for (const entry of readdirSync(dir)) {
+    const full = join(dir, entry);
+    if (statSync(full).isDirectory()) collectTsFiles(full, out);
+    else if (full.endsWith(".ts")) out.push(full);
+  }
+  return out;
+}
+
+function maxRegisteredTimeoutMs(): { value: number; file: string } {
+  let best = { value: 0, file: "" };
+  for (const file of collectTsFiles(REGISTRY_ROOT)) {
+    const source = readFileSync(file, "utf8");
+    for (const match of source.matchAll(TIMEOUT_LITERAL)) {
+      const value = Number(match[1].replaceAll("_", ""));
+      if (Number.isFinite(value) && value > best.value) best = { value, file };
+    }
+  }
+  return best;
+}
+
+test("the default active-stream budget covers every per-model timeoutMs in the registry", () => {
+  const max = maxRegisteredTimeoutMs();
+
+  assert.ok(max.value > 0, "expected at least one timeoutMs literal in the provider registry");
+  assert.ok(
+    DEFAULT_STREAM_ACTIVE_TIMEOUT_MS >= max.value,
+    `DEFAULT_STREAM_ACTIVE_TIMEOUT_MS (${DEFAULT_STREAM_ACTIVE_TIMEOUT_MS}) must stay >= the largest ` +
+      `registered model timeoutMs (${max.value}, declared in ${max.file}); otherwise the active ` +
+      `watchdog kills a model that is still inside its own budget.`
+  );
+  assert.equal(
+    DEFAULT_STREAM_ACTIVE_TIMEOUT_MS,
+    max.value + MAX_REGISTERED_MODEL_TIMEOUT_MARGIN_MS,
+    "the default is documented as the largest registered budget plus the margin — update both together"
+  );
+});

tests/unit/stream-handler.test.ts

@@ -767,6 +767,73 @@ test("pipeWithDisconnect does NOT flag a slow but progressing upstream as stalle
   assert.doesNotMatch(text, /"finish_reason":"error"/);
 });
 
+test("pipeWithDisconnect caps a continuously active non-terminal stream", async () => {
+  let interval: ReturnType<typeof setInterval> | null = null;
+  let upstreamCancelled = false;
+  const source = new ReadableStream({
+    start(controller) {
+      interval = setInterval(() => controller.enqueue(encoder.encode("delta")), 10);
+    },
+    cancel() {
+      upstreamCancelled = true;
+      if (interval) clearInterval(interval);
+    },
+  });
+
+  let onErrorCalls = 0;
+  let onErrorMessage = "";
+  const streamController = createStreamController({
+    onError(event) {
+      onErrorCalls += 1;
+      onErrorMessage = event.message;
+      return true;
+    },
+  });
+
+  const stream = pipeWithDisconnect(new Response(source), new TransformStream(), streamController, {
+    stallTimeoutMs: 150,
+    activeTimeoutMs: 70,
+  });
+  const text = await readStreamText(stream);
+
+  assert.equal(onErrorCalls, 1);
+  assert.equal(onErrorMessage, "stream active timeout");
+  assert.equal(streamController.signal.aborted, true);
+  assert.equal(upstreamCancelled, true);
+  assert.match(text, /stream active timeout/);
+  assert.match(text, /"finish_reason":"error"/);
+  assert.match(text, /\[DONE\]/);
+});
+
+test("pipeWithDisconnect allows a completed stream with active timeout disabled", async () => {
+  const source = new ReadableStream({
+    async start(controller) {
+      controller.enqueue(encoder.encode("a"));
+      await new Promise((resolve) => setTimeout(resolve, 30));
+      controller.enqueue(encoder.encode("b"));
+      await new Promise((resolve) => setTimeout(resolve, 30));
+      controller.close();
+    },
+  });
+  let onErrorCalled = false;
+  const streamController = createStreamController({
+    onError() {
+      onErrorCalled = true;
+      return true;
+    },
+  });
+
+  const text = await readStreamText(
+    pipeWithDisconnect(new Response(source), new TransformStream(), streamController, {
+      stallTimeoutMs: 120,
+      activeTimeoutMs: 0,
+    })
+  );
+
+  assert.equal(text, "ab");
+  assert.equal(onErrorCalled, false);
+});
+
 test("pipeWithDisconnect flags a truly stalled upstream (no bytes for the full stall budget)", async () => {
   // Upstream emits one byte and then goes silent forever. Stall budget is
   // 80ms — the watchdog must fire and surface a stream-stall error.