feat: opt-in pre-content stall retries (DEV-723 4/5)

src/lib/model-result.ts

@@ -74,7 +74,11 @@ import {
 import { StreamFailedError } from './stream-errors.js';
 import {
   applyResponsesStreamWatchdog,
+  awaitFirstContent,
   hasActiveStreamTimeouts,
+  isContentBearingStreamEvent,
+  isTerminalStreamEvent,
+  normalizeStallRetries,
   type StreamTimeoutOptions,
 } from './stream-watchdog.js';
 import { combineSignals } from './primitives.js';
@@ -913,32 +917,17 @@ export class ModelResult<
     };
 
     // Stall deadlines re-arm independently for every turn.
-    const turnWatch = this.createTurnWatchContext();
-    const newResult = await responsesSend(
-      this.options.client,
-      { responsesRequest: newRequest },
-      turnWatch.requestOptions,
-    );
-
-    if (!newResult.ok) {
-      throw newResult.error;
-    }
-
-    // Handle streaming or non-streaming response
-    const value = newResult.value;
-    if (isEventStream(value)) {
-      const followUpStream = new ReusableReadableStream(turnWatch.watch(value));
+    const turnResult = await this.sendTurnRequest(newRequest);
+    if (turnResult.kind === 'stream') {
+      const followUpStream = new ReusableReadableStream(turnResult.stream);
 
       if (this.turnBroadcaster) {
         return this.pipeAndConsumeStream(followUpStream, turnNumber);
       }
 
       return consumeStreamForCompletion(followUpStream);
-    } else if (this.isNonStreamingResponse(value)) {
-      return value;
-    } else {
-      throw new Error('Unexpected response type from API');
     }
+    return turnResult.response;
   }
 
   /**
@@ -958,6 +947,70 @@ export class ModelResult<
     }
   }
 
+  /**
+   * Send one turn's request with stall protection.
+   *
+   * Wraps `responsesSend` with the per-turn watchdog, and — when
+   * `timeout.maxStallRetries` is set — transparently re-issues the
+   * request on pre-content stalls. Retries are provably safe: the stream
+   * is only handed to the caller after its first content-bearing (or
+   * terminal) event arrives, so a discarded stalled attempt never leaked
+   * events downstream. Stalls after content started are never retried.
+   */
+  private async sendTurnRequest(
+    request: models.ResponsesRequest,
+  ): Promise<
+    | { kind: 'stream'; stream: ReadableStream<models.StreamEvents> }
+    | { kind: 'response'; response: models.OpenResponsesResult }
+  > {
+    const maxStallRetries = normalizeStallRetries(this.options.timeout);
+
+    for (let attempt = 0; ; attempt++) {
+      const turnWatch = this.createTurnWatchContext();
+      const apiResult = await responsesSend(
+        this.options.client,
+        { responsesRequest: request },
+        turnWatch.requestOptions,
+      );
+
+      if (!apiResult.ok) {
+        throw apiResult.error;
+      }
+
+      const value = apiResult.value;
+      if (this.isNonStreamingResponse(value)) {
+        return { kind: 'response', response: value };
+      }
+      if (!isEventStream(value)) {
+        throw new Error('Unexpected response type from API');
+      }
+
+      const watched = turnWatch.watch(value);
+
+      // Without retries, hand the watched stream straight through — events
+      // flow to consumers in real time exactly as before.
+      if (maxStallRetries === 0) {
+        return { kind: 'stream', stream: watched };
+      }
+
+      // With retries, hold the stream back until it proves alive. Metadata
+      // events are buffered and replayed, so consumers still see the
+      // complete event sequence of the winning attempt only.
+      const outcome = await awaitFirstContent(
+        watched,
+        (event) => isContentBearingStreamEvent(event) || isTerminalStreamEvent(event),
+      );
+      if (outcome.kind === 'live') {
+        return { kind: 'stream', stream: outcome.stream };
+      }
+      if (attempt >= maxStallRetries) {
+        throw outcome.error;
+      }
+      // Pre-content stall with budget left: the previous attempt's request
+      // was already aborted by its watchdog; go again.
+    }
+  }
+
   /**
    * Build the stall-detection context for one turn.
    *
@@ -1156,27 +1209,15 @@ export class ModelResult<
       // Force stream mode for initial request
       const request = this.resolvedRequest;
 
-      // Make the API request (with per-turn stall detection when configured)
-      const turnWatch = this.createTurnWatchContext();
-      const apiResult = await responsesSend(
-        this.options.client,
-        { responsesRequest: request },
-        turnWatch.requestOptions,
-      );
-
-      if (!apiResult.ok) {
-        throw apiResult.error;
-      }
-
-      // 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(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;
+      // Make the API request (with per-turn stall detection when configured).
+      // The API may return a non-streaming response even when stream: true
+      // is requested.
+      const turnResult = await this.sendTurnRequest(request);
+      if (turnResult.kind === 'stream') {
+        this.reusableStream = new ReusableReadableStream(turnResult.stream);
       } else {
-        throw new Error('Unexpected response type from API');
+        // API returned a complete response directly - use it as the final response
+        this.finalResponse = turnResult.response;
       }
     })();
 
@@ -1304,24 +1345,11 @@ export class ModelResult<
     this.resolvedRequest = 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 },
-      turnWatch.requestOptions,
-    );
-
-    if (!apiResult.ok) {
-      throw apiResult.error;
-    }
-
-    // Handle both streaming and non-streaming responses
-    if (isEventStream(apiResult.value)) {
-      this.reusableStream = new ReusableReadableStream(turnWatch.watch(apiResult.value));
-    } else if (this.isNonStreamingResponse(apiResult.value)) {
-      this.finalResponse = apiResult.value;
+    const turnResult = await this.sendTurnRequest(request);
+    if (turnResult.kind === 'stream') {
+      this.reusableStream = new ReusableReadableStream(turnResult.stream);
     } else {
-      throw new Error('Unexpected response type from API');
+      this.finalResponse = turnResult.response;
     }
   }
 

src/lib/stream-watchdog.ts

@@ -45,6 +45,14 @@ export type StreamTimeoutOptions = {
    * window — that is `firstContentMs`'s job. Unset by default.
    */
   contentIntervalMs?: number | undefined;
