diff --git a/.changeset/typed-stream-errors.md b/.changeset/typed-stream-errors.md new file mode 100644 index 00000000..ffed9ca0 --- /dev/null +++ b/.changeset/typed-stream-errors.md @@ -0,0 +1,21 @@ +--- +'@openrouter/agent': minor +--- + +Stream failures are now typed. `response.failed` events, stream-level `error` events, and streams that end without a terminal event surface as `StreamFailedError` (with `kind`, `code`, `errorType`, `response` and a `retryable` hint) on every consumption path, instead of a bare `Error`. Both consume paths (the initial stream and follow-up tool turns) now handle `error` events the same way: an `error` event followed by a completion still resolves, and one followed by the stream ending rejects with that error. + +Without tools, the stream getters (`getTextStream()`, `getFullResponsesStream()`, `getItemsStream()`, `getReasoningStream()`, `getToolStream()`, `getToolCallsStream()`) used to end silently when the response failed or was truncated. They now reject with the same `StreamFailedError` that `getText()` throws. + +Also exports `StreamStalledError`, which the opt-in stream watchdog raises (wired into `callModel` in a follow-up release). + +```ts +import { StreamFailedError } from '@openrouter/agent'; + +try { + await result.getText(); +} catch (error) { + if (error instanceof StreamFailedError && error.retryable) { + // transient server failure (server_error, rate limit, provider overloaded, ...) + } +} +``` diff --git a/packages/agent/README.md b/packages/agent/README.md index adf4067d..52f480df 100644 --- a/packages/agent/README.md +++ b/packages/agent/README.md @@ -353,6 +353,31 @@ 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. +#### Stream failures + +A response stream that fails surfaces as a typed `StreamFailedError` on every +consumption path (`getText()`, `getResponse()` and all the stream getters): + +```typescript +import { StreamFailedError } from '@openrouter/agent'; + +try { + for await (const delta of result.getTextStream()) process.stdout.write(delta); +} catch (error) { + if (error instanceof StreamFailedError) { + error.kind; // 'response_failed' | 'error_event' | 'truncated' + error.code; // e.g. 'server_error', when the server reported one + error.retryable; // true for transient codes (server_error, rate limits, ...) + } +} +``` + +- `response_failed`: the server sent `response.failed` (`error.response` holds it). +- `error_event`: the server sent a stream-level `error` event and the stream + ended without completing. An `error` event followed by a completion is not a + failure. +- `truncated`: the stream ended with no terminal event at all. + ### Doom-Loop Detection Catch runs that stop making progress while continuing to spend: the model diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index ca6853fa..820901e2 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -219,6 +219,8 @@ export { maxTokensUsed, stepCountIs, } from './lib/stop-conditions.js'; +export type { StreamFailureKind, StreamStallPhase } from './lib/stream-errors.js'; +export { StreamFailedError, StreamStalledError } from './lib/stream-errors.js'; export type { StreamableOutputItem } from './lib/stream-transformers.js'; export { extractUnsupportedContent, diff --git a/packages/agent/src/lib/model-result.ts b/packages/agent/src/lib/model-result.ts index bbc4f047..ca285176 100644 --- a/packages/agent/src/lib/model-result.ts +++ b/packages/agent/src/lib/model-result.ts @@ -59,9 +59,9 @@ import { applyNextTurnParamsToRequest, executeNextTurnParamsFunctions, } from './next-turn-params.js'; -import type { StreamReplay } from './reusable-stream.js'; -import { ReusableReadableStream } from './reusable-stream.js'; import { isStopConditionMet } from './stop-conditions.js'; +import type { StreamReplay } from './stream-runtime.js'; +import { ReusableReadableStream, StreamFailedError } from './stream-runtime.js'; import type { ItemInProgress, StreamableOutputItem } from './stream-transformers.js'; import { buildItemsStream, @@ -82,6 +82,7 @@ import { } from './stream-transformers.js'; import { hasTypeProperty, + isErrorEvent, isFunctionCallItem, isFunctionCallOutputItem, isOutputTextDeltaEvent, @@ -646,6 +647,11 @@ export class ModelResult< private initialPipePromise: Promise | null = null; private initialResponse: models.OpenResponsesResult | null = null; private initialResponseError: Error | null = null; + // Last stream-level `error` event seen on the initial stream. Surfaced as + // a StreamFailedError only if the stream ends without a terminal event. + private initialErrorEvent: models.ErrorEvent | null = null; + // Set by cancel(): a deliberately cancelled stream is not a failure. + private streamCancelled = false; private readonly streamReplay: StreamReplay; // Context store for typed tool context (persists across turns) @@ -954,15 +960,38 @@ export class ModelResult< } if (isResponseFailedEvent(event)) { - this.initialResponseError = new Error( - `Response failed: ${JSON.stringify(event.response.error)}`, - ); + this.initialResponseError = StreamFailedError.fromFailedResponse(event.response); + return; + } + + if (isErrorEvent(event)) { + this.initialErrorEvent = event; + } + } + + /** + * No-tools stream getters iterate the initial stream directly, and their + * consumers end cleanly on any terminal event or source close, including a + * failed response or a truncated stream. Re-surface those failures so the + * getters reject exactly like `getText()` does instead of ending silently. + * Call only after the consumer ran to its natural end (not after `break`). + */ + private throwIfInitialStreamFailed(): void { + if (this.streamCancelled || this.finalResponse || !this.reusableStream) { + return; + } + if (this.initialResponseError) { + throw this.initialResponseError; + } + if (!this.initialResponse) { + throw StreamFailedError.fromStreamEnd(this.initialErrorEvent); } } private setReusableStream(stream: ReadableStream): void { this.initialResponse = null; this.initialResponseError = null; + this.initialErrorEvent = null; this.reusableStream = new ReusableReadableStream(stream, { streamReplay: this.streamReplay, onValue: (event) => this.captureInitialStreamEvent(event), @@ -988,6 +1017,9 @@ export class ModelResult< const consumer = stream.createConsumer(); let completedResponse: models.OpenResponsesResult | null = null; + // Recorded, not thrown: the stream may still complete after an error + // event. Surfaced only when it ends without a completed response. + let errorEvent: models.ErrorEvent | null = null; for await (const event of consumer) { broadcaster.push(event); @@ -995,12 +1027,14 @@ export class ModelResult< completedResponse = event.response; } if (isResponseFailedEvent(event)) { - const errorMsg = 'message' in event ? String(event.message) : 'Response failed'; - throw new Error(errorMsg); + throw StreamFailedError.fromFailedResponse(event.response); } if (isResponseIncompleteEvent(event)) { completedResponse = event.response; } + if (isErrorEvent(event)) { + errorEvent = event; + } } broadcaster.push({ @@ -1010,7 +1044,10 @@ export class ModelResult< } satisfies TurnEndEvent); if (!completedResponse) { - throw new Error('Follow-up stream ended without a completed response'); + throw StreamFailedError.fromStreamEnd( + errorEvent, + 'Follow-up stream ended without a completed response', + ); } return completedResponse; @@ -1170,7 +1207,18 @@ export class ModelResult< } if (this.reusableStream) { - const response = await consumeStreamForCompletion(this.reusableStream); + const response = await consumeStreamForCompletion(this.reusableStream).catch((error) => { + // With `streamReplay: 'active-consumers'` the replay may no longer + // hold the error event; the onValue capture still does. + if ( + error instanceof StreamFailedError && + error.kind === 'truncated' && + this.initialErrorEvent + ) { + throw StreamFailedError.fromErrorEvent(this.initialErrorEvent); + } + throw error; + }); await this.emitPendingModelCallOnce(response); return response; } @@ -6819,6 +6867,7 @@ export class ModelResult< yield event; } } + this.throwIfInitialStreamFailed(); } catch (error) { streamFailed = true; throw error; @@ -6853,6 +6902,7 @@ export class ModelResult< if (this.reusableStream) { yield* extractTextDeltas(this.reusableStream); } + this.throwIfInitialStreamFailed(); } catch (error) { streamFailed = true; throw error; @@ -6942,6 +6992,7 @@ export class ModelResult< } } } + this.throwIfInitialStreamFailed(); } catch (error) { streamFailed = true; throw error; @@ -7177,6 +7228,7 @@ export class ModelResult< if (this.reusableStream) { yield* extractReasoningDeltas(this.reusableStream); } + this.throwIfInitialStreamFailed(); } catch (error) { streamFailed = true; throw error; @@ -7219,6 +7271,7 @@ export class ModelResult< }; } } + this.throwIfInitialStreamFailed(); } catch (error) { streamFailed = true; throw error; @@ -7300,6 +7353,7 @@ export class ModelResult< ParsedToolCall >; } + this.throwIfInitialStreamFailed(); }.call(this); } @@ -7386,6 +7440,7 @@ export class ModelResult< * new with async tool support. */ async cancel(): Promise { + this.streamCancelled = true; this.abortAllToolWork('Cancelled via ModelResult.cancel()'); if (this.reusableStream) { await this.reusableStream.cancel(); diff --git a/packages/agent/src/lib/stream-errors.ts b/packages/agent/src/lib/stream-errors.ts new file mode 100644 index 00000000..77f2633c --- /dev/null +++ b/packages/agent/src/lib/stream-errors.ts @@ -0,0 +1,203 @@ +import type * as models from '@openrouter/sdk/models'; + +/** + * Typed stream errors (DEV-723). + * + * - {@link StreamStalledError}: the client-side stream watchdog + * (`stream-watchdog.ts`) fired because a response stream stayed open + * without producing content within a configured deadline. + * - {@link StreamFailedError}: the server reported a failure mid-stream + * (`response.failed`, a stream-level `error` event) or the stream ended + * without any terminal event. + * + * Both replace bare `Error`s so callers can branch on `instanceof` and on + * the `retryable` hint instead of string-matching messages. Message text is + * informational, not a contract. + * + * This module only has type imports from `@openrouter/sdk`, so it is safe to + * load from startup-sensitive code paths. + */ + +/** + * Which watchdog deadline expired. + * + * - `first_content`: the response stream started (headers received) but no + * content-bearing event arrived within `firstContentMs`. + * - `between_content`: content started flowing, but the gap since the last + * content-bearing event exceeded `contentIntervalMs`. + */ +export type StreamStallPhase = 'first_content' | 'between_content'; + +/** + * Raised when a streaming response stalls: the connection stays open (and + * may still receive keep-alive framing or metadata events) but no + * content-bearing event arrives within the configured deadline. + */ +export class StreamStalledError extends Error { + override readonly name = 'StreamStalledError'; + + /** Which deadline expired. */ + readonly phase: StreamStallPhase; + /** The configured deadline that expired, in milliseconds. */ + readonly timeoutMs: number; + /** + * Milliseconds elapsed since the deadline was armed (stream start for + * `first_content`, the last content-bearing event for `between_content`). + */ + readonly elapsedMs: number; + /** Whether any content-bearing event was received before the stall. */ + readonly receivedAnyContent: boolean; + + constructor(options: { + phase: StreamStallPhase; + timeoutMs: number; + elapsedMs: number; + receivedAnyContent: boolean; + }) { + const message = + options.phase === 'first_content' + ? `Stream stalled: no content received within ${options.timeoutMs}ms of the response stream starting` + : `Stream stalled: no content received for ${options.timeoutMs}ms after the last content event`; + super(message); + this.phase = options.phase; + this.timeoutMs = options.timeoutMs; + this.elapsedMs = options.elapsedMs; + this.receivedAnyContent = options.receivedAnyContent; + Object.setPrototypeOf(this, StreamStalledError.prototype); + } + + /** + * Whether re-issuing the request cannot duplicate output: true only when + * the stall happened before any content was received. + */ + get retryable(): boolean { + return !this.receivedAnyContent; + } +} + +/** + * Where a {@link StreamFailedError} came from. + * + * - `response_failed`: a `response.failed` event (carries the failed response). + * - `error_event`: a stream-level `error` event, followed by the stream ending + * without a completion. + * - `truncated`: the stream ended with no terminal event and no error event + * (the connection closed mid-response). + */ +export type StreamFailureKind = 'response_failed' | 'error_event' | 'truncated'; + +/** + * Error codes that indicate a transient condition where retrying the request + * (ideally with backoff) is reasonable. Covers the Responses error-field codes + * (`server_error`, ...) and canonical OpenRouter error types + * (`provider_overloaded`, ...). + */ +const TRANSIENT_FAILURE_CODES: ReadonlySet = new Set([ + 'server_error', + 'server', + 'timeout', + 'rate_limit_exceeded', + 'provider_overloaded', + 'provider_unavailable', +]); + +/** + * Raised when a response stream fails: the server reported a failure + * (`response.failed` or a stream-level `error` event) or the stream ended + * without any terminal event. + */ +export class StreamFailedError extends Error { + override readonly name = 'StreamFailedError'; + + /** Where the failure came from. */ + readonly kind: StreamFailureKind; + /** Provider/API error code (e.g. `server_error`), when reported. */ + readonly code: string | null; + /** + * Canonical OpenRouter error type (e.g. `provider_unavailable`), when the + * failed response carries one. + */ + readonly errorType: string | null; + /** The failed response, when the failure came from `response.failed`. */ + readonly response: models.OpenResponsesResult | null; + + constructor(options: { + kind: StreamFailureKind; + message: string; + code?: string | null | undefined; + errorType?: string | null | undefined; + response?: models.OpenResponsesResult | null | undefined; + }) { + super(options.message); + this.kind = options.kind; + this.code = options.code ?? null; + this.errorType = options.errorType ?? null; + this.response = options.response ?? null; + Object.setPrototypeOf(this, StreamFailedError.prototype); + } + + /** + * Whether the failure looks transient (server error, timeout, rate limit, + * provider overloaded/unavailable), so a retry with backoff is reasonable. + * Validation-style failures and truncated streams return false: a truncated + * stream may already have delivered partial output. + */ + get retryable(): boolean { + return ( + (this.code !== null && TRANSIENT_FAILURE_CODES.has(this.code)) || + (this.errorType !== null && TRANSIENT_FAILURE_CODES.has(this.errorType)) + ); + } + + /** Build from a `response.failed` event's response payload. */ + static fromFailedResponse(response: models.OpenResponsesResult): StreamFailedError { + const code = typeof response.error?.code === 'string' ? response.error.code : null; + // Not modelled by every SDK version: read defensively. + const rawErrorType = ( + response as { + errorType?: unknown; + } + ).errorType; + const errorType = typeof rawErrorType === 'string' ? rawErrorType : null; + const detail = response.error?.message ?? 'no error detail provided'; + return new StreamFailedError({ + kind: 'response_failed', + message: `Response failed${code ? ` (${code})` : ''}: ${detail}`, + code, + errorType, + response, + }); + } + + /** Build from a stream-level `error` event. */ + static fromErrorEvent(event: { code: string | null; message: string }): StreamFailedError { + return new StreamFailedError({ + kind: 'error_event', + message: `Stream error${event.code ? ` (${event.code})` : ''}: ${event.message}`, + code: event.code, + }); + } + + /** + * Build for a stream that ended without a terminal event. Uses the most + * recent stream-level `error` event as the cause when one was seen. + */ + static fromStreamEnd( + errorEvent: + | { + code: string | null; + message: string; + } + | null + | undefined, + message = 'Stream ended without completion event', + ): StreamFailedError { + if (errorEvent) { + return StreamFailedError.fromErrorEvent(errorEvent); + } + return new StreamFailedError({ + kind: 'truncated', + message, + }); + } +} diff --git a/packages/agent/src/lib/stream-runtime.ts b/packages/agent/src/lib/stream-runtime.ts new file mode 100644 index 00000000..d7b7439a --- /dev/null +++ b/packages/agent/src/lib/stream-runtime.ts @@ -0,0 +1,18 @@ +/** + * The response-stream runtime's façade: multi-consumer buffering + * (`ReusableReadableStream`) 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 + * each. That also keeps it under the structural gate's fan-out ceiling + * (`no_god_files` trips above 15 outbound edges). Same pattern as + * `async-tools.ts`. + * + * Re-exports only, no logic. Anything needing a subset (tests, other lib + * modules) should still import the specific module directly. + */ + +export type { StreamReplay } from './reusable-stream.js'; +export { ReusableReadableStream } from './reusable-stream.js'; +export { StreamFailedError } from './stream-errors.js'; diff --git a/packages/agent/src/lib/stream-transformers.ts b/packages/agent/src/lib/stream-transformers.ts index 66dac32c..46024db1 100644 --- a/packages/agent/src/lib/stream-transformers.ts +++ b/packages/agent/src/lib/stream-transformers.ts @@ -7,7 +7,9 @@ import type { UnsupportedContent, } from '../api-shape-helpers/claude-message.js'; import type { ReusableReadableStream } from './reusable-stream.js'; +import { StreamFailedError } from './stream-errors.js'; import { + isErrorEvent, isFileCitationAnnotation, isFilePathAnnotation, isFileSearchCallOutputItem, @@ -662,12 +664,13 @@ export function extractCompletionFromBuffer( const terminal = findBufferedTerminalEvent(stream); if (terminal && isResponseFailedEvent(terminal)) { - throw new Error(`Response failed: ${JSON.stringify(terminal.response.error)}`); + throw StreamFailedError.fromFailedResponse(terminal.response); } if (terminal && (isResponseCompletedEvent(terminal) || isResponseIncompleteEvent(terminal))) { return terminal.response; } - throw new Error('Stream ended without completion event'); + const errorEvent = stream.findLastBuffered((event) => isErrorEvent(event)); + throw StreamFailedError.fromStreamEnd(errorEvent && isErrorEvent(errorEvent) ? errorEvent : null); } /** @@ -697,6 +700,9 @@ export async function consumeStreamForCompletion( stream: ReusableReadableStream, ): Promise { const consumer = stream.createConsumer(); + // Stream-level `error` events are recorded, not thrown: the stream may + // still complete after one. Surfaced only if it ends without a terminal. + let errorEvent: models.ErrorEvent | null = null; for await (const event of consumer) { if (!('type' in event)) { @@ -709,16 +715,20 @@ export async function consumeStreamForCompletion( if (isResponseFailedEvent(event)) { // The failed event contains the full response with error information - throw new Error(`Response failed: ${JSON.stringify(event.response.error)}`); + throw StreamFailedError.fromFailedResponse(event.response); } if (isResponseIncompleteEvent(event)) { // Return the incomplete response return event.response; } + + if (isErrorEvent(event)) { + errorEvent = event; + } } - throw new Error('Stream ended without completion event'); + throw StreamFailedError.fromStreamEnd(errorEvent); } /** diff --git a/packages/agent/src/lib/stream-type-guards.ts b/packages/agent/src/lib/stream-type-guards.ts index db29fccb..7fd6d14d 100644 --- a/packages/agent/src/lib/stream-type-guards.ts +++ b/packages/agent/src/lib/stream-type-guards.ts @@ -47,6 +47,11 @@ export function isResponseFailedEvent( return 'type' in event && event.type === 'response.failed'; } +/** A stream-level `error` event (not terminal: the stream may still complete). */ +export function isErrorEvent(event: models.StreamEvents): event is models.ErrorEvent { + return 'type' in event && event.type === 'error'; +} + export function isResponseIncompleteEvent( event: models.StreamEvents, ): event is models.StreamEventsResponseIncomplete { diff --git a/packages/agent/src/lib/stream-watchdog.ts b/packages/agent/src/lib/stream-watchdog.ts new file mode 100644 index 00000000..c5c0e12a --- /dev/null +++ b/packages/agent/src/lib/stream-watchdog.ts @@ -0,0 +1,342 @@ +import type * as models from '@openrouter/sdk/models'; +import type { StreamStallPhase } from './stream-errors.js'; +import { StreamStalledError } from './stream-errors.js'; + +/** + * Client-side stalled-stream detection (DEV-723). + * + * A streaming response can return headers quickly, emit only keep-alive + * framing or metadata events, and then stall indefinitely without producing + * content. Transport timeouts cannot catch this: SSE keep-alive comments + * reset socket idle timers even though no content is flowing. + * + * The watchdog wraps a parsed event stream and enforces two opt-in semantic + * deadlines: + * + * - `firstContentMs`: armed when the stream starts; satisfied (and + * permanently disarmed) by the first content-bearing event. + * - `contentIntervalMs`: once content has started, the maximum gap allowed + * between content-bearing events. + * + * Classification follows the parsed event type, not transport activity: + * empty role preludes (`response.output_item.added` for message shells), + * status events (`response.created`, `response.in_progress`, ...), and SSE + * keep-alive comments (dropped by the SSE parser before events reach this + * layer) neither satisfy nor reset a deadline. + * + * When a deadline expires the wrapped stream errors with + * {@link StreamStalledError} and the source stream is cancelled. + */ + +/** + * Opt-in stream stall deadlines. Both are unset by default (no watchdog). + * Non-finite or non-positive values disable the corresponding deadline. + */ +export type StreamTimeoutOptions = { + /** + * Maximum milliseconds between the response stream starting (headers + * received) and the first content-bearing event. + */ + firstContentMs?: number | undefined; + /** + * Maximum milliseconds between consecutive content-bearing events once + * content has started. Does not cover the pre-first-content window; that + * is `firstContentMs`. + */ + contentIntervalMs?: number | undefined; +}; + +/** Event classification and lifecycle hooks for {@link applyStreamWatchdog}. */ +export type StreamWatchdogHooks = { + /** + * True when the event carries model output (or a completed sub-result). + * Content events satisfy and re-arm deadlines. + */ + isContentEvent: (event: T) => boolean; + /** + * True when the event carries the server's final verdict (completed / + * failed / incomplete). Terminal events permanently disarm the watchdog + * so trailing bookkeeping events are never treated as stalls. + */ + isTerminalEvent?: ((event: T) => boolean) | undefined; + /** + * Invoked once, before the wrapped stream errors, when a deadline expires. + * Used to abort the underlying HTTP request. + */ + onStall?: ((error: StreamStalledError) => void) | undefined; + /** + * The request's abort signal. When it aborts, the watchdog disarms for + * good: the abort reaches consumers through the source stream erroring, + * and must never be reported as a stall. + */ + signal?: AbortSignal | undefined; +}; + +/** + * Wrap `source` with stall deadlines. Returns `source` unchanged when no + * deadline is configured. + * + * The wrapper pumps eagerly (it does not propagate backpressure): deadlines + * measure the server, not how fast the caller reads. Parsed SSE events are + * small and downstream consumers buffer them regardless. + */ +export function applyStreamWatchdog( + source: ReadableStream, + timeouts: StreamTimeoutOptions, + hooks: StreamWatchdogHooks, +): ReadableStream { + const firstContentMs = normalizeTimeout(timeouts.firstContentMs); + const contentIntervalMs = normalizeTimeout(timeouts.contentIntervalMs); + if (firstContentMs === undefined && contentIntervalMs === undefined) { + return source; + } + const reader = source.getReader(); + let watchdog: StreamWatchdog | undefined; + return new ReadableStream({ + start(controller) { + watchdog = new StreamWatchdog(reader, controller, hooks, firstContentMs, contentIntervalMs); + void watchdog.pump(); + }, + cancel(reason) { + watchdog?.settle(); + return reader.cancel(reason); + }, + }); +} + +/** Deadline state for one watched stream. See {@link applyStreamWatchdog}. */ +class StreamWatchdog { + private timer: ReturnType | undefined; + private armedPhase: StreamStallPhase = 'first_content'; + private armedTimeoutMs = 0; + private armedAtMs = 0; + private receivedContent = false; + /** Set by a terminal event or an abort: deadlines stop applying. */ + private disarmed = false; + /** Set once the wrapped stream has closed, errored, stalled or been cancelled. */ + private settled = false; + private readonly onAbort = (): void => { + this.disarmed = true; + this.clearTimer(); + }; + + constructor( + private readonly reader: ReadableStreamDefaultReader, + private readonly controller: ReadableStreamDefaultController, + private readonly hooks: StreamWatchdogHooks, + firstContentMs: number | undefined, + private readonly contentIntervalMs: number | undefined, + ) { + const signal = hooks.signal; + if (signal?.aborted) { + this.disarmed = true; + return; + } + signal?.addEventListener('abort', this.onAbort, { + once: true, + }); + // The pre-first-content deadline arms the moment the stream starts. + if (firstContentMs !== undefined) { + this.arm('first_content', firstContentMs); + } + } + + settle(): void { + this.settled = true; + this.clearTimer(); + this.hooks.signal?.removeEventListener('abort', this.onAbort); + } + + async pump(): Promise { + try { + while (true) { + const result = await this.reader.read(); + if (this.settled) { + return; + } + if (result.done) { + this.settle(); + this.controller.close(); + return; + } + this.observe(result.value); + this.controller.enqueue(result.value); + } + } catch (error) { + if (!this.settled) { + this.settle(); + this.controller.error(error); + } + } + } + + private observe(event: T): void { + if (this.hooks.isTerminalEvent?.(event)) { + this.disarmed = true; + this.clearTimer(); + return; + } + if (this.disarmed || !this.hooks.isContentEvent(event)) { + return; + } + this.receivedContent = true; + if (this.contentIntervalMs !== undefined) { + this.arm('between_content', this.contentIntervalMs); + } else { + this.clearTimer(); + } + } + + private arm(phase: StreamStallPhase, timeoutMs: number): void { + this.clearTimer(); + this.armedPhase = phase; + this.armedTimeoutMs = timeoutMs; + this.armedAtMs = Date.now(); + this.timer = setTimeout(() => this.expire(), timeoutMs); + } + + private clearTimer(): void { + if (this.timer !== undefined) { + clearTimeout(this.timer); + this.timer = undefined; + } + } + + private expire(): void { + this.timer = undefined; + if (this.settled || this.disarmed) { + return; + } + this.settle(); + const error = new StreamStalledError({ + phase: this.armedPhase, + timeoutMs: this.armedTimeoutMs, + elapsedMs: Date.now() - this.armedAtMs, + receivedAnyContent: this.receivedContent, + }); + this.hooks.onStall?.(error); + this.controller.error(error); + // Resolves the pump's pending read; `settled` makes the pump bail + // without touching the (already errored) controller. + this.reader.cancel(error).catch(() => { + // Irrelevant once the wrapped stream has errored. + }); + } +} + +/** + * Response stream event types that carry model output or a completed + * sub-result. These satisfy and re-arm watchdog deadlines. + * + * Deliberately excluded (neutral: neither satisfy nor reset): + * - `response.created` / `response.in_progress`: status. + * - `response.output_item.added`: an empty item shell (role prelude); the + * content it introduces arrives as subsequent delta events. + * - `response.content_part.added` / `response.reasoning_summary_part.added`: + * part shells preceding their deltas. + * - in-progress markers of server-side tools (web search, image generation, + * fusion): status, not output. + * - `error`: a stream-level error is not output, and the stream may still + * recover and complete, so it must not disarm the watchdog either. + * - unknown event types: a stream of only unknown types is indistinguishable + * from a stall; mixed streams are kept alive by the known events. + */ +const CONTENT_BEARING_EVENT_TYPES: ReadonlySet = new Set([ + 'response.output_text.delta', + 'response.output_text.done', + 'response.output_text.annotation.added', + 'response.refusal.delta', + 'response.refusal.done', + 'response.reasoning_text.delta', + 'response.reasoning_text.done', + 'response.reasoning_summary_text.delta', + 'response.reasoning_summary_text.done', + 'response.reasoning_summary_part.done', + 'response.function_call_arguments.delta', + 'response.function_call_arguments.done', + 'response.custom_tool_call_input.delta', + 'response.custom_tool_call_input.done', + 'response.apply_patch_call_operation_diff.delta', + 'response.apply_patch_call_operation_diff.done', + 'response.content_part.done', + 'response.output_item.done', + 'response.web_search_call.completed', + 'response.image_generation_call.partial_image', + 'response.image_generation_call.completed', + 'response.fusion_call.panel.delta', + 'response.fusion_call.panel.reasoning.delta', + 'response.fusion_call.panel.completed', + 'response.fusion_call.panel.failed', + 'response.fusion_call.analysis.completed', + 'response.fusion_call.completed', +]); + +/** + * Event types carrying the server's final verdict. They disarm the watchdog + * permanently. `error` is NOT terminal: the stream can continue (and even + * complete) after a stream-level error event, so the watchdog keeps + * policing it; otherwise an `error` followed by silence would hang forever. + */ +const TERMINAL_EVENT_TYPES: ReadonlySet = new Set([ + 'response.completed', + 'response.failed', + 'response.incomplete', +]); + +function eventType(event: models.StreamEvents): string | undefined { + return 'type' in event && typeof event.type === 'string' ? event.type : undefined; +} + +/** + * True when the event carries model output (text / refusal / reasoning + * deltas, tool-call arguments, completed output items or sub-results). + */ +export function isContentBearingStreamEvent(event: models.StreamEvents): boolean { + const type = eventType(event); + return type !== undefined && CONTENT_BEARING_EVENT_TYPES.has(type); +} + +/** True for `response.completed`, `response.failed` and `response.incomplete`. */ +export function isTerminalStreamEvent(event: models.StreamEvents): boolean { + const type = eventType(event); + return type !== undefined && TERMINAL_EVENT_TYPES.has(type); +} + +/** + * {@link applyStreamWatchdog} for Responses event streams, with the standard + * event classification. + */ +export function applyResponsesStreamWatchdog( + source: ReadableStream, + timeouts: StreamTimeoutOptions, + hooks?: Pick, 'onStall' | 'signal'>, +): ReadableStream { + return applyStreamWatchdog(source, timeouts, { + isContentEvent: isContentBearingStreamEvent, + isTerminalEvent: isTerminalStreamEvent, + onStall: hooks?.onStall, + signal: hooks?.signal, + }); +} + +/** True when at least one deadline is enabled (set, finite, > 0). */ +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. + */ +export function normalizeTimeout(value: number | undefined): number | undefined { + if (value === undefined || !Number.isFinite(value) || value <= 0) { + return undefined; + } + return value; +} diff --git a/packages/agent/tests/unit/hooks-post-model-call.test.ts b/packages/agent/tests/unit/hooks-post-model-call.test.ts index 8f98d80f..98ebd2c7 100644 --- a/packages/agent/tests/unit/hooks-post-model-call.test.ts +++ b/packages/agent/tests/unit/hooks-post-model-call.test.ts @@ -424,9 +424,20 @@ describe('PostModelCall hook', () => { input: 'hi', hooks, }); - for await (const _chunk of result.getTextStream()) { - // drain + // The truncated stream itself surfaces as a typed StreamFailedError + // (DEV-723) instead of the getter ending silently. + let streamError: unknown; + try { + for await (const _chunk of result.getTextStream()) { + // drain + } + } catch (error) { + streamError = error; } + expect(streamError).toMatchObject({ + name: 'StreamFailedError', + kind: 'truncated', + }); expect(calls).toHaveLength(0); expect(ends).toHaveLength(1); diff --git a/packages/agent/tests/unit/stream-failed-error.test.ts b/packages/agent/tests/unit/stream-failed-error.test.ts new file mode 100644 index 00000000..1f8dfc08 --- /dev/null +++ b/packages/agent/tests/unit/stream-failed-error.test.ts @@ -0,0 +1,338 @@ +/** + * Typed StreamFailedError (DEV-723): `response.failed`, stream-level `error` + * events, and truncated streams surface the same typed error on every + * consume path (getText / getResponse, the follow-up pipe, and every stream + * getter) instead of bare `Error`s or silently-ending iterators. + */ +import type { OpenRouterCore } from '@openrouter/sdk/core'; +import type * as models from '@openrouter/sdk/models'; +import { 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 { StreamFailedError } from '../../src/index.js'; +import { callModel } from '../../src/inner-loop/call-model.js'; +import type { ModelResult } from '../../src/lib/model-result.js'; +import { tool } from '../../src/lib/tool.js'; +import { + closedStream, + completedEvent, + createdEvent, + drain, + emptyMessageShell, + errorEvent, + failedEvent, + healthyToolCallEvents, + textDelta, + textResponse, +} 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, + }), +}); + +function respondWith(...streams: models.StreamEvents[][]): void { + for (const events of streams) { + mockBetaResponsesSend.mockResolvedValueOnce({ + ok: true, + value: closedStream(events), + }); + } +} + +function noToolsRun() { + return callModel(client, { + model: 'test-model', + input: 'hi', + }); +} + +function toolsRun() { + return callModel(client, { + model: 'test-model', + input: 'hi', + tools: [ + echoTool, + ] as const, + }); +} + +type AnyResult = ModelResult | ReturnType; + +/** Every consumption surface, each on a fresh run. */ +const getters: Array< + [ + string, + (r: AnyResult) => Promise, + ] +> = [ + [ + 'getText', + (r) => r.getText(), + ], + [ + 'getResponse', + (r) => r.getResponse(), + ], + [ + 'getTextStream', + (r) => throwIfDrainFailed(r.getTextStream()), + ], + [ + 'getFullResponsesStream', + (r) => throwIfDrainFailed(r.getFullResponsesStream()), + ], + [ + 'getItemsStream', + (r) => throwIfDrainFailed(r.getItemsStream()), + ], + [ + 'getReasoningStream', + (r) => throwIfDrainFailed(r.getReasoningStream()), + ], + [ + 'getToolStream', + (r) => throwIfDrainFailed(r.getToolStream()), + ], + [ + 'getNewMessagesStream', + (r) => throwIfDrainFailed(r.getNewMessagesStream()), + ], + [ + 'getToolCallsStream', + (r) => throwIfDrainFailed(r.getToolCallsStream()), + ], +]; + +async function throwIfDrainFailed(iterable: AsyncIterable): Promise { + const result = await drain(iterable); + if (result.threw) { + throw result.error; + } +} + +beforeEach(() => { + mockBetaResponsesSend.mockReset(); +}); + +describe('StreamFailedError', () => { + it.each([ + 'server_error', + 'server', + 'timeout', + 'rate_limit_exceeded', + 'provider_overloaded', + 'provider_unavailable', + ])('classifies %s as retryable (as code or errorType)', (code) => { + expect( + new StreamFailedError({ + kind: 'error_event', + message: 'x', + code, + }).retryable, + ).toBe(true); + expect( + new StreamFailedError({ + kind: 'response_failed', + message: 'x', + errorType: code, + }).retryable, + ).toBe(true); + }); + + it('classifies validation failures and truncated streams as non-retryable', () => { + expect( + new StreamFailedError({ + kind: 'response_failed', + message: 'x', + code: 'invalid_prompt', + }).retryable, + ).toBe(false); + expect(StreamFailedError.fromStreamEnd(null).retryable).toBe(false); + expect(StreamFailedError.fromStreamEnd(null).kind).toBe('truncated'); + }); + + it('fromFailedResponse carries code, errorType and the failed response', () => { + const event = failedEvent('server_error', 'upstream died', { + errorType: 'provider_unavailable', + }) as models.StreamEventsResponseFailed; + const error = StreamFailedError.fromFailedResponse(event.response); + expect(error).toBeInstanceOf(Error); + expect(error.name).toBe('StreamFailedError'); + expect(error.kind).toBe('response_failed'); + expect(error.code).toBe('server_error'); + expect(error.errorType).toBe('provider_unavailable'); + expect(error.response).toBe(event.response); + expect(error.message).toBe('Response failed (server_error): upstream died'); + expect(error.retryable).toBe(true); + }); +}); + +describe('response.failed surfaces as StreamFailedError on every consume path', () => { + const failed = () => [ + createdEvent(), + textDelta('partial'), + failedEvent('invalid_prompt', 'bad prompt'), + ]; + + it.each(getters)('no tools: %s rejects', async (_name, consume) => { + respondWith(failed()); + const error = await consume(noToolsRun()).then( + () => undefined, + (e: unknown) => e, + ); + expect(error).toBeInstanceOf(StreamFailedError); + expect(error).toMatchObject({ + kind: 'response_failed', + code: 'invalid_prompt', + retryable: false, + }); + }); + + it.each(getters)('tools, initial turn: %s rejects', async (_name, consume) => { + respondWith(failed()); + await expect(consume(toolsRun())).rejects.toBeInstanceOf(StreamFailedError); + }); + + it.each( + getters.filter(([name]) => name !== 'getToolCallsStream'), + )('tools, follow-up turn: %s rejects', async (_name, consume) => { + // getToolCallsStream only reads the initial turn, so it is excluded. + respondWith( + healthyToolCallEvents('echo', { + value: 'a', + }), + failed(), + ); + await expect(consume(toolsRun())).rejects.toMatchObject({ + name: 'StreamFailedError', + kind: 'response_failed', + code: 'invalid_prompt', + }); + }); +}); + +describe('stream-level error events', () => { + const errorThenClose = () => [ + createdEvent(), + errorEvent('server_error', 'provider hiccup'), + ]; + + it.each(getters)('no tools: error then close makes %s reject', async (_name, consume) => { + respondWith(errorThenClose()); + await expect(consume(noToolsRun())).rejects.toMatchObject({ + name: 'StreamFailedError', + kind: 'error_event', + code: 'server_error', + retryable: true, + }); + }); + + it('tools, follow-up turn: error then close rejects with the error event', async () => { + respondWith( + healthyToolCallEvents('echo', { + value: 'a', + }), + errorThenClose(), + ); + await expect(toolsRun().getText()).rejects.toMatchObject({ + name: 'StreamFailedError', + kind: 'error_event', + message: 'Stream error (server_error): provider hiccup', + }); + }); + + it('an error event followed by a completion still resolves', async () => { + respondWith([ + createdEvent(), + errorEvent('server_error', 'transient'), + emptyMessageShell(), + textDelta('recovered'), + completedEvent(textResponse('recovered')), + ]); + await expect(noToolsRun().getText()).resolves.toBe('recovered'); + }); + + it('keeps the error event with streamReplay: active-consumers', async () => { + // An error then close must still name the error event even though + // active-consumers replay drops consumed events before the run + // materializes the initial response. + respondWith(errorThenClose()); + const result = callModel(client, { + model: 'test-model', + input: 'hi', + tools: [ + echoTool, + ] as const, + streamReplay: 'active-consumers', + }); + const drained = await drain(result.getFullResponsesStream()); + expect(drained.error).toMatchObject({ + name: 'StreamFailedError', + kind: 'error_event', + }); + }); +}); + +describe('truncated streams (no terminal event)', () => { + const truncated = () => [ + createdEvent(), + emptyMessageShell(), + textDelta('cut off mid-'), + ]; + + it.each(getters)('no tools: %s rejects with kind truncated', async (_name, consume) => { + respondWith(truncated()); + await expect(consume(noToolsRun())).rejects.toMatchObject({ + name: 'StreamFailedError', + kind: 'truncated', + code: null, + retryable: false, + }); + }); + + it('getTextStream still yields the partial text before rejecting', async () => { + respondWith(truncated()); + const drained = await drain(noToolsRun().getTextStream()); + expect(drained.items).toEqual([ + 'cut off mid-', + ]); + expect(drained.error).toBeInstanceOf(StreamFailedError); + }); + + it('breaking out of a getter early is not a failure', async () => { + respondWith(truncated()); + const result = noToolsRun(); + for await (const _delta of result.getTextStream()) { + break; + } + }); + + it('a cancelled run does not report its stream as truncated', async () => { + respondWith(truncated()); + const result = noToolsRun(); + const iterator = result.getFullResponsesStream(); + await iterator.next(); + await result.cancel(); + const rest = await drain(iterator); + expect(rest.threw).toBe(false); + }); +}); diff --git a/packages/agent/tests/unit/stream-watchdog.test.ts b/packages/agent/tests/unit/stream-watchdog.test.ts new file mode 100644 index 00000000..80339a6e --- /dev/null +++ b/packages/agent/tests/unit/stream-watchdog.test.ts @@ -0,0 +1,519 @@ +import type * as models from '@openrouter/sdk/models'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { StreamStalledError } from '../../src/lib/stream-errors.js'; +import { + applyResponsesStreamWatchdog, + applyStreamWatchdog, + hasActiveStreamTimeouts, + isContentBearingStreamEvent, + isTerminalStreamEvent, +} from '../../src/lib/stream-watchdog.js'; +import { + completedEvent, + controlledStream, + createdEvent, + emptyMessageShell, + errorEvent, + failedEvent, + functionCallArgsDelta, + reasoningDelta, + textDelta, + textResponse, +} from '../utils/stream-fixtures.js'; + +/** + * Read a stream in the background, recording every event and how it ended, + * so fake-timer tests can advance the clock while a read is pending. + */ +function observe(stream: ReadableStream) { + const events: T[] = []; + const state: { + status: 'open' | 'closed' | 'errored'; + error: unknown; + } = { + status: 'open', + error: undefined, + }; + const reader = stream.getReader(); + const readAll = async (): Promise => { + try { + while (true) { + const result = await reader.read(); + if (result.done) { + state.status = 'closed'; + return; + } + events.push(result.value); + } + } catch (error) { + state.status = 'errored'; + state.error = error; + } + }; + const done = readAll(); + return { + events, + state, + done, + }; +} + +const hooks = { + isContentEvent: (e: string) => e.startsWith('content'), + isTerminalEvent: (e: string) => e === 'terminal', +}; + +beforeEach(() => { + vi.useFakeTimers(); +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +describe('event classification', () => { + it('classifies output deltas and completed sub-results as content', () => { + expect(isContentBearingStreamEvent(textDelta('hi'))).toBe(true); + expect(isContentBearingStreamEvent(reasoningDelta('hmm'))).toBe(true); + expect(isContentBearingStreamEvent(functionCallArgsDelta('{'))).toBe(true); + expect( + isContentBearingStreamEvent({ + type: 'response.fusion_call.analysis.completed', + } as unknown as models.StreamEvents), + ).toBe(true); + }); + + it('treats status events, item shells, errors and unknown types as neutral', () => { + expect(isContentBearingStreamEvent(createdEvent())).toBe(false); + expect(isContentBearingStreamEvent(emptyMessageShell())).toBe(false); + expect(isContentBearingStreamEvent(errorEvent('server_error', 'boom'))).toBe(false); + expect( + isContentBearingStreamEvent({ + type: 'response.mystery', + } as unknown as models.StreamEvents), + ).toBe(false); + expect(isTerminalStreamEvent(createdEvent())).toBe(false); + expect(isTerminalStreamEvent(textDelta('hi'))).toBe(false); + }); + + it('treats completed / failed / incomplete as terminal, but not error events', () => { + expect(isTerminalStreamEvent(completedEvent(textResponse('x')))).toBe(true); + expect(isTerminalStreamEvent(failedEvent('server_error', 'boom'))).toBe(true); + expect( + isTerminalStreamEvent({ + type: 'response.incomplete', + } as unknown as models.StreamEvents), + ).toBe(true); + // An error event can be followed by more events (even a completion), so + // it must keep the watchdog armed. + expect(isTerminalStreamEvent(errorEvent('server_error', 'boom'))).toBe(false); + }); + + it('hasActiveStreamTimeouts ignores unset, zero, negative and non-finite values', () => { + expect(hasActiveStreamTimeouts(undefined)).toBe(false); + expect(hasActiveStreamTimeouts({})).toBe(false); + expect( + hasActiveStreamTimeouts({ + firstContentMs: 0, + contentIntervalMs: -1, + }), + ).toBe(false); + expect( + hasActiveStreamTimeouts({ + firstContentMs: Number.POSITIVE_INFINITY, + }), + ).toBe(false); + expect( + hasActiveStreamTimeouts({ + contentIntervalMs: 10, + }), + ).toBe(true); + }); +}); + +describe('applyStreamWatchdog', () => { + it('returns the source unchanged when no deadline is configured', () => { + for (const timeouts of [ + {}, + { + firstContentMs: 0, + contentIntervalMs: -5, + }, + { + firstContentMs: Number.NaN, + }, + ]) { + const source = controlledStream().stream; + expect(applyStreamWatchdog(source, timeouts, hooks)).toBe(source); + } + }); + + it('passes a healthy stream through and leaves no timers behind', async () => { + const source = controlledStream(); + const out = observe( + applyStreamWatchdog( + source.stream, + { + firstContentMs: 1000, + contentIntervalMs: 1000, + }, + hooks, + ), + ); + source.push('content-1'); + await vi.advanceTimersByTimeAsync(500); + source.push('content-2'); + await vi.advanceTimersByTimeAsync(500); + source.close(); + await out.done; + expect(out.state.status).toBe('closed'); + expect(out.events).toEqual([ + 'content-1', + 'content-2', + ]); + expect(vi.getTimerCount()).toBe(0); + }); + + it('fails with first_content when no content arrives at all', async () => { + const source = controlledStream(); + const out = observe( + applyStreamWatchdog( + source.stream, + { + firstContentMs: 40, + }, + hooks, + ), + ); + await vi.advanceTimersByTimeAsync(39); + expect(out.state.status).toBe('open'); + await vi.advanceTimersByTimeAsync(1); + await out.done; + expect(out.state.status).toBe('errored'); + const stall = out.state.error as StreamStalledError; + expect(stall).toBeInstanceOf(StreamStalledError); + expect(stall.name).toBe('StreamStalledError'); + expect(stall.phase).toBe('first_content'); + expect(stall.timeoutMs).toBe(40); + expect(stall.elapsedMs).toBe(40); + expect(stall.receivedAnyContent).toBe(false); + expect(stall.retryable).toBe(true); + }); + + it('metadata heartbeats neither satisfy nor extend the first_content deadline', async () => { + const source = controlledStream(); + const out = observe( + applyStreamWatchdog( + source.stream, + { + firstContentMs: 90, + }, + hooks, + ), + ); + for (let i = 0; i < 4; i++) { + source.push('meta'); + await vi.advanceTimersByTimeAsync(20); + } + expect(out.state.status).toBe('open'); + await vi.advanceTimersByTimeAsync(10); + await out.done; + // Non-content events still flow through to the consumer. + expect(out.events).toEqual([ + 'meta', + 'meta', + 'meta', + 'meta', + ]); + expect((out.state.error as StreamStalledError).phase).toBe('first_content'); + }); + + it('the first content event disarms firstContentMs permanently', async () => { + const source = controlledStream(); + const out = observe( + applyStreamWatchdog( + source.stream, + { + firstContentMs: 50, + }, + hooks, + ), + ); + source.push('content-1'); + await vi.advanceTimersByTimeAsync(10_000); + source.push('content-2'); + source.close(); + await out.done; + expect(out.state.status).toBe('closed'); + expect(out.events).toEqual([ + 'content-1', + 'content-2', + ]); + }); + + it('fails with between_content when the stream stalls mid-generation', async () => { + const source = controlledStream(); + const out = observe( + applyStreamWatchdog( + source.stream, + { + firstContentMs: 200, + contentIntervalMs: 50, + }, + hooks, + ), + ); + source.push('content-1'); + await vi.advanceTimersByTimeAsync(10); + source.push('content-2'); + await vi.advanceTimersByTimeAsync(50); + await out.done; + expect(out.events).toEqual([ + 'content-1', + 'content-2', + ]); + const stall = out.state.error as StreamStalledError; + expect(stall.phase).toBe('between_content'); + expect(stall.timeoutMs).toBe(50); + expect(stall.receivedAnyContent).toBe(true); + // Content already flowed: not safe to blind-retry. + expect(stall.retryable).toBe(false); + }); + + it('each content event re-arms the between_content deadline', async () => { + const source = controlledStream(); + const out = observe( + applyStreamWatchdog( + source.stream, + { + contentIntervalMs: 50, + }, + hooks, + ), + ); + for (let i = 0; i < 5; i++) { + source.push(`content-${i}`); + await vi.advanceTimersByTimeAsync(40); + } + source.close(); + await out.done; + expect(out.state.status).toBe('closed'); + expect(out.events).toHaveLength(5); + }); + + it('contentIntervalMs alone does not police the pre-first-content window', async () => { + const source = controlledStream(); + const out = observe( + applyStreamWatchdog( + source.stream, + { + contentIntervalMs: 30, + }, + hooks, + ), + ); + await vi.advanceTimersByTimeAsync(5_000); + source.push('content-1'); + source.close(); + await out.done; + expect(out.state.status).toBe('closed'); + }); + + it('a terminal event disarms deadlines so trailing events are never stalls', async () => { + const source = controlledStream(); + const out = observe( + applyStreamWatchdog( + source.stream, + { + firstContentMs: 50, + contentIntervalMs: 30, + }, + hooks, + ), + ); + source.push('content-1', 'terminal'); + await vi.advanceTimersByTimeAsync(1_000); + source.push('meta-trailer'); + source.close(); + await out.done; + expect(out.state.status).toBe('closed'); + expect(out.events).toEqual([ + 'content-1', + 'terminal', + 'meta-trailer', + ]); + }); + + it('cancels the source and calls onStall exactly once when a deadline expires', async () => { + const source = controlledStream(); + const onStall = vi.fn(); + const out = observe( + applyStreamWatchdog( + source.stream, + { + firstContentMs: 30, + }, + { + ...hooks, + onStall, + }, + ), + ); + await vi.advanceTimersByTimeAsync(30); + await out.done; + await vi.advanceTimersByTimeAsync(1_000); + expect(onStall).toHaveBeenCalledTimes(1); + expect(onStall.mock.calls[0]?.[0]).toBe(out.state.error); + expect(source.cancelled).toBe(true); + expect(source.cancelReason).toBe(out.state.error); + expect(vi.getTimerCount()).toBe(0); + }); + + it('propagates upstream errors as-is, not as stalls', async () => { + const source = controlledStream(); + const out = observe( + applyStreamWatchdog( + source.stream, + { + firstContentMs: 1000, + }, + hooks, + ), + ); + const upstream = new Error('upstream exploded'); + source.error(upstream); + await out.done; + expect(out.state.error).toBe(upstream); + expect(vi.getTimerCount()).toBe(0); + }); + + it('cancelling the wrapped stream stops timers and cancels the source', async () => { + const source = controlledStream(); + const wrapped = applyStreamWatchdog( + source.stream, + { + firstContentMs: 30, + }, + hooks, + ); + await wrapped.cancel('consumer walked away'); + expect(source.cancelled).toBe(true); + expect(source.cancelReason).toBe('consumer walked away'); + expect(vi.getTimerCount()).toBe(0); + }); + + it('an abort on the request signal disarms the watchdog (aborts are never stalls)', async () => { + const source = controlledStream(); + const controller = new AbortController(); + const onStall = vi.fn(); + const out = observe( + applyStreamWatchdog( + source.stream, + { + firstContentMs: 30, + }, + { + ...hooks, + onStall, + signal: controller.signal, + }, + ), + ); + await vi.advanceTimersByTimeAsync(10); + const reason = new Error('user cancelled'); + controller.abort(reason); + expect(vi.getTimerCount()).toBe(0); + // The transport surfaces the abort by erroring the body stream. + source.error(reason); + await out.done; + await vi.advanceTimersByTimeAsync(1_000); + expect(out.state.error).toBe(reason); + expect(onStall).not.toHaveBeenCalled(); + }); + + it('an already-aborted signal never arms a deadline', async () => { + const source = controlledStream(); + const controller = new AbortController(); + controller.abort(); + const out = observe( + applyStreamWatchdog( + source.stream, + { + firstContentMs: 30, + }, + { + ...hooks, + signal: controller.signal, + }, + ), + ); + expect(vi.getTimerCount()).toBe(0); + source.close(); + await out.done; + expect(out.state.status).toBe('closed'); + }); +}); + +describe('applyResponsesStreamWatchdog', () => { + it('stalls on the keep-alive pathology: created + role prelude, then silence', async () => { + const source = controlledStream(); + const out = observe( + applyResponsesStreamWatchdog(source.stream, { + firstContentMs: 60, + }), + ); + source.push(createdEvent(), emptyMessageShell()); + await vi.advanceTimersByTimeAsync(60); + await out.done; + expect(out.events).toHaveLength(2); + const stall = out.state.error as StreamStalledError; + expect(stall).toBeInstanceOf(StreamStalledError); + expect(stall.phase).toBe('first_content'); + expect(stall.retryable).toBe(true); + }); + + it('reasoning deltas satisfy the first-content deadline', async () => { + const source = controlledStream(); + const out = observe( + applyResponsesStreamWatchdog(source.stream, { + firstContentMs: 50, + }), + ); + source.push(createdEvent()); + await vi.advanceTimersByTimeAsync(10); + source.push(reasoningDelta('thinking...')); + await vi.advanceTimersByTimeAsync(500); + source.push(textDelta('answer'), completedEvent(textResponse('answer'))); + source.close(); + await out.done; + expect(out.state.status).toBe('closed'); + expect(out.events).toHaveLength(4); + }); + + it('keeps policing after a stream-level error event (error then silence stalls)', async () => { + const source = controlledStream(); + const out = observe( + applyResponsesStreamWatchdog(source.stream, { + firstContentMs: 50, + }), + ); + source.push(createdEvent(), errorEvent('server_error', 'transient')); + await vi.advanceTimersByTimeAsync(50); + await out.done; + expect(out.state.error).toBeInstanceOf(StreamStalledError); + }); + + it('response.failed disarms: the failure surfaces through its own error path', async () => { + const source = controlledStream(); + const out = observe( + applyResponsesStreamWatchdog(source.stream, { + firstContentMs: 50, + }), + ); + source.push(createdEvent(), failedEvent('server_error', 'boom')); + await vi.advanceTimersByTimeAsync(1_000); + source.close(); + await out.done; + expect(out.state.status).toBe('closed'); + }); +}); diff --git a/packages/agent/tests/utils/stream-fixtures.ts b/packages/agent/tests/utils/stream-fixtures.ts new file mode 100644 index 00000000..f56ef8f6 --- /dev/null +++ b/packages/agent/tests/utils/stream-fixtures.ts @@ -0,0 +1,275 @@ +/** + * Shared Responses stream fixtures for the stream watchdog / typed stream + * error tests (DEV-723). Events are in the parsed (camelCase) shape that + * `betaResponsesSend` yields, so tests can mock it with a ReadableStream. + */ +import type * as models from '@openrouter/sdk/models'; + +let counter = 0; + +export function baseResponse( + output: unknown[], + overrides: Partial = {}, +): models.OpenResponsesResult { + counter++; + return { + id: `resp_${counter}`, + object: 'response', + createdAt: 0, + model: 'test-model', + status: 'completed', + completedAt: 0, + output, + error: null, + incompleteDetails: null, + temperature: null, + topP: null, + presencePenalty: null, + frequencyPenalty: null, + metadata: null, + instructions: null, + tools: [], + toolChoice: 'auto', + parallelToolCalls: false, + ...overrides, + } as models.OpenResponsesResult; +} + +export function textResponse(text: string): models.OpenResponsesResult { + return baseResponse([ + { + id: `msg_${counter}`, + type: 'message', + role: 'assistant', + status: 'completed', + content: [ + { + type: 'output_text', + text, + annotations: [], + }, + ], + }, + ]); +} + +export function toolCallResponse(name: string, args: unknown): models.OpenResponsesResult { + return baseResponse([ + { + type: 'function_call', + id: `fc_${counter}`, + callId: `call_${counter}`, + name, + arguments: JSON.stringify(args), + status: 'completed', + }, + ]); +} + +export function createdEvent(): models.StreamEvents { + return { + type: 'response.created', + response: baseResponse([], { + status: 'in_progress', + }), + sequenceNumber: 0, + } as models.StreamEvents; +} + +/** An empty assistant-message shell: the role prelude that must NOT count as content. */ +export function emptyMessageShell(): models.StreamEvents { + return { + type: 'response.output_item.added', + outputIndex: 0, + item: { + type: 'message', + id: 'msg_shell', + role: 'assistant', + status: 'in_progress', + content: [], + }, + sequenceNumber: 0, + } as models.StreamEvents; +} + +export function textDelta(delta: string): models.StreamEvents { + return { + type: 'response.output_text.delta', + delta, + itemId: 'msg_shell', + contentIndex: 0, + outputIndex: 0, + logprobs: [], + sequenceNumber: 1, + } as models.StreamEvents; +} + +export function reasoningDelta(delta: string): models.StreamEvents { + return { + type: 'response.reasoning_text.delta', + delta, + itemId: 'rs_1', + contentIndex: 0, + outputIndex: 0, + sequenceNumber: 1, + } as models.StreamEvents; +} + +export function functionCallArgsDelta(delta: string): models.StreamEvents { + return { + type: 'response.function_call_arguments.delta', + delta, + itemId: 'fc_1', + outputIndex: 0, + sequenceNumber: 1, + } as models.StreamEvents; +} + +export function completedEvent(response: models.OpenResponsesResult): models.StreamEvents { + return { + type: 'response.completed', + response, + sequenceNumber: 9, + } as models.StreamEvents; +} + +export function failedEvent( + code: string, + message: string, + extra: Record = {}, +): models.StreamEvents { + return { + type: 'response.failed', + response: baseResponse([], { + status: 'failed', + error: { + code, + message, + } as models.OpenResponsesResult['error'], + ...extra, + }), + sequenceNumber: 9, + } as models.StreamEvents; +} + +export function errorEvent(code: string | null, message: string): models.StreamEvents { + return { + type: 'error', + code, + message, + param: null, + sequenceNumber: 5, + } as models.StreamEvents; +} + +/** A healthy single-turn stream: created, shell, delta, completed, close. */ +export function healthyTextEvents(text: string): models.StreamEvents[] { + return [ + createdEvent(), + emptyMessageShell(), + textDelta(text), + completedEvent(textResponse(text)), + ]; +} + +export function healthyToolCallEvents(name: string, args: unknown): models.StreamEvents[] { + return [ + createdEvent(), + functionCallArgsDelta(JSON.stringify(args)), + completedEvent(toolCallResponse(name, args)), + ]; +} + +/** + * A ReadableStream driven by the test. Never closes on its own: a test that + * stops pushing simulates a hung connection. + */ +export interface ControlledStream { + stream: ReadableStream; + push: (...events: T[]) => void; + close: () => void; + error: (reason: unknown) => void; + readonly cancelled: boolean; + readonly cancelReason: unknown; +} + +export function controlledStream(initial: T[] = []): ControlledStream { + let controller!: ReadableStreamDefaultController; + let cancelled = false; + let cancelReason: unknown; + let done = false; + const stream = new ReadableStream({ + start(c) { + controller = c; + for (const event of initial) { + c.enqueue(event); + } + }, + cancel(reason) { + cancelled = true; + cancelReason = reason; + done = true; + }, + }); + return { + stream, + push: (...events) => { + if (done) { + return; + } + for (const event of events) { + controller.enqueue(event); + } + }, + close: () => { + if (!done) { + done = true; + controller.close(); + } + }, + error: (reason) => { + if (!done) { + done = true; + controller.error(reason); + } + }, + get cancelled() { + return cancelled; + }, + get cancelReason() { + return cancelReason; + }, + }; +} + +/** A stream that replays `events` then closes. */ +export function closedStream(events: T[]): ReadableStream { + const controlled = controlledStream(events); + controlled.close(); + return controlled.stream; +} + +/** Drain an async iterable, returning what it yielded and how it ended. */ +export async function drain(iterable: AsyncIterable): Promise<{ + items: T[]; + error: unknown; + threw: boolean; +}> { + const items: T[] = []; + try { + for await (const item of iterable) { + items.push(item); + } + return { + items, + error: undefined, + threw: false, + }; + } catch (error) { + return { + items, + error, + threw: true, + }; + } +}