feat(agent): callModel streamTimeout, armed per model request (DEV-723 2/3)

.changeset/stream-timeout-option.md

@@ -0,0 +1,17 @@
+---
+'@openrouter/agent': minor
+---
+
+Add an opt-in `streamTimeout` option to `callModel` that fails fast on stalled response streams. A stream can return headers and then send only keep-alives or metadata events while no content flows. Transport timeouts never fire in that case, so the run used to hang.
+
+```ts
+const result = callModel(client, {
+  model: 'openai/gpt-4o',
+  input: 'hi',
+  streamTimeout: { firstContentMs: 15_000, contentIntervalMs: 30_000 },
+});
+```
+
+The deadlines re-arm for every model request in the run: the initial request, each tool round, and final, retry and resume requests. When one expires, the in-flight request is aborted and every consumer rejects with `StreamStalledError`. Its `retryable` getter is true only when no content had arrived. The option composes with `signal` and `RequestOptions.timeoutMs`, and an abort through either is reported as itself, never as a stall. The option is never sent to the API.
+
+Also fixes an unhandled promise rejection when a stream consumer exits early from a tool-enabled run that later fails.

packages/agent/README.md

@@ -353,6 +353,32 @@ over passing a raw `signal` through `RequestOptions` — the underlying SDK
 skips its `timeoutMs` wiring whenever a request already carries a signal,
 so the engine re-composes them for you only on the `signal` option path.
 
+#### Stalled streams
+
+`timeoutMs` cannot catch a stream that returns headers and then stalls: the
+router's SSE keep-alive comments keep the connection busy while no content
+flows. `streamTimeout` adds opt-in deadlines measured on parsed events:
+
+```typescript
+const result = callModel(client, {
+  model: 'openai/gpt-4o',
+  input: 'Research this topic',
+  streamTimeout: {
+    firstContentMs: 15_000, // response stream start -> first content event
+    contentIntervalMs: 30_000, // max gap between content events afterwards
+  },
+});
+```
+
+Content means text, refusal and reasoning deltas, tool-call arguments and
+completed output items. Keep-alives, `response.created`, empty item shells and
+`error` events neither satisfy nor reset a deadline. The deadlines re-arm for
+every model request in the run. On expiry the request is aborted and every
+consumer rejects with `StreamStalledError` (`phase`, `timeoutMs`, `elapsedMs`,
+`receivedAnyContent`). Its `retryable` getter is true only when no content had
+arrived, so re-issuing the request cannot duplicate output. An abort through
+`signal` or `timeoutMs` is reported as itself, never as a stall.
+
 #### Stream failures
 
 A response stream that fails surfaces as a typed `StreamFailedError` on every

packages/agent/src/index.ts

@@ -227,6 +227,7 @@ export {
   getUnsupportedContentSummary,
   hasUnsupportedContent,
 } from './lib/stream-transformers.js';
+export type { StreamTimeoutOptions } from './lib/stream-watchdog.js';
 export type {
   BuiltDeferredTool,
   DeferredToolMethods,

packages/agent/src/inner-loop/call-model.ts

@@ -115,6 +115,7 @@ export function callModel<
     doomLoop,
     signal,
     toolTimeoutMs,
+    streamTimeout,
     toolConcurrency,
     asyncTools,
     ...apiRequest
@@ -197,6 +198,7 @@ export function callModel<
     doomLoop,
     signal,
     toolTimeoutMs,
+    streamTimeout,
     toolConcurrency,
     asyncTools,
   };

packages/agent/src/lib/async-params.ts

@@ -5,6 +5,7 @@ import type { HooksManager } from './hooks-manager.js';
 import type { InlineHookConfig } from './hooks-types.js';
 import type { Item } from './item-types.js';
 import type { StreamReplay } from './reusable-stream.js';
