From 069d7fb4f0b33942f9405eea6a2b52a8343bc1bf Mon Sep 17 00:00:00 2001 From: Abdelrahman Awad Date: Wed, 15 Jul 2026 14:35:14 -0400 Subject: [PATCH] fix(core): Instrument Anthropic client in place instead of via a deep proxy The Anthropic integration wrapped the client in a deep Proxy that ran every method with the raw object as `this`. That changed the client's observable semantics: internal delegation (`messages.stream()` calling `this.create()`) resolved against the raw object, so it escaped any outer wrapper and behaved differently than an uninstrumented client. Instrument the registered methods in place instead (own properties shadowing the prototype), invoking them with the caller's `this`. Instrumentation is now observationally transparent: private-field access and internal delegation work as they do on a plain client. A re-entrancy guard keyed on the active streaming helper span keeps `stream()`'s internal `create()` call from producing a duplicate span. --- .../core/src/tracing/anthropic-ai/index.ts | 113 ++++++++++++------ 1 file changed, 76 insertions(+), 37 deletions(-) diff --git a/packages/core/src/tracing/anthropic-ai/index.ts b/packages/core/src/tracing/anthropic-ai/index.ts index 4238bcb870db..1249c9f7d2d0 100644 --- a/packages/core/src/tracing/anthropic-ai/index.ts +++ b/packages/core/src/tracing/anthropic-ai/index.ts @@ -3,6 +3,7 @@ import { SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN } from '../../semanticAttributes'; import { SPAN_STATUS_ERROR } from '../../tracing'; import { startSpan, startSpanManual } from '../../tracing/trace'; import type { Span, SpanAttributeValue } from '../../types/span'; +import { getActiveSpan } from '../../utils/spanUtils'; import { GEN_AI_OPERATION_NAME_ATTRIBUTE, GEN_AI_PROMPT_ATTRIBUTE, @@ -22,7 +23,6 @@ import { } from '../ai/gen-ai-attributes'; import type { InstrumentedMethodEntry } from '../ai/utils'; import { - buildMethodPath, resolveAIRecordingOptions, setTokenUsageAttributes, shouldEnableTruncation, @@ -33,6 +33,16 @@ import { instrumentAsyncIterableStream, instrumentMessageStream } from './stream import type { AnthropicAiOptions, AnthropicAiResponse, AnthropicAiStreamingEvent, ContentBlock } from './types'; import { handleResponseError, messagesFromParams, setMessagesAttribute } from './utils'; +// Spans created for streaming helper methods (e.g. `messages.stream()`). The Anthropic SDK +// implements these helpers by internally calling `this.create()`, which we also instrument. +// We track the helper span here so that the internal `create` call does not produce a duplicate +// child span. This relies on the SDK invoking `create` synchronously while the helper span is +// active, which is the case for the currently supported SDK versions. +const STREAMING_HELPER_SPANS = new WeakSet(); + +// 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 +198,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 +221,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)); @@ -236,10 +245,14 @@ function handleStreamingRequest( } else { return startSpanManual(spanConfig, span => { try { + // Mark this as a streaming-helper span so the SDK's internal `create` delegation + // does not create a duplicate child span (see the dedup gate in `instrumentMethod`). + STREAMING_HELPER_SPANS.add(span); + if (options.recordInputs && params) { addPrivateRequestAttributes(span, params, shouldEnableTruncation(options.enableTruncation)); } - const messageStream = target.apply(context, args); + const messageStream = target.apply(invocationThis, args); return instrumentMessageStream(messageStream, span, options.recordOutputs ?? false); } catch (error) { return handleStreamingError(error, span, methodPath); @@ -262,19 +275,35 @@ 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 call is the SDK's internal 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) { + const activeSpan = getActiveSpan(); + if (activeSpan && STREAMING_HELPER_SPANS.has(activeSpan)) { + 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 +324,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 +357,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 +410,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)); }