feat: stream watchdog core for stalled-stream detection (DEV-723 1/5)

[LukasParke]

Stack layer 1/5 — stream watchdog core

Part of a stacked series for DEV-723 (client-side counterpart of DEV-721; prior art vercel/ai#17315 firstChunkMs):

1. This PR — watchdog + typed errors
2. feat: callModel per-turn stall timeouts (DEV-723 2/5) #773 — callModel integration (timeout option, per-turn re-arm, abort-signal merge)
3. feat: typed StreamFailedError for server stream failures (DEV-723 3/5) #774 — typed StreamFailedError for server stream failures
4. feat: opt-in pre-content stall retries (DEV-723 4/5) #775 — opt-in pre-content stall retries (maxStallRetries)
5. feat: raw-stream watchdog helpers + docs (DEV-723 5/5) #776 — raw-stream helpers (applyChatStreamWatchdog / applyResponsesStreamWatchdog) + docs

Problem

A streaming response can return headers quickly, emit only keep-alive framing or metadata events, and then stall indefinitely. Transport timeouts can't catch this — the router's SSE keep-alive comments reset socket idle timers while no content flows.

Changes

• src/lib/stream-errors.ts — StreamStalledError with phase (first_content | between_content), timeoutMs, elapsedMs, receivedAnyContent, and a retryable getter (true only when no content was received, so retrying cannot duplicate output).

• src/lib/stream-watchdog.ts — applyStreamWatchdog wraps a parsed event stream with two opt-in semantic deadlines:

  • firstContentMs: stream start → first content-bearing event
  • contentIntervalMs: max gap between content events afterwards

  Classification is by parsed event type (mirroring isOutputChunkType semantics): deltas / tool-call args / completed items satisfy and re-arm; response.created, response.in_progress, empty message shells, and unknown event types are neutral; response.completed/failed/incomplete/error disarm permanently. On expiry the wrapped stream errors and the source is cancelled. applyResponsesStreamWatchdog packages the standard OpenResponses classification.

• tests/unit/stream-watchdog.test.ts — 24 tests: the DEV-723 pathology (created + role prelude → silence), heartbeats-don't-reset-deadlines, re-arm behavior, terminal disarm, source cancellation, timer cleanup, error passthrough, retryability.

Unset timeouts = no watchdog, source returned unchanged → fully non-breaking. Hand-written src/lib/ files only — safe under Speakeasy persistent edits.

Verification

• pnpm run lint / typecheck / build clean
• Full unit suite: 214 passed, vitest typecheck no errors

[perry-the-pr-reviewer]

Perry's Review

Verdict: 💬 Comments / questions
Risk: 🟢 Low

Well-structured core with thorough test coverage. The watchdog logic is race-safe, timer cleanup covers every exit path (close, error, stall, consumer cancel), and the opt-in defaults make this fully non-breaking. One classification gap to address before the integration PRs land.

Findings

Suggestion — response.fusion_call.analysis.completed missing from content-bearing event set

FusionCallAnalysisCompletedEvent carries analysis: FusionAnalysisResult — the fusion analyst's structured output. The code's own documentation says content events include "completed output items or sub-results," and every other fusion sub-result (panel.completed, panel.failed, fusion_call.completed) is already in CONTENT_BEARING_EVENT_TYPES. Omitting analysis.completed means a fusion stream that delivers only the analysis result (no panel deltas) would never satisfy the firstContentMs deadline, producing a false first_content stall. The fix is one line — add 'response.fusion_call.analysis.completed' to the set.

What I checked and found clean

• Race safety: The async pump yields only at await reader.read(), so the timer callback (onDeadlineExpired) can only fire between pump iterations, never mid-iteration. The settled flag guards every exit (close, error, stall, cancel), and clearTimer() is called on all settled paths. No leaked timers.
• Event classification: Cross-referenced all 45 event types in the StreamEvents discriminated union against the content and terminal sets. Every type is correctly classified (status/shell events neutral, deltas/completions content, response-end events terminal) except the one above.
• Error passthrough: Upstream errors propagate as-is, not wrapped in StreamStalledError. Correct.
• Retryability: retryable getter returns !receivedAnyContent — safe to retry only when no content was emitted. Correct and important for avoiding duplicated output.
• normalizeTimeout: Treats NaN, Infinity, and <= 0 as disabled. Callers can pass raw user input without pre-validating. Good defensive design.
• Test coverage: 24 tests covering the DEV-723 pathology (created + role prelude → silence), heartbeats-don't-reset, first-content disarm, between-content re-arm, terminal disarm, source cancellation, timer cleanup, error passthrough, consumer cancel. Uses real timers (documented reason: vitest fake timers can't race with microtask-driven stream reads).
• Non-breaking: Both timeouts default to undefined → applyStreamWatchdog returns the source unchanged. No existing callers affected.

Risk: 🟢 Low

Risk assessment:

Dimension	Severity	Risk	Reasoning
Implementation risk	🟩	Low	Code is race-safe and thoroughly tested; the one classification gap is trivial to fix and only affects a specific event type.
Premise risk	🟩	Low	The semantic-deadline approach (classify by event type, not transport activity) is sound and validated by prior art (vercel/ai firstChunkMs).
Estimated impact	🟩	Low	The watchdog is opt-in (defaults to no-op) and not yet wired into the SDK request flow; worst case is a false StreamStalledError for fusion analysis streams, which is retryable: true.

Risk Factor	Severity	Risk	Reasoning
Reversibility	🟩	Low	Removing or changing a timeout option is a one-line change with no persisted state.
Detectability	🟩	Low	A false stall produces an immediate StreamStalledError — not a silent failure.
Blast radius	🟩	Low	Only affects callers who opt in to the watchdog; no impact on existing streams.
Data integrity		None	No persisted state is touched.
Financial exposure		None	No billing or payment paths involved.
Security and privacy exposure		None	No credentials, auth, or tenant isolation involved.
Propagation	🟩	Low	This is the core layer; integration PRs (#773–776) will wire it, so a classification bug here propagates to all callers.
Availability	🟩	Low	A false stall would abort a stream, but the error is retryable: true for the affected scenario.
Recovery cost	🟩	Low	The fix is adding one string to a Set.
Time to correct	🟩	Low	The missing event type can be added in the same PR or a quick follow-up before integration.

[perry-the-pr-reviewer]

Suggestion: add response.fusion_call.analysis.completed to the content-bearing event set

FusionCallAnalysisCompletedEvent carries analysis: FusionAnalysisResult — the fusion analyst's structured output. This is a completed sub-result, which the code's own comment says should be content-bearing. Every other fusion sub-result (panel.completed, panel.failed, fusion_call.completed) is already in this set.

Omitting analysis.completed means a fusion stream that delivers only the analysis result (no panel deltas) would never satisfy the firstContentMs deadline — the watchdog would fire a false first_content stall even though the analysis was delivered.

The fix is one line:

'response.fusion_call.analysis.completed',

Consider adding a test case that scripts a fusion analysis-only stream (created → fusion_call.in_progress → analysis.in_progress → analysis.completed) and verifies the deadline is satisfied.

▶ Prompt for agents: Add 'response.fusion_call.analysis.completed' to the CONTENT_BEARING_EVENT_TYPES set in src/lib/stream-watchdog.ts. Add a test in tests/unit/stream-watchdog.test.ts that scripts a fusion analysis-only stream and verifies it satisfies the firstContentMs deadline without a false stall.