feat(server-utils): Add first-party Flue instrumentation - #24265
feat(server-utils): Add first-party Flue instrumentation#24265RulaKhaled wants to merge 4 commits into
Conversation
size-limit report 📦
|
9fb3fcd to
d62f4f2
Compare
isaacs
left a comment
isaacs
left a comment
There was a problem hiding this comment.
I think this is on a good path, but there are still some gaps. It surfaced an interesting shortcoming of @apm-js-collab/tracing-hooks that could save around some lines on this side. apm-js-collab/tracing-hooks#52
| // 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) { |
There was a problem hiding this comment.
This approach is not parallel-safe.
agentDepth and agentSpan are single closure variables, and this treats any re-entry as a re-entry of the same submission. However, if an HTTP server or some other concurrency-heavy application had two parallel operations, those could clobber each other.
This can probably be fixed by using the operationId and submissionId to link the operations together. Keep agent spans in a Map keyed by operation.operationId, and resolve the parent in observe from observation.operationId. That also removes the depth counter and the re-entry special case entirely.
There was a problem hiding this comment.
Also, I'm not sure this is actually guaranteed to enter exactly twice? Looking at the flue code, the two dispatch sites I found are the coordinator and the session's runOperation (for operationKind of 'prompt' and 'skill').
runOperation is behind runExclusive, so it can't nest within one session. But subagent delegation runs in a separate session (DelegationDepthExceededError and defineSubagent are both exported), and a nested session's own runOperation('prompt') would be a third-level agent operation. Under the current code every subagent invocation is swallowed into the parent's single invoke_agent span.
I had the clanker clank up a test, written to packages/server-utils/test/ai/flue/nested-agent-operations.test.ts that seems to demonstrate this, if I'm understanding the behavior here properly: https://gist.github.com/isaacs/8f703ebae6adc26696b54c5d4f76b50e
There was a problem hiding this comment.
Good catch, confirmed and fixed, i added a test for this. One detail: the submission opens with a wrapper operation whose operationId is the submissionId, so that one is skipped or every invocation double-counts
| } | ||
| }, | ||
| ); | ||
| }, |
There was a problem hiding this comment.
FlueExecutionContext.traceCarrier is { traceparent, tracestate }, populated by extractTraceCarrier from the incoming request headers and passed on the outer agent operation.
But, the code here never reads it, so it seems like dispatched submissions will start orphan traces? For a durable or dispatched submission the coordinator runs the work later, possibly in another isolate, Durable Object, or process. Without traceCarrier, the invoke_agent span starts a brand new trace with no link to the request that enqueued it.
We could do this with a small helper, like:
function sentryTraceFromTraceparent(traceparent: string): string | undefined {
const [version, traceId, spanId, flags] = traceparent.split('-');
if (version !== '00' || !traceId || !spanId || !flags) {
return undefined;
}
return `${traceId}-${spanId}-${parseInt(flags, 16) & 0x01 ? '1' : '0'}`;
}(This could perhaps be reasonable to put in @sentry/core somewhere, near generateTraceparentHeader? I didn't see any w3c traceparent parser there already, and it's tiny, so we could also wait until there's a second use for it before abstracting.)
And then apply it only when nothing already continued the trace:
const sentryTrace = ctx.traceCarrier?.traceparent
? sentryTraceFromTraceparent(ctx.traceCarrier.traceparent)
: undefined;
return sentryTrace && !getActiveSpan()
? continueTrace({ sentryTrace, baggage: undefined }, openAgentSpan)
: openAgentSpan();There was a problem hiding this comment.
implemented your helper and the !getActiveSpan() guard as written, plus three tests. it rarely fires though 😅 extractTraceCarrier reads only W3C traceparent and our propagateTraceparent is @default false, so sentry-to-sentry the carrier is undefined. flue says dispatch(...) doesn't propagate trace context at all. and on cloudflare the alarm invocation has an active span (invoke_agent's parent is the alarm span) so the guard skips it there (correctly i guess? since we set startNewTrace: true for alarms on purpose)
digging into it turned up something separate, storeSpanContext is keyed by method name and only runs for startNewTrace methods, so the chain is alarm → previous alarm and the request that scheduled it is never linked. that's a general DO gap rather than a flue one, filing it separately.
| // 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); |
There was a problem hiding this comment.
This calls _INTERNAL_skipAiProviderWrapping once, when createFlueInstrumentation is constructed. On Cloudflare the documented usage is a manual call in module or Durable Object scope, so it runs once per isolate.
But, packages/cloudflare/src/client.ts line 188 calls _INTERNAL_clearAiProviderSkips() in _setupIntegrations(), and its comment states that Cloudflare calls init() per request. So the skip registered at isolate load is wiped by the first init() and never re-registered. Every request after the first in that isolate would double report gen_ai.chat for any provider client that is channel-instrumented, which is what the skip is there to prevent.
I think the skip needs to be re-applied per client, not once per instrumentation object. Registering it from an integration setup(client) would do that.
Also, the skip is a side effect of building the object, but the object is only useful once instrument() accepts it.
packages/server-utils/src/integrations/flue.ts lines 59-67 catch any throw from instrument() and log it.
So on the failure path the SDK has suppressed the other AI instrumentations and installed nothing in its place: the user gets no gen_ai.chat spans at all, only a debug log. Reading Flue's instrument function shows it throws InstrumentationAlreadyInstalledError when the key is taken and isDevMode() is false, and can also rethrow from registerExecutionInterceptor.
Recommendation: Move the skip to after a successful instrument() call, or restore the previous state on failure.
There was a problem hiding this comment.
I dropped the integration for manual registration, so instead the skip is applied on first observe/interceptor call and re-applied if the registry has been cleared, that covers your second point too: constructing the object no longer suppresses anything, so a rejected instrument() can't leave the app with providers off and nothing installed
| // per-Durable-Object isolates — gets it too. | ||
| _INTERNAL_skipAiProviderWrapping(SKIPPED_PROVIDERS); | ||
|
|
||
| const { recordInputs, recordOutputs } = resolveAIRecordingOptions(options); |
There was a problem hiding this comment.
Every other AI integration resolves these values at span time (see packages/server-utils/src/integrations/openai.ts line 82 and 124), because resolveAIRecordingOptions reads getClient()?.getDataCollectionOptions() (packages/server-utils/src/ai/core/utils.ts line 72).
On Cloudflare the client is replaced per request, so the values captured at isolate load are the wrong ones for every later request. On Node it happens to work because the client is stable. Resolving lazily would remove the divergence.
| } | ||
|
|
||
| const open = (): Span => | ||
| startInactiveSpan({ |
There was a problem hiding this comment.
startToolSpan uses startInactiveSpan and parents off the agent span. Anything the tool does (database query, HTTP call, a nested call etc), then lands under invoke_agent as a sibling of execute_tool rather than inside it.
The interceptor already receives type: 'tool' with toolCallId and toolName, and type: 'model' with turnId. Wrapping next() in withActiveSpan(existingSpan, next) for those operation types would fix the nesting without creating extra spans, and would reuse the span the observe path already opened.
The comment at packages/server-utils/src/ai/flue/index.ts lines 173-174 says the
active span during observe "is whatever the provider SDK last opened". With the providers skipped, that is worth re-checking; if it no longer holds, the whole agentSpan plumbing could be replaced by the current active span.
There was a problem hiding this comment.
Very nice catch!
There was a problem hiding this comment.
- I fixed the tool spans structure (tool work lands beside execute_tool instead of inside it)
- took your fix: withActiveSpan(existingSpan, next) in the interceptor for tool and model operations, reusing the spans observe already opened, no extra spans. db.query now nests under execute_tool, and the model half also fixes http.client sitting beside gen_ai.chat.
- explicit parenting is gone (the agentSpans map stays though, text_delta and message_start also carry conversationId and fire during a model op where the active span is chat, so using the active span for attribution would stamp it on the wrong span. parenting implicit, attribution still explicit)
| span.setAttribute(GEN_AI_RESPONSE_MODEL, responseModel); | ||
| } | ||
|
|
||
| const provider = observation.request?.providerId ?? observation.request?.providerName; |
There was a problem hiding this comment.
Both fields are required strings on ModelRequestInfo, so the fallback seems like it's not doing any work? I think just providerId is the correct value here.
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 <noreply@anthropic.com>
d62f4f2 to
a0e8143
Compare
- 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 <noreply@anthropic.com>
thanks for reviewing Isaac! this was still a WIP when you looked at it, hence the draft. i discussed it with Francesco and we landed on manual instrumentation rather than the module bindings, so anything related to that is moot now, i closed comments related to that change — the four shared-infra files are back to develop untouched and the orchestrion config and integration are gone. users call instrument(Sentry.createFlueInstrumentation()) themselves, same shape as eveConversationHook() and flue's own documented pattern. |
… 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>
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 <noreply@anthropic.com>
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Want reviews to match your repository better? Bugbot Learning can learn team-specific rules from PR activity. A team admin can enable Learning in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit 0c24d2b. Configure here.
| } | ||
| if (finishReason) { | ||
| span.setAttribute(GEN_AI_RESPONSE_FINISH_REASONS, [finishReason]); | ||
| } |
There was a problem hiding this comment.
Finish reasons stored as array
Medium Severity
gen_ai.response.finish_reasons is set as a raw string array. Every other AI integration in this package stores a JSON string, and existing integration tests assert a string value like ["stop"]. The AI Agents view and any consumer that parses this field as a string will not read Flue finish reasons correctly.
Reviewed by Cursor Bugbot for commit 0c24d2b. Configure here.
2 participants
Instruments the Flue agent framework (
@flue/runtime) through its owninstrument()hook, producing theinvoke_agent→chat/execute_toolhierarchy with token usage, Flue-computed cost and message content.Verified against a scaffolded
flue initapp driven over HTTP against a real provider, on Node and Cloudflare:Root cause of the shape: Flue is not instrumented at a call site — it exposes
instrument(), a registration API whose registry is module-scope state. An auto-registering integration would need a reference to that module's own binding, and no channel payload carries one (instrumentappears in Flue's build only as the function definition and in itsexport {}list, never as a property, argument or return value). Registration is therefore left to the user, which is also Flue's documented pattern for observability providers, and 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;
observeowns the turn and tool spans, becauseturn_start/turnare the only signal one-to-one with a model call andturnis what carries usage. Spanning themodeloperation instead does not work — 12 fire per turn, and the first resolves long before usage is known.Flue reaches providers through
@earendil-works/pi-ai, which bundles theopenai,@anthropic-ai/sdkand@google/genaiclients, so those are skipped while Flue is instrumented; the skip lives increateFlueInstrumentationso it applies however the instrumentation is registered.Turn and tool spans are tracked in
LRUMaps rather than plain maps. Both are keyed off an id that only the matching end observation removes, and a stream abandoned mid-turn never emits one, so an uncapped map would grow for the lifetime of the process. Eviction ends the span it drops instead of letting it disappear unsent, which is the same trade Mastra's exporter makes.On the Sentry bundler plugin: not required, and not recommended for Flue's sake. Span trees are identical with and without it on both targets — HTTP spans come from Node's native
diagnostics_channel, and the provider skip fires either way. It is worth adding only if the app also uses libraries that need orchestrion (pg, redis, kafka…); for a pure Flue app it force-bundles@flue/runtime(55KB → 4.3MB measured) for no telemetry gain.The export is added to every runtime that re-exports
@sentry/node.astroandelysianeed it named explicitly because they keep hand-maintained export lists (Vite puts a wildcard re-export underdefaultin Astro prod builds);nextjs,remixandsveltekituse a realexport *and pick it up on their own.Tests are stacked in #24266.
Known gaps: Flue ships its own Sentry blueprint (
flue add tooling sentry) targeting@sentry/node@^10.64.0, which our docs currently point at — needs a docs update, same shape as the@mastra/sentrymigration.Fixes #24017
🤖 Generated with Claude Code