From 3e080877f2cb7a5e75426b0586e23a9bfa1efe95 Mon Sep 17 00:00:00 2001 From: initguru Date: Thu, 17 Sep 2026 22:44:48 +0900 Subject: [PATCH] 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 | 4 +- .../fixes/12913-stream-active-timeout.md | 1 + docs/reference/ENVIRONMENT.md | 5 +- open-sse/config/constants.ts | 5 + open-sse/utils/streamHandler.ts | 120 ++++++++++-------- src/shared/utils/runtimeTimeouts.ts | 22 ++++ tests/unit/runtime-timeouts.test.ts | 28 ++++ ...ctive-timeout-covers-model-budgets.test.ts | 60 +++++++++ tests/unit/stream-handler.test.ts | 67 ++++++++++ 9 files changed, 259 insertions(+), 53 deletions(-) create mode 100644 changelog.d/fixes/12913-stream-active-timeout.md create mode 100644 tests/unit/stream-active-timeout-covers-model-budgets.test.ts diff --git a/.env.example b/.env.example index 32ad544497..2aa3eee4ef 100644 --- a/.env.example +++ b/.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). diff --git a/changelog.d/fixes/12913-stream-active-timeout.md b/changelog.d/fixes/12913-stream-active-timeout.md new file mode 100644 index 0000000000..145fd85da4 --- /dev/null +++ b/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) diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index 2ada70c885..ddce801335 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/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) | diff --git a/open-sse/config/constants.ts b/open-sse/config/constants.ts index ece185e129..82aeb0f3ba 100644 --- a/open-sse/config/constants.ts +++ b/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. diff --git a/open-sse/utils/streamHandler.ts b/open-sse/utils/streamHandler.ts index 8499a95b80..3227eac886 100644 --- a/open-sse/utils/streamHandler.ts +++ b/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, 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 | 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 | 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 | 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(); }, }); diff --git a/src/shared/utils/runtimeTimeouts.ts b/src/shared/utils/runtimeTimeouts.ts index 74c7c9e96c..8ef930050a 100644 --- a/src/shared/utils/runtimeTimeouts.ts +++ b/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, diff --git a/tests/unit/runtime-timeouts.test.ts b/tests/unit/runtime-timeouts.test.ts index 3e1af2a5a2..d50f131c4d 100644 --- a/tests/unit/runtime-timeouts.test.ts +++ b/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). diff --git a/tests/unit/stream-active-timeout-covers-model-budgets.test.ts b/tests/unit/stream-active-timeout-covers-model-budgets.test.ts new file mode 100644 index 0000000000..1f58d86eab --- /dev/null +++ b/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" + ); +}); diff --git a/tests/unit/stream-handler.test.ts b/tests/unit/stream-handler.test.ts index 3f2f988e4c..9d1c1e56c0 100644 --- a/tests/unit/stream-handler.test.ts +++ b/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 | 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.