From a2c78cf53c6f7005ec4368e1926f881409ffff7d Mon Sep 17 00:00:00 2001 From: Luke Parke <5702154+LukasParke@users.noreply.github.com> Date: Fri, 7 Aug 2026 23:26:22 -0500 Subject: [PATCH] feat: add stream watchdog for stalled-stream fail-fast detection (DEV-723 phase 1) A streaming response can return headers quickly, emit only keep-alive framing or metadata events (response.created, empty message shells), and then stall indefinitely. Transport-level timeouts cannot catch this because keep-alives reset socket idle timers without any content flowing. This adds the self-contained detection layer: - src/lib/stream-errors.ts: StreamStalledError with phase ('first_content' | 'between_content'), timeoutMs, elapsedMs, receivedAnyContent, and a retryable getter (true only when no content was received, so a retry cannot duplicate output). - src/lib/stream-watchdog.ts: applyStreamWatchdog wraps a parsed event stream with two opt-in semantic deadlines: firstContentMs (stream start -> first content-bearing event) and contentIntervalMs (max gap between content events). Classification is based on parsed event types, mirroring the isOutputChunkType approach from vercel/ai#17315: status events and item shells neither satisfy nor reset deadlines; terminal events (response.completed/failed/incomplete/error) disarm the watchdog permanently. On expiry the wrapped stream errors and the source is cancelled. applyResponsesStreamWatchdog packages the standard OpenResponses event classification. Unset timeouts mean no watchdog and the source stream is returned unchanged, so this is fully non-breaking. callModel/raw-stream integration lands in phase 2. --- src/lib/stream-errors.ts | 69 ++++ src/lib/stream-watchdog.ts | 295 ++++++++++++++++ tests/unit/stream-watchdog.test.ts | 530 +++++++++++++++++++++++++++++ 3 files changed, 894 insertions(+) create mode 100644 src/lib/stream-errors.ts create mode 100644 src/lib/stream-watchdog.ts create mode 100644 tests/unit/stream-watchdog.test.ts diff --git a/src/lib/stream-errors.ts b/src/lib/stream-errors.ts new file mode 100644 index 000000000..117ecc0ac --- /dev/null +++ b/src/lib/stream-errors.ts @@ -0,0 +1,69 @@ +/** + * Errors for client-side stalled-stream detection (DEV-723). + * + * A streaming response can return headers quickly and then never emit a + * content-bearing event (a "stalled stream"). The stream watchdog + * (`stream-watchdog.ts`) enforces the configured deadlines and raises + * `StreamStalledError` when one expires. + */ + +/** + * Which watchdog deadline expired. + * + * - `first_content`: the response stream started (headers received) but no + * content-bearing event arrived within `firstContentMs`. + * - `between_content`: content started flowing, but the gap since the last + * content-bearing event exceeded `contentIntervalMs`. + */ +export type StreamStallPhase = 'first_content' | 'between_content'; + +/** + * Raised when a streaming response stalls: the connection stays open (and may + * even receive keep-alive framing) but no content-bearing event arrives within + * the configured deadline. + */ +export class StreamStalledError extends Error { + override readonly name = 'StreamStalledError'; + + /** Which deadline expired. */ + readonly phase: StreamStallPhase; + /** The configured deadline that expired, in milliseconds. */ + readonly timeoutMs: number; + /** + * Milliseconds elapsed since the deadline was armed (stream start for + * `first_content`, the last content-bearing event for `between_content`). + */ + readonly elapsedMs: number; + /** Whether any content-bearing event was received before the stall. */ + readonly receivedAnyContent: boolean; + + constructor(options: { + phase: StreamStallPhase; + timeoutMs: number; + elapsedMs: number; + receivedAnyContent: boolean; + }) { + const message = + options.phase === 'first_content' + ? `Stream stalled: no content received within ${options.timeoutMs}ms of the response stream starting` + : `Stream stalled: no content received for ${options.timeoutMs}ms after the last content event`; + super(message); + + this.phase = options.phase; + this.timeoutMs = options.timeoutMs; + this.elapsedMs = options.elapsedMs; + this.receivedAnyContent = options.receivedAnyContent; + + // In older runtimes the prototype chain is not set up correctly by + // super() calls on subclasses of built-ins. + Object.setPrototypeOf(this, StreamStalledError.prototype); + } + + /** + * Whether the request is safe to retry without risking duplicated output. + * Only true when the stall happened before any content was received. + */ + get retryable(): boolean { + return !this.receivedAnyContent; + } +} diff --git a/src/lib/stream-watchdog.ts b/src/lib/stream-watchdog.ts new file mode 100644 index 000000000..f3e071a5e --- /dev/null +++ b/src/lib/stream-watchdog.ts @@ -0,0 +1,295 @@ +/** + * Client-side stalled-stream detection (DEV-723). + * + * A streaming response can return headers quickly, emit only keep-alive + * framing or metadata events, and then stall indefinitely without producing + * content. Transport-level timeouts cannot catch this: keep-alive SSE + * comments reset socket idle timers even though no content is flowing. + * + * The watchdog wraps a parsed event stream and enforces two semantic, + * opt-in deadlines: + * + * - `firstContentMs` — armed when the stream starts; satisfied (and + * permanently disarmed) by the first content-bearing event. + * - `contentIntervalMs` — after content has started, the maximum gap + * allowed between content-bearing events. + * + * Classification follows the parsed event's type, not transport activity: + * empty role preludes (`response.output_item.added` for message shells), + * status events (`response.created`, `response.in_progress`, ...), and + * SSE keep-alive comments (which the SSE parser drops before events reach + * this layer) neither satisfy nor reset a deadline. + * + * When a deadline expires, the wrapped stream errors with + * {@link StreamStalledError} and the source stream is cancelled. + */ + +import type * as models from '../models/index.js'; + +import { StreamStalledError, type StreamStallPhase } from './stream-errors.js'; + +/** + * Opt-in stream stall timeouts. Both are unset by default (no watchdog). + * Non-finite or non-positive values disable the corresponding deadline. + */ +export type StreamTimeoutOptions = { + /** + * Maximum milliseconds between the response stream starting (headers + * received, body stream available) and the first content-bearing event. + * Unset by default. + */ + firstContentMs?: number | undefined; + /** + * Maximum milliseconds between consecutive content-bearing events once + * content has started flowing. Does not govern the pre-first-content + * window — that is `firstContentMs`'s job. Unset by default. + */ + contentIntervalMs?: number | undefined; +}; + +/** + * Event classification hooks for {@link applyStreamWatchdog}. + */ +export type StreamWatchdogHooks = { + /** + * Returns true when the event carries model-generated output (or a + * completed sub-result). Content events satisfy and re-arm deadlines. + */ + isContentEvent: (event: T) => boolean; + /** + * Returns true when the event signals the response is finishing + * (completed / failed / incomplete / error). Terminal events permanently + * disarm the watchdog so trailing bookkeeping events are never killed. + */ + isTerminalEvent?: ((event: T) => boolean) | undefined; + /** + * Invoked once, before the wrapped stream errors, when a deadline + * expires. Phase 2 uses this to abort the underlying HTTP request. + */ + onStall?: ((error: StreamStalledError) => void) | undefined; +}; + +/** + * Wrap `source` with stall deadlines. Returns `source` unchanged when no + * deadline is configured. + * + * The wrapper pumps eagerly (it does not propagate backpressure); parsed + * SSE events are small and downstream consumers buffer regardless. + */ +export function applyStreamWatchdog( + source: ReadableStream, + timeouts: StreamTimeoutOptions, + hooks: StreamWatchdogHooks, +): ReadableStream { + const firstContentMs = normalizeTimeout(timeouts.firstContentMs); + const contentIntervalMs = normalizeTimeout(timeouts.contentIntervalMs); + + if (firstContentMs === undefined && contentIntervalMs === undefined) { + return source; + } + + const reader = source.getReader(); + + let timer: ReturnType | undefined; + let armedPhase: StreamStallPhase = 'first_content'; + let armedTimeoutMs = 0; + let armedAtMs = 0; + let receivedContent = false; + /** Set by a terminal event: deadlines permanently stop applying. */ + let disarmedForever = false; + /** Set once the wrapped stream has closed, errored, or stalled. */ + let settled = false; + + const clearTimer = (): void => { + if (timer !== undefined) { + clearTimeout(timer); + timer = undefined; + } + }; + + return new ReadableStream({ + start(controller) { + const onDeadlineExpired = (): void => { + if (settled) { + return; + } + settled = true; + timer = undefined; + const error = new StreamStalledError({ + phase: armedPhase, + timeoutMs: armedTimeoutMs, + elapsedMs: Date.now() - armedAtMs, + receivedAnyContent: receivedContent, + }); + hooks.onStall?.(error); + controller.error(error); + // Cancelling the reader resolves the pump's pending read; the + // settled flag makes the pump bail without touching the controller. + void reader.cancel(error).catch(() => { + // Cancellation failures are irrelevant once the stream errored. + }); + }; + + const arm = (phase: StreamStallPhase, timeoutMs: number): void => { + clearTimer(); + armedPhase = phase; + armedTimeoutMs = timeoutMs; + armedAtMs = Date.now(); + timer = setTimeout(onDeadlineExpired, timeoutMs); + }; + + // The pre-first-content deadline arms the moment the stream starts. + // `contentIntervalMs` alone does not cover that window. + if (firstContentMs !== undefined) { + arm('first_content', firstContentMs); + } + + void (async () => { + try { + while (true) { + const result = await reader.read(); + if (settled) { + return; + } + if (result.done) { + settled = true; + clearTimer(); + controller.close(); + return; + } + + const event = result.value; + if (hooks.isTerminalEvent?.(event)) { + // Response is finishing; deadlines permanently stop applying + // so trailing bookkeeping events are never treated as stalls. + disarmedForever = true; + clearTimer(); + } else if (!disarmedForever && hooks.isContentEvent(event)) { + receivedContent = true; + if (contentIntervalMs !== undefined) { + arm('between_content', contentIntervalMs); + } else { + clearTimer(); + } + } + controller.enqueue(event); + } + } catch (error) { + if (settled) { + return; + } + settled = true; + clearTimer(); + controller.error(error); + } + })(); + }, + cancel(reason) { + settled = true; + clearTimer(); + return reader.cancel(reason); + }, + }); +} + +/** + * Stream event types that carry model-generated output or a completed + * sub-result. These satisfy and re-arm watchdog deadlines. + * + * Deliberately excluded (neutral — neither satisfy nor reset): + * - `response.created` / `response.in_progress` / `response.debug` — status. + * - `response.output_item.added` — an empty item shell (role prelude); the + * content it introduces arrives as subsequent delta events. + * - `response.content_part.added` / `response.reasoning_summary_part.added` + * — part shells preceding their deltas. + * - in-progress markers of server-side tools (web search, image + * generation, fusion) — status, not output. + * - Unknown event types — a healthy stream of only unknown types is + * indistinguishable from a stall; mixed streams disarm via known events. + */ +const CONTENT_BEARING_EVENT_TYPES: ReadonlySet = new Set([ + 'response.output_text.delta', + 'response.output_text.done', + 'response.output_text.annotation.added', + 'response.refusal.delta', + 'response.refusal.done', + 'response.reasoning_text.delta', + 'response.reasoning_text.done', + 'response.reasoning_summary_text.delta', + 'response.reasoning_summary_text.done', + 'response.reasoning_summary_part.done', + 'response.function_call_arguments.delta', + 'response.function_call_arguments.done', + 'response.custom_tool_call_input.delta', + 'response.custom_tool_call_input.done', + 'response.apply_patch_call_operation_diff.delta', + 'response.apply_patch_call_operation_diff.done', + 'response.content_part.done', + 'response.output_item.done', + 'response.web_search_call.completed', + 'response.image_generation_call.partial_image', + 'response.image_generation_call.completed', + 'response.fusion_call.panel.delta', + 'response.fusion_call.panel.reasoning.delta', + 'response.fusion_call.panel.completed', + 'response.fusion_call.panel.failed', + 'response.fusion_call.completed', +]); + +/** + * Stream event types that signal the response is finishing. They disarm + * the watchdog permanently: the server has produced its verdict, so stall + * deadlines no longer apply (server-reported failures surface through + * their own error paths, not as stalls). + */ +const TERMINAL_EVENT_TYPES: ReadonlySet = new Set([ + 'response.completed', + 'response.failed', + 'response.incomplete', + 'error', +]); + +/** + * True when the event carries model output (text / refusal / reasoning + * deltas, tool-call arguments, completed output items or sub-results). + */ +export function isContentBearingStreamEvent(event: models.StreamEvents): boolean { + return ( + 'type' in event && typeof event.type === 'string' && CONTENT_BEARING_EVENT_TYPES.has(event.type) + ); +} + +/** + * True when the event signals the response is finishing (completed, + * failed, incomplete, or a server-emitted error event). + */ +export function isTerminalStreamEvent(event: models.StreamEvents): boolean { + return 'type' in event && typeof event.type === 'string' && TERMINAL_EVENT_TYPES.has(event.type); +} + +/** + * Convenience wrapper of {@link applyStreamWatchdog} for OpenResponses + * event streams, using the standard event classification. + */ +export function applyResponsesStreamWatchdog( + source: ReadableStream, + timeouts: StreamTimeoutOptions, + hooks?: { onStall?: ((error: StreamStalledError) => void) | undefined }, +): ReadableStream { + return applyStreamWatchdog(source, timeouts, { + isContentEvent: isContentBearingStreamEvent, + isTerminalEvent: isTerminalStreamEvent, + onStall: hooks?.onStall, + }); +} + +/** + * Treat non-finite and non-positive values as "disabled" so callers can + * pass raw user input without pre-validating. + */ +function normalizeTimeout(value: number | undefined): number | undefined { + if (value === undefined || !Number.isFinite(value) || value <= 0) { + return undefined; + } + return value; +} diff --git a/tests/unit/stream-watchdog.test.ts b/tests/unit/stream-watchdog.test.ts new file mode 100644 index 000000000..077debd0f --- /dev/null +++ b/tests/unit/stream-watchdog.test.ts @@ -0,0 +1,530 @@ +import type { StreamEvents } from '../../src/models/streamevents.js'; +import type { TextDeltaEvent } from '../../src/models/textdeltaevent.js'; +import type { ReasoningDeltaEvent } from '../../src/models/reasoningdeltaevent.js'; +import type { FunctionCallArgsDeltaEvent } from '../../src/models/functioncallargsdeltaevent.js'; +import type { OpenResponsesCreatedEvent } from '../../src/models/openresponsescreatedevent.js'; +import type { OpenResponsesResult } from '../../src/models/openresponsesresult.js'; +import type { StreamEventsResponseCompleted } from '../../src/models/streamevents.js'; +import type { StreamEventsResponseOutputItemAdded } from '../../src/models/streamevents.js'; + +import { describe, expect, it } from 'vitest'; +import { StreamStalledError } from '../../src/lib/stream-errors.js'; +import { + applyResponsesStreamWatchdog, + applyStreamWatchdog, + isContentBearingStreamEvent, + isTerminalStreamEvent, +} from '../../src/lib/stream-watchdog.js'; + +// ============================================================================ +// Scripted stream helpers +// +// The watchdog measures real elapsed time between events, so these tests use +// short real delays (tens of ms) rather than fake timers: vitest fake timers +// cannot advance a timer that races a genuinely pending microtask-driven +// stream read without also stepping the stream's own delays. +// ============================================================================ + +type ScriptStep = + | { kind: 'event'; value: T; delayMs: number } + | { kind: 'close'; delayMs: number } + | { kind: 'silence'; delayMs: number }; + +function event(value: T, delayMs = 0): ScriptStep { + return { kind: 'event', value, delayMs }; +} + +function close(delayMs = 0): ScriptStep { + return { kind: 'close', delayMs }; +} + +/** A gap with no events and no close — the stream just goes quiet. */ +function silence(delayMs: number): ScriptStep { + return { kind: 'silence', delayMs }; +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/** + * Build a ReadableStream that plays back the scripted steps. If the script + * ends without an explicit `close`, the stream stays open forever (hung + * connection) — exactly the pathology the watchdog exists to catch. + */ +function scriptedStream(steps: ScriptStep[]): ReadableStream { + let cancelled = false; + return new ReadableStream({ + start(controller) { + void (async () => { + for (const step of steps) { + await sleep(step.delayMs); + if (cancelled) { + return; + } + if (step.kind === 'event') { + controller.enqueue(step.value); + } else if (step.kind === 'close') { + controller.close(); + return; + } + // 'silence' steps only consume time. + } + // No explicit close: leave the stream hanging open. + })(); + }, + cancel() { + cancelled = true; + }, + }); +} + +async function collect(stream: ReadableStream): Promise { + const reader = stream.getReader(); + const out: T[] = []; + while (true) { + const result = await reader.read(); + if (result.done) { + return out; + } + out.push(result.value); + } +} + +async function collectError(stream: ReadableStream): Promise<{ events: T[]; error: unknown }> { + const reader = stream.getReader(); + const events: T[] = []; + try { + while (true) { + const result = await reader.read(); + if (result.done) { + throw new Error('Expected the stream to error, but it closed cleanly'); + } + events.push(result.value); + } + } catch (error) { + return { events, error }; + } +} + +// ============================================================================ +// Typed OpenResponses event fixtures +// ============================================================================ + +function fakeResponse(status: OpenResponsesResult['status']): OpenResponsesResult { + return { + id: 'resp_1', + object: 'response', + createdAt: 0, + model: 'test-model', + status, + completedAt: null, + output: [], + error: null, + incompleteDetails: null, + temperature: null, + topP: null, + presencePenalty: null, + frequencyPenalty: null, + metadata: null, + tools: [], + toolChoice: 'auto', + parallelToolCalls: false, + }; +} + +function createdEvent(): OpenResponsesCreatedEvent { + return { type: 'response.created', response: fakeResponse('in_progress'), sequenceNumber: 0 }; +} + +function textDelta(delta: string): TextDeltaEvent { + return { + type: 'response.output_text.delta', + delta, + itemId: 'item_1', + contentIndex: 0, + outputIndex: 0, + logprobs: [], + sequenceNumber: 1, + }; +} + +function reasoningDelta(delta: string): ReasoningDeltaEvent { + return { + type: 'response.reasoning_text.delta', + delta, + itemId: 'item_1', + contentIndex: 0, + outputIndex: 0, + sequenceNumber: 1, + }; +} + +function functionCallArgsDelta(delta: string): FunctionCallArgsDeltaEvent { + return { + type: 'response.function_call_arguments.delta', + delta, + itemId: 'item_1', + outputIndex: 0, + sequenceNumber: 1, + }; +} + +/** An empty assistant-message shell: the "role prelude" that must NOT count as content. */ +function emptyMessageShell(): StreamEventsResponseOutputItemAdded { + return { + type: 'response.output_item.added', + outputIndex: 0, + item: { + type: 'message', + id: 'item_1', + role: 'assistant', + status: 'in_progress', + content: [], + }, + sequenceNumber: 0, + }; +} + +function completedEvent(): StreamEventsResponseCompleted { + return { + type: 'response.completed', + response: fakeResponse('completed'), + sequenceNumber: 9, + }; +} + +// ============================================================================ +// Event classification +// ============================================================================ + +describe('isContentBearingStreamEvent', () => { + it('classifies output deltas as content', () => { + expect(isContentBearingStreamEvent(textDelta('hi'))).toBe(true); + expect(isContentBearingStreamEvent(reasoningDelta('hmm'))).toBe(true); + expect(isContentBearingStreamEvent(functionCallArgsDelta('{'))).toBe(true); + }); + + it('does not classify status events or item shells as content', () => { + expect(isContentBearingStreamEvent(createdEvent())).toBe(false); + expect(isContentBearingStreamEvent(emptyMessageShell())).toBe(false); + expect(isContentBearingStreamEvent(completedEvent())).toBe(false); + }); + + it('does not classify unknown event types as content', () => { + const unknown = { type: 'UNKNOWN', raw: { type: 'response.mystery' }, isUnknown: true } as StreamEvents; + expect(isContentBearingStreamEvent(unknown)).toBe(false); + }); +}); + +describe('isTerminalStreamEvent', () => { + it('classifies completion / failure / error events as terminal', () => { + expect(isTerminalStreamEvent(completedEvent())).toBe(true); + expect( + isTerminalStreamEvent({ + type: 'response.failed', + response: fakeResponse('failed'), + sequenceNumber: 3, + }), + ).toBe(true); + expect( + isTerminalStreamEvent({ + type: 'error', + code: 'server_error', + message: 'boom', + param: null, + sequenceNumber: 3, + }), + ).toBe(true); + }); + + it('does not classify content or status events as terminal', () => { + expect(isTerminalStreamEvent(textDelta('hi'))).toBe(false); + expect(isTerminalStreamEvent(createdEvent())).toBe(false); + }); +}); + +// ============================================================================ +// Watchdog behavior (generic) +// ============================================================================ + +describe('applyStreamWatchdog', () => { + const hooks = { + isContentEvent: (e: string) => e.startsWith('content'), + isTerminalEvent: (e: string) => e === 'terminal', + }; + + it('returns the source stream unchanged when no timeout is configured', () => { + const source = scriptedStream([close()]); + expect(applyStreamWatchdog(source, {}, hooks)).toBe(source); + // Also for explicitly disabled values: + const source2 = scriptedStream([close()]); + expect(applyStreamWatchdog(source2, { firstContentMs: 0, contentIntervalMs: -5 }, hooks)).toBe( + source2, + ); + const source3 = scriptedStream([close()]); + expect(applyStreamWatchdog(source3, { firstContentMs: Number.NaN }, hooks)).toBe(source3); + }); + + it('passes a healthy stream through untouched', async () => { + const wrapped = applyStreamWatchdog( + scriptedStream([event('content-1', 5), event('content-2', 5), close(5)]), + { firstContentMs: 1000, contentIntervalMs: 1000 }, + hooks, + ); + expect(await collect(wrapped)).toEqual(['content-1', 'content-2']); + }); + + it('fails with first_content when no content arrives at all', async () => { + const wrapped = applyStreamWatchdog( + scriptedStream([]), // hangs forever + { firstContentMs: 40 }, + hooks, + ); + const { events, error } = await collectError(wrapped); + expect(events).toEqual([]); + expect(error).toBeInstanceOf(StreamStalledError); + const stall = error as StreamStalledError; + expect(stall.phase).toBe('first_content'); + expect(stall.timeoutMs).toBe(40); + expect(stall.receivedAnyContent).toBe(false); + expect(stall.retryable).toBe(true); + expect(stall.elapsedMs).toBeGreaterThanOrEqual(30); + }); + + it('fails with first_content when only non-content events arrive (metadata-then-stall)', async () => { + const wrapped = applyStreamWatchdog( + scriptedStream([event('meta-1', 5), event('meta-2', 10)]), // then hangs + { firstContentMs: 50 }, + hooks, + ); + const { events, error } = await collectError(wrapped); + // Non-content events still flow through to the consumer... + expect(events).toEqual(['meta-1', 'meta-2']); + // ...but they do not reset the first-content deadline. + expect(error).toBeInstanceOf(StreamStalledError); + expect((error as StreamStalledError).phase).toBe('first_content'); + expect((error as StreamStalledError).retryable).toBe(true); + }); + + it('non-content events do not extend the first_content deadline', async () => { + // Steady metadata heartbeats every 20ms, forever. If they reset the + // deadline the watchdog would never fire. + const heartbeats: ScriptStep[] = Array.from({ length: 50 }, () => event('meta', 20)); + const wrapped = applyStreamWatchdog(scriptedStream(heartbeats), { firstContentMs: 90 }, hooks); + const startedAt = Date.now(); + const { error } = await collectError(wrapped); + expect(error).toBeInstanceOf(StreamStalledError); + expect((error as StreamStalledError).phase).toBe('first_content'); + // Fired near the deadline, not after 50 * 20ms of heartbeats. + expect(Date.now() - startedAt).toBeLessThan(500); + }); + + it('first content event disarms firstContentMs permanently', async () => { + const wrapped = applyStreamWatchdog( + // Content at 10ms, then a 100ms quiet gap, then close — the gap is + // longer than firstContentMs but there is no content-interval timeout. + scriptedStream([event('content-1', 10), event('content-2', 100), close()]), + { firstContentMs: 50 }, + hooks, + ); + expect(await collect(wrapped)).toEqual(['content-1', 'content-2']); + }); + + it('fails with between_content when the stream stalls mid-generation', async () => { + const wrapped = applyStreamWatchdog( + scriptedStream([event('content-1', 5), event('content-2', 10), silence(1000)]), + { firstContentMs: 200, contentIntervalMs: 50 }, + hooks, + ); + const { events, error } = await collectError(wrapped); + expect(events).toEqual(['content-1', 'content-2']); + expect(error).toBeInstanceOf(StreamStalledError); + const stall = error as StreamStalledError; + expect(stall.phase).toBe('between_content'); + expect(stall.timeoutMs).toBe(50); + expect(stall.receivedAnyContent).toBe(true); + // Content already flowed: NOT safe to blind-retry. + expect(stall.retryable).toBe(false); + }); + + it('content events re-arm the between_content deadline', async () => { + // Four content events, each 30ms apart, with a 50ms interval timeout: + // each event must reset the clock or this would fail spuriously. + const wrapped = applyStreamWatchdog( + scriptedStream([ + event('content-1', 30), + event('content-2', 30), + event('content-3', 30), + event('content-4', 30), + close(10), + ]), + { firstContentMs: 100, contentIntervalMs: 50 }, + hooks, + ); + expect(await collect(wrapped)).toEqual(['content-1', 'content-2', 'content-3', 'content-4']); + }); + + it('contentIntervalMs alone does not police the pre-first-content window', async () => { + // 80ms of silence before the first content, but only contentIntervalMs + // configured — the pre-content window is firstContentMs's job. + const wrapped = applyStreamWatchdog( + scriptedStream([event('content-1', 80), close()]), + { contentIntervalMs: 30 }, + hooks, + ); + expect(await collect(wrapped)).toEqual(['content-1']); + }); + + it('a terminal event disarms deadlines so trailing events are never stalls', async () => { + const wrapped = applyStreamWatchdog( + scriptedStream([ + event('content-1', 5), + event('terminal', 5), + // Long post-terminal gap before bookkeeping + close. + event('meta-trailer', 100), + close(), + ]), + { firstContentMs: 50, contentIntervalMs: 30 }, + hooks, + ); + expect(await collect(wrapped)).toEqual(['content-1', 'terminal', 'meta-trailer']); + }); + + it('a clean close cancels all deadlines', async () => { + const wrapped = applyStreamWatchdog( + scriptedStream([event('meta', 5), close(5)]), + { firstContentMs: 50 }, + hooks, + ); + expect(await collect(wrapped)).toEqual(['meta']); + // Wait past the deadline to catch a leaked timer firing on a closed stream. + await sleep(80); + }); + + it('cancels the source stream when a deadline expires', async () => { + let sourceCancelled = false; + const source = new ReadableStream({ + cancel() { + sourceCancelled = true; + }, + }); + const wrapped = applyStreamWatchdog(source, { firstContentMs: 30 }, hooks); + const { error } = await collectError(wrapped); + expect(error).toBeInstanceOf(StreamStalledError); + // Reader.cancel resolves asynchronously. + await sleep(10); + expect(sourceCancelled).toBe(true); + }); + + it('invokes onStall exactly once before erroring', async () => { + const stalls: StreamStalledError[] = []; + const wrapped = applyStreamWatchdog( + scriptedStream([]), + { firstContentMs: 30 }, + { ...hooks, onStall: (e) => stalls.push(e) }, + ); + const { error } = await collectError(wrapped); + await sleep(50); // room for any duplicate firing + expect(stalls).toHaveLength(1); + expect(stalls[0]).toBe(error); + }); + + it('propagates upstream errors as-is (not wrapped in StreamStalledError)', async () => { + const upstreamFailure = new Error('upstream exploded'); + const source = new ReadableStream({ + start(controller) { + setTimeout(() => controller.error(upstreamFailure), 10); + }, + }); + const wrapped = applyStreamWatchdog(source, { firstContentMs: 1000 }, hooks); + const { error } = await collectError(wrapped); + expect(error).toBe(upstreamFailure); + }); + + it('cancelling the wrapped stream stops timers and cancels the source', async () => { + let sourceCancelled = false; + const source = new ReadableStream({ + cancel() { + sourceCancelled = true; + }, + }); + const wrapped = applyStreamWatchdog(source, { firstContentMs: 30 }, hooks); + await wrapped.cancel('consumer walked away'); + expect(sourceCancelled).toBe(true); + // Wait past the deadline: the timer must not fire after cancellation. + await sleep(60); + }); +}); + +// ============================================================================ +// Responses-flavored wrapper +// ============================================================================ + +describe('applyResponsesStreamWatchdog', () => { + it('stalls on the OpenRouter keepalive pathology: created + role prelude, then silence', async () => { + // This is the exact DEV-723 scenario: headers arrive, response.created + // and an empty message shell stream in, then nothing. (SSE keep-alive + // comments never reach this layer — the SSE parser drops them.) + const wrapped = applyResponsesStreamWatchdog( + scriptedStream([event(createdEvent(), 5), event(emptyMessageShell(), 5)]), + { firstContentMs: 60 }, + ); + const { events, error } = await collectError(wrapped); + expect(events).toHaveLength(2); + expect(error).toBeInstanceOf(StreamStalledError); + const stall = error as StreamStalledError; + expect(stall.phase).toBe('first_content'); + expect(stall.retryable).toBe(true); + }); + + it('passes a healthy responses stream through to completion', async () => { + const wrapped = applyResponsesStreamWatchdog( + scriptedStream([ + event(createdEvent(), 5), + event(emptyMessageShell(), 5), + event(textDelta('Hello'), 5), + event(textDelta(' world'), 5), + event(completedEvent(), 5), + close(5), + ]), + { firstContentMs: 100, contentIntervalMs: 100 }, + ); + const collected = await collect(wrapped); + expect(collected).toHaveLength(5); + expect(collected.at(-1)?.type).toBe('response.completed'); + }); + + it('reasoning deltas satisfy the first-content deadline (reasoning models)', async () => { + const wrapped = applyResponsesStreamWatchdog( + scriptedStream([ + event(createdEvent(), 5), + // Reasoning streams for a while before any output text appears. + event(reasoningDelta('thinking...'), 10), + event(textDelta('answer'), 80), + event(completedEvent(), 5), + close(), + ]), + { firstContentMs: 50 }, + ); + const collected = await collect(wrapped); + expect(collected).toHaveLength(4); + }); + + it('stalls mid-generation when deltas stop and nothing terminal arrives', async () => { + const wrapped = applyResponsesStreamWatchdog( + scriptedStream([ + event(createdEvent(), 5), + event(textDelta('partial outp'), 5), + silence(1000), + ]), + { firstContentMs: 100, contentIntervalMs: 60 }, + ); + const { error } = await collectError(wrapped); + expect(error).toBeInstanceOf(StreamStalledError); + const stall = error as StreamStalledError; + expect(stall.phase).toBe('between_content'); + expect(stall.receivedAnyContent).toBe(true); + expect(stall.retryable).toBe(false); + }); +});