From 82e3903aea2182d2546a99cf535276769a0821f3 Mon Sep 17 00:00:00 2001 From: Chris Trzesniewski Date: Thu, 15 Oct 2026 08:32:47 +0200 Subject: [PATCH 1/3] fix(ai-anthropic): keep the input usage when a stream fails after message_start - Put the message_start usage on the RUN_ERROR that processAnthropicStream yields when the stream throws - Skip it when the closing message_delta already reported usage, so a call never reports it twice - Keep the output count that message_start reported - Say in the messageStartUsage comment that a failed stream reads it too - Cover a throw after message_start, before message_start and after the closing message_delta in usage-extraction.test.ts - Cover an Anthropic SSE error event through the real SDK in stream-failure-usage-wire.test.ts - Add a patch changeset for @tanstack/ai-anthropic Fixes #1735 --- .changeset/anthropic-stream-failure-usage.md | 5 + packages/ai-anthropic/src/adapters/text.ts | 7 +- .../tests/stream-failure-usage-wire.test.ts | 68 ++++++++++++++ .../tests/usage-extraction.test.ts | 94 +++++++++++++++++++ 4 files changed, 173 insertions(+), 1 deletion(-) create mode 100644 .changeset/anthropic-stream-failure-usage.md create mode 100644 packages/ai-anthropic/tests/stream-failure-usage-wire.test.ts diff --git a/.changeset/anthropic-stream-failure-usage.md b/.changeset/anthropic-stream-failure-usage.md new file mode 100644 index 0000000000..6c95a603cd --- /dev/null +++ b/.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. diff --git a/packages/ai-anthropic/src/adapters/text.ts b/packages/ai-anthropic/src/adapters/text.ts index d02313231b..69bf55d4bd 100644 --- a/packages/ai-anthropic/src/adapters/text.ts +++ b/packages/ai-anthropic/src/adapters/text.ts @@ -1548,7 +1548,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 { @@ -2154,6 +2154,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, @@ -2169,6 +2173,7 @@ export class AnthropicTextAdapter< // Forward the Anthropic SDK error's `.error` response body when present. ...(rawEvent !== undefined && { rawEvent }), ...(retryAfterMs !== undefined && { retryAfterMs }), + ...(usage && { usage }), error: { message: err.message || 'Unknown error occurred', code, diff --git a/packages/ai-anthropic/tests/stream-failure-usage-wire.test.ts b/packages/ai-anthropic/tests/stream-failure-usage-wire.test.ts new file mode 100644 index 0000000000..0790764712 --- /dev/null +++ b/packages/ai-anthropic/tests/stream-failure-usage-wire.test.ts @@ -0,0 +1,68 @@ +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 = [] + for await (const chunk of adapter.chatStream({ + logger: resolveDebugOption(false), + model, + 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 }, + }) + }) +}) diff --git a/packages/ai-anthropic/tests/usage-extraction.test.ts b/packages/ai-anthropic/tests/usage-extraction.test.ts index 048bf6b5e9..6a70a86189 100644 --- a/packages/ai-anthropic/tests/usage-extraction.test.ts +++ b/packages/ai-anthropic/tests/usage-extraction.test.ts @@ -64,6 +64,21 @@ function createMockStream( } } +function createFailingStream( + chunks: Array>, + error: Error, +): AsyncIterable> { + 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() @@ -627,4 +642,83 @@ 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 = [] + for await (const chunk of createAdapter().chatStream({ + model: 'claude-opus-4-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) + }, + ) + }) }) From 9cfa013ca1a73d6a11346158eb58cc64fcb268bb Mon Sep 17 00:00:00 2001 From: Chris Trzesniewski Date: Thu, 15 Oct 2026 08:32:47 +0200 Subject: [PATCH 2/3] docs(ai-anthropic): say that a stream that fails mid-way reports its usage - Extend the usage on RUN_ERROR paragraph of the Anthropic adapter page to a stream that fails after it starts - Say that the input counts are complete and the output count can be lower than the real one - Refresh the page's updatedAt in docs/config.json --- docs/adapters/anthropic.md | 2 +- docs/config.json | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/adapters/anthropic.md b/docs/adapters/anthropic.md index 8b8b2b73a8..e5145798f1 100644 --- a/docs/adapters/anthropic.md +++ b/docs/adapters/anthropic.md @@ -298,7 +298,7 @@ const stream = chat({ Anthropic's Messages API _requires_ `max_tokens` on every request, so the adapter always sends a value. When you don't set `modelOptions.max_tokens`, it defaults to the selected model's full output ceiling (`max_output_tokens` from the model metadata — e.g. 64K for Sonnet, 128K for Opus), falling back to a safe constant for unrecognized models. `max_tokens` is a ceiling, not a reservation — billing is on tokens actually generated — so this default costs nothing extra and avoids the silent mid-response truncation (`stop_reason: "max_tokens"`) that a low default would cause. Set `max_tokens` explicitly only when you want to _cap_ output below the model ceiling. If a response is truncated while using the default cap, the adapter logs a warning (visible with [debug logging](../advanced/debug-logging) enabled). -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 clamps this default to ~21K tokens. The Anthropic SDK rejects a non-streaming request whose `max_tokens` could exceed its 10-minute timeout, so the full ceiling can't be used there. Streaming chat is unaffected. To raise the structured-output ceiling toward a model's true max, stream the response. diff --git a/docs/config.json b/docs/config.json index aee4f36442..6374a4f655 100644 --- a/docs/config.json +++ b/docs/config.json @@ -1215,7 +1215,7 @@ "label": "Anthropic", "to": "adapters/anthropic", "addedAt": "2026-04-15", - "updatedAt": "2026-10-09" + "updatedAt": "2026-10-15" }, { "label": "Google Gemini", From 6d5b0d6b1e8194b075624274ad59514cafca5aed Mon Sep 17 00:00:00 2001 From: Chris Trzesniewski Date: Thu, 15 Oct 2026 08:32:47 +0200 Subject: [PATCH 3/3] test(e2e): cover the usage on an Anthropic stream that fails after message_start - Add an aimock fixture that drops the connection after message_start, the text block start and one text delta - Add the /api/anthropic-stream-failure-usage route, which returns the usage of the RUN_ERROR that chat() yields - Assert that the RUN_ERROR carries the message_start input and output counts - Regenerate routeTree.gen.ts with the dev server --- .../fixtures/stream-failure-usage/basic.json | 14 ++++++ testing/e2e/src/routeTree.gen.ts | 22 +++++++++ .../api.anthropic-stream-failure-usage.ts | 48 +++++++++++++++++++ .../anthropic-stream-failure-usage.spec.ts | 27 +++++++++++ 4 files changed, 111 insertions(+) create mode 100644 testing/e2e/fixtures/stream-failure-usage/basic.json create mode 100644 testing/e2e/src/routes/api.anthropic-stream-failure-usage.ts create mode 100644 testing/e2e/tests/anthropic-stream-failure-usage.spec.ts diff --git a/testing/e2e/fixtures/stream-failure-usage/basic.json b/testing/e2e/fixtures/stream-failure-usage/basic.json new file mode 100644 index 0000000000..d72f1c301f --- /dev/null +++ b/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 + } + ] +} diff --git a/testing/e2e/src/routeTree.gen.ts b/testing/e2e/src/routeTree.gen.ts index 780d06b4dc..0ade0da49c 100644 --- a/testing/e2e/src/routeTree.gen.ts +++ b/testing/e2e/src/routeTree.gen.ts @@ -60,6 +60,7 @@ import { Route as ApiAnthropicOpus5CombinedWireRouteImport } from './routes/api. import { Route as ApiAnthropicRedactedThinkingWireRouteImport } from './routes/api.anthropic-redacted-thinking-wire' 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 ApiAnthropicStructuredUsageRouteImport } from './routes/api.anthropic-structured-usage' import { Route as ApiAnthropicThinkingOrderWireRouteImport } from './routes/api.anthropic-thinking-order-wire' import { Route as ApiAnthropicTruncatedToolInputWireRouteImport } from './routes/api.anthropic-truncated-tool-input-wire' @@ -446,6 +447,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 ApiAnthropicStructuredUsageRoute = ApiAnthropicStructuredUsageRouteImport.update({ id: '/api/anthropic-structured-usage', @@ -1137,6 +1144,7 @@ export interface FileRoutesByFullPath { '/api/anthropic-redacted-thinking-wire': typeof ApiAnthropicRedactedThinkingWireRoute '/api/anthropic-skills-wire': typeof ApiAnthropicSkillsWireRoute '/api/anthropic-sonnet-5-5-wire': typeof ApiAnthropicSonnet55WireRoute + '/api/anthropic-stream-failure-usage': typeof ApiAnthropicStreamFailureUsageRoute '/api/anthropic-structured-usage': typeof ApiAnthropicStructuredUsageRoute '/api/anthropic-thinking-order-wire': typeof ApiAnthropicThinkingOrderWireRoute '/api/anthropic-truncated-tool-input-wire': typeof ApiAnthropicTruncatedToolInputWireRoute @@ -1311,6 +1319,7 @@ export interface FileRoutesByTo { '/api/anthropic-redacted-thinking-wire': typeof ApiAnthropicRedactedThinkingWireRoute '/api/anthropic-skills-wire': typeof ApiAnthropicSkillsWireRoute '/api/anthropic-sonnet-5-5-wire': typeof ApiAnthropicSonnet55WireRoute + '/api/anthropic-stream-failure-usage': typeof ApiAnthropicStreamFailureUsageRoute '/api/anthropic-structured-usage': typeof ApiAnthropicStructuredUsageRoute '/api/anthropic-thinking-order-wire': typeof ApiAnthropicThinkingOrderWireRoute '/api/anthropic-truncated-tool-input-wire': typeof ApiAnthropicTruncatedToolInputWireRoute @@ -1486,6 +1495,7 @@ export interface FileRoutesById { '/api/anthropic-redacted-thinking-wire': typeof ApiAnthropicRedactedThinkingWireRoute '/api/anthropic-skills-wire': typeof ApiAnthropicSkillsWireRoute '/api/anthropic-sonnet-5-5-wire': typeof ApiAnthropicSonnet55WireRoute + '/api/anthropic-stream-failure-usage': typeof ApiAnthropicStreamFailureUsageRoute '/api/anthropic-structured-usage': typeof ApiAnthropicStructuredUsageRoute '/api/anthropic-thinking-order-wire': typeof ApiAnthropicThinkingOrderWireRoute '/api/anthropic-truncated-tool-input-wire': typeof ApiAnthropicTruncatedToolInputWireRoute @@ -1662,6 +1672,7 @@ export interface FileRouteTypes { | '/api/anthropic-redacted-thinking-wire' | '/api/anthropic-skills-wire' | '/api/anthropic-sonnet-5-5-wire' + | '/api/anthropic-stream-failure-usage' | '/api/anthropic-structured-usage' | '/api/anthropic-thinking-order-wire' | '/api/anthropic-truncated-tool-input-wire' @@ -1836,6 +1847,7 @@ export interface FileRouteTypes { | '/api/anthropic-redacted-thinking-wire' | '/api/anthropic-skills-wire' | '/api/anthropic-sonnet-5-5-wire' + | '/api/anthropic-stream-failure-usage' | '/api/anthropic-structured-usage' | '/api/anthropic-thinking-order-wire' | '/api/anthropic-truncated-tool-input-wire' @@ -2010,6 +2022,7 @@ export interface FileRouteTypes { | '/api/anthropic-redacted-thinking-wire' | '/api/anthropic-skills-wire' | '/api/anthropic-sonnet-5-5-wire' + | '/api/anthropic-stream-failure-usage' | '/api/anthropic-structured-usage' | '/api/anthropic-thinking-order-wire' | '/api/anthropic-truncated-tool-input-wire' @@ -2185,6 +2198,7 @@ export interface RootRouteChildren { ApiAnthropicRedactedThinkingWireRoute: typeof ApiAnthropicRedactedThinkingWireRoute ApiAnthropicSkillsWireRoute: typeof ApiAnthropicSkillsWireRoute ApiAnthropicSonnet55WireRoute: typeof ApiAnthropicSonnet55WireRoute + ApiAnthropicStreamFailureUsageRoute: typeof ApiAnthropicStreamFailureUsageRoute ApiAnthropicStructuredUsageRoute: typeof ApiAnthropicStructuredUsageRoute ApiAnthropicThinkingOrderWireRoute: typeof ApiAnthropicThinkingOrderWireRoute ApiAnthropicTruncatedToolInputWireRoute: typeof ApiAnthropicTruncatedToolInputWireRoute @@ -2663,6 +2677,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-structured-usage': { id: '/api/anthropic-structured-usage' path: '/api/anthropic-structured-usage' @@ -3623,6 +3644,7 @@ const rootRouteChildren: RootRouteChildren = { ApiAnthropicRedactedThinkingWireRoute: ApiAnthropicRedactedThinkingWireRoute, ApiAnthropicSkillsWireRoute: ApiAnthropicSkillsWireRoute, ApiAnthropicSonnet55WireRoute: ApiAnthropicSonnet55WireRoute, + ApiAnthropicStreamFailureUsageRoute: ApiAnthropicStreamFailureUsageRoute, ApiAnthropicStructuredUsageRoute: ApiAnthropicStructuredUsageRoute, ApiAnthropicThinkingOrderWireRoute: ApiAnthropicThinkingOrderWireRoute, ApiAnthropicTruncatedToolInputWireRoute: diff --git a/testing/e2e/src/routes/api.anthropic-stream-failure-usage.ts b/testing/e2e/src/routes/api.anthropic-stream-failure-usage.ts new file mode 100644 index 0000000000..df210da37a --- /dev/null +++ b/testing/e2e/src/routes/api.anthropic-stream-failure-usage.ts @@ -0,0 +1,48 @@ +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 | undefined + try { + for await (const chunk of chat({ + adapter, + 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 }) + }, + }, + }, +}) diff --git a/testing/e2e/tests/anthropic-stream-failure-usage.spec.ts b/testing/e2e/tests/anthropic-stream-failure-usage.spec.ts new file mode 100644 index 0000000000..ec4830e0b9 --- /dev/null +++ b/testing/e2e/tests/anthropic-stream-failure-usage.spec.ts @@ -0,0 +1,27 @@ +import { test, expect } from './fixtures' + +/** + * Regression for #1735. An Anthropic stream that fails after `message_start` + * ends in `RUN_ERROR`, and Anthropic bills the input tokens of that call. The + * `RUN_ERROR` must carry the counts that `message_start` reported. + */ +test.describe('anthropic stream that fails after message_start', () => { + test('RUN_ERROR carries the message_start usage', async ({ request }) => { + const res = await request.post('/api/anthropic-stream-failure-usage') + expect(res.ok()).toBe(true) + + const { ok, runError, error } = (await res.json()) as { + ok: boolean + error?: string + runError?: { usage?: Record } + } + + expect(error ?? null).toBeNull() + expect(ok).toBe(true) + expect(runError?.usage).toEqual({ + promptTokens: 18, + completionTokens: 1, + totalTokens: 19, + }) + }) +})