From 41d2b587f621b41327e1b95d76c3942096d821d9 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Fri, 30 Oct 2026 11:10:12 +0100 Subject: [PATCH 1/5] 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 48cf3bd9bde0..83868bfc4c35 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 @@ -3,11 +3,15 @@ import { GEN_AI_CONVERSATION_ID, 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, GEN_AI_INVOKE_AGENT } from '@sentry/conventions/op'; +import { GEN_AI_CHAT, GEN_AI_EXECUTE_TOOL, GEN_AI_INVOKE_AGENT } 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'; @@ -116,6 +120,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 1adc3413722d..92f18d3b95b2 100644 --- a/packages/server-utils/src/ai/anthropic-ai/streaming.ts +++ b/packages/server-utils/src/ai/anthropic-ai/streaming.ts @@ -287,6 +287,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 811424a2430bade9c2c909ae002277fd65aa259f Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Mon, 2 Nov 2026 15:05:32 +0100 Subject: [PATCH 2/5] 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 83868bfc4c35..19782bbbd66b 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,6 +129,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 0c8c828eda4135eba33b7ecedc76b34d598f3d79 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Mon, 2 Nov 2026 15:52:09 +0100 Subject: [PATCH 3/5] 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 | 120 +++++++++++------- 1 file changed, 72 insertions(+), 48 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 19782bbbd66b..8d7d468765b2 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 @@ -130,61 +130,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, + ); + } - await runner.start().completed(); + test("emits one chat span per turn and none of the SDK's own spans", async () => { + const spans = await getSpans(); - 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.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([]); + }); - 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)('nests one chat span per $mode turn under the caller', async ({ mode, turns }) => { + const chatSpans = await chatSpansUnder(mode); + + expect(chatSpans.map(span => span.attributes[GEN_AI_RESPONSE_ID]?.value).sort()).toEqual( + turns.map(turn => `msg_${mode}_${turn}`).sort(), + ); }); - } + + 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) => { From 7df1252d80f58bdc2867ffc67358c01775a99a96 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Thu, 5 Nov 2026 14:39:24 +0100 Subject: [PATCH 4/5] fix(server-utils): Skip finished thread streams when matching a request A create() sent with the signal of a thread stream whose request failed was still matched to that stream, so its span never ended. A finished stream sends no more requests, so skip it. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../tracing/anthropic/v1/scenario-threads.mjs | 31 +++++++++++++------ .../suites/tracing/anthropic/v1/test.ts | 12 ++++++- .../src/integrations/anthropic.ts | 5 +-- 3 files changed, 36 insertions(+), 12 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 ea166274dab4..be583a717179 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 @@ -78,6 +78,10 @@ function startMockAnthropicServer() { const answersTool = req.body.messages.some( message => Array.isArray(message.content) && message.content.some(block => block.type === 'tool_result'), ); + if (mode === 'thread_stream_failed_signal_reuse' && !answersTool) { + res.status(400).json({ type: 'error', error: { type: 'invalid_request_error', message: 'Bad thread' } }); + return; + } const message = reply(mode, answersTool ? 'final' : 'tools'); if (!req.body.stream) { @@ -86,7 +90,7 @@ function startMockAnthropicServer() { } 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`. + // The body closes after the first tool input delta, without `message_stop`. const events = mode === 'thread_stream_truncated' ? streamEvents(message).slice(0, 3) : streamEvents(message); events.forEach((event, index) => { setTimeout(() => { @@ -115,6 +119,17 @@ const tools = [ ]; const params = { model: MODEL, max_tokens: 100, messages: [{ role: 'user', content: 'Weather in Vienna?' }] }; +async function streamFollowUp(client, signal) { + 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 }, + ); + for await (const _event of followUp) { + // consume + } +} + const MODES = { async runner_stream_eager(client) { for await (const stream of client.beta.messages.toolRunner({ @@ -152,14 +167,12 @@ const MODES = { 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 - } + await streamFollowUp(client, stream.controller.signal); + }, + async thread_stream_failed_signal_reuse(client) { + const stream = client.beta.messages.thread({ model: MODEL, max_tokens: 100 }).stream(params.messages); + await stream.finalMessage().catch(() => {}); + await streamFollowUp(client, stream.controller.signal); }, async thread_stream_abort(client) { const stream = client.beta.messages.thread({ model: MODEL, max_tokens: 100 }).stream(params.messages); 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 8d7d468765b2..a345e3d932ca 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 @@ -166,7 +166,8 @@ describe('Anthropic integration (1.x)', () => { 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, + // Plus the failed request and the follow-up of `thread_stream_failed_signal_reuse`. + MODES.flatMap(({ turns }) => turns).length + STATUS_ONLY_MODES.length + 2, ); expect(spans.filter(span => span.name.startsWith('anthropic.'))).toEqual([]); }); @@ -208,6 +209,15 @@ describe('Anthropic integration (1.x)', () => { expect(chatSpans).toHaveLength(1); expect(chatSpans[0]!.status).toBe(status); }); + + test("ends the span of a request sent with a failed stream's signal", async () => { + const chatSpans = await chatSpansUnder('thread_stream_failed_signal_reuse'); + const followUp = spanForCall(chatSpans, 'msg_thread_stream_failed_signal_reuse_final'); + + expect(chatSpans.map(span => span.status).sort()).toEqual(['error', 'ok']); + expect(followUp.attributes[GEN_AI_USAGE_OUTPUT_TOKENS]?.value).toBe(5); + expect(followUp.attributes[GEN_AI_RESPONSE_TEXT]?.value).toBe('Sunny.'); + }); }); }); diff --git a/packages/server-utils/src/integrations/anthropic.ts b/packages/server-utils/src/integrations/anthropic.ts index 1828f62e2bde..a7f80cb04ab7 100644 --- a/packages/server-utils/src/integrations/anthropic.ts +++ b/packages/server-utils/src/integrations/anthropic.ts @@ -211,7 +211,7 @@ function isMessageStream(value: unknown): value is MessageStreamEmitter { return !!value && typeof (value as MessageStreamEmitter).on === 'function'; } -type ThreadStream = MessageStreamEmitter & { controller?: { signal?: AbortSignal } }; +type ThreadStream = MessageStreamEmitter & { ended?: boolean; controller?: { signal?: AbortSignal } }; /** * The streams of threads and tool runners on 1.2+, by the signal they send each `create()` call with. Such a @@ -232,7 +232,8 @@ function wrapThreadStreamRequest( 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') { + // A finished stream sends no more requests, so a later call with its signal is the caller's own. + if (!stream || stream.ended || typeof request?._responseProps !== 'function') { return false; } From 7c140a0fb4c7ee250abfa566e8121d2074dc0611 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Tue, 10 Nov 2026 13:28:35 +0100 Subject: [PATCH 5/5] fix(server-utils): End aborted Anthropic thread streams as cancelled An aborted thread or tool-runner stream ended its chat span ok with nothing to show it was cut short. Mark it cancelled, as Google GenAI does since #26023, both mid-stream and before the response arrives. Also pin the error span of an attempt the thread recovered from, which is deliberate. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../tracing/anthropic/v1/scenario-threads.mjs | 20 +++++++- .../suites/tracing/anthropic/v1/test.ts | 18 ++++++- .../src/ai/anthropic-ai/streaming.ts | 3 ++ .../src/integrations/anthropic.ts | 10 +++- .../ai/lib/tracing/anthropic-ai-usage.test.ts | 51 +++++++++++++------ 5 files changed, 82 insertions(+), 20 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 be583a717179..c2286913e5ff 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 @@ -65,6 +65,8 @@ function streamEvents({ id, content, stop_reason, usage }) { return events; } +let recoveredReplies = 0; + function startMockAnthropicServer() { const app = express(); app.use(express.json()); @@ -78,11 +80,22 @@ function startMockAnthropicServer() { const answersTool = req.body.messages.some( message => Array.isArray(message.content) && message.content.some(block => block.type === 'tool_result'), ); + // The server lost the thread, so the SDK replays the turn that continues it. + if (mode === 'thread_stream_recovered' && req.body.thread?.type === 'continue') { + res.status(404).json({ + type: 'error', + error: { type: 'not_found_error', message: 'Thread not found', details: { error_code: 'thread_not_found' } }, + }); + return; + } if (mode === 'thread_stream_failed_signal_reuse' && !answersTool) { res.status(400).json({ type: 'error', error: { type: 'invalid_request_error', message: 'Bad thread' } }); return; } - const message = reply(mode, answersTool ? 'final' : 'tools'); + const message = + mode === 'thread_stream_recovered' + ? { ...reply(mode, 'final'), id: `msg_${mode}_${(recoveredReplies += 1)}` } + : reply(mode, answersTool ? 'final' : 'tools'); if (!req.body.stream) { res.json({ type: 'message', role: 'assistant', model: MODEL, stop_sequence: null, ...message }); @@ -191,6 +204,11 @@ const MODES = { .finalMessage() .catch(() => {}); }, + async thread_stream_recovered(client) { + const thread = client.beta.messages.thread({ model: MODEL, max_tokens: 100 }); + await thread.stream(params.messages).finalMessage(); + await thread.stream([{ role: 'user', content: 'And tomorrow?' }]).finalMessage(); + }, async thread_stream_error(client) { await client.beta.messages .thread({ model: MODEL, max_tokens: 100 }) 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 a345e3d932ca..afda0052828f 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 @@ -166,8 +166,9 @@ describe('Anthropic integration (1.x)', () => { const spans = await getSpans(); expect(spans.filter(span => span.attributes[SENTRY_OP]?.value === GEN_AI_CHAT)).toHaveLength( - // Plus the failed request and the follow-up of `thread_stream_failed_signal_reuse`. - MODES.flatMap(({ turns }) => turns).length + STATUS_ONLY_MODES.length + 2, + // Plus the failed request and the follow-up of `thread_stream_failed_signal_reuse`, and the three requests + // of `thread_stream_recovered`. + MODES.flatMap(({ turns }) => turns).length + STATUS_ONLY_MODES.length + 2 + 3, ); expect(spans.filter(span => span.name.startsWith('anthropic.'))).toEqual([]); }); @@ -210,6 +211,19 @@ describe('Anthropic integration (1.x)', () => { expect(chatSpans[0]!.status).toBe(status); }); + // Deliberate. The rejected attempt was a real request that failed, like an http.client span that a retry + // recovers from. Hiding it would mean copying the SDK's private recovery rules: they replay only some error codes, + // only on a `continue`, and may change in any release, so a copy would silently hide real errors. + test('keeps an error span for the attempt of a turn the thread recovered from', async () => { + const chatSpans = await chatSpansUnder('thread_stream_recovered'); + + expect(chatSpans.map(span => [span.attributes[GEN_AI_RESPONSE_ID]?.value, span.status])).toEqual([ + ['msg_thread_stream_recovered_1', 'ok'], + [undefined, 'error'], + ['msg_thread_stream_recovered_2', 'ok'], + ]); + }); + test("ends the span of a request sent with a failed stream's signal", async () => { const chatSpans = await chatSpansUnder('thread_stream_failed_signal_reuse'); const followUp = spanForCall(chatSpans, 'msg_thread_stream_failed_signal_reuse_final'); diff --git a/packages/server-utils/src/ai/anthropic-ai/streaming.ts b/packages/server-utils/src/ai/anthropic-ai/streaming.ts index 92f18d3b95b2..9d0b1a3def11 100644 --- a/packages/server-utils/src/ai/anthropic-ai/streaming.ts +++ b/packages/server-utils/src/ai/anthropic-ai/streaming.ts @@ -290,6 +290,9 @@ export function instrumentMessageStream // 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', () => { + if (span.isRecording() && (stream as { aborted?: boolean }).aborted) { + span.setStatus({ code: SPAN_STATUS_ERROR, message: 'cancelled' }); + } endStreamSpan(span, state, recordOutputs); }); diff --git a/packages/server-utils/src/integrations/anthropic.ts b/packages/server-utils/src/integrations/anthropic.ts index a7f80cb04ab7..ff9cc0c7440f 100644 --- a/packages/server-utils/src/integrations/anthropic.ts +++ b/packages/server-utils/src/integrations/anthropic.ts @@ -245,8 +245,14 @@ function wrapThreadStreamRequest( threadStreams.delete(signal); instrumentMessageStream(stream, span, recordOutputs); }, - // A caller abort is not a failed request. - error => end(signal.aborted ? undefined : error), + error => { + // A caller abort is not a failed request. The stream still emits 'end', which ends the span as cancelled. + if (signal.aborted) { + instrumentMessageStream(stream, span, recordOutputs); + } else { + end(error); + } + }, ); return true; } 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 109537394426..78569bc446ec 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 @@ -11,6 +11,7 @@ import { import { EventEmitter } from 'node:events'; import { describe, expect, it, vi } from 'vitest'; import type { Span } from '@sentry/core'; +import { SPAN_STATUS_ERROR } from '@sentry/core'; import { addResponseAttributes } from '../../../../src/ai/anthropic-ai'; import { instrumentAsyncIterableStream, instrumentMessageStream } from '../../../../src/ai/anthropic-ai/streaming'; import type { AnthropicAiStreamingEvent } from '../../../../src/ai/anthropic-ai/types'; @@ -273,28 +274,34 @@ 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 messageStart = { + 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 }, + }, + }; + + it('ends an aborted stream once, as cancelled, with the usage streamed so far', () => { const { span, attributes } = createMockSpan(); + const setStatus = vi.fn(); const endSpy = vi.spyOn(span, 'end'); - const stream = instrumentMessageStream(new EventEmitter(), span, false); + Object.assign(span, { setStatus }); + const stream = instrumentMessageStream(Object.assign(new EventEmitter(), { aborted: true }), 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('streamEvent', messageStart); stream.emit('abort', new Error('Request was aborted.')); stream.emit('end'); stream.emit('end'); + expect(setStatus).toHaveBeenCalledTimes(1); + expect(setStatus).toHaveBeenCalledWith({ code: SPAN_STATUS_ERROR, message: 'cancelled' }); expect(endSpy).toHaveBeenCalledTimes(1); expect(attributes).toEqual({ [GEN_AI_RESPONSE_STREAMING]: true, @@ -304,4 +311,18 @@ describe('instrumentMessageStream', () => { [GEN_AI_USAGE_TOTAL_TOKENS]: 42, }); }); + + it('leaves the status alone when the body closes early without an abort', () => { + const { span } = createMockSpan(); + const setStatus = vi.fn(); + const endSpy = vi.spyOn(span, 'end'); + Object.assign(span, { setStatus }); + const stream = instrumentMessageStream(Object.assign(new EventEmitter(), { aborted: false }), span, false); + + stream.emit('streamEvent', messageStart); + stream.emit('end'); + + expect(setStatus).not.toHaveBeenCalled(); + expect(endSpy).toHaveBeenCalledTimes(1); + }); });