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 <noreply@anthropic.com>

Author: RulaKhaled

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;

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,17 +55,19 @@ 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);
     }
   };
 
   // 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<string, Span>();
-  const turnSpans = new Map<string, Span>();
-  const toolSpans = new Map<string, Span>();
+  // 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();
       }
 

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<string, Span>): 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<string, Span>;
+
+/**
+ * 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<strin
   );
 }
 
-export function endTurnSpan(observation: FlueObservation, turnSpans: Map<string, Span>, 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<string, Span>, 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<strin
   );
 }
 
-export function endToolSpan(observation: FlueObservation, toolSpans: Map<string, Span>, 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<string,
  * `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<string, Span>): void {
+export function recordRequestContent(observation: FlueObservation, turnSpans: SpanTracker): void {
   const { turnId } = observation;
   const span = turnId ? turnSpans.get(turnId) : undefined;
   const input = observation.request?.input;