fix(server-utils): Cover every way an Anthropic SSE body gets drained

Handing the span to the SSE body wrapper left several paths uncovered.

A body that is not a web ReadableStream, which is what an injected
node-fetch or undici shim hands back, could not be wrapped at all, so the
span ended the moment create() resolved with request attributes only.
Patching the SDK Stream's iterator now serves as the fallback for those.

clone() tees the response's internal body and swaps in one branch, which
left the stream the wrapper had captured locked, so the next read threw.
The wrapper now resolves the source through the prototype getter on every
read instead of holding on to the stream it was handed.

text(), json() and arrayBuffer() read the internal body and never touch
the property we shadow, so the span never ended. Those are wrapped too,
and text() and arrayBuffer() feed their result through the accumulator so
the response attributes survive.

Three smaller ones: recordOutputs is resolved per call rather than once at
subscribe time, since subscribeToSseStream runs once per process and a
later client can carry different options. The frame parser's try now
covers a single frame, so one unparsable line no longer costs us the rest
of its chunk, where message_delta and message_stop ride. An unsampled span
skips the wrap entirely, and settle() flushes a trailing frame left
without a newline.

The pass-through is a byte stream now, so getReader({ mode: 'byob' })
keeps working on a response that supported it before.

instrumentRawSseBody moved to its own file to stay under the line cap.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: RulaKhaled

dev-packages/node-integration-tests/suites/tracing/anthropic/scenario-stream-node-body.mjs

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

dev-packages/node-integration-tests/suites/tracing/anthropic/scenario-stream-raw-body.mjs

@@ -89,6 +89,34 @@ async function run() {
     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);

dev-packages/node-integration-tests/suites/tracing/anthropic/test.ts

