fix(server-utils): Trace streamObject calls on AI SDK v4-v6

dev-packages/node-integration-tests/suites/tracing/vercelai/v4-static/scenario-stream-object.mjs

@@ -0,0 +1,38 @@
+import * as Sentry from '@sentry/node';
+import { streamObject } from 'ai';
+import { convertArrayToReadableStream, MockLanguageModelV1 } from 'ai/test';
+import { z } from 'zod';
+
+async function run() {
+  await Sentry.startSpan({ op: 'function', name: 'main' }, async () => {
+    const result = streamObject({
+      model: new MockLanguageModelV1({
+        defaultObjectGenerationMode: 'json',
+        doStream: async () => ({
+          stream: convertArrayToReadableStream([
+            { type: 'response-metadata', id: 'resp-1', modelId: 'mock-model-id' },
+            { type: 'text-delta', textDelta: '{ "name": "John Doe",' },
+            { type: 'text-delta', textDelta: ' "age": 30 }' },
+            {
+              type: 'finish',
+              finishReason: 'stop',
+              usage: { promptTokens: 15, completionTokens: 25 },
+            },
+          ]),
+          rawCall: { rawPrompt: null, rawSettings: {} },
+        }),
+      }),
+      schema: z.object({ name: z.string(), age: z.number() }),
+      prompt: 'Generate a person object',
+    });
+
+    // streamObject returns synchronously; drive the lazy stream to completion so the spans finish.
+    for await (const _part of result.partialObjectStream) {
+      void _part;
+    }
+  });
+
+  await Sentry.flush(2000);
+}
+
+run();

dev-packages/node-integration-tests/suites/tracing/vercelai/v4-static/test.ts

@@ -697,6 +697,36 @@ describe('Vercel AI integration (v4)', () => {
     });
   });
 
