feat(core): Emit gen_ai.output.messages from Workers AI instrumentation

JPeer264 · claude · JPeer264 · commit 609fe63f95cc · 2026-07-20T10:07:18.000+02:00
The Sentry product reads model output from `gen_ai.output.messages`, treating
`gen_ai.response.text` and `gen_ai.response.tool_calls` as deprecated. Relay
migrates `response.text` into `output.messages` at ingestion, but the tool-calls
half of that migration is lossy, so tool-call turns rendered an empty Output tab.

Build `gen_ai.output.messages` directly in both the streaming and non-streaming
paths (mirroring the Vercel AI integration), normalizing the OpenAI-compatible
(`function.{name,arguments}`) and native (top-level `name`/`arguments`) tool-call
shapes. The deprecated attributes are still written for backward compatibility.

The streaming parser is also extended to read the OpenAI-compatible SSE shape
(`choices[].delta.content` / `choices[].delta.tool_calls`) that models routed
through the OpenAI-compatible endpoint emit, which previously dropped both text
and tool calls while still capturing usage.

Co-Authored-By: Claude Opus 4.8 &lt;noreply@anthropic.com&gt;
diff --git a/packages/core/src/tracing/workers-ai/streaming.ts b/packages/core/src/tracing/workers-ai/streaming.ts
@@ -2,17 +2,87 @@ import { SPAN_STATUS_ERROR } from '../../tracing';
 import type { Span } from '../../types/span';
 import { endStreamSpan, type StreamResponseState } from '../ai/utils';
 import type { WorkersAiUsage } from './types';
+import { setOutputMessagesAttribute } from './utils';
+
+interface WorkersAiStreamingToolCall {
+  index?: number;
+  id?: string;
+  type?: string;
+  function?: { name?: string; arguments?: string };
+  // Some Workers AI models stream tool calls with the name/arguments at the top
+  // level of the tool-call object instead of nested under `function`.
+  name?: string;
+  arguments?: string;
+}
 
 interface WorkersAiStreamChunk {
+  // Native Workers AI streaming shape (`env.AI.run` with `stream: true`).
   response?: unknown;
-  usage?: WorkersAiUsage;
   tool_calls?: unknown[];
+  // OpenAI-compatible streaming shape emitted for models routed through the
+  // OpenAI-compatible endpoint (e.g. via `workers-ai-provider`).
+  choices?: Array<{
+    delta?: { content?: unknown; tool_calls?: WorkersAiStreamingToolCall[] };
+    finish_reason?: unknown;
+  }>;
+  usage?: WorkersAiUsage & { prompt_tokens?: number; completion_tokens?: number; total_tokens?: number };
+}
+
+/**
+ * Accumulate a fragmented OpenAI-compatible tool call (delivered across multiple
+ * `choices[].delta.tool_calls` chunks) into the index-keyed accumulator.
+ */
+function accumulateStreamingToolCalls(
+  toolCalls: WorkersAiStreamingToolCall[],
+  accumulator: Record<number, WorkersAiStreamingToolCall>,
+): void {
+  for (const toolCall of toolCalls) {
+    // Normalize both shapes: name/arguments nested under `function`, or at the top level.
+    const name = toolCall.function?.name ?? toolCall.name;
+    const args = toolCall.function?.arguments ?? toolCall.arguments;
+
+    // A tool call must carry at least a name or argument fragment to be meaningful.
+    if (name == null && args == null) {
+      continue;
+    }
+
+    const index = toolCall.index ?? 0;
+    const existing = accumulator[index];
+
+    if (!existing) {
+      accumulator[index] = {
+        index,
+        id: toolCall.id,
+        type: toolCall.type,
+        function: {
+          name,
+          arguments: args ?? '',
+        },
+      };
+    } else if (existing.function) {
+      if (name && !existing.function.name) {
+        existing.function.name = name;
+      }
+      if (args) {
+        existing.function.arguments = `${existing.function.arguments ?? ''}${args}`;
+      }
+    }
+  }
 }
 
 /**
  * Parse a single SSE line (`data: {...}`) and accumulate its data into the streaming state.
+ *
+ * Handles both the native Workers AI shape (top-level `response`/`tool_calls`) and the
+ * OpenAI-compatible shape (`choices[].delta.content`/`choices[].delta.tool_calls`), because
+ * the same `run()` call transparently yields either format depending on the model.
  */
