fix(v10/node): Stop leaking unhandled rejections on aborted Vercel AI streams

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();

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,
+      },
+    },
+  );
 });

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<object> };
   if (!Array.isArray(resultObj.content)) {
     return;

packages/server-utils/src/tracing-channel.ts

@@ -23,6 +23,9 @@ export type TracingChannelPayloadWithSpan<TData extends object> = 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<TData extends object = object> {
   deferSpanEnd?: (args: {
     span: Span;
     data: TracingChannelPayloadWithSpan<TData>;
-    /** 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<TData extends object>(
       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);

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 });