From a7e4378683e11168ce35b6bfbb9ef1604323c1fa Mon Sep 17 00:00:00 2001 From: Luke Parke <5702154+LukasParke@users.noreply.github.com> Date: Mon, 10 Aug 2026 10:15:30 -0500 Subject: [PATCH] feat: wire stream watchdog into callModel with per-turn stall timeouts (DEV-723 phase 2) Adds the opt-in 'timeout' option to callModel: client.callModel({ model, input, timeout: { firstContentMs: 15_000, contentIntervalMs: 30_000 }, }) Integration points: - CallModelInput gains 'timeout' (client-only field, stripped from the API request in resolveAsyncFunctions and resolveRequestForContext). - ModelResult.createTurnWatchContext() builds a per-turn AbortController whose signal is merged (combineSignals) with the caller's signal and, when the caller set none, a replicated timeoutMs signal - configuring the watchdog never silently disables existing timeout behavior. The watchdog's onStall aborts the turn's HTTP request, tearing down the hung connection. - All three responsesSend call sites (initial request, follow-up tool turns, approval-resume continuation) route through the watch context, so deadlines re-arm independently for every turn, matching the per-step semantics of vercel/ai firstChunkMs. - ToolEventBroadcaster.complete() is now idempotent (first call wins): the pipe's error completion must not be overwritten by the later unconditional .finally() completion, or concurrent consumers would see a clean close instead of the stall error. - StreamStalledError, StreamStallPhase, and StreamTimeoutOptions are exported from the package root via the sdk.ts custom region. Tests drive the real transport path (HTTPClient fetcher serving scripted SSE bytes through responsesSend -> EventStream -> ModelResult): stall before content, keepalive-comment immunity, healthy streams, default-off, mid-generation stall, multi-consumer rejection, per-turn re-arm across a tool loop (turn 1 passes, turn 2 stalls), and caller abort signals surviving signal merging. --- src/funcs/call-model.ts | 2 + src/lib/async-params.ts | 19 + src/lib/model-result.ts | 99 ++++- src/lib/stream-watchdog.ts | 15 + src/lib/tool-event-broadcaster.ts | 10 + src/sdk/sdk.ts | 2 + tests/unit/call-model-stream-timeout.test.ts | 403 +++++++++++++++++++ 7 files changed, 541 insertions(+), 9 deletions(-) create mode 100644 tests/unit/call-model-stream-timeout.test.ts diff --git a/src/funcs/call-model.ts b/src/funcs/call-model.ts index 58132fc47..3de7d6c25 100644 --- a/src/funcs/call-model.ts +++ b/src/funcs/call-model.ts @@ -87,6 +87,7 @@ export function callModel< sharedContextSchema, onTurnStart, onTurnEnd, + timeout, ...apiRequest } = request; @@ -128,5 +129,6 @@ export function callModel< ...(sharedContextSchema !== undefined && { sharedContextSchema }), ...(onTurnStart !== undefined && { onTurnStart }), ...(onTurnEnd !== undefined && { onTurnEnd }), + ...(timeout !== undefined && { timeout }), } as GetResponseOptions); } diff --git a/src/lib/async-params.ts b/src/lib/async-params.ts index 364ae2153..61b283f70 100644 --- a/src/lib/async-params.ts +++ b/src/lib/async-params.ts @@ -2,6 +2,7 @@ import type * as models from '../models/index.js'; import type { ToolContextMapWithShared, ParsedToolCall, StateAccessor, StopWhen, Tool, TurnContext } from './tool-types.js'; import type { OpenResponsesResult } from '../models/index.js'; import type { ContextInput } from './tool-context.js'; +import type { StreamTimeoutOptions } from './stream-watchdog.js'; // Re-export Tool type for convenience export type { Tool } from './tool-types.js'; @@ -46,6 +47,23 @@ type BaseCallModelInput< } & { tools?: TTools; stopWhen?: StopWhen; + /** + * Opt-in stalled-stream detection (fail fast instead of hanging on a + * stream that returns headers but never produces content). + * + * - `firstContentMs`: max milliseconds between a turn's response stream + * starting and its first content-bearing event (text/reasoning delta, + * tool-call arguments, ...). Keep-alives and metadata events do not + * satisfy or reset it. + * - `contentIntervalMs`: max gap between content-bearing events once + * content has started. + * + * Deadlines re-arm for every turn in multi-turn tool loops. On expiry the + * in-flight request is aborted and all consumers reject with + * `StreamStalledError` (its `retryable` getter is true only when no + * content had been received). Unset by default: no watchdog runs. + */ + timeout?: StreamTimeoutOptions; /** Typed context data passed to tools via contextSchema. Includes optional `shared` key. */ context?: ContextInput>; /** @@ -162,6 +180,7 @@ export async function resolveAsyncFunctions void | Promise; /** Callback invoked at the end of each tool execution turn */ onTurnEnd?: (context: TurnContext, response: models.OpenResponsesResult) => void | Promise; + + /** + * Opt-in stalled-stream detection. Deadlines are armed per turn; on + * expiry the turn's HTTP request is aborted and consumers reject with + * `StreamStalledError`. See `StreamTimeoutOptions`. + */ + timeout?: StreamTimeoutOptions; } +/** + * Per-turn stall-detection context: request options carrying the merged + * abort signal, and a wrapper that arms the watchdog on the turn's stream. + * When no stall timeout is active, `requestOptions` is the caller's + * options unchanged and `watch` is the identity function. + */ +type TurnWatchContext = { + requestOptions: RequestOptions | undefined; + watch: ( + stream: ReadableStream, + ) => ReadableStream; +}; + /** * A wrapper around a streaming response that provides multiple consumption patterns. * @@ -875,10 +901,12 @@ export class ModelResult< stream: true, }; + // Stall deadlines re-arm independently for every turn. + const turnWatch = this.createTurnWatchContext(); const newResult = await responsesSend( this.options.client, { responsesRequest: newRequest }, - this.options.options, + turnWatch.requestOptions, ); if (!newResult.ok) { @@ -888,7 +916,7 @@ export class ModelResult< // Handle streaming or non-streaming response const value = newResult.value; if (isEventStream(value)) { - const followUpStream = new ReusableReadableStream(value); + const followUpStream = new ReusableReadableStream(turnWatch.watch(value)); if (this.turnBroadcaster) { return this.pipeAndConsumeStream(followUpStream, turnNumber); @@ -919,6 +947,57 @@ export class ModelResult< } } + /** + * Build the stall-detection context for one turn. + * + * When a stall timeout is configured, the turn gets its own + * AbortController (merged with the caller's signal so neither is lost). + * The returned `watch` wrapper arms the watchdog on the turn's parsed + * event stream; on expiry the watchdog aborts the turn's HTTP request, + * tearing down the hung connection, and the stream errors with + * `StreamStalledError`. + */ + private createTurnWatchContext(): TurnWatchContext { + const timeouts = this.options.timeout; + if (!timeouts || !hasActiveStreamTimeouts(timeouts)) { + return { + requestOptions: this.options.options, + watch: (stream) => stream, + }; + } + + const turnAbort = new AbortController(); + const baseOptions = this.options.options; + const callerSignal = baseOptions?.signal ?? baseOptions?.fetchOptions?.signal ?? null; + + /* + * The generated request builder only applies `timeoutMs` (via + * AbortSignal.timeout) when no signal is provided. Since we are about + * to provide one, replicate that behavior here so configuring a stall + * watchdog never silently disables the caller's overall timeout. + */ + const timeoutMs = baseOptions?.timeoutMs ?? this.options.client._options.timeoutMs; + const timeoutSignal = + !callerSignal && timeoutMs && timeoutMs > 0 ? AbortSignal.timeout(timeoutMs) : null; + + const mergedSignal = combineSignals(turnAbort.signal, callerSignal, timeoutSignal); + + const requestOptions: RequestOptions = { + ...baseOptions, + ...(mergedSignal ? { signal: mergedSignal } : {}), + }; + + return { + requestOptions, + watch: (stream) => + applyResponsesStreamWatchdog(stream, timeouts, { + onStall: (error) => { + turnAbort.abort(error); + }, + }), + }; + } + /** * Resolve async functions in the request for a given turn context. * Extracts non-function fields and resolves any async parameter functions. @@ -932,7 +1011,7 @@ export class ModelResult< } // Already resolved, extract non-function fields // Filter out stopWhen and state-related fields that aren't part of the API request - const { stopWhen: _, state: _s, requireApproval: _r, approveToolCalls: _a, rejectToolCalls: _rj, context: _c, ...rest } = this.options.request; + const { stopWhen: _, state: _s, requireApproval: _r, approveToolCalls: _a, rejectToolCalls: _rj, context: _c, timeout: _t, ...rest } = this.options.request; return rest as ResolvedCallModelInput; } @@ -1066,11 +1145,12 @@ export class ModelResult< // Force stream mode for initial request const request = this.resolvedRequest; - // Make the API request + // Make the API request (with per-turn stall detection when configured) + const turnWatch = this.createTurnWatchContext(); const apiResult = await responsesSend( this.options.client, { responsesRequest: request }, - this.options.options, + turnWatch.requestOptions, ); if (!apiResult.ok) { @@ -1080,7 +1160,7 @@ export class ModelResult< // Handle both streaming and non-streaming responses // The API may return a non-streaming response even when stream: true is requested if (isEventStream(apiResult.value)) { - this.reusableStream = new ReusableReadableStream(apiResult.value); + this.reusableStream = new ReusableReadableStream(turnWatch.watch(apiResult.value)); } else if (this.isNonStreamingResponse(apiResult.value)) { // API returned a complete response directly - use it as the final response this.finalResponse = apiResult.value; @@ -1212,11 +1292,12 @@ export class ModelResult< this.resolvedRequest = request; - // Make the API request + // Make the API request (stall deadlines re-arm for the resumed turn) + const turnWatch = this.createTurnWatchContext(); const apiResult = await responsesSend( this.options.client, { responsesRequest: request }, - this.options.options, + turnWatch.requestOptions, ); if (!apiResult.ok) { @@ -1225,7 +1306,7 @@ export class ModelResult< // Handle both streaming and non-streaming responses if (isEventStream(apiResult.value)) { - this.reusableStream = new ReusableReadableStream(apiResult.value); + this.reusableStream = new ReusableReadableStream(turnWatch.watch(apiResult.value)); } else if (this.isNonStreamingResponse(apiResult.value)) { this.finalResponse = apiResult.value; } else { diff --git a/src/lib/stream-watchdog.ts b/src/lib/stream-watchdog.ts index f3e071a5e..1be495a19 100644 --- a/src/lib/stream-watchdog.ts +++ b/src/lib/stream-watchdog.ts @@ -283,6 +283,21 @@ export function applyResponsesStreamWatchdog( }); } +/** + * True when at least one watchdog deadline is enabled (set, finite, > 0). + * Callers use this to skip per-turn abort plumbing entirely when the + * watchdog would be a no-op. + */ +export function hasActiveStreamTimeouts(timeouts: StreamTimeoutOptions | undefined): boolean { + if (!timeouts) { + return false; + } + return ( + normalizeTimeout(timeouts.firstContentMs) !== undefined || + normalizeTimeout(timeouts.contentIntervalMs) !== undefined + ); +} + /** * Treat non-finite and non-positive values as "disabled" so callers can * pass raw user input without pre-validating. diff --git a/src/lib/tool-event-broadcaster.ts b/src/lib/tool-event-broadcaster.ts index 7a5ecb537..122339c5a 100644 --- a/src/lib/tool-event-broadcaster.ts +++ b/src/lib/tool-event-broadcaster.ts @@ -31,8 +31,18 @@ export class ToolEventBroadcaster { * Mark the broadcaster as complete - no more events will be pushed. * Optionally pass an error to signal failure to all consumers. * Cleans up buffer and consumers after completion. + * + * Idempotent: the first call wins. This matters when a stream error + * completes the broadcaster with an error (e.g. a stalled-stream abort + * from the pipe's catch handler) and an unconditional `.finally()` + * completion runs afterwards - the later error-less call must not wipe + * the recorded failure, or late consumers would see a clean close + * instead of the error. */ complete(error?: Error): void { + if (this.isComplete) { + return; + } this.isComplete = true; this.completionError = error ?? null; this.notifyWaitingConsumers(); diff --git a/src/sdk/sdk.ts b/src/sdk/sdk.ts index c95283152..e61fc0d40 100644 --- a/src/sdk/sdk.ts +++ b/src/sdk/sdk.ts @@ -43,6 +43,8 @@ import type { RequestOptions } from "../lib/sdks.js"; import { type Tool, ToolType } from "../lib/tool-types.js"; export { ToolType }; +export { StreamStalledError, type StreamStallPhase } from "../lib/stream-errors.js"; +export type { StreamTimeoutOptions } from "../lib/stream-watchdog.js"; // #endregion imports export class OpenRouter extends ClientSDK { diff --git a/tests/unit/call-model-stream-timeout.test.ts b/tests/unit/call-model-stream-timeout.test.ts new file mode 100644 index 000000000..a7c343dc7 --- /dev/null +++ b/tests/unit/call-model-stream-timeout.test.ts @@ -0,0 +1,403 @@ +import { describe, expect, it } from 'vitest'; +import { z } from 'zod/v4'; +import { OpenRouter } from '../../src/index.js'; +import { HTTPClient } from '../../src/lib/http.js'; +import { StreamStalledError } from '../../src/lib/stream-errors.js'; +import { tool } from '../../src/lib/tool.js'; + +// ============================================================================ +// SSE fixture helpers (snake_case wire shapes, parsed by the real inbound +// schemas via responsesSend -> EventStream -> ModelResult) +// ============================================================================ + +/** Minimal OpenResponsesResult JSON in wire (snake_case) shape. */ +function resultJson(options: { + status: 'in_progress' | 'completed'; + output?: unknown[]; + outputText?: string; +}): Record { + return { + id: 'resp_1', + object: 'response', + created_at: 0, + completed_at: options.status === 'completed' ? 1 : null, + model: 'test-model', + status: options.status, + output: options.output ?? [], + ...(options.outputText !== undefined ? { output_text: options.outputText } : {}), + error: null, + incomplete_details: null, + instructions: null, + metadata: null, + temperature: null, + top_p: null, + presence_penalty: null, + frequency_penalty: null, + tools: [], + tool_choice: 'auto', + parallel_tool_calls: false, + }; +} + +function messageItem(text: string): Record { + return { + type: 'message', + id: 'msg_1', + role: 'assistant', + status: 'completed', + content: [{ type: 'output_text', text, annotations: [] }], + }; +} + +function functionCallItem(name: string): Record { + return { + type: 'function_call', + id: 'fc_1', + call_id: 'call_1', + name, + arguments: '{}', + status: 'completed', + }; +} + +function sse(event: Record): string { + return `data: ${JSON.stringify(event)}\n\n`; +} + +const createdFrame = sse({ + type: 'response.created', + sequence_number: 0, + response: resultJson({ status: 'in_progress' }), +}); + +function textDeltaFrame(delta: string): string { + return sse({ + type: 'response.output_text.delta', + delta, + item_id: 'msg_1', + content_index: 0, + output_index: 0, + logprobs: [], + sequence_number: 1, + }); +} + +function completedFrame(output: unknown[], outputText?: string): string { + return sse({ + type: 'response.completed', + sequence_number: 9, + response: resultJson({ + status: 'completed', + output, + ...(outputText !== undefined ? { outputText } : {}), + }), + }); +} + +const DONE_FRAME = 'data: [DONE]\n\n'; + +/** The router's SSE keep-alive comment; dropped by the SSE parser. */ +const KEEPALIVE_FRAME = ': OPENROUTER PROCESSING\n\n'; + +// ============================================================================ +// Scripted SSE transport +// ============================================================================ + +type BodyStep = + | { kind: 'frame'; text: string; delayMs: number } + | { kind: 'close'; delayMs: number }; + +function frame(text: string, delayMs = 0): BodyStep { + return { kind: 'frame', text, delayMs }; +} + +function closeBody(delayMs = 0): BodyStep { + return { kind: 'close', delayMs }; +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +type TransportObservations = { + requests: Request[]; + /** Body teardowns: reader cancellations and abort-driven body errors. */ + bodyTeardowns: unknown[]; +}; + +/** + * Build an OpenRouter client whose fetch plays back one scripted SSE body + * per request. A script without a `close` step hangs forever — the stalled + * connection this feature exists to detect. + */ +function scriptedClient(scripts: BodyStep[][]): { + client: OpenRouter; + observed: TransportObservations; +} { + const observed: TransportObservations = { requests: [], bodyTeardowns: [] }; + const encoder = new TextEncoder(); + let call = 0; + + const fetcher = async (input: RequestInfo | URL): Promise => { + if (!(input instanceof Request)) { + throw new Error('Expected a Request instance from the SDK'); + } + observed.requests.push(input); + const script = scripts[call++]; + if (!script) { + throw new Error(`No scripted response for request #${call}`); + } + + let cancelled = false; + const body = new ReadableStream({ + start(controller) { + // Real fetch implementations fail in-flight body reads when the + // request is aborted; mirror that so abort semantics are realistic. + input.signal.addEventListener('abort', () => { + if (!cancelled) { + cancelled = true; + observed.bodyTeardowns.push(input.signal.reason); + try { + controller.error( + input.signal.reason ?? new DOMException('The operation was aborted.', 'AbortError'), + ); + } catch { + // Controller may already be closed. + } + } + }); + void (async () => { + for (const step of script) { + await sleep(step.delayMs); + if (cancelled) { + return; + } + if (step.kind === 'frame') { + controller.enqueue(encoder.encode(step.text)); + } else { + controller.close(); + return; + } + } + // No close step: hang, holding the connection open. + })(); + }, + cancel(reason) { + cancelled = true; + observed.bodyTeardowns.push(reason); + }, + }); + + return new Response(body, { + status: 200, + headers: { 'content-type': 'text/event-stream' }, + }); + }; + + const client = new OpenRouter({ + apiKey: 'test-api-key', + httpClient: new HTTPClient({ fetcher }), + retryConfig: { strategy: 'none' }, + }); + + return { client, observed }; +} + +const HEALTHY_SCRIPT: BodyStep[] = [ + frame(createdFrame, 5), + frame(textDeltaFrame('Hello'), 5), + frame(textDeltaFrame(' world'), 5), + frame(completedFrame([messageItem('Hello world')], 'Hello world'), 5), + frame(DONE_FRAME), + closeBody(), +]; + +// ============================================================================ +// Tests +// ============================================================================ + +describe('callModel stream timeout integration', () => { + it('getText rejects with StreamStalledError when the stream never produces content', async () => { + const { client, observed } = scriptedClient([ + [frame(createdFrame, 5)], // created, then hangs forever + ]); + + const result = client.callModel({ + model: 'test-model', + input: 'hi', + timeout: { firstContentMs: 80 }, + }); + + await expect(result.getText()).rejects.toThrow(StreamStalledError); + + // The watchdog must tear the connection down, not just reject: + // the in-flight request is aborted and the response body cancelled. + await sleep(20); + expect(observed.requests).toHaveLength(1); + expect(observed.requests[0]?.signal.aborted).toBe(true); + expect(observed.bodyTeardowns.length).toBeGreaterThan(0); + }); + + it('keep-alive comment frames do not satisfy or reset the first-content deadline', async () => { + // Keepalives every 30ms forever — a socket-idle timer would never fire. + const keepalives: BodyStep[] = [ + frame(createdFrame, 5), + ...Array.from({ length: 40 }, () => frame(KEEPALIVE_FRAME, 30)), + ]; + const { client } = scriptedClient([keepalives]); + + const result = client.callModel({ + model: 'test-model', + input: 'hi', + timeout: { firstContentMs: 100 }, + }); + + const startedAt = Date.now(); + const error = await result.getText().then( + () => null, + (e: unknown) => e, + ); + expect(error).toBeInstanceOf(StreamStalledError); + expect((error as StreamStalledError).phase).toBe('first_content'); + expect((error as StreamStalledError).retryable).toBe(true); + // Fired near the 100ms deadline, not after the full keepalive script. + expect(Date.now() - startedAt).toBeLessThan(600); + }); + + it('resolves normally on a healthy stream well within deadlines', async () => { + const { client, observed } = scriptedClient([HEALTHY_SCRIPT]); + + const result = client.callModel({ + model: 'test-model', + input: 'hi', + timeout: { firstContentMs: 1000, contentIntervalMs: 1000 }, + }); + + expect(await result.getText()).toBe('Hello world'); + expect(observed.requests[0]?.signal.aborted).toBe(false); + }); + + it('resolves normally when no timeout is configured (default off)', async () => { + const { client } = scriptedClient([HEALTHY_SCRIPT]); + const result = client.callModel({ model: 'test-model', input: 'hi' }); + expect(await result.getText()).toBe('Hello world'); + }); + + it('fails with between_content when deltas stop mid-generation', async () => { + const { client } = scriptedClient([ + [ + frame(createdFrame, 5), + frame(textDeltaFrame('partial'), 5), + // then hangs with no completed event + ], + ]); + + const result = client.callModel({ + model: 'test-model', + input: 'hi', + timeout: { firstContentMs: 500, contentIntervalMs: 60 }, + }); + + const error = await result.getText().then( + () => null, + (e: unknown) => e, + ); + expect(error).toBeInstanceOf(StreamStalledError); + expect((error as StreamStalledError).phase).toBe('between_content'); + expect((error as StreamStalledError).receivedAnyContent).toBe(true); + expect((error as StreamStalledError).retryable).toBe(false); + }); + + it('propagates the stall to all concurrent stream consumers', async () => { + const { client } = scriptedClient([[frame(createdFrame, 5)]]); + + const result = client.callModel({ + model: 'test-model', + input: 'hi', + timeout: { firstContentMs: 60 }, + }); + + const consumeTextStream = (async () => { + for await (const _delta of result.getTextStream()) { + // drain + } + })(); + const consumeFullStream = (async () => { + for await (const _event of result.getFullResponsesStream()) { + // drain + } + })(); + + await expect(consumeTextStream).rejects.toThrow(StreamStalledError); + await expect(consumeFullStream).rejects.toThrow(StreamStalledError); + }); + + it('re-arms deadlines per turn and catches a stall in a follow-up turn', async () => { + const toolTurnScript: BodyStep[] = [ + frame(createdFrame, 5), + frame( + sse({ + type: 'response.function_call_arguments.delta', + delta: '{}', + item_id: 'fc_1', + output_index: 0, + sequence_number: 1, + }), + 5, + ), + // Turn 1 total duration exceeds firstContentMs — only a per-turn + // deadline (re-armed for turn 2) lets this pass while turn 2 fails. + frame(completedFrame([functionCallItem('get_thing')]), 120), + frame(DONE_FRAME), + closeBody(), + ]; + const stalledFollowupScript: BodyStep[] = [frame(createdFrame, 5)]; // hangs + + const { client, observed } = scriptedClient([toolTurnScript, stalledFollowupScript]); + + const getThing = tool({ + name: 'get_thing', + description: 'returns a thing', + inputSchema: z.object({}), + execute: async () => ({ thing: 42 }), + }); + + const result = client.callModel({ + model: 'test-model', + input: 'hi', + tools: [getThing], + timeout: { firstContentMs: 80 }, + }); + + await expect(result.getText()).rejects.toThrow(StreamStalledError); + // Both turns went out; the second one was aborted by its own watchdog. + expect(observed.requests).toHaveLength(2); + expect(observed.requests[0]?.signal.aborted).toBe(false); + expect(observed.requests[1]?.signal.aborted).toBe(true); + }); + + it('caller abort signals still work alongside the watchdog', async () => { + const { client } = scriptedClient([ + [frame(createdFrame, 5)], // hangs, but watchdog is generous + ]); + + const userAbort = new AbortController(); + const result = client.callModel( + { model: 'test-model', input: 'hi', timeout: { firstContentMs: 5000 } }, + { signal: userAbort.signal }, + ); + + const textPromise = result.getText(); + const pendingRejection = expect(textPromise).rejects.toThrow(); + await sleep(30); + userAbort.abort(new Error('user cancelled')); + + await pendingRejection; + const error = await textPromise.then( + () => null, + (e: unknown) => e, + ); + // The user's abort wins — it must not be reported as a stall. + expect(error).not.toBeInstanceOf(StreamStalledError); + }); +});