-function processLine(line: string, state: StreamResponseState, recordOutputs: boolean): void {
+function processLine(
+  line: string,
+  state: StreamResponseState,
+  recordOutputs: boolean,
+  toolCallAccumulator: Record<number, WorkersAiStreamingToolCall>,
+): void {
   const trimmed = line.trim();
   if (!trimmed.startsWith('data:')) {
     return;
@@ -49,6 +119,20 @@ function processLine(line: string, state: StreamResponseState, recordOutputs: bo
   if (recordOutputs && Array.isArray(parsed.tool_calls) && parsed.tool_calls.length > 0) {
     state.toolCalls.push(...parsed.tool_calls);
   }
+
+  if (Array.isArray(parsed.choices)) {
+    for (const choice of parsed.choices) {
+      if (recordOutputs && typeof choice.delta?.content === 'string' && choice.delta.content) {
+        state.responseTexts.push(choice.delta.content);
+      }
+      if (recordOutputs && Array.isArray(choice.delta?.tool_calls)) {
+        accumulateStreamingToolCalls(choice.delta.tool_calls, toolCallAccumulator);
+      }
+      if (typeof choice.finish_reason === 'string') {
+        state.finishReasons.push(choice.finish_reason);
+      }
+    }
+  }
 }
 
 /**
@@ -76,6 +160,10 @@ export function instrumentWorkersAiStream(
     totalTokens: undefined,
   };
 
+  // OpenAI-compatible tool calls arrive fragmented across chunks and are keyed by index;
+  // accumulate them here and flatten into `state.toolCalls` once the stream ends.
+  const toolCallAccumulator: Record<number, WorkersAiStreamingToolCall> = {};
+
   let buffer = '';
   let spanEnded = false;
 
@@ -84,6 +172,21 @@ export function instrumentWorkersAiStream(
       return;
     }
     spanEnded = true;
+
+    if (recordOutputs) {
+      const accumulatedToolCalls = Object.values(toolCallAccumulator);
+      if (accumulatedToolCalls.length > 0) {
+        state.toolCalls.push(...accumulatedToolCalls);
+      }
+
+      // Set the authoritative `gen_ai.output.messages` alongside the deprecated response
+      // attributes `endStreamSpan` writes, so tool calls survive Relay's lossy migration.
+      setOutputMessagesAttribute(span, {
+        responseText: state.responseTexts.join(''),
+        toolCalls: state.toolCalls,
+      });
+    }
+
     endStreamSpan(span, state, recordOutputs);
   };
 
@@ -92,7 +195,7 @@ export function instrumentWorkersAiStream(
     // Keep the last (potentially incomplete) line in the buffer unless the stream is done.
     buffer = isDone ? '' : (lines.pop() ?? '');
     for (const line of lines) {
-      processLine(line, state, recordOutputs);
+      processLine(line, state, recordOutputs, toolCallAccumulator);
     }
   };
 
diff --git a/packages/core/src/tracing/workers-ai/utils.ts b/packages/core/src/tracing/workers-ai/utils.ts
@@ -10,14 +10,16 @@ import {
   GEN_AI_REQUEST_TOP_K,
   GEN_AI_REQUEST_TOP_P,
   GEN_AI_PROVIDER_NAME,
-  GEN_AI_OUTPUT_MESSAGES,
   GEN_AI_SYSTEM_INSTRUCTIONS,
 } from '@sentry/conventions/attributes';
 import { SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN } from '../../semanticAttributes';
 import type { Span, SpanAttributeValue } from '../../types/span';
 import {
   GEN_AI_INPUT_MESSAGES_ORIGINAL_LENGTH_ATTRIBUTE,
+  GEN_AI_OUTPUT_MESSAGES_ATTRIBUTE,
   GEN_AI_REQUEST_STREAM_ATTRIBUTE,
+  GEN_AI_RESPONSE_TEXT_ATTRIBUTE,
+  GEN_AI_RESPONSE_TOOL_CALLS_ATTRIBUTE,
 } from '../ai/gen-ai-attributes';
 import { extractSystemInstructions, getTruncatedJsonString, setTokenUsageAttributes } from '../ai/utils';
 import { stringify } from '../../utils/string';
@@ -134,6 +136,55 @@ export function addRequestAttributes(
   );
 }
 
+/**
+ * Build the `gen_ai.output.messages` value (a single assistant message with text and/or
+ * tool-call parts) from the response text and tool calls.
+ *
+ * We set this in addition to the deprecated `gen_ai.response.text` / `gen_ai.response.tool_calls`
+ * attributes because Sentry's product reads the model output from `gen_ai.output.messages` first.
+ * Relay migrates `gen_ai.response.text` into `gen_ai.output.messages`, but the tool-calls half of
+ * that migration is lossy — so tool-call turns would otherwise render an empty Output. Emitting the
+ * normalized message here (mirroring the Vercel AI integration) keeps tool calls visible.
+ */
+export function setOutputMessagesAttribute(
+  span: Span,
+  { responseText, toolCalls }: { responseText?: string; toolCalls?: unknown[] },
+): void {
+  const parts: Array<Record<string, unknown>> = [];
+
+  if (typeof responseText === 'string' && responseText.length > 0) {
+    parts.push({ type: 'text', content: responseText });
+  }
+
+  if (Array.isArray(toolCalls)) {
+    for (const toolCall of toolCalls) {
+      if (!toolCall || typeof toolCall !== 'object') {
+        continue;
+      }
+      const call = toolCall as {
+        id?: unknown;
+        function?: { name?: unknown; arguments?: unknown };
+        name?: unknown;
+        arguments?: unknown;
+      };
+      // Normalize both the OpenAI-compatible shape (name/arguments nested under `function`)
+      // and the native Workers AI shape (name/arguments at the top level).
+      const name = call.function?.name ?? call.name;
+      const args = call.function?.arguments ?? call.arguments;
+      parts.push({
+        type: 'tool_call',
+        id: call.id,
+        name,
+        arguments: typeof args === 'string' ? args : JSON.stringify(args ?? {}),
+      });
+    }
+  }
+
+  if (parts.length > 0) {
+    span.setAttribute(GEN_AI_OUTPUT_MESSAGES_ATTRIBUTE, JSON.stringify([{ role: 'assistant', parts }]));
+  }
+}
+
 /**
  * Record the response attributes (token usage, response text, tool calls) on the span.
  */
@@ -154,14 +205,21 @@ export function addResponseAttributes(span: Span, result: unknown, recordOutputs
   }
 
   if (recordOutputs) {
+    let responseText: string | undefined;
     if (typeof response.response === 'string') {
-      span.setAttribute(GEN_AI_OUTPUT_MESSAGES, response.response);
+      responseText = response.response;
+      span.setAttribute(GEN_AI_RESPONSE_TEXT_ATTRIBUTE, response.response);
     } else if (response.response != null) {
-      span.setAttribute(GEN_AI_OUTPUT_MESSAGES, JSON.stringify(response.response));
+      responseText = JSON.stringify(response.response);
+      span.setAttribute(GEN_AI_RESPONSE_TEXT_ATTRIBUTE, responseText);
     }
 
-    if (Array.isArray(response.tool_calls) && response.tool_calls.length > 0) {
-      span.setAttribute(GEN_AI_OUTPUT_MESSAGES, JSON.stringify(response.tool_calls));
+    const toolCalls =
+      Array.isArray(response.tool_calls) && response.tool_calls.length > 0 ? response.tool_calls : undefined;
+    if (toolCalls) {
+      span.setAttribute(GEN_AI_RESPONSE_TOOL_CALLS_ATTRIBUTE, JSON.stringify(toolCalls));
     }
+
+    setOutputMessagesAttribute(span, { responseText, toolCalls });
   }
 }
diff --git a/packages/core/test/lib/tracing/workers-ai-streaming.test.ts b/packages/core/test/lib/tracing/workers-ai-streaming.test.ts
@@ -1,8 +1,10 @@
 import { describe, expect, it } from 'vitest';
 import type { Span } from '../../../src';
 import {
+  GEN_AI_OUTPUT_MESSAGES_ATTRIBUTE,
   GEN_AI_RESPONSE_STREAMING_ATTRIBUTE,
   GEN_AI_RESPONSE_TEXT_ATTRIBUTE,
+  GEN_AI_RESPONSE_TOOL_CALLS_ATTRIBUTE,
   GEN_AI_USAGE_INPUT_TOKENS_ATTRIBUTE,
   GEN_AI_USAGE_OUTPUT_TOKENS_ATTRIBUTE,
   GEN_AI_USAGE_TOTAL_TOKENS_ATTRIBUTE,
@@ -99,6 +101,114 @@ describe('instrumentWorkersAiStream', () => {
     expect(ended()).toBe(true);
   });
 
+  // Models routed through the OpenAI-compatible endpoint (e.g. via `workers-ai-provider`,
+  // which the Cloudflare Agents SDK uses) stream `choices[].delta.content` instead of the
+  // native top-level `response` field. Usage still arrives top-level, which is why the
+  // pre-fix parser captured tokens but dropped the response text entirely.
+  it('accumulates response text from OpenAI-compatible choices[].delta.content chunks', async () => {
+    const { span, attributes, ended } = createMockSpan();
+    const chunks = [
+      'data: {"choices":[{"delta":{"content":"The capital "},"finish_reason":null}]}\n\n',
+      'data: {"choices":[{"delta":{"content":"of France "},"finish_reason":null}]}\n\n',
+      'data: {"choices":[{"delta":{"content":"is Paris."},"finish_reason":"stop"}]}\n\n',
+      'data: {"usage":{"prompt_tokens":12,"completion_tokens":7,"total_tokens":19}}\n\ndata: [DONE]\n\n',
+    ];
+
+    const instrumented = instrumentWorkersAiStream(streamFromChunks(chunks), span, true);
+    await new Response(instrumented).text();
+
+    expect(attributes[GEN_AI_RESPONSE_STREAMING_ATTRIBUTE]).toBe(true);
+    expect(attributes[GEN_AI_RESPONSE_TEXT_ATTRIBUTE]).toBe('The capital of France is Paris.');
+    expect(attributes[GEN_AI_USAGE_INPUT_TOKENS_ATTRIBUTE]).toBe(12);
+    expect(attributes[GEN_AI_USAGE_OUTPUT_TOKENS_ATTRIBUTE]).toBe(7);
+    expect(attributes[GEN_AI_USAGE_TOTAL_TOKENS_ATTRIBUTE]).toBe(19);
+    expect(ended()).toBe(true);
+  });
+
+  it('assembles fragmented OpenAI-compatible tool calls from choices[].delta.tool_calls', async () => {
+    const { span, attributes } = createMockSpan();
+    const chunks = [
+      'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"getRepoInfo","arguments":"{\\"owner\\":"}}]}}]}\n\n',
+      'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\\"cloudflare\\",\\"name\\":\\"agents\\"}"}}]}}]}\n\n',
+      'data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}]}\n\ndata: [DONE]\n\n',
+    ];
+
+    const instrumented = instrumentWorkersAiStream(streamFromChunks(chunks), span, true);
+    await new Response(instrumented).text();
+
+    const toolCalls = JSON.parse(attributes[GEN_AI_RESPONSE_TOOL_CALLS_ATTRIBUTE] as string);
+    expect(toolCalls).toEqual([
+      {
+        index: 0,
+        id: 'call_1',
+        type: 'function',
+        function: { name: 'getRepoInfo', arguments: '{"owner":"cloudflare","name":"agents"}' },
+      },
+    ]);
+
+    // The product reads model output from `gen_ai.output.messages`; tool calls must appear there too.
+    expect(JSON.parse(attributes[GEN_AI_OUTPUT_MESSAGES_ATTRIBUTE] as string)).toEqual([
+      {
+        role: 'assistant',
+        parts: [
+          {
+            type: 'tool_call',
+            id: 'call_1',
+            name: 'getRepoInfo',
+            arguments: '{"owner":"cloudflare","name":"agents"}',
+          },
+        ],
+      },
+    ]);
+  });
+
+  // Some Workers AI models (e.g. `@cf/moonshotai/kimi-k2.6`) stream tool calls with the
+  // name/arguments at the top level of the tool-call object rather than nested under `function`.
+  it('assembles tool calls whose name/arguments are at the top level of the delta', async () => {
+    const { span, attributes } = createMockSpan();
+    const chunks = [
+      'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","name":"getRepoInfo","arguments":"{\\"owner\\":"}]}}]}\n\n',
+      'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"arguments":"\\"cloudflare\\"}"}]}}]}\n\n',
+      'data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}]}\n\ndata: [DONE]\n\n',
+    ];
+
+    const instrumented = instrumentWorkersAiStream(streamFromChunks(chunks), span, true);
+    await new Response(instrumented).text();
+
+    const toolCalls = JSON.parse(attributes[GEN_AI_RESPONSE_TOOL_CALLS_ATTRIBUTE] as string);
+    expect(toolCalls).toEqual([
+      {
+        index: 0,
+        id: 'call_1',
+        type: undefined,
+        function: { name: 'getRepoInfo', arguments: '{"owner":"cloudflare"}' },
+      },
+    ]);
+
+    expect(JSON.parse(attributes[GEN_AI_OUTPUT_MESSAGES_ATTRIBUTE] as string)).toEqual([
+      {
+        role: 'assistant',
+        parts: [{ type: 'tool_call', id: 'call_1', name: 'getRepoInfo', arguments: '{"owner":"cloudflare"}' }],
+      },
+    ]);
+  });
+
+  it('does not record OpenAI-compatible output when recordOutputs is false', async () => {
+    const { span, attributes } = createMockSpan();
+    const chunks = [
+      'data: {"choices":[{"delta":{"content":"secret"}}]}\n\n',
+      'data: {"usage":{"prompt_tokens":3,"completion_tokens":1,"total_tokens":4}}\n\ndata: [DONE]\n\n',
+    ];
+
+    const instrumented = instrumentWorkersAiStream(streamFromChunks(chunks), span, false);
+    await new Response(instrumented).text();
+
+    expect(attributes[GEN_AI_RESPONSE_TEXT_ATTRIBUTE]).toBeUndefined();
+    expect(attributes[GEN_AI_RESPONSE_TOOL_CALLS_ATTRIBUTE]).toBeUndefined();
+    expect(attributes[GEN_AI_OUTPUT_MESSAGES_ATTRIBUTE]).toBeUndefined();
+    expect(attributes[GEN_AI_USAGE_OUTPUT_TOKENS_ATTRIBUTE]).toBe(1);
+  });
+
   it('ends the span when the consumer cancels the stream', async () => {
     const { span, attributes, ended } = createMockSpan();
 
diff --git a/packages/core/test/lib/utils/workers-ai-utils.test.ts b/packages/core/test/lib/utils/workers-ai-utils.test.ts
