diff --git a/desktop/src/stores/chatStore.test.ts b/desktop/src/stores/chatStore.test.ts index 9cf8421a..f0ae09d9 100644 --- a/desktop/src/stores/chatStore.test.ts +++ b/desktop/src/stores/chatStore.test.ts @@ -4396,13 +4396,17 @@ describe('chatStore history mapping', () => { }) useChatStore.getState().handleServerMessage(TEST_SESSION_ID, { type: 'error', - message: 'provider failed', - code: 'provider_error', + message: 'API Error: Provider stream stalled after partial response - no new chunks for 240s', + code: 'STREAM_IDLE_TIMEOUT', }) expect(useChatStore.getState().sessions[TEST_SESSION_ID]?.messages).toMatchObject([ { type: 'assistant_text', content: 'partial answer before error' }, - { type: 'error', message: 'provider failed' }, + { + type: 'error', + message: 'API Error: Provider stream stalled after partial response - no new chunks for 240s', + code: 'STREAM_IDLE_TIMEOUT', + }, ]) vi.runOnlyPendingTimers() diff --git a/src/server/__tests__/ws-memory-events.test.ts b/src/server/__tests__/ws-memory-events.test.ts index a1002621..8556bd96 100644 --- a/src/server/__tests__/ws-memory-events.test.ts +++ b/src/server/__tests__/ws-memory-events.test.ts @@ -27,6 +27,46 @@ describe('WebSocket memory events', () => { ]) }) + it('maps watchdog API errors to stable desktop error codes', () => { + expect(translateCliMessage({ + type: 'assistant', + error: 'server_error', + isApiErrorMessage: true, + message: { + role: 'assistant', + content: [{ + type: 'text', + text: 'API Error: Provider stream stalled after partial response - no new chunks for 240s (last event: text_delta, content deltas: 1)', + }], + }, + }, 'session-1')).toEqual([ + { + type: 'error', + message: 'API Error: Provider stream stalled after partial response - no new chunks for 240s (last event: text_delta, content deltas: 1)', + code: 'STREAM_IDLE_TIMEOUT', + }, + ]) + + expect(translateCliMessage({ + type: 'assistant', + error: 'server_error', + isApiErrorMessage: true, + message: { + role: 'assistant', + content: [{ + type: 'text', + text: 'API Error: Stream max duration exceeded - no completion received after 600s (last event: text_delta)', + }], + }, + }, 'session-1')).toEqual([ + { + type: 'error', + message: 'API Error: Stream max duration exceeded - no completion received after 600s (last event: text_delta)', + code: 'STREAM_MAX_DURATION', + }, + ]) + }) + it('replays slash-command breadcrumbs as readable user messages', () => { expect(translateCliMessage({ type: 'user', diff --git a/src/server/ws/handler.ts b/src/server/ws/handler.ts index 7b3f78eb..ad112718 100644 --- a/src/server/ws/handler.ts +++ b/src/server/ws/handler.ts @@ -1357,6 +1357,19 @@ function isDuplicateOfLastApiError( ) } +function classifyRuntimeErrorCode(message: string, fallbackCode: string): string { + if (/Stream max duration exceeded/i.test(message)) { + return 'STREAM_MAX_DURATION' + } + if ( + /Provider stream stalled after partial response/i.test(message) || + /Stream idle timeout/i.test(message) + ) { + return 'STREAM_IDLE_TIMEOUT' + } + return fallbackCode +} + function bindPrewarmMetadataCapture(sessionId: string) { for (const msg of conversationService.getRecentSdkMessages(sessionId)) { cacheSessionInitMetadata(sessionId, msg) @@ -1444,7 +1457,8 @@ export function translateCliMessage(cliMsg: any, sessionId: string): ServerMessa return [] } const message = extractAssistantText(cliMsg) || cliMsg.error || 'Unknown API error' - const code = typeof cliMsg.error === 'string' ? cliMsg.error : 'API_ERROR' + const fallbackCode = typeof cliMsg.error === 'string' ? cliMsg.error : 'API_ERROR' + const code = classifyRuntimeErrorCode(message, fallbackCode) streamState.lastApiError = { message, code } return [{ type: 'error', @@ -1744,7 +1758,7 @@ export function translateCliMessage(cliMsg: any, sessionId: string): ServerMessa { type: 'error', message: resultMessage, - code: 'CLI_ERROR', + code: classifyRuntimeErrorCode(resultMessage, 'CLI_ERROR'), }, { type: 'message_complete', usage }, ] diff --git a/src/services/api/claude.ts b/src/services/api/claude.ts index 200134b1..4250f586 100644 --- a/src/services/api/claude.ts +++ b/src/services/api/claude.ts @@ -212,6 +212,10 @@ import { stopSessionActivity, } from "../../utils/sessionActivity.js"; import { shouldTriggerNonStreamingFallbackForEmptyStream } from "./streamFallback.js"; +import { + StreamWatchdogTimeoutError, + createStreamWatchdogState, +} from "./streamWatchdog.js"; import { jsonStringify } from "../../utils/slowOperations.js"; import { isBetaTracingEnabled, @@ -2009,6 +2013,8 @@ async function* queryModel( let streamIdleAborted = false; // Which watchdog tripped, so the thrown error message is accurate. let streamAbortReason: "idle" | "max_duration" | null = null; + const streamWatchdogState = createStreamWatchdogState(); + let streamWatchdogTimeoutError: StreamWatchdogTimeoutError | null = null; // performance.now() snapshot when watchdog fires, for measuring abort propagation delay let streamWatchdogFiredAt: number | null = null; let streamIdleWarningTimer: ReturnType | null = null; @@ -2035,11 +2041,19 @@ async function* queryModel( const warningMs = idleMs / 2; streamIdleWarningTimer = setTimeout( (warnMs) => { + const warningError = streamWatchdogState.createTimeoutError( + "idle", + warnMs, + ); logForDebugging( - `Streaming idle warning: no chunks received for ${warnMs / 1000}s`, + `Streaming idle warning: ${warningError.message}`, { level: "warn" }, ); - logForDiagnosticsNoPII("warn", "cli_streaming_idle_warning"); + logForDiagnosticsNoPII( + "warn", + "cli_streaming_idle_warning", + warningError.toDiagnosticData(), + ); }, warningMs, warningMs, @@ -2048,17 +2062,31 @@ async function* queryModel( streamIdleAborted = true; streamAbortReason = "idle"; streamWatchdogFiredAt = performance.now(); + streamWatchdogTimeoutError = streamWatchdogState.createTimeoutError( + "idle", + idleMs, + ); logForDebugging( - `Streaming idle timeout: no chunks received for ${idleMs / 1000}s, aborting stream`, + `Streaming idle timeout: ${streamWatchdogTimeoutError.message}, aborting stream`, { level: "error" }, ); - logForDiagnosticsNoPII("error", "cli_streaming_idle_timeout"); + logForDiagnosticsNoPII( + "error", + "cli_streaming_idle_timeout", + streamWatchdogTimeoutError.toDiagnosticData(), + ); logEvent("tengu_streaming_idle_timeout", { model: options.model as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, request_id: (streamRequestId ?? "unknown") as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, timeout_ms: idleMs, + reason: + streamWatchdogTimeoutError.reason as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, + phase: + streamWatchdogTimeoutError.phase as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, + content_delta_count: + streamWatchdogTimeoutError.streamSnapshot.contentDeltaCount, }); releaseStreamResources(); }, idleMs); @@ -2072,19 +2100,29 @@ async function* queryModel( streamIdleAborted = true; streamAbortReason = "max_duration"; streamWatchdogFiredAt = performance.now(); + streamWatchdogTimeoutError = streamWatchdogState.createTimeoutError( + "max_duration", + STREAM_MAX_DURATION_MS, + ); logForDebugging( - `Streaming max duration exceeded: no completion after ${STREAM_MAX_DURATION_MS / 1000}s, aborting stream`, + `Streaming max duration exceeded: ${streamWatchdogTimeoutError.message}, aborting stream`, { level: "error" }, ); logForDiagnosticsNoPII("error", "cli_streaming_max_duration_exceeded", { - timeoutMs: STREAM_MAX_DURATION_MS, + ...streamWatchdogTimeoutError.toDiagnosticData(), }); - logEvent("tengu_streaming_idle_timeout", { + logEvent("tengu_streaming_max_duration_timeout", { model: options.model as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, request_id: (streamRequestId ?? "unknown") as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, timeout_ms: STREAM_MAX_DURATION_MS, + reason: + streamWatchdogTimeoutError.reason as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, + phase: + streamWatchdogTimeoutError.phase as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, + content_delta_count: + streamWatchdogTimeoutError.streamSnapshot.contentDeltaCount, }); releaseStreamResources(); }, STREAM_MAX_DURATION_MS); @@ -2100,6 +2138,7 @@ async function* queryModel( let stallCount = 0; for await (const part of stream) { + const receivedFirstContentDelta = streamWatchdogState.recordEvent(part); resetStreamIdleTimer(); const now = Date.now(); @@ -2136,13 +2175,18 @@ async function* queryModel( } endQueryProfile(); isFirstChunk = false; - // Tokens are flowing — switch the watchdog from the generous - // first-token (prefill) budget to the shorter mid-stream idle budget, - // so a stall *after* output started is still reclaimed quickly (#826). - if (currentStreamIdleTimeoutMs !== STREAM_IDLE_TIMEOUT_MS) { - currentStreamIdleTimeoutMs = STREAM_IDLE_TIMEOUT_MS; - resetStreamIdleTimer(); - } + } + + // Content is flowing - switch the watchdog from the generous first-token + // (prefill) budget to the shorter mid-stream idle budget. Non-content + // events such as message_start must not end the first-token phase: some + // gateways open SSE and emit bookkeeping long before useful content. + if ( + receivedFirstContentDelta && + currentStreamIdleTimeoutMs !== STREAM_IDLE_TIMEOUT_MS + ) { + currentStreamIdleTimeoutMs = STREAM_IDLE_TIMEOUT_MS; + resetStreamIdleTimer(); } switch (part.type) { @@ -2504,11 +2548,13 @@ async function* queryModel( // Prevent double-emit: this throw lands in the catch block below, // whose exit_path='error' probe guards on streamWatchdogFiredAt. streamWatchdogFiredAt = null; - throw new Error( - streamAbortReason === "max_duration" - ? "Stream max duration exceeded - no completion received" - : "Stream idle timeout - no chunks received", - ); + throw streamWatchdogTimeoutError ?? + streamWatchdogState.createTimeoutError( + streamAbortReason ?? "idle", + streamAbortReason === "max_duration" + ? STREAM_MAX_DURATION_MS + : currentStreamIdleTimeoutMs, + ); } // Detect when the stream completed without producing any assistant messages. @@ -2660,9 +2706,26 @@ async function* queryModel( // (newMessages.length === 0): a zero-output stream means no tool_use block // ever completed, so query.ts never started a tool — no double-execution // risk (cf. #766 / inc-4258), the same precondition the zero-output - // fallback below relies on. Watchdog aborts are excluded (they have their - // own fallback/timeout handling). Thrown past the outer catch — which + // fallback below relies on. Watchdog aborts without content/tool output + // are handled first; remaining watchdog aborts keep their partial-output + // boundary and are not retried. Thrown past the outer catch — which // re-throws it — up to withStreamRetry. + if ( + streamIdleAborted && + streamingError instanceof StreamWatchdogTimeoutError && + streamingError.safeToRetryStream() && + newMessages.length === 0 && + !signal.aborted + ) { + logForDebugging( + `Watchdog timeout before content/tool output, will retry stream: ${errorMessage( + streamingError, + )}`, + { level: "warn" }, + ); + throw new RetriableStreamError(streamingError); + } + if ( newMessages.length === 0 && !streamIdleAborted && diff --git a/src/services/api/errors.ts b/src/services/api/errors.ts index bb482455..ca175317 100644 --- a/src/services/api/errors.ts +++ b/src/services/api/errors.ts @@ -51,6 +51,7 @@ import { } from '../claudeAiLimits.js' import { shouldProcessRateLimits } from '../rateLimitMocking.js' // Used for /mock-limits command import { extractConnectionErrorDetails, formatAPIError } from './errorUtils.js' +import { StreamWatchdogTimeoutError } from './streamWatchdog.js' export const API_ERROR_MESSAGE_PREFIX = 'API Error' @@ -987,6 +988,14 @@ export function getAssistantMessageFromError( }) } + if (error instanceof StreamWatchdogTimeoutError) { + return createAssistantAPIErrorMessage({ + content: `${API_ERROR_MESSAGE_PREFIX}: ${error.message}`, + error: 'server_error', + errorDetails: JSON.stringify(error.toDiagnosticData()), + }) + } + // Connection errors (non-timeout) — use formatAPIError for detailed messages if (error instanceof APIConnectionError) { return createAssistantAPIErrorMessage({ diff --git a/src/services/api/streamWatchdog.test.ts b/src/services/api/streamWatchdog.test.ts new file mode 100644 index 00000000..15e20bbb --- /dev/null +++ b/src/services/api/streamWatchdog.test.ts @@ -0,0 +1,118 @@ +import { describe, expect, test } from 'bun:test' +import { + StreamWatchdogTimeoutError, + createStreamWatchdogState, +} from './streamWatchdog.js' + +describe('stream watchdog state', () => { + test('keeps the first-token budget until content deltas arrive', () => { + const state = createStreamWatchdogState() + + expect(state.recordEvent({ type: 'message_start' })).toBe(false) + expect(state.recordEvent({ + type: 'content_block_start', + index: 0, + content_block: { type: 'thinking' }, + })).toBe(false) + + expect(state.hasContentDelta()).toBe(false) + expect(state.snapshot().phase).toBe('before_content') + + expect(state.recordEvent({ + type: 'content_block_delta', + index: 0, + delta: { type: 'thinking_delta', thinking: 'working' }, + })).toBe(true) + + expect(state.hasContentDelta()).toBe(true) + expect(state.snapshot()).toMatchObject({ + phase: 'mid_stream', + eventCount: 3, + contentDeltaCount: 1, + thinkingDeltaCount: 1, + lastEventType: 'content_block_delta', + lastDeltaType: 'thinking_delta', + lastBlockType: 'thinking', + }) + }) + + test('describes a stream that started but never produced content', () => { + const state = createStreamWatchdogState() + state.recordEvent({ type: 'message_start' }) + state.recordEvent({ + type: 'content_block_start', + index: 0, + content_block: { type: 'thinking' }, + }) + + const error = state.createTimeoutError('idle', 600_000) + + expect(error).toBeInstanceOf(StreamWatchdogTimeoutError) + expect(error.code).toBe('STREAM_IDLE_TIMEOUT') + expect(error.phase).toBe('before_content') + expect(error.safeToRetryStream()).toBe(true) + expect(error.message).toContain('stream started but no content was received') + expect(error.message).toContain('last event: content_block_start') + }) + + test('describes a provider stream that stalls after partial output', () => { + const state = createStreamWatchdogState() + state.recordEvent({ type: 'message_start' }) + state.recordEvent({ + type: 'content_block_start', + index: 0, + content_block: { type: 'text' }, + }) + state.recordEvent({ + type: 'content_block_delta', + index: 0, + delta: { type: 'text_delta', text: 'writing HTML report, partial' }, + }) + + const error = state.createTimeoutError('idle', 240_000) + + expect(error.code).toBe('STREAM_IDLE_TIMEOUT') + expect(error.phase).toBe('mid_stream') + expect(error.safeToRetryStream()).toBe(false) + expect(error.message).toContain('Provider stream stalled after partial response') + expect(error.message).toContain('last event: text_delta') + expect(error.message).not.toContain('no chunks received') + }) + + test('does not retry after a tool use has started', () => { + const state = createStreamWatchdogState() + state.recordEvent({ type: 'message_start' }) + state.recordEvent({ + type: 'content_block_start', + index: 0, + content_block: { type: 'tool_use' }, + }) + + const error = state.createTimeoutError('idle', 240_000) + + expect(error.phase).toBe('before_content') + expect(error.safeToRetryStream()).toBe(false) + }) + + test('classifies max-duration aborts separately from idle aborts', () => { + const state = createStreamWatchdogState() + state.recordEvent({ type: 'message_start' }) + state.recordEvent({ + type: 'content_block_start', + index: 0, + content_block: { type: 'text' }, + }) + state.recordEvent({ + type: 'content_block_delta', + index: 0, + delta: { type: 'text_delta', text: 'partial' }, + }) + + const error = state.createTimeoutError('max_duration', 600_000) + + expect(error.code).toBe('STREAM_MAX_DURATION') + expect(error.phase).toBe('mid_stream') + expect(error.message).toContain('Stream max duration exceeded') + expect(error.message).toContain('last event: text_delta') + }) +}) diff --git a/src/services/api/streamWatchdog.ts b/src/services/api/streamWatchdog.ts new file mode 100644 index 00000000..43925086 --- /dev/null +++ b/src/services/api/streamWatchdog.ts @@ -0,0 +1,270 @@ +export type StreamWatchdogAbortReason = 'idle' | 'max_duration' +export type StreamWatchdogErrorCode = 'STREAM_IDLE_TIMEOUT' | 'STREAM_MAX_DURATION' +export type StreamWatchdogPhase = + | 'before_first_event' + | 'before_content' + | 'mid_stream' + +export type StreamWatchdogSnapshot = { + phase: StreamWatchdogPhase + eventCount: number + contentDeltaCount: number + textDeltaCount: number + thinkingDeltaCount: number + toolInputDeltaCount: number + lastEventType?: string + lastDeltaType?: string + lastBlockType?: string + messageStopReceived: boolean + toolUseStarted: boolean +} + +type StreamEventLike = { + type?: unknown + index?: unknown + content_block?: unknown + delta?: unknown +} + +function asObject(value: unknown): Record | null { + return value && typeof value === 'object' + ? value as Record + : null +} + +function readString(value: unknown): string | undefined { + return typeof value === 'string' ? value : undefined +} + +function readIndex(value: unknown): number | undefined { + return typeof value === 'number' && Number.isInteger(value) ? value : undefined +} + +function hasNonEmptyStringField( + value: Record, + field: string, +): boolean { + const text = value[field] + return typeof text === 'string' && text.length > 0 +} + +function isContentDelta(delta: Record): boolean { + switch (delta.type) { + case 'text_delta': + return hasNonEmptyStringField(delta, 'text') + case 'thinking_delta': + return hasNonEmptyStringField(delta, 'thinking') + case 'input_json_delta': + return hasNonEmptyStringField(delta, 'partial_json') + case 'connector_text_delta': + return hasNonEmptyStringField(delta, 'connector_text') + default: + return false + } +} + +function countContentDelta( + snapshot: StreamWatchdogSnapshot, + deltaType: string | undefined, +): StreamWatchdogSnapshot { + switch (deltaType) { + case 'text_delta': + case 'connector_text_delta': + return { + ...snapshot, + contentDeltaCount: snapshot.contentDeltaCount + 1, + textDeltaCount: snapshot.textDeltaCount + 1, + } + case 'thinking_delta': + return { + ...snapshot, + contentDeltaCount: snapshot.contentDeltaCount + 1, + thinkingDeltaCount: snapshot.thinkingDeltaCount + 1, + } + case 'input_json_delta': + return { + ...snapshot, + contentDeltaCount: snapshot.contentDeltaCount + 1, + toolInputDeltaCount: snapshot.toolInputDeltaCount + 1, + } + default: + return snapshot + } +} + +function formatSeconds(ms: number): string { + return `${Math.round(ms / 1000)}s` +} + +function formatDetails(snapshot: StreamWatchdogSnapshot): string { + const lastEvent = snapshot.lastDeltaType ?? snapshot.lastEventType ?? 'none' + const parts = [ + `last event: ${lastEvent}`, + `events: ${snapshot.eventCount}`, + `content deltas: ${snapshot.contentDeltaCount}`, + ] + if (snapshot.lastBlockType) { + parts.push(`last block: ${snapshot.lastBlockType}`) + } + return parts.join(', ') +} + +function buildMessage( + reason: StreamWatchdogAbortReason, + timeoutMs: number, + snapshot: StreamWatchdogSnapshot, +): string { + const seconds = formatSeconds(timeoutMs) + if (reason === 'max_duration') { + return `Stream max duration exceeded - no completion received after ${seconds} (${formatDetails(snapshot)})` + } + + switch (snapshot.phase) { + case 'before_first_event': + return `Stream idle timeout - no stream events received for ${seconds}` + case 'before_content': + return `Stream idle timeout - stream started but no content was received for ${seconds} (${formatDetails(snapshot)})` + case 'mid_stream': + return `Provider stream stalled after partial response - no new chunks for ${seconds} (${formatDetails(snapshot)})` + } +} + +export class StreamWatchdogTimeoutError extends Error { + readonly code: StreamWatchdogErrorCode + readonly phase: StreamWatchdogPhase + + constructor( + readonly reason: StreamWatchdogAbortReason, + readonly timeoutMs: number, + readonly streamSnapshot: StreamWatchdogSnapshot, + ) { + super(buildMessage(reason, timeoutMs, streamSnapshot)) + this.name = 'StreamWatchdogTimeoutError' + this.code = reason === 'max_duration' + ? 'STREAM_MAX_DURATION' + : 'STREAM_IDLE_TIMEOUT' + this.phase = streamSnapshot.phase + } + + safeToRetryStream(): boolean { + return ( + this.streamSnapshot.contentDeltaCount === 0 && + !this.streamSnapshot.toolUseStarted + ) + } + + toDiagnosticData(): Record { + return { + reason: this.reason, + code: this.code, + phase: this.phase, + timeoutMs: this.timeoutMs, + safeToRetry: this.safeToRetryStream(), + ...this.streamSnapshot, + } + } +} + +class StreamWatchdogState { + private snapshotValue: StreamWatchdogSnapshot = { + phase: 'before_first_event', + eventCount: 0, + contentDeltaCount: 0, + textDeltaCount: 0, + thinkingDeltaCount: 0, + toolInputDeltaCount: 0, + messageStopReceived: false, + toolUseStarted: false, + } + + private readonly blockTypesByIndex = new Map() + + recordEvent(event: unknown): boolean { + const beforeContentDeltaCount = this.snapshotValue.contentDeltaCount + const rawEvent = asObject(event) as StreamEventLike | null + const type = readString(rawEvent?.type) + if (!type) { + return false + } + + this.snapshotValue = { + ...this.snapshotValue, + eventCount: this.snapshotValue.eventCount + 1, + lastEventType: type, + } + + if (type === 'message_stop') { + this.snapshotValue = { + ...this.snapshotValue, + messageStopReceived: true, + } + } + + if (type === 'content_block_start') { + const block = asObject(rawEvent?.content_block) + const blockType = readString(block?.type) + const index = readIndex(rawEvent?.index) + if (blockType) { + this.snapshotValue = { + ...this.snapshotValue, + lastBlockType: blockType, + toolUseStarted: + this.snapshotValue.toolUseStarted || + blockType === 'tool_use' || + blockType === 'server_tool_use', + } + if (index !== undefined) { + this.blockTypesByIndex.set(index, blockType) + } + } + } + + if (type === 'content_block_delta') { + const delta = asObject(rawEvent?.delta) + const deltaType = readString(delta?.type) + const index = readIndex(rawEvent?.index) + const blockType = index === undefined + ? undefined + : this.blockTypesByIndex.get(index) + this.snapshotValue = { + ...this.snapshotValue, + ...(deltaType ? { lastDeltaType: deltaType } : {}), + ...(blockType ? { lastBlockType: blockType } : {}), + } + if (delta && isContentDelta(delta)) { + this.snapshotValue = countContentDelta(this.snapshotValue, deltaType) + } + } + + this.snapshotValue = { + ...this.snapshotValue, + phase: this.snapshotValue.contentDeltaCount > 0 + ? 'mid_stream' + : 'before_content', + } + + return ( + beforeContentDeltaCount === 0 && + this.snapshotValue.contentDeltaCount > 0 + ) + } + + hasContentDelta(): boolean { + return this.snapshotValue.contentDeltaCount > 0 + } + + snapshot(): StreamWatchdogSnapshot { + return { ...this.snapshotValue } + } + + createTimeoutError( + reason: StreamWatchdogAbortReason, + timeoutMs: number, + ): StreamWatchdogTimeoutError { + return new StreamWatchdogTimeoutError(reason, timeoutMs, this.snapshot()) + } +} + +export function createStreamWatchdogState(): StreamWatchdogState { + return new StreamWatchdogState() +}