@@ -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,238 @@
+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;
+}
+
+let recoveredReplies = 0;
+
+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'),
+    );
+    // The server lost the thread, so the SDK replays the turn that continues it.
+    if (mode === 'thread_stream_recovered' && req.body.thread?.type === 'continue') {
+      res.status(404).json({
+        type: 'error',
+        error: { type: 'not_found_error', message: 'Thread not found', details: { error_code: 'thread_not_found' } },
+      });
+      return;
+    }
+    if (mode === 'thread_stream_failed_signal_reuse' && !answersTool) {
+      res.status(400).json({ type: 'error', error: { type: 'invalid_request_error', message: 'Bad thread' } });
+      return;
+    }
+    const message =
+      mode === 'thread_stream_recovered'
+        ? { ...reply(mode, 'final'), id: `msg_${mode}_${(recoveredReplies += 1)}` }
+        : 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 tool input 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?' }] };
+
+async function streamFollowUp(client, signal) {
+  const answer = { role: 'user', content: [{ type: 'tool_result', tool_use_id: 'toolu_1', content: 'sunny' }] };
+  const followUp = await client.beta.messages.create(
+    { ...params, messages: [...params.messages, answer], stream: true },
+    { signal },
+  );
+  for await (const _event of followUp) {
+    // consume
+  }
+}
+
+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();
+  },
+  // A later request sent with a finished thread stream's signal is not that stream's request.
+  async thread_stream_signal_reuse(client) {
+    const stream = client.beta.messages.thread({ model: MODEL, max_tokens: 100 }).stream(params.messages);
+    await stream.finalMessage();
+    await streamFollowUp(client, stream.controller.signal);
+  },
+  async thread_stream_failed_signal_reuse(client) {
+    const stream = client.beta.messages.thread({ model: MODEL, max_tokens: 100 }).stream(params.messages);
+    await stream.finalMessage().catch(() => {});
+    await streamFollowUp(client, stream.controller.signal);
+  },
+  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_recovered(client) {
+    const thread = client.beta.messages.thread({ model: MODEL, max_tokens: 100 });
+    await thread.stream(params.messages).finalMessage();
+    await thread.stream([{ role: 'user', content: 'And tomorrow?' }]).finalMessage();
+  },
+  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();
@@ -3,11 +3,15 @@ import {
   GEN_AI_CONVERSATION_ID,
   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, GEN_AI_INVOKE_AGENT } from '@sentry/conventions/op';
