From a42d55e1410f6e0c4a12c22fee42e1ef941b62b9 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Fri, 11 Sep 2026 08:11:32 +0200 Subject: [PATCH 01/10] 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 ba7b2e95e8ed..06794f7a380f 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 caf4e1116045..b98e4b5eb2f1 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 f75e4c429e78..5fb8ed0863e3 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 0225063baeed..f99ad0f7c85c 100644 --- a/packages/deno/src/index.ts +++ b/packages/deno/src/index.ts @@ -137,6 +137,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 3d08265de05f..c328ec171666 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 5fa4d57e2466..fde46f43ed33 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 74171e9f047b..a2369110ba68 100644 --- a/packages/server-utils/src/index.ts +++ b/packages/server-utils/src/index.ts @@ -42,6 +42,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 3a6b1a77fde379f16380010912ee35b142eba4da Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Fri, 11 Sep 2026 10:19:01 +0200 Subject: [PATCH 02/10] 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 477df590904e5a4781305531c49351b6cd89cfe6 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Fri, 11 Sep 2026 13:45:05 +0200 Subject: [PATCH 03/10] 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 d78c1af0c676806ab243e93f3d1f240a53cbf0dc Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Fri, 11 Sep 2026 14:52:50 +0200 Subject: [PATCH 04/10] 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 8e22aa0f5dd4..03150cf4619f 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'; /** From 736634fae35f1a8750e16fa8f1d32be75aff265b Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Mon, 14 Sep 2026 10:50:08 +0300 Subject: [PATCH 05/10] fix(server-utils): Serialize the Flue `finish_reasons` attribute `@sentry/conventions` declares `GEN_AI_RESPONSE_FINISH_REASONS_TYPE = string`, and Flue wrote a raw `string[]`, so any consumer parsing the field as a string would not read it. `ai/core`, Mastra, OpenAI and Vercel AI all serialize it. Bedrock and LangGraph also write raw arrays on this attribute and look wrong for the same reason, but they are outside this change. Co-Authored-By: Claude Opus 5 --- packages/server-utils/src/ai/flue/utils.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/packages/server-utils/src/ai/flue/utils.ts b/packages/server-utils/src/ai/flue/utils.ts index c666a858719a..aea15d6eb658 100644 --- a/packages/server-utils/src/ai/flue/utils.ts +++ b/packages/server-utils/src/ai/flue/utils.ts @@ -129,7 +129,9 @@ export function endTurnSpan(observation: FlueObservation, turnSpans: SpanTracker span.setAttribute(GEN_AI_RESPONSE_ID, responseId); } if (finishReason) { - span.setAttribute(GEN_AI_RESPONSE_FINISH_REASONS, [finishReason]); + // Serialized, not a raw array: the conventions declare this attribute's value type as `string`, + // and that is what `ai/core`, Mastra, OpenAI and Vercel AI all write. + span.setAttribute(GEN_AI_RESPONSE_FINISH_REASONS, stringify([finishReason])); } const output = observation.response?.output; From af0f86f9ec2f16928bd3c70db1664d92c4657a37 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Mon, 14 Sep 2026 16:12:55 +0300 Subject: [PATCH 06/10] feat(server-utils): Capture errors thrown inside Flue tools and turns Flue catches whatever a tool throws and feeds it back to the model as a tool result, so nothing propagates for the SDK's global handlers to see. A throwing tool produced an errored `execute_tool` span and no error event at all, which an e2e run against a real agent confirmed. The observation carries `errorInfo` with the original name, message and stack, so rebuild an `Error` from it rather than capturing the serialized shape. Marked handled, because Flue does catch it. Co-Authored-By: Claude Opus 5 --- packages/server-utils/src/ai/flue/types.ts | 12 +++++++ packages/server-utils/src/ai/flue/utils.ts | 40 ++++++++++++++++++++-- 2 files changed, 50 insertions(+), 2 deletions(-) diff --git a/packages/server-utils/src/ai/flue/types.ts b/packages/server-utils/src/ai/flue/types.ts index 525cda38a8a3..bdbd58455518 100644 --- a/packages/server-utils/src/ai/flue/types.ts +++ b/packages/server-utils/src/ai/flue/types.ts @@ -55,6 +55,17 @@ export interface FlueModelResponse { finishReason?: string; } +/** + * Mirrors Flue's `errorInfo`, which it attaches to a failed tool or turn observation. Flue catches + * the throw and turns it into a tool result, so this serialized form is the only trace of it. + */ +export interface FlueErrorInfo { + type?: string; + name?: string; + message?: string; + stack?: 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. @@ -71,6 +82,7 @@ export interface FlueObservation { toolName?: string; toolCallId?: string; isError?: boolean; + errorInfo?: FlueErrorInfo; purpose?: string; durationMs?: number; request?: FlueModelRequest; diff --git a/packages/server-utils/src/ai/flue/utils.ts b/packages/server-utils/src/ai/flue/utils.ts index aea15d6eb658..a5973e71ce2e 100644 --- a/packages/server-utils/src/ai/flue/utils.ts +++ b/packages/server-utils/src/ai/flue/utils.ts @@ -1,5 +1,11 @@ import type { LRUMap, Span } from '@sentry/core'; -import { SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, SPAN_STATUS_ERROR, startInactiveSpan, stringify } from '@sentry/core'; +import { + captureException, + SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, + SPAN_STATUS_ERROR, + startInactiveSpan, + stringify, +} from '@sentry/core'; import { GEN_AI_CONVERSATION_ID, GEN_AI_COST_CACHE_CREATION_INPUT_TOKENS, @@ -33,7 +39,7 @@ import { } from '@sentry/conventions/attributes'; import { getGenAiSpanOp } from '../core/utils'; import { FLUE_ORIGIN, MAX_TRACKED_FLUE_SPANS } from './constants'; -import type { FlueModelRequestInfo, FlueObservation, FlueUsage } from './types'; +import type { FlueErrorInfo, FlueModelRequestInfo, FlueObservation, FlueUsage } from './types'; /** * Flue persists the incoming W3C `traceparent` at admission and replays it on the agent operation. @@ -71,6 +77,34 @@ function trackSpan(tracker: SpanTracker, key: string, span: Span): void { tracker.set(key, span); } +/** + * Report a failed tool or turn as an error event. + * + * Flue catches whatever the tool threw and reports it as a tool result, so nothing propagates for + * the SDK's global handlers to see: without this, a throwing tool produces an errored span and no + * error at all. `errorInfo` carries the original name, message and stack, so rebuild an `Error` + * from it rather than capturing the serialized shape. + */ +function captureFlueError(errorInfo: FlueErrorInfo | undefined, kind: 'tool' | 'turn'): void { + if (!errorInfo) { + return; + } + + const error = new Error(errorInfo.message ?? `Flue ${kind} failed`); + error.name = errorInfo.name ?? errorInfo.type ?? 'Error'; + if (errorInfo.stack) { + error.stack = errorInfo.stack; + } + + captureException(error, { + mechanism: { + // Handled: Flue caught it and fed it back to the model as a tool result. + handled: true, + type: `${FLUE_ORIGIN}.${kind}_error`, + }, + }); +} + export function startTurnSpan(observation: FlueObservation, turnSpans: SpanTracker): void { const { turnId } = observation; if (!turnId || turnSpans.get(turnId)) { @@ -143,6 +177,7 @@ export function endTurnSpan(observation: FlueObservation, turnSpans: SpanTracker if (observation.isError) { span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); + captureFlueError(observation.errorInfo, 'turn'); } span.end(); } @@ -224,6 +259,7 @@ export function endToolSpan(observation: FlueObservation, toolSpans: SpanTracker if (observation.isError) { span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); + captureFlueError(observation.errorInfo, 'tool'); } span.end(); } From ccec047d7b7bbb4c70d98f8745e1179c65bd3b84 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Mon, 14 Sep 2026 16:17:49 +0300 Subject: [PATCH 07/10] ref(server-utils): Align Flue error capture with the Mastra integration Matches #24374: mechanism `auto.ai.flue` with `handled: false`, and the capture runs under the operation's own span so the issue lands on the right trace. Unlike Mastra, the rebuild from `errorInfo` stays. Mastra's `errorInfo` is `{ name, message }` with no stack, which is why it has to reach for the channel's real `Error`; Flue's carries the original stack, so there is nothing to gain from a second seam that might not fire on every path. Co-Authored-By: Claude Opus 5 --- packages/server-utils/src/ai/flue/utils.ts | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/packages/server-utils/src/ai/flue/utils.ts b/packages/server-utils/src/ai/flue/utils.ts index a5973e71ce2e..4ceab8279a48 100644 --- a/packages/server-utils/src/ai/flue/utils.ts +++ b/packages/server-utils/src/ai/flue/utils.ts @@ -5,6 +5,7 @@ import { SPAN_STATUS_ERROR, startInactiveSpan, stringify, + withActiveSpan, } from '@sentry/core'; import { GEN_AI_CONVERSATION_ID, @@ -85,7 +86,7 @@ function trackSpan(tracker: SpanTracker, key: string, span: Span): void { * error at all. `errorInfo` carries the original name, message and stack, so rebuild an `Error` * from it rather than capturing the serialized shape. */ -function captureFlueError(errorInfo: FlueErrorInfo | undefined, kind: 'tool' | 'turn'): void { +function captureFlueError(span: Span, errorInfo: FlueErrorInfo | undefined, kind: 'tool' | 'turn'): void { if (!errorInfo) { return; } @@ -96,12 +97,10 @@ function captureFlueError(errorInfo: FlueErrorInfo | undefined, kind: 'tool' | ' error.stack = errorInfo.stack; } - captureException(error, { - mechanism: { - // Handled: Flue caught it and fed it back to the model as a tool result. - handled: true, - type: `${FLUE_ORIGIN}.${kind}_error`, - }, + // Captured under the operation's own span so the issue lands on the right trace, matching how the + // Mastra integration attaches its captures. + withActiveSpan(span, () => { + captureException(error, { mechanism: { handled: false, type: FLUE_ORIGIN } }); }); } @@ -177,7 +176,7 @@ export function endTurnSpan(observation: FlueObservation, turnSpans: SpanTracker if (observation.isError) { span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); - captureFlueError(observation.errorInfo, 'turn'); + captureFlueError(span, observation.errorInfo, 'turn'); } span.end(); } @@ -259,7 +258,7 @@ export function endToolSpan(observation: FlueObservation, toolSpans: SpanTracker if (observation.isError) { span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); - captureFlueError(observation.errorInfo, 'tool'); + captureFlueError(span, observation.errorInfo, 'tool'); } span.end(); } From 17ed22c2c170a3ad8d5b3a1122933214aadcb94f Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Mon, 14 Sep 2026 16:33:28 +0300 Subject: [PATCH 08/10] docs(server-utils): Explain what the Flue instrumentation key actually buys The previous comment justified it by coexistence with `@flue/opentelemetry`, which is free either way since the keys differ. The real reason is repeat registration: Flue deduplicates on object identity and the factory returns a new object each call, so without a key a second `instrument()` silently doubles every span. With one, Flue throws in production and swaps the registration in dev, which is what keeps `vite dev` from stacking observers. Co-Authored-By: Claude Opus 5 --- packages/server-utils/src/ai/flue/constants.ts | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/packages/server-utils/src/ai/flue/constants.ts b/packages/server-utils/src/ai/flue/constants.ts index 4f85fb7280f4..ae568d24312a 100644 --- a/packages/server-utils/src/ai/flue/constants.ts +++ b/packages/server-utils/src/ai/flue/constants.ts @@ -1,9 +1,16 @@ 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`. + * Identifies our registration in Flue's keyed instrumentation registry. + * + * `key` is optional, but without one there is no protection against registering twice: Flue + * deduplicates on object identity, and `createFlueInstrumentation()` returns a new object each + * call, so a second `instrument()` would silently add a second observer and interceptor and + * duplicate every span. + * + * With a key, Flue handles the repeat itself — it throws `InstrumentationAlreadyInstalledError` in + * production, and in dev disposes the previous registration and swaps in the new one, which is what + * stops `vite dev` from stacking observers across reloads. */ export const FLUE_INSTRUMENTATION_KEY = Symbol.for('sentry.flue.instrumentation'); From d0e40c6caa542be92572b802e07e7e4873628fd7 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Mon, 14 Sep 2026 16:52:54 +0300 Subject: [PATCH 09/10] fix(server-utils): Narrow Flue error capture to tools A failed turn carries no `errorInfo`: forcing a provider auth failure produced an errored `chat` span and zero error events, so the turn branch could never fire. Dropping it also removes the double-report concern raised in review, since the tool path is the only one that reports. Also records why `dispose()` ends only the turn and tool spans: those come from `startInactiveSpan` so nothing else ends them, while the agent spans come from `startSpan`, which ends them when its callback settles. Ending those here would stamp an early end time on work still in flight. Co-Authored-By: Claude Opus 5 --- packages/server-utils/src/ai/flue/index.ts | 4 ++++ packages/server-utils/src/ai/flue/utils.ts | 20 +++++++++++--------- 2 files changed, 15 insertions(+), 9 deletions(-) diff --git a/packages/server-utils/src/ai/flue/index.ts b/packages/server-utils/src/ai/flue/index.ts index e98720446786..67e59c9fcdc6 100644 --- a/packages/server-utils/src/ai/flue/index.ts +++ b/packages/server-utils/src/ai/flue/index.ts @@ -182,6 +182,10 @@ export function createFlueInstrumentation(options: FlueOptions = {}): FlueInstru } }, + // Only the turn and tool spans are ended here. They come from `startInactiveSpan`, so nothing + // else will. The agent spans come from `startSpan`, which ends them when its callback settles — + // ending them here would stamp an early end time on work that is still running, and the later + // `end()` would be ignored. dispose: () => { for (const span of [...turnSpans.values(), ...toolSpans.values()]) { span.end(); diff --git a/packages/server-utils/src/ai/flue/utils.ts b/packages/server-utils/src/ai/flue/utils.ts index 4ceab8279a48..8dbed9b34590 100644 --- a/packages/server-utils/src/ai/flue/utils.ts +++ b/packages/server-utils/src/ai/flue/utils.ts @@ -79,19 +79,22 @@ function trackSpan(tracker: SpanTracker, key: string, span: Span): void { } /** - * Report a failed tool or turn as an error event. + * Report a failed tool as an error event. * - * Flue catches whatever the tool threw and reports it as a tool result, so nothing propagates for - * the SDK's global handlers to see: without this, a throwing tool produces an errored span and no - * error at all. `errorInfo` carries the original name, message and stack, so rebuild an `Error` - * from it rather than capturing the serialized shape. + * Flue catches whatever the tool threw and hands it back to the model as a tool result, so nothing + * propagates for the SDK's global handlers to see: without this a throwing tool produces an errored + * span and no issue at all. `errorInfo` carries the original name, message and stack, so rebuild an + * `Error` from it rather than capturing the serialized shape. + * + * Tools only. A failed turn carries no `errorInfo` — a provider auth failure produces an errored + * `chat` span and nothing to rebuild from — so there is no turn equivalent to capture. */ -function captureFlueError(span: Span, errorInfo: FlueErrorInfo | undefined, kind: 'tool' | 'turn'): void { +function captureToolError(span: Span, errorInfo: FlueErrorInfo | undefined): void { if (!errorInfo) { return; } - const error = new Error(errorInfo.message ?? `Flue ${kind} failed`); + const error = new Error(errorInfo.message ?? 'Flue tool failed'); error.name = errorInfo.name ?? errorInfo.type ?? 'Error'; if (errorInfo.stack) { error.stack = errorInfo.stack; @@ -176,7 +179,6 @@ export function endTurnSpan(observation: FlueObservation, turnSpans: SpanTracker if (observation.isError) { span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); - captureFlueError(span, observation.errorInfo, 'turn'); } span.end(); } @@ -258,7 +260,7 @@ export function endToolSpan(observation: FlueObservation, toolSpans: SpanTracker if (observation.isError) { span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); - captureFlueError(span, observation.errorInfo, 'tool'); + captureToolError(span, observation.errorInfo); } span.end(); } From d843c78803e2a52475bb4a2992773932e6574e41 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Mon, 14 Sep 2026 23:03:38 +0300 Subject: [PATCH 10/10] fix(server-utils): Mark the Flue tool error as handled Copied `handled: false` from the Mastra integration, but the two capture from different places. Mastra observes a tracing channel rejection, where the handled state is unknowable. Flue reports the failure on its observation stream after catching the throw and returning it to the model as a tool result, so it is definitively handled and no global hook will see it. Co-Authored-By: Claude Opus 5 --- packages/server-utils/src/ai/flue/utils.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/server-utils/src/ai/flue/utils.ts b/packages/server-utils/src/ai/flue/utils.ts index 8dbed9b34590..a1b463cbfbd7 100644 --- a/packages/server-utils/src/ai/flue/utils.ts +++ b/packages/server-utils/src/ai/flue/utils.ts @@ -103,7 +103,10 @@ function captureToolError(span: Span, errorInfo: FlueErrorInfo | undefined): voi // Captured under the operation's own span so the issue lands on the right trace, matching how the // Mastra integration attaches its captures. withActiveSpan(span, () => { - captureException(error, { mechanism: { handled: false, type: FLUE_ORIGIN } }); + // Handled: this is not a rejection observed on a tracing channel, where the handled state is + // unknowable. Flue caught the throw and returned it to the model as a tool result, so it is + // definitively handled and no global hook will ever see it. + captureException(error, { mechanism: { handled: true, type: FLUE_ORIGIN } }); }); }