+import type { StreamTimeoutOptions } from './stream-watchdog.js';
 import type { ContextInput } from './tool-context.js';
 import type {
   ParsedToolCall,
@@ -198,6 +199,29 @@ type BaseCallModelInput<
    * detached and its result is discarded.
    */
   toolTimeoutMs?: number;
+  /**
+   * Opt-in stalled-stream detection: fail fast on a response stream that
+   * returned headers but never produces content (keep-alives and metadata
+   * events only), instead of hanging.
+   *
+   * - `firstContentMs`: max milliseconds between a turn's response stream
+   *   starting and its first content-bearing event (text / reasoning delta,
+   *   tool-call arguments, completed output item, ...). Keep-alive comments
+   *   and status events (`response.created`, empty item shells) neither
+   *   satisfy nor reset it.
+   * - `contentIntervalMs`: max gap between content-bearing events once
+   *   content has started.
+   *
+   * Deadlines re-arm for every model request of the run (initial, each tool
+   * round, final and retry turns, resumes). On expiry the in-flight request
+   * is aborted and the run rejects with `StreamStalledError` on every
+   * consumption path; its `retryable` getter is true only when no content
+   * had been received. Unset by default: no watchdog runs.
+   *
+   * Composes with `signal` and `RequestOptions.timeoutMs`: an abort or a
+   * request timeout is reported as itself, never as a stall.
+   */
+  streamTimeout?: StreamTimeoutOptions;
   /**
    * Concurrency limits for tool execution. A bare number is shorthand for
    * `{ round: n }`.
@@ -347,6 +371,7 @@ export async function resolveAsyncFunctions<TTools extends readonly Tool[] = rea
     'doomLoop', // Client-side doom-loop detection config
     'signal', // Client-side run cancellation
     'toolTimeoutMs', // Client-side per-tool deadline default
+    'streamTimeout', // Client-side stalled-stream watchdog deadlines
     'toolConcurrency', // Client-side tool concurrency limits
     'asyncTools', // Client-side async tool (background/deferred) behavior
     'activeTools', // Applied client-side to filter tools before conversion

packages/agent/src/lib/model-result.ts

@@ -60,8 +60,13 @@ import {
   executeNextTurnParamsFunctions,
 } from './next-turn-params.js';
 import { isStopConditionMet } from './stop-conditions.js';
-import type { StreamReplay } from './stream-runtime.js';
-import { ReusableReadableStream, StreamFailedError } from './stream-runtime.js';
+import type { StreamReplay, StreamTimeoutOptions } from './stream-runtime.js';
+import {
+  applyResponsesStreamWatchdog,
+  hasActiveStreamTimeouts,
+  ReusableReadableStream,
+  StreamFailedError,
+} from './stream-runtime.js';
 import type { ItemInProgress, StreamableOutputItem } from './stream-transformers.js';
 import {
   buildItemsStream,
@@ -509,6 +514,11 @@ export interface GetResponseOptions<
    * semantics (bounds the round's wait, not the tool body).
    */
   toolTimeoutMs?: number;
+  /**
+   * Opt-in stalled-stream detection, armed per model request. See
+   * `CallModelInput.streamTimeout`.
+   */
+  streamTimeout?: StreamTimeoutOptions;
   /**
    * Concurrency limits for tool execution. A bare number is shorthand for
    * `{ round: n }`. `round` bounds simultaneous executions within a tool
@@ -1135,6 +1145,9 @@ export class ModelResult<
       }
       broadcaster.complete();
     });
+    // Attach an immediate rejection handler so early consumer exit (such as an abort)
+    // does not leave executionPromise as an unhandled promise rejection.
+    void executionPromise.catch(() => {});
     return {
       consumer,
       executionPromise,
@@ -4902,19 +4915,8 @@ export class ModelResult<
 
     this.beginToolChoiceDispatch(newRequest.toolChoice);
     const startedAt = performance.now();
-    const newResult = await betaResponsesSend(
-      this.options.client,
-      {
-        responsesRequest: newRequest,
-      },
-      this.dispatchRequestOptions(),
-    );
-
-    if (!newResult.ok) {
-      throw newResult.error;
-    }
-
-    const response = await this.materializeResponse(newResult.value, turnNumber);
+    const newValue = await this.sendModelRequest(newRequest);
+    const response = await this.materializeResponse(newValue, turnNumber);
     this.commitToolChoiceDispatch(responseHasToolCalls(response));
     await this.emitPostModelCall(response, startedAt, 'tool_round', turnNumber);
     return response;
@@ -5204,19 +5206,8 @@ export class ModelResult<
     this.resolvedRequest = finalRequest;
 
     const startedAt = performance.now();
-    const result = await betaResponsesSend(
-      this.options.client,
-      {
-        responsesRequest: finalRequest,
-      },
-      this.dispatchRequestOptions(),
-    );
-
-    if (!result.ok) {
-      throw result.error;
-    }
-
-    const response = await this.materializeResponse(result.value, turnNumber);
+    const finalValue = await this.sendModelRequest(finalRequest);
+    const response = await this.materializeResponse(finalValue, turnNumber);
     await this.emitPostModelCall(response, startedAt, 'final', turnNumber);
     return response;
   }
@@ -5290,19 +5281,8 @@ export class ModelResult<
     };
 
     const startedAt = performance.now();
-    const newResult = await betaResponsesSend(
-      this.options.client,
-      {
-        responsesRequest: newRequest,
-      },
-      this.dispatchRequestOptions(),
-    );
-
-    if (!newResult.ok) {
-      throw newResult.error;
-    }
-
-    const response = await this.materializeResponse(newResult.value, turnNumber);
+    const newValue = await this.sendModelRequest(newRequest);
+    const response = await this.materializeResponse(newValue, turnNumber);
     await this.emitPostModelCall(response, startedAt, 'retry', turnNumber);
     return response;
   }
@@ -5322,17 +5302,22 @@ export class ModelResult<
    * Also the per-request fail-fast checkpoint: an already-aborted run
    * signal throws its abort reason here, before any network I/O.
    */
-  private dispatchRequestOptions(): RequestOptions | undefined {
+  private dispatchRequestOptions(turnSignal?: AbortSignal): RequestOptions | undefined {
     const base = this.options.options;
     const runSignal = this.options.signal;
-    if (!runSignal) {
+    if (!runSignal && !turnSignal) {
       return base;
     }
-    runSignal.throwIfAborted();
+    runSignal?.throwIfAborted();
 
-    const signals: AbortSignal[] = [
-      runSignal,
-    ];
+    const signals: AbortSignal[] = [];
+    if (runSignal) {
+      signals.push(runSignal);
+    }
+    // Per-request watchdog controller (see sendModelRequest).
+    if (turnSignal) {
+      signals.push(turnSignal);
+    }
     const callerSignal = base?.signal ?? base?.fetchOptions?.signal;
     if (callerSignal) {
       signals.push(callerSignal);
@@ -5346,10 +5331,50 @@ export class ModelResult<
     }
     return {
       ...base,
-      signal: signals.length === 1 ? runSignal : AbortSignal.any(signals),
+      signal: signals.length === 1 && signals[0] ? signals[0] : AbortSignal.any(signals),
     };
   }
 
+  /**
+   * Dispatch one model request: the single entry point for every
+   * betaResponsesSend call of the run (initial, tool-round follow-up, final,
+   * retry, resume), so stall deadlines re-arm per request.
+   *
+   * With `streamTimeout` configured, the request gets its own
+   * AbortController (composed with the run signal, caller signal and
+   * timeoutMs by `dispatchRequestOptions`), and a streaming value is wrapped
+   * in the watchdog. On a stall the watchdog aborts that controller, tearing
+   * down the hung connection, and the stream errors with StreamStalledError,
+   * which every consumer of the turn then observes. The watchdog is disarmed
+   * by an abort of the composed signal, so a user abort or `timeoutMs`
+   * surfaces as itself.
+   *
+   * Returns the raw value (stream or non-streaming response); throws the
+   * SDK error on a non-ok result.
+   */
+  private async sendModelRequest(request: models.ResponsesRequest): Promise<unknown> {
+    const timeouts = this.options.streamTimeout;
+    const turnAbort = hasActiveStreamTimeouts(timeouts) ? new AbortController() : undefined;
+    const requestOptions = this.dispatchRequestOptions(turnAbort?.signal);
+    const result = await betaResponsesSend(
+      this.options.client,
+      {
+        responsesRequest: request,
+      },
+      requestOptions,
+    );
+    if (!result.ok) {
+      throw result.error;
+    }
+    if (!turnAbort || !timeouts || !isEventStream(result.value)) {
+      return result.value;
+    }
+    return applyResponsesStreamWatchdog(result.value, timeouts, {
+      signal: requestOptions?.signal ?? undefined,
+      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.
@@ -5383,6 +5408,7 @@ export class ModelResult<
       streamReplay: _sr,
       signal: _sig,
       toolTimeoutMs: _ttm,
+      streamTimeout: _st,
       toolConcurrency: _tc,
       asyncTools: _at,
       activeTools: _activeTools,
@@ -5969,17 +5995,7 @@ export class ModelResult<
 
       // Make the API request
       this.beginToolChoiceDispatch(request.toolChoice);
-      const apiResult = await betaResponsesSend(
-        this.options.client,
-        {
-          responsesRequest: request,
-        },
-        this.dispatchRequestOptions(),
-      );
-
-      if (!apiResult.ok) {
-        throw apiResult.error;
-      }
+      const apiValue = await this.sendModelRequest(request);
 
       // Stash fresh user items so saveResponseToState can persist them
       // atomically with the assistant output. Writing them here would leave
@@ -5991,11 +6007,11 @@ 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.setReusableStream(apiResult.value);
-      } else if (this.isNonStreamingResponse(apiResult.value)) {
+      if (isEventStream(apiValue)) {
+        this.setReusableStream(apiValue);
+      } else if (this.isNonStreamingResponse(apiValue)) {
         // API returned a complete response directly - use it as the final response
-        this.finalResponse = apiResult.value;
+        this.finalResponse = apiValue;
         await this.emitPendingModelCallOnce(this.finalResponse);
       } else {
         throw new Error('Unexpected response type from API');
@@ -6285,23 +6301,13 @@ export class ModelResult<
 
     // Make the API request
     this.beginToolChoiceDispatch(request.toolChoice);
-    const apiResult = await betaResponsesSend(
-      this.options.client,
-      {
-        responsesRequest: request,
-      },
-      this.dispatchRequestOptions(),
-    );
-
-    if (!apiResult.ok) {
-      throw apiResult.error;
-    }
+    const apiValue = await this.sendModelRequest(request);
 
     // Handle both streaming and non-streaming responses
-    if (isEventStream(apiResult.value)) {
-      this.setReusableStream(apiResult.value);
-    } else if (this.isNonStreamingResponse(apiResult.value)) {
-      this.finalResponse = apiResult.value;
+    if (isEventStream(apiValue)) {
+      this.setReusableStream(apiValue);
+    } else if (this.isNonStreamingResponse(apiValue)) {
+      this.finalResponse = apiValue;
       await this.emitPendingModelCallOnce(this.finalResponse);
     } else {
       throw new Error('Unexpected response type from API');

packages/agent/src/lib/stream-runtime.ts

@@ -1,7 +1,7 @@
 /**
  * The response-stream runtime's façade: multi-consumer buffering
- * (`ReusableReadableStream`) and typed stream failures (`StreamFailedError`,
- * `StreamStalledError`).
+ * (`ReusableReadableStream`), the stall watchdog, and typed stream failures
+ * (`StreamFailedError`, `StreamStalledError`).
  *
  * These modules are one feature and the engine uses them together, so
  * `model-result.ts` imports them through here rather than reaching into
@@ -16,3 +16,5 @@
 export type { StreamReplay } from './reusable-stream.js';
 export { ReusableReadableStream } from './reusable-stream.js';
 export { StreamFailedError } from './stream-errors.js';
+export type { StreamTimeoutOptions } from './stream-watchdog.js';
+export { applyResponsesStreamWatchdog, hasActiveStreamTimeouts } from './stream-watchdog.js';