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..ea166274dab4 --- /dev/null +++ b/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/scenario-threads.mjs @@ -0,0 +1,207 @@ +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(); + }, + // 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()); + 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..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 @@ -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,97 @@ 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'] }, + { mode: 'thread_stream_signal_reuse', streaming: true, turns: ['tools', 'final'] }, + ]; + 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' }, + ]; + + 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([]); + }); + + 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) => { 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..1828f62e2bde 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,13 +211,52 @@ 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( + () => { + // 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), + ); + 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`). * * - `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/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; 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, + }); + }); +});