diff --git a/dev-packages/node-integration-tests/suites/tracing/anthropic/scenario-outer-wrapper.mjs b/dev-packages/node-integration-tests/suites/tracing/anthropic/scenario-outer-wrapper.mjs new file mode 100644 index 000000000000..2b7dc3737a7e --- /dev/null +++ b/dev-packages/node-integration-tests/suites/tracing/anthropic/scenario-outer-wrapper.mjs @@ -0,0 +1,110 @@ +import Anthropic from '@anthropic-ai/sdk'; +import * as Sentry from '@sentry/node'; +import express from 'express'; + +function startMockAnthropicServer() { + const app = express(); + app.use(express.json()); + + app.post('/anthropic/v1/messages', (req, res) => { + res.writeHead(200, { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + }); + + const model = req.body.model; + const events = [ + { + type: 'message_start', + message: { + id: 'msg_stream_1', + type: 'message', + role: 'assistant', + model, + content: [], + usage: { input_tokens: 10 }, + }, + }, + { type: 'content_block_start', index: 0, content_block: { type: 'text', text: '' } }, + { type: 'content_block_delta', index: 0, delta: { type: 'text_delta', text: 'Hello ' } }, + { type: 'content_block_delta', index: 0, delta: { type: 'text_delta', text: 'from ' } }, + { type: 'content_block_delta', index: 0, delta: { type: 'text_delta', text: 'stream!' } }, + { type: 'content_block_stop', index: 0 }, + { + type: 'message_delta', + delta: { stop_reason: 'end_turn', stop_sequence: null }, + usage: { output_tokens: 15 }, + }, + { type: 'message_stop' }, + ]; + + events.forEach((event, index) => { + setTimeout(() => { + res.write(`event: ${event.type}\n`); + res.write(`data: ${JSON.stringify(event)}\n\n`); + if (index === events.length - 1) { + res.end(); + } + }, index * 10); + }); + }); + + return new Promise(resolve => { + const server = app.listen(0, () => { + resolve(server); + }); + }); +} + +// Minimal stand-in for a third-party library that also instruments the client by wrapping +// `messages.create` (e.g. Braintrust's `wrapAnthropic`). Our instrumentation must not hide the +// client's own `create` from it when the SDK's `messages.stream()` helper delegates internally. +function wrapClient(client) { + return new Proxy(client, { + get(target, prop, receiver) { + if (prop !== 'messages') { + return Reflect.get(target, prop, receiver); + } + return new Proxy(Reflect.get(target, prop, receiver), { + get(messages, messagesProp, messagesReceiver) { + if (messagesProp !== 'create') { + return Reflect.get(messages, messagesProp, messagesReceiver); + } + const originalCreate = Reflect.get(messages, messagesProp, messagesReceiver); + return function (...args) { + Sentry.captureMessage('third-party wrapper observed messages.create'); + return originalCreate.apply(this, args); + }; + }, + }); + }, + }); +} + +async function run() { + const server = await startMockAnthropicServer(); + + await Sentry.startSpan({ op: 'function', name: 'main' }, async () => { + const client = wrapClient( + new Anthropic({ + apiKey: 'mock-api-key', + baseURL: `http://localhost:${server.address().port}/anthropic`, + }), + ); + + const stream = client.messages.stream({ + model: 'claude-3-haiku-20240307', + messages: [{ role: 'user', content: 'Stream this please' }], + }); + for await (const _ of stream) { + void _; + } + }); + + await Sentry.flush(2000); + + server.close(); +} + +run(); diff --git a/dev-packages/node-integration-tests/suites/tracing/anthropic/scenario-stream-nested-create.mjs b/dev-packages/node-integration-tests/suites/tracing/anthropic/scenario-stream-nested-create.mjs new file mode 100644 index 000000000000..8f335c4ce794 --- /dev/null +++ b/dev-packages/node-integration-tests/suites/tracing/anthropic/scenario-stream-nested-create.mjs @@ -0,0 +1,110 @@ +import Anthropic from '@anthropic-ai/sdk'; +import * as Sentry from '@sentry/node'; +import express from 'express'; + +function startMockAnthropicServer() { + const app = express(); + app.use(express.json()); + + app.post('/anthropic/v1/messages', (req, res) => { + const model = req.body.model; + + // Non-streaming request (the nested call made from the stream event handler). + if (!req.body.stream) { + res.send({ + id: 'msg_nested', + type: 'message', + model, + role: 'assistant', + content: [{ type: 'text', text: 'nested reply' }], + stop_reason: 'end_turn', + usage: { input_tokens: 3, output_tokens: 4 }, + }); + return; + } + + res.writeHead(200, { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + }); + const events = [ + { + type: 'message_start', + message: { + id: 'msg_stream_1', + type: 'message', + role: 'assistant', + model, + content: [], + usage: { input_tokens: 10 }, + }, + }, + { type: 'content_block_start', index: 0, content_block: { type: 'text', text: '' } }, + { type: 'content_block_delta', index: 0, delta: { type: 'text_delta', text: 'Hello!' } }, + { type: 'content_block_stop', index: 0 }, + { type: 'message_delta', delta: { stop_reason: 'end_turn', stop_sequence: null }, usage: { output_tokens: 15 } }, + { type: 'message_stop' }, + ]; + events.forEach((event, index) => { + setTimeout(() => { + res.write(`event: ${event.type}\n`); + res.write(`data: ${JSON.stringify(event)}\n\n`); + if (index === events.length - 1) { + res.end(); + } + }, index * 10); + }); + }); + + return new Promise(resolve => { + const server = app.listen(0, () => { + resolve(server); + }); + }); +} + +async function run() { + const server = await startMockAnthropicServer(); + + await Sentry.startSpan({ op: 'function', name: 'main' }, async () => { + const client = new Anthropic({ + apiKey: 'mock-api-key', + baseURL: `http://localhost:${server.address().port}/anthropic`, + }); + + const stream = client.messages.stream({ + model: 'claude-3-haiku-20240307', + messages: [{ role: 'user', content: 'Stream this please' }], + }); + + // Fire a separate, non-streaming request from a stream event handler. This runs inside the + // stream's async continuation, so the streaming-helper span is still the active span here. + // It must still be traced (the dedup must only suppress the helper's own internal `create`). + let resolveNested; + const nested = new Promise(resolve => (resolveNested = resolve)); + let fired = false; + stream.on('streamEvent', () => { + if (fired) return; + fired = true; + client.messages + .create({ + model: 'claude-3-haiku-20240307', + messages: [{ role: 'user', content: 'Nested call from handler' }], + max_tokens: 10, + }) + .then(resolveNested, resolveNested); + }); + + for await (const _ of stream) { + void _; + } + await nested; + }); + + await Sentry.flush(2000); + + server.close(); +} + +run(); diff --git a/dev-packages/node-integration-tests/suites/tracing/anthropic/test.ts b/dev-packages/node-integration-tests/suites/tracing/anthropic/test.ts index e8129890c978..5546aa8abd66 100644 --- a/dev-packages/node-integration-tests/suites/tracing/anthropic/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/anthropic/test.ts @@ -396,6 +396,48 @@ describe('Anthropic integration', () => { }); }); + // Instrumenting the client must not hide its own methods from an outer wrapper (e.g. another + // library instrumenting the same client). `messages.stream()` delegates to `create` through + // `this`, so if our instrumentation rebinds `this` away from the client, that internal call is + // never observed by the wrapper. Regression test for the deep-proxy `this` rebinding. + createEsmAndCjsTests(__dirname, 'scenario-outer-wrapper.mjs', 'instrument.mjs', (createRunner, test) => { + test('does not hide the client methods from an outer wrapper when stream() delegates internally', async () => { + await createRunner() + .expect({ event: { message: 'third-party wrapper observed messages.create' } }) + .start() + .completed(); + }); + }); + + // The stream dedup must only suppress the helper's own internal `create` delegation, not a + // separate `create` a user makes from a stream event handler (which runs while the streaming + // helper span is still the active span). Regression test for over-suppression. + createEsmAndCjsTests(__dirname, 'scenario-stream-nested-create.mjs', 'instrument.mjs', (createRunner, test) => { + test('traces a create() invoked from a stream event handler (dedup does not over-suppress)', async () => { + await createRunner() + .ignore('event') + .expect({ transaction: { transaction: 'main' } }) + .expect({ + span: container => { + const nestedSpan = container.items.find( + span => span.attributes[GEN_AI_RESPONSE_ID_ATTRIBUTE]?.value === 'msg_nested', + ); + expect(nestedSpan).toBeDefined(); + expect(nestedSpan.attributes['sentry.op'].value).toBe('gen_ai.chat'); + + // The helper's own internal `create` delegation must be deduped: exactly one span + // for the streamed response, not a duplicate child span. + const streamingSpans = container.items.filter( + span => span.attributes[GEN_AI_RESPONSE_ID_ATTRIBUTE]?.value === 'msg_stream_1', + ); + expect(streamingSpans).toHaveLength(1); + }, + }) + .start() + .completed(); + }); + }); + // Non-streaming tool calls + available tools (PII true) createEsmAndCjsTests(__dirname, 'scenario-tools.mjs', 'instrument-with-pii.mjs', (createRunner, test) => { test('non-streaming sets available tools and tool calls with PII', async () => { diff --git a/packages/core/src/tracing/anthropic-ai/index.ts b/packages/core/src/tracing/anthropic-ai/index.ts index 4238bcb870db..64cd105905bc 100644 --- a/packages/core/src/tracing/anthropic-ai/index.ts +++ b/packages/core/src/tracing/anthropic-ai/index.ts @@ -22,7 +22,6 @@ import { } from '../ai/gen-ai-attributes'; import type { InstrumentedMethodEntry } from '../ai/utils'; import { - buildMethodPath, resolveAIRecordingOptions, setTokenUsageAttributes, shouldEnableTruncation, @@ -33,6 +32,15 @@ import { instrumentAsyncIterableStream, instrumentMessageStream } from './stream import type { AnthropicAiOptions, AnthropicAiResponse, AnthropicAiStreamingEvent, ContentBlock } from './types'; import { handleResponseError, messagesFromParams, setMessagesAttribute } from './utils'; +// Set only while a streaming helper (e.g. `messages.stream()`) synchronously delegates to the +// underlying `create`. The SDK invokes that internal `create` synchronously, so a plain flag +// suppresses exactly the duplicate delegation and nothing else: a `create` made later from a +// stream event handler runs in a separate async continuation with the flag already cleared. +let suppressDelegatedCreate = false; + +// Methods that have already been wrapped, so instrumenting the same client twice is a no-op. +const INSTRUMENTED_METHODS = new WeakSet(); + /** * Extract request attributes from method arguments */ @@ -188,9 +196,8 @@ function handleStreamingError(error: unknown, span: Span, methodPath: string): n * Handle streaming cases with common logic */ function handleStreamingRequest( - originalMethod: (...args: T) => R | Promise, target: (...args: T) => R | Promise, - context: unknown, + invocationThis: unknown, args: T, requestAttributes: Record, operationName: string, @@ -212,7 +219,7 @@ function handleStreamingRequest( let originalResult!: Promise; const instrumentedPromise = startSpanManual(spanConfig, (span: Span) => { - originalResult = originalMethod.apply(context, args) as Promise; + originalResult = target.apply(invocationThis, args) as Promise; if (options.recordInputs && params) { addPrivateRequestAttributes(span, params, shouldEnableTruncation(options.enableTruncation)); @@ -239,9 +246,14 @@ function handleStreamingRequest( if (options.recordInputs && params) { addPrivateRequestAttributes(span, params, shouldEnableTruncation(options.enableTruncation)); } - const messageStream = target.apply(context, args); + // The helper synchronously delegates to `create`; suppress that one internal call so it + // does not produce a duplicate child span (see the dedup gate in `instrumentMethod`). + suppressDelegatedCreate = true; + const messageStream = target.apply(invocationThis, args); + suppressDelegatedCreate = false; return instrumentMessageStream(messageStream, span, options.recordOutputs ?? false); } catch (error) { + suppressDelegatedCreate = false; return handleStreamingError(error, span, methodPath); } }); @@ -262,19 +274,32 @@ function instrumentMethod( ): (...args: T) => R | Promise { return new Proxy(originalMethod, { apply(target, thisArg, args: T): R | Promise { + // Preserve the caller's `this` so instrumentation stays transparent: the SDK's methods + // rely on private fields bound to the real instance, and internal delegation (e.g. + // `messages.stream()` calling `this.create()`) must resolve against the same object it + // would on an uninstrumented client. Fall back to the wrap-time owner for unbound calls. + const invocationThis = thisArg !== undefined ? thisArg : context; + + const isStreamingMethod = instrumentedMethod.streaming === true; + + // If this is the SDK's internal `create` delegation from a streaming helper (e.g. + // `messages.stream()` invoking `this.create()`), skip instrumentation: the helper span + // already represents this operation, so a second span would be a duplicate. + if (!isStreamingMethod && suppressDelegatedCreate) { + return target.apply(invocationThis, args); + } + const operationName = instrumentedMethod.operation || 'unknown'; const requestAttributes = extractRequestAttributes(args, methodPath, operationName); const model = requestAttributes[GEN_AI_REQUEST_MODEL_ATTRIBUTE] ?? 'unknown'; const params = typeof args[0] === 'object' ? (args[0] as Record) : undefined; const isStreamRequested = Boolean(params?.stream); - const isStreamingMethod = instrumentedMethod.streaming === true; if (isStreamRequested || isStreamingMethod) { return handleStreamingRequest( - originalMethod, target, - context, + invocationThis, args, requestAttributes, operationName, @@ -295,7 +320,7 @@ function instrumentMethod( attributes: requestAttributes as Record, }, span => { - originalResult = target.apply(context, args) as Promise; + originalResult = target.apply(invocationThis, args) as Promise; if (options.recordInputs && params) { addPrivateRequestAttributes(span, params, shouldEnableTruncation(options.enableTruncation)); @@ -328,37 +353,47 @@ function instrumentMethod( } /** - * Create a deep proxy for Anthropic AI client instrumentation + * Instrument the Anthropic client's methods in place. + * + * We deliberately do not wrap the client in a Proxy. The Anthropic SDK relies on private class + * fields (`this.#field`), which are invisible to a Proxy and throw if a method runs with a + * proxied `this`. Wrapping the registered methods in place (as own properties shadowing the + * prototype) keeps `this` bound to the real instance, so instrumentation stays observationally + * transparent: internal delegation (e.g. `messages.stream()` calling `this.create()`) and + * `instanceof` checks behave exactly as on an uninstrumented client, and non-instrumented + * methods are left untouched. */ -function createDeepProxy(target: T, currentPath = '', options: AnthropicAiOptions): T { - return new Proxy(target, { - get(obj: object, prop: string): unknown { - const value = (obj as Record)[prop]; - const methodPath = buildMethodPath(currentPath, String(prop)); - - const instrumentedMethod = ANTHROPIC_METHOD_REGISTRY[methodPath as keyof typeof ANTHROPIC_METHOD_REGISTRY]; - if (typeof value === 'function' && instrumentedMethod) { - return instrumentMethod( - value as (...args: unknown[]) => unknown | Promise, - methodPath, - instrumentedMethod, - obj, - options, - ); - } +function instrumentClientInPlace(client: T, options: AnthropicAiOptions): T { + for (const methodPath of Object.keys(ANTHROPIC_METHOD_REGISTRY) as Array) { + const segments = methodPath.split('.'); + const methodName = segments.pop() as string; + + let owner = client as Record | undefined; + for (const segment of segments) { + owner = owner?.[segment] as Record | undefined; + } - if (typeof value === 'function') { - // Bind non-instrumented functions to preserve the original `this` context, - return value.bind(obj); - } + if (!owner || typeof owner[methodName] !== 'function') { + continue; + } - if (value && typeof value === 'object') { - return createDeepProxy(value, methodPath, options); - } + const originalMethod = owner[methodName] as (...args: unknown[]) => unknown; + if (INSTRUMENTED_METHODS.has(originalMethod)) { + continue; + } - return value; - }, - }) as T; + const instrumented = instrumentMethod( + originalMethod, + methodPath, + ANTHROPIC_METHOD_REGISTRY[methodPath], + owner, + options, + ); + INSTRUMENTED_METHODS.add(instrumented); + owner[methodName] = instrumented; + } + + return client; } /** @@ -371,5 +406,5 @@ function createDeepProxy(target: T, currentPath = '', options: * @returns The instrumented client with the same type as the input */ export function instrumentAnthropicAiClient(anthropicAiClient: T, options?: AnthropicAiOptions): T { - return createDeepProxy(anthropicAiClient, '', resolveAIRecordingOptions(options)); + return instrumentClientInPlace(anthropicAiClient, resolveAIRecordingOptions(options)); }