+  createEsmAndCjsTests(__dirname, 'scenario-stream-object.mjs', 'instrument.mjs', (createRunner, test) => {
+    test('creates ai spans for streamObject (doStream)', async () => {
+      await createRunner()
+        .expect({ transaction: { transaction: 'main' } })
+        .expect({
+          span: container => {
+            expect(container.items).toHaveLength(2);
+
+            const invokeAgentSpan = container.items.find(span => span.name === 'invoke_agent');
+            expect(invokeAgentSpan!.status).toBe('ok');
+            expect(invokeAgentSpan!.attributes[SENTRY_OP].value).toBe('gen_ai.invoke_agent');
+            expect(invokeAgentSpan!.attributes[SENTRY_ORIGIN].value).toBe(expectedOrigin);
+            expect(invokeAgentSpan!.attributes['vercel.ai.operationId'].value).toBe('ai.streamObject');
+            expect(invokeAgentSpan!.attributes[GEN_AI_USAGE_INPUT_TOKENS].value).toBe(15);
+            expect(invokeAgentSpan!.attributes[GEN_AI_USAGE_OUTPUT_TOKENS].value).toBe(25);
+            expect(invokeAgentSpan!.attributes[GEN_AI_USAGE_TOTAL_TOKENS].value).toBe(40);
+
+            const generateContentSpan = container.items.find(span => span.name === 'generate_content mock-model-id');
+            expect(generateContentSpan!.status).toBe('ok');
+            expect(generateContentSpan!.parent_span_id).toBe(invokeAgentSpan!.span_id);
+            expect(generateContentSpan!.attributes['vercel.ai.operationId'].value).toBe('ai.streamObject.doStream');
+            expect(generateContentSpan!.attributes[GEN_AI_USAGE_INPUT_TOKENS].value).toBe(15);
+            expect(generateContentSpan!.attributes[GEN_AI_USAGE_OUTPUT_TOKENS].value).toBe(25);
+          },
+        })
+        .start()
+        .completed();
+    });
+  });
+
   createEsmAndCjsTests(__dirname, 'scenario-generate-object.mjs', 'instrument-with-pii.mjs', (createRunner, test) => {
     test('captures generateObject spans with schema attributes', async () => {
       await createRunner()

dev-packages/node-integration-tests/suites/tracing/vercelai/v6_v7/scenario-stream-object-invalid.mjs

@@ -0,0 +1,40 @@
+import * as Sentry from '@sentry/node';
+import { streamObject } from 'ai';
+import { MockLanguageModelV3, simulateReadableStream } from 'ai/test';
+import { z } from 'zod';
+
+async function run() {
+  await Sentry.startSpan({ op: 'function', name: 'main' }, async () => {
+    const result = streamObject({
+      model: new MockLanguageModelV3({
+        doStream: async () => ({
+          stream: simulateReadableStream({
+            chunks: [
+              { type: 'stream-start', warnings: [] },
+              { type: 'text-start', id: '0' },
+              { type: 'text-delta', id: '0', delta: 'not json' },
+              { type: 'text-end', id: '0' },
+              {
+                type: 'finish',
+                finishReason: { unified: 'stop', raw: 'stop' },
+                usage: {
+                  inputTokens: { total: 15, noCache: 15, cached: 0 },
+                  outputTokens: { total: 25, noCache: 25, cached: 0 },
+                  totalTokens: { total: 40, noCache: 40, cached: 0 },
+                },
+              },
+            ],
+          }),
+        }),
+      }),
+      schema: z.object({ name: z.string(), age: z.number() }),
+      prompt: 'Generate a person object',
+    });
+
+    for await (const _part of result.partialObjectStream) {
+      void _part;
+    }
+  });
+}
+
+run();

dev-packages/node-integration-tests/suites/tracing/vercelai/v6_v7/scenario-stream-object.mjs

@@ -0,0 +1,42 @@
+import * as Sentry from '@sentry/node';
+import { streamObject } from 'ai';
+import { MockLanguageModelV3, simulateReadableStream } from 'ai/test';
+import { z } from 'zod';
+
+async function run() {
+  await Sentry.startSpan({ op: 'function', name: 'main' }, async () => {
+    const result = streamObject({
+      model: new MockLanguageModelV3({
+        doStream: async () => ({
+          stream: simulateReadableStream({
+            chunks: [
+              { type: 'stream-start', warnings: [] },
+              { type: 'text-start', id: '0' },
+              { type: 'text-delta', id: '0', delta: '{ "name": "John Doe",' },
+              { type: 'text-delta', id: '0', delta: ' "age": 30 }' },
+              { type: 'text-end', id: '0' },
+              {
+                type: 'finish',
+                finishReason: { unified: 'stop', raw: 'stop' },
+                usage: {
+                  inputTokens: { total: 15, noCache: 15, cached: 0 },
+                  outputTokens: { total: 25, noCache: 25, cached: 0 },
+                  totalTokens: { total: 40, noCache: 40, cached: 0 },
+                },
+              },
+            ],
+          }),
+        }),
+      }),
+      schema: z.object({ name: z.string(), age: z.number() }),
+      prompt: 'Generate a person object',
+    });
+
+    // streamObject returns synchronously; drive the lazy stream to completion so the spans finish.
+    for await (const _part of result.partialObjectStream) {
+      void _part;
+    }
+  });
+}
+
+run();

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

@@ -1236,6 +1236,85 @@ describe.each(matrix)('Vercel AI integration (version %s)', (version, vercelAiVe
     },
   );
 
+  createEsmTests(
+    __dirname,
+    'scenario-stream-object.mjs',
+    'instrument.mjs',
+    (createRunner, test) => {
+      // Like `generateObject`, v7's native channel never publishes `streamObject`; v6 uses orchestrion.
+      test.skipIf(version === '7')(
+        'creates streamObject spans with the model call parented to invoke_agent',
+        async () => {
+          await createRunner()
+            .expect({
+              span: container => {
+                const segment = container.items.find(span => span.is_segment && span.name === 'main');
+                expect(segment).toBeDefined();
+                const spans = container.items.filter(
+                  span => span.attributes[SENTRY_ORIGIN]?.value === 'auto.ai.vercelai',
+                );
+
+                const invokeAgent = spans.find(span => span.attributes[SENTRY_OP]?.value === 'gen_ai.invoke_agent')!;
+                expect(invokeAgent.status).toBe('ok');
+                expect(invokeAgent.attributes['vercel.ai.operationId']?.value).toBe('ai.streamObject');
+                expect(invokeAgent.attributes[GEN_AI_USAGE_INPUT_TOKENS]?.value).toBe(15);
+                expect(invokeAgent.attributes[GEN_AI_USAGE_OUTPUT_TOKENS]?.value).toBe(25);
+                expect(invokeAgent.attributes[GEN_AI_USAGE_TOTAL_TOKENS]?.value).toBe(40);
+
+                const generateContent = spans.find(
+                  span => span.attributes[SENTRY_OP]?.value === 'gen_ai.generate_content',
+                )!;
+                expect(generateContent.status).toBe('ok');
+                expect(generateContent.parent_span_id).toBe(invokeAgent.span_id);
+                expect(generateContent.attributes['vercel.ai.operationId']?.value).toBe('ai.streamObject.doStream');
+                expect(generateContent.attributes[GEN_AI_REQUEST_MODEL]?.value).toBe('mock-model-id');
+                expect(generateContent.attributes[GEN_AI_USAGE_INPUT_TOKENS]?.value).toBe(15);
+                expect(generateContent.attributes[GEN_AI_USAGE_OUTPUT_TOKENS]?.value).toBe(25);
+              },
+            })
+            .unordered()
+            .start()
+            .completed();
+        },
+      );
+    },
+    {
+      additionalDependencies: {
+        ai: vercelAiVersion,
+      },
+    },
+  );
+
+  createEsmTests(
+    __dirname,
+    'scenario-stream-object-invalid.mjs',
+    'instrument.mjs',
+    (createRunner, test) => {
+      test.skipIf(version === '7')('marks streamObject as failed when the output fails the schema', async () => {
+        await createRunner()
+          .expect({
+            span: container => {
+              const invokeAgent = container.items.find(
+                span =>
+                  span.attributes[SENTRY_ORIGIN]?.value === 'auto.ai.vercelai' &&
+                  span.attributes[SENTRY_OP]?.value === 'gen_ai.invoke_agent',
+              )!;
+              expect(invokeAgent.attributes['vercel.ai.operationId']?.value).toBe('ai.streamObject');
+              expect(invokeAgent.status).toBe('error');
+            },
+          })
+          .unordered()
+          .start()
+          .completed();
+      });
+    },
+    {
+      additionalDependencies: {
+        ai: vercelAiVersion,
+      },
+    },
+  );
+
   createEsmTests(
     __dirname,
     'scenario-aborted-stream-text.mjs',

packages/server-utils/src/integrations/vercel-ai/vercel-ai-dc-subscriber.ts

@@ -114,6 +114,7 @@ const ROOT_OPERATION_TYPES = new Set<ChannelEventType>([
   'generateText',
   'streamText',
   'generateObject',
+  'streamObject',
   'embed',
   'embedMany',
   'rerank',
@@ -235,6 +236,7 @@ export type ChannelEventType =
   | 'generateText'
   | 'streamText'
   | 'generateObject'
+  | 'streamObject'
   | 'step'
   | 'languageModelCall'
   | 'executeTool'
@@ -463,10 +465,16 @@ export function createSpanFromMessage(
     case 'generateText':
     case 'streamText':
     case 'generateObject':
-      // `generateObject` builds the same `invoke_agent` span as `generateText` (non-streaming); its
-      // distinct `ai.generateObject` operationId rides on `event.operationId`. The JSON-schema attribute
-      // the OTel path derives from the SDK's Zod schema is not reconstructed on the channel path.
-      return buildInvokeAgentSpan(event, baseAttributes, recordInputs, callId, type === 'streamText');
+    case 'streamObject':
+      // The object variants' operationId rides on `event.operationId`. The JSON-schema attribute the OTel
+      // path derives from the SDK's Zod schema is not reconstructed on the channel path.
+      return buildInvokeAgentSpan(
+        event,
+        baseAttributes,
+        recordInputs,
+        callId,
+        type === 'streamText' || type === 'streamObject',
+      );
     case 'languageModelCall':
       _INTERNAL_skipAiProviderWrapping([WORKERS_AI_INTEGRATION_NAME]);
 

packages/server-utils/src/integrations/vercel-ai/vercel-ai-orchestrion-subscriber.ts

@@ -150,6 +150,7 @@ export function subscribeVercelAiOrchestrionChannels(
     bindOperation(tracingChannel, CHANNELS.VERCEL_AI_GENERATE_TEXT, buildTextMessage('generateText'), options);
     bindOperation(tracingChannel, CHANNELS.VERCEL_AI_STREAM_TEXT, buildTextMessage('streamText'), options);
     bindOperation(tracingChannel, CHANNELS.VERCEL_AI_GENERATE_OBJECT, buildTextMessage('generateObject'), options);
+    bindOperation(tracingChannel, CHANNELS.VERCEL_AI_STREAM_OBJECT, buildTextMessage('streamObject'), options);
     bindOperation(
       tracingChannel,
       CHANNELS.VERCEL_AI_EMBED,
@@ -305,33 +306,35 @@ function bindOperation(
           message.result = message.type === 'executeTool' ? { output: data.result } : data.result;
           enrichSpanOnEnd(span, message, options);
         }
-        // A `streamText` model call runs after this (synchronously-returning) operation's span has
-        // already ended, so its `callId` entry must outlive the operation — it's cleared once the model
-        // call settles (see `patchModelMethod`). Every other operation can clear here.
-        if (message.type !== 'streamText') {
+        // A `streamText`/`streamObject` model call runs after this (synchronously-returning) operation's
+        // span has already ended, so its `callId` entry must outlive the operation — it's cleared once the
+        // model call settles (see `patchModelMethod`). Every other operation can clear here.
+        if (!isStreamOperation(message.type)) {
           clearOperationId(message);
         }
         messages.delete(data);
       },
-      // `streamText` returns synchronously, so its operation span would otherwise end before the stream
-      // drains — losing the aggregate usage/output. Defer the end and await the result's completion
-      // promises (`totalUsage`/`text`/…, which resolve on drain), mirroring how v7's channel defers the
-      // operation span on the SDK's total-usage promise.
-      deferSpanEnd: ({ data, end }) => deferStreamTextOperationEnd(data, end),
+      // `streamText`/`streamObject` return synchronously, so their operation span would otherwise end
+      // before the stream drains — losing the aggregate usage/output. Defer the end and await the result's
+      // completion promises (`totalUsage`/`text`/…, which resolve on drain), mirroring how v7's channel
+      // defers the operation span on the SDK's total-usage promise.
+      deferSpanEnd: ({ data, end }) => deferStreamOperationEnd(data, end),
     },
   );
 }
 
 /**
- * Keep a streamed `streamText` operation span open until the stream drains, then enrich it from the
- * `StreamTextResult`'s completion promises (usage/output/finish reason) and end it. Returns `false` for
- * anything that isn't a streamed `streamText` result, so the helper ends the span as usual.
+ * Keep a streamed `streamText`/`streamObject` operation span open until the stream drains, then enrich it
+ * from the result's completion promises (usage/output/finish reason) and end it. A `StreamObjectResult`
+ * has no `text`/`toolCalls`/`totalUsage`, so those resolve to `undefined`. Returns `false` for anything
+ * that isn't a streamed result, so the helper ends the span as usual.
  */
-function deferStreamTextOperationEnd(
+function deferStreamOperationEnd(
   data: TracingChannelPayloadWithSpan<OrchestrionChannelContext>,
   end: (error?: unknown) => void,
 ): boolean {
-  if (messages.get(data)?.type !== 'streamText' || 'error' in data || !isStreamingResult(data.result)) {
+  const type = messages.get(data)?.type;
+  if (!type || !isStreamOperation(type) || 'error' in data || !isStreamingResult(data.result)) {
     return false;
   }
 
@@ -344,6 +347,8 @@ function deferStreamTextOperationEnd(
         streamResult.toolCalls,
         streamResult.finishReason,
         streamResult.response,
+        // Only `object` rejects when the output fails the schema; awaiting it fails the span like `generateObject`.
+        type === 'streamObject' ? streamResult.object : undefined,
       ]);
       // Feed the resolved values back through the shared enrichment: `beforeSpanEnd` reads `data.result`.
       data.result = { usage, text, toolCalls, finishReason, response };
@@ -356,7 +361,11 @@ function deferStreamTextOperationEnd(
   return true;
 }
 
-/** A `StreamTextResult` exposes its aggregates as promises (`totalUsage`/`usage`); a settled result does not. */
+function isStreamOperation(type: VercelAiChannelMessage['type']): boolean {
+  return type === 'streamText' || type === 'streamObject';
+}
+
+/** A streamed result exposes its aggregates as promises (`totalUsage`/`usage`); a settled result does not. */
 function isStreamingResult(result: unknown): result is Record<string, PromiseLike<unknown> | undefined> {
   return isObjectLike(result) && (isThenable(result.totalUsage) || isThenable(result.usage));
 }
@@ -502,8 +511,8 @@ function patchModelMethod(
       return Promise.resolve(original.apply(this, args));
     }
 
-    // `streamText` ends its operation span synchronously, so its `callId` entry was deliberately left in
-    // place for this later model call; drop it now that we've used it.
+    // `streamText`/`streamObject` end their operation span synchronously, so the `callId` entry was
+    // deliberately left in place for this later model call; drop it now that we've used it.
     const clearStreamCallId = (): void => {
       if (method === 'doStream' && callId) {
         clearOperationCallId(callId);
@@ -525,7 +534,7 @@ function patchModelMethod(
         // A streamed model call resolves to `{ stream, ... }` before the stream is consumed, so its
         // usage/finish/output only arrive as the stream drains. Tap it (same helper as the v7 path) and
         // defer ending this model-call span until then. The parent `invoke_agent` span is enriched
-        // separately by `deferStreamTextOperationEnd`, which awaits the operation's own result promises.
+        // separately by `deferStreamOperationEnd`, which awaits the operation's own result promises.
         if (method === 'doStream' && isObjectLike(value) && isReadableStream(value.stream)) {
           value.stream = tapModelCallStream(
             value.stream,
@@ -654,7 +663,7 @@ function patchToolExecute(
   };
 }
 
-function buildTextMessage(type: 'generateText' | 'streamText' | 'generateObject'): MessageBuilder {
+function buildTextMessage(type: 'generateText' | 'streamText' | 'generateObject' | 'streamObject'): MessageBuilder {
   return (options, telemetry) => ({
     type,
     event: {