diff --git a/src/lib/model-result.ts b/src/lib/model-result.ts index 637fb09e2..8d0df98bb 100644 --- a/src/lib/model-result.ts +++ b/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 } + | { 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; } } diff --git a/src/lib/stream-watchdog.ts b/src/lib/stream-watchdog.ts index 1be495a19..1ae6f1729 100644 --- a/src/lib/stream-watchdog.ts +++ b/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, + timeouts: StreamTimeoutOptions, + hooks?: { onStall?: ((error: StreamStalledError) => void) | undefined }, +): ReadableStream { + 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 = + | { kind: 'live'; stream: ReadableStream } + | { 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( + source: ReadableStream, + isCommitEvent: (event: T) => boolean, +): Promise> { + 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(events: T[], reader: ReadableStreamDefaultReader): ReadableStream { + return new ReadableStream({ + 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; } + diff --git a/tests/unit/call-model-stream-timeout.test.ts b/tests/unit/call-model-stream-timeout.test.ts index a7c343dc7..622659fbc 100644 --- a/tests/unit/call-model-stream-timeout.test.ts +++ b/tests/unit/call-model-stream-timeout.test.ts @@ -376,6 +376,85 @@ describe('callModel stream timeout integration', () => { expect(observed.requests[1]?.signal.aborted).toBe(true); }); + it('maxStallRetries re-issues a pre-content stall and succeeds on the retry', async () => { + const { client, observed } = scriptedClient([ + [frame(createdFrame, 5)], // attempt 1: stalls before content + HEALTHY_SCRIPT, // attempt 2: healthy + ]); + + const result = client.callModel({ + model: 'test-model', + input: 'hi', + timeout: { firstContentMs: 60, maxStallRetries: 1 }, + }); + + expect(await result.getText()).toBe('Hello world'); + expect(observed.requests).toHaveLength(2); + expect(observed.requests[0]?.signal.aborted).toBe(true); // stalled attempt torn down + expect(observed.requests[1]?.signal.aborted).toBe(false); + }); + + it('maxStallRetries respects the retry budget and rethrows the final stall', async () => { + const { client, observed } = scriptedClient([ + [frame(createdFrame, 5)], // attempt 1: stalls + [frame(createdFrame, 5)], // attempt 2 (retry 1): stalls + ]); + + const result = client.callModel({ + model: 'test-model', + input: 'hi', + timeout: { firstContentMs: 50, maxStallRetries: 1 }, + }); + + const error = await result.getText().then( + () => null, + (e: unknown) => e, + ); + expect(error).toBeInstanceOf(StreamStalledError); + expect(observed.requests).toHaveLength(2); // initial + exactly one retry + }); + + it('maxStallRetries never retries a mid-content stall', async () => { + const { client, observed } = scriptedClient([ + [frame(createdFrame, 5), frame(textDeltaFrame('partial'), 5)], // content, then stall + ]); + + const result = client.callModel({ + model: 'test-model', + input: 'hi', + timeout: { firstContentMs: 300, contentIntervalMs: 60, maxStallRetries: 3 }, + }); + + const error = await result.getText().then( + () => null, + (e: unknown) => e, + ); + expect(error).toBeInstanceOf(StreamStalledError); + expect((error as StreamStalledError).phase).toBe('between_content'); + // No second request: output was already observed, retrying could duplicate it. + expect(observed.requests).toHaveLength(1); + }); + + it('streaming consumers see each event exactly once when a retry succeeds', async () => { + const { client } = scriptedClient([ + [frame(createdFrame, 5)], // attempt 1: stalls + HEALTHY_SCRIPT, // attempt 2: healthy + ]); + + const result = client.callModel({ + model: 'test-model', + input: 'hi', + timeout: { firstContentMs: 60, maxStallRetries: 1 }, + }); + + const deltas: string[] = []; + for await (const delta of result.getTextStream()) { + deltas.push(delta); + } + // Only the winning attempt's deltas, no duplicates from the stalled one. + expect(deltas).toEqual(['Hello', ' world']); + }); + it('caller abort signals still work alongside the watchdog', async () => { const { client } = scriptedClient([ [frame(createdFrame, 5)], // hangs, but watchdog is generous