RulaKhaled · Oct 30, 2026
@@ -5,7 +5,7 @@ Sentry.init({
   dsn: 'https://public@dsn.ingest.sentry.io/1337',
   release: '1.0',
   tracesSampleRate: 1.0,
-  dataCollection: { genAI: { inputs: false, outputs: false } },
+  dataCollection: { genAI: { inputs: false, outputs: process.env.RECORD_OUTPUTS === 'true' } },
   transport: loggingTransport,
   // Registers Sentry as the global tracer provider, which is what 1.x traces itself through.
   enableOpenTelemetrySetup: process.env.OTEL_SETUP === 'true',

@@ -0,0 +1,194 @@
+import Anthropic from '@anthropic-ai/sdk';
+import * as Sentry from '@sentry/node';
+import express from 'express';
+
+const MODEL = 'claude-3-haiku-20240307';
+
+// The first turn calls the weather tool and the second answers. Ids are `msg_<mode>_<turn>`, so each span can be told apart.
+function reply(mode, turn) {
+  const id = `msg_${mode}_${turn}`;
+  return turn === 'tools'
+    ? {
+        id,
+        content: [{ type: 'tool_use', id: `toolu_${mode}`, name: 'weather', input: { city: 'Vienna' } }],
+        stop_reason: 'tool_use',
+        usage: { input_tokens: 10, output_tokens: 20 },
+      }
+    : {
+        id,
+        content: [{ type: 'text', text: 'Sunny.' }],
+        stop_reason: 'end_turn',
+        usage: { input_tokens: 30, output_tokens: 5 },
+      };
+}
+
+function streamEvents({ id, content, stop_reason, usage }) {
+  const events = [
+    {
+      type: 'message_start',
+      message: {
+        id,
+        type: 'message',
+        role: 'assistant',
+        model: MODEL,
+        content: [],
+        usage: { input_tokens: usage.input_tokens },
+      },
+    },
+  ];
+  content.forEach((block, index) => {
+    if (block.type === 'tool_use') {
+      events.push(
+        { type: 'content_block_start', index, content_block: { ...block, input: {} } },
+        {
+          type: 'content_block_delta',
+          index,
+          delta: { type: 'input_json_delta', partial_json: JSON.stringify(block.input) },
+        },
+      );
+    } else {
+      events.push(
+        { type: 'content_block_start', index, content_block: { type: 'text', text: '' } },
+        { type: 'content_block_delta', index, delta: { type: 'text_delta', text: block.text } },
+      );
+    }
+    events.push({ type: 'content_block_stop', index });
+  });
+  events.push(
+    {
+      type: 'message_delta',
+      delta: { stop_reason, stop_sequence: null },
+      usage: { output_tokens: usage.output_tokens },
+    },
+    { type: 'message_stop' },
+  );
+  return events;
+}
+
+function startMockAnthropicServer() {
+  const app = express();
+  app.use(express.json());
+
+  app.post('/anthropic/v1/messages', (req, res) => {
+    const mode = req.headers['x-mode'];
+    if (mode === 'thread_stream_error') {
+      res.status(400).json({ type: 'error', error: { type: 'invalid_request_error', message: 'Bad thread' } });
+      return;
+    }
+    const answersTool = req.body.messages.some(
+      message => Array.isArray(message.content) && message.content.some(block => block.type === 'tool_result'),
+    );
+    const message = reply(mode, answersTool ? 'final' : 'tools');
+
+    if (!req.body.stream) {
+      res.json({ type: 'message', role: 'assistant', model: MODEL, stop_sequence: null, ...message });
+      return;
+    }
+
+    res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', Connection: 'keep-alive' });
+    // The body closes after the first text delta, without `message_stop`.
+    const events = mode === 'thread_stream_truncated' ? streamEvents(message).slice(0, 3) : streamEvents(message);
+    events.forEach((event, index) => {
+      setTimeout(() => {
+        res.write(`event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`);
+        if (index === events.length - 1) {
+          res.end();
+        }
+      }, index * 5);
+    });
+  });
+
+  return new Promise(resolve => {
+    const server = app.listen(0, () => {
+      resolve(server);
+    });
+  });
+}
+
+const tools = [
+  {
+    name: 'weather',
+    description: 'Get the weather by city',
+    input_schema: { type: 'object', properties: { city: { type: 'string' } } },
+    run: () => 'sunny',
+  },
+];
+const params = { model: MODEL, max_tokens: 100, messages: [{ role: 'user', content: 'Weather in Vienna?' }] };
+
+const MODES = {
+  async runner_stream_eager(client) {
+    for await (const stream of client.beta.messages.toolRunner({
+      ...params,
+      tools,
+      stream: true,
+      runToolsEagerly: true,
+    })) {
+      await stream.finalMessage();
+    }
+  },
+  async runner_stream(client) {
+    for await (const stream of client.beta.messages.toolRunner({ ...params, tools, stream: true })) {
+      await stream.finalMessage();
+    }
+  },
+  // How Bedrock and Vertex clients, which have no threads, run every tool runner.
+  async runner_stream_unthreaded(client) {
+    for await (const stream of client.beta.messages.toolRunner({ ...params, tools, stream: true, thread: false })) {
+      await stream.finalMessage();
+    }
+  },
+  async runner(client) {
+    for await (const _message of client.beta.messages.toolRunner({ ...params, tools })) {
+      // consume
+    }
+  },
+  async thread_send(client) {
+    await client.beta.messages.thread({ model: MODEL, max_tokens: 100 }).send(params.messages);
+  },
+  async thread_stream(client) {
+    await client.beta.messages.thread({ model: MODEL, max_tokens: 100 }).stream(params.messages).finalMessage();
+  },
+  async thread_stream_abort(client) {
+    const stream = client.beta.messages.thread({ model: MODEL, max_tokens: 100 }).stream(params.messages);
+    stream.once('streamEvent', () => stream.abort());
+    await stream.finalMessage().catch(() => {});
+  },
+  async thread_stream_abort_early(client) {
+    const stream = client.beta.messages.thread({ model: MODEL, max_tokens: 100 }).stream(params.messages);
+    stream.abort();
+    await stream.finalMessage().catch(() => {});
+  },
+  async thread_stream_truncated(client) {
+    await client.beta.messages
+      .thread({ model: MODEL, max_tokens: 100 })
+      .stream(params.messages)
+      .finalMessage()
+      .catch(() => {});
+  },
+  async thread_stream_error(client) {
+    await client.beta.messages
+      .thread({ model: MODEL, max_tokens: 100 })
+      .stream(params.messages)
+      .finalMessage()
+      .catch(() => {});
+  },
+};
+
+async function run() {
+  const server = await startMockAnthropicServer();
+  const baseURL = `http://localhost:${server.address().port}/anthropic`;
+
+  await Sentry.startSpan({ op: 'function', name: 'main' }, async () => {
+    for (const [mode, call] of Object.entries(MODES)) {
+      await Sentry.startSpan({ name: mode }, () =>
+        call(new Anthropic({ apiKey: 'mock-api-key', baseURL, maxRetries: 0, defaultHeaders: { 'x-mode': mode } })),
+      );
+    }
+  });
+
+  await Sentry.flush(2000);
+
+  server.close();
+}
+
+run();
@@ -2,11 +2,15 @@ import type { Event, SerializedStreamedSpanContainer } from '@sentry/core';
 import {
   GEN_AI_RESPONSE_ID,
   GEN_AI_RESPONSE_STREAMING,
+  GEN_AI_RESPONSE_TEXT,
+  GEN_AI_RESPONSE_TOOL_CALLS,
   GEN_AI_TOOL_NAME,
+  GEN_AI_USAGE_INPUT_TOKENS,
+  GEN_AI_USAGE_OUTPUT_TOKENS,
   SENTRY_OP,
   SENTRY_ORIGIN,
 } from '@sentry/conventions/attributes';
-import { GEN_AI_EXECUTE_TOOL } from '@sentry/conventions/op';
+import { GEN_AI_CHAT, GEN_AI_EXECUTE_TOOL } from '@sentry/conventions/op';
 import { afterAll, describe, expect } from 'vitest';
 import { GEN_AI_TOOL_CALL_ID_ATTRIBUTE } from '../../../../../../packages/server-utils/src/ai/core/gen-ai-attributes';
 import { cleanupChildProcesses, createEsmAndCjsTests } from '../../../../utils/runner';
@@ -115,6 +119,72 @@ describe('Anthropic integration (1.x)', () => {
     }
   });
 
+  // Since 1.2, tool runners and `thread().stream()` send each turn through a thread, which parses the response itself.
+  createEsmAndCjsTests(__dirname, 'scenario-threads.mjs', 'instrument.mjs', (createRunner, test) => {
+    const MODES = [
+      { mode: 'runner_stream_eager', streaming: true, turns: ['tools', 'final'] },
+      { mode: 'runner_stream', streaming: true, turns: ['tools', 'final'] },
+      { mode: 'runner_stream_unthreaded', streaming: true, turns: ['tools', 'final'] },
+      { mode: 'runner', streaming: false, turns: ['tools', 'final'] },
+      { mode: 'thread_send', streaming: false, turns: ['tools'] },
+      { mode: 'thread_stream', streaming: true, turns: ['tools'] },
+      // Only the span's status is known for these. The stream is aborted on its first event or before the response,
+      // its body closes before `message_stop`, or the request fails.
+      { mode: 'thread_stream_abort', streaming: true, turns: ['abort'], status: 'ok' },
+      { mode: 'thread_stream_abort_early', streaming: true, turns: ['abort'], status: 'ok' },
+      { mode: 'thread_stream_truncated', streaming: true, turns: ['truncated'], status: 'ok' },
+      { mode: 'thread_stream_error', streaming: true, turns: ['error'], status: 'error' },
+    ];
+    const TURNS: Record<string, { input?: number; output?: number; text?: string; toolCall?: boolean }> = {
+      tools: { input: 10, output: 20, toolCall: true },
+      final: { input: 30, output: 5, text: 'Sunny.' },
+    };
+
+    for (const env of [{}, { OTEL_SETUP: 'true' }]) {
+      test(`emits one chat span per thread turn with ${JSON.stringify(env)}`, async () => {
+        const runner = createRunner().withEnv({ ...env, RECORD_OUTPUTS: 'true' });
+        const spansPromise = runner.collectStreamedSpansUntilSegment('main');
+
+        await runner.start().completed();
+
+        const spans = await spansPromise;
+        const chatSpans = spans.filter(span => span.attributes[SENTRY_OP]?.value === GEN_AI_CHAT);
+        expect(chatSpans).toHaveLength(MODES.flatMap(({ turns }) => turns).length);
+        expect(spans.filter(span => span.name.startsWith('anthropic.'))).toEqual([]);
+
+        for (const { mode, streaming, turns, status } of MODES) {
+          const modeSpan = spans.find(span => span.name === mode)!;
+          const modeChatSpans = chatSpans.filter(span => span.parent_span_id === modeSpan.span_id);
+          expect(modeChatSpans, mode).toHaveLength(turns.length);
+
+          if (status) {
+            expect(modeChatSpans[0]!.status, mode).toBe(status);
+            continue;
+          }
+
+          for (const turn of turns) {
+            const id = `msg_${mode}_${turn}`;
+            const chatSpan = spanForCall(modeChatSpans, id);
+            const { input, output, text, toolCall } = TURNS[turn]!;
+            expect(chatSpan.status, id).toBe('ok');
+            expect(chatSpan.attributes[GEN_AI_RESPONSE_STREAMING]?.value ?? false, id).toBe(streaming);
+            expect(chatSpan.attributes[GEN_AI_USAGE_INPUT_TOKENS]?.value, id).toBe(input);
+            expect(chatSpan.attributes[GEN_AI_USAGE_OUTPUT_TOKENS]?.value, id).toBe(output);
+            if (text) {
+              expect(chatSpan.attributes[GEN_AI_RESPONSE_TEXT]?.value, id).toBe(text);
+            }
+            const toolCalls = chatSpan.attributes[GEN_AI_RESPONSE_TOOL_CALLS]?.value as string | undefined;
+            expect(toolCalls && JSON.parse(toolCalls), id).toEqual(
+              toolCall
+                ? [{ type: 'tool_use', id: `toolu_${mode}`, name: 'weather', input: { city: 'Vienna' } }]
+                : undefined,
+            );
+          }
+        }
+      });
+    }
+  });
+
   createEsmAndCjsTests(__dirname, 'scenario-session-tool-runner.mjs', 'instrument.mjs', (createRunner, test) => {
     for (const env of [{}, { OTEL_SETUP: 'true' }]) {
       test(`emits an execute_tool span for each tool a session runner runs with ${JSON.stringify(env)}`, async () => {

@@ -284,6 +284,12 @@ export function instrumentMessageStream<R extends { on: (...args: unknown[]) =>
     endStreamSpan(span, state, recordOutputs);
   });
 
+  // Fires last on every stream, so a stream that stops without 'message' (aborted, or the body closed early) still
+  // ends its span. An 'abort' listener would also stop the SDK from raising an unobserved abort as a rejection.
+  stream.on('end', () => {
+    endStreamSpan(span, state, recordOutputs);
+  });
+
   stream.on('error', (error: unknown) => {
     // Attaching this listener stops the stream error from being raised as an unhandled rejection, so
     // we capture it here to avoid swallowing it (e.g. for callers that don't await/iterate the stream).

@@ -72,6 +72,7 @@ function instrumentAnthropic(options: AnthropicAiOptions): void {
           );
         },
         deferSpanEnd: ({ span, data, end }) =>
+          wrapThreadStreamRequest(span, data, end, options) ||
           wrapApiPromiseResponse(
             data.result,
             response => {
@@ -81,7 +82,8 @@ function instrumentAnthropic(options: AnthropicAiOptions): void {
               }
             },
             end,
-          ) || wrapStreamResult(span, data, stream, options),
+          ) ||
+          wrapStreamResult(span, data, stream, options),
       },
     );
   }
@@ -96,6 +98,15 @@ function instrumentAnthropic(options: AnthropicAiOptions): void {
     .end.subscribe(message => {
       instrumentSessionToolRunner((message as OrchestrionChannelContext).self, options);
     });
+  // On `start`, because `_start()` sends the first request before it returns.
+  diagnosticsChannel
+    .tracingChannel<OrchestrionChannelContext>(CHANNELS.ANTHROPIC_THREAD_STREAM)
+    .start.subscribe(message => {
+      const stream = (message as OrchestrionChannelContext).self as ThreadStream | undefined;
+      if (isMessageStream(stream) && stream.controller?.signal) {
+        threadStreams.set(stream.controller.signal, stream);
+      }
+    });
 }
 
 /**
@@ -199,6 +210,41 @@ function isMessageStream(value: unknown): value is MessageStreamEmitter {
   return !!value && typeof (value as MessageStreamEmitter).on === 'function';
 }
 
+type ThreadStream = MessageStreamEmitter & { controller?: { signal?: AbortSignal } };
+
+/**
+ * The streams of threads and tool runners on 1.2+, by the signal they send each `create()` call with. Such a
+ * stream parses the response itself, so the `create()` result is never parsed and never ends its span.
+ */
+const threadStreams = new WeakMap<AbortSignal, ThreadStream>();
+
+/**
+ * End the span of a thread stream's `create()` call from the stream once the response arrives. A thread may
+ * retry a rejected call, so a failed call ends its own span and only the one that succeeds reads the stream.
+ */
+function wrapThreadStreamRequest(
+  span: Span,
+  data: OrchestrionChannelContext,
+  end: (error?: unknown) => void,
+  options: AnthropicAiOptions,
+): boolean {
+  const signal = (data.arguments?.[1] as { signal?: AbortSignal } | undefined)?.signal;
+  const stream = signal && threadStreams.get(signal);
+  const request = data.result as { _responseProps?: () => Promise<unknown> } | undefined;
+  if (!stream || typeof request?._responseProps !== 'function') {
+    return false;
+  }
+
+  const { recordOutputs } = resolveAIRecordingOptions(options);
+  // Not `asResponse()`, which on 1.x also ends the SDK's own span when the user turned its tracing on.
+  void request._responseProps().then(
+    () => instrumentMessageStream(stream, span, recordOutputs),
+    // A caller abort is not a failed request.
+    error => end(signal.aborted ? undefined : error),
+  );
+  return true;
+}
+
 /**
  * Hand span-ending ownership to a streamed result: returns `true` to skip the normal `beforeSpanEnd`,
  * `false` for non-streaming results (which end via `beforeSpanEnd`).