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/core/src/utils/worldwide.ts b/packages/core/src/utils/worldwide.ts index 82f302bec5c3..04134db1c8b0 100644 --- a/packages/core/src/utils/worldwide.ts +++ b/packages/core/src/utils/worldwide.ts @@ -86,6 +86,13 @@ export type InternalGlobal = { * `init()` and instantiates them. */ integrations?: Map Integration>; + /** + * Named exports handed back by an instrumented module's injected snippet, keyed by module + * name. For libraries instrumented through a registration API they expose rather than through + * channels: that API closes over module-scope state, so an integration must call the app's own + * copy, and the snippet runs inside it. + */ + moduleBindings?: Map>; /** * Set once `registerDiagnosticsChannelInjection()` has run but could not * install the runtime module hooks — most commonly because diff --git a/packages/node/src/index.ts b/packages/node/src/index.ts index 03532223a506..c1f704fd2909 100644 --- a/packages/node/src/index.ts +++ b/packages/node/src/index.ts @@ -20,6 +20,7 @@ export { langChainIntegration, langGraphIntegration, lruMemoizerIntegration, + flueIntegration, mastraIntegration, SentryMastraExporter, mongoIntegration, diff --git a/packages/server-utils/src/ai/flue/constants.ts b/packages/server-utils/src/ai/flue/constants.ts new file mode 100644 index 000000000000..2591db9587f9 --- /dev/null +++ b/packages/server-utils/src/ai/flue/constants.ts @@ -0,0 +1,20 @@ +export const FLUE_INTEGRATION_NAME = 'Flue' as const; + +export const FLUE_MODULE_NAME = '@flue/runtime'; + +export const FLUE_ORIGIN = 'auto.ai.flue'; + +/** + * Identifies our registration in Flue's keyed instrumentation registry. A distinct key lets us + * coexist with `@flue/opentelemetry` (which registers under its own key) and makes a repeated + * `instrument()` call a no-op instead of throwing `InstrumentationAlreadyInstalledError`. + */ +export const FLUE_INSTRUMENTATION_KEY = Symbol.for('sentry.flue.instrumentation'); + +/** + * Flue drives one LLM call through many `model` operations (one per stream read), and its `agent` + * operation nests inside itself once per submission. Only `agent` is spanned from the interceptor, + * and only at the outermost depth; the turn span is driven from the observation stream instead, + * where `turn_start`/`turn` are exactly one-to-one with a model call. + */ +export const SPANNED_OPERATION_TYPE = 'agent'; diff --git a/packages/server-utils/src/ai/flue/index.ts b/packages/server-utils/src/ai/flue/index.ts new file mode 100644 index 000000000000..ec6cbe74316e --- /dev/null +++ b/packages/server-utils/src/ai/flue/index.ts @@ -0,0 +1,331 @@ +import type { Span } from '@sentry/core'; +import { + _INTERNAL_skipAiProviderWrapping, + SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, + SPAN_STATUS_ERROR, + startInactiveSpan, + startSpan, + stringify, + withActiveSpan, +} from '@sentry/core'; +import { + GEN_AI_AGENT_NAME, + GEN_AI_CONVERSATION_ID, + GEN_AI_INPUT_MESSAGES, + GEN_AI_OUTPUT_MESSAGES, + GEN_AI_COST_CACHE_CREATION_INPUT_TOKENS, + GEN_AI_COST_CACHE_READ_INPUT_TOKENS, + GEN_AI_COST_INPUT_TOKENS, + GEN_AI_COST_OUTPUT_TOKENS, + GEN_AI_COST_TOTAL_TOKENS, + GEN_AI_OPERATION_NAME, + GEN_AI_PROVIDER_NAME, + GEN_AI_REQUEST_MODEL, + GEN_AI_RESPONSE_FINISH_REASONS, + GEN_AI_RESPONSE_ID, + GEN_AI_RESPONSE_MODEL, + GEN_AI_SYSTEM_INSTRUCTIONS, + GEN_AI_TOOL_CALL_ARGUMENTS, + GEN_AI_TOOL_CALL_RESULT, + GEN_AI_TOOL_DEFINITIONS, + GEN_AI_TOOL_NAME, + GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS, + GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS, + GEN_AI_USAGE_INPUT_TOKENS, + GEN_AI_USAGE_OUTPUT_TOKENS, + GEN_AI_USAGE_TOTAL_TOKENS, +} from '@sentry/conventions/attributes'; +import { ANTHROPIC_AI_INTEGRATION_NAME } from '../anthropic-ai/constants'; +import type { GenAiOptions } from '../core/utils'; +import { getGenAiSpanOp, resolveAIRecordingOptions } from '../core/utils'; +import { GOOGLE_GENAI_INTEGRATION_NAME } from '../google-genai/constants'; +import { OPENAI_INTEGRATION_NAME } from '../openai/constants'; +import { FLUE_INSTRUMENTATION_KEY, FLUE_ORIGIN, SPANNED_OPERATION_TYPE } from './constants'; +import type { FlueInstrumentation, FlueObservation, FlueUsage } from './types'; + +export type FlueOptions = GenAiOptions; + +const SKIPPED_PROVIDERS = [OPENAI_INTEGRATION_NAME, ANTHROPIC_AI_INTEGRATION_NAME, GOOGLE_GENAI_INTEGRATION_NAME]; + +/** + * Build the object to hand to `instrument()` from `@flue/runtime`. + * + * The two callbacks own different halves of the result: + * + * - `interceptor` wraps agent execution, so the agent span is *active* for its duration and every + * span opened underneath parents correctly. + * - `observe` opens and closes the turn span, because Flue's `turn_start`/`turn` events are the + * only one-to-one signal for a model call and `turn` is what carries usage and cost. + * + * Message content, tool arguments and tool results are gated on `recordInputs`/`recordOutputs`, + * which fall back to the client's `dataCollection.genAI` settings. + */ +export function createFlueInstrumentation(options: FlueOptions = {}): FlueInstrumentation { + // Flue drives the providers through `@earendil-works/pi-ai`, which bundles the `openai`, + // `@anthropic-ai/sdk` and `@google/genai` clients. Left alone they instrument the same call this + // reports as a turn, emitting a second `gen_ai.chat` beside ours. Done here rather than in + // `flueIntegration` so registering by hand — the only option on Cloudflare, where agents run in + // per-Durable-Object isolates — gets it too. + _INTERNAL_skipAiProviderWrapping(SKIPPED_PROVIDERS); + + const { recordInputs, recordOutputs } = resolveAIRecordingOptions(options); + const turnSpans = new Map(); + const toolSpans = new Map(); + let agentSpan: Span | undefined; + let agentDepth = 0; + + return { + key: FLUE_INSTRUMENTATION_KEY, + + interceptor: async (operation, ctx, next) => { + if (operation?.type !== SPANNED_OPERATION_TYPE) { + return next(); + } + + // A submission's agent operation re-enters once, and the two carry different halves of the + // agent's identity: the outer context names the agent, the inner one names the conversation. + // Only the outer becomes a span, so the conversation id is lifted onto it from the re-entry. + if (agentDepth++ > 0) { + if (ctx.conversationId) { + agentSpan?.setAttribute(GEN_AI_CONVERSATION_ID, ctx.conversationId); + } + try { + return await next(); + } finally { + agentDepth--; + } + } + + return startSpan( + { + name: `invoke_agent ${ctx.agentName ?? 'agent'}`, + op: getGenAiSpanOp('invoke_agent'), + attributes: { + [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: FLUE_ORIGIN, + [GEN_AI_OPERATION_NAME]: 'invoke_agent', + ...(ctx.agentName ? { [GEN_AI_AGENT_NAME]: ctx.agentName } : {}), + ...(ctx.conversationId ? { [GEN_AI_CONVERSATION_ID]: ctx.conversationId } : {}), + }, + }, + async span => { + agentSpan = span; + try { + return await next(); + } finally { + agentDepth--; + agentSpan = undefined; + } + }, + ); + }, + + observe: observation => { + switch (observation.type) { + case 'turn_start': + startTurnSpan(observation, turnSpans, agentSpan); + return; + case 'turn_request': + if (recordInputs) { + recordRequestContent(observation, turnSpans); + } + return; + case 'turn': + endTurnSpan(observation, turnSpans, recordOutputs); + return; + case 'tool_start': + startToolSpan(observation, toolSpans, agentSpan, recordInputs); + return; + case 'tool': + endToolSpan(observation, toolSpans, recordOutputs); + return; + default: + return; + } + }, + + dispose: () => { + for (const span of [...turnSpans.values(), ...toolSpans.values()]) { + span.end(); + } + turnSpans.clear(); + toolSpans.clear(); + }, + }; +} + +function startTurnSpan(observation: FlueObservation, turnSpans: Map, agentSpan: Span | undefined): void { + const { turnId } = observation; + if (!turnId || turnSpans.has(turnId)) { + return; + } + + const open = (): Span => + startInactiveSpan({ + name: 'chat', + op: getGenAiSpanOp('chat'), + attributes: { + [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: FLUE_ORIGIN, + [GEN_AI_OPERATION_NAME]: 'chat', + ...(observation.conversationId ? { [GEN_AI_CONVERSATION_ID]: observation.conversationId } : {}), + }, + }); + + // `observe` runs inside the agent operation, but the active span there is whatever the provider + // SDK last opened — parent explicitly so the turn always hangs off the agent. + turnSpans.set(turnId, agentSpan ? withActiveSpan(agentSpan, open) : open()); +} + +function endTurnSpan(observation: FlueObservation, turnSpans: Map, recordOutputs: boolean): void { + const { turnId } = observation; + const span = turnId ? turnSpans.get(turnId) : undefined; + if (!span || !turnId) { + return; + } + turnSpans.delete(turnId); + + const requestedModel = observation.request?.requestedModel; + const responseModel = observation.response?.responseModel; + const model = responseModel ?? requestedModel; + if (model) { + span.updateName(`chat ${model}`); + } + if (requestedModel) { + span.setAttribute(GEN_AI_REQUEST_MODEL, requestedModel); + } + if (responseModel) { + span.setAttribute(GEN_AI_RESPONSE_MODEL, responseModel); + } + + const provider = observation.request?.providerId ?? observation.request?.providerName; + if (provider) { + span.setAttribute(GEN_AI_PROVIDER_NAME, provider); + } + + const { responseId, finishReason } = observation.response ?? {}; + if (responseId) { + span.setAttribute(GEN_AI_RESPONSE_ID, responseId); + } + if (finishReason) { + span.setAttribute(GEN_AI_RESPONSE_FINISH_REASONS, [finishReason]); + } + + const output = observation.response?.output; + if (recordOutputs && output !== undefined) { + span.setAttribute(GEN_AI_OUTPUT_MESSAGES, stringify(output)); + } + + setUsageAttributes(span, observation.response?.usage, observation.isError); + + if (observation.isError) { + span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); + } + span.end(); +} + +/** + * Flue reports token counts and its own computed costs on the same `usage` object, so both are set + * here. The cost figures have no equivalent in the provider SDKs' own instrumentation. + */ +function setUsageAttributes(span: Span, usage: FlueUsage | undefined, isError?: boolean): void { + // A turn that failed before the provider billed anything reports every counter as 0. Writing + // those is noise that reads as a real zero-cost call, so skip the block entirely. + if (!usage || (isError && !usage.totalTokens)) { + return; + } + + const attributes: Record = {}; + const set = (key: string, value: number | undefined): void => { + if (typeof value === 'number') { + attributes[key] = value; + } + }; + + set(GEN_AI_USAGE_INPUT_TOKENS, usage.input); + set(GEN_AI_USAGE_OUTPUT_TOKENS, usage.output); + set(GEN_AI_USAGE_TOTAL_TOKENS, usage.totalTokens); + set(GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS, usage.cacheRead); + set(GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS, usage.cacheWrite); + + set(GEN_AI_COST_INPUT_TOKENS, usage.cost?.input); + set(GEN_AI_COST_OUTPUT_TOKENS, usage.cost?.output); + set(GEN_AI_COST_TOTAL_TOKENS, usage.cost?.total); + set(GEN_AI_COST_CACHE_READ_INPUT_TOKENS, usage.cost?.cacheRead); + set(GEN_AI_COST_CACHE_CREATION_INPUT_TOKENS, usage.cost?.cacheWrite); + + span.setAttributes(attributes); +} + +/** + * Tool spans hang off the agent invocation rather than the turn, matching how Flue's own + * OpenTelemetry adapter projects them: siblings of `chat`, correlated to model output by tool call + * id. Keyed by `toolCallId` so concurrent tool calls in one turn cannot cross-attribute. + */ +function startToolSpan( + observation: FlueObservation, + toolSpans: Map, + agentSpan: Span | undefined, + recordInputs: boolean, +): void { + const { toolCallId, toolName } = observation; + if (!toolCallId || toolSpans.has(toolCallId)) { + return; + } + + const open = (): Span => + startInactiveSpan({ + name: `execute_tool ${toolName ?? 'unknown'}`, + op: getGenAiSpanOp('execute_tool'), + attributes: { + [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: FLUE_ORIGIN, + [GEN_AI_OPERATION_NAME]: 'execute_tool', + ...(toolName ? { [GEN_AI_TOOL_NAME]: toolName } : {}), + ...(observation.conversationId ? { [GEN_AI_CONVERSATION_ID]: observation.conversationId } : {}), + ...(recordInputs && observation.args !== undefined + ? { [GEN_AI_TOOL_CALL_ARGUMENTS]: stringify(observation.args) } + : {}), + }, + }); + + toolSpans.set(toolCallId, agentSpan ? withActiveSpan(agentSpan, open) : open()); +} + +function endToolSpan(observation: FlueObservation, toolSpans: Map, recordOutputs: boolean): void { + const { toolCallId } = observation; + const span = toolCallId ? toolSpans.get(toolCallId) : undefined; + if (!span || !toolCallId) { + return; + } + toolSpans.delete(toolCallId); + + if (recordOutputs && observation.result !== undefined) { + span.setAttribute(GEN_AI_TOOL_CALL_RESULT, stringify(observation.result)); + } + + if (observation.isError) { + span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); + } + span.end(); +} + +/** + * `turn_request` is the only event carrying the request's content — the settled `turn` reports + * metadata alone — so input messages, system prompt and tool definitions are read from it. + */ +function recordRequestContent(observation: FlueObservation, turnSpans: Map): void { + const { turnId } = observation; + const span = turnId ? turnSpans.get(turnId) : undefined; + const input = observation.request?.input; + if (!span || !input) { + return; + } + + if (input.systemPrompt) { + span.setAttribute(GEN_AI_SYSTEM_INSTRUCTIONS, input.systemPrompt); + } + if (input.messages) { + span.setAttribute(GEN_AI_INPUT_MESSAGES, stringify(input.messages)); + } + if (input.tools?.length) { + span.setAttribute(GEN_AI_TOOL_DEFINITIONS, stringify(input.tools)); + } +} diff --git a/packages/server-utils/src/ai/flue/types.ts b/packages/server-utils/src/ai/flue/types.ts new file mode 100644 index 000000000000..05784d30c8e6 --- /dev/null +++ b/packages/server-utils/src/ai/flue/types.ts @@ -0,0 +1,101 @@ +/** + * Structural types for the subset of `@flue/runtime`'s instrumentation contract we consume. + * + * Declared locally rather than imported: `@flue/runtime` is ESM-only and not a dependency of this + * package, and the SDK must not import it. Mirrors `FlueInstrumentation`, `FlueObservation` and + * `FlueExecutionContext` as of `@flue/runtime` 2.x. + */ + +/** Token counts and Flue-computed costs on a settled turn. */ +export interface FlueUsage { + input?: number; + output?: number; + cacheRead?: number; + cacheWrite?: number; + totalTokens?: number; + cost?: { + input?: number; + output?: number; + cacheRead?: number; + cacheWrite?: number; + total?: number; + }; +} + +/** Mirrors `ModelRequestInfo`. */ +export interface FlueModelRequestInfo { + requestedModel?: string; + providerId?: string; + providerName?: string; +} + +/** Mirrors `ModelRequestInput` — the content half of `ModelRequest`, on `turn_request` only. */ +export interface FlueModelRequestInput { + systemPrompt?: string; + messages?: unknown[]; + tools?: unknown[]; +} + +/** `turn_request` carries `ModelRequest`, which is `ModelRequestInfo` plus the input. */ +export interface FlueModelRequest extends FlueModelRequestInfo { + input?: FlueModelRequestInput; +} + +/** Mirrors `ModelResponse`. */ +export interface FlueModelResponse { + responseId?: string; + responseModel?: string; + output?: unknown; + usage?: FlueUsage; + finishReason?: string; +} + +/** + * One event from Flue's observation stream. Only the fields we read are declared; Flue emits more + * event types than are handled here, and unknown types are ignored. + */ +export interface FlueObservation { + type: string; + agentName?: string; + conversationId?: string; + session?: string; + turnId?: string; + taskId?: string; + toolName?: string; + toolCallId?: string; + isError?: boolean; + purpose?: string; + durationMs?: number; + request?: FlueModelRequest; + args?: unknown; + result?: unknown; + response?: FlueModelResponse; +} + +/** The execution unit an interceptor wraps. */ +export interface FlueExecutionOperation { + type: string; + operationId?: string; + operationKind?: string; + turnId?: string; +} + +export interface FlueExecutionContext { + agentName?: string; + conversationId?: string; + session?: string; + turnId?: string; + taskId?: string; +} + +export interface FlueEventContext { + agentName?: string; +} + +/** The object `instrument()` accepts. */ +export interface FlueInstrumentation { + key: symbol; + observe: (observation: FlueObservation, ctx: FlueEventContext) => void; + interceptor: (operation: FlueExecutionOperation, ctx: FlueExecutionContext, next: () => Promise) => Promise; + dispose: () => void; +} diff --git a/packages/server-utils/src/ai/index.ts b/packages/server-utils/src/ai/index.ts index 9fd995466027..f082646cd454 100644 --- a/packages/server-utils/src/ai/index.ts +++ b/packages/server-utils/src/ai/index.ts @@ -11,3 +11,5 @@ export { instrumentWorkersAiClient } from './workers-ai'; export { createLangChainCallbackHandler, instrumentLangChainEmbeddings } from './langchain'; export { instrumentStateGraph, instrumentStateGraphCompile, instrumentCreateReactAgent } from './langgraph'; export { SentryMastraExporter } from './mastra'; +export { createFlueInstrumentation } from './flue'; +export type { FlueOptions } from './flue'; diff --git a/packages/server-utils/src/index.ts b/packages/server-utils/src/index.ts index 2f918e2d5422..aceefeef5566 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 { flueIntegration } from './integrations/flue'; +export type { FlueOptions } from './integrations/flue'; export { mastraIntegration } from './integrations/mastra'; export { SentryMastraExporter } from './ai/mastra'; export { lruMemoizerIntegration } from './integrations/lru-memoizer'; diff --git a/packages/server-utils/src/integrations/flue.ts b/packages/server-utils/src/integrations/flue.ts new file mode 100644 index 000000000000..9d359667f450 --- /dev/null +++ b/packages/server-utils/src/integrations/flue.ts @@ -0,0 +1,78 @@ +import type { IntegrationFn } from '@sentry/core'; +import { debug, defineIntegration, GLOBAL_OBJ } from '@sentry/core'; +import { createFlueInstrumentation } from '../ai/flue'; +import { FLUE_INTEGRATION_NAME, FLUE_MODULE_NAME } from '../ai/flue/constants'; +import type { FlueOptions } from '../ai/flue'; +import type { FlueInstrumentation } from '../ai/flue/types'; +import { DEBUG_BUILD } from '../debug-build'; +import { flueModuleNames } from '../orchestrion/config/flue'; +import { invokeOrchestrionInstrumentation } from '../orchestrion/instrumentation'; +import { getOrchestrionModuleBindings } from '../utils/moduleInjected'; + +type FlueInstrumentFn = (instrumentation: FlueInstrumentation) => unknown; + +const _flueIntegration = ((options: FlueOptions = {}) => { + return { + name: FLUE_INTEGRATION_NAME, + setup(client) { + invokeOrchestrionInstrumentation(client, flueModuleNames, registerFlueInstrumentation, [options], { + // Nothing is bound to a tracing channel: the interceptor opens the agent span itself, so + // the async-context binding is not a precondition for registering. + requiresTracingChannelBinding: false, + }); + }, + }; +}) satisfies IntegrationFn; + +/** + * The two injection paths surface the app's `instrument` differently, and neither records what the + * other relies on. Under a bundler plugin the module is inlined, so the injected snippet hands its + * own binding out. Under the runtime hook there is no snippet, but the resolved file is recorded, + * and ESM keys its module registry by URL — so importing that URL yields the running namespace. + */ +function registerFlueInstrumentation(options: FlueOptions): void { + const bound = getOrchestrionModuleBindings(FLUE_MODULE_NAME)?.instrument as FlueInstrumentFn | undefined; + if (typeof bound === 'function') { + install(bound, options); + return; + } + + const url = GLOBAL_OBJ.__SENTRY_ORCHESTRION__?.runtimeFiles?.[FLUE_MODULE_NAME]; + if (!url) { + DEBUG_BUILD && + debug.log(`[${FLUE_INTEGRATION_NAME}] no \`instrument\` binding or resolved file, not instrumenting`); + return; + } + + import(url).then( + (mod: { instrument?: FlueInstrumentFn }) => { + if (mod.instrument) { + install(mod.instrument, options); + } + }, + (error: unknown) => { + DEBUG_BUILD && debug.log(`[${FLUE_INTEGRATION_NAME}] could not load the app's \`@flue/runtime\``, error); + }, + ); +} + +function install(instrument: FlueInstrumentFn, options: FlueOptions): void { + try { + instrument(createFlueInstrumentation(options)); + } catch (error) { + // Flue throws `InstrumentationAlreadyInstalledError` if something already registered under our + // key. The earlier registration is live, so log rather than surface it. + DEBUG_BUILD && debug.log(`[${FLUE_INTEGRATION_NAME}] \`instrument()\` rejected the registration`, error); + } +} + +/** + * Instruments the Flue agent framework (`@flue/runtime`), registering a Sentry observer through the + * runtime's own `instrument()` hook. Enabled by default. + * + * Requires the Sentry runtime hook or bundler plugin: both are what make the app's own + * `@flue/runtime` reachable, and neither path can instrument without one. + */ +export const flueIntegration = defineIntegration(_flueIntegration); + +export type { FlueOptions }; diff --git a/packages/server-utils/src/integrations/index.ts b/packages/server-utils/src/integrations/index.ts index 849a873cb1e3..8715f23e1958 100644 --- a/packages/server-utils/src/integrations/index.ts +++ b/packages/server-utils/src/integrations/index.ts @@ -13,6 +13,7 @@ import { mongooseIntegration } from './mongoose'; import { lruMemoizerIntegration } from './lru-memoizer'; import { langChainIntegration } from './langchain'; import { langGraphIntegration } from './langgraph'; +import { flueIntegration } from './flue'; import { mastraIntegration } from './mastra'; import { vercelAIIntegration } from './vercel-ai'; import { openAIIntegration } from './openai'; @@ -49,6 +50,7 @@ export function getTracingIntegrations(): Integration[] { langChainIntegration(), langGraphIntegration(), mastraIntegration(), + flueIntegration(), vercelAIIntegration(), openAIIntegration(), anthropicAIIntegration(), diff --git a/packages/server-utils/src/orchestrion/bundler/moduleInjectedTransform.ts b/packages/server-utils/src/orchestrion/bundler/moduleInjectedTransform.ts index 87a9fa6ddad6..1d9d5473f3e1 100644 --- a/packages/server-utils/src/orchestrion/bundler/moduleInjectedTransform.ts +++ b/packages/server-utils/src/orchestrion/bundler/moduleInjectedTransform.ts @@ -1,6 +1,6 @@ import type { CustomTransform } from '../apmTypes'; import { parse } from 'meriyah'; -import { subscriberExportForModule } from '../config/channel-integration-definitions'; +import { moduleBindingsForModule, subscriberExportForModule } from '../config/channel-integration-definitions'; import { MODULE_REGISTRATION_TRANSFORM } from '../config/registration-only'; // Tracks Program nodes we already injected into, so a package with several @@ -70,14 +70,22 @@ function moduleInjectedSnippet( exportName: string | undefined, esm: boolean, importSpecifier: string, + moduleBindings: readonly string[] | undefined, ): string { - const bindings = exportName ? `orchestrionModuleInjected, ${exportName}` : 'orchestrionModuleInjected'; + const imported = exportName ? `orchestrionModuleInjected, ${exportName}` : 'orchestrionModuleInjected'; const importStmt = esm - ? `import { ${bindings} } from ${JSON.stringify(importSpecifier)};` - : `const { ${bindings} } = require(${JSON.stringify(importSpecifier)});`; - - const args = exportName ? `${JSON.stringify(moduleName)}, ${exportName}` : JSON.stringify(moduleName); - return `${importStmt}\n${MODULE_INJECTED_SINK} = orchestrionModuleInjected(${args});`; + ? `import { ${imported} } from ${JSON.stringify(importSpecifier)};` + : `const { ${imported} } = require(${JSON.stringify(importSpecifier)});`; + + // A third argument passes named exports of THIS module back to its subscriber. The snippet is + // spliced into the module itself, so each name is a local binding here and is guaranteed to be + // the app's copy — the only copy whose module-scope state the running app reads. + const args = [ + JSON.stringify(moduleName), + ...(exportName ? [exportName] : []), + ...(moduleBindings?.length ? [`{ ${moduleBindings.join(', ')} }`] : []), + ]; + return `${importStmt}\n${MODULE_INJECTED_SINK} = orchestrionModuleInjected(${args.join(', ')});`; } /** @@ -127,10 +135,14 @@ export function moduleInjectedTransforms( const specifier = (typeof importSpecifier === 'function' ? importSpecifier() : importSpecifier) ?? DEFAULT_IMPORT_SPECIFIER; const exportName = subscriberExportForModule(moduleName); - const statements = parse(moduleInjectedSnippet(moduleName, exportName, moduleType === 'esm', specifier), { - module: moduleType === 'esm', - next: true, - }).body as ProgramNode['body']; + const moduleBindings = moduleBindingsForModule(moduleName); + const statements = parse( + moduleInjectedSnippet(moduleName, exportName, moduleType === 'esm', specifier, moduleBindings), + { + module: moduleType === 'esm', + next: true, + }, + ).body as ProgramNode['body']; const directiveIndex = node.body.findIndex(n => n.type === 'ExpressionStatement' && n.directive === 'use strict'); node.body.splice(directiveIndex + 1, 0, ...statements); diff --git a/packages/server-utils/src/orchestrion/config/channel-integration-definitions.ts b/packages/server-utils/src/orchestrion/config/channel-integration-definitions.ts index d293b9d22baf..bd0ea1e236fe 100644 --- a/packages/server-utils/src/orchestrion/config/channel-integration-definitions.ts +++ b/packages/server-utils/src/orchestrion/config/channel-integration-definitions.ts @@ -52,9 +52,28 @@ export const CHANNEL_INTEGRATION_DEFINITIONS = [ { exportName: 'kafkaIntegration', modules: ['kafkajs'] }, { exportName: 'redisIntegration', modules: ['redis', '@redis/client', 'ioredis'] }, { exportName: 'dataloaderIntegration', modules: ['dataloader'] }, -] as const satisfies ReadonlyArray<{ exportName: string; modules: readonly string[] }>; + // `bindings`: named exports of the instrumented module that the injected snippet hands back to + // the subscriber. Needed when a library is instrumented through a registration API it exposes + // (Flue's `instrument()`) rather than through channels at its call sites: that API closes over + // module-scope state, so it only works on the copy the app itself loaded. This covers the bundler + // path, where the module is inlined and no resolved file is recorded to import instead. + { exportName: 'flueIntegration', modules: ['@flue/runtime'], bindings: ['instrument'] }, +] as const satisfies ReadonlyArray<{ + exportName: string; + modules: readonly string[]; + bindings?: readonly string[]; +}>; /** Look up the subscriber export name for an instrumented package, if any. */ export function subscriberExportForModule(moduleName: string): string | undefined { return CHANNEL_INTEGRATION_DEFINITIONS.find(d => (d.modules as readonly string[]).includes(moduleName))?.exportName; } + +/** + * Named exports of the instrumented module that its injected snippet should pass back, if any. + * These are read from inside the module's own scope, so they are always the app's copy. + */ +export function moduleBindingsForModule(moduleName: string): readonly string[] | undefined { + const definition = CHANNEL_INTEGRATION_DEFINITIONS.find(d => (d.modules as readonly string[]).includes(moduleName)); + return definition && 'bindings' in definition ? definition.bindings : undefined; +} diff --git a/packages/server-utils/src/orchestrion/config/flue.ts b/packages/server-utils/src/orchestrion/config/flue.ts new file mode 100644 index 000000000000..20dfd0c9bdd9 --- /dev/null +++ b/packages/server-utils/src/orchestrion/config/flue.ts @@ -0,0 +1,30 @@ +import type { InstrumentationConfig } from '../apmTypes'; +import { FLUE_MODULE_NAME } from '../../ai/flue/constants'; +import { getModuleNames } from './module-names'; + +/** + * `@flue/runtime` publishes no diagnostics channels of its own, and is not instrumented at a call + * site: it exposes `instrument()`, a keyed registry taking `{ observe, interceptor }`. + * + * That registry is module-scope state, so it only takes effect on the copy of `@flue/runtime` the + * app actually loaded — registering into a second evaluated copy fails silently. `flueIntegration` + * reaches that copy differently per injection path: under a bundler plugin from the binding the + * injected snippet passes out, and under the runtime hook by importing the resolved URL the hook + * records, which ESM guarantees resolves to the same instance. (`createRequire`, how + * `mastraIntegration` reaches the app's copy, cannot resolve an ESM-only package whose `exports` + * map has no `require` condition.) + * + * `init` is wrapped only as the anchor that makes this file transform, which is what records that + * URL — nothing subscribes to the channel, and the spans come from `instrument()`. The transform + * must be a built-in one: a custom transform (as `registrationOnly` uses) cannot be registered on + * the runtime `--import` path, which would limit this to bundled apps. + */ +export const flueConfig: InstrumentationConfig[] = [ + { + channelName: 'flueInit', + module: { name: FLUE_MODULE_NAME, versionRange: '>=2.0.0 <3', filePath: 'dist/index.mjs' }, + functionQuery: { functionName: 'init', kind: 'Sync' }, + }, +]; + +export const flueModuleNames = getModuleNames(flueConfig); diff --git a/packages/server-utils/src/orchestrion/config/index.ts b/packages/server-utils/src/orchestrion/config/index.ts index 1fec4fb2c5ad..27f1766bfa23 100644 --- a/packages/server-utils/src/orchestrion/config/index.ts +++ b/packages/server-utils/src/orchestrion/config/index.ts @@ -8,6 +8,7 @@ import { anthropicAiConfig } from './anthropic-ai'; import { dataloaderConfig } from './dataloader'; import { expressConfig } from './express'; import { firebaseConfig } from './firebase'; +import { flueConfig } from './flue'; import { genericPoolConfig } from './generic-pool'; import { googleGenAiConfig } from './google-genai'; import { graphqlConfig } from './graphql'; @@ -55,6 +56,7 @@ export const SENTRY_INSTRUMENTATIONS: InstrumentationConfig[] = [ ...dataloaderConfig, ...expressConfig, ...firebaseConfig, + ...flueConfig, ...genericPoolConfig, ...googleGenAiConfig, ...graphqlConfig, diff --git a/packages/server-utils/src/utils/moduleInjected.ts b/packages/server-utils/src/utils/moduleInjected.ts index 599d6292d825..63a8bd48d34a 100644 --- a/packages/server-utils/src/utils/moduleInjected.ts +++ b/packages/server-utils/src/utils/moduleInjected.ts @@ -28,7 +28,11 @@ import { getClient, GLOBAL_OBJ } from '@sentry/core'; * that evaluate later (e.g. a lazily-required driver after a per-request * `init()` already snapshotted the marker). */ -export function orchestrionModuleInjected(moduleName: string, integrationFn?: () => Integration): void { +export function orchestrionModuleInjected( + moduleName: string, + integrationFn?: () => Integration, + moduleBindings?: Record, +): void { const marker = (GLOBAL_OBJ.__SENTRY_ORCHESTRION__ ??= {}); // Runtime guard, not just type narrowing: a banner from another SDK copy or @@ -41,5 +45,17 @@ export function orchestrionModuleInjected(moduleName: string, integrationFn?: () (marker.integrations ??= new Map()).set(moduleName, integrationFn); } + if (moduleBindings) { + (marker.moduleBindings ??= new Map()).set(moduleName, moduleBindings); + } + getClient()?.emit('orchestrion.module-injected', moduleName); } + +/** + * Named exports an instrumented module handed back through its injected snippet. Recorded from + * inside that module's own scope, so they always belong to the copy the app loaded. + */ +export function getOrchestrionModuleBindings(moduleName: string): Record | undefined { + return GLOBAL_OBJ.__SENTRY_ORCHESTRION__?.moduleBindings?.get(moduleName); +}