fix(server-utils): Skip finished thread streams when matching a request

A create() sent with the signal of a thread stream whose request failed was
still matched to that stream, so its span never ended. A finished stream sends
no more requests, so skip it.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: RulaKhaled

dev-packages/node-integration-tests/suites/tracing/anthropic/v1/scenario-threads.mjs

@@ -78,6 +78,10 @@ function startMockAnthropicServer() {
     const answersTool = req.body.messages.some(
       message => Array.isArray(message.content) && message.content.some(block => block.type === 'tool_result'),
     );
+    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 = reply(mode, answersTool ? 'final' : 'tools');
 
     if (!req.body.stream) {
@@ -86,7 +90,7 @@ function startMockAnthropicServer() {
     }
 
     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`.
+    // 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(() => {
@@ -115,6 +119,17 @@ const tools = [
 ];
 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({
@@ -152,14 +167,12 @@ const MODES = {
   async thread_stream_signal_reuse(client) {
     const stream = client.beta.messages.thread({ model: MODEL, max_tokens: 100 }).stream(params.messages);
     await stream.finalMessage();
-    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: stream.controller.signal },
-    );
-    for await (const _event of followUp) {
-      // consume
-    }
+    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);

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

@@ -165,7 +165,8 @@ describe('Anthropic integration (1.x)', () => {
         const spans = await getSpans();
 
         expect(spans.filter(span => span.attributes[SENTRY_OP]?.value === GEN_AI_CHAT)).toHaveLength(
-          MODES.flatMap(({ turns }) => turns).length + STATUS_ONLY_MODES.length,
+          // Plus the failed request and the follow-up of `thread_stream_failed_signal_reuse`.
+          MODES.flatMap(({ turns }) => turns).length + STATUS_ONLY_MODES.length + 2,
         );
         expect(spans.filter(span => span.name.startsWith('anthropic.'))).toEqual([]);
       });
@@ -207,6 +208,15 @@ describe('Anthropic integration (1.x)', () => {
         expect(chatSpans).toHaveLength(1);
         expect(chatSpans[0]!.status).toBe(status);
       });
+
+      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.');
+      });
     });
   });
 

packages/server-utils/src/integrations/anthropic.ts

@@ -211,7 +211,7 @@ function isMessageStream(value: unknown): value is MessageStreamEmitter {
   return !!value && typeof (value as MessageStreamEmitter).on === 'function';
 }
 
-type ThreadStream = MessageStreamEmitter & { controller?: { signal?: AbortSignal } };
+type ThreadStream = MessageStreamEmitter & { ended?: boolean; controller?: { signal?: AbortSignal } };
 
 /**
  * The streams of threads and tool runners on 1.2+, by the signal they send each `create()` call with. Such a
@@ -232,7 +232,8 @@ function wrapThreadStreamRequest(
   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') {
+  // A finished stream sends no more requests, so a later call with its signal is the caller's own.
+  if (!stream || stream.ended || typeof request?._responseProps !== 'function') {
     return false;
   }