From 754c9f2d0f9c3c6e61de0e2be327e574158b88c7 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Fri, 11 Sep 2026 08:12:06 +0200 Subject: [PATCH 01/12] test(server-utils): Cover the Flue instrumentation Unit tests over `createFlueInstrumentation` for the span shapes, the conversation id lifted off the re-entered agent operation, the usage/cost mapping, the all-zero-usage guard on failed turns, tool spans, content recording and its `recordInputs`/`recordOutputs` gating, and dispose. The integration test drives a real agent through a tool call using `pi-ai`'s `faux` provider, so the run is deterministic and needs no provider key or mock server. ESM only: `@flue/runtime` has no `require` export condition, and it is installed per-suite because its `engines.node >= 22.19` would break `yarn install` on the Node 20 CI matrix. Co-Authored-By: Claude Opus 5 --- .../suites/tracing/flue/instrument.mjs | 11 + .../suites/tracing/flue/scenario.mjs | 38 +++ .../suites/tracing/flue/test.ts | 86 ++++++ .../test/ai/lib/tracing/flue.test.ts | 272 ++++++++++++++++++ 4 files changed, 407 insertions(+) create mode 100644 dev-packages/node-integration-tests/suites/tracing/flue/instrument.mjs create mode 100644 dev-packages/node-integration-tests/suites/tracing/flue/scenario.mjs create mode 100644 dev-packages/node-integration-tests/suites/tracing/flue/test.ts create mode 100644 packages/server-utils/test/ai/lib/tracing/flue.test.ts diff --git a/dev-packages/node-integration-tests/suites/tracing/flue/instrument.mjs b/dev-packages/node-integration-tests/suites/tracing/flue/instrument.mjs new file mode 100644 index 000000000000..42052d281304 --- /dev/null +++ b/dev-packages/node-integration-tests/suites/tracing/flue/instrument.mjs @@ -0,0 +1,11 @@ +import * as Sentry from '@sentry/node'; +import { loggingTransport } from '@sentry-internal/node-integration-tests'; + +Sentry.init({ + traceLifecycle: 'static', + dsn: 'https://public@dsn.ingest.sentry.io/1337', + release: '1.0', + tracesSampleRate: 1.0, + dataCollection: { genAI: { inputs: false, outputs: false } }, + transport: loggingTransport, +}); diff --git a/dev-packages/node-integration-tests/suites/tracing/flue/scenario.mjs b/dev-packages/node-integration-tests/suites/tracing/flue/scenario.mjs new file mode 100644 index 000000000000..625abf964ee0 --- /dev/null +++ b/dev-packages/node-integration-tests/suites/tracing/flue/scenario.mjs @@ -0,0 +1,38 @@ +import * as Sentry from '@sentry/node'; +import { __flueBindAgentModule, init, instrument, useModel, useTool } from '@flue/runtime'; +import { start } from '@flue/runtime/node'; +import { fauxAssistantMessage, fauxProvider, fauxToolCall } from '@earendil-works/pi-ai/providers/faux'; + +// `pi-ai`'s faux provider scripts model responses in-process, so the run is deterministic and needs +// no provider key or mock server. Two steps: a tool call, then the final answer. +instrument(Sentry.createFlueInstrumentation()); + +const faux = fauxProvider({ + provider: 'faux', + models: [{ id: 'faux-model', cost: { input: 1, output: 2, cacheRead: 0, cacheWrite: 0 } }], +}); +faux.setResponses([ + fauxAssistantMessage(fauxToolCall('get_weather', { city: 'Berlin' }, { id: 'call_1' }), { stopReason: 'toolUse' }), + fauxAssistantMessage('It is 21 degrees and sunny in Berlin.'), +]); + +function Hello() { + useModel('faux/faux-model'); + useTool({ + name: 'get_weather', + description: 'Get the current weather for a city.', + run: ({ city }) => `It is 21 degrees and sunny in ${city}.`, + }); + return 'You are a helpful assistant.'; +} +__flueBindAgentModule(Hello, { identity: 'Hello' }); + +await Sentry.startSpan({ name: 'flue-test', op: 'function' }, async () => { + const flue = await start({ agents: [Hello], providers: [faux.provider] }); + const agent = init(Hello, { id: 'e2e' }); + const receipt = await agent.dispatch('What is the weather in Berlin?'); + await agent.read(receipt); + await flue[Symbol.asyncDispose]?.(); +}); + +await Sentry.flush(2000); diff --git a/dev-packages/node-integration-tests/suites/tracing/flue/test.ts b/dev-packages/node-integration-tests/suites/tracing/flue/test.ts new file mode 100644 index 000000000000..ec7a26f71ad1 --- /dev/null +++ b/dev-packages/node-integration-tests/suites/tracing/flue/test.ts @@ -0,0 +1,86 @@ +import { + GEN_AI_AGENT_NAME, + GEN_AI_CONVERSATION_ID, + GEN_AI_COST_TOTAL_TOKENS, + GEN_AI_OPERATION_NAME, + GEN_AI_TOOL_NAME, + GEN_AI_USAGE_INPUT_TOKENS, + GEN_AI_USAGE_OUTPUT_TOKENS, + GEN_AI_USAGE_TOTAL_TOKENS, +} from '@sentry/conventions/attributes'; +import { afterAll, expect } from 'vitest'; +import { conditionalTest } from '../../../utils'; +import { cleanupChildProcesses, createEsmAndCjsTests } from '../../../utils/runner'; + +// `@flue/runtime` declares `engines.node >= 22.19`, so it can't live in the package's root +// `devDependencies` (that would break `yarn install` on the 20.19 CI matrix). Install it per-suite +// instead, guarded by the `min: 22` skip below. +const FLUE_DEPENDENCIES = { + additionalDependencies: { + '@flue/runtime': '2.0.3', + '@earendil-works/pi-ai': '0.85.1', + }, +}; + +conditionalTest({ min: 22 })('Flue integration', () => { + afterAll(() => { + cleanupChildProcesses(); + }); + + createEsmAndCjsTests( + __dirname, + 'scenario.mjs', + 'instrument.mjs', + (createRunner, test, mode) => { + // `@flue/runtime` is ESM-only — its `exports` map has no `require` condition, so there is no + // CJS variant of this scenario to run. + if (mode === 'cjs') { + return; + } + + test('creates the invoke_agent / chat / execute_tool hierarchy', async () => { + await createRunner() + .expect({ transaction: { transaction: 'flue-test' } }) + .expect({ + span: container => { + const spans = container.items; + const names = spans.map(span => span.name); + + expect(names).toContain('invoke_agent Hello'); + expect(names).toContain('execute_tool get_weather'); + expect(names.filter(name => name?.startsWith('chat'))).toHaveLength(2); + + const agent = spans.find(span => span.name === 'invoke_agent Hello')!; + expect(agent.attributes['sentry.op'].value).toBe('gen_ai.invoke_agent'); + expect(agent.attributes['sentry.origin'].value).toBe('auto.ai.flue'); + expect(agent.attributes[GEN_AI_OPERATION_NAME].value).toBe('invoke_agent'); + expect(agent.attributes[GEN_AI_AGENT_NAME].value).toBe('Hello'); + expect(agent.attributes[GEN_AI_CONVERSATION_ID].value).toEqual(expect.any(String)); + + const chat = spans.find(span => span.name?.startsWith('chat'))!; + expect(chat.attributes['sentry.op'].value).toBe('gen_ai.chat'); + expect(chat.attributes['sentry.origin'].value).toBe('auto.ai.flue'); + expect(chat.attributes[GEN_AI_USAGE_INPUT_TOKENS].value).toEqual(expect.any(Number)); + expect(chat.attributes[GEN_AI_USAGE_OUTPUT_TOKENS].value).toEqual(expect.any(Number)); + expect(chat.attributes[GEN_AI_USAGE_TOTAL_TOKENS].value).toEqual(expect.any(Number)); + // Flue computes cost itself; no provider SDK reports it. + expect(chat.attributes[GEN_AI_COST_TOTAL_TOKENS].value).toEqual(expect.any(Number)); + + const tool = spans.find(span => span.name === 'execute_tool get_weather')!; + expect(tool.attributes['sentry.op'].value).toBe('gen_ai.execute_tool'); + expect(tool.attributes['sentry.origin'].value).toBe('auto.ai.flue'); + expect(tool.attributes[GEN_AI_TOOL_NAME].value).toBe('get_weather'); + + // Tool spans are siblings of `chat` under the agent invocation, matching how Flue's + // own OpenTelemetry adapter projects them. + expect(tool.parent_span_id).toBe(agent.span_id); + expect(chat.parent_span_id).toBe(agent.span_id); + }, + }) + .start() + .completed(); + }); + }, + FLUE_DEPENDENCIES, + ); +}); diff --git a/packages/server-utils/test/ai/lib/tracing/flue.test.ts b/packages/server-utils/test/ai/lib/tracing/flue.test.ts new file mode 100644 index 000000000000..97485285282e --- /dev/null +++ b/packages/server-utils/test/ai/lib/tracing/flue.test.ts @@ -0,0 +1,272 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import type { Span } from '@sentry/core'; +import { + _INTERNAL_clearAiProviderSkips, + _INTERNAL_shouldSkipAiProviderWrapping, + getMainCarrier, + setCurrentClient, + spanToStaticSpanJSON, +} from '@sentry/core'; +import { ANTHROPIC_AI_INTEGRATION_NAME } from '../../../../src/ai/anthropic-ai/constants'; +import { createFlueInstrumentation } from '../../../../src/ai/flue'; +import type { FlueInstrumentation, FlueObservation } from '../../../../src/ai/flue/types'; +import { OPENAI_INTEGRATION_NAME } from '../../../../src/ai/openai/constants'; +import { getDefaultTestClientOptions, TestClient } from '../../../mocks/client'; + +const AGENT_CTX = { agentName: 'Hello' }; +const INNER_CTX = { conversationId: 'conv_1' }; + +/** A settled turn as Flue reports it, with the field names `ModelRequestInfo`/`ModelResponse` use. */ +function turn(overrides: Partial = {}): FlueObservation { + return { + type: 'turn', + turnId: 'turn_1', + request: { requestedModel: 'claude-haiku-4.5', providerId: 'anthropic' }, + response: { + responseId: 'resp_1', + finishReason: 'stop', + usage: { + input: 924, + output: 57, + totalTokens: 981, + cacheRead: 0, + cacheWrite: 0, + cost: { input: 0.000924, output: 0.000275, total: 0.001199, cacheRead: 0, cacheWrite: 0 }, + }, + }, + ...overrides, + }; +} + +describe('createFlueInstrumentation', () => { + let endedSpans: Span[]; + let instrumentation: FlueInstrumentation; + + beforeEach(() => { + _INTERNAL_clearAiProviderSkips(); + getMainCarrier().__SENTRY__ = undefined; + const client = new TestClient( + getDefaultTestClientOptions({ + dsn: 'https://public@dsn.ingest.sentry.io/1337', + tracesSampleRate: 1, + traceLifecycle: 'stream', + }), + ); + setCurrentClient(client); + client.init(); + + endedSpans = []; + client.on('spanEnd', span => endedSpans.push(span)); + instrumentation = createFlueInstrumentation(); + }); + + afterEach(() => { + _INTERNAL_clearAiProviderSkips(); + getMainCarrier().__SENTRY__ = undefined; + }); + + /** Run `fn` inside an agent operation, the way Flue's interceptor would. */ + function withAgent(fn: () => Promise | T): Promise { + return instrumentation.interceptor({ type: 'agent' }, AGENT_CTX, async () => fn()); + } + + function findSpan(description: string): ReturnType | undefined { + return endedSpans.map(span => spanToStaticSpanJSON(span)).find(json => json.description === description); + } + + // Flue drives the providers through `pi-ai`, which bundles the `openai` / `@anthropic-ai/sdk` / + // `@google/genai` clients those integrations patch, so their spans duplicate the turn span. + it('skips raw provider wrapping as soon as the instrumentation is built', () => { + expect(_INTERNAL_shouldSkipAiProviderWrapping(OPENAI_INTEGRATION_NAME)).toBe(true); + expect(_INTERNAL_shouldSkipAiProviderWrapping(ANTHROPIC_AI_INTEGRATION_NAME)).toBe(true); + }); + + it('names agent spans `invoke_agent {name}` and sets the gen_ai op', async () => { + await withAgent(() => undefined); + + const json = findSpan('invoke_agent Hello'); + expect(json?.data['sentry.op']).toBe('gen_ai.invoke_agent'); + expect(json?.data['sentry.origin']).toBe('auto.ai.flue'); + expect(json?.data['gen_ai.operation.name']).toBe('invoke_agent'); + expect(json?.data['gen_ai.agent.name']).toBe('Hello'); + }); + + // The agent operation re-enters once, and only the inner context names the conversation. + it('lifts the conversation id off the re-entered agent operation', async () => { + await withAgent(() => instrumentation.interceptor({ type: 'agent' }, INNER_CTX, async () => undefined)); + + expect(findSpan('invoke_agent Hello')?.data['gen_ai.conversation.id']).toBe('conv_1'); + }); + + it('does not span operations other than `agent`', async () => { + await instrumentation.interceptor({ type: 'model', turnId: 'turn_1' }, {}, async () => undefined); + + expect(endedSpans).toHaveLength(0); + }); + + it('opens a chat span on turn_start and completes it from the settled turn', async () => { + await withAgent(() => { + instrumentation.observe({ type: 'turn_start', turnId: 'turn_1', conversationId: 'conv_1' }, {}); + instrumentation.observe(turn(), {}); + }); + + const json = findSpan('chat claude-haiku-4.5'); + expect(json?.data['sentry.op']).toBe('gen_ai.chat'); + expect(json?.data['sentry.origin']).toBe('auto.ai.flue'); + expect(json?.data['gen_ai.request.model']).toBe('claude-haiku-4.5'); + expect(json?.data['gen_ai.provider.name']).toBe('anthropic'); + expect(json?.data['gen_ai.response.id']).toBe('resp_1'); + expect(json?.data['gen_ai.response.finish_reasons']).toEqual(['stop']); + expect(json?.data['gen_ai.conversation.id']).toBe('conv_1'); + }); + + // Flue computes costs itself; the provider SDKs report none. + it('records token usage and Flue-computed cost on the chat span', async () => { + await withAgent(() => { + instrumentation.observe({ type: 'turn_start', turnId: 'turn_1' }, {}); + instrumentation.observe(turn(), {}); + }); + + const json = findSpan('chat claude-haiku-4.5'); + expect(json?.data['gen_ai.usage.input_tokens']).toBe(924); + expect(json?.data['gen_ai.usage.output_tokens']).toBe(57); + expect(json?.data['gen_ai.usage.total_tokens']).toBe(981); + expect(json?.data['gen_ai.cost.total_tokens']).toBe(0.001199); + }); + + it('marks a failed turn as errored', async () => { + await withAgent(() => { + instrumentation.observe({ type: 'turn_start', turnId: 'turn_1' }, {}); + instrumentation.observe(turn({ isError: true }), {}); + }); + + expect(findSpan('chat claude-haiku-4.5')?.status).toBe('internal_error'); + }); + + // A turn that fails before the provider bills anything reports every counter as 0; writing those + // reads as a real zero-cost call. + it('omits usage entirely when a failed turn produced no tokens', async () => { + const empty = { + input: 0, + output: 0, + totalTokens: 0, + cacheRead: 0, + cacheWrite: 0, + cost: { input: 0, output: 0, total: 0, cacheRead: 0, cacheWrite: 0 }, + }; + + await withAgent(() => { + instrumentation.observe({ type: 'turn_start', turnId: 'turn_1' }, {}); + instrumentation.observe(turn({ isError: true, response: { usage: empty } }), {}); + }); + + const json = findSpan('chat claude-haiku-4.5'); + expect(json).toBeDefined(); + expect(json?.data['gen_ai.usage.total_tokens']).toBeUndefined(); + expect(json?.data['gen_ai.cost.total_tokens']).toBeUndefined(); + }); + + describe('content recording', () => { + const requestContent = { + type: 'turn_request', + turnId: 'turn_1', + request: { + requestedModel: 'claude-haiku-4.5', + input: { + systemPrompt: 'You are helpful.', + messages: [{ role: 'user', content: 'hi' }], + tools: [{ name: 'get_weather', description: 'weather', parameters: {} }], + }, + }, + } satisfies FlueObservation; + + async function record(instr: FlueInstrumentation): Promise { + await instr.interceptor({ type: 'agent' }, AGENT_CTX, async () => { + instr.observe({ type: 'turn_start', turnId: 'turn_1' }, {}); + instr.observe(requestContent, {}); + instr.observe(turn({ response: { ...turn().response, output: { role: 'assistant' } } }), {}); + instr.observe({ type: 'tool_start', toolCallId: 'c1', toolName: 'get_weather', args: { city: 'Berlin' } }, {}); + instr.observe({ type: 'tool', toolCallId: 'c1', toolName: 'get_weather', result: 'sunny' }, {}); + }); + } + + it('records messages, instructions, tool definitions, arguments and results by default', async () => { + await record(instrumentation); + + const chat = findSpan('chat claude-haiku-4.5'); + expect(chat?.data['gen_ai.system_instructions']).toBe('You are helpful.'); + expect(chat?.data['gen_ai.input.messages']).toContain('"role":"user"'); + expect(chat?.data['gen_ai.output.messages']).toContain('"role":"assistant"'); + expect(chat?.data['gen_ai.tool.definitions']).toContain('get_weather'); + + const tool = findSpan('execute_tool get_weather'); + expect(tool?.data['gen_ai.tool.call.arguments']).toBe('{"city":"Berlin"}'); + expect(tool?.data['gen_ai.tool.call.result']).toBe('sunny'); + }); + + it('omits inputs when recordInputs is false but keeps outputs', async () => { + await record(createFlueInstrumentation({ recordInputs: false })); + + const chat = findSpan('chat claude-haiku-4.5'); + expect(chat?.data['gen_ai.input.messages']).toBeUndefined(); + expect(chat?.data['gen_ai.system_instructions']).toBeUndefined(); + expect(chat?.data['gen_ai.tool.definitions']).toBeUndefined(); + expect(chat?.data['gen_ai.output.messages']).toBeDefined(); + expect(findSpan('execute_tool get_weather')?.data['gen_ai.tool.call.arguments']).toBeUndefined(); + }); + + it('omits outputs when recordOutputs is false but keeps inputs', async () => { + await record(createFlueInstrumentation({ recordOutputs: false })); + + const chat = findSpan('chat claude-haiku-4.5'); + expect(chat?.data['gen_ai.output.messages']).toBeUndefined(); + expect(chat?.data['gen_ai.input.messages']).toBeDefined(); + expect(findSpan('execute_tool get_weather')?.data['gen_ai.tool.call.result']).toBeUndefined(); + }); + }); + + it('emits execute_tool spans keyed by tool call id', async () => { + await withAgent(() => { + instrumentation.observe({ type: 'tool_start', toolCallId: 'call_1', toolName: 'get_weather' }, {}); + instrumentation.observe({ type: 'tool', toolCallId: 'call_1', toolName: 'get_weather' }, {}); + }); + + const json = findSpan('execute_tool get_weather'); + expect(json?.data['sentry.op']).toBe('gen_ai.execute_tool'); + expect(json?.data['sentry.origin']).toBe('auto.ai.flue'); + expect(json?.data['gen_ai.operation.name']).toBe('execute_tool'); + expect(json?.data['gen_ai.tool.name']).toBe('get_weather'); + }); + + it('marks a failed tool call as errored', async () => { + await withAgent(() => { + instrumentation.observe({ type: 'tool_start', toolCallId: 'call_1', toolName: 'boom' }, {}); + instrumentation.observe({ type: 'tool', toolCallId: 'call_1', toolName: 'boom', isError: true }, {}); + }); + + expect(findSpan('execute_tool boom')?.status).toBe('internal_error'); + }); + + it('ignores a settled turn or tool it never opened a span for', async () => { + await withAgent(() => { + instrumentation.observe(turn({ turnId: 'never_started' }), {}); + instrumentation.observe({ type: 'tool', toolCallId: 'never_started', toolName: 'x' }, {}); + }); + + expect(endedSpans.map(span => spanToStaticSpanJSON(span).description)).toEqual(['invoke_agent Hello']); + }); + + it('ends spans still open at dispose', async () => { + await withAgent(() => { + instrumentation.observe({ type: 'turn_start', turnId: 'turn_1' }, {}); + instrumentation.observe({ type: 'tool_start', toolCallId: 'call_1', toolName: 'get_weather' }, {}); + }); + // Never settled, so the span keeps the unqualified name it opened with. + expect(findSpan('chat')).toBeUndefined(); + + instrumentation.dispose(); + + expect(findSpan('chat')).toBeDefined(); + expect(findSpan('execute_tool get_weather')).toBeDefined(); + }); +}); From 840aca8e6d503dd75973b33248557431e4c79bbe Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Fri, 11 Sep 2026 10:19:27 +0200 Subject: [PATCH 02/12] test(server-utils): Cover the Flue review fixes Concurrent agent runs, subagent delegation (adopted from Isaac's repro, moved into the suite and asserted through a helper so a span-order assumption cannot creep back), the agent name arriving via the observations, trace continuation from the replayed traceparent, the provider skip applying on first use and re-applying after a registry reset, the recording options following the current client, and the conventional request attributes. Co-Authored-By: Claude Opus 5 --- .../test/ai/lib/tracing/flue.test.ts | 276 ++++++++++++++++-- 1 file changed, 255 insertions(+), 21 deletions(-) diff --git a/packages/server-utils/test/ai/lib/tracing/flue.test.ts b/packages/server-utils/test/ai/lib/tracing/flue.test.ts index 97485285282e..066d94b16f76 100644 --- a/packages/server-utils/test/ai/lib/tracing/flue.test.ts +++ b/packages/server-utils/test/ai/lib/tracing/flue.test.ts @@ -6,6 +6,7 @@ import { getMainCarrier, setCurrentClient, spanToStaticSpanJSON, + startSpan, } from '@sentry/core'; import { ANTHROPIC_AI_INTEGRATION_NAME } from '../../../../src/ai/anthropic-ai/constants'; import { createFlueInstrumentation } from '../../../../src/ai/flue'; @@ -13,7 +14,8 @@ import type { FlueInstrumentation, FlueObservation } from '../../../../src/ai/fl import { OPENAI_INTEGRATION_NAME } from '../../../../src/ai/openai/constants'; import { getDefaultTestClientOptions, TestClient } from '../../../mocks/client'; -const AGENT_CTX = { agentName: 'Hello' }; +const AGENT_OP = { type: 'agent', operationId: 'op_1' }; +const AGENT_CTX = { agentName: 'Hello', submissionId: 'sub_1' }; const INNER_CTX = { conversationId: 'conv_1' }; /** A settled turn as Flue reports it, with the field names `ModelRequestInfo`/`ModelResponse` use. */ @@ -21,7 +23,17 @@ function turn(overrides: Partial = {}): FlueObservation { return { type: 'turn', turnId: 'turn_1', - request: { requestedModel: 'claude-haiku-4.5', providerId: 'anthropic' }, + submissionId: 'sub_1', + operationId: 'op_1', + request: { + requestedModel: 'claude-haiku-4.5', + providerId: 'anthropic', + temperature: 0.7, + maxTokens: 1024, + reasoningLevel: 'high', + serverAddress: 'api.anthropic.com', + serverPort: 443, + }, response: { responseId: 'resp_1', finishReason: 'stop', @@ -67,7 +79,13 @@ describe('createFlueInstrumentation', () => { /** Run `fn` inside an agent operation, the way Flue's interceptor would. */ function withAgent(fn: () => Promise | T): Promise { - return instrumentation.interceptor({ type: 'agent' }, AGENT_CTX, async () => fn()); + return instrumentation.interceptor(AGENT_OP, AGENT_CTX, async () => fn()); + } + + function agentSpans(): ReturnType[] { + return endedSpans + .map(span => spanToStaticSpanJSON(span)) + .filter(json => json.data['sentry.op'] === 'gen_ai.invoke_agent'); } function findSpan(description: string): ReturnType | undefined { @@ -76,11 +94,30 @@ describe('createFlueInstrumentation', () => { // Flue drives the providers through `pi-ai`, which bundles the `openai` / `@anthropic-ai/sdk` / // `@google/genai` clients those integrations patch, so their spans duplicate the turn span. - it('skips raw provider wrapping as soon as the instrumentation is built', () => { + // Not at construction: if `instrument()` rejects the object, suppressing the provider + // integrations would leave the app with no `gen_ai.chat` spans at all. + it('skips raw provider wrapping on first use, not on construction', async () => { + expect(_INTERNAL_shouldSkipAiProviderWrapping(OPENAI_INTEGRATION_NAME)).toBe(false); + + await withAgent(() => undefined); + expect(_INTERNAL_shouldSkipAiProviderWrapping(OPENAI_INTEGRATION_NAME)).toBe(true); expect(_INTERNAL_shouldSkipAiProviderWrapping(ANTHROPIC_AI_INTEGRATION_NAME)).toBe(true); }); + // The registry is reset per client (`_setupIntegrations` clears it, and Cloudflare calls `init()` + // per request), so a one-shot call at construction is wiped by the next reset. + it('re-applies the provider skip after the registry is cleared', async () => { + await withAgent(() => undefined); + expect(_INTERNAL_shouldSkipAiProviderWrapping(OPENAI_INTEGRATION_NAME)).toBe(true); + + _INTERNAL_clearAiProviderSkips(); + expect(_INTERNAL_shouldSkipAiProviderWrapping(OPENAI_INTEGRATION_NAME)).toBe(false); + + await withAgent(() => undefined); + expect(_INTERNAL_shouldSkipAiProviderWrapping(OPENAI_INTEGRATION_NAME)).toBe(true); + }); + it('names agent spans `invoke_agent {name}` and sets the gen_ai op', async () => { await withAgent(() => undefined); @@ -91,13 +128,113 @@ describe('createFlueInstrumentation', () => { expect(json?.data['gen_ai.agent.name']).toBe('Hello'); }); - // The agent operation re-enters once, and only the inner context names the conversation. - it('lifts the conversation id off the re-entered agent operation', async () => { - await withAgent(() => instrumentation.interceptor({ type: 'agent' }, INNER_CTX, async () => undefined)); + // The agent span opens before the conversation is known — the submission-scoped operation names + // the agent, and the conversation arrives on the observations that follow. + it('sets the conversation id on the agent span from the observations', async () => { + await withAgent(() => { + instrumentation.observe( + { type: 'turn_start', turnId: 'turn_1', operationId: 'op_1', conversationId: 'conv_1' }, + {}, + ); + }); expect(findSpan('invoke_agent Hello')?.data['gen_ai.conversation.id']).toBe('conv_1'); }); + // The re-entered agent operation carries no `submissionId` and must not open a second span. + it('does not open a second agent span for the re-entry', async () => { + await withAgent(() => instrumentation.interceptor(AGENT_OP, INNER_CTX, async () => undefined)); + + const agentSpans = endedSpans + .map(span => spanToStaticSpanJSON(span)) + .filter(json => json.data['sentry.op'] === 'gen_ai.invoke_agent'); + expect(agentSpans).toHaveLength(1); + }); + + // A durable submission is resumed later, with nothing linking it to the request that enqueued it. + // Flue replays that request's `traceparent`, so the agent span should continue from it. + describe('trace continuation', () => { + const TRACE_ID = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'; + const CARRIER = { traceparent: `00-${TRACE_ID}-bbbbbbbbbbbbbbbb-01` }; + + function agentTraceId(): string | undefined { + return endedSpans.map(span => spanToStaticSpanJSON(span)).find(json => json.description === 'invoke_agent Hello') + ?.trace_id; + } + + it('continues the trace from the replayed traceparent', async () => { + await instrumentation.interceptor(AGENT_OP, { ...AGENT_CTX, traceCarrier: CARRIER }, async () => undefined); + + expect(agentTraceId()).toBe(TRACE_ID); + }); + + it('ignores a malformed traceparent', async () => { + await instrumentation.interceptor( + AGENT_OP, + { ...AGENT_CTX, traceCarrier: { traceparent: 'not-a-traceparent' } }, + async () => undefined, + ); + + expect(agentTraceId()).not.toBe(TRACE_ID); + }); + + // An in-process dispatch is genuinely part of the surrounding trace; continuing the persisted + // one would detach it from the request it is actually running inside. + it('keeps the active trace when one is already running', async () => { + await startSpan({ name: 'incoming request' }, async () => { + await instrumentation.interceptor(AGENT_OP, { ...AGENT_CTX, traceCarrier: CARRIER }, async () => undefined); + }); + + expect(agentTraceId()).not.toBe(TRACE_ID); + }); + }); + + // The spanned operation carries no `agentName` — only the submission wrapper does, and that opens + // no span. The name arrives on the observations instead. + it('names the agent span from the observations', async () => { + await instrumentation.interceptor(AGENT_OP, {}, async () => { + instrumentation.observe({ type: 'agent_start', operationId: 'op_1', agentName: 'Hello' }, {}); + }); + + const json = findSpan('invoke_agent Hello'); + expect(json).toBeDefined(); + expect(json?.data['gen_ai.agent.name']).toBe('Hello'); + }); + + // Delegation nests a second agent operation inside the first, in its own session. Flue defers + // each to a microtask, so the helper mirrors that rather than calling the interceptor directly. + describe('subagent delegation', () => { + function runNested(operationId: string, ctx: Record, next: () => Promise): Promise { + return Promise.resolve().then(() => instrumentation.interceptor({ type: 'agent', operationId }, ctx, next)); + } + + it('opens one agent span per invocation rather than folding the delegate into its parent', async () => { + await runNested('op_parent', {}, async () => { + instrumentation.observe({ type: 'agent_start', operationId: 'op_parent', conversationId: 'conv_parent' }, {}); + // The tool's task delegation, then the delegate's own prompt. + return runNested('op_child', {}, async () => { + instrumentation.observe({ type: 'agent_start', operationId: 'op_child', conversationId: 'conv_child' }, {}); + }); + }); + + expect(agentSpans()).toHaveLength(2); + }); + + // Asserted on conversation rather than agent name: the observation stream reports the root + // agent's name for both operations, so the delegate's own name never reaches us. + it('keeps each invocation on its own conversation', async () => { + await runNested('op_parent', {}, async () => { + instrumentation.observe({ type: 'agent_start', operationId: 'op_parent', conversationId: 'conv_parent' }, {}); + return runNested('op_child', {}, async () => { + instrumentation.observe({ type: 'agent_start', operationId: 'op_child', conversationId: 'conv_child' }, {}); + }); + }); + + // The delegate ends first, so order is inner-to-outer. + expect(agentSpans().map(json => json.data['gen_ai.conversation.id'])).toEqual(['conv_child', 'conv_parent']); + }); + }); + it('does not span operations other than `agent`', async () => { await instrumentation.interceptor({ type: 'model', turnId: 'turn_1' }, {}, async () => undefined); @@ -106,7 +243,10 @@ describe('createFlueInstrumentation', () => { it('opens a chat span on turn_start and completes it from the settled turn', async () => { await withAgent(() => { - instrumentation.observe({ type: 'turn_start', turnId: 'turn_1', conversationId: 'conv_1' }, {}); + instrumentation.observe( + { type: 'turn_start', turnId: 'turn_1', operationId: 'op_1', conversationId: 'conv_1' }, + {}, + ); instrumentation.observe(turn(), {}); }); @@ -123,7 +263,7 @@ describe('createFlueInstrumentation', () => { // Flue computes costs itself; the provider SDKs report none. it('records token usage and Flue-computed cost on the chat span', async () => { await withAgent(() => { - instrumentation.observe({ type: 'turn_start', turnId: 'turn_1' }, {}); + instrumentation.observe({ type: 'turn_start', turnId: 'turn_1', operationId: 'op_1' }, {}); instrumentation.observe(turn(), {}); }); @@ -134,9 +274,25 @@ describe('createFlueInstrumentation', () => { expect(json?.data['gen_ai.cost.total_tokens']).toBe(0.001199); }); + it('records the model-call tuning and provider endpoint', async () => { + await withAgent(() => { + instrumentation.observe({ type: 'turn_start', turnId: 'turn_1', operationId: 'op_1', purpose: 'agent' }, {}); + instrumentation.observe(turn(), {}); + }); + + const json = findSpan('chat claude-haiku-4.5'); + expect(json?.data['gen_ai.request.temperature']).toBe(0.7); + expect(json?.data['gen_ai.request.max_tokens']).toBe(1024); + expect(json?.data['gen_ai.request.reasoning.level']).toBe('high'); + expect(json?.data['server.address']).toBe('api.anthropic.com'); + expect(json?.data['server.port']).toBe(443); + // Distinguishes a compaction turn from a user-facing one. + expect(json?.data['flue.turn.purpose']).toBe('agent'); + }); + it('marks a failed turn as errored', async () => { await withAgent(() => { - instrumentation.observe({ type: 'turn_start', turnId: 'turn_1' }, {}); + instrumentation.observe({ type: 'turn_start', turnId: 'turn_1', operationId: 'op_1' }, {}); instrumentation.observe(turn({ isError: true }), {}); }); @@ -156,7 +312,7 @@ describe('createFlueInstrumentation', () => { }; await withAgent(() => { - instrumentation.observe({ type: 'turn_start', turnId: 'turn_1' }, {}); + instrumentation.observe({ type: 'turn_start', turnId: 'turn_1', operationId: 'op_1' }, {}); instrumentation.observe(turn({ isError: true, response: { usage: empty } }), {}); }); @@ -181,12 +337,24 @@ describe('createFlueInstrumentation', () => { } satisfies FlueObservation; async function record(instr: FlueInstrumentation): Promise { - await instr.interceptor({ type: 'agent' }, AGENT_CTX, async () => { - instr.observe({ type: 'turn_start', turnId: 'turn_1' }, {}); + await instr.interceptor(AGENT_OP, AGENT_CTX, async () => { + instr.observe({ type: 'turn_start', turnId: 'turn_1', operationId: 'op_1' }, {}); instr.observe(requestContent, {}); instr.observe(turn({ response: { ...turn().response, output: { role: 'assistant' } } }), {}); - instr.observe({ type: 'tool_start', toolCallId: 'c1', toolName: 'get_weather', args: { city: 'Berlin' } }, {}); - instr.observe({ type: 'tool', toolCallId: 'c1', toolName: 'get_weather', result: 'sunny' }, {}); + instr.observe( + { + type: 'tool_start', + toolCallId: 'c1', + toolName: 'get_weather', + args: { city: 'Berlin' }, + operationId: 'op_1', + }, + {}, + ); + instr.observe( + { type: 'tool', toolCallId: 'c1', toolName: 'get_weather', result: 'sunny', operationId: 'op_1' }, + {}, + ); }); } @@ -215,6 +383,33 @@ describe('createFlueInstrumentation', () => { expect(findSpan('execute_tool get_weather')?.data['gen_ai.tool.call.arguments']).toBeUndefined(); }); + // The client is replaced per request on Cloudflare, so options captured once at construction + // would be the wrong ones for every later request. + it('follows the current client when it is replaced', async () => { + const instr = createFlueInstrumentation(); + await record(instr); + expect(findSpan('chat claude-haiku-4.5')?.data['gen_ai.input.messages']).toBeDefined(); + + endedSpans.length = 0; + const strict = new TestClient( + getDefaultTestClientOptions({ + dsn: 'https://public@dsn.ingest.sentry.io/1337', + tracesSampleRate: 1, + traceLifecycle: 'stream', + dataCollection: { genAI: { inputs: false, outputs: false } }, + }), + ); + setCurrentClient(strict); + strict.init(); + strict.on('spanEnd', span => endedSpans.push(span)); + + await record(instr); + + const chat = findSpan('chat claude-haiku-4.5'); + expect(chat).toBeDefined(); + expect(chat?.data['gen_ai.input.messages']).toBeUndefined(); + }); + it('omits outputs when recordOutputs is false but keeps inputs', async () => { await record(createFlueInstrumentation({ recordOutputs: false })); @@ -227,8 +422,11 @@ describe('createFlueInstrumentation', () => { it('emits execute_tool spans keyed by tool call id', async () => { await withAgent(() => { - instrumentation.observe({ type: 'tool_start', toolCallId: 'call_1', toolName: 'get_weather' }, {}); - instrumentation.observe({ type: 'tool', toolCallId: 'call_1', toolName: 'get_weather' }, {}); + instrumentation.observe( + { type: 'tool_start', toolCallId: 'call_1', toolName: 'get_weather', operationId: 'op_1' }, + {}, + ); + instrumentation.observe({ type: 'tool', toolCallId: 'call_1', toolName: 'get_weather', operationId: 'op_1' }, {}); }); const json = findSpan('execute_tool get_weather'); @@ -240,13 +438,46 @@ describe('createFlueInstrumentation', () => { it('marks a failed tool call as errored', async () => { await withAgent(() => { - instrumentation.observe({ type: 'tool_start', toolCallId: 'call_1', toolName: 'boom' }, {}); - instrumentation.observe({ type: 'tool', toolCallId: 'call_1', toolName: 'boom', isError: true }, {}); + instrumentation.observe({ type: 'tool_start', toolCallId: 'call_1', toolName: 'boom', operationId: 'op_1' }, {}); + instrumentation.observe( + { type: 'tool', toolCallId: 'call_1', toolName: 'boom', isError: true, operationId: 'op_1' }, + {}, + ); }); expect(findSpan('execute_tool boom')?.status).toBe('internal_error'); }); + // Two agent runs overlap on a busy server. With shared closure state the second run is mistaken + // for a re-entry of the first: it gets no span, and its conversation id lands on the first's span. + it('keeps concurrent agent runs separate', async () => { + const runA = instrumentation.interceptor({ type: 'agent', operationId: 'op_a' }, { agentName: 'A' }, async () => { + instrumentation.observe( + { type: 'turn_start', turnId: 'turn_a', operationId: 'op_a', conversationId: 'conv_a' }, + {}, + ); + // B starts while A is still open. + await instrumentation.interceptor({ type: 'agent', operationId: 'op_b' }, { agentName: 'B' }, async () => { + instrumentation.observe( + { type: 'turn_start', turnId: 'turn_b', operationId: 'op_b', conversationId: 'conv_b' }, + {}, + ); + instrumentation.observe(turn({ turnId: 'turn_b', operationId: 'op_b' }), {}); + }); + instrumentation.observe(turn({ turnId: 'turn_a', operationId: 'op_a' }), {}); + }); + await runA; + + const agentA = findSpan('invoke_agent A'); + const agentB = findSpan('invoke_agent B'); + expect(agentA).toBeDefined(); + expect(agentB).toBeDefined(); + + // Each run keeps its own conversation; neither is overwritten by the other. + expect(agentA?.data['gen_ai.conversation.id']).toBe('conv_a'); + expect(agentB?.data['gen_ai.conversation.id']).toBe('conv_b'); + }); + it('ignores a settled turn or tool it never opened a span for', async () => { await withAgent(() => { instrumentation.observe(turn({ turnId: 'never_started' }), {}); @@ -258,8 +489,11 @@ describe('createFlueInstrumentation', () => { it('ends spans still open at dispose', async () => { await withAgent(() => { - instrumentation.observe({ type: 'turn_start', turnId: 'turn_1' }, {}); - instrumentation.observe({ type: 'tool_start', toolCallId: 'call_1', toolName: 'get_weather' }, {}); + instrumentation.observe({ type: 'turn_start', turnId: 'turn_1', operationId: 'op_1' }, {}); + instrumentation.observe( + { type: 'tool_start', toolCallId: 'call_1', toolName: 'get_weather', operationId: 'op_1' }, + {}, + ); }); // Never settled, so the span keeps the unqualified name it opened with. expect(findSpan('chat')).toBeUndefined(); From d5346b28b4e6965463084b2dbae418cc39596d71 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Fri, 11 Sep 2026 13:45:12 +0200 Subject: [PATCH 03/12] test(server-utils): Cover the Flue span cap and provider skip guard Both tests fail against the previous implementation: the first leaves `openai` registered before the run so the old first-entry-only guard short-circuits, and the second overflows the turn tracker to prove the evicted span is ended rather than dropped unsent. Co-Authored-By: Claude Opus 5 --- .../test/ai/lib/tracing/flue.test.ts | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/packages/server-utils/test/ai/lib/tracing/flue.test.ts b/packages/server-utils/test/ai/lib/tracing/flue.test.ts index 066d94b16f76..5d29e7424759 100644 --- a/packages/server-utils/test/ai/lib/tracing/flue.test.ts +++ b/packages/server-utils/test/ai/lib/tracing/flue.test.ts @@ -3,6 +3,7 @@ import type { Span } from '@sentry/core'; import { _INTERNAL_clearAiProviderSkips, _INTERNAL_shouldSkipAiProviderWrapping, + _INTERNAL_skipAiProviderWrapping, getMainCarrier, setCurrentClient, spanToStaticSpanJSON, @@ -10,6 +11,8 @@ import { } from '@sentry/core'; import { ANTHROPIC_AI_INTEGRATION_NAME } from '../../../../src/ai/anthropic-ai/constants'; import { createFlueInstrumentation } from '../../../../src/ai/flue'; +import { MAX_TRACKED_FLUE_SPANS } from '../../../../src/ai/flue/constants'; +import { GOOGLE_GENAI_INTEGRATION_NAME } from '../../../../src/ai/google-genai/constants'; import type { FlueInstrumentation, FlueObservation } from '../../../../src/ai/flue/types'; import { OPENAI_INTEGRATION_NAME } from '../../../../src/ai/openai/constants'; import { getDefaultTestClientOptions, TestClient } from '../../../mocks/client'; @@ -118,6 +121,29 @@ describe('createFlueInstrumentation', () => { expect(_INTERNAL_shouldSkipAiProviderWrapping(OPENAI_INTEGRATION_NAME)).toBe(true); }); + // The guard has to hold for every provider, not just the first: another integration may have + // registered a skip for one of them already, which would otherwise short-circuit the rest. + it('applies the skip to every provider when only some are already registered', async () => { + _INTERNAL_skipAiProviderWrapping([OPENAI_INTEGRATION_NAME]); + + await withAgent(() => undefined); + + expect(_INTERNAL_shouldSkipAiProviderWrapping(ANTHROPIC_AI_INTEGRATION_NAME)).toBe(true); + expect(_INTERNAL_shouldSkipAiProviderWrapping(GOOGLE_GENAI_INTEGRATION_NAME)).toBe(true); + }); + + // A turn whose stream is abandoned never emits the settled `turn` that would remove it, so the + // tracker is capped. Eviction has to end the span it drops, or it is never sent. + it('ends the oldest chat span when the turn tracker overflows', async () => { + await withAgent(() => { + for (let i = 0; i <= MAX_TRACKED_FLUE_SPANS; i++) { + instrumentation.observe({ type: 'turn_start', turnId: `turn_${i}`, operationId: 'op_1' }, {}); + } + }); + + expect(endedSpans.filter(span => spanToStaticSpanJSON(span).data['sentry.op'] === 'gen_ai.chat')).toHaveLength(1); + }); + it('names agent spans `invoke_agent {name}` and sets the gen_ai op', async () => { await withAgent(() => undefined); From 734cec300e99d39672f32cb6751d91c095e551eb Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Fri, 11 Sep 2026 15:09:46 +0200 Subject: [PATCH 04/12] test(server-utils): Cover the active turn and tool spans The `model` and `tool` interceptor branches open no span; they make the span `observe` already opened active so the provider's HTTP call and the tool's own work nest inside it. Deleting both branches left all 28 tests green, and the e2e does not reach it either: its parent assertions come from the observation stream firing inside the agent operation, and nothing in that scenario opens a span inside a tool or model operation. Each case fails when its own branch is removed, and neither fails for the other's. Co-Authored-By: Claude Opus 5 --- .../test/ai/lib/tracing/flue.test.ts | 31 +++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/packages/server-utils/test/ai/lib/tracing/flue.test.ts b/packages/server-utils/test/ai/lib/tracing/flue.test.ts index 5d29e7424759..ce3d4e8e46ce 100644 --- a/packages/server-utils/test/ai/lib/tracing/flue.test.ts +++ b/packages/server-utils/test/ai/lib/tracing/flue.test.ts @@ -7,6 +7,7 @@ import { getMainCarrier, setCurrentClient, spanToStaticSpanJSON, + startInactiveSpan, startSpan, } from '@sentry/core'; import { ANTHROPIC_AI_INTEGRATION_NAME } from '../../../../src/ai/anthropic-ai/constants'; @@ -267,6 +268,36 @@ describe('createFlueInstrumentation', () => { expect(endedSpans).toHaveLength(0); }); + // The `model` and `tool` operations open no span of their own; they make the span `observe` + // already opened active, so the provider's HTTP call and the tool's own work nest inside it + // rather than landing beside it as siblings of the agent invocation. + it('makes the turn span active for the model operation it wraps', async () => { + await withAgent(async () => { + instrumentation.observe({ type: 'turn_start', turnId: 'turn_1', operationId: 'op_1' }, {}); + await instrumentation.interceptor({ type: 'model', turnId: 'turn_1' }, {}, async () => { + startInactiveSpan({ name: 'provider request' }).end(); + }); + instrumentation.observe(turn(), {}); + }); + + expect(findSpan('provider request')?.parent_span_id).toBe(findSpan('chat claude-haiku-4.5')?.span_id); + }); + + it('makes the tool span active for the tool operation it wraps', async () => { + await withAgent(async () => { + instrumentation.observe( + { type: 'tool_start', toolCallId: 'call_1', toolName: 'get_weather', operationId: 'op_1' }, + {}, + ); + await instrumentation.interceptor({ type: 'tool', toolCallId: 'call_1' }, {}, async () => { + startInactiveSpan({ name: 'tool work' }).end(); + }); + instrumentation.observe({ type: 'tool', toolCallId: 'call_1', toolName: 'get_weather', operationId: 'op_1' }, {}); + }); + + expect(findSpan('tool work')?.parent_span_id).toBe(findSpan('execute_tool get_weather')?.span_id); + }); + it('opens a chat span on turn_start and completes it from the settled turn', async () => { await withAgent(() => { instrumentation.observe( From 753b0299f7982c30a0476efd31f8feb7f45b78c2 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Mon, 14 Sep 2026 10:50:42 +0300 Subject: [PATCH 05/12] test(server-utils): Assert the serialized `finish_reasons` value Co-Authored-By: Claude Opus 5 --- packages/server-utils/test/ai/lib/tracing/flue.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/server-utils/test/ai/lib/tracing/flue.test.ts b/packages/server-utils/test/ai/lib/tracing/flue.test.ts index ce3d4e8e46ce..3e215913236d 100644 --- a/packages/server-utils/test/ai/lib/tracing/flue.test.ts +++ b/packages/server-utils/test/ai/lib/tracing/flue.test.ts @@ -313,7 +313,7 @@ describe('createFlueInstrumentation', () => { expect(json?.data['gen_ai.request.model']).toBe('claude-haiku-4.5'); expect(json?.data['gen_ai.provider.name']).toBe('anthropic'); expect(json?.data['gen_ai.response.id']).toBe('resp_1'); - expect(json?.data['gen_ai.response.finish_reasons']).toEqual(['stop']); + expect(json?.data['gen_ai.response.finish_reasons']).toBe('["stop"]'); expect(json?.data['gen_ai.conversation.id']).toBe('conv_1'); }); From b29ef5f03e59215a7c3b1689ed09c8c7f830c7a8 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Mon, 14 Sep 2026 10:57:30 +0300 Subject: [PATCH 06/12] test(server-utils): Make the trace continuation assertions provable Both cases passed vacuously. `not.toBe(TRACE_ID)` on the malformed carrier also held when no agent span was opened at all, since `undefined` is not the carrier's id either; suppressing span creation entirely failed nine other tests and left that one green. And the active-trace case never checked the span landed on the surrounding request's trace, so continuing an unrelated new trace passed too. Now the first asserts a well-formed trace id, and the second asserts the agent span carries the `incoming request` span's own trace id. Dropping the `!getActiveSpan()` guard fails the second. Co-Authored-By: Claude Opus 5 --- packages/server-utils/test/ai/lib/tracing/flue.test.ts | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/packages/server-utils/test/ai/lib/tracing/flue.test.ts b/packages/server-utils/test/ai/lib/tracing/flue.test.ts index 3e215913236d..60756df70bd6 100644 --- a/packages/server-utils/test/ai/lib/tracing/flue.test.ts +++ b/packages/server-utils/test/ai/lib/tracing/flue.test.ts @@ -202,17 +202,23 @@ describe('createFlueInstrumentation', () => { async () => undefined, ); + // Asserting a well-formed id, not just "not the carrier's": the span has to still be opened + // on a fresh trace, and `not.toBe` alone would also pass if no span were opened at all. + expect(agentTraceId()).toMatch(/^[0-9a-f]{32}$/); expect(agentTraceId()).not.toBe(TRACE_ID); }); // An in-process dispatch is genuinely part of the surrounding trace; continuing the persisted // one would detach it from the request it is actually running inside. it('keeps the active trace when one is already running', async () => { - await startSpan({ name: 'incoming request' }, async () => { + let requestTraceId: string | undefined; + + await startSpan({ name: 'incoming request' }, async span => { + requestTraceId = spanToStaticSpanJSON(span).trace_id; await instrumentation.interceptor(AGENT_OP, { ...AGENT_CTX, traceCarrier: CARRIER }, async () => undefined); }); - expect(agentTraceId()).not.toBe(TRACE_ID); + expect(agentTraceId()).toBe(requestTraceId); }); }); From 41408e62e3ee2921032c26a02eca5abcf70f0720 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Mon, 14 Sep 2026 16:14:33 +0300 Subject: [PATCH 07/12] test(server-utils): Cover error capture for a failed Flue tool Fails without the capture: the span is still errored, so only the error event distinguishes the two. Co-Authored-By: Claude Opus 5 --- .../test/ai/lib/tracing/flue.test.ts | 40 ++++++++++++++++++- 1 file changed, 39 insertions(+), 1 deletion(-) diff --git a/packages/server-utils/test/ai/lib/tracing/flue.test.ts b/packages/server-utils/test/ai/lib/tracing/flue.test.ts index 60756df70bd6..10d24cb26329 100644 --- a/packages/server-utils/test/ai/lib/tracing/flue.test.ts +++ b/packages/server-utils/test/ai/lib/tracing/flue.test.ts @@ -56,12 +56,13 @@ function turn(overrides: Partial = {}): FlueObservation { describe('createFlueInstrumentation', () => { let endedSpans: Span[]; + let client: TestClient; let instrumentation: FlueInstrumentation; beforeEach(() => { _INTERNAL_clearAiProviderSkips(); getMainCarrier().__SENTRY__ = undefined; - const client = new TestClient( + client = new TestClient( getDefaultTestClientOptions({ dsn: 'https://public@dsn.ingest.sentry.io/1337', tracesSampleRate: 1, @@ -499,6 +500,43 @@ describe('createFlueInstrumentation', () => { expect(json?.data['gen_ai.tool.name']).toBe('get_weather'); }); + // Flue catches whatever the tool threw and feeds it back to the model as a tool result, so + // nothing reaches the SDK's global handlers. Without an explicit capture there is an errored span + // and no error event at all. + it('captures an error event for a failed tool, rebuilt from `errorInfo`', async () => { + await withAgent(() => { + instrumentation.observe({ type: 'tool_start', toolCallId: 'c1', toolName: 'boom', operationId: 'op_1' }, {}); + instrumentation.observe( + { + type: 'tool', + toolCallId: 'c1', + toolName: 'boom', + operationId: 'op_1', + isError: true, + errorInfo: { type: 'Error', name: 'TypeError', message: 'kaboom', stack: 'TypeError: kaboom\n at run' }, + }, + {}, + ); + }); + + await client.flush(); + + const exception = client.event?.exception?.values?.[0]; + expect(exception?.type).toBe('TypeError'); + expect(exception?.value).toBe('kaboom'); + expect(exception?.mechanism?.type).toBe('auto.ai.flue.tool_error'); + }); + + it('does not capture an error event for a tool that succeeded', async () => { + await withAgent(() => { + instrumentation.observe({ type: 'tool_start', toolCallId: 'c1', toolName: 'ok', operationId: 'op_1' }, {}); + instrumentation.observe({ type: 'tool', toolCallId: 'c1', toolName: 'ok', operationId: 'op_1' }, {}); + }); + await client.flush(); + + expect(client.event).toBeUndefined(); + }); + it('marks a failed tool call as errored', async () => { await withAgent(() => { instrumentation.observe({ type: 'tool_start', toolCallId: 'call_1', toolName: 'boom', operationId: 'op_1' }, {}); From 93ddf47952ca7d73aa169470a5a4e607f1e9192f Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Mon, 14 Sep 2026 16:18:03 +0300 Subject: [PATCH 08/12] test(server-utils): Assert the aligned Flue error mechanism Co-Authored-By: Claude Opus 5 --- packages/server-utils/test/ai/lib/tracing/flue.test.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/packages/server-utils/test/ai/lib/tracing/flue.test.ts b/packages/server-utils/test/ai/lib/tracing/flue.test.ts index 10d24cb26329..2ffbfef371cc 100644 --- a/packages/server-utils/test/ai/lib/tracing/flue.test.ts +++ b/packages/server-utils/test/ai/lib/tracing/flue.test.ts @@ -524,7 +524,8 @@ describe('createFlueInstrumentation', () => { const exception = client.event?.exception?.values?.[0]; expect(exception?.type).toBe('TypeError'); expect(exception?.value).toBe('kaboom'); - expect(exception?.mechanism?.type).toBe('auto.ai.flue.tool_error'); + expect(exception?.mechanism?.type).toBe('auto.ai.flue'); + expect(exception?.mechanism?.handled).toBe(false); }); it('does not capture an error event for a tool that succeeded', async () => { From b1192bdf32ca65a570ba3ec8f7db544809203bdf Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Mon, 14 Sep 2026 23:04:02 +0300 Subject: [PATCH 09/12] test(server-utils): Assert a failed Flue tool is captured exactly once Guards the concern raised on the Mastra error-capture PR: if a second capture path is added, or the error starts reaching the global handlers, this fails. Co-Authored-By: Claude Opus 5 --- .../test/ai/lib/tracing/flue.test.ts | 34 +++++++++++++++++-- 1 file changed, 32 insertions(+), 2 deletions(-) diff --git a/packages/server-utils/test/ai/lib/tracing/flue.test.ts b/packages/server-utils/test/ai/lib/tracing/flue.test.ts index 2ffbfef371cc..5817e1033871 100644 --- a/packages/server-utils/test/ai/lib/tracing/flue.test.ts +++ b/packages/server-utils/test/ai/lib/tracing/flue.test.ts @@ -1,5 +1,5 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; -import type { Span } from '@sentry/core'; +import type { Event, Span } from '@sentry/core'; import { _INTERNAL_clearAiProviderSkips, _INTERNAL_shouldSkipAiProviderWrapping, @@ -525,7 +525,37 @@ describe('createFlueInstrumentation', () => { expect(exception?.type).toBe('TypeError'); expect(exception?.value).toBe('kaboom'); expect(exception?.mechanism?.type).toBe('auto.ai.flue'); - expect(exception?.mechanism?.handled).toBe(false); + // Handled: Flue caught the throw and returned it to the model, so no global hook sees it. + expect(exception?.mechanism?.handled).toBe(true); + }); + + // Flue swallows the throw, so nothing else reports it. If a second capture path is ever added, + // or the error starts propagating to the global handlers, this catches the duplicate. + it('captures a failed tool exactly once', async () => { + const events: Event[] = []; + TestClient.sendEventCalled = event => events.push(event); + + try { + await withAgent(() => { + instrumentation.observe({ type: 'tool_start', toolCallId: 'c1', toolName: 'boom', operationId: 'op_1' }, {}); + instrumentation.observe( + { + type: 'tool', + toolCallId: 'c1', + toolName: 'boom', + operationId: 'op_1', + isError: true, + errorInfo: { name: 'Error', message: 'kaboom' }, + }, + {}, + ); + }); + await client.flush(); + } finally { + TestClient.sendEventCalled = undefined; + } + + expect(events.filter(event => event.exception?.values?.length)).toHaveLength(1); }); it('does not capture an error event for a tool that succeeded', async () => { From 211fe8ebc9a5334323ea2bda2b1d522eeb4d3f3b Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Tue, 15 Sep 2026 12:53:28 +0300 Subject: [PATCH 10/12] test(node): Bound the Flue span assertions to exact counts `toContain` plus `find` left the integration test blind to duplicate spans and to everything about the second turn. Count each span kind, assert both `chat` spans rather than whichever one `find` returned first, and tie the agent span to the surrounding transaction. Mutation-checked: dropping the submission wrapper guard, mis-parenting the second turn, and dropping its usage attributes each fail now; the last two passed before. Co-Authored-By: Claude Opus 5 --- .../suites/tracing/flue/test.ts | 77 +++++++++++++------ 1 file changed, 53 insertions(+), 24 deletions(-) diff --git a/dev-packages/node-integration-tests/suites/tracing/flue/test.ts b/dev-packages/node-integration-tests/suites/tracing/flue/test.ts index ec7a26f71ad1..b8b8a71a9a38 100644 --- a/dev-packages/node-integration-tests/suites/tracing/flue/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/flue/test.ts @@ -3,6 +3,7 @@ import { GEN_AI_CONVERSATION_ID, GEN_AI_COST_TOTAL_TOKENS, GEN_AI_OPERATION_NAME, + GEN_AI_RESPONSE_FINISH_REASONS, GEN_AI_TOOL_NAME, GEN_AI_USAGE_INPUT_TOKENS, GEN_AI_USAGE_OUTPUT_TOKENS, @@ -39,42 +40,70 @@ conditionalTest({ min: 22 })('Flue integration', () => { } test('creates the invoke_agent / chat / execute_tool hierarchy', async () => { + let rootSpanId: string | undefined; + await createRunner() - .expect({ transaction: { transaction: 'flue-test' } }) + .expect({ + transaction: event => { + expect(event.transaction).toBe('flue-test'); + rootSpanId = event.contexts?.trace?.span_id; + }, + }) .expect({ span: container => { const spans = container.items; - const names = spans.map(span => span.name); - expect(names).toContain('invoke_agent Hello'); - expect(names).toContain('execute_tool get_weather'); - expect(names.filter(name => name?.startsWith('chat'))).toHaveLength(2); + // Counted rather than looked up: the interceptor skips the submission wrapper + // operation, so one dispatch opens exactly one agent span, and each turn and tool call + // is spanned once. `find` passes just as happily on a duplicate. + expect(spans.filter(span => span.attributes['sentry.origin']?.value === 'auto.ai.flue')).toHaveLength(4); + + const agents = spans.filter(span => span.name === 'invoke_agent Hello'); + const chats = spans.filter(span => span.name === 'chat faux-model'); + const tools = spans.filter(span => span.name === 'execute_tool get_weather'); + expect(agents).toHaveLength(1); + expect(tools).toHaveLength(1); + // One turn asks for the tool, the second answers with its result. + expect(chats).toHaveLength(2); + + const agent = agents[0]!; + expect(agent.attributes['sentry.op']?.value).toBe('gen_ai.invoke_agent'); + expect(agent.attributes[GEN_AI_OPERATION_NAME]?.value).toBe('invoke_agent'); + expect(agent.attributes[GEN_AI_AGENT_NAME]?.value).toBe('Hello'); + expect(agent.parent_span_id).toBe(rootSpanId); + + const conversationId = agent.attributes[GEN_AI_CONVERSATION_ID]?.value; + expect(conversationId).toEqual(expect.any(String)); - const agent = spans.find(span => span.name === 'invoke_agent Hello')!; - expect(agent.attributes['sentry.op'].value).toBe('gen_ai.invoke_agent'); - expect(agent.attributes['sentry.origin'].value).toBe('auto.ai.flue'); - expect(agent.attributes[GEN_AI_OPERATION_NAME].value).toBe('invoke_agent'); - expect(agent.attributes[GEN_AI_AGENT_NAME].value).toBe('Hello'); - expect(agent.attributes[GEN_AI_CONVERSATION_ID].value).toEqual(expect.any(String)); + // Both turns, not just the first: they leave the provider by different paths (a tool + // call, then a final answer) and resolve their parent through separate tracker lookups. + for (const chat of chats) { + expect(chat.attributes['sentry.op']?.value).toBe('gen_ai.chat'); + expect(chat.attributes['sentry.origin']?.value).toBe('auto.ai.flue'); + expect(chat.attributes[GEN_AI_OPERATION_NAME]?.value).toBe('chat'); + expect(chat.attributes[GEN_AI_CONVERSATION_ID]?.value).toBe(conversationId); + expect(chat.parent_span_id).toBe(agent.span_id); + expect(chat.attributes[GEN_AI_USAGE_INPUT_TOKENS]?.value).toBeGreaterThan(0); + expect(chat.attributes[GEN_AI_USAGE_OUTPUT_TOKENS]?.value).toBeGreaterThan(0); + expect(chat.attributes[GEN_AI_USAGE_TOTAL_TOKENS]?.value).toBeGreaterThan(0); + // Flue computes cost itself; no provider SDK reports it. The faux provider prices + // every model at zero, so this only proves the attribute is mapped. + expect(chat.attributes[GEN_AI_COST_TOTAL_TOKENS]?.value).toEqual(expect.any(Number)); + } - const chat = spans.find(span => span.name?.startsWith('chat'))!; - expect(chat.attributes['sentry.op'].value).toBe('gen_ai.chat'); - expect(chat.attributes['sentry.origin'].value).toBe('auto.ai.flue'); - expect(chat.attributes[GEN_AI_USAGE_INPUT_TOKENS].value).toEqual(expect.any(Number)); - expect(chat.attributes[GEN_AI_USAGE_OUTPUT_TOKENS].value).toEqual(expect.any(Number)); - expect(chat.attributes[GEN_AI_USAGE_TOTAL_TOKENS].value).toEqual(expect.any(Number)); - // Flue computes cost itself; no provider SDK reports it. - expect(chat.attributes[GEN_AI_COST_TOTAL_TOKENS].value).toEqual(expect.any(Number)); + expect(chats.map(chat => chat.attributes[GEN_AI_RESPONSE_FINISH_REASONS]?.value).sort()).toEqual([ + '["stop"]', + '["toolUse"]', + ]); - const tool = spans.find(span => span.name === 'execute_tool get_weather')!; - expect(tool.attributes['sentry.op'].value).toBe('gen_ai.execute_tool'); - expect(tool.attributes['sentry.origin'].value).toBe('auto.ai.flue'); - expect(tool.attributes[GEN_AI_TOOL_NAME].value).toBe('get_weather'); + const tool = tools[0]!; + expect(tool.attributes['sentry.op']?.value).toBe('gen_ai.execute_tool'); + expect(tool.attributes['sentry.origin']?.value).toBe('auto.ai.flue'); + expect(tool.attributes[GEN_AI_TOOL_NAME]?.value).toBe('get_weather'); // Tool spans are siblings of `chat` under the agent invocation, matching how Flue's // own OpenTelemetry adapter projects them. expect(tool.parent_span_id).toBe(agent.span_id); - expect(chat.parent_span_id).toBe(agent.span_id); }, }) .start() From bf25c8336e7baf68153b9d301f6b22ff8c3485e6 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Thu, 17 Sep 2026 13:54:20 +0300 Subject: [PATCH 11/12] test(node): Make the Flue scenario's tool call succeed The tool declared no `input` schema, so Flue validated the model's arguments against an empty one and rejected them with "must not have additional properties". The tool body never ran. Nothing caught it: a rejected call still produces an `execute_tool` span with the right name and parent, and the suite only asserted those. Declare the schema, read the arguments from `ctx.data` where Flue passes them, and assert the span settles `ok` so the same silent failure can't come back. Co-Authored-By: Claude Opus 5 --- .../node-integration-tests/suites/tracing/flue/scenario.mjs | 6 +++++- .../node-integration-tests/suites/tracing/flue/test.ts | 4 ++++ 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/dev-packages/node-integration-tests/suites/tracing/flue/scenario.mjs b/dev-packages/node-integration-tests/suites/tracing/flue/scenario.mjs index 625abf964ee0..b71b000bf18f 100644 --- a/dev-packages/node-integration-tests/suites/tracing/flue/scenario.mjs +++ b/dev-packages/node-integration-tests/suites/tracing/flue/scenario.mjs @@ -2,6 +2,7 @@ import * as Sentry from '@sentry/node'; import { __flueBindAgentModule, init, instrument, useModel, useTool } from '@flue/runtime'; import { start } from '@flue/runtime/node'; import { fauxAssistantMessage, fauxProvider, fauxToolCall } from '@earendil-works/pi-ai/providers/faux'; +import * as v from 'valibot'; // `pi-ai`'s faux provider scripts model responses in-process, so the run is deterministic and needs // no provider key or mock server. Two steps: a tool call, then the final answer. @@ -21,7 +22,10 @@ function Hello() { useTool({ name: 'get_weather', description: 'Get the current weather for a city.', - run: ({ city }) => `It is 21 degrees and sunny in ${city}.`, + // Without an `input` schema Flue validates the call against an empty one and rejects the + // model's arguments, so the tool never runs and its span settles as an error. + input: v.object({ city: v.string() }), + run: ({ data }) => `It is 21 degrees and sunny in ${data.city}.`, }); return 'You are a helpful assistant.'; } diff --git a/dev-packages/node-integration-tests/suites/tracing/flue/test.ts b/dev-packages/node-integration-tests/suites/tracing/flue/test.ts index b8b8a71a9a38..be217a4505d5 100644 --- a/dev-packages/node-integration-tests/suites/tracing/flue/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/flue/test.ts @@ -20,6 +20,7 @@ const FLUE_DEPENDENCIES = { additionalDependencies: { '@flue/runtime': '2.0.3', '@earendil-works/pi-ai': '0.85.1', + valibot: '1.1.0', }, }; @@ -97,6 +98,9 @@ conditionalTest({ min: 22 })('Flue integration', () => { ]); const tool = tools[0]!; + // The tool has to actually run: a schema mismatch still produces a correctly named and + // parented span, so only the status separates a real call from a rejected one. + expect(tool.status).toBe('ok'); expect(tool.attributes['sentry.op']?.value).toBe('gen_ai.execute_tool'); expect(tool.attributes['sentry.origin']?.value).toBe('auto.ai.flue'); expect(tool.attributes[GEN_AI_TOOL_NAME]?.value).toBe('get_weather'); From 1cfa8dc6e744a710f5e0ede823cfdf2e84cd5e61 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Thu, 17 Sep 2026 13:54:41 +0300 Subject: [PATCH 12/12] test(node): Run the Flue suite on the streaming trace lifecycle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The suite inherited `traceLifecycle: 'static'` from the other gen-AI suites, where gen_ai spans reach the assertions by a detour: `extractGenAiSpansFromEvent` lifts them out of the transaction and re-emits them as a span container. Nothing in the Flue instrumentation depends on that, and streaming is what users get by default. On `stream` there is no transaction envelope — the `flue-test` root arrives in the same container as a segment span, so the root id comes from there instead. Co-Authored-By: Claude Opus 5 --- .../suites/tracing/flue/instrument.mjs | 2 +- .../suites/tracing/flue/test.ts | 13 ++++--------- 2 files changed, 5 insertions(+), 10 deletions(-) diff --git a/dev-packages/node-integration-tests/suites/tracing/flue/instrument.mjs b/dev-packages/node-integration-tests/suites/tracing/flue/instrument.mjs index 42052d281304..31609dbe3960 100644 --- a/dev-packages/node-integration-tests/suites/tracing/flue/instrument.mjs +++ b/dev-packages/node-integration-tests/suites/tracing/flue/instrument.mjs @@ -2,7 +2,7 @@ import * as Sentry from '@sentry/node'; import { loggingTransport } from '@sentry-internal/node-integration-tests'; Sentry.init({ - traceLifecycle: 'static', + traceLifecycle: 'stream', dsn: 'https://public@dsn.ingest.sentry.io/1337', release: '1.0', tracesSampleRate: 1.0, diff --git a/dev-packages/node-integration-tests/suites/tracing/flue/test.ts b/dev-packages/node-integration-tests/suites/tracing/flue/test.ts index be217a4505d5..98ad59dcc4df 100644 --- a/dev-packages/node-integration-tests/suites/tracing/flue/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/flue/test.ts @@ -41,19 +41,14 @@ conditionalTest({ min: 22 })('Flue integration', () => { } test('creates the invoke_agent / chat / execute_tool hierarchy', async () => { - let rootSpanId: string | undefined; - await createRunner() - .expect({ - transaction: event => { - expect(event.transaction).toBe('flue-test'); - rootSpanId = event.contexts?.trace?.span_id; - }, - }) .expect({ span: container => { const spans = container.items; + const root = spans.find(span => span.name === 'flue-test')!; + expect(root.is_segment).toBe(true); + // Counted rather than looked up: the interceptor skips the submission wrapper // operation, so one dispatch opens exactly one agent span, and each turn and tool call // is spanned once. `find` passes just as happily on a duplicate. @@ -71,7 +66,7 @@ conditionalTest({ min: 22 })('Flue integration', () => { expect(agent.attributes['sentry.op']?.value).toBe('gen_ai.invoke_agent'); expect(agent.attributes[GEN_AI_OPERATION_NAME]?.value).toBe('invoke_agent'); expect(agent.attributes[GEN_AI_AGENT_NAME]?.value).toBe('Hello'); - expect(agent.parent_span_id).toBe(rootSpanId); + expect(agent.parent_span_id).toBe(root.span_id); const conversationId = agent.attributes[GEN_AI_CONVERSATION_ID]?.value; expect(conversationId).toEqual(expect.any(String));