@@ -272,9 +272,10 @@ describe('Anthropic integration', () => {
         .expect({
           span: container => {
             const genAiSpans = container.items.filter(span => span.attributes['sentry.op']?.value === 'gen_ai.chat');
-            // Two calls, one drained via `.asResponse()` and one via the SDK `Stream`. Both must end,
-            // and both must carry the response attributes accumulated off the SSE frames.
-            expect(genAiSpans).toHaveLength(2);
+            // 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');
@@ -294,6 +295,31 @@ describe('Anthropic integration', () => {
     });
   });
 
+  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()

packages/server-utils/src/ai/anthropic-ai/sse-body.ts

@@ -0,0 +1,202 @@
+import type { Span } from '@sentry/core';
+import { endStreamSpan } from '../core/utils';
+import { createStreamingState, processEvent } from './streaming';
+import type { AnthropicAiStreamingEvent } from './types';
+
+/** The slice of a stream controller the SSE pass-through uses, shared by the byte and default variants. */
+interface SseStreamController {
+  enqueue: (chunk: Uint8Array) => void;
+  close: () => void;
+  error: (reason?: unknown) => void;
+}
+
+/** Walks the prototype chain for a getter, so a shadowing own property doesn't hide the original. */
+function findBoundGetter(target: object, key: string): (() => unknown) | undefined {
+  for (let proto = Object.getPrototypeOf(target); proto; proto = Object.getPrototypeOf(proto)) {
+    const descriptor = Object.getOwnPropertyDescriptor(proto, key);
+    if (descriptor?.get) {
+      return descriptor.get.bind(target);
+    }
+  }
+  return undefined;
+}
+
+/**
+ * Replace `response.body` with a pass-through that accumulates the SSE frames flowing through it and
+ * ends `span` once the body is exhausted, cancelled or errors.
+ *
+ * Every way of draining an Anthropic stream bottoms out in `response.body`: the SDK `Stream`'s async
+ * iterator, `tee()`, and a caller reading the raw `Response` from `.asResponse()`/`.withResponse()`.
+ * Instrumenting the body instead of the `Stream` covers all of them with one accumulator.
+ *
+ * Returns `false`, leaving the response untouched, for a body we can't wrap.
+ *
+ * @internal Exported for the diagnostics-channel integration.
+ */
+export function instrumentRawSseBody(response: { body?: unknown }, span: Span, recordOutputs: boolean): boolean {
+  const body = response.body as ReadableStream<Uint8Array> | null | undefined;
+  // An unsampled span is discarded by `endStreamSpan`, so decoding and parsing every frame for it buys
+  // nothing — leave the body alone and let the caller's response stay exactly as the SDK built it.
+  if (!body || typeof body.getReader !== 'function' || !span.isRecording()) {
+    return false;
+  }
+
+  const state = createStreamingState();
+  const decoder = new TextDecoder();
+  let buffered = '';
+  let settled = false;
+
+  // Never lets an accumulation failure reach the caller: their stream matters more than our attributes.
+  // Scoped to a single frame so one line we can't parse doesn't cost us its neighbours — `message_delta`
+  // (token usage) and `message_stop` ride in the last chunk, where a bad frame would hurt most.
+  const consumeFrame = (line: string): void => {
+    // An SSE frame's `event:` line only repeats the `type` already carried by the JSON payload.
+    if (!line.startsWith('data:')) {
+      return;
+    }
+    try {
+      processEvent(JSON.parse(line.slice(5)) as AnthropicAiStreamingEvent, state, recordOutputs, span);
+    } catch {
+      // A frame we can't parse is not worth breaking the caller's stream over.
+    }
+  };
+
+  const consumeText = (text: string): void => {
+    buffered += text;
+
+    let newline = buffered.indexOf('\n');
+    while (newline !== -1) {
+      consumeFrame(buffered.slice(0, newline).trim());
+      buffered = buffered.slice(newline + 1);
+      newline = buffered.indexOf('\n');
+    }
+  };
+
+  const consumeBytes = (chunk: Uint8Array): void => {
+    try {
+      consumeText(decoder.decode(chunk, { stream: true }));
+    } catch {
+      // As above: a chunk we can't decode is not worth breaking the caller's stream over.
+    }
+  };
+
+  // No error status on a torn-down body, matching `instrumentAsyncIterableStream`: the SDK surfaces the
+  // failure to the caller, and an `error` SSE frame already marks the span through `isErrorEvent`.
+  const settle = (): void => {
+    if (settled) {
+      return;
+    }
+    settled = true;
+    // A body that ends without a trailing newline leaves its last frame sitting in `buffered`.
+    const trailing = buffered.trim();
+    buffered = '';
+    if (trailing) {
+      consumeFrame(trailing);
+    }
+    endStreamSpan(span, state, recordOutputs);
+  };
+
+  // Resolved on every read rather than captured at wrap time: `clone()` tees the response's *internal*
+  // body and swaps in one branch, which leaves the stream we were handed permanently locked.
+  const readBody = findBoundGetter(response, 'body');
+  const source = (): ReadableStream<Uint8Array> => (readBody?.() as ReadableStream<Uint8Array> | null) ?? body;
+
+  // Acquired on the first read, never at wrap time: taking a reader disturbs the body, which would
+  // make `response.text()`, `arrayBuffer()` and `clone()` throw on a response nobody has read yet.
+  let reader: ReadableStreamDefaultReader<Uint8Array> | undefined;
+
+  const underlyingSource = {
+    async pull(controller: SseStreamController): Promise<void> {
+      try {
+        reader ??= source().getReader();
+        const { done, value } = await reader.read();
+        if (done) {
+          settle();
+          controller.close();
+          return;
+        }
+        // Must precede the enqueue: a byte stream transfers the chunk's buffer, detaching it.
+        consumeBytes(value);
+        controller.enqueue(value);
+      } catch (error) {
+        settle();
+        controller.error(error);
+      }
+    },
+    async cancel(reason: unknown): Promise<void> {
+      settle();
+      await (reader ? reader.cancel(reason) : source().cancel(reason));
+    },
+  };
+
+  // A high-water mark of 0 keeps the stream from pulling a chunk before anyone asks for one. The
+  // default of 1 would read ahead the moment we wrap, disturbing a body the caller may never read.
+  const strategy = { highWaterMark: 0 };
+  let instrumented: ReadableStream<Uint8Array>;
+  try {
+    // `response.body` is a byte stream, so a plain one here would break `getReader({ mode: 'byob' })`
+    // on a response that supported it before we touched it.
+    instrumented = new ReadableStream<Uint8Array>(
+      { ...underlyingSource, type: 'bytes' } as unknown as UnderlyingSource<Uint8Array>,
+      strategy,
+    );
+  } catch {
+    instrumented = new ReadableStream<Uint8Array>(underlyingSource as UnderlyingSource<Uint8Array>, strategy);
+  }
+
+  try {
+    // `body` is a prototype getter, so an own data property shadows it for every later read.
+    Object.defineProperty(response, 'body', { value: instrumented, configurable: true });
+  } catch {
+    return false;
+  }
+
+  // `text()`, `json()` and `arrayBuffer()` read the response's internal body and never touch the
+  // property we just shadowed. Without wrapping them too, a caller draining the stream that way would
+  // leave the span unended forever, since nothing else ends it once we take ownership.
+  instrumentBodyConsumers(response, consumeText, settle);
+
+  return true;
+}
+
+const BODY_CONSUMERS = ['text', 'json', 'arrayBuffer'] as const;
+
+function instrumentBodyConsumers(
+  response: Record<string, unknown>,
+  consumeText: (text: string) => void,
+  settle: () => void,
+): void {
+  for (const name of BODY_CONSUMERS) {
+    const original = response[name];
+    if (typeof original !== 'function') {
+      continue;
+    }
+
+    const wrapped = function (this: unknown, ...args: unknown[]): Promise<unknown> {
+      return Promise.resolve((original as (...a: unknown[]) => unknown).apply(this ?? response, args)).then(
+        result => {
+          // Whatever came back is the stream we would have accumulated off the body, so read the frames
+          // out of it rather than settling for a span with request attributes only. `json()` returns
+          // neither shape, since parsing an SSE body as JSON is a caller error to begin with.
+          if (typeof result === 'string') {
+            consumeText(result);
+          } else if (result instanceof ArrayBuffer) {
+            consumeText(new TextDecoder().decode(result));
+          }
+          settle();
+          return result;
+        },
+        error => {
+          settle();
+          throw error;
+        },
+      );
+    };
+
+    try {
+      Object.defineProperty(response, name, { value: wrapped, configurable: true, writable: true });
+    } catch {
+      // A consumer we can't wrap just means the span leans on the body wrapper to end.
+    }
+  }
+}