fix(server-utils): Mark AI spans as cancelled when the caller aborts

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

@@ -0,0 +1,69 @@
+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());
+
+  // Never responds, so each request only ends when the client gives up on it.
+  app.post('/anthropic/v1/messages', () => {});
+
+  return new Promise(resolve => {
+    const server = app.listen(0, () => {
+      resolve(server);
+    });
+  });
+}
+
+function abortAfter(ms) {
+  const controller = new AbortController();
+  setTimeout(() => controller.abort(), ms);
+  return controller.signal;
+}
+
+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`,
+      maxRetries: 0,
+    });
+    const messages = [{ role: 'user', content: 'Aborted by the caller' }];
+
+    const requests = [
+      [{ model: 'abort-before-stream', stream: true }, { signal: abortAfter(50) }],
+      [{ model: 'abort-non-streaming' }, { signal: abortAfter(50) }],
+      [{ model: 'timeout-before-stream', stream: true }, { timeout: 50 }],
+      [{ model: 'timeout-non-streaming' }, { timeout: 50 }],
+    ];
+    for (const [params, requestOptions] of requests) {
+      try {
+        await client.messages.create({ ...params, max_tokens: 100, messages }, requestOptions);
+      } catch {
+        // Error expected
+      }
+    }
+
+    const streamRequests = [
+      [{ model: 'stream-helper-abort' }, { signal: abortAfter(50) }],
+      [{ model: 'stream-helper-signal-timeout' }, { signal: AbortSignal.timeout(50) }],
+    ];
+    for (const [params, requestOptions] of streamRequests) {
+      try {
+        await client.messages.stream({ ...params, max_tokens: 100, messages }, requestOptions).finalMessage();
+      } catch {
+        // Error expected
+      }
+    }
+  });
+
+  await Sentry.flush(2000);
+
+  server.closeAllConnections();
+  server.close();
+}
+
+run();

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

@@ -586,4 +586,31 @@ describe('Anthropic integration', () => {
       });
     });
   });
+
+  createEsmAndCjsTests(__dirname, 'scenario-abort.mjs', 'instrument.mjs', (createRunner, test) => {
+    test('ends a call the caller aborts as cancelled, but a timeout as an error', async () => {
+      await createRunner()
+        .unordered()
+        .expect({
+          span: container => {
+            const spans = container.items.filter(span => span.attributes[SENTRY_ORIGIN]?.value === 'auto.ai.anthropic');
+            expect(spans).toHaveLength(6);
+            const statusByModel = Object.fromEntries(
+              spans.map(span => [span.attributes[GEN_AI_REQUEST_MODEL]?.value, span.status]),
+            );
+            // Span streaming sends `cancelled` as `ok`.
+            expect(statusByModel).toEqual({
+              'abort-before-stream': 'ok',
+              'abort-non-streaming': 'ok',
+              'timeout-before-stream': 'error',
+              'timeout-non-streaming': 'error',
+              'stream-helper-abort': 'ok',
+              'stream-helper-signal-timeout': 'error',
+            });
+          },
+        })
+        .start()
+        .completed();
+    });
+  });
 });

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

@@ -73,6 +73,10 @@ function startMockAnthropicServer() {
 
   app.post('/anthropic/v1/messages', (req, res) => {
     const mode = req.headers['x-mode'];
+    // Never responds, so the request only ends when the caller's signal fires.
+    if (mode === 'thread_stream_signal_abort' || mode === 'thread_stream_signal_timeout') {
+      return;
+    }
     if (mode === 'thread_stream_error') {
       res.status(400).json({ type: 'error', error: { type: 'invalid_request_error', message: 'Bad thread' } });
       return;
@@ -197,6 +201,22 @@ const MODES = {
     stream.abort();
     await stream.finalMessage().catch(() => {});
   },
+  async thread_stream_signal_abort(client) {
+    const controller = new AbortController();
+    setTimeout(() => controller.abort(), 50);
+    await client.beta.messages
+      .thread({ model: MODEL, max_tokens: 100 })
+      .stream(params.messages, {}, { signal: controller.signal })
+      .finalMessage()
+      .catch(() => {});
+  },
+  async thread_stream_signal_timeout(client) {
+    await client.beta.messages
+      .thread({ model: MODEL, max_tokens: 100 })
+      .stream(params.messages, {}, { signal: AbortSignal.timeout(50) })
+      .finalMessage()
+      .catch(() => {});
+  },
   async thread_stream_truncated(client) {
     await client.beta.messages
       .thread({ model: MODEL, max_tokens: 100 })

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

@@ -137,6 +137,8 @@ describe('Anthropic integration (1.x)', () => {
     const STATUS_ONLY_MODES = [
       { mode: 'thread_stream_abort', status: 'ok' },
       { mode: 'thread_stream_abort_early', status: 'ok' },
+      { mode: 'thread_stream_signal_abort', status: 'ok' },
+      { mode: 'thread_stream_signal_timeout', status: 'error' },
       { mode: 'thread_stream_truncated', status: 'ok' },
       { mode: 'thread_stream_error', status: 'error' },
     ];

dev-packages/node-integration-tests/suites/tracing/google-genai/scenario-stream-midway-error.mjs

@@ -7,6 +7,11 @@ function startMockGoogleGenAIServer() {
   app.use(express.json());
 
   app.post('/v1beta/models/:model\\:streamGenerateContent', (req, res) => {
+    // Never responds, so the request only ends when the client gives up on it.
+    if (req.params.model.endsWith('before-stream')) {
+      return;
+    }
+
     res.setHeader('Content-Type', 'application/json');
     res.setHeader('Transfer-Encoding', 'chunked');
 
@@ -81,6 +86,28 @@ async function run() {
       // Error expected
     }
 
+    const beforeStream = new AbortController();
+    setTimeout(() => beforeStream.abort(), 50);
+    try {
+      await client.models.generateContentStream({
+        model: 'caller-abort-before-stream',
+        contents: [{ role: 'user', parts: [{ text: 'The caller aborts before the first chunk' }] }],
+        config: { abortSignal: beforeStream.signal },
+      });
+    } catch {
+      // Error expected
+    }
+
+    try {
+      await client.models.generateContentStream({
+        model: 'timeout-before-stream',
+        contents: [{ role: 'user', parts: [{ text: 'The SDK timeout fires before the first chunk' }] }],
+        config: { httpOptions: { timeout: 50 } },
+      });
+    } catch {
+      // Error expected
+    }
+
     try {
       const timedOutStream = await client.models.generateContentStream({
         model: 'timeout',

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

@@ -597,7 +597,7 @@ describe('Google GenAI integration', () => {
           .expect({
             span: container => {
               const spans = container.items.filter(span => span.attributes[SENTRY_ORIGIN]?.value === EXPECTED_ORIGIN);
-              expect(spans).toHaveLength(4);
+              expect(spans).toHaveLength(6);
 
               const midwaySpan = spans.find(span => span.name === 'generate_content midway-error');
               expect(midwaySpan!.status).toBe('error');
@@ -613,6 +613,10 @@ describe('Google GenAI integration', () => {
               // `cancelled` is sent as `ok` under span streaming, so a timeout must not be recorded as one.
               expect(spans.find(span => span.name === 'generate_content caller-abort')!.status).toBe('ok');
               expect(spans.find(span => span.name === 'generate_content timeout')!.status).toBe('error');
+              expect(spans.find(span => span.name === 'generate_content caller-abort-before-stream')!.status).toBe(
+                'ok',
+              );
+              expect(spans.find(span => span.name === 'generate_content timeout-before-stream')!.status).toBe('error');
             },
           })
           .start()

dev-packages/node-integration-tests/suites/tracing/openai/scenario-abort.mjs

@@ -0,0 +1,57 @@
+import * as Sentry from '@sentry/node';
+import express from 'express';
+import OpenAI from 'openai';
+
+function startMockServer() {
+  const app = express();
+  app.use(express.json());
+
+  // Never responds, so each request only ends when the client gives up on it.
+  app.post('/openai/chat/completions', () => {});
+
+  return new Promise(resolve => {
+    const server = app.listen(0, () => {
+      resolve(server);
+    });
+  });
+}
+
+function abortAfter(ms) {
+  const controller = new AbortController();
+  setTimeout(() => controller.abort(), ms);
+  return controller.signal;
+}
+
+async function run() {
+  const server = await startMockServer();
+
+  await Sentry.startSpan({ op: 'function', name: 'main' }, async () => {
+    const client = new OpenAI({
+      baseURL: `http://localhost:${server.address().port}/openai`,
+      apiKey: 'mock-api-key',
+      maxRetries: 0,
+    });
+    const messages = [{ role: 'user', content: 'Aborted by the caller' }];
+
+    const requests = [
+      [{ model: 'abort-before-stream', stream: true }, { signal: abortAfter(50) }],
+      [{ model: 'abort-non-streaming' }, { signal: abortAfter(50) }],
+      [{ model: 'timeout-before-stream', stream: true }, { timeout: 50 }],
+      [{ model: 'timeout-non-streaming' }, { timeout: 50 }],
+    ];
+    for (const [params, requestOptions] of requests) {
+      try {
+        await client.chat.completions.create({ ...params, messages }, requestOptions);
+      } catch {
+        // Error expected
+      }
+    }
+  });
+
+  await Sentry.flush(2000);
+
+  server.closeAllConnections();
+  server.close();
+}
+
+run();

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

