From 6c394c7c1ae91ccc354d1ec74a0f66f48e800ddd Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Wed, 9 Sep 2026 20:48:33 +0200 Subject: [PATCH 1/2] 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, 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 | 36 +++ .../suites/tracing/flue/test.ts | 86 +++++++ .../test/ai/lib/tracing/flue.test.ts | 213 ++++++++++++++++++ 4 files changed, 346 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..849a542947db --- /dev/null +++ b/dev-packages/node-integration-tests/suites/tracing/flue/scenario.mjs @@ -0,0 +1,36 @@ +import * as Sentry from '@sentry/node'; +import { __flueBindAgentModule, init, 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. +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..452a50475be0 --- /dev/null +++ b/packages/server-utils/test/ai/lib/tracing/flue.test.ts @@ -0,0 +1,213 @@ +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(); + }); + + 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 612798a1ce8e1b3a20b811ee19c25e5346972c07 Mon Sep 17 00:00:00 2001 From: RulaKhaled Date: Wed, 9 Sep 2026 20:59:42 +0200 Subject: [PATCH 2/2] test(server-utils): Cover Flue message content recording Co-Authored-By: Claude Opus 5 --- .../test/ai/lib/tracing/flue.test.ts | 59 +++++++++++++++++++ 1 file changed, 59 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 452a50475be0..97485285282e 100644 --- a/packages/server-utils/test/ai/lib/tracing/flue.test.ts +++ b/packages/server-utils/test/ai/lib/tracing/flue.test.ts @@ -166,6 +166,65 @@ describe('createFlueInstrumentation', () => { 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' }, {});