+  /**
+   * How many times to transparently re-issue a turn's request when it
+   * stalls before producing any content (`callModel` only; raw-stream
+   * helpers ignore it). Only pre-content stalls are retried — they are
+   * provably safe because no output has been observed. Stalls after
+   * content started are never retried. Defaults to 0 (no retries).
+   */
+  maxStallRetries?: number | undefined;
 };
 
 /**
@@ -283,6 +291,145 @@ export function applyResponsesStreamWatchdog(
   });
 }
 
+/**
+ * True when a chat-completions stream chunk carries model output: a choice
+ * delta with content, reasoning, refusal, tool-call arguments, or audio.
+ * Role-only preludes (`delta: { role: 'assistant' }`) and usage-only
+ * chunks are neutral.
+ */
+export function isContentBearingChatChunk(chunk: models.ChatStreamChunk): boolean {
+  return chunk.choices.some((choice) => {
+    const delta = choice.delta;
+    if (!delta) {
+      return false;
+    }
+    return (
+      (typeof delta.content === 'string' && delta.content.length > 0) ||
+      (typeof delta.reasoning === 'string' && delta.reasoning.length > 0) ||
+      (typeof delta.refusal === 'string' && delta.refusal.length > 0) ||
+      (delta.reasoningDetails !== undefined && delta.reasoningDetails.length > 0) ||
+      (delta.toolCalls !== undefined && delta.toolCalls.length > 0) ||
+      delta.audio !== undefined
+    );
+  });
+}
+
+/**
+ * True when a chat-completions stream chunk signals the response is
+ * finishing: a non-null finish reason on any choice, or a chunk-level
+ * error payload.
+ */
+export function isTerminalChatChunk(chunk: models.ChatStreamChunk): boolean {
+  if (chunk.error !== undefined) {
+    return true;
+  }
+  return chunk.choices.some((choice) => choice.finishReason !== null && choice.finishReason !== undefined);
+}
+
+/**
+ * Convenience wrapper of {@link applyStreamWatchdog} for chat-completions
+ * chunk streams, using the standard chunk classification.
+ */
+export function applyChatStreamWatchdog(
+  source: ReadableStream<models.ChatStreamChunk>,
+  timeouts: StreamTimeoutOptions,
+  hooks?: { onStall?: ((error: StreamStalledError) => void) | undefined },
+): ReadableStream<models.ChatStreamChunk> {
+  return applyStreamWatchdog(source, timeouts, {
+    isContentEvent: isContentBearingChatChunk,
+    isTerminalEvent: isTerminalChatChunk,
+    onStall: hooks?.onStall,
+  });
+}
+
+/**
+ * Outcome of waiting for a stream's first committed (content or terminal)
+ * event. `live` carries a stream that replays everything observed so far
+ * followed by the remainder of the source. `stalled` means the watchdog
+ * fired before any content: nothing was handed downstream, so the caller
+ * can safely retry the whole request.
+ */
+export type FirstContentOutcome<T> =
+  | { kind: 'live'; stream: ReadableStream<T> }
+  | { kind: 'stalled'; error: StreamStalledError };
+
+/**
+ * Read from `source` until an event satisfying `isCommitEvent` arrives
+ * (or the stream closes), buffering everything seen. Used to make
+ * pre-content stall retries safe: the returned stream only exists once
+ * the attempt has proven alive, so a discarded attempt never leaks
+ * events downstream.
+ *
+ * Non-stall errors and post-content stalls propagate as rejections.
+ */
+export async function awaitFirstContent<T>(
+  source: ReadableStream<T>,
+  isCommitEvent: (event: T) => boolean,
+): Promise<FirstContentOutcome<T>> {
+  const reader = source.getReader();
+  const buffered: T[] = [];
+  try {
+    while (true) {
+      const result = await reader.read();
+      if (result.done) {
+        return { kind: 'live', stream: replayThenPipe(buffered, reader) };
+      }
+      buffered.push(result.value);
+      if (isCommitEvent(result.value)) {
+        return { kind: 'live', stream: replayThenPipe(buffered, reader) };
+      }
+    }
+  } catch (error) {
+    if (error instanceof StreamStalledError && !error.receivedAnyContent) {
+      return { kind: 'stalled', error };
+    }
+    throw error;
+  }
+}
+
+/**
+ * A stream that replays `events`, then pipes the remainder of `reader`.
+ * Reading a finished reader resolves `done`, so this also covers sources
+ * that closed during buffering.
+ */
+function replayThenPipe<T>(events: T[], reader: ReadableStreamDefaultReader<T>): ReadableStream<T> {
+  return new ReadableStream<T>({
+    start(controller) {
+      for (const event of events) {
+        controller.enqueue(event);
+      }
+      void (async () => {
+        try {
+          while (true) {
+            const result = await reader.read();
+            if (result.done) {
+              controller.close();
+              return;
+            }
+            controller.enqueue(result.value);
+          }
+        } catch (error) {
+          controller.error(error);
+        }
+      })();
+    },
+    cancel(reason) {
+      return reader.cancel(reason);
+    },
+  });
+}
+
+/**
+ * Clamp `maxStallRetries` to a non-negative integer (0 = disabled).
+ */
+export function normalizeStallRetries(timeouts: StreamTimeoutOptions | undefined): number {
+  const value = timeouts?.maxStallRetries;
+  if (value === undefined || !Number.isFinite(value) || value <= 0) {
+    return 0;
+  }
+  return Math.floor(value);
+}
+
 /**
  * True when at least one watchdog deadline is enabled (set, finite, > 0).
  * Callers use this to skip per-turn abort plumbing entirely when the
@@ -308,3 +455,4 @@ function normalizeTimeout(value: number | undefined): number | undefined {
   }
   return value;
 }
+