fix(streaming): retry transient mid-stream errors instead of failing the turn

Mid-stream api_error/overloaded_error SSE events (e.g. a local provider such
as LM Studio rejecting a malformed tool_call) arrive inside the 200 SSE body,
bypassing withRetry — which only guards stream creation — so they were
surfaced as a terminal API error that ended the turn with no retry.

Add withStreamRetry around queryModel: when an attempt produced zero output
(no tool_use block completed, so query.ts never started a tool — safe w.r.t.
the double-tool-execution hazard, inc-4258), re-establish the stream and retry
up to CLAUDE_STREAM_TRANSIENT_RETRY_MAX times (default 2) before surfacing the
error. Covers both the streaming and non-streaming query wrappers; adds unit
tests for the retry loop and the transient-error classifier.

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

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<any, void>): Promise<any[]> {
+  // 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<any, void> {
+        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<any, void> {
+        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<any, void> {
+        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<any, void> {
+        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<any, void> {
+        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')
+  })
+})

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<StreamQueryMessage, void>,
+  model: string,
+  messages: Message[],
+): AsyncGenerator<StreamQueryMessage, void> {
+  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,
+      });
+    }
+  }
+}

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')
+  })
+})

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<T>(
   getClient: () => Promise<Anthropic>,
   operation: (