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/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/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/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..4f85fb7280f4 --- /dev/null +++ b/packages/server-utils/src/ai/flue/constants.ts @@ -0,0 +1,30 @@ +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'); + +/** + * 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 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 new file mode 100644 index 000000000000..e98720446786 --- /dev/null +++ b/packages/server-utils/src/ai/flue/index.ts @@ -0,0 +1,194 @@ +import type { Span } from '@sentry/core'; +import { + _INTERNAL_shouldSkipAiProviderWrapping, + _INTERNAL_skipAiProviderWrapping, + continueTrace, + getActiveSpan, + LRUMap, + SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, + startSpan, + withActiveSpan, +} from '@sentry/core'; +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_OPERATION, FLUE_ORIGIN, MAX_TRACKED_FLUE_SPANS } from './constants'; +import type { SpanTracker } from './utils'; +import { + endToolSpan, + endTurnSpan, + recordRequestContent, + sentryTraceFromTraceparent, + startToolSpan, + startTurnSpan, +} from './utils'; +import type { FlueInstrumentation } 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 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. + // + // 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 (!SKIPPED_PROVIDERS.every(provider => _INTERNAL_shouldSkipAiProviderWrapping(provider))) { + _INTERNAL_skipAiProviderWrapping(SKIPPED_PROVIDERS); + } + }; + + // 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(); + // 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, + + 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 === FLUE_OPERATION.TOOL) { + const toolSpan = operation.toolCallId ? toolSpans.get(operation.toolCallId) : undefined; + return toolSpan ? withActiveSpan(toolSpan, next) : next(); + } + if (operation?.type === FLUE_OPERATION.MODEL) { + const turnSpan = operation.turnId ? turnSpans.get(operation.turnId) : undefined; + return turnSpan ? withActiveSpan(turnSpan, next) : next(); + } + + if (operation?.type !== FLUE_OPERATION.AGENT) { + return next(); + } + + // 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(); + } + + 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: 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); + return; + case 'turn_request': + if (recordInputs) { + recordRequestContent(observation, turnSpans); + } + return; + case 'turn': + endTurnSpan(observation, turnSpans, recordOutputs); + return; + case 'tool_start': + startToolSpan(observation, toolSpans, recordInputs); + return; + case 'tool': + endToolSpan(observation, toolSpans, recordOutputs); + return; + default: + return; + } + }, + + dispose: () => { + for (const span of [...turnSpans.values(), ...toolSpans.values()]) { + span.end(); + } + agentSpans.clear(); + turnSpans.clear(); + toolSpans.clear(); + }, + }; +} 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..525cda38a8a3 --- /dev/null +++ b/packages/server-utils/src/ai/flue/types.ts @@ -0,0 +1,117 @@ +/** + * 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; + /** Provider slug (`anthropic`, `openrouter`) — what `gen_ai.provider.name` wants. */ + providerId?: string; + temperature?: number; + maxTokens?: number; + reasoningLevel?: string; + serverAddress?: string; + serverPort?: number; +} + +/** 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; + submissionId?: string; + operationId?: 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; + 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; + 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/flue/utils.ts b/packages/server-utils/src/ai/flue/utils.ts new file mode 100644 index 000000000000..c666a858719a --- /dev/null +++ b/packages/server-utils/src/ai/flue/utils.ts @@ -0,0 +1,275 @@ +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, + 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, MAX_TRACKED_FLUE_SPANS } 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'}`; +} + +/** + * 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.get(turnId)) { + return; + } + + trackSpan( + turnSpans, + 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: SpanTracker, recordOutputs: boolean): void { + const { turnId } = observation; + const span = turnId ? turnSpans.get(turnId) : undefined; + if (!span || !turnId) { + return; + } + turnSpans.remove(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: SpanTracker, recordInputs: boolean): void { + const { toolCallId, toolName } = observation; + if (!toolCallId || toolSpans.get(toolCallId)) { + return; + } + + trackSpan( + toolSpans, + 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: SpanTracker, recordOutputs: boolean): void { + const { toolCallId } = observation; + const span = toolCallId ? toolSpans.get(toolCallId) : undefined; + if (!span || !toolCallId) { + return; + } + toolSpans.remove(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: SpanTracker): 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); +} 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';