@@ -1722,4 +1722,29 @@ describe('OpenAI integration', () => {
       });
     },
   );
+
+  createEsmAndCjsTests(__dirname, 'scenario-abort.mjs', 'instrument.mjs', (createRunner, test) => {
+    test('ends a call the caller aborts as cancelled, but an SDK timeout as an error', async () => {
+      await createRunner()
+        .unordered()
+        .expect({
+          span: container => {
+            const spans = container.items.filter(span => span.attributes[SENTRY_ORIGIN]?.value === 'auto.ai.openai');
+            expect(spans).toHaveLength(4);
+            const statusByModel = Object.fromEntries(
+              spans.map(span => [span.attributes[GEN_AI_REQUEST_MODEL]?.value, span.status]),
+            );
+            // Span streaming sends `cancelled` as `ok`.
+            expect(statusByModel).toEqual({
+              'abort-before-stream': 'ok',
+              'abort-non-streaming': 'ok',
+              'timeout-before-stream': 'error',
+              'timeout-non-streaming': 'error',
+            });
+          },
+        })
+        .start()
+        .completed();
+    });
+  });
 });

packages/server-utils/src/ai/anthropic-ai/index.ts

@@ -1,5 +1,5 @@
 /* eslint-disable typescript-eslint/no-deprecated */
