diff --git a/dev-packages/node-integration-tests/suites/tracing/vercelai/v6_v7/scenario-aborted-stream-text.mjs b/dev-packages/node-integration-tests/suites/tracing/vercelai/v6_v7/scenario-aborted-stream-text.mjs new file mode 100644 index 000000000000..d7f4e7755f32 --- /dev/null +++ b/dev-packages/node-integration-tests/suites/tracing/vercelai/v6_v7/scenario-aborted-stream-text.mjs @@ -0,0 +1,34 @@ +import * as Sentry from '@sentry/node'; +import { streamText } from 'ai'; +import { MockLanguageModelV3 } from 'ai/test'; + +async function run() { + await Sentry.startSpan({ op: 'function', name: 'main' }, async () => { + const controller = new AbortController(); + + // Abort the moment the model is asked for a stream, so the operation fails before its first + // chunk. `fetch()` rejects with the signal's reason on abort, and `@hono/node-server` aborts + // with a plain string — so there is no `AbortError` name to suppress by. + const model = new MockLanguageModelV3({ + doStream: ({ abortSignal }) => + new Promise((_, reject) => { + abortSignal.addEventListener('abort', () => reject(abortSignal.reason), { once: true }); + controller.abort('Client connection prematurely closed.'); + }), + }); + + const result = streamText({ + experimental_telemetry: { isEnabled: true, recordInputs: true, recordOutputs: true }, + maxRetries: 0, + model, + prompt: 'Stream me a response', + abortSignal: controller.signal, + }); + + for await (const _part of result.textStream) { + void _part; + } + }); +} + +run(); diff --git a/dev-packages/node-integration-tests/suites/tracing/vercelai/v6_v7/test.ts b/dev-packages/node-integration-tests/suites/tracing/vercelai/v6_v7/test.ts index 6963de59bcc6..ddb5dcdfa146 100644 --- a/dev-packages/node-integration-tests/suites/tracing/vercelai/v6_v7/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/vercelai/v6_v7/test.ts @@ -935,4 +935,20 @@ describe.each(matrix)('Vercel AI integration (version %s)', (version, vercelAiVe }, }, ); + + createEsmTests( + __dirname, + 'scenario-aborted-stream-text.mjs', + 'instrument.mjs', + (createRunner, test) => { + test('aborting a stream with a non-AbortError reason leaves no unhandled rejection', async () => { + await createRunner().ensureNoErrorOutput().start().completed(); + }); + }, + { + additionalDependencies: { + ai: vercelAiVersion, + }, + }, + ); }); diff --git a/packages/node/src/integrations/tracing/vercelai/instrumentation.ts b/packages/node/src/integrations/tracing/vercelai/instrumentation.ts index 7b3deb2b820f..1edc04352f2f 100644 --- a/packages/node/src/integrations/tracing/vercelai/instrumentation.ts +++ b/packages/node/src/integrations/tracing/vercelai/instrumentation.ts @@ -80,6 +80,13 @@ export function processToolCallResults(result: unknown): void { return; } + // A streamed result derives `content` from a stream that hasn't settled, so reading it both + // starts draining the stream and hands back a promise we drop — which surfaces as an unhandled + // rejection once the stream fails. Streamed tool errors were never processed here anyway. + if (typeof (result as { consumeStream?: unknown }).consumeStream === 'function') { + return; + } + const resultObj = result as { content: Array }; if (!Array.isArray(resultObj.content)) { return; diff --git a/packages/server-utils/src/tracing-channel.ts b/packages/server-utils/src/tracing-channel.ts index 097a3da6018a..b9fb9ab5eb0e 100644 --- a/packages/server-utils/src/tracing-channel.ts +++ b/packages/server-utils/src/tracing-channel.ts @@ -23,6 +23,9 @@ export type TracingChannelPayloadWithSpan = TData & { * The context's active store value, used to restore the context for asyncStart continuations for callback-based tracing. */ _sentryCallerStore?: unknown; + + /** Set by Node's tracing channel when the traced operation failed. */ + error?: unknown; }; /* @@ -63,7 +66,7 @@ export interface TracingChannelLifeCycleOptions { deferSpanEnd?: (args: { span: Span; data: TracingChannelPayloadWithSpan; - /** Ends the span: `end()` on success, `end(error)` on failure. Idempotent. */ + /** Ends the span: `end()` on success, `end(error)` on failure (which marks `data` as errored). Idempotent. */ end: (error?: unknown) => void; }) => boolean; @@ -145,6 +148,9 @@ export function bindTracingChannelToSpan( ended = true; if (error !== undefined) { annotateSpanError(span, error); + // Without this the payload still looks successful, so `beforeSpanEnd` enriches the span from + // a `result` the operation never produced. + data.error = error; } endBoundSpan(data, beforeSpanEnd); diff --git a/packages/server-utils/test/tracing-channel.test.ts b/packages/server-utils/test/tracing-channel.test.ts index 81c67b34e82f..960b1db54a7c 100644 --- a/packages/server-utils/test/tracing-channel.test.ts +++ b/packages/server-utils/test/tracing-channel.test.ts @@ -838,6 +838,32 @@ describe('bindTracingChannelToSpan', () => { expect(endSpy).toHaveBeenCalledTimes(1); }); + it('`end(error)` marks the payload as failed for `beforeSpanEnd`', () => { + installTestAsyncContextStrategy(); + initTestClient(); + const span = startInactiveSpan({ name: 'channel-span' }); + const beforeSpanEnd = vi.fn(); + let captured: (error?: unknown) => void = () => undefined; + const { channel } = bindTracingChannelToSpan( + tracingChannel<{ operation: string }>('test:defer:payload-error'), + () => span, + { + beforeSpanEnd, + deferSpanEnd({ end }) { + captured = end; + return true; + }, + }, + ); + + channel.traceSync(() => 'stream', { operation: 'read' }); + const error = new Error('stream aborted'); + captured(error); + + expect(beforeSpanEnd).toHaveBeenCalledTimes(1); + expect(beforeSpanEnd).toHaveBeenCalledWith(span, expect.objectContaining({ error })); + }); + it('captures the error via `end(error)` when `captureError` is set', () => { const captureExceptionSpy = vi.spyOn(SentryCore, 'captureException').mockReturnValue('event-id'); const { end } = setupDeferred('test:defer:capture', { captureError: true });