@tanstack/ai-anthropic: a stream that fails after message_start loses the call's input usage

Package / version: @tanstack/ai-anthropic 0.21.1, @tanstack/ai 0.68.0, @anthropic-ai/sdk 0.97.1, Node 24.20.0. The same code is on main.

Summary

Anthropic reports a call's input side (input_tokens, cache_creation_input_tokens, cache_read_input_tokens) in the stream's first event, message_start. processAnthropicStream stores it in messageStartUsage and only uses it when the closing message_delta arrives (packages/ai-anthropic/src/adapters/text.ts, around lines 1569 and 2025 on main). If the stream fails between those two events, a connection reset, a mid-stream error event, an overloaded_error, the catch block yields RUN_ERROR with no usage, and the input tokens, which Anthropic has already counted and bills, are lost to the consumer. No chunk before RUN_FINISHED carries usage either, so there is nothing earlier to read it from.

This is the sibling of #1597, which fixed the max_tokens stop: that case has a closing message_delta to build from. The failure case has only message_start.

Reproduction

A fetch whose response body drops after the second content_block_delta. Everything up to the cut is Anthropic's real stream.

import { chat } from '@tanstack/ai'
import { createAnthropicChatWithClient } from '@tanstack/ai-anthropic'
import Anthropic from '@anthropic-ai/sdk'

const MODEL = 'claude-haiku-4-5'
const messages = [{ role: 'user', content: 'Count from one to thirty, one number per line.' }]

const cuttingFetch = async (url, init) => {
  const res = await fetch(url, init)
  let deltas = 0
  const decoder = new TextDecoder()
  const body = res.body.pipeThrough(
    new TransformStream({
      transform(chunk, controller) {
        deltas += (decoder.decode(chunk, { stream: true }).match(/event: content_block_delta/g) ?? []).length
        if (deltas > 2) controller.error(Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' }))
        else controller.enqueue(chunk)
      },
    }),
  )
  return new Response(body, { status: res.status, headers: res.headers })
}

const client = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY, fetch: cuttingFetch, maxRetries: 0 })
const adapter = createAnthropicChatWithClient(MODEL, client)
const seen = []
for await (const c of chat({ adapter, messages })) seen.push(c)
console.log('chunk types seen:', seen.map((c) => c.type).join(' '))
console.log('chunks carrying usage:', seen.filter((c) => c.usage !== undefined).map((c) => c.type).join(' ') || '(none)')
console.log('last chunk =', JSON.stringify(seen.at(-1)))

// What Anthropic sent in its first event for the same request.
const plain = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY })
const stream = await plain.messages.create({ model: MODEL, max_tokens: 64, messages, stream: true })
for await (const e of stream) {
  if (e.type === 'message_start') { console.log('raw message_start usage =', JSON.stringify(e.message.usage)); break }
}

Observed

chunk types seen: RUN_STARTED TEXT_MESSAGE_START TEXT_MESSAGE_CONTENT RUN_ERROR
chunks carrying usage: (none)
last chunk = {"type":"RUN_ERROR","timestamp":1791850527495,"message":"read ECONNRESET","code":"ECONNRESET","metadata":{"tanstack":{"model":"claude-haiku-4-5"}}}
raw message_start usage = {"input_tokens":18,"cache_creation_input_tokens":0,"cache_read_input_tokens":0,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":0},"output_tokens":1,"service_tier":"standard","inference_geo":"not_available"}

The adapter held input_tokens: 18 from the first event and discarded it. With a cached system prompt the discarded number is the cache read or write count for the whole prompt, which is most of the call's price.

Expected

The input side reaches the consumer when the stream fails after message_start. Two ways that could look, not exclusive:

1. Usage on the error. In the catch of processAnthropicStream, when messageStartUsage is set, attach usage: buildAnthropicUsage(messageStartUsage) (output count zero or whatever message_start carried) to the RUN_ERROR, the way #1597 put usage on the max_tokens error. Smallest change, and it covers every mid-stream failure the adapter sees.

2. Opening usage as an event. Yield the message_start usage when it arrives, so a consumer can record the input side before the run ends. That also covers the case the adapter cannot see: the process itself dying mid-stream. This is what I'm carrying as a local patch to the built file today:

if (event.type === 'message_start') {
  messageStartUsage = event.message.usage
  yield {
    type: EventType.CUSTOM,
    name: 'anthropic.message_start',
    value: { model: event.message.model, usage: event.message.usage },
    model,
    timestamp: Date.now(),
  }
} else if (event.type === 'content_block_start') {

A provider-named CUSTOM event is the least intrusive shape for a patch, but a cross-adapter one (usage on RUN_STARTED, or a shared early-usage event the OpenAI Responses and Bedrock adapters could also emit) would be the better API.

Do you intend to try to help solve this bug with your own PR? No, because I do not have time to dig into it. The two shapes above are what I found while patching locally; I won't be sending a PR.

• I agree to follow this project's Code of Conduct
• I understand that if my bug cannot be reliably reproduced in a debuggable environment, it will probably not be fixed and this issue may even be closed.