feat(server-utils): Record Flue message content behind the genAI options

Flue exposes the content on its event stream — `turn_request` carries the full
`ModelRequest` (system prompt, messages, tool definitions), the settled `turn`
carries `response.output`, and the tool events carry arguments and results — so
there is no reason to omit it.

Recorded as `gen_ai.input.messages`, `gen_ai.output.messages`,
`gen_ai.system_instructions`, `gen_ai.tool.definitions`,
`gen_ai.tool.call.arguments` and `gen_ai.tool.call.result`, gated on
`recordInputs`/`recordOutputs` via `resolveAIRecordingOptions`, which falls back
to the client's `dataCollection.genAI` settings. `FlueOptions` is threaded back
through `flueIntegration` and now does something.

Note the request content is only on `turn_request`; the settled `turn` reports
request metadata alone.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: RulaKhaled

packages/server-utils/src/ai/flue/index.ts

@@ -5,11 +5,14 @@ import {
   SPAN_STATUS_ERROR,
   startInactiveSpan,
   startSpan,
+  stringify,
   withActiveSpan,
 } from '@sentry/core';
 import {
   GEN_AI_AGENT_NAME,
   GEN_AI_CONVERSATION_ID,
+  GEN_AI_INPUT_MESSAGES,
+  GEN_AI_OUTPUT_MESSAGES,
   GEN_AI_COST_CACHE_CREATION_INPUT_TOKENS,
   GEN_AI_COST_CACHE_READ_INPUT_TOKENS,
   GEN_AI_COST_INPUT_TOKENS,
@@ -21,6 +24,10 @@ import {
   GEN_AI_RESPONSE_FINISH_REASONS,
   GEN_AI_RESPONSE_ID,
   GEN_AI_RESPONSE_MODEL,
+  GEN_AI_SYSTEM_INSTRUCTIONS,
+  GEN_AI_TOOL_CALL_ARGUMENTS,
+  GEN_AI_TOOL_CALL_RESULT,
+  GEN_AI_TOOL_DEFINITIONS,
   GEN_AI_TOOL_NAME,
   GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS,
   GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS,
@@ -29,12 +36,15 @@ import {
   GEN_AI_USAGE_TOTAL_TOKENS,
 } from '@sentry/conventions/attributes';
 import { ANTHROPIC_AI_INTEGRATION_NAME } from '../anthropic-ai/constants';
-import { getGenAiSpanOp } from '../core/utils';
+import type { GenAiOptions } from '../core/utils';
+import { getGenAiSpanOp, resolveAIRecordingOptions } from '../core/utils';
 import { GOOGLE_GENAI_INTEGRATION_NAME } from '../google-genai/constants';
 import { OPENAI_INTEGRATION_NAME } from '../openai/constants';
 import { FLUE_INSTRUMENTATION_KEY, FLUE_ORIGIN, SPANNED_OPERATION_TYPE } from './constants';
 import type { FlueInstrumentation, FlueObservation, FlueUsage } from './types';
 
+export type FlueOptions = GenAiOptions;
+
 const SKIPPED_PROVIDERS = [OPENAI_INTEGRATION_NAME, ANTHROPIC_AI_INTEGRATION_NAME, GOOGLE_GENAI_INTEGRATION_NAME];
 
 /**
@@ -47,17 +57,18 @@ const SKIPPED_PROVIDERS = [OPENAI_INTEGRATION_NAME, ANTHROPIC_AI_INTEGRATION_NAM
  * - `observe` opens and closes the turn span, because Flue's `turn_start`/`turn` events are the
  *   only one-to-one signal for a model call and `turn` is what carries usage and cost.
  *
- * Takes no options yet: no message content is recorded, so there is nothing for
- * `recordInputs`/`recordOutputs` to gate.
+ * Message content, tool arguments and tool results are gated on `recordInputs`/`recordOutputs`,
+ * which fall back to the client's `dataCollection.genAI` settings.
  */
-export function createFlueInstrumentation(): FlueInstrumentation {
+export function createFlueInstrumentation(options: FlueOptions = {}): FlueInstrumentation {
   // Flue drives the providers through `@earendil-works/pi-ai`, which bundles the `openai`,
   // `@anthropic-ai/sdk` and `@google/genai` clients. Left alone they instrument the same call this
   // reports as a turn, emitting a second `gen_ai.chat` beside ours. Done here rather than in
   // `flueIntegration` so registering by hand — the only option on Cloudflare, where agents run in
   // per-Durable-Object isolates — gets it too.
   _INTERNAL_skipAiProviderWrapping(SKIPPED_PROVIDERS);
 
+  const { recordInputs, recordOutputs } = resolveAIRecordingOptions(options);
   const turnSpans = new Map<string, Span>();
   const toolSpans = new Map<string, Span>();
   let agentSpan: Span | undefined;
@@ -113,14 +124,19 @@ export function createFlueInstrumentation(): FlueInstrumentation {
         case 'turn_start':
           startTurnSpan(observation, turnSpans, agentSpan);
           return;
+        case 'turn_request':
+          if (recordInputs) {
+            recordRequestContent(observation, turnSpans);
+          }
+          return;
         case 'turn':
-          endTurnSpan(observation, turnSpans);
+          endTurnSpan(observation, turnSpans, recordOutputs);
           return;
         case 'tool_start':
-          startToolSpan(observation, toolSpans, agentSpan);
+          startToolSpan(observation, toolSpans, agentSpan, recordInputs);
           return;
         case 'tool':
-          endToolSpan(observation, toolSpans);
+          endToolSpan(observation, toolSpans, recordOutputs);
           return;
         default:
           return;
@@ -159,7 +175,7 @@ function startTurnSpan(observation: FlueObservation, turnSpans: Map<string, Span
   turnSpans.set(turnId, agentSpan ? withActiveSpan(agentSpan, open) : open());
 }
 
-function endTurnSpan(observation: FlueObservation, turnSpans: Map<string, Span>): void {
+function endTurnSpan(observation: FlueObservation, turnSpans: Map<string, Span>, recordOutputs: boolean): void {
   const { turnId } = observation;
   const span = turnId ? turnSpans.get(turnId) : undefined;
   if (!span || !turnId) {
@@ -193,6 +209,11 @@ function endTurnSpan(observation: FlueObservation, turnSpans: Map<string, Span>)
     span.setAttribute(GEN_AI_RESPONSE_FINISH_REASONS, [finishReason]);
   }
 
+  const output = observation.response?.output;
+  if (recordOutputs && output !== undefined) {
+    span.setAttribute(GEN_AI_OUTPUT_MESSAGES, stringify(output));
+  }
+
   setUsageAttributes(span, observation.response?.usage, observation.isError);
 
   if (observation.isError) {
@@ -239,7 +260,12 @@ function setUsageAttributes(span: Span, usage: FlueUsage | undefined, isError?:
  * OpenTelemetry adapter projects them: siblings of `chat`, correlated to model output by tool call
  * id. Keyed by `toolCallId` so concurrent tool calls in one turn cannot cross-attribute.
  */
-function startToolSpan(observation: FlueObservation, toolSpans: Map<string, Span>, agentSpan: Span | undefined): void {
+function startToolSpan(
+  observation: FlueObservation,
+  toolSpans: Map<string, Span>,
+  agentSpan: Span | undefined,
+  recordInputs: boolean,
+): void {
   const { toolCallId, toolName } = observation;
   if (!toolCallId || toolSpans.has(toolCallId)) {
     return;
@@ -254,22 +280,52 @@ function startToolSpan(observation: FlueObservation, toolSpans: Map<string, Span
         [GEN_AI_OPERATION_NAME]: 'execute_tool',
         ...(toolName ? { [GEN_AI_TOOL_NAME]: toolName } : {}),
         ...(observation.conversationId ? { [GEN_AI_CONVERSATION_ID]: observation.conversationId } : {}),
+        ...(recordInputs && observation.args !== undefined
+          ? { [GEN_AI_TOOL_CALL_ARGUMENTS]: stringify(observation.args) }
+          : {}),
       },
     });
 
   toolSpans.set(toolCallId, agentSpan ? withActiveSpan(agentSpan, open) : open());
 }
 
-function endToolSpan(observation: FlueObservation, toolSpans: Map<string, Span>): void {
+function endToolSpan(observation: FlueObservation, toolSpans: Map<string, Span>, recordOutputs: boolean): void {
   const { toolCallId } = observation;
   const span = toolCallId ? toolSpans.get(toolCallId) : undefined;
   if (!span || !toolCallId) {
     return;
   }
   toolSpans.delete(toolCallId);
 
+  if (recordOutputs && observation.result !== undefined) {
+    span.setAttribute(GEN_AI_TOOL_CALL_RESULT, stringify(observation.result));
+  }
+
   if (observation.isError) {
     span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' });
   }
   span.end();
 }
+
+/**
+ * `turn_request` is the only event carrying the request's content — the settled `turn` reports
+ * metadata alone — so input messages, system prompt and tool definitions are read from it.
+ */
+function recordRequestContent(observation: FlueObservation, turnSpans: Map<string, Span>): void {
+  const { turnId } = observation;
+  const span = turnId ? turnSpans.get(turnId) : undefined;
+  const input = observation.request?.input;
+  if (!span || !input) {
+    return;
+  }
+
+  if (input.systemPrompt) {
+    span.setAttribute(GEN_AI_SYSTEM_INSTRUCTIONS, input.systemPrompt);
+  }
+  if (input.messages) {
+    span.setAttribute(GEN_AI_INPUT_MESSAGES, stringify(input.messages));
+  }
+  if (input.tools?.length) {
+    span.setAttribute(GEN_AI_TOOL_DEFINITIONS, stringify(input.tools));
+  }
+}

packages/server-utils/src/ai/flue/types.ts

@@ -29,10 +29,23 @@ export interface FlueModelRequestInfo {
   providerName?: string;
 }
 
+/** Mirrors `ModelRequestInput` — the content half of `ModelRequest`, on `turn_request` only. */
+export interface FlueModelRequestInput {
+  systemPrompt?: string;
+  messages?: unknown[];
+  tools?: unknown[];
+}
+
+/** `turn_request` carries `ModelRequest`, which is `ModelRequestInfo` plus the input. */
+export interface FlueModelRequest extends FlueModelRequestInfo {
+  input?: FlueModelRequestInput;
+}
+
 /** Mirrors `ModelResponse`. */
 export interface FlueModelResponse {
   responseId?: string;
   responseModel?: string;
+  output?: unknown;
   usage?: FlueUsage;
   finishReason?: string;
 }
@@ -53,7 +66,9 @@ export interface FlueObservation {
   isError?: boolean;
   purpose?: string;
   durationMs?: number;
-  request?: FlueModelRequestInfo;
+  request?: FlueModelRequest;
+  args?: unknown;
+  result?: unknown;
   response?: FlueModelResponse;
 }
 

packages/server-utils/src/ai/index.ts

@@ -12,3 +12,4 @@ export { createLangChainCallbackHandler, instrumentLangChainEmbeddings } from '.
 export { instrumentStateGraph, instrumentStateGraphCompile, instrumentCreateReactAgent } from './langgraph';
 export { SentryMastraExporter } from './mastra';
 export { createFlueInstrumentation } from './flue';
+export type { FlueOptions } from './flue';

packages/server-utils/src/index.ts

@@ -41,6 +41,7 @@ export { knexIntegration } from './integrations/knex';
 export { langChainIntegration } from './integrations/langchain';
 export { langGraphIntegration } from './integrations/langgraph';
 export { flueIntegration } from './integrations/flue';
+export type { FlueOptions } from './integrations/flue';
 export { mastraIntegration } from './integrations/mastra';
 export { SentryMastraExporter } from './ai/mastra';
 export { lruMemoizerIntegration } from './integrations/lru-memoizer';

packages/server-utils/src/integrations/flue.ts

@@ -2,6 +2,7 @@ import type { IntegrationFn } from '@sentry/core';
 import { debug, defineIntegration, GLOBAL_OBJ } from '@sentry/core';
 import { createFlueInstrumentation } from '../ai/flue';
 import { FLUE_INTEGRATION_NAME, FLUE_MODULE_NAME } from '../ai/flue/constants';
+import type { FlueOptions } from '../ai/flue';
 import type { FlueInstrumentation } from '../ai/flue/types';
 import { DEBUG_BUILD } from '../debug-build';
 import { flueModuleNames } from '../orchestrion/config/flue';
@@ -10,11 +11,11 @@ import { getOrchestrionModuleBindings } from '../utils/moduleInjected';
 
 type FlueInstrumentFn = (instrumentation: FlueInstrumentation) => unknown;
 
-const _flueIntegration = (() => {
+const _flueIntegration = ((options: FlueOptions = {}) => {
   return {
     name: FLUE_INTEGRATION_NAME,
     setup(client) {
-      invokeOrchestrionInstrumentation(client, flueModuleNames, registerFlueInstrumentation, [], {
+      invokeOrchestrionInstrumentation(client, flueModuleNames, registerFlueInstrumentation, [options], {
         // Nothing is bound to a tracing channel: the interceptor opens the agent span itself, so
         // the async-context binding is not a precondition for registering.
         requiresTracingChannelBinding: false,
@@ -29,10 +30,10 @@ const _flueIntegration = (() => {
  * own binding out. Under the runtime hook there is no snippet, but the resolved file is recorded,
  * and ESM keys its module registry by URL — so importing that URL yields the running namespace.
  */
-function registerFlueInstrumentation(): void {
+function registerFlueInstrumentation(options: FlueOptions): void {
   const bound = getOrchestrionModuleBindings(FLUE_MODULE_NAME)?.instrument as FlueInstrumentFn | undefined;
   if (typeof bound === 'function') {
-    install(bound);
+    install(bound, options);
     return;
   }
 
@@ -46,7 +47,7 @@ function registerFlueInstrumentation(): void {
   import(url).then(
     (mod: { instrument?: FlueInstrumentFn }) => {
       if (mod.instrument) {
-        install(mod.instrument);
+        install(mod.instrument, options);
       }
     },
     (error: unknown) => {
@@ -55,9 +56,9 @@ function registerFlueInstrumentation(): void {
   );
 }
 
-function install(instrument: FlueInstrumentFn): void {
+function install(instrument: FlueInstrumentFn, options: FlueOptions): void {
   try {
-    instrument(createFlueInstrumentation());
+    instrument(createFlueInstrumentation(options));
   } catch (error) {
     // Flue throws `InstrumentationAlreadyInstalledError` if something already registered under our
     // key. The earlier registration is live, so log rather than surface it.
@@ -73,3 +74,5 @@ function install(instrument: FlueInstrumentFn): void {
  * `@flue/runtime` reachable, and neither path can instrument without one.
  */
 export const flueIntegration = defineIntegration(_flueIntegration);
+
+export type { FlueOptions };