fix(ai-anthropic): keep the input usage when a stream fails after message_start

.changeset/anthropic-stream-failure-usage.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/ai-anthropic': patch
+---
+
+Report the input token usage when a streamed response fails after its first event. A dropped connection or an Anthropic `error` event in the stream ended in a `RUN_ERROR` with no `usage`. Anthropic bills the input tokens of that call, so this `RUN_ERROR` now carries the `usage` that `message_start` reported. The output count is the one from `message_start`, so it can be lower than the real output count.

docs/adapters/anthropic-advanced.md

@@ -180,7 +180,7 @@ The Messages API of Anthropic requires `max_tokens` on every request, so the ada
 - Set `max_tokens` only to cap the output below the model limit.
 - If a response stops at the default limit, the adapter logs a warning. You see it with [debug logging](../advanced/debug-logging) on.
 
-A streamed response that stops at `max_tokens` ends in a `RUN_ERROR` with `code: 'max_tokens'`. Anthropic bills the tokens of that call, so this `RUN_ERROR` carries the `usage` of the call. The `onUsage` middleware hook fires only for `RUN_FINISHED`. To count these tokens too, read `usage` from the `RUN_ERROR` chunk, in the stream or in an `onChunk` middleware.
+A streamed response that stops at `max_tokens` ends in a `RUN_ERROR` with `code: 'max_tokens'`. Anthropic bills the tokens of that call, so this `RUN_ERROR` carries the `usage` of the call. A stream can also fail after it starts, for example when the connection drops. That `RUN_ERROR` carries the `usage` from Anthropic's first event (`message_start`). The input counts are complete. The output count can be lower than the real one. The `onUsage` middleware hook fires only for `RUN_FINISHED`. To count these tokens too, read `usage` from the `RUN_ERROR` chunk, in the stream or in an `onChunk` middleware.
 
 One exception: structured output (`chat({ outputSchema })`) on models that use the non-streaming finalization path caps this default at about 21K tokens. The Anthropic SDK rejects a non-streaming request whose `max_tokens` can go past its 10-minute timeout. Streaming chat does not have this cap. To raise the structured output limit toward the true maximum of a model, stream the response.
 

docs/config.json

