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/docs/adapters/anthropic-advanced.md b/docs/adapters/anthropic-advanced.md index f47fd085e6..a92787918c 100644 --- a/docs/adapters/anthropic-advanced.md +++ b/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. diff --git a/docs/config.json b/docs/config.json index ba16ab89da..5d4255995c 100644 --- a/docs/config.json +++ b/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", diff --git a/packages/ai-anthropic/src/adapters/text.ts b/packages/ai-anthropic/src/adapters/text.ts index faf733481f..1b019fddd6 100644 --- a/packages/ai-anthropic/src/adapters/text.ts +++ b/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', 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..826fa28d8f --- /dev/null +++ b/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 = [] + 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 }, + }) + }) +}) diff --git a/packages/ai-anthropic/tests/usage-extraction.test.ts b/packages/ai-anthropic/tests/usage-extraction.test.ts index 4c10690f38..e5b75a6702 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() @@ -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 = [] + 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) + }, + ) + }) }) 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 e5ec31ba17..bffd62fa7b 100644 --- a/testing/e2e/src/routeTree.gen.ts +++ b/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, 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..ba055a86f5 --- /dev/null +++ b/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 | 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 }) + }, + }, + }, +}) 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, + }) + }) +})