getsentry · RulaKhaled · Sep 18, 2026 · Sep 18, 2026 · Sep 18, 2026 · Sep 18, 2026
@@ -0,0 +1,10 @@
+import * as Sentry from '@sentry/node';
+import { loggingTransport } from '@sentry-internal/node-integration-tests';
+
+Sentry.init({
+  dsn: 'https://public@dsn.ingest.sentry.io/1337',
+  release: '1.0',
+  tracesSampleRate: 1.0,
+  dataCollection: { genAI: { inputs: true, outputs: true } },
+  transport: loggingTransport,
+});
@@ -0,0 +1,97 @@
+import { Readable } from 'node:stream';
+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_node_body',
+          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: 'Node ' } },
+      { type: 'content_block_delta', index: 0, delta: { type: 'text_delta', text: 'body!' } },
+      { 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);
+    });
+  });
+}
+
+// Stands in for the node-fetch/undici-compat shims callers pass as `fetch`, whose responses carry a
+// Node `Readable` body the SSE body wrapper can't wrap.
+async function fetchWithNodeStreamBody(url, init) {
+  const response = await fetch(url, init);
+  Object.defineProperty(response, 'body', {
+    value: Readable.fromWeb(response.body),
+    configurable: true,
+  });
+  return response;
+}
+
+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`,
+      fetch: fetchWithNodeStreamBody,
+    });
+
+    const stream = await client.messages.create({
+      model: 'claude-3-haiku-20240307',
+      messages: [{ role: 'user', content: 'Stream this please' }],
+      stream: true,
+    });
+
+    for await (const _ of stream) {
+      void _;
+    }
+  });
+
+  await Sentry.flush(2000);
+
+  server.close();
+}
+
+run();
@@ -0,0 +1,127 @@
+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_raw_body',
+          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: 'Raw ' } },
+      { type: 'content_block_delta', index: 0, delta: { type: 'text_delta', text: 'body!' } },
+      { 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 params = {
+      model: 'claude-3-haiku-20240307',
+      messages: [{ role: 'user', content: 'Stream this please' }],
+      stream: true,
+    };
+
+    // 1) Drain the raw `Response` body, never touching the SDK `Stream`
+    const response = await client.messages.create({ ...params }).asResponse();
+
+    // Wrapping the body must not disturb it, or `text()`, `arrayBuffer()` and `clone()` would throw
+    // on a response the caller has not read yet.
+    if (response.bodyUsed) {
+      throw new Error('raw Response body was consumed before the caller read it');
+    }
+
+    for await (const _ of response.body) {
+      void _;
+    }
+
+    // 2) Drain the SDK `Stream`, so both consumption styles are covered in one run
+    const stream = await client.messages.create({ ...params });
+    for await (const _ of stream) {
+      void _;
+    }
+
+    // 3) Clone first, then drain. `clone()` tees the response's internal body and swaps in one branch,
+    // so the wrapper has to re-read the body rather than hold on to the stream it was handed.
+    const cloned = await client.messages.create({ ...params }).asResponse();
+    const copy = cloned.clone();
+    for await (const _ of cloned.body) {
+      void _;
+    }
+    void copy;
+
+    // 4) Read the body as text, which never touches the `body` property at all
+    const asText = await client.messages.create({ ...params }).asResponse();
+    const text = await asText.text();
+    if (!text.includes('message_stop')) {
+      throw new Error('raw Response text did not contain the streamed frames');
+    }
+
+    // 5) Read the body through a BYOB reader, which only a byte stream supports
+    const byob = await client.messages.create({ ...params }).asResponse();
+    const reader = byob.body.getReader({ mode: 'byob' });
+    let buffer = new ArrayBuffer(1024);
+    for (;;) {
+      const { done, value } = await reader.read(new Uint8Array(buffer, 0, 1024));
+      if (done) {
+        break;
+      }
+      buffer = value.buffer;
+    }
+  });
+
+  await Sentry.flush(2000);
+
+  server.close();
+}
+
+run();
@@ -265,6 +265,61 @@ describe('Anthropic integration', () => {
     });
   });
 
+  createEsmAndCjsTests(__dirname, 'scenario-stream-raw-body.mjs', 'instrument-raw-body.mjs', (createRunner, test) => {
+    test('ends the span when a stream is drained through the raw Response body', async () => {
+      await createRunner()
+        .unordered()
+        .expect({
+          span: container => {
+            const genAiSpans = container.items.filter(span => span.attributes['sentry.op']?.value === 'gen_ai.chat');
+            // One call per way of draining the stream: the raw `.asResponse()` body, the SDK `Stream`,
+            // a cloned response, `text()`, and a BYOB reader. Every one of them must end its span with
+            // the response attributes accumulated off the SSE frames.
+            expect(genAiSpans).toHaveLength(5);
+            for (const span of genAiSpans) {
+              expect(span.name).toBe('chat claude-3-haiku-20240307');
+              expect(span.status).toBe('ok');
+              expect(span.attributes[GEN_AI_RESPONSE_STREAMING].value).toBe(true);
+              expect(span.attributes[GEN_AI_RESPONSE_ID].value).toBe('msg_raw_body');
+              expect(span.attributes[GEN_AI_RESPONSE_MODEL].value).toBe('claude-3-haiku-20240307');
+              expect(span.attributes[GEN_AI_RESPONSE_FINISH_REASONS].value).toBe('["end_turn"]');
+              expect(span.attributes[GEN_AI_RESPONSE_TEXT].value).toBe('Raw body!');
+              expect(span.attributes[GEN_AI_USAGE_INPUT_TOKENS].value).toBe(10);
+              expect(span.attributes[GEN_AI_USAGE_OUTPUT_TOKENS].value).toBe(15);
+              expect(span.attributes[GEN_AI_USAGE_TOTAL_TOKENS].value).toBe(25);
+            }
+          },
+        })
+        .start()
+        .completed();
+    });
+  });
+
+  createEsmAndCjsTests(__dirname, 'scenario-stream-node-body.mjs', 'instrument-raw-body.mjs', (createRunner, test) => {
+    test('ends the span when the response body is not a web ReadableStream', async () => {
+      await createRunner()
+        .unordered()
+        .expect({
+          span: container => {
+            const genAiSpan = container.items.find(span => span.attributes['sentry.op']?.value === 'gen_ai.chat');
+            // The body wrapper has nothing to hold on to here, so the SDK `Stream`'s iterator has to
+            // carry the span instead — otherwise it would end with request attributes only.
+            expect(genAiSpan).toBeDefined();
+            expect(genAiSpan!.status).toBe('ok');
+            expect(genAiSpan!.attributes[GEN_AI_RESPONSE_STREAMING].value).toBe(true);
+            expect(genAiSpan!.attributes[GEN_AI_RESPONSE_ID].value).toBe('msg_node_body');
+            expect(genAiSpan!.attributes[GEN_AI_RESPONSE_FINISH_REASONS].value).toBe('["end_turn"]');
+            expect(genAiSpan!.attributes[GEN_AI_RESPONSE_TEXT].value).toBe('Node body!');
+            expect(genAiSpan!.attributes[GEN_AI_USAGE_INPUT_TOKENS].value).toBe(10);
+            expect(genAiSpan!.attributes[GEN_AI_USAGE_OUTPUT_TOKENS].value).toBe(15);
+            expect(genAiSpan!.attributes[GEN_AI_USAGE_TOTAL_TOKENS].value).toBe(25);
+          },
+        })
+        .start()
+        .completed();
+    });
+  });
+
   createEsmAndCjsTests(__dirname, 'scenario-stream.mjs', 'instrument-with-pii.mjs', (createRunner, test) => {
     test('streams record response text when PII true', async () => {
       await createRunner()