From f5bdf2b75d3b1bfaac3d2316ab87cf035d39dc1c Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Fri, 30 Oct 2026 11:10:12 +0100 Subject: [PATCH 1/3] fix(server-utils): Record Anthropic chat spans for thread and tool-runner streams on 1.2+ On @anthropic-ai/sdk 1.2.0+, streamed turns from beta.messages.toolRunner({ stream: true }) and beta.messages.thread().stream() had no gen_ai.chat spans. The thread parses the response of each create() call through a new APIPromise our hook never sees, so the span never ended. Hook BetaMessageStream._start(), match each request to its stream by abort signal, and end the span from the stream's events. instrumentMessageStream now also ends the span on 'end', so aborted or truncated streams no longer leave it open. Fixes #25816 Co-Authored-By: Claude Opus 5.5 (1M context) --- .../tracing/anthropic/v1/instrument.mjs | 2 +- .../tracing/anthropic/v1/scenario-threads.mjs | 194 ++++++++++++++++++ .../suites/tracing/anthropic/v1/test.ts | 72 ++++++- .../src/ai/anthropic-ai/streaming.ts | 6 + .../src/integrations/anthropic.ts | 48 ++++- .../src/orchestrion/config/anthropic-ai.ts | 8 + 6 files changed, 327 insertions(+), 3 deletions(-) create mode 100644 dev-packages/node-integration-tests/suites/tracing/anthropic/v1/scenario-threads.mjs diff --git a/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/instrument.mjs b/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/instrument.mjs index b5472310228b..f09ea5121842 100644 --- a/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/instrument.mjs +++ b/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/instrument.mjs @@ -5,7 +5,7 @@ Sentry.init({ dsn: 'https://public@dsn.ingest.sentry.io/1337', release: '1.0', tracesSampleRate: 1.0, - dataCollection: { genAI: { inputs: false, outputs: false } }, + dataCollection: { genAI: { inputs: false, outputs: process.env.RECORD_OUTPUTS === 'true' } }, transport: loggingTransport, // Registers Sentry as the global tracer provider, which is what 1.x traces itself through. enableOpenTelemetrySetup: process.env.OTEL_SETUP === 'true', diff --git a/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/scenario-threads.mjs b/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/scenario-threads.mjs new file mode 100644 index 000000000000..acdc2a1f743a --- /dev/null +++ b/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/scenario-threads.mjs @@ -0,0 +1,194 @@ +import Anthropic from '@anthropic-ai/sdk'; +import * as Sentry from '@sentry/node'; +import express from 'express'; + +const MODEL = 'claude-3-haiku-20240307'; + +// The first turn calls the weather tool and the second answers. Ids are `msg__`, so each span can be told apart. +function reply(mode, turn) { + const id = `msg_${mode}_${turn}`; + return turn === 'tools' + ? { + id, + content: [{ type: 'tool_use', id: `toolu_${mode}`, name: 'weather', input: { city: 'Vienna' } }], + stop_reason: 'tool_use', + usage: { input_tokens: 10, output_tokens: 20 }, + } + : { + id, + content: [{ type: 'text', text: 'Sunny.' }], + stop_reason: 'end_turn', + usage: { input_tokens: 30, output_tokens: 5 }, + }; +} + +function streamEvents({ id, content, stop_reason, usage }) { + const events = [ + { + type: 'message_start', + message: { + id, + type: 'message', + role: 'assistant', + model: MODEL, + content: [], + usage: { input_tokens: usage.input_tokens }, + }, + }, + ]; + content.forEach((block, index) => { + if (block.type === 'tool_use') { + events.push( + { type: 'content_block_start', index, content_block: { ...block, input: {} } }, + { + type: 'content_block_delta', + index, + delta: { type: 'input_json_delta', partial_json: JSON.stringify(block.input) }, + }, + ); + } else { + events.push( + { type: 'content_block_start', index, content_block: { type: 'text', text: '' } }, + { type: 'content_block_delta', index, delta: { type: 'text_delta', text: block.text } }, + ); + } + events.push({ type: 'content_block_stop', index }); + }); + events.push( + { + type: 'message_delta', + delta: { stop_reason, stop_sequence: null }, + usage: { output_tokens: usage.output_tokens }, + }, + { type: 'message_stop' }, + ); + return events; +} + +function startMockAnthropicServer() { + const app = express(); + app.use(express.json()); + + app.post('/anthropic/v1/messages', (req, res) => { + const mode = req.headers['x-mode']; + if (mode === 'thread_stream_error') { + res.status(400).json({ type: 'error', error: { type: 'invalid_request_error', message: 'Bad thread' } }); + return; + } + const answersTool = req.body.messages.some( + message => Array.isArray(message.content) && message.content.some(block => block.type === 'tool_result'), + ); + const message = reply(mode, answersTool ? 'final' : 'tools'); + + if (!req.body.stream) { + res.json({ type: 'message', role: 'assistant', model: MODEL, stop_sequence: null, ...message }); + return; + } + + res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', Connection: 'keep-alive' }); + // The body closes after the first text delta, without `message_stop`. + const events = mode === 'thread_stream_truncated' ? streamEvents(message).slice(0, 3) : streamEvents(message); + events.forEach((event, index) => { + setTimeout(() => { + res.write(`event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`); + if (index === events.length - 1) { + res.end(); + } + }, index * 5); + }); + }); + + return new Promise(resolve => { + const server = app.listen(0, () => { + resolve(server); + }); + }); +} + +const tools = [ + { + name: 'weather', + description: 'Get the weather by city', + input_schema: { type: 'object', properties: { city: { type: 'string' } } }, + run: () => 'sunny', + }, +]; +const params = { model: MODEL, max_tokens: 100, messages: [{ role: 'user', content: 'Weather in Vienna?' }] }; + +const MODES = { + async runner_stream_eager(client) { + for await (const stream of client.beta.messages.toolRunner({ + ...params, + tools, + stream: true, + runToolsEagerly: true, + })) { + await stream.finalMessage(); + } + }, + async runner_stream(client) { + for await (const stream of client.beta.messages.toolRunner({ ...params, tools, stream: true })) { + await stream.finalMessage(); + } + }, + // How Bedrock and Vertex clients, which have no threads, run every tool runner. + async runner_stream_unthreaded(client) { + for await (const stream of client.beta.messages.toolRunner({ ...params, tools, stream: true, thread: false })) { + await stream.finalMessage(); + } + }, + async runner(client) { + for await (const _message of client.beta.messages.toolRunner({ ...params, tools })) { + // consume + } + }, + async thread_send(client) { + await client.beta.messages.thread({ model: MODEL, max_tokens: 100 }).send(params.messages); + }, + async thread_stream(client) { + await client.beta.messages.thread({ model: MODEL, max_tokens: 100 }).stream(params.messages).finalMessage(); + }, + async thread_stream_abort(client) { + const stream = client.beta.messages.thread({ model: MODEL, max_tokens: 100 }).stream(params.messages); + stream.once('streamEvent', () => stream.abort()); + await stream.finalMessage().catch(() => {}); + }, + async thread_stream_abort_early(client) { + const stream = client.beta.messages.thread({ model: MODEL, max_tokens: 100 }).stream(params.messages); + stream.abort(); + await stream.finalMessage().catch(() => {}); + }, + async thread_stream_truncated(client) { + await client.beta.messages + .thread({ model: MODEL, max_tokens: 100 }) + .stream(params.messages) + .finalMessage() + .catch(() => {}); + }, + async thread_stream_error(client) { + await client.beta.messages + .thread({ model: MODEL, max_tokens: 100 }) + .stream(params.messages) + .finalMessage() + .catch(() => {}); + }, +}; + +async function run() { + const server = await startMockAnthropicServer(); + const baseURL = `http://localhost:${server.address().port}/anthropic`; + + await Sentry.startSpan({ op: 'function', name: 'main' }, async () => { + for (const [mode, call] of Object.entries(MODES)) { + await Sentry.startSpan({ name: mode }, () => + call(new Anthropic({ apiKey: 'mock-api-key', baseURL, maxRetries: 0, defaultHeaders: { 'x-mode': mode } })), + ); + } + }); + + await Sentry.flush(2000); + + server.close(); +} + +run(); diff --git a/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/test.ts b/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/test.ts index 0983283d178f..2c6e1b216cab 100644 --- a/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/test.ts @@ -2,11 +2,15 @@ import type { Event, SerializedStreamedSpanContainer } from '@sentry/core'; import { GEN_AI_RESPONSE_ID, GEN_AI_RESPONSE_STREAMING, + GEN_AI_RESPONSE_TEXT, + GEN_AI_RESPONSE_TOOL_CALLS, GEN_AI_TOOL_NAME, + GEN_AI_USAGE_INPUT_TOKENS, + GEN_AI_USAGE_OUTPUT_TOKENS, SENTRY_OP, SENTRY_ORIGIN, } from '@sentry/conventions/attributes'; -import { GEN_AI_EXECUTE_TOOL } from '@sentry/conventions/op'; +import { GEN_AI_CHAT, GEN_AI_EXECUTE_TOOL } from '@sentry/conventions/op'; import { afterAll, describe, expect } from 'vitest'; import { GEN_AI_TOOL_CALL_ID_ATTRIBUTE } from '../../../../../../packages/server-utils/src/ai/core/gen-ai-attributes'; import { cleanupChildProcesses, createEsmAndCjsTests } from '../../../../utils/runner'; @@ -115,6 +119,72 @@ describe('Anthropic integration (1.x)', () => { } }); + // Since 1.2, tool runners and `thread().stream()` send each turn through a thread, which parses the response itself. + createEsmAndCjsTests(__dirname, 'scenario-threads.mjs', 'instrument.mjs', (createRunner, test) => { + const MODES = [ + { mode: 'runner_stream_eager', streaming: true, turns: ['tools', 'final'] }, + { mode: 'runner_stream', streaming: true, turns: ['tools', 'final'] }, + { mode: 'runner_stream_unthreaded', streaming: true, turns: ['tools', 'final'] }, + { mode: 'runner', streaming: false, turns: ['tools', 'final'] }, + { mode: 'thread_send', streaming: false, turns: ['tools'] }, + { mode: 'thread_stream', streaming: true, turns: ['tools'] }, + // Only the span's status is known for these. The stream is aborted on its first event or before the response, + // its body closes before `message_stop`, or the request fails. + { mode: 'thread_stream_abort', streaming: true, turns: ['abort'], status: 'ok' }, + { mode: 'thread_stream_abort_early', streaming: true, turns: ['abort'], status: 'ok' }, + { mode: 'thread_stream_truncated', streaming: true, turns: ['truncated'], status: 'ok' }, + { mode: 'thread_stream_error', streaming: true, turns: ['error'], status: 'error' }, + ]; + const TURNS: Record = { + tools: { input: 10, output: 20, toolCall: true }, + final: { input: 30, output: 5, text: 'Sunny.' }, + }; + + for (const env of [{}, { OTEL_SETUP: 'true' }]) { + test(`emits one chat span per thread turn with ${JSON.stringify(env)}`, async () => { + const runner = createRunner().withEnv({ ...env, RECORD_OUTPUTS: 'true' }); + const spansPromise = runner.collectStreamedSpansUntilSegment('main'); + + await runner.start().completed(); + + const spans = await spansPromise; + const chatSpans = spans.filter(span => span.attributes[SENTRY_OP]?.value === GEN_AI_CHAT); + expect(chatSpans).toHaveLength(MODES.flatMap(({ turns }) => turns).length); + expect(spans.filter(span => span.name.startsWith('anthropic.'))).toEqual([]); + + for (const { mode, streaming, turns, status } of MODES) { + const modeSpan = spans.find(span => span.name === mode)!; + const modeChatSpans = chatSpans.filter(span => span.parent_span_id === modeSpan.span_id); + expect(modeChatSpans, mode).toHaveLength(turns.length); + + if (status) { + expect(modeChatSpans[0]!.status, mode).toBe(status); + continue; + } + + for (const turn of turns) { + const id = `msg_${mode}_${turn}`; + const chatSpan = spanForCall(modeChatSpans, id); + const { input, output, text, toolCall } = TURNS[turn]!; + expect(chatSpan.status, id).toBe('ok'); + expect(chatSpan.attributes[GEN_AI_RESPONSE_STREAMING]?.value ?? false, id).toBe(streaming); + expect(chatSpan.attributes[GEN_AI_USAGE_INPUT_TOKENS]?.value, id).toBe(input); + expect(chatSpan.attributes[GEN_AI_USAGE_OUTPUT_TOKENS]?.value, id).toBe(output); + if (text) { + expect(chatSpan.attributes[GEN_AI_RESPONSE_TEXT]?.value, id).toBe(text); + } + const toolCalls = chatSpan.attributes[GEN_AI_RESPONSE_TOOL_CALLS]?.value as string | undefined; + expect(toolCalls && JSON.parse(toolCalls), id).toEqual( + toolCall + ? [{ type: 'tool_use', id: `toolu_${mode}`, name: 'weather', input: { city: 'Vienna' } }] + : undefined, + ); + } + } + }); + } + }); + createEsmAndCjsTests(__dirname, 'scenario-session-tool-runner.mjs', 'instrument.mjs', (createRunner, test) => { for (const env of [{}, { OTEL_SETUP: 'true' }]) { test(`emits an execute_tool span for each tool a session runner runs with ${JSON.stringify(env)}`, async () => { diff --git a/packages/server-utils/src/ai/anthropic-ai/streaming.ts b/packages/server-utils/src/ai/anthropic-ai/streaming.ts index 1c3ed58674d6..caebf609cf40 100644 --- a/packages/server-utils/src/ai/anthropic-ai/streaming.ts +++ b/packages/server-utils/src/ai/anthropic-ai/streaming.ts @@ -284,6 +284,12 @@ export function instrumentMessageStream endStreamSpan(span, state, recordOutputs); }); + // Fires last on every stream, so a stream that stops without 'message' (aborted, or the body closed early) still + // ends its span. An 'abort' listener would also stop the SDK from raising an unobserved abort as a rejection. + stream.on('end', () => { + endStreamSpan(span, state, recordOutputs); + }); + stream.on('error', (error: unknown) => { // Attaching this listener stops the stream error from being raised as an unhandled rejection, so // we capture it here to avoid swallowing it (e.g. for callers that don't await/iterate the stream). diff --git a/packages/server-utils/src/integrations/anthropic.ts b/packages/server-utils/src/integrations/anthropic.ts index c39ebcdfdeb2..0ea4068dd69e 100644 --- a/packages/server-utils/src/integrations/anthropic.ts +++ b/packages/server-utils/src/integrations/anthropic.ts @@ -73,6 +73,7 @@ function instrumentAnthropic(options: AnthropicAiOptions): void { ); }, deferSpanEnd: ({ span, data, end }) => + wrapThreadStreamRequest(span, data, end, options) || wrapApiPromiseResponse( data.result, response => { @@ -82,7 +83,8 @@ function instrumentAnthropic(options: AnthropicAiOptions): void { } }, end, - ) || wrapStreamResult(span, data, stream, options), + ) || + wrapStreamResult(span, data, stream, options), }, ); } @@ -97,6 +99,15 @@ function instrumentAnthropic(options: AnthropicAiOptions): void { .end.subscribe(message => { instrumentSessionToolRunner((message as OrchestrionChannelContext).self, options); }); + // On `start`, because `_start()` sends the first request before it returns. + diagnosticsChannel + .tracingChannel(CHANNELS.ANTHROPIC_THREAD_STREAM) + .start.subscribe(message => { + const stream = (message as OrchestrionChannelContext).self as ThreadStream | undefined; + if (isMessageStream(stream) && stream.controller?.signal) { + threadStreams.set(stream.controller.signal, stream); + } + }); } /** @@ -200,6 +211,41 @@ function isMessageStream(value: unknown): value is MessageStreamEmitter { return !!value && typeof (value as MessageStreamEmitter).on === 'function'; } +type ThreadStream = MessageStreamEmitter & { controller?: { signal?: AbortSignal } }; + +/** + * The streams of threads and tool runners on 1.2+, by the signal they send each `create()` call with. Such a + * stream parses the response itself, so the `create()` result is never parsed and never ends its span. + */ +const threadStreams = new WeakMap(); + +/** + * End the span of a thread stream's `create()` call from the stream once the response arrives. A thread may + * retry a rejected call, so a failed call ends its own span and only the one that succeeds reads the stream. + */ +function wrapThreadStreamRequest( + span: Span, + data: OrchestrionChannelContext, + end: (error?: unknown) => void, + options: AnthropicAiOptions, +): boolean { + const signal = (data.arguments?.[1] as { signal?: AbortSignal } | undefined)?.signal; + const stream = signal && threadStreams.get(signal); + const request = data.result as { _responseProps?: () => Promise } | undefined; + if (!stream || typeof request?._responseProps !== 'function') { + return false; + } + + const { recordOutputs } = resolveAIRecordingOptions(options); + // Not `asResponse()`, which on 1.x also ends the SDK's own span when the user turned its tracing on. + void request._responseProps().then( + () => instrumentMessageStream(stream, span, recordOutputs), + // A caller abort is not a failed request. + error => end(signal.aborted ? undefined : error), + ); + return true; +} + /** * Hand span-ending ownership to a streamed result: returns `true` to skip the normal `beforeSpanEnd`, * `false` for non-streaming results (which end via `beforeSpanEnd`). diff --git a/packages/server-utils/src/orchestrion/config/anthropic-ai.ts b/packages/server-utils/src/orchestrion/config/anthropic-ai.ts index f0a953e03ef1..422b3e367f78 100644 --- a/packages/server-utils/src/orchestrion/config/anthropic-ai.ts +++ b/packages/server-utils/src/orchestrion/config/anthropic-ai.ts @@ -38,6 +38,13 @@ export const anthropicAiConfig = [ module: { name: '@anthropic-ai/sdk', versionRange: '>=0.97.0 <2', filePath }, functionQuery: { className: 'SessionToolRunner' }, })), + // Thread and tool-runner streams parse the response of their `create()` call themselves, so the chat channel + // never sees that stream. + ...['lib/BetaMessageStream.js', 'lib/BetaMessageStream.mjs'].map(filePath => ({ + channelName: 'thread-stream', + module: { name: '@anthropic-ai/sdk', versionRange: '>=1.2.0 <2', filePath }, + functionQuery: { className: 'BetaMessageStream', methodName: '_start', kind: 'Sync' as const }, + })), ] satisfies InstrumentationConfig[]; export const anthropicAiModuleNames = getModuleNames(anthropicAiConfig); @@ -47,4 +54,5 @@ export const anthropicAiChannels = { ANTHROPIC_MESSAGES_STREAM: 'orchestrion:@anthropic-ai/sdk:messages-stream', ANTHROPIC_TOOL_RUNNER: 'orchestrion:@anthropic-ai/sdk:tool-runner', ANTHROPIC_SESSION_TOOL_RUNNER: 'orchestrion:@anthropic-ai/sdk:session-tool-runner', + ANTHROPIC_THREAD_STREAM: 'orchestrion:@anthropic-ai/sdk:thread-stream', } as const; From 419870519d844e93287f79ddb72d35c5df1ccf19 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Mon, 2 Nov 2026 15:05:32 +0100 Subject: [PATCH 2/3] fix(server-utils): Release a thread stream once its request is matched A later create() sent with a finished thread stream's signal was wired to that stream and its span never ended. Drop the entry once an attempt succeeds, since only rejected attempts are retried. Also cover the 'end' listener with a unit test and update the wrapStreamResult JSDoc. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../tracing/anthropic/v1/scenario-threads.mjs | 13 +++++++ .../suites/tracing/anthropic/v1/test.ts | 1 + .../src/integrations/anthropic.ts | 8 +++-- .../ai/lib/tracing/anthropic-ai-usage.test.ts | 36 ++++++++++++++++++- 4 files changed, 55 insertions(+), 3 deletions(-) diff --git a/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/scenario-threads.mjs b/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/scenario-threads.mjs index acdc2a1f743a..ea166274dab4 100644 --- a/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/scenario-threads.mjs +++ b/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/scenario-threads.mjs @@ -148,6 +148,19 @@ const MODES = { async thread_stream(client) { await client.beta.messages.thread({ model: MODEL, max_tokens: 100 }).stream(params.messages).finalMessage(); }, + // A later request sent with a finished thread stream's signal is not that stream's request. + async thread_stream_signal_reuse(client) { + const stream = client.beta.messages.thread({ model: MODEL, max_tokens: 100 }).stream(params.messages); + await stream.finalMessage(); + const answer = { role: 'user', content: [{ type: 'tool_result', tool_use_id: 'toolu_1', content: 'sunny' }] }; + const followUp = await client.beta.messages.create( + { ...params, messages: [...params.messages, answer], stream: true }, + { signal: stream.controller.signal }, + ); + for await (const _event of followUp) { + // consume + } + }, async thread_stream_abort(client) { const stream = client.beta.messages.thread({ model: MODEL, max_tokens: 100 }).stream(params.messages); stream.once('streamEvent', () => stream.abort()); diff --git a/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/test.ts b/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/test.ts index 2c6e1b216cab..01ce273ea538 100644 --- a/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/test.ts @@ -128,6 +128,7 @@ describe('Anthropic integration (1.x)', () => { { mode: 'runner', streaming: false, turns: ['tools', 'final'] }, { mode: 'thread_send', streaming: false, turns: ['tools'] }, { mode: 'thread_stream', streaming: true, turns: ['tools'] }, + { mode: 'thread_stream_signal_reuse', streaming: true, turns: ['tools', 'final'] }, // Only the span's status is known for these. The stream is aborted on its first event or before the response, // its body closes before `message_stop`, or the request fails. { mode: 'thread_stream_abort', streaming: true, turns: ['abort'], status: 'ok' }, diff --git a/packages/server-utils/src/integrations/anthropic.ts b/packages/server-utils/src/integrations/anthropic.ts index 0ea4068dd69e..1828f62e2bde 100644 --- a/packages/server-utils/src/integrations/anthropic.ts +++ b/packages/server-utils/src/integrations/anthropic.ts @@ -239,7 +239,11 @@ function wrapThreadStreamRequest( const { recordOutputs } = resolveAIRecordingOptions(options); // Not `asResponse()`, which on 1.x also ends the SDK's own span when the user turned its tracing on. void request._responseProps().then( - () => instrumentMessageStream(stream, span, recordOutputs), + () => { + // Only a rejected attempt is retried, so nothing else may claim this stream. + threadStreams.delete(signal); + instrumentMessageStream(stream, span, recordOutputs); + }, // A caller abort is not a failed request. error => end(signal.aborted ? undefined : error), ); @@ -252,7 +256,7 @@ function wrapThreadStreamRequest( * * - `async-iterable`: patch the `Stream`'s async iterator in place so `instrumentAsyncIterableStream` ends * the span when iteration finishes. - * - `message-stream`: `instrumentMessageStream` attaches `'message'`/`'error'` listeners that end the span. + * - `message-stream`: `instrumentMessageStream` attaches `'message'`/`'error'`/`'end'` listeners that end the span. */ function wrapStreamResult( span: Span, diff --git a/packages/server-utils/test/ai/lib/tracing/anthropic-ai-usage.test.ts b/packages/server-utils/test/ai/lib/tracing/anthropic-ai-usage.test.ts index c506fbcb9387..109537394426 100644 --- a/packages/server-utils/test/ai/lib/tracing/anthropic-ai-usage.test.ts +++ b/packages/server-utils/test/ai/lib/tracing/anthropic-ai-usage.test.ts @@ -9,7 +9,7 @@ import { GEN_AI_USAGE_TOTAL_TOKENS, } from '@sentry/conventions/attributes'; import { EventEmitter } from 'node:events'; -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, vi } from 'vitest'; import type { Span } from '@sentry/core'; import { addResponseAttributes } from '../../../../src/ai/anthropic-ai'; import { instrumentAsyncIterableStream, instrumentMessageStream } from '../../../../src/ai/anthropic-ai/streaming'; @@ -271,3 +271,37 @@ describe.each(['async iterable', 'message stream'] as const)( }); }, ); + +describe('instrumentMessageStream', () => { + it('ends the span once with the streamed usage when an aborted stream ends without a message', () => { + const { span, attributes } = createMockSpan(); + const endSpy = vi.spyOn(span, 'end'); + const stream = instrumentMessageStream(new EventEmitter(), span, false); + + stream.emit('streamEvent', { + type: 'message_start', + message: { + id: 'msg_aborted', + type: 'message', + model: 'claude-sonnet-4-6', + role: 'assistant', + content: [], + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 42, output_tokens: 1 }, + }, + }); + stream.emit('abort', new Error('Request was aborted.')); + stream.emit('end'); + stream.emit('end'); + + expect(endSpy).toHaveBeenCalledTimes(1); + expect(attributes).toEqual({ + [GEN_AI_RESPONSE_STREAMING]: true, + [GEN_AI_RESPONSE_ID]: 'msg_aborted', + [GEN_AI_RESPONSE_MODEL]: 'claude-sonnet-4-6', + [GEN_AI_USAGE_INPUT_TOKENS]: 42, + [GEN_AI_USAGE_TOTAL_TOKENS]: 42, + }); + }); +}); From aedfeab53f79f17d56080b3e406c05af94facc5a Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Mon, 2 Nov 2026 15:52:09 +0100 Subject: [PATCH 3/3] test(node): Split the Anthropic thread checks into separate cases Each mode and turn is now its own test.each case, with no conditionals, so a failure points at one path. The cases share one scenario run per env, since every run is a Node process. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../suites/tracing/anthropic/v1/test.ts | 122 +++++++++++------- 1 file changed, 73 insertions(+), 49 deletions(-) diff --git a/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/test.ts b/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/test.ts index 01ce273ea538..f04311e849eb 100644 --- a/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/test.ts @@ -129,61 +129,85 @@ describe('Anthropic integration (1.x)', () => { { mode: 'thread_send', streaming: false, turns: ['tools'] }, { mode: 'thread_stream', streaming: true, turns: ['tools'] }, { mode: 'thread_stream_signal_reuse', streaming: true, turns: ['tools', 'final'] }, - // Only the span's status is known for these. The stream is aborted on its first event or before the response, - // its body closes before `message_stop`, or the request fails. - { mode: 'thread_stream_abort', streaming: true, turns: ['abort'], status: 'ok' }, - { mode: 'thread_stream_abort_early', streaming: true, turns: ['abort'], status: 'ok' }, - { mode: 'thread_stream_truncated', streaming: true, turns: ['truncated'], status: 'ok' }, - { mode: 'thread_stream_error', streaming: true, turns: ['error'], status: 'error' }, ]; - const TURNS: Record = { - tools: { input: 10, output: 20, toolCall: true }, - final: { input: 30, output: 5, text: 'Sunny.' }, - }; + const FINAL_TURN_MODES = MODES.filter(({ turns }) => turns.includes('final')); + // Only the span's status is known for these. The stream is aborted on its first event or before the response, + // its body closes before `message_stop`, or the request fails. + const STATUS_ONLY_MODES = [ + { mode: 'thread_stream_abort', status: 'ok' }, + { mode: 'thread_stream_abort_early', status: 'ok' }, + { mode: 'thread_stream_truncated', status: 'ok' }, + { mode: 'thread_stream_error', status: 'error' }, + ]; - for (const env of [{}, { OTEL_SETUP: 'true' }]) { - test(`emits one chat span per thread turn with ${JSON.stringify(env)}`, async () => { - const runner = createRunner().withEnv({ ...env, RECORD_OUTPUTS: 'true' }); - const spansPromise = runner.collectStreamedSpansUntilSegment('main'); + describe.each([{}, { OTEL_SETUP: 'true' }])('with %j', env => { + // Every scenario run is a Node process, so the cases below share one run. + let run: Promise | undefined; + function getSpans(): Promise { + run ??= (async () => { + const runner = createRunner().withEnv({ ...env, RECORD_OUTPUTS: 'true' }); + const spansPromise = runner.collectStreamedSpansUntilSegment('main'); + await runner.start().completed(); + return spansPromise; + })(); + return run; + } + + async function chatSpansUnder(mode: string): Promise { + const spans = await getSpans(); + const modeSpan = spans.find(span => span.name === mode)!; + return spans.filter( + span => span.parent_span_id === modeSpan.span_id && span.attributes[SENTRY_OP]?.value === GEN_AI_CHAT, + ); + } + + test("emits one chat span per turn and none of the SDK's own spans", async () => { + const spans = await getSpans(); + + expect(spans.filter(span => span.attributes[SENTRY_OP]?.value === GEN_AI_CHAT)).toHaveLength( + MODES.flatMap(({ turns }) => turns).length + STATUS_ONLY_MODES.length, + ); + expect(spans.filter(span => span.name.startsWith('anthropic.'))).toEqual([]); + }); - await runner.start().completed(); + test.each(MODES)('nests one chat span per $mode turn under the caller', async ({ mode, turns }) => { + const chatSpans = await chatSpansUnder(mode); - const spans = await spansPromise; - const chatSpans = spans.filter(span => span.attributes[SENTRY_OP]?.value === GEN_AI_CHAT); - expect(chatSpans).toHaveLength(MODES.flatMap(({ turns }) => turns).length); - expect(spans.filter(span => span.name.startsWith('anthropic.'))).toEqual([]); + expect(chatSpans.map(span => span.attributes[GEN_AI_RESPONSE_ID]?.value).sort()).toEqual( + turns.map(turn => `msg_${mode}_${turn}`).sort(), + ); + }); - for (const { mode, streaming, turns, status } of MODES) { - const modeSpan = spans.find(span => span.name === mode)!; - const modeChatSpans = chatSpans.filter(span => span.parent_span_id === modeSpan.span_id); - expect(modeChatSpans, mode).toHaveLength(turns.length); - - if (status) { - expect(modeChatSpans[0]!.status, mode).toBe(status); - continue; - } - - for (const turn of turns) { - const id = `msg_${mode}_${turn}`; - const chatSpan = spanForCall(modeChatSpans, id); - const { input, output, text, toolCall } = TURNS[turn]!; - expect(chatSpan.status, id).toBe('ok'); - expect(chatSpan.attributes[GEN_AI_RESPONSE_STREAMING]?.value ?? false, id).toBe(streaming); - expect(chatSpan.attributes[GEN_AI_USAGE_INPUT_TOKENS]?.value, id).toBe(input); - expect(chatSpan.attributes[GEN_AI_USAGE_OUTPUT_TOKENS]?.value, id).toBe(output); - if (text) { - expect(chatSpan.attributes[GEN_AI_RESPONSE_TEXT]?.value, id).toBe(text); - } - const toolCalls = chatSpan.attributes[GEN_AI_RESPONSE_TOOL_CALLS]?.value as string | undefined; - expect(toolCalls && JSON.parse(toolCalls), id).toEqual( - toolCall - ? [{ type: 'tool_use', id: `toolu_${mode}`, name: 'weather', input: { city: 'Vienna' } }] - : undefined, - ); - } - } + test.each(MODES)('records the tool call of the first $mode turn', async ({ mode, streaming }) => { + const chatSpan = spanForCall(await chatSpansUnder(mode), `msg_${mode}_tools`); + + expect(chatSpan.status).toBe('ok'); + expect(chatSpan.attributes[GEN_AI_RESPONSE_STREAMING]?.value ?? false).toBe(streaming); + expect(chatSpan.attributes[GEN_AI_USAGE_INPUT_TOKENS]?.value).toBe(10); + expect(chatSpan.attributes[GEN_AI_USAGE_OUTPUT_TOKENS]?.value).toBe(20); + expect(JSON.parse(chatSpan.attributes[GEN_AI_RESPONSE_TOOL_CALLS]!.value as string)).toEqual([ + { type: 'tool_use', id: `toolu_${mode}`, name: 'weather', input: { city: 'Vienna' } }, + ]); }); - } + + test.each(FINAL_TURN_MODES)('records the text of the final $mode turn', async ({ mode, streaming }) => { + const chatSpan = spanForCall(await chatSpansUnder(mode), `msg_${mode}_final`); + + expect(chatSpan.status).toBe('ok'); + expect(chatSpan.attributes[GEN_AI_RESPONSE_STREAMING]?.value ?? false).toBe(streaming); + expect(chatSpan.attributes[GEN_AI_USAGE_INPUT_TOKENS]?.value).toBe(30); + expect(chatSpan.attributes[GEN_AI_USAGE_OUTPUT_TOKENS]?.value).toBe(5); + expect(chatSpan.attributes[GEN_AI_RESPONSE_TEXT]?.value).toBe('Sunny.'); + expect(chatSpan.attributes[GEN_AI_RESPONSE_TOOL_CALLS]).toBeUndefined(); + }); + + test.each(STATUS_ONLY_MODES)('ends the $mode chat span with status $status', async ({ mode, status }) => { + const chatSpans = await chatSpansUnder(mode); + + expect(chatSpans).toHaveLength(1); + expect(chatSpans[0]!.status).toBe(status); + }); + }); }); createEsmAndCjsTests(__dirname, 'scenario-session-tool-runner.mjs', 'instrument.mjs', (createRunner, test) => {