diff --git a/README.md b/README.md index 7411a9be6..cd7f84eac 100644 --- a/README.md +++ b/README.md @@ -100,6 +100,83 @@ for await (const chunk of result) { +## Stalled-stream detection + +A streaming response can return headers quickly, then never emit a content +chunk — the connection stays open (keep-alive comments may even keep +arriving) while no output flows. Transport-level timeouts cannot catch +this. The SDK ships an opt-in watchdog with two semantic deadlines based +on parsed events, not socket activity: + +- `firstContentMs` — max time between the response stream starting and its + first content-bearing event (text/reasoning/refusal delta, tool-call + arguments). Keep-alives, `response.created`, and empty role preludes do + not satisfy or reset it. +- `contentIntervalMs` — max gap between content-bearing events once + content has started. + +With `callModel`, pass `timeout` (deadlines re-arm for every turn in a +tool loop, and the stalled turn's HTTP request is aborted): + +```typescript +import { OpenRouter, StreamStalledError } from "@openrouter/sdk"; + +const openRouter = new OpenRouter(); + +const result = openRouter.callModel({ + model: "openai/gpt-5", + input: "Hello!", + timeout: { + firstContentMs: 15_000, + contentIntervalMs: 30_000, + // Optional: transparently re-issue a turn that stalls before any + // content arrived (never retries after content started, so output + // cannot be duplicated). + maxStallRetries: 1, + }, +}); + +try { + console.log(await result.getText()); +} catch (error) { + if (error instanceof StreamStalledError) { + // error.phase: "first_content" | "between_content" + // error.retryable: true only if no content was received + console.error(`Stream stalled after ${error.elapsedMs}ms`, error.phase); + } + throw error; +} +``` + +For raw streams (`chat.send` / `responses.send` with `stream: true`), wrap +the event stream yourself: + +```typescript +import { OpenRouter, applyChatStreamWatchdog } from "@openrouter/sdk"; + +const openRouter = new OpenRouter(); + +const stream = await openRouter.chat.send({ + model: "openai/gpt-5", + messages: [{ role: "user", content: "Hello!" }], + stream: true, +}); + +if (stream instanceof ReadableStream) { + const watched = applyChatStreamWatchdog(stream, { firstContentMs: 15_000 }); + for await (const chunk of watched) { + console.log(chunk.choices[0]?.delta.content); + } +} +``` + +(`applyResponsesStreamWatchdog` is the equivalent for the Responses API.) + +Separately from stalls, server-reported stream failures (`response.failed` +or stream `error` events) throw `StreamFailedError` carrying `code`, +`errorType`, the failed `response`, and a `retryable` hint — instead of a +bare `Error`. + diff --git a/src/sdk/sdk.ts b/src/sdk/sdk.ts index d107db20f..452c8105d 100644 --- a/src/sdk/sdk.ts +++ b/src/sdk/sdk.ts @@ -48,7 +48,11 @@ export { StreamStalledError, type StreamStallPhase, } from "../lib/stream-errors.js"; -export type { StreamTimeoutOptions } from "../lib/stream-watchdog.js"; +export { + applyChatStreamWatchdog, + applyResponsesStreamWatchdog, + type StreamTimeoutOptions, +} from "../lib/stream-watchdog.js"; // #endregion imports export class OpenRouter extends ClientSDK { diff --git a/tests/unit/chat-stream-watchdog.test.ts b/tests/unit/chat-stream-watchdog.test.ts new file mode 100644 index 000000000..2219b19d8 --- /dev/null +++ b/tests/unit/chat-stream-watchdog.test.ts @@ -0,0 +1,197 @@ +import type { ChatStreamChunk } from '../../src/models/chatstreamchunk.js'; + +import { describe, expect, it } from 'vitest'; +import { StreamStalledError } from '../../src/lib/stream-errors.js'; +import { + applyChatStreamWatchdog, + isContentBearingChatChunk, + isTerminalChatChunk, +} from '../../src/lib/stream-watchdog.js'; + +// ============================================================================ +// Chunk fixtures +// ============================================================================ + +function chunk(overrides: { + delta?: Partial; + finishReason?: 'stop' | 'length' | null; + error?: { code: number; message: string }; + noChoices?: boolean; +}): ChatStreamChunk { + return { + id: 'gen-1', + object: 'chat.completion.chunk', + created: 0, + model: 'test-model', + ...(overrides.error !== undefined ? { error: overrides.error } : {}), + choices: overrides.noChoices + ? [] + : [ + { + index: 0, + delta: { ...overrides.delta }, + finishReason: overrides.finishReason ?? null, + }, + ], + } as ChatStreamChunk; +} + +/** The role-only prelude chunk every chat stream starts with. */ +const ROLE_PRELUDE = chunk({ delta: { role: 'assistant', content: '' } }); + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function scriptedChunkStream( + steps: Array<{ chunk?: ChatStreamChunk; delayMs: number; close?: boolean }>, +): ReadableStream { + let cancelled = false; + return new ReadableStream({ + start(controller) { + void (async () => { + for (const step of steps) { + await sleep(step.delayMs); + if (cancelled) { + return; + } + if (step.chunk) { + controller.enqueue(step.chunk); + } + if (step.close) { + controller.close(); + return; + } + } + // No close: hang. + })(); + }, + cancel() { + cancelled = true; + }, + }); +} + +// ============================================================================ +// Classification +// ============================================================================ + +describe('isContentBearingChatChunk', () => { + it('classifies content, reasoning, refusal, tool-call, and audio deltas as content', () => { + expect(isContentBearingChatChunk(chunk({ delta: { content: 'hi' } }))).toBe(true); + expect(isContentBearingChatChunk(chunk({ delta: { reasoning: 'hmm' } }))).toBe(true); + expect(isContentBearingChatChunk(chunk({ delta: { refusal: 'no' } }))).toBe(true); + expect( + isContentBearingChatChunk( + chunk({ + delta: { + toolCalls: [ + { index: 0, id: 'c1', type: 'function', function: { name: 'f', arguments: '' } }, + ], + }, + }), + ), + ).toBe(true); + }); + + it('does not classify role preludes, empty content, or empty chunks as content', () => { + expect(isContentBearingChatChunk(ROLE_PRELUDE)).toBe(false); + expect(isContentBearingChatChunk(chunk({ delta: { content: '' } }))).toBe(false); + expect(isContentBearingChatChunk(chunk({ delta: {} }))).toBe(false); + expect(isContentBearingChatChunk(chunk({ noChoices: true }))).toBe(false); + }); +}); + +describe('isTerminalChatChunk', () => { + it('classifies finish reasons and error payloads as terminal', () => { + expect(isTerminalChatChunk(chunk({ delta: {}, finishReason: 'stop' }))).toBe(true); + expect( + isTerminalChatChunk(chunk({ noChoices: true, error: { code: 500, message: 'boom' } })), + ).toBe(true); + }); + + it('does not classify ordinary delta chunks as terminal', () => { + expect(isTerminalChatChunk(chunk({ delta: { content: 'hi' } }))).toBe(false); + expect(isTerminalChatChunk(ROLE_PRELUDE)).toBe(false); + }); +}); + +// ============================================================================ +// applyChatStreamWatchdog +// ============================================================================ + +describe('applyChatStreamWatchdog', () => { + it('stalls on a role prelude followed by silence', async () => { + const wrapped = applyChatStreamWatchdog( + scriptedChunkStream([{ chunk: ROLE_PRELUDE, delayMs: 5 }]), // then hangs + { firstContentMs: 60 }, + ); + + const reader = wrapped.getReader(); + const seen: ChatStreamChunk[] = []; + const error = await (async () => { + try { + while (true) { + const result = await reader.read(); + if (result.done) return null; + seen.push(result.value); + } + } catch (e) { + return e; + } + })(); + + expect(seen).toHaveLength(1); // prelude flowed through + expect(error).toBeInstanceOf(StreamStalledError); + expect((error as StreamStalledError).phase).toBe('first_content'); + expect((error as StreamStalledError).retryable).toBe(true); + }); + + it('passes a healthy chat stream through, with the finish chunk disarming deadlines', async () => { + const wrapped = applyChatStreamWatchdog( + scriptedChunkStream([ + { chunk: ROLE_PRELUDE, delayMs: 5 }, + { chunk: chunk({ delta: { content: 'Hello' } }), delayMs: 5 }, + { chunk: chunk({ delta: {}, finishReason: 'stop' }), delayMs: 5 }, + // usage chunk arriving late, after the terminal chunk disarmed timers + { chunk: chunk({ noChoices: true }), delayMs: 100, close: true }, + ]), + { firstContentMs: 60, contentIntervalMs: 40 }, + ); + + const collected: ChatStreamChunk[] = []; + const reader = wrapped.getReader(); + while (true) { + const result = await reader.read(); + if (result.done) break; + collected.push(result.value); + } + expect(collected).toHaveLength(4); + }); + + it('stalls when content deltas stop mid-generation', async () => { + const wrapped = applyChatStreamWatchdog( + scriptedChunkStream([ + { chunk: chunk({ delta: { content: 'partial' } }), delayMs: 5 }, + // then hangs — no finish chunk + ]), + { contentIntervalMs: 50 }, + ); + + const reader = wrapped.getReader(); + const error = await (async () => { + try { + while (true) { + const result = await reader.read(); + if (result.done) return null; + } + } catch (e) { + return e; + } + })(); + + expect(error).toBeInstanceOf(StreamStalledError); + expect((error as StreamStalledError).phase).toBe('between_content'); + expect((error as StreamStalledError).retryable).toBe(false); + }); +});