fix(core): Instrument Anthropic client in place instead of via a deep proxy

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();

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();

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 () => {