+import { GEN_AI_CHAT, GEN_AI_EXECUTE_TOOL, GEN_AI_INVOKE_AGENT } 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';
@@ -116,6 +120,121 @@ 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'] },
+      { mode: 'thread_stream_signal_reuse', streaming: true, turns: ['tools', 'final'] },
+    ];
+    const FINAL_TURN_MODES = MODES.filter(({ turns }) => turns.includes('final'));
+    // 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.
+    const STATUS_ONLY_MODES = [
+      { mode: 'thread_stream_abort', status: 'ok' },
+      { mode: 'thread_stream_abort_early', status: 'ok' },
+      { mode: 'thread_stream_truncated', status: 'ok' },
+      { mode: 'thread_stream_error', status: 'error' },
+    ];
+
+    describe.each([{}, { OTEL_SETUP: 'true' }])('with %j', env => {
+      // Every scenario run is a Node process, so the cases below share one run.
+      let run: Promise<SerializedSpan[]> | undefined;
+      function getSpans(): Promise<SerializedSpan[]> {
+        run ??= (async () => {
+          const runner = createRunner().withEnv({ ...env, RECORD_OUTPUTS: 'true' });
+          const spansPromise = runner.collectStreamedSpansUntilSegment('main');
+          await runner.start().completed();
+          return spansPromise;
+        })();
+        return run;
+      }
+
+      async function chatSpansUnder(mode: string): Promise<SerializedSpan[]> {
+        const spans = await getSpans();
+        const modeSpan = spans.find(span => span.name === mode)!;
+        return spans.filter(
+          span => span.parent_span_id === modeSpan.span_id && span.attributes[SENTRY_OP]?.value === GEN_AI_CHAT,
+        );
+      }
+
+      test("emits one chat span per turn and none of the SDK's own spans", async () => {
+        const spans = await getSpans();
+
+        expect(spans.filter(span => span.attributes[SENTRY_OP]?.value === GEN_AI_CHAT)).toHaveLength(
+          // Plus the failed request and the follow-up of `thread_stream_failed_signal_reuse`, and the three requests
+          // of `thread_stream_recovered`.
+          MODES.flatMap(({ turns }) => turns).length + STATUS_ONLY_MODES.length + 2 + 3,
+        );
+        expect(spans.filter(span => span.name.startsWith('anthropic.'))).toEqual([]);
+      });
+
+      test.each(MODES)('nests one chat span per $mode turn under the caller', async ({ mode, turns }) => {
+        const chatSpans = await chatSpansUnder(mode);
+
+        expect(chatSpans.map(span => span.attributes[GEN_AI_RESPONSE_ID]?.value).sort()).toEqual(
+          turns.map(turn => `msg_${mode}_${turn}`).sort(),
+        );
+      });
+
+      test.each(MODES)('records the tool call of the first $mode turn', async ({ mode, streaming }) => {
+        const chatSpan = spanForCall(await chatSpansUnder(mode), `msg_${mode}_tools`);
+
+        expect(chatSpan.status).toBe('ok');
+        expect(chatSpan.attributes[GEN_AI_RESPONSE_STREAMING]?.value ?? false).toBe(streaming);
+        expect(chatSpan.attributes[GEN_AI_USAGE_INPUT_TOKENS]?.value).toBe(10);
+        expect(chatSpan.attributes[GEN_AI_USAGE_OUTPUT_TOKENS]?.value).toBe(20);
+        expect(JSON.parse(chatSpan.attributes[GEN_AI_RESPONSE_TOOL_CALLS]!.value as string)).toEqual([
+          { type: 'tool_use', id: `toolu_${mode}`, name: 'weather', input: { city: 'Vienna' } },
+        ]);
+      });
+
+      test.each(FINAL_TURN_MODES)('records the text of the final $mode turn', async ({ mode, streaming }) => {
+        const chatSpan = spanForCall(await chatSpansUnder(mode), `msg_${mode}_final`);
+
+        expect(chatSpan.status).toBe('ok');
+        expect(chatSpan.attributes[GEN_AI_RESPONSE_STREAMING]?.value ?? false).toBe(streaming);
+        expect(chatSpan.attributes[GEN_AI_USAGE_INPUT_TOKENS]?.value).toBe(30);
+        expect(chatSpan.attributes[GEN_AI_USAGE_OUTPUT_TOKENS]?.value).toBe(5);
+        expect(chatSpan.attributes[GEN_AI_RESPONSE_TEXT]?.value).toBe('Sunny.');
+        expect(chatSpan.attributes[GEN_AI_RESPONSE_TOOL_CALLS]).toBeUndefined();
+      });
+
+      test.each(STATUS_ONLY_MODES)('ends the $mode chat span with status $status', async ({ mode, status }) => {
+        const chatSpans = await chatSpansUnder(mode);
+
+        expect(chatSpans).toHaveLength(1);
+        expect(chatSpans[0]!.status).toBe(status);
+      });
+
+      // Deliberate. The rejected attempt was a real request that failed, like an http.client span that a retry
+      // recovers from. Hiding it would mean copying the SDK's private recovery rules: they replay only some error codes,
+      // only on a `continue`, and may change in any release, so a copy would silently hide real errors.
+      test('keeps an error span for the attempt of a turn the thread recovered from', async () => {
+        const chatSpans = await chatSpansUnder('thread_stream_recovered');
+
+        expect(chatSpans.map(span => [span.attributes[GEN_AI_RESPONSE_ID]?.value, span.status])).toEqual([
+          ['msg_thread_stream_recovered_1', 'ok'],
+          [undefined, 'error'],
+          ['msg_thread_stream_recovered_2', 'ok'],
+        ]);
+      });
+
+      test("ends the span of a request sent with a failed stream's signal", async () => {
+        const chatSpans = await chatSpansUnder('thread_stream_failed_signal_reuse');
+        const followUp = spanForCall(chatSpans, 'msg_thread_stream_failed_signal_reuse_final');
+
+        expect(chatSpans.map(span => span.status).sort()).toEqual(['error', 'ok']);
+        expect(followUp.attributes[GEN_AI_USAGE_OUTPUT_TOKENS]?.value).toBe(5);
+        expect(followUp.attributes[GEN_AI_RESPONSE_TEXT]?.value).toBe('Sunny.');
+      });
+    });
+  });
+
   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 () => {