feat: raw-stream watchdog helpers + docs (DEV-723 5/5)

README.md

@@ -100,6 +100,83 @@ for await (const chunk of result) {
 
 <!-- No Server-sent event streaming [eventstream] -->
 
+## 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`.
+
 <!-- No Retries [retries] -->
 
 <!-- No Error Handling [errors] -->

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 {

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<ChatStreamChunk['choices'][number]['delta']>;
+  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<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+function scriptedChunkStream(
+  steps: Array<{ chunk?: ChatStreamChunk; delayMs: number; close?: boolean }>,
+): ReadableStream<ChatStreamChunk> {
+  let cancelled = false;
+  return new ReadableStream<ChatStreamChunk>({
+    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);
+  });
+});