From a0e81430ba32e9573236d73e372e0e02c4d6b691 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Fri, 11 Sep 2026 08:11:32 +0200 Subject: [PATCH 1/4] feat(server-utils): Add first-party Flue instrumentation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Instruments the Flue agent framework (`@flue/runtime`) through the runtime's own `instrument()` hook, producing the `invoke_agent` -> `chat` / `execute_tool` hierarchy with token usage, Flue-computed cost and message content: import { instrument } from '@flue/runtime'; import * as Sentry from '@sentry/node'; instrument(Sentry.createFlueInstrumentation()); Registration is left to the user rather than done through orchestrion. Flue is not instrumented at a call site — `instrument()` is a registration API whose registry is module-scope state, so an auto-registering integration would need a reference to that module's binding, which no channel payload carries. Flue documents this same pattern for observability providers, and it needs neither the runtime hook nor a bundler plugin. The two callbacks own different halves: the interceptor owns the agent span and the active context, so spans opened underneath parent correctly; `observe` owns the turn and tool spans, because `turn_start`/`turn` are the only signal one-to-one with a model call and `turn` carries usage. Also skips the raw provider integrations: Flue reaches providers through `@earendil-works/pi-ai`, which bundles the `openai`, `@anthropic-ai/sdk` and `@google/genai` clients, so those would emit a second `gen_ai.chat` beside ours. Co-Authored-By: Claude Opus 5 --- packages/aws-serverless/src/index.ts | 1 + packages/bun/src/index.ts | 1 + packages/cloudflare/src/index.ts | 2 +- packages/deno/src/index.ts | 1 + packages/google-cloud-serverless/src/index.ts | 1 + packages/node/src/index.ts | 1 + .../server-utils/src/ai/flue/constants.ts | 20 ++ packages/server-utils/src/ai/flue/index.ts | 331 ++++++++++++++++++ packages/server-utils/src/ai/flue/types.ts | 101 ++++++ packages/server-utils/src/ai/index.ts | 2 + packages/server-utils/src/index.ts | 2 + 11 files changed, 462 insertions(+), 1 deletion(-) create mode 100644 packages/server-utils/src/ai/flue/constants.ts create mode 100644 packages/server-utils/src/ai/flue/index.ts create mode 100644 packages/server-utils/src/ai/flue/types.ts diff --git a/packages/aws-serverless/src/index.ts b/packages/aws-serverless/src/index.ts index aa5aaaf33cb2..b9c46e0c702f 100644 --- a/packages/aws-serverless/src/index.ts +++ b/packages/aws-serverless/src/index.ts @@ -64,6 +64,7 @@ export { langGraphIntegration, mastraIntegration, SentryMastraExporter, + createFlueInstrumentation, modulesIntegration, nodeRuntimeMetricsIntegration, type NodeRuntimeMetricsOptions, diff --git a/packages/bun/src/index.ts b/packages/bun/src/index.ts index fe427aa97424..d82053dec02e 100644 --- a/packages/bun/src/index.ts +++ b/packages/bun/src/index.ts @@ -86,6 +86,7 @@ export { langGraphIntegration, mastraIntegration, SentryMastraExporter, + createFlueInstrumentation, modulesIntegration, contextLinesIntegration, nodeContextIntegration, diff --git a/packages/cloudflare/src/index.ts b/packages/cloudflare/src/index.ts index e347cbc9fdab..fdf54a1222b8 100644 --- a/packages/cloudflare/src/index.ts +++ b/packages/cloudflare/src/index.ts @@ -107,7 +107,7 @@ export { spanStreamingIntegration, } from '@sentry/core'; export { trpcMiddleware, wrapMcpServerWithSentry } from '@sentry/core/server'; -export { instrumentPostgresJsSql } from '@sentry/server-utils'; +export { createFlueInstrumentation, instrumentPostgresJsSql } from '@sentry/server-utils'; export { withSentry } from './withSentry'; export { defineCloudflareOptions } from './defineCloudflareOptions'; diff --git a/packages/deno/src/index.ts b/packages/deno/src/index.ts index 56840b35f654..d6b1b71566da 100644 --- a/packages/deno/src/index.ts +++ b/packages/deno/src/index.ts @@ -135,6 +135,7 @@ export { langGraphIntegration, mastraIntegration, SentryMastraExporter, + createFlueInstrumentation, lruMemoizerIntegration, mongoIntegration, mongooseIntegration, diff --git a/packages/google-cloud-serverless/src/index.ts b/packages/google-cloud-serverless/src/index.ts index 0540b2e6879b..3493fe6c222a 100644 --- a/packages/google-cloud-serverless/src/index.ts +++ b/packages/google-cloud-serverless/src/index.ts @@ -64,6 +64,7 @@ export { langGraphIntegration, mastraIntegration, SentryMastraExporter, + createFlueInstrumentation, modulesIntegration, nodeRuntimeMetricsIntegration, type NodeRuntimeMetricsOptions, diff --git a/packages/node/src/index.ts b/packages/node/src/index.ts index 03532223a506..33903c85bf7d 100644 --- a/packages/node/src/index.ts +++ b/packages/node/src/index.ts @@ -20,6 +20,7 @@ export { langChainIntegration, langGraphIntegration, lruMemoizerIntegration, + createFlueInstrumentation, mastraIntegration, SentryMastraExporter, mongoIntegration, diff --git a/packages/server-utils/src/ai/flue/constants.ts b/packages/server-utils/src/ai/flue/constants.ts new file mode 100644 index 000000000000..2591db9587f9 --- /dev/null +++ b/packages/server-utils/src/ai/flue/constants.ts @@ -0,0 +1,20 @@ +export const FLUE_INTEGRATION_NAME = 'Flue' as const; + +export const FLUE_MODULE_NAME = '@flue/runtime'; + +export const FLUE_ORIGIN = 'auto.ai.flue'; + +/** + * Identifies our registration in Flue's keyed instrumentation registry. A distinct key lets us + * coexist with `@flue/opentelemetry` (which registers under its own key) and makes a repeated + * `instrument()` call a no-op instead of throwing `InstrumentationAlreadyInstalledError`. + */ +export const FLUE_INSTRUMENTATION_KEY = Symbol.for('sentry.flue.instrumentation'); + +/** + * Flue drives one LLM call through many `model` operations (one per stream read), and its `agent` + * operation nests inside itself once per submission. Only `agent` is spanned from the interceptor, + * and only at the outermost depth; the turn span is driven from the observation stream instead, + * where `turn_start`/`turn` are exactly one-to-one with a model call. + */ +export const SPANNED_OPERATION_TYPE = 'agent'; diff --git a/packages/server-utils/src/ai/flue/index.ts b/packages/server-utils/src/ai/flue/index.ts new file mode 100644 index 000000000000..ec6cbe74316e --- /dev/null +++ b/packages/server-utils/src/ai/flue/index.ts @@ -0,0 +1,331 @@ +import type { Span } from '@sentry/core'; +import { + _INTERNAL_skipAiProviderWrapping, + SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, + 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, + GEN_AI_COST_OUTPUT_TOKENS, + GEN_AI_COST_TOTAL_TOKENS, + GEN_AI_OPERATION_NAME, + GEN_AI_PROVIDER_NAME, + GEN_AI_REQUEST_MODEL, + 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, + GEN_AI_USAGE_INPUT_TOKENS, + GEN_AI_USAGE_OUTPUT_TOKENS, + GEN_AI_USAGE_TOTAL_TOKENS, +} from '@sentry/conventions/attributes'; +import { ANTHROPIC_AI_INTEGRATION_NAME } from '../anthropic-ai/constants'; +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]; + +/** + * Build the object to hand to `instrument()` from `@flue/runtime`. + * + * The two callbacks own different halves of the result: + * + * - `interceptor` wraps agent execution, so the agent span is *active* for its duration and every + * span opened underneath parents correctly. + * - `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. + * + * 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(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(); + const toolSpans = new Map(); + let agentSpan: Span | undefined; + let agentDepth = 0; + + return { + key: FLUE_INSTRUMENTATION_KEY, + + interceptor: async (operation, ctx, next) => { + if (operation?.type !== SPANNED_OPERATION_TYPE) { + return next(); + } + + // A submission's agent operation re-enters once, and the two carry different halves of the + // agent's identity: the outer context names the agent, the inner one names the conversation. + // Only the outer becomes a span, so the conversation id is lifted onto it from the re-entry. + if (agentDepth++ > 0) { + if (ctx.conversationId) { + agentSpan?.setAttribute(GEN_AI_CONVERSATION_ID, ctx.conversationId); + } + try { + return await next(); + } finally { + agentDepth--; + } + } + + return startSpan( + { + name: `invoke_agent ${ctx.agentName ?? 'agent'}`, + op: getGenAiSpanOp('invoke_agent'), + attributes: { + [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: FLUE_ORIGIN, + [GEN_AI_OPERATION_NAME]: 'invoke_agent', + ...(ctx.agentName ? { [GEN_AI_AGENT_NAME]: ctx.agentName } : {}), + ...(ctx.conversationId ? { [GEN_AI_CONVERSATION_ID]: ctx.conversationId } : {}), + }, + }, + async span => { + agentSpan = span; + try { + return await next(); + } finally { + agentDepth--; + agentSpan = undefined; + } + }, + ); + }, + + observe: observation => { + switch (observation.type) { + case 'turn_start': + startTurnSpan(observation, turnSpans, agentSpan); + return; + case 'turn_request': + if (recordInputs) { + recordRequestContent(observation, turnSpans); + } + return; + case 'turn': + endTurnSpan(observation, turnSpans, recordOutputs); + return; + case 'tool_start': + startToolSpan(observation, toolSpans, agentSpan, recordInputs); + return; + case 'tool': + endToolSpan(observation, toolSpans, recordOutputs); + return; + default: + return; + } + }, + + dispose: () => { + for (const span of [...turnSpans.values(), ...toolSpans.values()]) { + span.end(); + } + turnSpans.clear(); + toolSpans.clear(); + }, + }; +} + +function startTurnSpan(observation: FlueObservation, turnSpans: Map, agentSpan: Span | undefined): void { + const { turnId } = observation; + if (!turnId || turnSpans.has(turnId)) { + return; + } + + const open = (): Span => + startInactiveSpan({ + name: 'chat', + op: getGenAiSpanOp('chat'), + attributes: { + [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: FLUE_ORIGIN, + [GEN_AI_OPERATION_NAME]: 'chat', + ...(observation.conversationId ? { [GEN_AI_CONVERSATION_ID]: observation.conversationId } : {}), + }, + }); + + // `observe` runs inside the agent operation, but the active span there is whatever the provider + // SDK last opened — parent explicitly so the turn always hangs off the agent. + turnSpans.set(turnId, agentSpan ? withActiveSpan(agentSpan, open) : open()); +} + +function endTurnSpan(observation: FlueObservation, turnSpans: Map, recordOutputs: boolean): void { + const { turnId } = observation; + const span = turnId ? turnSpans.get(turnId) : undefined; + if (!span || !turnId) { + return; + } + turnSpans.delete(turnId); + + const requestedModel = observation.request?.requestedModel; + const responseModel = observation.response?.responseModel; + const model = responseModel ?? requestedModel; + if (model) { + span.updateName(`chat ${model}`); + } + if (requestedModel) { + span.setAttribute(GEN_AI_REQUEST_MODEL, requestedModel); + } + if (responseModel) { + span.setAttribute(GEN_AI_RESPONSE_MODEL, responseModel); + } + + const provider = observation.request?.providerId ?? observation.request?.providerName; + if (provider) { + span.setAttribute(GEN_AI_PROVIDER_NAME, provider); + } + + const { responseId, finishReason } = observation.response ?? {}; + if (responseId) { + span.setAttribute(GEN_AI_RESPONSE_ID, responseId); + } + if (finishReason) { + 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) { + span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); + } + span.end(); +} + +/** + * Flue reports token counts and its own computed costs on the same `usage` object, so both are set + * here. The cost figures have no equivalent in the provider SDKs' own instrumentation. + */ +function setUsageAttributes(span: Span, usage: FlueUsage | undefined, isError?: boolean): void { + // A turn that failed before the provider billed anything reports every counter as 0. Writing + // those is noise that reads as a real zero-cost call, so skip the block entirely. + if (!usage || (isError && !usage.totalTokens)) { + return; + } + + const attributes: Record = {}; + const set = (key: string, value: number | undefined): void => { + if (typeof value === 'number') { + attributes[key] = value; + } + }; + + set(GEN_AI_USAGE_INPUT_TOKENS, usage.input); + set(GEN_AI_USAGE_OUTPUT_TOKENS, usage.output); + set(GEN_AI_USAGE_TOTAL_TOKENS, usage.totalTokens); + set(GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS, usage.cacheRead); + set(GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS, usage.cacheWrite); + + set(GEN_AI_COST_INPUT_TOKENS, usage.cost?.input); + set(GEN_AI_COST_OUTPUT_TOKENS, usage.cost?.output); + set(GEN_AI_COST_TOTAL_TOKENS, usage.cost?.total); + set(GEN_AI_COST_CACHE_READ_INPUT_TOKENS, usage.cost?.cacheRead); + set(GEN_AI_COST_CACHE_CREATION_INPUT_TOKENS, usage.cost?.cacheWrite); + + span.setAttributes(attributes); +} + +/** + * Tool spans hang off the agent invocation rather than the turn, matching how Flue's own + * 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, + agentSpan: Span | undefined, + recordInputs: boolean, +): void { + const { toolCallId, toolName } = observation; + if (!toolCallId || toolSpans.has(toolCallId)) { + return; + } + + const open = (): Span => + startInactiveSpan({ + name: `execute_tool ${toolName ?? 'unknown'}`, + op: getGenAiSpanOp('execute_tool'), + attributes: { + [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: FLUE_ORIGIN, + [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, 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): 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)); + } +} diff --git a/packages/server-utils/src/ai/flue/types.ts b/packages/server-utils/src/ai/flue/types.ts new file mode 100644 index 000000000000..05784d30c8e6 --- /dev/null +++ b/packages/server-utils/src/ai/flue/types.ts @@ -0,0 +1,101 @@ +/** + * Structural types for the subset of `@flue/runtime`'s instrumentation contract we consume. + * + * Declared locally rather than imported: `@flue/runtime` is ESM-only and not a dependency of this + * package, and the SDK must not import it. Mirrors `FlueInstrumentation`, `FlueObservation` and + * `FlueExecutionContext` as of `@flue/runtime` 2.x. + */ + +/** Token counts and Flue-computed costs on a settled turn. */ +export interface FlueUsage { + input?: number; + output?: number; + cacheRead?: number; + cacheWrite?: number; + totalTokens?: number; + cost?: { + input?: number; + output?: number; + cacheRead?: number; + cacheWrite?: number; + total?: number; + }; +} + +/** Mirrors `ModelRequestInfo`. */ +export interface FlueModelRequestInfo { + requestedModel?: string; + providerId?: string; + 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; +} + +/** + * One event from Flue's observation stream. Only the fields we read are declared; Flue emits more + * event types than are handled here, and unknown types are ignored. + */ +export interface FlueObservation { + type: string; + agentName?: string; + conversationId?: string; + session?: string; + turnId?: string; + taskId?: string; + toolName?: string; + toolCallId?: string; + isError?: boolean; + purpose?: string; + durationMs?: number; + request?: FlueModelRequest; + args?: unknown; + result?: unknown; + response?: FlueModelResponse; +} + +/** The execution unit an interceptor wraps. */ +export interface FlueExecutionOperation { + type: string; + operationId?: string; + operationKind?: string; + turnId?: string; +} + +export interface FlueExecutionContext { + agentName?: string; + conversationId?: string; + session?: string; + turnId?: string; + taskId?: string; +} + +export interface FlueEventContext { + agentName?: string; +} + +/** The object `instrument()` accepts. */ +export interface FlueInstrumentation { + key: symbol; + observe: (observation: FlueObservation, ctx: FlueEventContext) => void; + interceptor: (operation: FlueExecutionOperation, ctx: FlueExecutionContext, next: () => Promise) => Promise; + dispose: () => void; +} diff --git a/packages/server-utils/src/ai/index.ts b/packages/server-utils/src/ai/index.ts index 9fd995466027..f082646cd454 100644 --- a/packages/server-utils/src/ai/index.ts +++ b/packages/server-utils/src/ai/index.ts @@ -11,3 +11,5 @@ export { instrumentWorkersAiClient } from './workers-ai'; export { createLangChainCallbackHandler, instrumentLangChainEmbeddings } from './langchain'; export { instrumentStateGraph, instrumentStateGraphCompile, instrumentCreateReactAgent } from './langgraph'; export { SentryMastraExporter } from './mastra'; +export { createFlueInstrumentation } from './flue'; +export type { FlueOptions } from './flue'; diff --git a/packages/server-utils/src/index.ts b/packages/server-utils/src/index.ts index 2f918e2d5422..471062fae85b 100644 --- a/packages/server-utils/src/index.ts +++ b/packages/server-utils/src/index.ts @@ -40,6 +40,8 @@ export { kafkaIntegration } from './integrations/kafkajs'; export { knexIntegration } from './integrations/knex'; export { langChainIntegration } from './integrations/langchain'; export { langGraphIntegration } from './integrations/langgraph'; +export { createFlueInstrumentation } from './ai/flue'; +export type { FlueOptions } from './ai/flue'; export { mastraIntegration } from './integrations/mastra'; export { SentryMastraExporter } from './ai/mastra'; export { lruMemoizerIntegration } from './integrations/lru-memoizer'; From c01cb3f39efac678cba4cbd72c1d2773421cc95f Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Fri, 11 Sep 2026 10:19:01 +0200 Subject: [PATCH 2/4] fix(server-utils): Address review on the Flue instrumentation - Key agent spans by `operation.operationId` instead of shared closure state, so concurrent runs cannot clobber each other and a delegated subagent gets its own span rather than being folded into its parent's. Removes the depth counter and the re-entry special case. - Take the agent name and conversation id from the observations: the operation that gets the span carries neither, since the submission wrapper holds the name and the re-entry holds the conversation, and neither opens a span. - Continue the trace from `ctx.traceCarrier` when a durable submission resumes with no active trace, so it links back to the request that enqueued it. - Apply the AI provider skip on first use rather than at construction, and re-apply it once the registry has been cleared. Constructing the object no longer suppresses the provider integrations, so a rejected `instrument()` cannot leave an app with no `gen_ai.chat` spans at all. - Make the turn and tool spans active for their operations, so a tool's own work and the provider's HTTP call nest inside them instead of beside them. - Resolve the recording options per event rather than once, since the client is replaced per request on Cloudflare. - Record the conventional request attributes (`temperature`, `max_tokens`, `reasoning.level`, `server.address`, `server.port`) and the turn purpose, and drop the dead `providerId ?? providerName` fallback. Co-Authored-By: Claude Opus 5 --- packages/server-utils/src/ai/flue/index.ts | 359 +++++++-------------- packages/server-utils/src/ai/flue/types.ts | 18 +- packages/server-utils/src/ai/flue/utils.ts | 251 ++++++++++++++ 3 files changed, 377 insertions(+), 251 deletions(-) create mode 100644 packages/server-utils/src/ai/flue/utils.ts diff --git a/packages/server-utils/src/ai/flue/index.ts b/packages/server-utils/src/ai/flue/index.ts index ec6cbe74316e..a87c7124bd1e 100644 --- a/packages/server-utils/src/ai/flue/index.ts +++ b/packages/server-utils/src/ai/flue/index.ts @@ -1,47 +1,29 @@ import type { Span } from '@sentry/core'; import { + _INTERNAL_shouldSkipAiProviderWrapping, _INTERNAL_skipAiProviderWrapping, + continueTrace, + getActiveSpan, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, - 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, - GEN_AI_COST_OUTPUT_TOKENS, - GEN_AI_COST_TOTAL_TOKENS, - GEN_AI_OPERATION_NAME, - GEN_AI_PROVIDER_NAME, - GEN_AI_REQUEST_MODEL, - 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, - GEN_AI_USAGE_INPUT_TOKENS, - GEN_AI_USAGE_OUTPUT_TOKENS, - GEN_AI_USAGE_TOTAL_TOKENS, -} from '@sentry/conventions/attributes'; +import { GEN_AI_AGENT_NAME, GEN_AI_CONVERSATION_ID, GEN_AI_OPERATION_NAME } from '@sentry/conventions/attributes'; import { ANTHROPIC_AI_INTEGRATION_NAME } from '../anthropic-ai/constants'; 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'; +import { + endToolSpan, + endTurnSpan, + recordRequestContent, + sentryTraceFromTraceparent, + startToolSpan, + startTurnSpan, +} from './utils'; +import type { FlueInstrumentation } from './types'; export type FlueOptions = GenAiOptions; @@ -58,71 +40,124 @@ const SKIPPED_PROVIDERS = [OPENAI_INTEGRATION_NAME, ANTHROPIC_AI_INTEGRATION_NAM * only one-to-one signal for a model call and `turn` is what carries usage and cost. * * Message content, tool arguments and tool results are gated on `recordInputs`/`recordOutputs`, - * which fall back to the client's `dataCollection.genAI` settings. + * which fall back to the current client's `dataCollection.genAI` settings and are read per event. */ 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); + // reports as a turn, emitting a second `gen_ai.chat` beside ours. + // + // Applied on first use rather than here, for two reasons. Constructing the object proves nothing + // — if `instrument()` rejects it, suppressing the provider integrations would leave the app with + // no `gen_ai.chat` spans at all. And the registry is reset per client (`_setupIntegrations` + // clears it, and Cloudflare calls `init()` per request), so a one-shot call at module scope is + // wiped by the next `init()` and every later request double-reports. + const skipProviders = (): void => { + if (!_INTERNAL_shouldSkipAiProviderWrapping(SKIPPED_PROVIDERS[0]!)) { + _INTERNAL_skipAiProviderWrapping(SKIPPED_PROVIDERS); + } + }; - const { recordInputs, recordOutputs } = resolveAIRecordingOptions(options); + // Keyed by the agent operation's own id, which is what the observations carry. That keeps + // concurrent runs apart and gives a delegated subagent its own span: Flue nests a second `agent` + // operation inside the parent's for `task` delegation, and the nesting is not bounded at two. + const agentSpans = new Map(); const turnSpans = new Map(); const toolSpans = new Map(); - let agentSpan: Span | undefined; - let agentDepth = 0; return { key: FLUE_INSTRUMENTATION_KEY, interceptor: async (operation, ctx, next) => { + skipProviders(); + + // `observe` has already opened the span for this unit of work; make it active for the + // duration so whatever the tool or model call does lands inside it rather than beside it. + if (operation?.type === 'tool') { + const toolSpan = operation.toolCallId ? toolSpans.get(operation.toolCallId) : undefined; + return toolSpan ? withActiveSpan(toolSpan, next) : next(); + } + if (operation?.type === 'model') { + const turnSpan = operation.turnId ? turnSpans.get(operation.turnId) : undefined; + return turnSpan ? withActiveSpan(turnSpan, next) : next(); + } + if (operation?.type !== SPANNED_OPERATION_TYPE) { return next(); } - // A submission's agent operation re-enters once, and the two carry different halves of the - // agent's identity: the outer context names the agent, the inner one names the conversation. - // Only the outer becomes a span, so the conversation id is lifted onto it from the re-entry. - if (agentDepth++ > 0) { - if (ctx.conversationId) { - agentSpan?.setAttribute(GEN_AI_CONVERSATION_ID, ctx.conversationId); - } - try { - return await next(); - } finally { - agentDepth--; - } + // A submission opens with a wrapper operation whose id *is* the submission id; the run it + // wraps gets its own operation id, and that is the one the observations reference. Spanning + // the wrapper too would double-count every agent invocation. + const operationId = operation.operationId; + if (!operationId || operationId === ctx.submissionId || agentSpans.has(operationId)) { + return next(); } - return startSpan( - { - name: `invoke_agent ${ctx.agentName ?? 'agent'}`, - op: getGenAiSpanOp('invoke_agent'), - attributes: { - [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: FLUE_ORIGIN, - [GEN_AI_OPERATION_NAME]: 'invoke_agent', - ...(ctx.agentName ? { [GEN_AI_AGENT_NAME]: ctx.agentName } : {}), - ...(ctx.conversationId ? { [GEN_AI_CONVERSATION_ID]: ctx.conversationId } : {}), + const openAgentSpan = () => + startSpan( + { + name: `invoke_agent ${ctx.agentName ?? 'agent'}`, + op: getGenAiSpanOp('invoke_agent'), + attributes: { + [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: FLUE_ORIGIN, + [GEN_AI_OPERATION_NAME]: 'invoke_agent', + ...(ctx.agentName ? { [GEN_AI_AGENT_NAME]: ctx.agentName } : {}), + ...(ctx.conversationId ? { [GEN_AI_CONVERSATION_ID]: ctx.conversationId } : {}), + }, }, - }, - async span => { - agentSpan = span; - try { - return await next(); - } finally { - agentDepth--; - agentSpan = undefined; - } - }, - ); + async (span: Span) => { + agentSpans.set(operationId, span); + try { + return await next(); + } finally { + agentSpans.delete(operationId); + } + }, + ); + + // A durable or dispatched submission is run later by the coordinator, possibly in another + // isolate or process, where nothing links it back to the request that enqueued it. Flue + // replays that request's `traceparent` here, so continue from it — but only when no trace is + // already active, so an in-process dispatch keeps the trace it is genuinely part of. + const sentryTrace = ctx.traceCarrier?.traceparent + ? sentryTraceFromTraceparent(ctx.traceCarrier.traceparent) + : undefined; + + return sentryTrace && !getActiveSpan() + ? continueTrace({ sentryTrace, baggage: undefined }, openAgentSpan) + : openAgentSpan(); }, observe: observation => { + skipProviders(); + + // Resolved per observation, not once at construction: `resolveAIRecordingOptions` reads the + // current client's `dataCollection.genAI`, and Cloudflare replaces the client per request, so + // values captured at isolate load would be the wrong ones for every later request. + const { recordInputs, recordOutputs } = resolveAIRecordingOptions(options); + + // Observations are emitted synchronously from inside the agent operation, so the active span + // here is the agent span the interceptor opened — turn and tool spans parent off it without + // any bookkeeping. + // + // The agent operation that gets the span is the one the observations reference, and it knows + // neither the agent's name nor the conversation: the submission wrapper carries the name, the + // re-entry carries the conversation, and neither opens a span. Both arrive here instead. + const agentSpan = observation.operationId ? agentSpans.get(observation.operationId) : undefined; + if (agentSpan) { + if (observation.conversationId) { + agentSpan.setAttribute(GEN_AI_CONVERSATION_ID, observation.conversationId); + } + if (observation.agentName) { + agentSpan.setAttribute(GEN_AI_AGENT_NAME, observation.agentName); + agentSpan.updateName(`invoke_agent ${observation.agentName}`); + } + } + switch (observation.type) { case 'turn_start': - startTurnSpan(observation, turnSpans, agentSpan); + startTurnSpan(observation, turnSpans); return; case 'turn_request': if (recordInputs) { @@ -133,7 +168,7 @@ export function createFlueInstrumentation(options: FlueOptions = {}): FlueInstru endTurnSpan(observation, turnSpans, recordOutputs); return; case 'tool_start': - startToolSpan(observation, toolSpans, agentSpan, recordInputs); + startToolSpan(observation, toolSpans, recordInputs); return; case 'tool': endToolSpan(observation, toolSpans, recordOutputs); @@ -147,185 +182,9 @@ export function createFlueInstrumentation(options: FlueOptions = {}): FlueInstru for (const span of [...turnSpans.values(), ...toolSpans.values()]) { span.end(); } + agentSpans.clear(); turnSpans.clear(); toolSpans.clear(); }, }; } - -function startTurnSpan(observation: FlueObservation, turnSpans: Map, agentSpan: Span | undefined): void { - const { turnId } = observation; - if (!turnId || turnSpans.has(turnId)) { - return; - } - - const open = (): Span => - startInactiveSpan({ - name: 'chat', - op: getGenAiSpanOp('chat'), - attributes: { - [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: FLUE_ORIGIN, - [GEN_AI_OPERATION_NAME]: 'chat', - ...(observation.conversationId ? { [GEN_AI_CONVERSATION_ID]: observation.conversationId } : {}), - }, - }); - - // `observe` runs inside the agent operation, but the active span there is whatever the provider - // SDK last opened — parent explicitly so the turn always hangs off the agent. - turnSpans.set(turnId, agentSpan ? withActiveSpan(agentSpan, open) : open()); -} - -function endTurnSpan(observation: FlueObservation, turnSpans: Map, recordOutputs: boolean): void { - const { turnId } = observation; - const span = turnId ? turnSpans.get(turnId) : undefined; - if (!span || !turnId) { - return; - } - turnSpans.delete(turnId); - - const requestedModel = observation.request?.requestedModel; - const responseModel = observation.response?.responseModel; - const model = responseModel ?? requestedModel; - if (model) { - span.updateName(`chat ${model}`); - } - if (requestedModel) { - span.setAttribute(GEN_AI_REQUEST_MODEL, requestedModel); - } - if (responseModel) { - span.setAttribute(GEN_AI_RESPONSE_MODEL, responseModel); - } - - const provider = observation.request?.providerId ?? observation.request?.providerName; - if (provider) { - span.setAttribute(GEN_AI_PROVIDER_NAME, provider); - } - - const { responseId, finishReason } = observation.response ?? {}; - if (responseId) { - span.setAttribute(GEN_AI_RESPONSE_ID, responseId); - } - if (finishReason) { - 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) { - span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); - } - span.end(); -} - -/** - * Flue reports token counts and its own computed costs on the same `usage` object, so both are set - * here. The cost figures have no equivalent in the provider SDKs' own instrumentation. - */ -function setUsageAttributes(span: Span, usage: FlueUsage | undefined, isError?: boolean): void { - // A turn that failed before the provider billed anything reports every counter as 0. Writing - // those is noise that reads as a real zero-cost call, so skip the block entirely. - if (!usage || (isError && !usage.totalTokens)) { - return; - } - - const attributes: Record = {}; - const set = (key: string, value: number | undefined): void => { - if (typeof value === 'number') { - attributes[key] = value; - } - }; - - set(GEN_AI_USAGE_INPUT_TOKENS, usage.input); - set(GEN_AI_USAGE_OUTPUT_TOKENS, usage.output); - set(GEN_AI_USAGE_TOTAL_TOKENS, usage.totalTokens); - set(GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS, usage.cacheRead); - set(GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS, usage.cacheWrite); - - set(GEN_AI_COST_INPUT_TOKENS, usage.cost?.input); - set(GEN_AI_COST_OUTPUT_TOKENS, usage.cost?.output); - set(GEN_AI_COST_TOTAL_TOKENS, usage.cost?.total); - set(GEN_AI_COST_CACHE_READ_INPUT_TOKENS, usage.cost?.cacheRead); - set(GEN_AI_COST_CACHE_CREATION_INPUT_TOKENS, usage.cost?.cacheWrite); - - span.setAttributes(attributes); -} - -/** - * Tool spans hang off the agent invocation rather than the turn, matching how Flue's own - * 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, - agentSpan: Span | undefined, - recordInputs: boolean, -): void { - const { toolCallId, toolName } = observation; - if (!toolCallId || toolSpans.has(toolCallId)) { - return; - } - - const open = (): Span => - startInactiveSpan({ - name: `execute_tool ${toolName ?? 'unknown'}`, - op: getGenAiSpanOp('execute_tool'), - attributes: { - [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: FLUE_ORIGIN, - [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, 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): 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)); - } -} diff --git a/packages/server-utils/src/ai/flue/types.ts b/packages/server-utils/src/ai/flue/types.ts index 05784d30c8e6..525cda38a8a3 100644 --- a/packages/server-utils/src/ai/flue/types.ts +++ b/packages/server-utils/src/ai/flue/types.ts @@ -25,8 +25,13 @@ export interface FlueUsage { /** Mirrors `ModelRequestInfo`. */ export interface FlueModelRequestInfo { requestedModel?: string; + /** Provider slug (`anthropic`, `openrouter`) — what `gen_ai.provider.name` wants. */ providerId?: string; - providerName?: string; + temperature?: number; + maxTokens?: number; + reasoningLevel?: string; + serverAddress?: string; + serverPort?: number; } /** Mirrors `ModelRequestInput` — the content half of `ModelRequest`, on `turn_request` only. */ @@ -61,6 +66,8 @@ export interface FlueObservation { session?: string; turnId?: string; taskId?: string; + submissionId?: string; + operationId?: string; toolName?: string; toolCallId?: string; isError?: boolean; @@ -78,10 +85,19 @@ export interface FlueExecutionOperation { operationId?: string; operationKind?: string; turnId?: string; + toolCallId?: string; +} + +/** Mirrors `FlueTraceCarrier` — the W3C headers Flue persists at admission. */ +export interface FlueTraceCarrier { + traceparent: string; + tracestate?: string; } export interface FlueExecutionContext { + traceCarrier?: FlueTraceCarrier; agentName?: string; + submissionId?: string; conversationId?: string; session?: string; turnId?: string; diff --git a/packages/server-utils/src/ai/flue/utils.ts b/packages/server-utils/src/ai/flue/utils.ts new file mode 100644 index 000000000000..289af9c18f6d --- /dev/null +++ b/packages/server-utils/src/ai/flue/utils.ts @@ -0,0 +1,251 @@ +import type { Span } from '@sentry/core'; +import { SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, SPAN_STATUS_ERROR, startInactiveSpan, stringify } from '@sentry/core'; +import { + GEN_AI_CONVERSATION_ID, + GEN_AI_COST_CACHE_CREATION_INPUT_TOKENS, + GEN_AI_COST_CACHE_READ_INPUT_TOKENS, + GEN_AI_COST_INPUT_TOKENS, + GEN_AI_COST_OUTPUT_TOKENS, + GEN_AI_COST_TOTAL_TOKENS, + GEN_AI_INPUT_MESSAGES, + GEN_AI_OPERATION_NAME, + GEN_AI_OUTPUT_MESSAGES, + GEN_AI_PROVIDER_NAME, + GEN_AI_REQUEST_MAX_TOKENS, + GEN_AI_REQUEST_MODEL, + GEN_AI_REQUEST_REASONING_LEVEL, + GEN_AI_REQUEST_TEMPERATURE, + 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, + GEN_AI_USAGE_INPUT_TOKENS, + GEN_AI_USAGE_OUTPUT_TOKENS, + GEN_AI_USAGE_TOTAL_TOKENS, + SERVER_ADDRESS, + SERVER_PORT, +} from '@sentry/conventions/attributes'; +import { getGenAiSpanOp } from '../core/utils'; +import { FLUE_ORIGIN } from './constants'; +import type { FlueModelRequestInfo, FlueObservation, FlueUsage } from './types'; + +/** + * Flue persists the incoming W3C `traceparent` at admission and replays it on the agent operation. + * Sentry's own propagation uses `sentry-trace`, so the carrier has to be converted before it can + * continue the trace. Kept local until something else needs a W3C parser. + */ +export function sentryTraceFromTraceparent(traceparent: string): string | undefined { + const [version, traceId, spanId, flags] = traceparent.split('-'); + if (version !== '00' || !traceId || !spanId || !flags) { + return undefined; + } + // The sampled bit is the low bit of the flags byte; `% 2` avoids a bitwise operator. + return `${traceId}-${spanId}-${parseInt(flags, 16) % 2 === 1 ? '1' : '0'}`; +} + +export function startTurnSpan(observation: FlueObservation, turnSpans: Map): void { + const { turnId } = observation; + if (!turnId || turnSpans.has(turnId)) { + return; + } + + turnSpans.set( + turnId, + startInactiveSpan({ + name: 'chat', + op: getGenAiSpanOp('chat'), + attributes: { + [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: FLUE_ORIGIN, + [GEN_AI_OPERATION_NAME]: 'chat', + ...(observation.conversationId ? { [GEN_AI_CONVERSATION_ID]: observation.conversationId } : {}), + // No conventional attribute for this; it is the only way to tell a compaction turn from a + // user-facing one. + ...(observation.purpose ? { 'flue.turn.purpose': observation.purpose } : {}), + }, + }), + ); +} + +export function endTurnSpan(observation: FlueObservation, turnSpans: Map, recordOutputs: boolean): void { + const { turnId } = observation; + const span = turnId ? turnSpans.get(turnId) : undefined; + if (!span || !turnId) { + return; + } + turnSpans.delete(turnId); + + const requestedModel = observation.request?.requestedModel; + const responseModel = observation.response?.responseModel; + const model = responseModel ?? requestedModel; + if (model) { + span.updateName(`chat ${model}`); + } + if (requestedModel) { + span.setAttribute(GEN_AI_REQUEST_MODEL, requestedModel); + } + if (responseModel) { + span.setAttribute(GEN_AI_RESPONSE_MODEL, responseModel); + } + + // `providerId` is the slug (`anthropic`, `openrouter`); `providerName` is the display name. + const provider = observation.request?.providerId; + if (provider) { + span.setAttribute(GEN_AI_PROVIDER_NAME, provider); + } + + setRequestAttributes(span, observation.request); + + const { responseId, finishReason } = observation.response ?? {}; + if (responseId) { + span.setAttribute(GEN_AI_RESPONSE_ID, responseId); + } + if (finishReason) { + 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) { + span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); + } + span.end(); +} + +/** + * Flue reports token counts and its own computed costs on the same `usage` object, so both are set + * here. The cost figures have no equivalent in the provider SDKs' own instrumentation. + */ +export function setUsageAttributes(span: Span, usage: FlueUsage | undefined, isError?: boolean): void { + // A turn that failed before the provider billed anything reports every counter as 0. Writing + // those is noise that reads as a real zero-cost call, so skip the block entirely. + if (!usage || (isError && !usage.totalTokens)) { + return; + } + + const attributes: Record = {}; + const set = (key: string, value: number | undefined): void => { + if (typeof value === 'number') { + attributes[key] = value; + } + }; + + set(GEN_AI_USAGE_INPUT_TOKENS, usage.input); + set(GEN_AI_USAGE_OUTPUT_TOKENS, usage.output); + set(GEN_AI_USAGE_TOTAL_TOKENS, usage.totalTokens); + set(GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS, usage.cacheRead); + set(GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS, usage.cacheWrite); + + set(GEN_AI_COST_INPUT_TOKENS, usage.cost?.input); + set(GEN_AI_COST_OUTPUT_TOKENS, usage.cost?.output); + set(GEN_AI_COST_TOTAL_TOKENS, usage.cost?.total); + set(GEN_AI_COST_CACHE_READ_INPUT_TOKENS, usage.cost?.cacheRead); + set(GEN_AI_COST_CACHE_CREATION_INPUT_TOKENS, usage.cost?.cacheWrite); + + span.setAttributes(attributes); +} + +/** + * Tool spans hang off the agent invocation rather than the turn, matching how Flue's own + * 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. + */ +export function startToolSpan(observation: FlueObservation, toolSpans: Map, recordInputs: boolean): void { + const { toolCallId, toolName } = observation; + if (!toolCallId || toolSpans.has(toolCallId)) { + return; + } + + toolSpans.set( + toolCallId, + startInactiveSpan({ + name: `execute_tool ${toolName ?? 'unknown'}`, + op: getGenAiSpanOp('execute_tool'), + attributes: { + [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: FLUE_ORIGIN, + [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) } + : {}), + }, + }), + ); +} + +export function endToolSpan(observation: FlueObservation, toolSpans: Map, 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. + */ +export function recordRequestContent(observation: FlueObservation, turnSpans: Map): 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)); + } +} + +/** + * Model-call tuning and the provider endpoint, all on the settled turn's `ModelRequestInfo`. These + * are the conventional attributes the other AI integrations in this package set. + */ +export function setRequestAttributes(span: Span, request: FlueModelRequestInfo | undefined): void { + if (!request) { + return; + } + + const attributes: Record = {}; + const set = (key: string, value: string | number | undefined): void => { + if (value !== undefined) { + attributes[key] = value; + } + }; + + set(GEN_AI_REQUEST_TEMPERATURE, request.temperature); + set(GEN_AI_REQUEST_MAX_TOKENS, request.maxTokens); + set(GEN_AI_REQUEST_REASONING_LEVEL, request.reasoningLevel); + set(SERVER_ADDRESS, request.serverAddress); + set(SERVER_PORT, request.serverPort); + + span.setAttributes(attributes); +} From 41b32ccd7732dc27cb39bef8e9d179e9c807261b Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Fri, 11 Sep 2026 13:45:05 +0200 Subject: [PATCH 3/4] fix(server-utils): Cap Flue span tracking and widen the provider skip guard The turn and tool maps are keyed off ids that only a matching end observation removes, so a stream abandoned mid-turn left an entry behind for the lifetime of the process. Both are now `LRUMap`s capped the same way Mastra caps its own tracker, and eviction ends the span it drops rather than letting it disappear unsent. The provider skip guard only tested the first entry of `SKIPPED_PROVIDERS`, so an unrelated integration registering a skip for `openai` first would suppress the call that registers the other two, and their spans would duplicate the turn span. It now requires every provider to be registered before it short-circuits. Also names the Flue operation types we branch on, instead of one named constant for `agent` beside inline literals for `model` and `tool`. Co-Authored-By: Claude Opus 5 --- .../server-utils/src/ai/flue/constants.ts | 24 +++++++-- packages/server-utils/src/ai/flue/index.ts | 18 ++++--- packages/server-utils/src/ai/flue/utils.ts | 50 ++++++++++++++----- 3 files changed, 67 insertions(+), 25 deletions(-) diff --git a/packages/server-utils/src/ai/flue/constants.ts b/packages/server-utils/src/ai/flue/constants.ts index 2591db9587f9..bde674b2572a 100644 --- a/packages/server-utils/src/ai/flue/constants.ts +++ b/packages/server-utils/src/ai/flue/constants.ts @@ -12,9 +12,23 @@ export const FLUE_ORIGIN = 'auto.ai.flue'; export const FLUE_INSTRUMENTATION_KEY = Symbol.for('sentry.flue.instrumentation'); /** - * Flue drives one LLM call through many `model` operations (one per stream read), and its `agent` - * operation nests inside itself once per submission. Only `agent` is spanned from the interceptor, - * and only at the outermost depth; the turn span is driven from the observation stream instead, - * where `turn_start`/`turn` are exactly one-to-one with a model call. + * The Flue execution operations we act on. + * + * Only `AGENT` is spanned from the interceptor, and only at the outermost depth: Flue drives one + * LLM call through many `MODEL` operations (one per stream read), and its `AGENT` operation nests + * inside itself once per submission. The turn span is driven from the observation stream instead, + * where `turn_start`/`turn` are exactly one-to-one with a model call. `MODEL` and `TOOL` are + * intercepted only to make the already-open span active for the duration of the operation. */ -export const SPANNED_OPERATION_TYPE = 'agent'; +export const FLUE_OPERATION = { + AGENT: 'agent', + MODEL: 'model', + TOOL: 'tool', +} as const; + +/** + * Cap on tracked turn and tool spans, matching `MAX_TRACKED_MASTRA_SPANS`. Both maps are keyed off + * an id that is only removed when the matching end observation arrives; a stream that is abandoned + * mid-turn never emits one, so without a cap the map grows for the lifetime of the process. + */ +export const MAX_TRACKED_FLUE_SPANS = 1000; diff --git a/packages/server-utils/src/ai/flue/index.ts b/packages/server-utils/src/ai/flue/index.ts index a87c7124bd1e..e98720446786 100644 --- a/packages/server-utils/src/ai/flue/index.ts +++ b/packages/server-utils/src/ai/flue/index.ts @@ -4,6 +4,7 @@ import { _INTERNAL_skipAiProviderWrapping, continueTrace, getActiveSpan, + LRUMap, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, startSpan, withActiveSpan, @@ -14,7 +15,8 @@ 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 { FLUE_INSTRUMENTATION_KEY, FLUE_OPERATION, FLUE_ORIGIN, MAX_TRACKED_FLUE_SPANS } from './constants'; +import type { SpanTracker } from './utils'; import { endToolSpan, endTurnSpan, @@ -53,7 +55,7 @@ export function createFlueInstrumentation(options: FlueOptions = {}): FlueInstru // clears it, and Cloudflare calls `init()` per request), so a one-shot call at module scope is // wiped by the next `init()` and every later request double-reports. const skipProviders = (): void => { - if (!_INTERNAL_shouldSkipAiProviderWrapping(SKIPPED_PROVIDERS[0]!)) { + if (!SKIPPED_PROVIDERS.every(provider => _INTERNAL_shouldSkipAiProviderWrapping(provider))) { _INTERNAL_skipAiProviderWrapping(SKIPPED_PROVIDERS); } }; @@ -61,9 +63,11 @@ export function createFlueInstrumentation(options: FlueOptions = {}): FlueInstru // Keyed by the agent operation's own id, which is what the observations carry. That keeps // concurrent runs apart and gives a delegated subagent its own span: Flue nests a second `agent` // operation inside the parent's for `task` delegation, and the nesting is not bounded at two. + // A plain map: the entry is removed in a `finally`, so it is bounded by concurrent agent runs. const agentSpans = new Map(); - const turnSpans = new Map(); - const toolSpans = new Map(); + // Capped, unlike the above: these are keyed off ids that only a matching end observation removes. + const turnSpans: SpanTracker = new LRUMap(MAX_TRACKED_FLUE_SPANS); + const toolSpans: SpanTracker = new LRUMap(MAX_TRACKED_FLUE_SPANS); return { key: FLUE_INSTRUMENTATION_KEY, @@ -73,16 +77,16 @@ export function createFlueInstrumentation(options: FlueOptions = {}): FlueInstru // `observe` has already opened the span for this unit of work; make it active for the // duration so whatever the tool or model call does lands inside it rather than beside it. - if (operation?.type === 'tool') { + if (operation?.type === FLUE_OPERATION.TOOL) { const toolSpan = operation.toolCallId ? toolSpans.get(operation.toolCallId) : undefined; return toolSpan ? withActiveSpan(toolSpan, next) : next(); } - if (operation?.type === 'model') { + if (operation?.type === FLUE_OPERATION.MODEL) { const turnSpan = operation.turnId ? turnSpans.get(operation.turnId) : undefined; return turnSpan ? withActiveSpan(turnSpan, next) : next(); } - if (operation?.type !== SPANNED_OPERATION_TYPE) { + if (operation?.type !== FLUE_OPERATION.AGENT) { return next(); } diff --git a/packages/server-utils/src/ai/flue/utils.ts b/packages/server-utils/src/ai/flue/utils.ts index 289af9c18f6d..c666a858719a 100644 --- a/packages/server-utils/src/ai/flue/utils.ts +++ b/packages/server-utils/src/ai/flue/utils.ts @@ -1,4 +1,4 @@ -import type { Span } from '@sentry/core'; +import type { LRUMap, Span } from '@sentry/core'; import { SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, SPAN_STATUS_ERROR, startInactiveSpan, stringify } from '@sentry/core'; import { GEN_AI_CONVERSATION_ID, @@ -32,7 +32,7 @@ import { SERVER_PORT, } from '@sentry/conventions/attributes'; import { getGenAiSpanOp } from '../core/utils'; -import { FLUE_ORIGIN } from './constants'; +import { FLUE_ORIGIN, MAX_TRACKED_FLUE_SPANS } from './constants'; import type { FlueModelRequestInfo, FlueObservation, FlueUsage } from './types'; /** @@ -49,13 +49,36 @@ export function sentryTraceFromTraceparent(traceparent: string): string | undefi return `${traceId}-${spanId}-${parseInt(flags, 16) % 2 === 1 ? '1' : '0'}`; } -export function startTurnSpan(observation: FlueObservation, turnSpans: Map): void { +/** + * Turn and tool spans, keyed by the id of the work they cover. Bounded, because the key is only + * removed when the matching end observation arrives and an abandoned stream never emits one. + */ +export type SpanTracker = LRUMap; + +/** + * Store a span under `key`. Callers only reach this with a key the tracker does not hold, so the + * one span at risk of being dropped without `end()` is the oldest entry, which `LRUMap.set` silently + * evicts once the tracker is full. + */ +function trackSpan(tracker: SpanTracker, key: string, span: Span): void { + if (tracker.size >= MAX_TRACKED_FLUE_SPANS) { + const oldestKey = tracker.keys()[0]; + if (oldestKey !== undefined) { + tracker.remove(oldestKey)?.end(); + } + } + + tracker.set(key, span); +} + +export function startTurnSpan(observation: FlueObservation, turnSpans: SpanTracker): void { const { turnId } = observation; - if (!turnId || turnSpans.has(turnId)) { + if (!turnId || turnSpans.get(turnId)) { return; } - turnSpans.set( + trackSpan( + turnSpans, turnId, startInactiveSpan({ name: 'chat', @@ -72,13 +95,13 @@ export function startTurnSpan(observation: FlueObservation, turnSpans: Map, recordOutputs: boolean): void { +export function endTurnSpan(observation: FlueObservation, turnSpans: SpanTracker, recordOutputs: boolean): void { const { turnId } = observation; const span = turnId ? turnSpans.get(turnId) : undefined; if (!span || !turnId) { return; } - turnSpans.delete(turnId); + turnSpans.remove(turnId); const requestedModel = observation.request?.requestedModel; const responseModel = observation.response?.responseModel; @@ -160,13 +183,14 @@ export function setUsageAttributes(span: Span, usage: FlueUsage | undefined, isE * 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. */ -export function startToolSpan(observation: FlueObservation, toolSpans: Map, recordInputs: boolean): void { +export function startToolSpan(observation: FlueObservation, toolSpans: SpanTracker, recordInputs: boolean): void { const { toolCallId, toolName } = observation; - if (!toolCallId || toolSpans.has(toolCallId)) { + if (!toolCallId || toolSpans.get(toolCallId)) { return; } - toolSpans.set( + trackSpan( + toolSpans, toolCallId, startInactiveSpan({ name: `execute_tool ${toolName ?? 'unknown'}`, @@ -184,13 +208,13 @@ export function startToolSpan(observation: FlueObservation, toolSpans: Map, recordOutputs: boolean): void { +export function endToolSpan(observation: FlueObservation, toolSpans: SpanTracker, recordOutputs: boolean): void { const { toolCallId } = observation; const span = toolCallId ? toolSpans.get(toolCallId) : undefined; if (!span || !toolCallId) { return; } - toolSpans.delete(toolCallId); + toolSpans.remove(toolCallId); if (recordOutputs && observation.result !== undefined) { span.setAttribute(GEN_AI_TOOL_CALL_RESULT, stringify(observation.result)); @@ -206,7 +230,7 @@ export function endToolSpan(observation: FlueObservation, toolSpans: Map): void { +export function recordRequestContent(observation: FlueObservation, turnSpans: SpanTracker): void { const { turnId } = observation; const span = turnId ? turnSpans.get(turnId) : undefined; const input = observation.request?.input; From 0c24d2ba583115daf981365ba08d4dc129c0c8bc Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Fri, 11 Sep 2026 14:52:50 +0200 Subject: [PATCH 4/4] fix(astro, elysia): Re-export `createFlueInstrumentation` Astro keeps a hand-maintained list of the `@sentry/node` re-exports, because Vite puts a wildcard re-export under `default` in prod builds. The list missed `createFlueInstrumentation`, which fails the `node-exports-test-app` check that compares every dependent against `@sentry/node`. Nextjs, remix and sveltekit use a real `export *` and pick it up on their own. Elysia has the same hand-maintained shape and the same gap. Nothing covers it in CI, but it sits next to `mastraIntegration` either way. Also drops `FLUE_INTEGRATION_NAME` and `FLUE_MODULE_NAME`. Both are left over from the module-binding approach and are referenced nowhere now that registration is the user's call. Co-Authored-By: Claude Opus 5 --- packages/astro/src/index.server.ts | 1 + packages/elysia/src/index.ts | 1 + packages/server-utils/src/ai/flue/constants.ts | 4 ---- 3 files changed, 2 insertions(+), 4 deletions(-) diff --git a/packages/astro/src/index.server.ts b/packages/astro/src/index.server.ts index 030d150878d7..b862b2588ec8 100644 --- a/packages/astro/src/index.server.ts +++ b/packages/astro/src/index.server.ts @@ -94,6 +94,7 @@ export { openAIIntegration, langChainIntegration, langGraphIntegration, + createFlueInstrumentation, mastraIntegration, SentryMastraExporter, parameterize, diff --git a/packages/elysia/src/index.ts b/packages/elysia/src/index.ts index 904b707b1a37..9cab170cce2b 100644 --- a/packages/elysia/src/index.ts +++ b/packages/elysia/src/index.ts @@ -63,6 +63,7 @@ export { openAIIntegration, langChainIntegration, langGraphIntegration, + createFlueInstrumentation, mastraIntegration, SentryMastraExporter, modulesIntegration, diff --git a/packages/server-utils/src/ai/flue/constants.ts b/packages/server-utils/src/ai/flue/constants.ts index bde674b2572a..4f85fb7280f4 100644 --- a/packages/server-utils/src/ai/flue/constants.ts +++ b/packages/server-utils/src/ai/flue/constants.ts @@ -1,7 +1,3 @@ -export const FLUE_INTEGRATION_NAME = 'Flue' as const; - -export const FLUE_MODULE_NAME = '@flue/runtime'; - export const FLUE_ORIGIN = 'auto.ai.flue'; /**