diff --git a/src/services/api/claude.ts b/src/services/api/claude.ts index a72d16b9..96456341 100644 --- a/src/services/api/claude.ts +++ b/src/services/api/claude.ts @@ -253,10 +253,13 @@ import { checkResponseForCacheBreak, recordPromptState, } from "./promptCacheBreakDetection.js"; +import { withStreamRetry } from "./streamRetry.js"; import { CannotRetryError, FallbackTriggeredError, is529Error, + isRetryableStreamError, + RetriableStreamError, type RetryContext, withRetry, } from "./withRetry.js"; @@ -730,13 +733,18 @@ export async function queryModelWithoutStreaming({ // logAPISuccessAndDuration gets called (which happens after all yields) let assistantMessage: AssistantMessage | undefined; for await (const message of withStreamingVCR(messages, async function* () { - yield* queryModel( + yield* withStreamRetry( + () => + queryModel( + messages, + systemPrompt, + thinkingConfig, + tools, + signal, + options, + ), + options.model, messages, - systemPrompt, - thinkingConfig, - tools, - signal, - options, ); })) { if (message.type === "assistant") { @@ -773,13 +781,18 @@ export async function* queryModelWithStreaming({ void > { return yield* withStreamingVCR(messages, async function* () { - yield* queryModel( + yield* withStreamRetry( + () => + queryModel( + messages, + systemPrompt, + thinkingConfig, + tools, + signal, + options, + ), + options.model, messages, - systemPrompt, - thinkingConfig, - tools, - signal, - options, ); }); } @@ -2602,6 +2615,31 @@ async function* queryModel( } } + // A transient, server-side error that arrived mid-stream (a local provider + // rejecting a malformed tool_call, or an upstream api_error / + // overloaded_error SSE event) is recoverable by re-establishing the + // stream. Only retry when this attempt produced NOTHING + // (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 + // re-throws it — up to withStreamRetry. + if ( + newMessages.length === 0 && + !streamIdleAborted && + !signal.aborted && + isRetryableStreamError(streamingError) + ) { + logForDebugging( + `Transient mid-stream error before any output, will retry stream: ${errorMessage( + streamingError, + )}`, + { level: "warn" }, + ); + throw new RetriableStreamError(streamingError); + } + // When the flag is enabled, skip the non-streaming fallback and let the // error propagate to withRetry. The mid-stream fallback causes double tool // execution when streaming tool execution is active: the partial stream @@ -2753,6 +2791,14 @@ async function* queryModel( throw errorFromRetry; } + // A transient mid-stream error flagged for stream-level retry: propagate up + // to withStreamRetry (the streaming wrapper), which re-establishes the + // stream. Must escape the terminal error handling below, which would + // otherwise yield an API-error message and end the turn. + if (errorFromRetry instanceof RetriableStreamError) { + throw errorFromRetry; + } + // Check if this is a 404 error during stream creation that should trigger // non-streaming fallback. This handles gateways that return 404 for streaming // endpoints but work fine with non-streaming. Before v2.1.8, BetaMessageStream diff --git a/src/services/api/streamRetry.test.ts b/src/services/api/streamRetry.test.ts new file mode 100644 index 00000000..99c8bb52 --- /dev/null +++ b/src/services/api/streamRetry.test.ts @@ -0,0 +1,139 @@ +import { afterAll, beforeAll, describe, expect, test } from 'bun:test' +import { APIError } from '@anthropic-ai/sdk' +import { withStreamRetry } from './streamRetry.js' +import { RetriableStreamError } from './withRetry.js' + +const RETRY_ENV = 'CLAUDE_STREAM_TRANSIENT_RETRY_MAX' + +// getAssistantMessageFromError() (invoked when retries are exhausted) consults +// isClaudeAISubscriber(), which throws if no auth is configured. We only assert +// that an assistant error message is produced, so a dummy key suffices. In +// production this path always runs with real auth already in place. +let priorApiKey: string | undefined +beforeAll(() => { + priorApiKey = process.env.ANTHROPIC_API_KEY + process.env.ANTHROPIC_API_KEY ??= 'sk-ant-test' +}) +afterAll(() => { + if (priorApiKey === undefined) { + delete process.env.ANTHROPIC_API_KEY + } +}) + +/** A RetriableStreamError wrapping a realistic mid-stream api_error (no status). */ +function retriableError(): RetriableStreamError { + const body = { + type: 'error', + error: { + type: 'api_error', + message: 'Failed to generate a valid tool call.', + }, + } + return new RetriableStreamError( + new APIError(undefined, body, JSON.stringify(body), undefined), + ) +} + +// biome-ignore lint/suspicious/noExplicitAny: test harness collects heterogeneous stream messages +async function collect(gen: AsyncGenerator): Promise { + // biome-ignore lint/suspicious/noExplicitAny: see above + const out: any[] = [] + for await (const m of gen) out.push(m) + return out +} + +describe('withStreamRetry', () => { + test('retries after a transient mid-stream error and yields the successful attempt', async () => { + process.env[RETRY_ENV] = '2' + let calls = 0 + const attempt = () => + // biome-ignore lint/suspicious/noExplicitAny: mock stream messages + (async function* (): AsyncGenerator { + calls++ + if (calls === 1) { + // A failed attempt may have already emitted partials before throwing. + yield { type: 'stream_event', event: { type: 'message_start' } } + throw retriableError() + } + yield { type: 'assistant', message: { content: [] }, uuid: 'ok' } + })() + + const out = await collect(withStreamRetry(attempt, 'test-model', [])) + + expect(calls).toBe(2) + const assistants = out.filter(m => m.type === 'assistant') + expect(assistants).toHaveLength(1) + expect(assistants[0].uuid).toBe('ok') + // The successful retry must NOT be reported as an API error. + expect(out.some(m => m.isApiErrorMessage)).toBe(false) + delete process.env[RETRY_ENV] + }) + + test('exhausts retries and surfaces an API-error assistant message', async () => { + process.env[RETRY_ENV] = '2' + let calls = 0 + const attempt = () => + // biome-ignore lint/suspicious/noExplicitAny: mock stream messages + (async function* (): AsyncGenerator { + calls++ + throw retriableError() + })() + + const out = await collect(withStreamRetry(attempt, 'test-model', [])) + + expect(calls).toBe(3) // 1 initial attempt + 2 retries + const last = out.at(-1) + expect(last?.type).toBe('assistant') + expect(last?.isApiErrorMessage).toBe(true) + delete process.env[RETRY_ENV] + }) + + test('does not retry a non-RetriableStreamError; rethrows it', async () => { + let calls = 0 + const attempt = () => + // biome-ignore lint/suspicious/noExplicitAny: mock stream messages + (async function* (): AsyncGenerator { + calls++ + throw new Error('fatal') + })() + + await expect( + collect(withStreamRetry(attempt, 'test-model', [])), + ).rejects.toThrow('fatal') + expect(calls).toBe(1) + }) + + test('maxRetries=0 makes a single attempt, then surfaces the error', async () => { + process.env[RETRY_ENV] = '0' + let calls = 0 + const attempt = () => + // biome-ignore lint/suspicious/noExplicitAny: mock stream messages + (async function* (): AsyncGenerator { + calls++ + throw retriableError() + })() + + const out = await collect(withStreamRetry(attempt, 'test-model', [])) + + expect(calls).toBe(1) + expect(out.at(-1)?.type).toBe('assistant') + expect(out.at(-1)?.isApiErrorMessage).toBe(true) + delete process.env[RETRY_ENV] + }) + + test('passes through a clean attempt without retrying', async () => { + let calls = 0 + const attempt = () => + // biome-ignore lint/suspicious/noExplicitAny: mock stream messages + (async function* (): AsyncGenerator { + calls++ + yield { type: 'assistant', message: { content: [] }, uuid: 'clean' } + })() + + const out = await collect(withStreamRetry(attempt, 'test-model', [])) + + expect(calls).toBe(1) + expect(out).toHaveLength(1) + expect(out[0].uuid).toBe('clean') + }) +}) diff --git a/src/services/api/streamRetry.ts b/src/services/api/streamRetry.ts new file mode 100644 index 00000000..84c4adfb --- /dev/null +++ b/src/services/api/streamRetry.ts @@ -0,0 +1,94 @@ +import type { + AssistantMessage, + Message, + StreamEvent, + SystemAPIErrorMessage, + SystemStreamingFallbackMessage, +} from "../../types/message.js"; +import { logForDebugging } from "../../utils/debug.js"; +import { errorMessage } from "../../utils/errors.js"; +import { + type AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, + logEvent, +} from "../analytics/index.js"; +import { getAssistantMessageFromError } from "./errors.js"; +import { + getMaxStreamTransientRetries, + RetriableStreamError, +} from "./withRetry.js"; + +/** Messages a streaming query attempt can emit. Mirrors queryModel's yield type. */ +type StreamQueryMessage = + | StreamEvent + | AssistantMessage + | SystemAPIErrorMessage + | SystemStreamingFallbackMessage; + +/** + * Wrap a single streaming query attempt with transient mid-stream retries. + * + * Mid-stream transient errors — a malformed tool_call a local provider rejects, + * or an upstream api_error/overloaded_error SSE event — arrive inside the 200 + * SSE body, so they never reach withRetry (which only guards stream creation). + * queryModel() detects them via isRetryableStreamError and throws + * RetriableStreamError; here we catch it and re-run the whole attempt by + * re-invoking the generator factory. Re-running queryModel() is a clean re-send: + * every per-request value is a fresh local, so there is nothing to reset by hand. + * + * Safe against the double-tool-execution hazard (#766 / inc-4258): queryModel + * only throws RetriableStreamError when the failed attempt produced zero + * assistant messages (no content_block_stop completed), so query.ts never handed + * a tool_use to the StreamingToolExecutor and no tool ran. + * + * A failed attempt may already have yielded raw stream_event partials; the retry + * re-emits message_start etc. queryModelWithoutStreaming ignores stream_event + * entirely, and the streaming UI resets its in-flight partial on the next + * message_start, so a re-emit at most causes a brief redraw. + */ +export async function* withStreamRetry( + attempt: () => AsyncGenerator, + model: string, + messages: Message[], +): AsyncGenerator { + const maxRetries = getMaxStreamTransientRetries(); + for (let i = 0; ; i++) { + try { + yield* attempt(); + return; + } catch (error) { + if (!(error instanceof RetriableStreamError)) { + throw error; + } + if (i >= maxRetries) { + // Retries exhausted — surface the original error as an assistant + // message, matching queryModel's normal terminal-error behavior. + logForDebugging( + `Transient mid-stream error: retries exhausted after ${maxRetries} attempt(s): ${errorMessage( + error.originalError, + )}`, + { level: "error" }, + ); + logEvent("tengu_stream_transient_retry_exhausted", { + attempts: maxRetries, + model: + model as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, + }); + yield getAssistantMessageFromError(error.originalError, model, { + messages, + }); + return; + } + logForDebugging( + `Transient mid-stream error, retrying (attempt ${i + 1}/${maxRetries}): ${errorMessage( + error.originalError, + )}`, + { level: "warn" }, + ); + logEvent("tengu_stream_transient_retry", { + attempt: i + 1, + model: + model as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, + }); + } + } +} diff --git a/src/services/api/withRetry.test.ts b/src/services/api/withRetry.test.ts index a4b12132..fab65b53 100644 --- a/src/services/api/withRetry.test.ts +++ b/src/services/api/withRetry.test.ts @@ -1,8 +1,13 @@ import { describe, expect, test } from 'bun:test' import type Anthropic from '@anthropic-ai/sdk' -import { APIConnectionError } from '@anthropic-ai/sdk' +import { APIConnectionError, APIError } from '@anthropic-ai/sdk' import { _resetKeepAliveForTesting, getProxyFetchOptions } from '../../utils/proxy.js' -import { withRetry } from './withRetry.js' +import { + getMaxStreamTransientRetries, + isRetryableStreamError, + RetriableStreamError, + withRetry, +} from './withRetry.js' describe('withRetry stale connections', () => { test('disables keep-alive before retrying ECONNRESET connection failures', async () => { @@ -47,3 +52,90 @@ describe('withRetry stale connections', () => { _resetKeepAliveForTesting() }) }) + +describe('isRetryableStreamError', () => { + // The SDK embeds the serialized error body in `error.message`; mirror that so + // the matcher sees the same shape it does in production. + function apiErrorWithBody(body: object, status?: number): APIError { + return new APIError(status, body, JSON.stringify(body), undefined) + } + + test('matches a mid-stream api_error with no HTTP status', () => { + const err = apiErrorWithBody({ + type: 'error', + error: { + type: 'api_error', + message: 'Failed to generate a valid tool call.', + }, + }) + expect(isRetryableStreamError(err)).toBe(true) + }) + + test('matches an overloaded_error', () => { + const err = apiErrorWithBody({ + type: 'error', + error: { type: 'overloaded_error', message: 'Overloaded' }, + }) + expect(isRetryableStreamError(err)).toBe(true) + }) + + test('does not match a client invalid_request_error', () => { + const err = apiErrorWithBody( + { + type: 'error', + error: { type: 'invalid_request_error', message: 'bad input' }, + }, + 400, + ) + expect(isRetryableStreamError(err)).toBe(false) + }) + + test('does not match a non-APIError', () => { + expect( + isRetryableStreamError(new Error('Failed to generate a valid tool call.')), + ).toBe(false) + }) + + test('does not match an APIError whose message lacks the markers', () => { + const err = new APIError( + 500, + { error: { type: 'internal', message: 'x' } }, + 'Internal Server Error', + undefined, + ) + expect(isRetryableStreamError(err)).toBe(false) + }) +}) + +describe('getMaxStreamTransientRetries', () => { + const ENV = 'CLAUDE_STREAM_TRANSIENT_RETRY_MAX' + + test('defaults to 2 when unset', () => { + delete process.env[ENV] + expect(getMaxStreamTransientRetries()).toBe(2) + }) + + test('honors a numeric override (including 0 to disable)', () => { + process.env[ENV] = '5' + expect(getMaxStreamTransientRetries()).toBe(5) + process.env[ENV] = '0' + expect(getMaxStreamTransientRetries()).toBe(0) + delete process.env[ENV] + }) + + test('falls back to 2 on non-numeric input', () => { + process.env[ENV] = 'abc' + expect(getMaxStreamTransientRetries()).toBe(2) + delete process.env[ENV] + }) +}) + +describe('RetriableStreamError', () => { + test('carries the original error and a faithful message', () => { + const original = new Error('boom') + const wrapped = new RetriableStreamError(original) + expect(wrapped.originalError).toBe(original) + expect(wrapped.name).toBe('RetriableStreamError') + expect(wrapped.message).toContain('boom') + }) +}) diff --git a/src/services/api/withRetry.ts b/src/services/api/withRetry.ts index f63b6f77..11494515 100644 --- a/src/services/api/withRetry.ts +++ b/src/services/api/withRetry.ts @@ -166,6 +166,63 @@ export class FallbackTriggeredError extends Error { } } +/** + * Raised inside the streaming path when a transient, server-side error arrives + * mid-stream (inside the 200 SSE body) and therefore bypasses withRetry — which + * only wraps stream *creation*, not stream *consumption*. Caught by + * withStreamRetry() in claude.ts, which re-establishes the stream and retries + * the whole request. Carries the original SDK error so the retries-exhausted + * path can surface a faithful API-error message. + */ +export class RetriableStreamError extends Error { + constructor(public readonly originalError: unknown) { + super(errorMessage(originalError)) + this.name = 'RetriableStreamError' + if (originalError instanceof Error && originalError.stack) { + this.stack = originalError.stack + } + } +} + +/** + * Detect transient, server-side errors that the SDK surfaces *during streaming* + * without a usable HTTP status. The upstream sends them inside a 200 SSE body as + * {"type":"error","error":{"type":"api_error"|"overloaded_error",...}} + * so `error.status` is undefined and every status-based check in shouldRetry() + * falls through to `return false`. Per Anthropic semantics both are retryable: + * `api_error` = "unexpected error internal to the server", `overloaded_error` = + * capacity. Local/proxy providers (e.g. LM Studio) also wrap transient + * generation failures — such as a malformed tool_call the runtime rejects + * mid-stream — as `api_error`, which a fresh attempt almost always clears. + * + * Matches the serialized error body embedded in `error.message`, the same + * technique the overloaded-error check has always used (see shouldRetry). + */ +export function isRetryableStreamError(error: unknown): boolean { + if (!(error instanceof APIError)) { + return false + } + const message = error.message + if (!message) { + return false + } + return ( + message.includes('"type":"api_error"') || + message.includes('"type":"overloaded_error"') + ) +} + +/** + * Max times withStreamRetry() re-establishes a stream after a transient + * mid-stream error (see RetriableStreamError). Small by default — a malformed + * tool_call or a one-off blip usually clears on the first retry; this is not a + * capacity backoff loop. Override with CLAUDE_STREAM_TRANSIENT_RETRY_MAX. + */ +export function getMaxStreamTransientRetries(): number { + const raw = parseInt(process.env.CLAUDE_STREAM_TRANSIENT_RETRY_MAX || '', 10) + return Number.isFinite(raw) && raw >= 0 ? raw : 2 +} + export async function* withRetry( getClient: () => Promise, operation: (