feat: callModel per-turn stall timeouts (DEV-723 2/5)

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<TTools, TShared>);
 }

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<TTools>;
+  /**
+   * 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<ToolContextMapWithShared<TTools, TShared>>;
   /**
@@ -162,6 +180,7 @@ export async function resolveAsyncFunctions<TTools extends readonly Tool[] = rea
     'sharedContextSchema',   // Client-side schema for shared context validation
     'onTurnStart',           // Client-side turn start callback
     'onTurnEnd',             // Client-side turn end callback
+    'timeout',               // Client-side stalled-stream watchdog config
   ]);
 
   // Iterate over all keys in the input

src/lib/model-result.ts

@@ -70,6 +70,12 @@ import {
   isReasoningDeltaEvent,
   hasTypeProperty,
 } from './stream-type-guards.js';
+import {
+  applyResponsesStreamWatchdog,
+  hasActiveStreamTimeouts,
+  type StreamTimeoutOptions,
+} from './stream-watchdog.js';
+import { combineSignals } from './primitives.js';
 
 /**
  * Default maximum number of tool execution steps if no stopWhen is specified.
@@ -129,8 +135,28 @@ export interface GetResponseOptions<
   onTurnStart?: (context: TurnContext) => void | Promise<void>;
   /** Callback invoked at the end of each tool execution turn */
   onTurnEnd?: (context: TurnContext, response: models.OpenResponsesResult) => void | Promise<void>;
+
+  /**
+   * 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<models.StreamEvents>,
+  ) => ReadableStream<models.StreamEvents>;
+};
+
 /**
  * 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 {

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.

src/lib/tool-event-broadcaster.ts

@@ -31,8 +31,18 @@ export class ToolEventBroadcaster<T> {
    * 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();

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 {