diff --git a/src/lib/stream-errors.ts b/src/lib/stream-errors.ts
new file mode 100644
index 000000000..117ecc0ac
--- /dev/null
+++ b/src/lib/stream-errors.ts
@@ -0,0 +1,69 @@
+/**
+ * Errors for client-side stalled-stream detection (DEV-723).
+ *
+ * A streaming response can return headers quickly and then never emit a
+ * content-bearing event (a "stalled stream"). The stream watchdog
+ * (`stream-watchdog.ts`) enforces the configured deadlines and raises
+ * `StreamStalledError` when one expires.
+ */
+
+/**
+ * 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
+ * even receive keep-alive framing) 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;
+
+    // In older runtimes the prototype chain is not set up correctly by
+    // super() calls on subclasses of built-ins.
+    Object.setPrototypeOf(this, StreamStalledError.prototype);
+  }
+
+  /**
+   * Whether the request is safe to retry without risking duplicated output.
+   * Only true when the stall happened before any content was received.
+   */
+  get retryable(): boolean {
+    return !this.receivedAnyContent;
+  }
+}
diff --git a/src/lib/stream-watchdog.ts b/src/lib/stream-watchdog.ts
new file mode 100644
index 000000000..f3e071a5e
--- /dev/null
+++ b/src/lib/stream-watchdog.ts
@@ -0,0 +1,295 @@
+/**
+ * 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-level timeouts cannot catch this: keep-alive SSE
+ * comments reset socket idle timers even though no content is flowing.
+ *
+ * The watchdog wraps a parsed event stream and enforces two semantic,
+ * opt-in deadlines:
+ *
+ * - `firstContentMs` — armed when the stream starts; satisfied (and
+ *   permanently disarmed) by the first content-bearing event.
+ * - `contentIntervalMs` — after content has started, the maximum gap
+ *   allowed between content-bearing events.
+ *
+ * Classification follows the parsed event's 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 (which the SSE parser drops 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.
+ */
+
+import type * as models from '../models/index.js';
+
+import { StreamStalledError, type StreamStallPhase } from './stream-errors.js';
+
+/**
+ * Opt-in stream stall timeouts. 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, body stream available) and the first content-bearing event.
+   * Unset by default.
+   */
+  firstContentMs?: number | undefined;
+  /**
+   * Maximum milliseconds between consecutive content-bearing events once
+   * content has started flowing. Does not govern the pre-first-content
+   * window — that is `firstContentMs`'s job. Unset by default.
+   */
+  contentIntervalMs?: number | undefined;
+};
+
+/**
+ * Event classification hooks for {@link applyStreamWatchdog}.
+ */
+export type StreamWatchdogHooks<T> = {
+  /**
+   * Returns true when the event carries model-generated output (or a
+   * completed sub-result). Content events satisfy and re-arm deadlines.
+   */
+  isContentEvent: (event: T) => boolean;
+  /**
+   * Returns true when the event signals the response is finishing
+   * (completed / failed / incomplete / error). Terminal events permanently
+   * disarm the watchdog so trailing bookkeeping events are never killed.
+   */
+  isTerminalEvent?: ((event: T) => boolean) | undefined;
+  /**
+   * Invoked once, before the wrapped stream errors, when a deadline
+   * expires. Phase 2 uses this to abort the underlying HTTP request.
+   */
+  onStall?: ((error: StreamStalledError) => void) | undefined;
+};
+
+/**
+ * Wrap `source` with stall deadlines. Returns `source` unchanged when no
+ * deadline is configured.
+ *
+ * The wrapper pumps eagerly (it does not propagate backpressure); parsed
+ * SSE events are small and downstream consumers buffer regardless.
+ */
+export function applyStreamWatchdog<T>(
+  source: ReadableStream<T>,
+  timeouts: StreamTimeoutOptions,
+  hooks: StreamWatchdogHooks<T>,
+): ReadableStream<T> {
+  const firstContentMs = normalizeTimeout(timeouts.firstContentMs);
+  const contentIntervalMs = normalizeTimeout(timeouts.contentIntervalMs);
+
+  if (firstContentMs === undefined && contentIntervalMs === undefined) {
+    return source;
+  }
+
+  const reader = source.getReader();
+
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  let armedPhase: StreamStallPhase = 'first_content';
+  let armedTimeoutMs = 0;
+  let armedAtMs = 0;
+  let receivedContent = false;
+  /** Set by a terminal event: deadlines permanently stop applying. */
+  let disarmedForever = false;
+  /** Set once the wrapped stream has closed, errored, or stalled. */
+  let settled = false;
+
+  const clearTimer = (): void => {
+    if (timer !== undefined) {
+      clearTimeout(timer);
+      timer = undefined;
+    }
+  };
+
+  return new ReadableStream<T>({
+    start(controller) {
+      const onDeadlineExpired = (): void => {
+        if (settled) {
+          return;
+        }
+        settled = true;
+        timer = undefined;
+        const error = new StreamStalledError({
+          phase: armedPhase,
+          timeoutMs: armedTimeoutMs,
+          elapsedMs: Date.now() - armedAtMs,
+          receivedAnyContent: receivedContent,
+        });
+        hooks.onStall?.(error);
+        controller.error(error);
+        // Cancelling the reader resolves the pump's pending read; the
+        // settled flag makes the pump bail without touching the controller.
+        void reader.cancel(error).catch(() => {
+          // Cancellation failures are irrelevant once the stream errored.
+        });
+      };
+
+      const arm = (phase: StreamStallPhase, timeoutMs: number): void => {
+        clearTimer();
+        armedPhase = phase;
+        armedTimeoutMs = timeoutMs;
+        armedAtMs = Date.now();
+        timer = setTimeout(onDeadlineExpired, timeoutMs);
+      };
+
+      // The pre-first-content deadline arms the moment the stream starts.
+      // `contentIntervalMs` alone does not cover that window.
+      if (firstContentMs !== undefined) {
+        arm('first_content', firstContentMs);
+      }
+
+      void (async () => {
+        try {
+          while (true) {
+            const result = await reader.read();
+            if (settled) {
+              return;
+            }
+            if (result.done) {
+              settled = true;
+              clearTimer();
+              controller.close();
+              return;
+            }
+
+            const event = result.value;
+            if (hooks.isTerminalEvent?.(event)) {
+              // Response is finishing; deadlines permanently stop applying
+              // so trailing bookkeeping events are never treated as stalls.
+              disarmedForever = true;
+              clearTimer();
+            } else if (!disarmedForever && hooks.isContentEvent(event)) {
+              receivedContent = true;
+              if (contentIntervalMs !== undefined) {
+                arm('between_content', contentIntervalMs);
+              } else {
+                clearTimer();
+              }
+            }
+            controller.enqueue(event);
+          }
+        } catch (error) {
+          if (settled) {
+            return;
+          }
+          settled = true;
+          clearTimer();
+          controller.error(error);
+        }
+      })();
+    },
+    cancel(reason) {
+      settled = true;
+      clearTimer();
+      return reader.cancel(reason);
+    },
+  });
+}
+
+/**
+ * Stream event types that carry model-generated 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` / `response.debug` — 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.
+ * - Unknown event types — a healthy stream of only unknown types is
+ *   indistinguishable from a stall; mixed streams disarm via known events.
+ */
+const CONTENT_BEARING_EVENT_TYPES: ReadonlySet<string> = 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.completed',
+]);
+
+/**
+ * Stream event types that signal the response is finishing. They disarm
+ * the watchdog permanently: the server has produced its verdict, so stall
+ * deadlines no longer apply (server-reported failures surface through
+ * their own error paths, not as stalls).
+ */
+const TERMINAL_EVENT_TYPES: ReadonlySet<string> = new Set([
+  'response.completed',
+  'response.failed',
+  'response.incomplete',
+  'error',
+]);
+
+/**
+ * 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 {
+  return (
+    'type' in event && typeof event.type === 'string' && CONTENT_BEARING_EVENT_TYPES.has(event.type)
+  );
+}
+
+/**
+ * True when the event signals the response is finishing (completed,
+ * failed, incomplete, or a server-emitted error event).
+ */
+export function isTerminalStreamEvent(event: models.StreamEvents): boolean {
+  return 'type' in event && typeof event.type === 'string' && TERMINAL_EVENT_TYPES.has(event.type);
+}
+
+/**
+ * Convenience wrapper of {@link applyStreamWatchdog} for OpenResponses
+ * event streams, using the standard event classification.
+ */
+export function applyResponsesStreamWatchdog(
+  source: ReadableStream<models.StreamEvents>,
+  timeouts: StreamTimeoutOptions,
+  hooks?: { onStall?: ((error: StreamStalledError) => void) | undefined },
+): ReadableStream<models.StreamEvents> {
+  return applyStreamWatchdog(source, timeouts, {
+    isContentEvent: isContentBearingStreamEvent,
+    isTerminalEvent: isTerminalStreamEvent,
+    onStall: hooks?.onStall,
+  });
+}
+
+/**
+ * Treat non-finite and non-positive values as "disabled" so callers can
+ * pass raw user input without pre-validating.
+ */
+function normalizeTimeout(value: number | undefined): number | undefined {
+  if (value === undefined || !Number.isFinite(value) || value <= 0) {
+    return undefined;
+  }
+  return value;
+}
diff --git a/tests/unit/stream-watchdog.test.ts b/tests/unit/stream-watchdog.test.ts
new file mode 100644
index 000000000..077debd0f
--- /dev/null
+++ b/tests/unit/stream-watchdog.test.ts
@@ -0,0 +1,530 @@
+import type { StreamEvents } from '../../src/models/streamevents.js';
+import type { TextDeltaEvent } from '../../src/models/textdeltaevent.js';
+import type { ReasoningDeltaEvent } from '../../src/models/reasoningdeltaevent.js';
+import type { FunctionCallArgsDeltaEvent } from '../../src/models/functioncallargsdeltaevent.js';
+import type { OpenResponsesCreatedEvent } from '../../src/models/openresponsescreatedevent.js';
+import type { OpenResponsesResult } from '../../src/models/openresponsesresult.js';
+import type { StreamEventsResponseCompleted } from '../../src/models/streamevents.js';
+import type { StreamEventsResponseOutputItemAdded } from '../../src/models/streamevents.js';
+
+import { describe, expect, it } from 'vitest';
+import { StreamStalledError } from '../../src/lib/stream-errors.js';
+import {
+  applyResponsesStreamWatchdog,
+  applyStreamWatchdog,
+  isContentBearingStreamEvent,
+  isTerminalStreamEvent,
+} from '../../src/lib/stream-watchdog.js';
+
+// ============================================================================
+// Scripted stream helpers
+//
+// The watchdog measures real elapsed time between events, so these tests use
+// short real delays (tens of ms) rather than fake timers: vitest fake timers
+// cannot advance a timer that races a genuinely pending microtask-driven
+// stream read without also stepping the stream's own delays.
+// ============================================================================
+
+type ScriptStep<T> =
+  | { kind: 'event'; value: T; delayMs: number }
+  | { kind: 'close'; delayMs: number }
+  | { kind: 'silence'; delayMs: number };
+
+function event<T>(value: T, delayMs = 0): ScriptStep<T> {
+  return { kind: 'event', value, delayMs };
+}
+
+function close<T>(delayMs = 0): ScriptStep<T> {
+  return { kind: 'close', delayMs };
+}
+
+/** A gap with no events and no close — the stream just goes quiet. */
+function silence<T>(delayMs: number): ScriptStep<T> {
+  return { kind: 'silence', delayMs };
+}
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+/**
+ * Build a ReadableStream that plays back the scripted steps. If the script
+ * ends without an explicit `close`, the stream stays open forever (hung
+ * connection) — exactly the pathology the watchdog exists to catch.
+ */
+function scriptedStream<T>(steps: ScriptStep<T>[]): ReadableStream<T> {
+  let cancelled = false;
+  return new ReadableStream<T>({
+    start(controller) {
+      void (async () => {
+        for (const step of steps) {
+          await sleep(step.delayMs);
+          if (cancelled) {
+            return;
+          }
+          if (step.kind === 'event') {
+            controller.enqueue(step.value);
+          } else if (step.kind === 'close') {
+            controller.close();
+            return;
+          }
+          // 'silence' steps only consume time.
+        }
+        // No explicit close: leave the stream hanging open.
+      })();
+    },
+    cancel() {
+      cancelled = true;
+    },
+  });
+}
+
+async function collect<T>(stream: ReadableStream<T>): Promise<T[]> {
+  const reader = stream.getReader();
+  const out: T[] = [];
+  while (true) {
+    const result = await reader.read();
+    if (result.done) {
+      return out;
+    }
+    out.push(result.value);
+  }
+}
+
+async function collectError<T>(stream: ReadableStream<T>): Promise<{ events: T[]; error: unknown }> {
+  const reader = stream.getReader();
+  const events: T[] = [];
+  try {
+    while (true) {
+      const result = await reader.read();
+      if (result.done) {
+        throw new Error('Expected the stream to error, but it closed cleanly');
+      }
+      events.push(result.value);
+    }
+  } catch (error) {
+    return { events, error };
+  }
+}
+
+// ============================================================================
+// Typed OpenResponses event fixtures
+// ============================================================================
+
+function fakeResponse(status: OpenResponsesResult['status']): OpenResponsesResult {
+  return {
+    id: 'resp_1',
+    object: 'response',
+    createdAt: 0,
+    model: 'test-model',
+    status,
+    completedAt: null,
+    output: [],
+    error: null,
+    incompleteDetails: null,
+    temperature: null,
+    topP: null,
+    presencePenalty: null,
+    frequencyPenalty: null,
+    metadata: null,
+    tools: [],
+    toolChoice: 'auto',
+    parallelToolCalls: false,
+  };
+}
+
+function createdEvent(): OpenResponsesCreatedEvent {
+  return { type: 'response.created', response: fakeResponse('in_progress'), sequenceNumber: 0 };
+}
+
+function textDelta(delta: string): TextDeltaEvent {
+  return {
+    type: 'response.output_text.delta',
+    delta,
+    itemId: 'item_1',
+    contentIndex: 0,
+    outputIndex: 0,
+    logprobs: [],
+    sequenceNumber: 1,
+  };
+}
+
+function reasoningDelta(delta: string): ReasoningDeltaEvent {
+  return {
+    type: 'response.reasoning_text.delta',
+    delta,
+    itemId: 'item_1',
+    contentIndex: 0,
+    outputIndex: 0,
+    sequenceNumber: 1,
+  };
+}
+
+function functionCallArgsDelta(delta: string): FunctionCallArgsDeltaEvent {
+  return {
+    type: 'response.function_call_arguments.delta',
+    delta,
+    itemId: 'item_1',
+    outputIndex: 0,
+    sequenceNumber: 1,
+  };
+}
+
+/** An empty assistant-message shell: the "role prelude" that must NOT count as content. */
+function emptyMessageShell(): StreamEventsResponseOutputItemAdded {
+  return {
+    type: 'response.output_item.added',
+    outputIndex: 0,
+    item: {
+      type: 'message',
+      id: 'item_1',
+      role: 'assistant',
+      status: 'in_progress',
+      content: [],
+    },
+    sequenceNumber: 0,
+  };
+}
+
+function completedEvent(): StreamEventsResponseCompleted {
+  return {
+    type: 'response.completed',
+    response: fakeResponse('completed'),
+    sequenceNumber: 9,
+  };
+}
+
+// ============================================================================
+// Event classification
+// ============================================================================
+
+describe('isContentBearingStreamEvent', () => {
+  it('classifies output deltas as content', () => {
+    expect(isContentBearingStreamEvent(textDelta('hi'))).toBe(true);
+    expect(isContentBearingStreamEvent(reasoningDelta('hmm'))).toBe(true);
+    expect(isContentBearingStreamEvent(functionCallArgsDelta('{'))).toBe(true);
+  });
+
+  it('does not classify status events or item shells as content', () => {
+    expect(isContentBearingStreamEvent(createdEvent())).toBe(false);
+    expect(isContentBearingStreamEvent(emptyMessageShell())).toBe(false);
+    expect(isContentBearingStreamEvent(completedEvent())).toBe(false);
+  });
+
+  it('does not classify unknown event types as content', () => {
+    const unknown = { type: 'UNKNOWN', raw: { type: 'response.mystery' }, isUnknown: true } as StreamEvents;
+    expect(isContentBearingStreamEvent(unknown)).toBe(false);
+  });
+});
+
+describe('isTerminalStreamEvent', () => {
+  it('classifies completion / failure / error events as terminal', () => {
+    expect(isTerminalStreamEvent(completedEvent())).toBe(true);
+    expect(
+      isTerminalStreamEvent({
+        type: 'response.failed',
+        response: fakeResponse('failed'),
+        sequenceNumber: 3,
+      }),
+    ).toBe(true);
+    expect(
+      isTerminalStreamEvent({
+        type: 'error',
+        code: 'server_error',
+        message: 'boom',
+        param: null,
+        sequenceNumber: 3,
+      }),
+    ).toBe(true);
+  });
+
+  it('does not classify content or status events as terminal', () => {
+    expect(isTerminalStreamEvent(textDelta('hi'))).toBe(false);
+    expect(isTerminalStreamEvent(createdEvent())).toBe(false);
+  });
+});
+
+// ============================================================================
+// Watchdog behavior (generic)
+// ============================================================================
+
+describe('applyStreamWatchdog', () => {
+  const hooks = {
+    isContentEvent: (e: string) => e.startsWith('content'),
+    isTerminalEvent: (e: string) => e === 'terminal',
+  };
+
+  it('returns the source stream unchanged when no timeout is configured', () => {
+    const source = scriptedStream<string>([close()]);
+    expect(applyStreamWatchdog(source, {}, hooks)).toBe(source);
+    // Also for explicitly disabled values:
+    const source2 = scriptedStream<string>([close()]);
+    expect(applyStreamWatchdog(source2, { firstContentMs: 0, contentIntervalMs: -5 }, hooks)).toBe(
+      source2,
+    );
+    const source3 = scriptedStream<string>([close()]);
+    expect(applyStreamWatchdog(source3, { firstContentMs: Number.NaN }, hooks)).toBe(source3);
+  });
+
+  it('passes a healthy stream through untouched', async () => {
+    const wrapped = applyStreamWatchdog(
+      scriptedStream([event('content-1', 5), event('content-2', 5), close(5)]),
+      { firstContentMs: 1000, contentIntervalMs: 1000 },
+      hooks,
+    );
+    expect(await collect(wrapped)).toEqual(['content-1', 'content-2']);
+  });
+
+  it('fails with first_content when no content arrives at all', async () => {
+    const wrapped = applyStreamWatchdog(
+      scriptedStream<string>([]), // hangs forever
+      { firstContentMs: 40 },
+      hooks,
+    );
+    const { events, error } = await collectError(wrapped);
+    expect(events).toEqual([]);
+    expect(error).toBeInstanceOf(StreamStalledError);
+    const stall = error as StreamStalledError;
+    expect(stall.phase).toBe('first_content');
+    expect(stall.timeoutMs).toBe(40);
+    expect(stall.receivedAnyContent).toBe(false);
+    expect(stall.retryable).toBe(true);
+    expect(stall.elapsedMs).toBeGreaterThanOrEqual(30);
+  });
+
+  it('fails with first_content when only non-content events arrive (metadata-then-stall)', async () => {
+    const wrapped = applyStreamWatchdog(
+      scriptedStream([event('meta-1', 5), event('meta-2', 10)]), // then hangs
+      { firstContentMs: 50 },
+      hooks,
+    );
+    const { events, error } = await collectError(wrapped);
+    // Non-content events still flow through to the consumer...
+    expect(events).toEqual(['meta-1', 'meta-2']);
+    // ...but they do not reset the first-content deadline.
+    expect(error).toBeInstanceOf(StreamStalledError);
+    expect((error as StreamStalledError).phase).toBe('first_content');
+    expect((error as StreamStalledError).retryable).toBe(true);
+  });
+
+  it('non-content events do not extend the first_content deadline', async () => {
+    // Steady metadata heartbeats every 20ms, forever. If they reset the
+    // deadline the watchdog would never fire.
+    const heartbeats: ScriptStep<string>[] = Array.from({ length: 50 }, () => event('meta', 20));
+    const wrapped = applyStreamWatchdog(scriptedStream(heartbeats), { firstContentMs: 90 }, hooks);
+    const startedAt = Date.now();
+    const { error } = await collectError(wrapped);
+    expect(error).toBeInstanceOf(StreamStalledError);
+    expect((error as StreamStalledError).phase).toBe('first_content');
+    // Fired near the deadline, not after 50 * 20ms of heartbeats.
+    expect(Date.now() - startedAt).toBeLessThan(500);
+  });
+
+  it('first content event disarms firstContentMs permanently', async () => {
+    const wrapped = applyStreamWatchdog(
+      // Content at 10ms, then a 100ms quiet gap, then close — the gap is
+      // longer than firstContentMs but there is no content-interval timeout.
+      scriptedStream([event('content-1', 10), event('content-2', 100), close()]),
+      { firstContentMs: 50 },
+      hooks,
+    );
+    expect(await collect(wrapped)).toEqual(['content-1', 'content-2']);
+  });
+
+  it('fails with between_content when the stream stalls mid-generation', async () => {
+    const wrapped = applyStreamWatchdog(
+      scriptedStream([event('content-1', 5), event('content-2', 10), silence(1000)]),
+      { firstContentMs: 200, contentIntervalMs: 50 },
+      hooks,
+    );
+    const { events, error } = await collectError(wrapped);
+    expect(events).toEqual(['content-1', 'content-2']);
+    expect(error).toBeInstanceOf(StreamStalledError);
+    const stall = 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('content events re-arm the between_content deadline', async () => {
+    // Four content events, each 30ms apart, with a 50ms interval timeout:
+    // each event must reset the clock or this would fail spuriously.
+    const wrapped = applyStreamWatchdog(
+      scriptedStream([
+        event('content-1', 30),
+        event('content-2', 30),
+        event('content-3', 30),
+        event('content-4', 30),
+        close(10),
+      ]),
+      { firstContentMs: 100, contentIntervalMs: 50 },
+      hooks,
+    );
+    expect(await collect(wrapped)).toEqual(['content-1', 'content-2', 'content-3', 'content-4']);
+  });
+
+  it('contentIntervalMs alone does not police the pre-first-content window', async () => {
+    // 80ms of silence before the first content, but only contentIntervalMs
+    // configured — the pre-content window is firstContentMs's job.
+    const wrapped = applyStreamWatchdog(
+      scriptedStream([event('content-1', 80), close()]),
+      { contentIntervalMs: 30 },
+      hooks,
+    );
+    expect(await collect(wrapped)).toEqual(['content-1']);
+  });
+
+  it('a terminal event disarms deadlines so trailing events are never stalls', async () => {
+    const wrapped = applyStreamWatchdog(
+      scriptedStream([
+        event('content-1', 5),
+        event('terminal', 5),
+        // Long post-terminal gap before bookkeeping + close.
+        event('meta-trailer', 100),
+        close(),
+      ]),
+      { firstContentMs: 50, contentIntervalMs: 30 },
+      hooks,
+    );
+    expect(await collect(wrapped)).toEqual(['content-1', 'terminal', 'meta-trailer']);
+  });
+
+  it('a clean close cancels all deadlines', async () => {
+    const wrapped = applyStreamWatchdog(
+      scriptedStream([event('meta', 5), close(5)]),
+      { firstContentMs: 50 },
+      hooks,
+    );
+    expect(await collect(wrapped)).toEqual(['meta']);
+    // Wait past the deadline to catch a leaked timer firing on a closed stream.
+    await sleep(80);
+  });
+
+  it('cancels the source stream when a deadline expires', async () => {
+    let sourceCancelled = false;
+    const source = new ReadableStream<string>({
+      cancel() {
+        sourceCancelled = true;
+      },
+    });
+    const wrapped = applyStreamWatchdog(source, { firstContentMs: 30 }, hooks);
+    const { error } = await collectError(wrapped);
+    expect(error).toBeInstanceOf(StreamStalledError);
+    // Reader.cancel resolves asynchronously.
+    await sleep(10);
+    expect(sourceCancelled).toBe(true);
+  });
+
+  it('invokes onStall exactly once before erroring', async () => {
+    const stalls: StreamStalledError[] = [];
+    const wrapped = applyStreamWatchdog(
+      scriptedStream<string>([]),
+      { firstContentMs: 30 },
+      { ...hooks, onStall: (e) => stalls.push(e) },
+    );
+    const { error } = await collectError(wrapped);
+    await sleep(50); // room for any duplicate firing
+    expect(stalls).toHaveLength(1);
+    expect(stalls[0]).toBe(error);
+  });
+
+  it('propagates upstream errors as-is (not wrapped in StreamStalledError)', async () => {
+    const upstreamFailure = new Error('upstream exploded');
+    const source = new ReadableStream<string>({
+      start(controller) {
+        setTimeout(() => controller.error(upstreamFailure), 10);
+      },
+    });
+    const wrapped = applyStreamWatchdog(source, { firstContentMs: 1000 }, hooks);
+    const { error } = await collectError(wrapped);
+    expect(error).toBe(upstreamFailure);
+  });
+
+  it('cancelling the wrapped stream stops timers and cancels the source', async () => {
+    let sourceCancelled = false;
+    const source = new ReadableStream<string>({
+      cancel() {
+        sourceCancelled = true;
+      },
+    });
+    const wrapped = applyStreamWatchdog(source, { firstContentMs: 30 }, hooks);
+    await wrapped.cancel('consumer walked away');
+    expect(sourceCancelled).toBe(true);
+    // Wait past the deadline: the timer must not fire after cancellation.
+    await sleep(60);
+  });
+});
+
+// ============================================================================
+// Responses-flavored wrapper
+// ============================================================================
+
+describe('applyResponsesStreamWatchdog', () => {
+  it('stalls on the OpenRouter keepalive pathology: created + role prelude, then silence', async () => {
+    // This is the exact DEV-723 scenario: headers arrive, response.created
+    // and an empty message shell stream in, then nothing. (SSE keep-alive
+    // comments never reach this layer — the SSE parser drops them.)
+    const wrapped = applyResponsesStreamWatchdog(
+      scriptedStream<StreamEvents>([event(createdEvent(), 5), event(emptyMessageShell(), 5)]),
+      { firstContentMs: 60 },
+    );
+    const { events, error } = await collectError(wrapped);
+    expect(events).toHaveLength(2);
+    expect(error).toBeInstanceOf(StreamStalledError);
+    const stall = error as StreamStalledError;
+    expect(stall.phase).toBe('first_content');
+    expect(stall.retryable).toBe(true);
+  });
+
+  it('passes a healthy responses stream through to completion', async () => {
+    const wrapped = applyResponsesStreamWatchdog(
+      scriptedStream<StreamEvents>([
+        event(createdEvent(), 5),
+        event(emptyMessageShell(), 5),
+        event(textDelta('Hello'), 5),
+        event(textDelta(' world'), 5),
+        event(completedEvent(), 5),
+        close(5),
+      ]),
+      { firstContentMs: 100, contentIntervalMs: 100 },
+    );
+    const collected = await collect(wrapped);
+    expect(collected).toHaveLength(5);
+    expect(collected.at(-1)?.type).toBe('response.completed');
+  });
+
+  it('reasoning deltas satisfy the first-content deadline (reasoning models)', async () => {
+    const wrapped = applyResponsesStreamWatchdog(
+      scriptedStream<StreamEvents>([
+        event(createdEvent(), 5),
+        // Reasoning streams for a while before any output text appears.
+        event(reasoningDelta('thinking...'), 10),
+        event(textDelta('answer'), 80),
+        event(completedEvent(), 5),
+        close(),
+      ]),
+      { firstContentMs: 50 },
+    );
+    const collected = await collect(wrapped);
+    expect(collected).toHaveLength(4);
+  });
+
+  it('stalls mid-generation when deltas stop and nothing terminal arrives', async () => {
+    const wrapped = applyResponsesStreamWatchdog(
+      scriptedStream<StreamEvents>([
+        event(createdEvent(), 5),
+        event(textDelta('partial outp'), 5),
+        silence(1000),
+      ]),
+      { firstContentMs: 100, contentIntervalMs: 60 },
+    );
+    const { error } = await collectError(wrapped);
+    expect(error).toBeInstanceOf(StreamStalledError);
+    const stall = error as StreamStalledError;
+    expect(stall.phase).toBe('between_content');
+    expect(stall.receivedAnyContent).toBe(true);
+    expect(stall.retryable).toBe(false);
+  });
+});