@@ -1691,7 +1691,8 @@
         {
           "label": "Anthropic: Advanced",
           "to": "adapters/anthropic-advanced",
-          "addedAt": "2026-10-21"
+          "addedAt": "2026-10-21",
+          "updatedAt": "2026-10-22"
         },
         {
           "label": "Google Gemini",

packages/ai-anthropic/src/adapters/text.ts

@@ -1719,7 +1719,7 @@ export class AnthropicTextAdapter<
     // Track current content block type for proper content_block_stop handling
     let currentBlockType: string | null = null
     // Input and cache counts from message_start, for a closing message_delta
-    // that leaves them out.
+    // that leaves them out and for a stream that fails before it arrives.
     let messageStartUsage: Anthropic_SDK.Beta.BetaUsage | undefined
 
     try {
@@ -2343,6 +2343,10 @@ export class AnthropicTextAdapter<
       const code = runErrorCode(err)
       const rawEvent = toRunErrorRawEvent(error)
       const retryAfterMs = toRetryAfterMs(error)
+      // The closing message_delta already reported usage on its terminal chunk.
+      const usage = hasEmittedRunFinished
+        ? undefined
+        : buildAnthropicUsage(messageStartUsage)
 
       logger.errors('anthropic.processAnthropicStream fatal', {
         error,
@@ -2358,6 +2362,7 @@ export class AnthropicTextAdapter<
         // Forward the Anthropic SDK error's `.error` response body when present.
         ...(rawEvent !== undefined && { rawEvent }),
         ...(retryAfterMs !== undefined && { retryAfterMs }),
+        ...(usage && { usage }),
         ...toRunErrorDetails(error),
         error: {
           message: err.message || 'Unknown error occurred',

packages/ai-anthropic/tests/stream-failure-usage-wire.test.ts

@@ -0,0 +1,69 @@
+import { describe, expect, it } from 'vitest'
+import { resolveDebugOption } from '@tanstack/ai/adapter-internals'
+import { createAnthropicChat } from '../src/adapters/text'
+import type { StreamChunk } from '@tanstack/ai'
+
+const model = 'claude-haiku-4-5'
+
+const sse = (events: Array<{ type: string }>) =>
+  new Response(
+    events
+      .map(
+        (event) => `event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`,
+      )
+      .join(''),
+    { headers: { 'content-type': 'text/event-stream' } },
+  )
+
+describe('Anthropic adapter on a stream that fails mid-way', () => {
+  it('reports the message_start usage on the RUN_ERROR of an SSE error event (#1735)', async () => {
+    const adapter = createAnthropicChat(model, 'test-key', {
+      maxRetries: 0,
+      fetch: async () =>
+        sse([
+          {
+            type: 'message_start',
+            message: {
+              id: 'msg_1',
+              type: 'message',
+              role: 'assistant',
+              model,
+              content: [],
+              stop_reason: null,
+              stop_sequence: null,
+              usage: { input_tokens: 18, output_tokens: 1 },
+            },
+          } as { type: string },
+          {
+            type: 'content_block_start',
+            index: 0,
+            content_block: { type: 'text', text: '' },
+          } as { type: string },
+          {
+            type: 'content_block_delta',
+            index: 0,
+            delta: { type: 'text_delta', text: 'One, two' },
+          } as { type: string },
+          {
+            type: 'error',
+            error: { type: 'overloaded_error', message: 'Overloaded' },
+          } as { type: string },
+        ]),
+    })
+    const chunks: Array<StreamChunk> = []
+    for await (const chunk of adapter.chatStream({
+      logger: resolveDebugOption(false),
+      model,
+      threadId: 'thread-1',
+      messages: [{ role: 'user', content: 'Count to thirty' }],
+    })) {
+      chunks.push(chunk)
+    }
+
+    const error = chunks.find((chunk) => chunk.type === 'RUN_ERROR')
+    expect(error?.message).toContain('Overloaded')
+    expect(error).toMatchObject({
+      usage: { promptTokens: 18, completionTokens: 1, totalTokens: 19 },
+    })
+  })
+})

packages/ai-anthropic/tests/usage-extraction.test.ts

@@ -64,6 +64,21 @@ function createMockStream(
   }
 }
 
+function createFailingStream(
+  chunks: Array<Record<string, unknown>>,
+  error: Error,
+): AsyncIterable<Record<string, unknown>> {
+  return {
+    // eslint-disable-next-line @typescript-eslint/require-await
+    async *[Symbol.asyncIterator]() {
+      for (const chunk of chunks) {
+        yield chunk
+      }
+      throw error
+    },
+  }
+}
+
 describe('Anthropic usage extraction', () => {
   beforeEach(() => {
     vi.clearAllMocks()
@@ -685,4 +700,84 @@ describe('Anthropic usage extraction', () => {
     expect(doneChunk).toBeDefined()
     expect(tokenUsageOf(doneChunk)).toBeUndefined()
   })
+  describe('when the stream fails (#1735)', () => {
+    const messageStart = {
+      type: 'message_start',
+      message: {
+        id: 'msg_123',
+        type: 'message',
+        role: 'assistant',
+        content: [],
+        model: 'claude-opus-4-1',
+        usage: {
+          input_tokens: 18,
+          output_tokens: 1,
+          cache_creation_input_tokens: 7,
+          cache_read_input_tokens: 40,
+        },
+      },
+    }
+    const textDelta = [
+      {
+        type: 'content_block_start',
+        index: 0,
+        content_block: { type: 'text', text: '' },
+      },
+      {
+        type: 'content_block_delta',
+        index: 0,
+        delta: { type: 'text_delta', text: 'One, two' },
+      },
+    ]
+    const closingDelta = {
+      type: 'message_delta',
+      delta: { stop_reason: 'end_turn' },
+      usage: { output_tokens: 5 },
+    }
+
+    it.each([
+      [
+        'after message_start',
+        [messageStart, ...textDelta],
+        {
+          promptTokens: 65,
+          completionTokens: 1,
+          totalTokens: 66,
+          promptTokensDetails: { cacheWriteTokens: 7, cachedTokens: 40 },
+        },
+      ],
+      ['before message_start', [], undefined],
+      // The RUN_FINISHED already carried the usage, so the RUN_ERROR must not
+      // report it a second time.
+      [
+        'after the closing message_delta',
+        [messageStart, ...textDelta, closingDelta],
+        undefined,
+      ],
+    ])(
+      'reports the expected usage on the RUN_ERROR %s',
+      async (_, events, expectedUsage) => {
+        mocks.betaMessagesCreate.mockResolvedValueOnce(
+          createFailingStream(
+            events,
+            Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' }),
+          ),
+        )
+
+        const chunks: Array<AdapterYieldChunk> = []
+        for await (const chunk of createAdapter().chatStream({
+          model: 'claude-opus-4-1',
+          threadId: 'thread-1',
+          messages: [{ role: 'user', content: 'Count to thirty' }],
+          logger: createSilentLogger(),
+        })) {
+          chunks.push(chunk)
+        }
+
+        const errorChunk = chunks.find((c) => c.type === 'RUN_ERROR')
+        expect(errorChunk).toMatchObject({ code: 'ECONNRESET' })
+        expect(tokenUsageOf(errorChunk)).toEqual(expectedUsage)
+      },
+    )
+  })
 })

testing/e2e/fixtures/stream-failure-usage/basic.json

@@ -0,0 +1,14 @@
+{
+  "fixtures": [
+    {
+      "match": { "userMessage": "[stream-failure-usage] count to thirty" },
+      "response": {
+        "content": "One, two, three, four, five, six, seven, eight, nine, ten",
+        "usage": { "input_tokens": 18, "output_tokens": 1 }
+      },
+      "chunkSize": 4,
+      "latency": 20,
+      "truncateAfterChunks": 4
+    }
+  ]
+}

testing/e2e/src/routeTree.gen.ts

@@ -64,6 +64,7 @@ import { Route as ApiAnthropicRedactedThinkingWireRouteImport } from './routes/a
 import { Route as ApiAnthropicRefusalRouteImport } from './routes/api.anthropic-refusal'
 import { Route as ApiAnthropicSkillsWireRouteImport } from './routes/api.anthropic-skills-wire'
 import { Route as ApiAnthropicSonnet55WireRouteImport } from './routes/api.anthropic-sonnet-5-5-wire'
+import { Route as ApiAnthropicStreamFailureUsageRouteImport } from './routes/api.anthropic-stream-failure-usage'
 import { Route as ApiAnthropicThinkingOrderWireRouteImport } from './routes/api.anthropic-thinking-order-wire'
 import { Route as ApiAnthropicTruncatedToolInputWireRouteImport } from './routes/api.anthropic-truncated-tool-input-wire'
 import { Route as ApiArktypeToolWireRouteImport } from './routes/api.arktype-tool-wire'
@@ -491,6 +492,12 @@ const ApiAnthropicSonnet55WireRoute =
     path: '/api/anthropic-sonnet-5-5-wire',
     getParentRoute: () => rootRouteImport,
   } as any)
+const ApiAnthropicStreamFailureUsageRoute =
+  ApiAnthropicStreamFailureUsageRouteImport.update({
+    id: '/api/anthropic-stream-failure-usage',
+    path: '/api/anthropic-stream-failure-usage',
+    getParentRoute: () => rootRouteImport,
+  } as any)
 const ApiAnthropicThinkingOrderWireRoute =
   ApiAnthropicThinkingOrderWireRouteImport.update({
     id: '/api/anthropic-thinking-order-wire',
@@ -1283,6 +1290,7 @@ export interface FileRoutesByFullPath {
   '/api/anthropic-refusal': typeof ApiAnthropicRefusalRoute
   '/api/anthropic-skills-wire': typeof ApiAnthropicSkillsWireRoute
   '/api/anthropic-sonnet-5-5-wire': typeof ApiAnthropicSonnet55WireRoute
+  '/api/anthropic-stream-failure-usage': typeof ApiAnthropicStreamFailureUsageRoute
   '/api/anthropic-thinking-order-wire': typeof ApiAnthropicThinkingOrderWireRoute
   '/api/anthropic-truncated-tool-input-wire': typeof ApiAnthropicTruncatedToolInputWireRoute
   '/api/arktype-tool-wire': typeof ApiArktypeToolWireRoute
@@ -1480,6 +1488,7 @@ export interface FileRoutesByTo {
   '/api/anthropic-refusal': typeof ApiAnthropicRefusalRoute
   '/api/anthropic-skills-wire': typeof ApiAnthropicSkillsWireRoute
   '/api/anthropic-sonnet-5-5-wire': typeof ApiAnthropicSonnet55WireRoute
+  '/api/anthropic-stream-failure-usage': typeof ApiAnthropicStreamFailureUsageRoute
   '/api/anthropic-thinking-order-wire': typeof ApiAnthropicThinkingOrderWireRoute
   '/api/anthropic-truncated-tool-input-wire': typeof ApiAnthropicTruncatedToolInputWireRoute
   '/api/arktype-tool-wire': typeof ApiArktypeToolWireRoute
@@ -1678,6 +1687,7 @@ export interface FileRoutesById {
   '/api/anthropic-refusal': typeof ApiAnthropicRefusalRoute
   '/api/anthropic-skills-wire': typeof ApiAnthropicSkillsWireRoute
   '/api/anthropic-sonnet-5-5-wire': typeof ApiAnthropicSonnet55WireRoute
+  '/api/anthropic-stream-failure-usage': typeof ApiAnthropicStreamFailureUsageRoute
   '/api/anthropic-thinking-order-wire': typeof ApiAnthropicThinkingOrderWireRoute
   '/api/anthropic-truncated-tool-input-wire': typeof ApiAnthropicTruncatedToolInputWireRoute
   '/api/arktype-tool-wire': typeof ApiArktypeToolWireRoute
@@ -1877,6 +1887,7 @@ export interface FileRouteTypes {
     | '/api/anthropic-refusal'
     | '/api/anthropic-skills-wire'
     | '/api/anthropic-sonnet-5-5-wire'
+    | '/api/anthropic-stream-failure-usage'
     | '/api/anthropic-thinking-order-wire'
     | '/api/anthropic-truncated-tool-input-wire'
     | '/api/arktype-tool-wire'
@@ -2074,6 +2085,7 @@ export interface FileRouteTypes {
     | '/api/anthropic-refusal'
     | '/api/anthropic-skills-wire'
     | '/api/anthropic-sonnet-5-5-wire'
+    | '/api/anthropic-stream-failure-usage'
     | '/api/anthropic-thinking-order-wire'
     | '/api/anthropic-truncated-tool-input-wire'
     | '/api/arktype-tool-wire'
@@ -2271,6 +2283,7 @@ export interface FileRouteTypes {
     | '/api/anthropic-refusal'
     | '/api/anthropic-skills-wire'
     | '/api/anthropic-sonnet-5-5-wire'
+    | '/api/anthropic-stream-failure-usage'
     | '/api/anthropic-thinking-order-wire'
     | '/api/anthropic-truncated-tool-input-wire'
     | '/api/arktype-tool-wire'
@@ -2469,6 +2482,7 @@ export interface RootRouteChildren {
   ApiAnthropicRefusalRoute: typeof ApiAnthropicRefusalRoute
   ApiAnthropicSkillsWireRoute: typeof ApiAnthropicSkillsWireRoute
   ApiAnthropicSonnet55WireRoute: typeof ApiAnthropicSonnet55WireRoute
+  ApiAnthropicStreamFailureUsageRoute: typeof ApiAnthropicStreamFailureUsageRoute
   ApiAnthropicThinkingOrderWireRoute: typeof ApiAnthropicThinkingOrderWireRoute
   ApiAnthropicTruncatedToolInputWireRoute: typeof ApiAnthropicTruncatedToolInputWireRoute
   ApiArktypeToolWireRoute: typeof ApiArktypeToolWireRoute
@@ -2994,6 +3008,13 @@ declare module '@tanstack/react-router' {
       preLoaderRoute: typeof ApiAnthropicSonnet55WireRouteImport
       parentRoute: typeof rootRouteImport
     }
+    '/api/anthropic-stream-failure-usage': {
+      id: '/api/anthropic-stream-failure-usage'
+      path: '/api/anthropic-stream-failure-usage'
+      fullPath: '/api/anthropic-stream-failure-usage'
+      preLoaderRoute: typeof ApiAnthropicStreamFailureUsageRouteImport
+      parentRoute: typeof rootRouteImport
+    }
     '/api/anthropic-thinking-order-wire': {
       id: '/api/anthropic-thinking-order-wire'
       path: '/api/anthropic-thinking-order-wire'
@@ -4091,6 +4112,7 @@ const rootRouteChildren: RootRouteChildren = {
   ApiAnthropicRefusalRoute: ApiAnthropicRefusalRoute,
   ApiAnthropicSkillsWireRoute: ApiAnthropicSkillsWireRoute,
   ApiAnthropicSonnet55WireRoute: ApiAnthropicSonnet55WireRoute,
+  ApiAnthropicStreamFailureUsageRoute: ApiAnthropicStreamFailureUsageRoute,
   ApiAnthropicThinkingOrderWireRoute: ApiAnthropicThinkingOrderWireRoute,
   ApiAnthropicTruncatedToolInputWireRoute:
     ApiAnthropicTruncatedToolInputWireRoute,

testing/e2e/src/routes/api.anthropic-stream-failure-usage.ts

@@ -0,0 +1,49 @@
+import { createFileRoute } from '@tanstack/react-router'
+import { chat } from '@tanstack/ai'
+import { createAnthropicChat } from '@tanstack/ai-anthropic'
+
+const LLMOCK_DEFAULT_BASE = process.env.LLMOCK_URL || 'http://127.0.0.1:4010'
+const DUMMY_KEY = 'sk-ant-e2e-test-dummy-key'
+
+/**
+ * Streams one Anthropic chat against the aimock fixture in
+ * `fixtures/stream-failure-usage`. aimock sends `message_start`, the start of
+ * a text block and one text delta, then drops the connection. Returns the
+ * `RUN_ERROR` that `chat()` yields (issue #1735).
+ */
+export const Route = createFileRoute('/api/anthropic-stream-failure-usage')({
+  server: {
+    handlers: {
+      POST: async () => {
+        const adapter = createAnthropicChat('claude-haiku-4-5', DUMMY_KEY, {
+          baseURL: LLMOCK_DEFAULT_BASE,
+        })
+
+        let runError: Record<string, unknown> | undefined
+        try {
+          for await (const chunk of chat({
+            adapter,
+            threadId: 'stream-failure-usage',
+            messages: [
+              {
+                role: 'user',
+                content: '[stream-failure-usage] count to thirty',
+              },
+            ],
+          })) {
+            if (chunk.type === 'RUN_ERROR') {
+              runError = { usage: chunk.usage }
+            }
+          }
+        } catch (error) {
+          return Response.json({
+            ok: false,
+            error: error instanceof Error ? error.message : String(error),
+          })
+        }
+
+        return Response.json({ ok: true, runError })
+      },
+    },
+  },
+})