diff --git a/dev-packages/node-integration-tests/suites/tracing/anthropic/scenario-abort.mjs b/dev-packages/node-integration-tests/suites/tracing/anthropic/scenario-abort.mjs new file mode 100644 index 000000000000..8a15d662c231 --- /dev/null +++ b/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(); diff --git a/dev-packages/node-integration-tests/suites/tracing/anthropic/test.ts b/dev-packages/node-integration-tests/suites/tracing/anthropic/test.ts index 87d1cd735458..54dd27321cd1 100644 --- a/dev-packages/node-integration-tests/suites/tracing/anthropic/test.ts +++ b/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(); + }); + }); }); diff --git a/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/scenario-threads.mjs b/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/scenario-threads.mjs index c2286913e5ff..b6419eebca33 100644 --- a/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/scenario-threads.mjs +++ b/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 }) diff --git a/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/test.ts b/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/test.ts index afda0052828f..383199d7f208 100644 --- a/dev-packages/node-integration-tests/suites/tracing/anthropic/v1/test.ts +++ b/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' }, ]; diff --git a/dev-packages/node-integration-tests/suites/tracing/google-genai/scenario-stream-midway-error.mjs b/dev-packages/node-integration-tests/suites/tracing/google-genai/scenario-stream-midway-error.mjs index 321928dc78bf..27cd1c9f25c7 100644 --- a/dev-packages/node-integration-tests/suites/tracing/google-genai/scenario-stream-midway-error.mjs +++ b/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', diff --git a/dev-packages/node-integration-tests/suites/tracing/google-genai/test.ts b/dev-packages/node-integration-tests/suites/tracing/google-genai/test.ts index c9a1944151d0..afc273ab86c5 100644 --- a/dev-packages/node-integration-tests/suites/tracing/google-genai/test.ts +++ b/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() diff --git a/dev-packages/node-integration-tests/suites/tracing/openai/scenario-abort.mjs b/dev-packages/node-integration-tests/suites/tracing/openai/scenario-abort.mjs new file mode 100644 index 000000000000..5f225f4b7908 --- /dev/null +++ b/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(); diff --git a/dev-packages/node-integration-tests/suites/tracing/openai/test.ts b/dev-packages/node-integration-tests/suites/tracing/openai/test.ts index a6be1844dbab..98b9f8755573 100644 --- a/dev-packages/node-integration-tests/suites/tracing/openai/test.ts +++ b/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(); + }); + }); }); diff --git a/packages/server-utils/src/ai/anthropic-ai/index.ts b/packages/server-utils/src/ai/anthropic-ai/index.ts index 6b4f7d357ce6..5ef9a40636b5 100644 --- a/packages/server-utils/src/ai/anthropic-ai/index.ts +++ b/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( isStreamRequested: boolean, isStreamingMethod: boolean, ): R | Promise { + const abortSignal = getRequestSignal(args[1]); // messages.stream() always returns a sync MessageStream, even with stream: true param if (isStreamRequested && !isStreamingMethod) { let originalResult!: Promise; @@ -205,7 +208,7 @@ function handleStreamingRequest( options.recordOutputs ?? false, ) as unknown as R; } catch (error) { - return handleStreamingError(error, span); + return handleStreamingError(error, span, abortSignal); } })(); }); @@ -222,10 +225,10 @@ function handleStreamingRequest( 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( ); } + const abortSignal = getRequestSignal(args[1]); let originalResult!: Promise; const instrumentedPromise = startSpan(spanConfig, span => { @@ -299,10 +303,16 @@ function instrumentMethod( 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); diff --git a/packages/server-utils/src/ai/anthropic-ai/streaming.ts b/packages/server-utils/src/ai/anthropic-ai/streaming.ts index 253ff094f829..8931d6098b8d 100644 --- a/packages/server-utils/src/ai/anthropic-ai/streaming.ts +++ b/packages/server-utils/src/ai/anthropic-ai/streaming.ts @@ -1,6 +1,6 @@ import { captureException, SPAN_STATUS_ERROR } from '@sentry/core'; import type { Span } from '@sentry/core'; -import { endStreamSpan } from '../core/utils'; +import { endStreamSpan, getErrorStatus } from '../core/utils'; import type { AnthropicAiStreamingEvent } from './types'; import { mapAnthropicErrorToStatusMessage } from './utils'; @@ -262,6 +262,7 @@ export function instrumentMessageStream stream: R, span: Span, recordOutputs: boolean, + abortSignal?: AbortSignal, ): R { const state: StreamingState = { responseTexts: [], @@ -290,8 +291,11 @@ export function instrumentMessageStream // 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', () => { - if (span.isRecording() && (stream as { aborted?: boolean }).aborted) { - span.setStatus({ code: SPAN_STATUS_ERROR, message: 'cancelled' }); + if (span.isRecording() && stream.aborted) { + // `aborted` is also set when the caller's `AbortSignal.timeout()` fires, which is a timeout, not a cancellation. + span.setStatus( + abortSignal?.aborted ? getErrorStatus(abortSignal) : { code: SPAN_STATUS_ERROR, message: 'cancelled' }, + ); } endStreamSpan(span, state, recordOutputs); }); @@ -312,14 +316,5 @@ export function instrumentMessageStream } }); - // An aborted stream emits only `abort` and `end`. We don't listen for `abort`: doing so would stop the SDK from - // raising an unhandled rejection when the caller never awaits the aborted stream. - stream.on('end', () => { - if (stream.aborted && span.isRecording()) { - span.setStatus({ code: SPAN_STATUS_ERROR, message: 'cancelled' }); - } - endStreamSpan(span, state, recordOutputs); - }); - return stream; } diff --git a/packages/server-utils/src/ai/anthropic-ai/tool-runner.ts b/packages/server-utils/src/ai/anthropic-ai/tool-runner.ts index ebf033a462a3..f373f79c2468 100644 --- a/packages/server-utils/src/ai/anthropic-ai/tool-runner.ts +++ b/packages/server-utils/src/ai/anthropic-ai/tool-runner.ts @@ -29,7 +29,7 @@ import { } from '@sentry/conventions/attributes'; import { GEN_AI_EXECUTE_TOOL, GEN_AI_INVOKE_AGENT } from '@sentry/conventions/op'; import { GEN_AI_TOOL_CALL_ID_ATTRIBUTE } from '../core/gen-ai-attributes'; -import { resolveAIRecordingOptions } from '../core/utils'; +import { getErrorStatus, resolveAIRecordingOptions } from '../core/utils'; import { withCleanScopes } from '../../utils/withCleanScopes'; import { ANTHROPIC_AI_ORIGIN } from './constants'; import type { AnthropicAiOptions } from './types'; @@ -217,13 +217,12 @@ function runWithSpan( } function handleToolError(span: Span, error: unknown, context: RunContext): void { + span.setStatus(getErrorStatus(context?.signal)); + // An aborted tool, caller abort or timeout, throws because the run stopped, not because the tool failed. if (context?.signal?.aborted) { - span.setStatus({ code: SPAN_STATUS_ERROR, message: 'cancelled' }); return; } - span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); - // Only a runner passes `toolUse`/`toolUseBlock` (SDK 0.85+), and it hands the throw to the model as a result. // A direct call rethrows to its caller, and a `ToolError` is sent on purpose, so neither is captured. const fromRunner = !!(context?.toolUse ?? context?.toolUseBlock); diff --git a/packages/server-utils/src/ai/core/utils.ts b/packages/server-utils/src/ai/core/utils.ts index 93949ec40eca..816c023ab08e 100644 --- a/packages/server-utils/src/ai/core/utils.ts +++ b/packages/server-utils/src/ai/core/utils.ts @@ -2,8 +2,8 @@ /** * Shared utils for AI integrations (OpenAI, Anthropic, Verce.AI, etc.) */ -import { getClient, isObjectLike, isThenable, stringify } from '@sentry/core'; -import type { Span } from '@sentry/core'; +import { getClient, isObjectLike, isThenable, SPAN_STATUS_ERROR, stringify } from '@sentry/core'; +import type { Span, SpanStatus } from '@sentry/core'; import { GEN_AI_OUTPUT_MESSAGES, GEN_AI_RESPONSE_FINISH_REASONS, @@ -233,6 +233,37 @@ export interface StreamResponseState { reasoningOutputTokens?: number; } +/** + * Reads `signal` from the request options OpenAI and Anthropic take as their second argument. + */ +export function getRequestSignal(requestOptions: unknown): AbortSignal | undefined { + return isObjectLike(requestOptions) ? (requestOptions.signal as AbortSignal | undefined) : undefined; +} + +/** + * Checks the signal rather than the error name, because SDK timeouts also throw an `AbortError`. A caller's + * own `AbortSignal.timeout()` is a timeout too, not a cancellation. + */ +export function isCallerAbort(abortSignal: AbortSignal | undefined): boolean { + return !!abortSignal?.aborted && (abortSignal.reason as { name?: unknown } | undefined)?.name !== 'TimeoutError'; +} + +/** + * A caller's abort is `cancelled`, anything else `internal_error`. + */ +export function getErrorStatus(abortSignal: AbortSignal | undefined): SpanStatus { + return { code: SPAN_STATUS_ERROR, message: isCallerAbort(abortSignal) ? 'cancelled' : 'internal_error' }; +} + +/** + * Overrides the status of a span whose call the caller aborted. + */ +export function markCancelledIfAborted(span: Span, abortSignal: AbortSignal | undefined): void { + if (isCallerAbort(abortSignal)) { + span.setStatus({ code: SPAN_STATUS_ERROR, message: 'cancelled' }); + } +} + /** * Ends a streaming span by setting all accumulated response attributes and ending the span. * Shared across OpenAI, Anthropic, and Google GenAI streaming implementations. diff --git a/packages/server-utils/src/ai/google-genai/index.ts b/packages/server-utils/src/ai/google-genai/index.ts index 8b0477ae40d4..cdc1fe022b62 100644 --- a/packages/server-utils/src/ai/google-genai/index.ts +++ b/packages/server-utils/src/ai/google-genai/index.ts @@ -4,7 +4,6 @@ import { getClient, handleCallbackErrors, hasSpanStreamingEnabled, - SPAN_STATUS_ERROR, startSpan, startSpanManual, stringify, @@ -31,7 +30,13 @@ import { SENTRY_ORIGIN, } from '@sentry/conventions/attributes'; import type { InstrumentedMethodEntry } from '../core/utils'; -import { buildMethodPath, getGenAiSpanOp, getTokenUsageAttributes, resolveAIRecordingOptions } from '../core/utils'; +import { + buildMethodPath, + getErrorStatus, + getGenAiSpanOp, + getTokenUsageAttributes, + resolveAIRecordingOptions, +} from '../core/utils'; import { GOOGLE_GENAI_METHOD_REGISTRY, GOOGLE_GENAI_SYSTEM_NAME } from './constants'; import { instrumentStream } from './streaming'; import type { Candidate, ContentPart, GoogleGenAIOptions, GoogleGenAIResponse } from './types'; @@ -285,6 +290,13 @@ export function resolveChatParams( return { ...params, config: chatConfig }; } +/** + * Reads the caller's `config.abortSignal` from params resolved by `resolveChatParams`. + */ +export function getAbortSignal(params: Record | undefined): AbortSignal | undefined { + return (params?.config as { abortSignal?: AbortSignal } | undefined)?.abortSignal; +} + /** * Instrument any async or synchronous genai method with Sentry spans * Handles operations like models.generateContent and chat.sendMessage and chats.create @@ -304,6 +316,7 @@ function instrumentMethod( const operationName = instrumentedMethod.operation || 'unknown'; const params = args[0] as Record | undefined; const attributeParams = resolveChatParams(operationName, params, context); + const abortSignal = getAbortSignal(attributeParams); const requestAttributes = extractRequestAttributes( operationName, attributeParams, @@ -333,14 +346,9 @@ function instrumentMethod( addPrivateRequestAttributes(span, attributeParams, operationName); } const stream = await target.apply(context, args); - return instrumentStream( - stream, - span, - Boolean(options.recordOutputs), - (attributeParams?.config as { abortSignal?: AbortSignal } | undefined)?.abortSignal, - ) as R; + return instrumentStream(stream, span, Boolean(options.recordOutputs), abortSignal) as R; } catch (error) { - span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); + span.setStatus(getErrorStatus(abortSignal)); span.end(); throw error; } @@ -359,11 +367,10 @@ function instrumentMethod( addPrivateRequestAttributes(span, attributeParams, operationName); } - // `onError` is a no-op because the rejection is rethrown to the caller and `startSpan` already - // marks the span errored; both leading callbacks are positional and only exist to reach `onSuccess`. + // The rejection is rethrown to the caller. `startSpan` keeps the status set here. return handleCallbackErrors( () => target.apply(context, args), - () => {}, + () => span.setStatus(getErrorStatus(abortSignal)), () => {}, result => { // Only add response attributes for content-producing methods, not for embeddings diff --git a/packages/server-utils/src/ai/google-genai/streaming.ts b/packages/server-utils/src/ai/google-genai/streaming.ts index 3b19a991e0d0..9aa416d650ad 100644 --- a/packages/server-utils/src/ai/google-genai/streaming.ts +++ b/packages/server-utils/src/ai/google-genai/streaming.ts @@ -1,6 +1,6 @@ import { captureException, SPAN_STATUS_ERROR } from '@sentry/core'; import type { Span } from '@sentry/core'; -import { endStreamSpan } from '../core/utils'; +import { endStreamSpan, getErrorStatus } from '../core/utils'; import type { GoogleGenAIResponse } from './types'; import type { MessagePart } from './utils'; import { candidatesToMessageParts, setOutputMessagesAttribute } from './utils'; @@ -137,7 +137,7 @@ export async function* instrumentStream( yield chunk; } } catch (error) { - span.setStatus({ code: SPAN_STATUS_ERROR, message: abortSignal?.aborted ? 'cancelled' : 'internal_error' }); + span.setStatus(getErrorStatus(abortSignal)); throw error; } finally { if (recordOutputs) { diff --git a/packages/server-utils/src/ai/openai/index.ts b/packages/server-utils/src/ai/openai/index.ts index ada47a5710c7..06d34be7b412 100644 --- a/packages/server-utils/src/ai/openai/index.ts +++ b/packages/server-utils/src/ai/openai/index.ts @@ -1,14 +1,6 @@ /* eslint-disable typescript-eslint/no-deprecated */ import { DEBUG_BUILD } from '../../debug-build'; -import { - getClient, - hasSpanStreamingEnabled, - SPAN_STATUS_ERROR, - startSpan, - startSpanManual, - debug, - stringify, -} from '@sentry/core'; +import { getClient, hasSpanStreamingEnabled, startSpan, startSpanManual, debug, stringify } from '@sentry/core'; import type { Span, SpanAttributeValue } from '@sentry/core'; import { GEN_AI_EMBEDDINGS_INPUT, @@ -24,7 +16,9 @@ import type { InstrumentedMethodEntry } from '../core/utils'; import { buildMethodPath, extractSystemInstructions, + getErrorStatus, getGenAiSpanOp, + getRequestSignal, resolveAIRecordingOptions, wrapPromiseWithMethods, } from '../core/utils'; @@ -158,6 +152,7 @@ function instrumentMethod( const params = args[0] as Record | undefined; const isStreamRequested = params && typeof params === 'object' && params.stream === true; + const abortSignal = getRequestSignal(args[1]); const client = getClient(); const spanConfig = { @@ -190,7 +185,7 @@ function instrumentMethod( options.recordOutputs ?? false, ) as unknown as R; } catch (error) { - span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); + span.setStatus(getErrorStatus(abortSignal)); span.end(); throw error; } @@ -211,10 +206,16 @@ function instrumentMethod( addRequestAttributes(span, params, operationName); } - return originalResult.then(result => { - addResponseAttributes(span, result, options.recordOutputs, operationName); - return result; - }); + return originalResult.then( + result => { + addResponseAttributes(span, result, options.recordOutputs, operationName); + return result; + }, + error => { + span.setStatus(getErrorStatus(abortSignal)); + throw error; + }, + ); }); return wrapPromiseWithMethods(originalResult, instrumentedPromise); diff --git a/packages/server-utils/src/integrations/anthropic.ts b/packages/server-utils/src/integrations/anthropic.ts index 4dfbff811c52..e9d2b1bf4485 100644 --- a/packages/server-utils/src/integrations/anthropic.ts +++ b/packages/server-utils/src/integrations/anthropic.ts @@ -2,7 +2,7 @@ import { GEN_AI_REQUEST_MODEL, SENTRY_ORIGIN } from '@sentry/conventions/attribu import * as diagnosticsChannel from '../utils/diagnosticsChannel'; import type { IntegrationFn, Span, SpanAttributeValue } from '@sentry/core'; import { defineIntegration, getActiveSpan, getClient, hasSpanStreamingEnabled, startInactiveSpan } from '@sentry/core'; -import { getGenAiSpanOp, resolveAIRecordingOptions } from '../ai/core/utils'; +import { getGenAiSpanOp, getRequestSignal, markCancelledIfAborted, resolveAIRecordingOptions } from '../ai/core/utils'; import { wrapApiPromiseResponse } from '../ai/core/apiPromise'; import { addPrivateRequestAttributes, addResponseAttributes, extractRequestAttributes } from '../ai/anthropic-ai'; import { ANTHROPIC_AI_ORIGIN } from '../ai/anthropic-ai/constants'; @@ -60,6 +60,9 @@ function instrumentAnthropic(options: AnthropicAiOptions): void { data => createGenAiSpan(data, operation, options, stream), { beforeSpanEnd: (span, data) => { + if ('error' in data) { + markCancelledIfAborted(span, getRequestSignal(data.arguments?.[1])); + } addResponseAttributes( span, data.result as AnthropicAiResponse, @@ -97,9 +100,12 @@ function instrumentAnthropic(options: AnthropicAiOptions): void { diagnosticsChannel .tracingChannel(CHANNELS.ANTHROPIC_THREAD_STREAM) .start.subscribe(message => { - const stream = (message as OrchestrionChannelContext).self as ThreadStream | undefined; + const data = message as OrchestrionChannelContext; + const stream = data.self as ThreadStream | undefined; if (isMessageStream(stream) && stream.controller?.signal) { - threadStreams.set(stream.controller.signal, stream); + // `_start(create, params, options)`: each `create()` gets only the stream's own signal, which doesn't carry + // the reason a caller's signal fired with. + threadStreams.set(stream.controller.signal, { stream, callerSignal: getRequestSignal(data.arguments?.[2]) }); } }); } @@ -209,7 +215,7 @@ type ThreadStream = MessageStreamEmitter & { ended?: boolean; controller?: { sig * 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(); +const threadStreams = new WeakMap(); /** * End the span of a thread stream's `create()` call from the stream once the response arrives. A thread may @@ -222,12 +228,13 @@ function wrapThreadStreamRequest( options: AnthropicAiOptions, ): boolean { const signal = (data.arguments?.[1] as { signal?: AbortSignal } | undefined)?.signal; - const stream = signal && threadStreams.get(signal); + const threadStream = signal && threadStreams.get(signal); const request = data.result as { _responseProps?: () => Promise } | undefined; // 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') { + if (!threadStream || threadStream.stream.ended || typeof request?._responseProps !== 'function') { return false; } + const { stream, callerSignal } = threadStream; const { recordOutputs } = resolveAIRecordingOptions(options); // Not `asResponse()`, which on 1.x also ends the SDK's own span when the user turned its tracing on. @@ -235,12 +242,12 @@ function wrapThreadStreamRequest( () => { // Only a rejected attempt is retried, so nothing else may claim this stream. threadStreams.delete(signal); - instrumentMessageStream(stream, span, recordOutputs); + instrumentMessageStream(stream, span, recordOutputs, callerSignal); }, error => { - // A caller abort is not a failed request. The stream still emits 'end', which ends the span as cancelled. + // An abort is not a failed request. The stream still emits 'end', which sets the status from the caller's signal. if (signal.aborted) { - instrumentMessageStream(stream, span, recordOutputs); + instrumentMessageStream(stream, span, recordOutputs, callerSignal); } else { end(error); } @@ -274,7 +281,7 @@ function wrapStreamResult( } if (stream === 'message-stream' && isMessageStream(result)) { - instrumentMessageStream(result, span, recordOutputs); + instrumentMessageStream(result, span, recordOutputs, getRequestSignal(data.arguments?.[1])); return true; } diff --git a/packages/server-utils/src/integrations/aws-sdk/services/bedrock-runtime.ts b/packages/server-utils/src/integrations/aws-sdk/services/bedrock-runtime.ts index f77e04f6f5ed..aab0bf1383c0 100644 --- a/packages/server-utils/src/integrations/aws-sdk/services/bedrock-runtime.ts +++ b/packages/server-utils/src/integrations/aws-sdk/services/bedrock-runtime.ts @@ -1,5 +1,5 @@ import type { Span } from '@sentry/core'; -import { debug, SPAN_STATUS_ERROR } from '@sentry/core'; +import { debug } from '@sentry/core'; import { GEN_AI_OPERATION_NAME, GEN_AI_PROVIDER_NAME, @@ -14,7 +14,12 @@ import { } from '@sentry/conventions/attributes'; import { GEN_AI_CHAT, GEN_AI_GENERATE_CONTENT } from '@sentry/conventions/op'; import type { StreamResponseState } from '../../../ai/core/utils'; -import { endStreamSpan, getTokenUsageAttributes, resolveAIRecordingOptions } from '../../../ai/core/utils'; +import { + endStreamSpan, + getErrorStatus, + getTokenUsageAttributes, + resolveAIRecordingOptions, +} from '../../../ai/core/utils'; import { isInsidePiAiSpan } from '../../../ai/pi-ai/providers'; import { DEBUG_BUILD } from '../../../debug-build'; import { @@ -459,7 +464,7 @@ async function* instrumentStream( yield item; } } catch (error) { - span.setStatus({ code: SPAN_STATUS_ERROR, message: abortSignal?.aborted ? 'cancelled' : 'internal_error' }); + span.setStatus(getErrorStatus(abortSignal)); throw error; } finally { endStreamSpan(span, state, false); diff --git a/packages/server-utils/src/integrations/google-genai.ts b/packages/server-utils/src/integrations/google-genai.ts index 2170f040a4e5..6ffae5f8df56 100644 --- a/packages/server-utils/src/integrations/google-genai.ts +++ b/packages/server-utils/src/integrations/google-genai.ts @@ -9,11 +9,12 @@ import { spanToJSON, startInactiveSpan, } from '@sentry/core'; -import { getGenAiSpanOp, resolveAIRecordingOptions } from '../ai/core/utils'; +import { getGenAiSpanOp, markCancelledIfAborted, resolveAIRecordingOptions } from '../ai/core/utils'; import { addPrivateRequestAttributes, addResponseAttributes, extractRequestAttributes, + getAbortSignal, resolveChatParams, } from '../ai/google-genai'; import { shouldSkipAiProviderSpan } from '../ai/pi-ai/providers'; @@ -54,6 +55,14 @@ function instrumentGoogleGenai(options: GoogleGenAIOptions): void { data => createGenAiSpan(data, operation, options), { beforeSpanEnd: (span, data) => { + if ('error' in data) { + const params = resolveChatParams( + operation, + data.arguments?.[0] as Record | undefined, + data.self, + ); + markCancelledIfAborted(span, getAbortSignal(params)); + } // Embeddings responses carry no content attributes. if (operation !== 'embeddings') { addResponseAttributes( @@ -152,7 +161,7 @@ function wrapStreamResult( { [Symbol.asyncIterator]: iterate }, span, recordOutputs ?? false, - (params?.config as { abortSignal?: AbortSignal } | undefined)?.abortSignal, + getAbortSignal(params), ); result[Symbol.asyncIterator] = () => instrumented; diff --git a/packages/server-utils/src/integrations/openai-compatible.ts b/packages/server-utils/src/integrations/openai-compatible.ts index 48bacae5bcf8..4e4a0301a8b8 100644 --- a/packages/server-utils/src/integrations/openai-compatible.ts +++ b/packages/server-utils/src/integrations/openai-compatible.ts @@ -7,7 +7,7 @@ import { startInactiveSpan, } from '@sentry/core'; import { GEN_AI_PROVIDER_NAME, SENTRY_ORIGIN } from '@sentry/conventions/attributes'; -import { getGenAiSpanOp, resolveAIRecordingOptions } from '../ai/core/utils'; +import { getGenAiSpanOp, getRequestSignal, markCancelledIfAborted, resolveAIRecordingOptions } from '../ai/core/utils'; import { wrapApiPromiseResponse } from '../ai/core/apiPromise'; import { addRequestAttributes, extractRequestAttributes } from '../ai/openai'; import { instrumentStream } from '../ai/openai/streaming'; @@ -56,6 +56,9 @@ export function createOpenAiCompatibleIntegration createGenAiSpan(data, operation, provider, options), { beforeSpanEnd: (span, data) => { + if ('error' in data) { + markCancelledIfAborted(span, getRequestSignal(data.arguments?.[1])); + } addResponseAttributes(span, data.result, resolveAIRecordingOptions(options).recordOutputs); }, deferSpanEnd: ({ span, data, end }) => diff --git a/packages/server-utils/src/integrations/openai.ts b/packages/server-utils/src/integrations/openai.ts index 88e58a172b0c..3dda9492956e 100644 --- a/packages/server-utils/src/integrations/openai.ts +++ b/packages/server-utils/src/integrations/openai.ts @@ -2,7 +2,7 @@ import { SENTRY_ORIGIN } from '@sentry/conventions/attributes'; import * as diagnosticsChannel from '../utils/diagnosticsChannel'; import type { IntegrationFn, Span, SpanAttributeValue } from '@sentry/core'; import { defineIntegration, getClient, hasSpanStreamingEnabled, startInactiveSpan } from '@sentry/core'; -import { getGenAiSpanOp, resolveAIRecordingOptions } from '../ai/core/utils'; +import { getGenAiSpanOp, getRequestSignal, markCancelledIfAborted, resolveAIRecordingOptions } from '../ai/core/utils'; import { addRequestAttributes, extractRequestAttributes } from '../ai/openai'; import { wrapApiPromiseResponse } from '../ai/core/apiPromise'; import { instrumentStream } from '../ai/openai/streaming'; @@ -44,6 +44,9 @@ function instrumentOpenai(options: OpenAiOptions): void { data => createGenAiSpan(data, operation, options), { beforeSpanEnd: (span, data) => { + if ('error' in data) { + markCancelledIfAborted(span, getRequestSignal(data.arguments?.[1])); + } addResponseAttributes(span, data.result, resolveAIRecordingOptions(options).recordOutputs, operation); }, deferSpanEnd: ({ span, data, end }) => diff --git a/packages/server-utils/test/ai/lib/tracing/anthropic-ai-tool-runner.test.ts b/packages/server-utils/test/ai/lib/tracing/anthropic-ai-tool-runner.test.ts index 5736a661188c..ff685f561b2e 100644 --- a/packages/server-utils/test/ai/lib/tracing/anthropic-ai-tool-runner.test.ts +++ b/packages/server-utils/test/ai/lib/tracing/anthropic-ai-tool-runner.test.ts @@ -256,7 +256,7 @@ describe('instrumentAnthropicAiClient tool runner', () => { expect(events).toHaveLength(0); }); - it('does not capture a ToolError or a call aborted by the runner', async () => { + it('does not capture a ToolError or a call aborted by the runner, but marks a timed-out one as an error', async () => { const toolError = Object.assign(new Error('bad input'), { name: 'ToolError' }); const refuses = tool('refuses', async () => { throw toolError; @@ -265,13 +265,27 @@ describe('instrumentAnthropicAiClient tool runner', () => { throw new Error('aborted'); }); - instrumentAnthropicAiClient(fakeClient()).beta.messages.toolRunner({ tools: [refuses, aborted] }); + const timedOut = tool('timedOut', async () => { + throw new Error('timed out'); + }); + + instrumentAnthropicAiClient(fakeClient()).beta.messages.toolRunner({ tools: [refuses, aborted, timedOut] }); await expect(refuses.run({}, { toolUse: { id: 'toolu_1' } })).rejects.toBe(toolError); await expect(aborted.run({}, { toolUse: { id: 'toolu_2' }, signal: AbortSignal.abort() })).rejects.toThrow( 'aborted', ); - expect(toolSpans().map(span => span.status)).toEqual(['internal_error', 'cancelled']); + await expect( + timedOut.run( + {}, + { + toolUse: { id: 'toolu_3' }, + signal: AbortSignal.abort(new DOMException('The operation timed out', 'TimeoutError')), + }, + ), + ).rejects.toThrow('timed out'); + + expect(toolSpans().map(span => span.status)).toEqual(['internal_error', 'cancelled', 'internal_error']); expect(events).toHaveLength(0); }); diff --git a/packages/server-utils/test/ai/lib/tracing/anthropic-ai-usage.test.ts b/packages/server-utils/test/ai/lib/tracing/anthropic-ai-usage.test.ts index 78569bc446ec..5d06a19c1e80 100644 --- a/packages/server-utils/test/ai/lib/tracing/anthropic-ai-usage.test.ts +++ b/packages/server-utils/test/ai/lib/tracing/anthropic-ai-usage.test.ts @@ -312,6 +312,25 @@ describe('instrumentMessageStream', () => { }); }); + it.each([ + { reason: 'the caller aborts its signal', signal: AbortSignal.abort(), status: 'cancelled' }, + { + reason: "the caller's AbortSignal.timeout() fired", + signal: AbortSignal.abort(new DOMException('The operation timed out', 'TimeoutError')), + status: 'internal_error', + }, + ])('ends an aborted stream as $status when $reason', ({ signal, status }) => { + const { span } = createMockSpan(); + const setStatus = vi.fn(); + Object.assign(span, { setStatus }); + const stream = instrumentMessageStream(Object.assign(new EventEmitter(), { aborted: true }), span, false, signal); + + stream.emit('end'); + + expect(setStatus).toHaveBeenCalledTimes(1); + expect(setStatus).toHaveBeenCalledWith({ code: SPAN_STATUS_ERROR, message: status }); + }); + it('leaves the status alone when the body closes early without an abort', () => { const { span } = createMockSpan(); const setStatus = vi.fn(); diff --git a/packages/server-utils/test/ai/lib/tracing/caller-abort.test.ts b/packages/server-utils/test/ai/lib/tracing/caller-abort.test.ts new file mode 100644 index 000000000000..e9e999e1b2fe --- /dev/null +++ b/packages/server-utils/test/ai/lib/tracing/caller-abort.test.ts @@ -0,0 +1,92 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { getMainCarrier, setCurrentClient, spanToStaticSpanJSON } from '@sentry/core'; +import type { Span } from '@sentry/core'; +import { instrumentAnthropicAiClient } from '../../../../src/ai/anthropic-ai'; +import { instrumentGoogleGenAIClient } from '../../../../src/ai/google-genai'; +import { instrumentOpenAiClient } from '../../../../src/ai/openai'; +import { getDefaultTestClientOptions, TestClient } from '../../../mocks/client'; + +type Call = (signal: AbortSignal, stream: boolean) => Promise; + +// Each provider's SDK rejects with an `AbortError` for both a caller abort and its own timeout. +function rejectingMethod(): ReturnType { + return vi.fn().mockRejectedValue(new DOMException('This operation was aborted', 'AbortError')); +} + +const providers: Array<{ name: string; call: () => Call }> = [ + { + name: 'openai', + call: () => { + const client = instrumentOpenAiClient({ chat: { completions: { create: rejectingMethod() } } }); + return (signal, stream) => client.chat.completions.create({ model: 'gpt-4o', messages: [], stream }, { signal }); + }, + }, + { + name: 'anthropic', + call: () => { + const client = instrumentAnthropicAiClient({ messages: { create: rejectingMethod() } }); + return (signal, stream) => + client.messages.create({ model: 'claude-sonnet-4-5', max_tokens: 10, messages: [], stream }, { signal }); + }, + }, + { + name: 'google-genai', + call: () => { + const client = instrumentGoogleGenAIClient({ + models: { generateContent: rejectingMethod(), generateContentStream: rejectingMethod() }, + }); + return (signal, stream) => { + const params = { model: 'gemini-2.5-flash', contents: 'Hello', config: { abortSignal: signal } }; + return stream ? client.models.generateContentStream(params) : client.models.generateContent(params); + }; + }, + }, +]; + +describe.each(providers)('$name caller aborts', ({ call }) => { + let endedSpans: Span[]; + + beforeEach(() => { + getMainCarrier().__SENTRY__ = undefined; + const client = new TestClient( + getDefaultTestClientOptions({ + dsn: 'https://public@dsn.ingest.sentry.io/1337', + tracesSampleRate: 1, + traceLifecycle: 'static', + }), + ); + setCurrentClient(client); + client.init(); + endedSpans = []; + client.on('spanEnd', span => endedSpans.push(span)); + }); + + afterEach(() => { + getMainCarrier().__SENTRY__ = undefined; + }); + + async function statusAfterRejection(signal: AbortSignal, stream: boolean): Promise { + await expect(call()(signal, stream)).rejects.toThrow('This operation was aborted'); + expect(endedSpans).toHaveLength(1); + return spanToStaticSpanJSON(endedSpans[0]!).status; + } + + describe.each([ + { kind: 'streaming', stream: true }, + { kind: 'non-streaming', stream: false }, + ])('$kind request', ({ stream }) => { + it('marks the span as cancelled when the caller aborts before the response', async () => { + expect(await statusAfterRejection(AbortSignal.abort(), stream)).toBe('cancelled'); + }); + + it("marks the span as errored when the caller's own timeout fired", async () => { + const timeoutSignal = AbortSignal.abort(new DOMException('The operation timed out', 'TimeoutError')); + + expect(await statusAfterRejection(timeoutSignal, stream)).toBe('internal_error'); + }); + + it("marks the span as errored when the request fails without the caller's signal firing", async () => { + expect(await statusAfterRejection(new AbortController().signal, stream)).toBe('internal_error'); + }); + }); +}); diff --git a/packages/server-utils/test/ai/lib/tracing/stream-errors.test.ts b/packages/server-utils/test/ai/lib/tracing/stream-errors.test.ts index 30c37aec9ba5..8e7d1f490169 100644 --- a/packages/server-utils/test/ai/lib/tracing/stream-errors.test.ts +++ b/packages/server-utils/test/ai/lib/tracing/stream-errors.test.ts @@ -176,6 +176,18 @@ describe('google-genai stream aborts', () => { expect(setStatus).toHaveBeenCalledWith({ code: SPAN_STATUS_ERROR, message: 'cancelled' }); }); + it("marks the span as errored when the caller's own timeout fired", async () => { + const { span, setStatus } = createMockSpan(); + const timeoutSignal = AbortSignal.abort(new DOMException('The operation timed out', 'TimeoutError')); + + await expect(drain(instrumentGoogleGenAIStream(abortedStream(), span, false, timeoutSignal))).rejects.toBe( + abortError, + ); + + expect(setStatus).toHaveBeenCalledTimes(1); + expect(setStatus).toHaveBeenCalledWith({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); + }); + it("marks the span as errored when an abort didn't come from the caller's signal", async () => { const { span, setStatus } = createMockSpan();