diff --git a/.changeset/stream-timeout-option.md b/.changeset/stream-timeout-option.md new file mode 100644 index 00000000..1fb9bb07 --- /dev/null +++ b/.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. diff --git a/packages/agent/README.md b/packages/agent/README.md index 52f480df..60d77d82 100644 --- a/packages/agent/README.md +++ b/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 diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index 820901e2..52a6a8b0 100644 --- a/packages/agent/src/index.ts +++ b/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, diff --git a/packages/agent/src/inner-loop/call-model.ts b/packages/agent/src/inner-loop/call-model.ts index 73d2d8ef..b0c682fe 100644 --- a/packages/agent/src/inner-loop/call-model.ts +++ b/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, }; diff --git a/packages/agent/src/lib/async-params.ts b/packages/agent/src/lib/async-params.ts index f2f72e71..fdba6566 100644 --- a/packages/agent/src/lib/async-params.ts +++ b/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 {}); 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 { + 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'); diff --git a/packages/agent/src/lib/stream-runtime.ts b/packages/agent/src/lib/stream-runtime.ts index d7b7439a..66c47573 100644 --- a/packages/agent/src/lib/stream-runtime.ts +++ b/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'; diff --git a/packages/agent/tests/unit/call-model-stream-timeout.test.ts b/packages/agent/tests/unit/call-model-stream-timeout.test.ts new file mode 100644 index 00000000..bc7ca6ca --- /dev/null +++ b/packages/agent/tests/unit/call-model-stream-timeout.test.ts @@ -0,0 +1,432 @@ +/** + * callModel `streamTimeout` (DEV-723): the stream watchdog armed per model + * request, composed with the run signal, aborting the hung request, and + * surfacing StreamStalledError on every consumption path. + * + * betaResponsesSend is mocked; each mocked request errors its body stream + * when its request signal aborts, like a real fetch body does. + */ +import type { OpenRouterCore } from '@openrouter/sdk/core'; +import type * as models from '@openrouter/sdk/models'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { z } from 'zod/v4'; + +const mockBetaResponsesSend = vi.hoisted(() => vi.fn()); + +vi.mock('@openrouter/sdk/funcs/betaResponsesSend', () => ({ + betaResponsesSend: mockBetaResponsesSend, +})); + +import { StreamStalledError } from '../../src/index.js'; +import { callModel } from '../../src/inner-loop/call-model.js'; +import { tool } from '../../src/lib/tool.js'; +import type { ControlledStream } from '../utils/stream-fixtures.js'; +import { + completedEvent, + controlledStream, + createdEvent, + drain, + emptyMessageShell, + functionCallArgsDelta, + healthyTextEvents, + healthyToolCallEvents, + textDelta, + textResponse, + toolCallResponse, +} from '../utils/stream-fixtures.js'; + +const client = { + _options: {}, +} as OpenRouterCore; + +const echoTool = tool({ + name: 'echo', + inputSchema: z.object({ + value: z.string(), + }), + outputSchema: z.object({ + echoed: z.string(), + }), + execute: async ({ value }) => ({ + echoed: value, + }), +}); + +interface Dispatch { + request: models.ResponsesRequest; + signal: AbortSignal | undefined; + options: Record | undefined; +} + +const dispatches: Dispatch[] = []; + +/** Answer the next requests with these streams, in order. */ +function serve(...streams: ControlledStream[]): void { + for (const source of streams) { + mockBetaResponsesSend.mockImplementationOnce( + async ( + _client: unknown, + params: { + responsesRequest: models.ResponsesRequest; + }, + options?: { + signal?: AbortSignal; + }, + ) => { + const signal = options?.signal; + dispatches.push({ + request: params.responsesRequest, + signal, + options, + }); + signal?.addEventListener('abort', () => source.error(signal.reason)); + return { + ok: true, + value: source.stream, + }; + }, + ); + } +} + +/** Settle a promise into a value without letting it reject unobserved. */ +function settle(promise: Promise): Promise<{ + value?: T; + error?: unknown; +}> { + return promise.then( + (value) => ({ + value, + }), + (error: unknown) => ({ + error, + }), + ); +} + +const TIMEOUT = { + firstContentMs: 100, + contentIntervalMs: 100, +}; + +beforeEach(() => { + vi.useFakeTimers(); + mockBetaResponsesSend.mockReset(); + dispatches.length = 0; +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +describe('streamTimeout: single turn', () => { + it('fails fast when a stream sends only metadata, then goes silent', async () => { + const source = controlledStream([ + createdEvent(), + emptyMessageShell(), + ]); + serve(source); + const outcome = settle( + callModel(client, { + model: 'test-model', + input: 'hi', + streamTimeout: TIMEOUT, + }).getText(), + ); + + await vi.advanceTimersByTimeAsync(99); + await vi.advanceTimersByTimeAsync(1); + const { error } = await outcome; + expect(error).toBeInstanceOf(StreamStalledError); + expect(error).toMatchObject({ + phase: 'first_content', + timeoutMs: 100, + receivedAnyContent: false, + retryable: true, + }); + // The hung request is torn down: its signal aborted with the stall, + // which errors (tears down) its body stream. + expect(dispatches[0]?.signal?.aborted).toBe(true); + expect(dispatches[0]?.signal?.reason).toBe(error); + expect(source.finished).toBe(true); + expect(vi.getTimerCount()).toBe(0); + }); + + it('a healthy stream completes and leaves no timers behind', async () => { + serve(controlledStream(healthyTextEvents('hello'))); + const text = await callModel(client, { + model: 'test-model', + input: 'hi', + streamTimeout: TIMEOUT, + }).getText(); + expect(text).toBe('hello'); + expect(vi.getTimerCount()).toBe(0); + }); + + it('is off by default: no watchdog, no extra request signal', async () => { + const source = controlledStream([ + createdEvent(), + ]); + serve(source); + const outcome = settle( + callModel(client, { + model: 'test-model', + input: 'hi', + }).getText(), + ); + await vi.advanceTimersByTimeAsync(10 * 60_000); + expect(dispatches[0]?.signal).toBeUndefined(); + source.push(emptyMessageShell(), textDelta('late'), completedEvent(textResponse('late'))); + source.close(); + expect((await outcome).value).toBe('late'); + }); + + it('stalls between content events once generation started (not retryable)', async () => { + const source = controlledStream([ + createdEvent(), + emptyMessageShell(), + textDelta('partial outp'), + ]); + serve(source); + const result = callModel(client, { + model: 'test-model', + input: 'hi', + streamTimeout: TIMEOUT, + }); + const streamed = settle(drain(result.getTextStream())); + await vi.advanceTimersByTimeAsync(100); + const { value } = await streamed; + expect(value?.items).toEqual([ + 'partial outp', + ]); + expect(value?.error).toBeInstanceOf(StreamStalledError); + expect(value?.error).toMatchObject({ + phase: 'between_content', + receivedAnyContent: true, + retryable: false, + }); + }); + + it('rejects every concurrent consumer with the same stall', async () => { + serve( + controlledStream([ + createdEvent(), + ]), + ); + const result = callModel(client, { + model: 'test-model', + input: 'hi', + streamTimeout: TIMEOUT, + }); + const outcomes = [ + settle(result.getText()), + settle(drain(result.getTextStream()).then((r) => Promise.reject(r.error))), + settle(drain(result.getFullResponsesStream()).then((r) => Promise.reject(r.error))), + settle(drain(result.getItemsStream()).then((r) => Promise.reject(r.error))), + settle(drain(result.getReasoningStream()).then((r) => Promise.reject(r.error))), + ]; + await vi.advanceTimersByTimeAsync(100); + const errors = (await Promise.all(outcomes)).map((o) => o.error); + for (const error of errors) { + expect(error).toBeInstanceOf(StreamStalledError); + } + expect(new Set(errors).size).toBe(1); + expect(mockBetaResponsesSend).toHaveBeenCalledTimes(1); + }); + + it('a run-signal abort is reported as the abort, never as a stall', async () => { + const controller = new AbortController(); + serve( + controlledStream([ + createdEvent(), + ]), + ); + const outcome = settle( + callModel(client, { + model: 'test-model', + input: 'hi', + signal: controller.signal, + streamTimeout: TIMEOUT, + }).getText(), + ); + await vi.advanceTimersByTimeAsync(50); + const reason = new Error('user cancelled'); + controller.abort(reason); + await vi.advanceTimersByTimeAsync(1_000); + expect((await outcome).error).toBe(reason); + expect(dispatches[0]?.signal?.reason).toBe(reason); + expect(vi.getTimerCount()).toBe(0); + }); + + it('keeps RequestOptions (timeoutMs, headers) and composes the watchdog signal in', async () => { + serve(controlledStream(healthyTextEvents('ok'))); + await callModel( + client, + { + model: 'test-model', + input: 'hi', + streamTimeout: TIMEOUT, + }, + { + timeoutMs: 60_000, + }, + ).getText(); + const dispatch = dispatches[0]; + expect(dispatch?.options?.['timeoutMs']).toBe(60_000); + expect(dispatch?.signal).toBeInstanceOf(AbortSignal); + expect(dispatch?.signal?.aborted).toBe(false); + }); + + it('never sends streamTimeout to the API (static and async-resolved requests)', async () => { + serve(controlledStream(healthyTextEvents('a')), controlledStream(healthyTextEvents('b'))); + await callModel(client, { + model: 'test-model', + input: 'hi', + streamTimeout: TIMEOUT, + }).getText(); + await callModel(client, { + model: 'test-model', + input: 'hi', + temperature: async () => 0.5, + streamTimeout: TIMEOUT, + }).getText(); + for (const dispatch of dispatches) { + expect(dispatch.request).not.toHaveProperty('streamTimeout'); + } + expect(dispatches[1]?.request.temperature).toBe(0.5); + }); +}); + +describe('streamTimeout: multi-turn tool loop', () => { + /** Turn 1 produces its tool call after `delayMs`, then completes. */ + function slowToolTurn(delayMs: number): { + source: ControlledStream; + release: () => Promise; + } { + const source = controlledStream([ + createdEvent(), + ]); + return { + source, + release: async () => { + await vi.advanceTimersByTimeAsync(delayMs); + source.push( + functionCallArgsDelta('{"value":"a"}'), + completedEvent( + toolCallResponse('echo', { + value: 'a', + }), + ), + ); + source.close(); + }, + }; + } + + function toolsRun() { + return callModel(client, { + model: 'test-model', + input: 'hi', + tools: [ + echoTool, + ] as const, + streamTimeout: TIMEOUT, + }); + } + + it('re-arms deadlines per request: turn 1 passes, a stalled turn 2 fails', async () => { + const turn1 = slowToolTurn(80); + const turn2 = controlledStream([ + createdEvent(), + ]); + serve(turn1.source, turn2); + const outcome = settle(toolsRun().getText()); + + await turn1.release(); + // Turn 2 is dispatched; its own deadline starts now. + await vi.advanceTimersByTimeAsync(99); + expect(mockBetaResponsesSend).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(1); + + const { error } = await outcome; + expect(error).toBeInstanceOf(StreamStalledError); + expect(dispatches[0]?.signal).not.toBe(dispatches[1]?.signal); + expect(dispatches[0]?.signal?.aborted).toBe(false); + expect(dispatches[1]?.signal?.aborted).toBe(true); + expect(turn2.finished).toBe(true); + }); + + it('total run time beyond the deadline is fine when every turn is within it', async () => { + const turn1 = slowToolTurn(80); + const turn2 = controlledStream([ + createdEvent(), + ]); + serve(turn1.source, turn2); + const outcome = settle(toolsRun().getText()); + await turn1.release(); + await vi.advanceTimersByTimeAsync(80); + turn2.push(emptyMessageShell(), textDelta('done'), completedEvent(textResponse('done'))); + turn2.close(); + expect((await outcome).value).toBe('done'); + }); + + it('a follow-up stall rejects concurrent stream consumers', async () => { + const turn2 = controlledStream([ + createdEvent(), + ]); + serve( + controlledStream( + healthyToolCallEvents('echo', { + value: 'a', + }), + ), + turn2, + ); + const result = toolsRun(); + const text = settle(drain(result.getTextStream())); + const full = settle(drain(result.getFullResponsesStream())); + const tools = settle(drain(result.getToolStream())); + await vi.advanceTimersByTimeAsync(100); + for (const outcome of await Promise.all([ + text, + full, + tools, + ])) { + expect(outcome.value?.error).toBeInstanceOf(StreamStalledError); + } + const fullEvents = (await full).value?.items ?? []; + // Turn 1 completed and its tool ran before turn 2 stalled. + expect(fullEvents.some((e) => e.type === 'tool.result')).toBe(true); + }); + + it('a consumer that walks away early does not leave an unhandled rejection', async () => { + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => { + unhandled.push(reason); + }; + process.on('unhandledRejection', onUnhandled); + try { + serve( + controlledStream( + healthyToolCallEvents('echo', { + value: 'a', + }), + ), + controlledStream([ + createdEvent(), + ]), + ); + const result = toolsRun(); + for await (const _event of result.getFullResponsesStream()) { + break; + } + await vi.advanceTimersByTimeAsync(500); + // Let Node flush its unhandled-rejection bookkeeping. + vi.useRealTimers(); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(unhandled).toEqual([]); + } finally { + process.off('unhandledRejection', onUnhandled); + } + }); +}); diff --git a/packages/agent/tests/utils/stream-fixtures.ts b/packages/agent/tests/utils/stream-fixtures.ts index f56ef8f6..1568d4a0 100644 --- a/packages/agent/tests/utils/stream-fixtures.ts +++ b/packages/agent/tests/utils/stream-fixtures.ts @@ -191,6 +191,8 @@ export interface ControlledStream { error: (reason: unknown) => void; readonly cancelled: boolean; readonly cancelReason: unknown; + /** Closed, errored or cancelled: the source is torn down either way. */ + readonly finished: boolean; } export function controlledStream(initial: T[] = []): ControlledStream { @@ -239,6 +241,9 @@ export function controlledStream(initial: T[] = []): Co get cancelReason() { return cancelReason; }, + get finished() { + return done; + }, }; }