-import { getClient, hasSpanStreamingEnabled, SPAN_STATUS_ERROR, startSpan, startSpanManual } from '@sentry/core';
+import { getClient, hasSpanStreamingEnabled, startSpan, startSpanManual } from '@sentry/core';
 import type { Span, SpanAttributeValue } from '@sentry/core';
 import {
   GEN_AI_OPERATION_NAME,
@@ -20,7 +20,9 @@ import {
 import { GEN_AI_REQUEST_STREAM_ATTRIBUTE } from '../core/gen-ai-attributes';
 import type { InstrumentedMethodEntry } from '../core/utils';
 import {
+  getErrorStatus,
   getGenAiSpanOp,
+  getRequestSignal,
   getTokenUsageAttributes,
   resolveAIRecordingOptions,
   wrapPromiseWithMethods,
@@ -164,9 +166,9 @@ export function addResponseAttributes(span: Span, response: AnthropicAiResponse,
 /**
  * Handle common error catching and reporting for streaming requests
  */
-function handleStreamingError(error: unknown, span: Span): never {
+function handleStreamingError(error: unknown, span: Span, abortSignal: AbortSignal | undefined): never {
   if (span.isRecording()) {
-    span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' });
+    span.setStatus(getErrorStatus(abortSignal));
     span.end();
   }
   throw error;
@@ -185,6 +187,7 @@ function handleStreamingRequest<T extends unknown[], R>(
   isStreamRequested: boolean,
   isStreamingMethod: boolean,
 ): R | Promise<R> {
+  const abortSignal = getRequestSignal(args[1]);
   // messages.stream() always returns a sync MessageStream, even with stream: true param
   if (isStreamRequested && !isStreamingMethod) {
     let originalResult!: Promise<R>;
@@ -205,7 +208,7 @@ function handleStreamingRequest<T extends unknown[], R>(
             options.recordOutputs ?? false,
           ) as unknown as R;
         } catch (error) {
-          return handleStreamingError(error, span);
+          return handleStreamingError(error, span, abortSignal);
         }
       })();
     });
@@ -222,10 +225,10 @@ function handleStreamingRequest<T extends unknown[], R>(
         suppressDelegatedCreate = true;
         const messageStream = target.apply(invocationThis, args);
         suppressDelegatedCreate = false;
-        return instrumentMessageStream(messageStream, span, options.recordOutputs ?? false);
+        return instrumentMessageStream(messageStream, span, options.recordOutputs ?? false, abortSignal);
       } catch (error) {
         suppressDelegatedCreate = false;
-        return handleStreamingError(error, span);
+        return handleStreamingError(error, span, abortSignal);
       }
     });
   }
@@ -290,6 +293,7 @@ function instrumentMethod<T extends unknown[], R>(
         );
       }
 
+      const abortSignal = getRequestSignal(args[1]);
       let originalResult!: Promise<R>;
 
       const instrumentedPromise = startSpan(spanConfig, span => {
@@ -299,10 +303,16 @@ function instrumentMethod<T extends unknown[], R>(
           addPrivateRequestAttributes(span, params);
         }
 
-        return originalResult.then(result => {
-          addResponseAttributes(span, result as AnthropicAiResponse, options.recordOutputs);
-          return result;
-        });
+        return originalResult.then(
+          result => {
+            addResponseAttributes(span, result as AnthropicAiResponse, options.recordOutputs);
+            return result;
+          },
+          error => {
+            span.setStatus(getErrorStatus(abortSignal));
+            throw error;
+          },
+        );
       });
 
       return wrapPromiseWithMethods(originalResult, instrumentedPromise);