Skip to content

Commit 38396de

Browse files
JPeer264claude
andcommitted
fix(server-utils): Map Flue messages and token usage to the gen_ai conventions
Flue records its turn messages in pi-ai's shape, with a `toolResult` role and `toolCall` parts, and Sentry renders neither, so tool calls and tool results were missing from the conversation views. Its token counts come from pi-ai too, whose `input` leaves the cached tokens out while the conventions count them in, so input tokens and input cost were too low whenever prompt caching was active. A shared pi-ai mapper now translates messages, usage and the finish reason to the conventions, and the tool span records the result the model receives instead of Flue's internal result wrapper. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1 parent d880cf0 commit 38396de

11 files changed

Lines changed: 729 additions & 95 deletions

File tree

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
import * as Sentry from '@sentry/node';
2+
import { loggingTransport } from '@sentry-internal/node-integration-tests';
3+
4+
Sentry.init({
5+
dsn: 'https://public@dsn.ingest.sentry.io/1337',
6+
release: '1.0',
7+
tracesSampleRate: 1.0,
8+
dataCollection: { genAI: { inputs: true, outputs: true } },
9+
transport: loggingTransport,
10+
});

‎dev-packages/node-integration-tests/suites/tracing/flue/test.ts‎

Lines changed: 65 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,13 @@ import {
22
GEN_AI_AGENT_NAME,
33
GEN_AI_CONVERSATION_ID,
44
GEN_AI_COST_TOTAL_TOKENS,
5+
GEN_AI_INPUT_MESSAGES,
56
GEN_AI_OPERATION_NAME,
7+
GEN_AI_OUTPUT_MESSAGES,
68
GEN_AI_RESPONSE_FINISH_REASONS,
9+
GEN_AI_SYSTEM_INSTRUCTIONS,
10+
GEN_AI_TOOL_CALL_ARGUMENTS,
11+
GEN_AI_TOOL_CALL_RESULT,
712
GEN_AI_TOOL_NAME,
813
GEN_AI_USAGE_INPUT_TOKENS,
914
GEN_AI_USAGE_OUTPUT_TOKENS,
@@ -87,9 +92,10 @@ conditionalTest({ min: 22 })('Flue integration', () => {
8792
expect(chat.attributes[GEN_AI_COST_TOTAL_TOKENS]?.value).toEqual(expect.any(Number));
8893
}
8994

95+
// pi-ai's `toolUse` is reported as the conventions' `tool_call`.
9096
expect(chats.map(chat => chat.attributes[GEN_AI_RESPONSE_FINISH_REASONS]?.value).sort()).toEqual([
9197
'["stop"]',
92-
'["toolUse"]',
98+
'["tool_call"]',
9399
]);
94100

95101
const tool = tools[0]!;
@@ -111,4 +117,62 @@ conditionalTest({ min: 22 })('Flue integration', () => {
111117
},
112118
FLUE_DEPENDENCIES,
113119
);
120+
121+
createEsmAndCjsTests(
122+
__dirname,
123+
'scenario.mjs',
124+
'instrument-with-pii.mjs',
125+
(createRunner, test, mode) => {
126+
if (mode === 'cjs') {
127+
return;
128+
}
129+
130+
test('records the messages of the turns in the gen_ai conventions shape', async () => {
131+
await createRunner()
132+
.expect({
133+
span: container => {
134+
const chats = container.items
135+
.filter(span => span.name === 'chat faux-model')
136+
.sort((a, b) => a.start_timestamp - b.start_timestamp);
137+
expect(chats).toHaveLength(2);
138+
139+
// Flue sends pi-ai's shape (`toolResult`, `toolCall`); Sentry renders the conventions.
140+
expect(JSON.parse(String(chats[0]!.attributes[GEN_AI_INPUT_MESSAGES]?.value))).toEqual([
141+
{ role: 'user', parts: [{ type: 'text', content: 'What is the weather in Berlin?' }] },
142+
]);
143+
expect(JSON.parse(String(chats[0]!.attributes[GEN_AI_OUTPUT_MESSAGES]?.value))).toEqual([
144+
{
145+
role: 'assistant',
146+
parts: [{ type: 'tool_call', id: 'call_1', name: 'get_weather', arguments: '{"city":"Berlin"}' }],
147+
finish_reason: 'tool_call',
148+
},
149+
]);
150+
151+
const answerInput = JSON.parse(String(chats[1]!.attributes[GEN_AI_INPUT_MESSAGES]?.value));
152+
expect(answerInput.map((message: { role: string }) => message.role)).toEqual([
153+
'user',
154+
'assistant',
155+
'tool',
156+
]);
157+
expect(answerInput[2].parts[0]).toMatchObject({
158+
type: 'tool_call_response',
159+
id: 'call_1',
160+
name: 'get_weather',
161+
});
162+
expect(answerInput[2].parts[0].result).toContain('sunny in Berlin');
163+
expect(chats[1]!.attributes[GEN_AI_SYSTEM_INSTRUCTIONS]?.value).toContain('You are a helpful assistant.');
164+
165+
// The tool span records the result the model receives, not Flue's internal wrapper.
166+
const tool = container.items.find(span => span.name === 'execute_tool get_weather')!;
167+
expect(tool.attributes[GEN_AI_TOOL_CALL_ARGUMENTS]?.value).toBe('{"city":"Berlin"}');
168+
expect(tool.attributes[GEN_AI_TOOL_CALL_RESULT]?.value).toContain('sunny in Berlin');
169+
expect(tool.attributes[GEN_AI_TOOL_CALL_RESULT]?.value).not.toContain('details');
170+
},
171+
})
172+
.start()
173+
.completed();
174+
});
175+
},
176+
FLUE_DEPENDENCIES,
177+
);
114178
});

‎packages/server-utils/src/ai/flue/index.ts‎

Lines changed: 8 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,5 @@
11
import type { Span } from '@sentry/core';
22
import {
3-
_INTERNAL_shouldSkipAiProviderWrapping,
4-
_INTERNAL_skipAiProviderWrapping,
53
continueTrace,
64
getActiveSpan,
75
LRUMap,
@@ -10,11 +8,9 @@ import {
108
withActiveSpan,
119
} from '@sentry/core';
1210
import { GEN_AI_AGENT_NAME, GEN_AI_CONVERSATION_ID, GEN_AI_OPERATION_NAME } from '@sentry/conventions/attributes';
13-
import { ANTHROPIC_AI_INTEGRATION_NAME } from '../anthropic-ai/constants';
1411
import type { GenAiOptions } from '../core/utils';
1512
import { getGenAiSpanOp, resolveAIRecordingOptions } from '../core/utils';
16-
import { GOOGLE_GENAI_INTEGRATION_NAME } from '../google-genai/constants';
17-
import { OPENAI_INTEGRATION_NAME } from '../openai/constants';
13+
import { skipPiAiProviderIntegrations } from '../pi-ai/providers';
1814
import { FLUE_INSTRUMENTATION_KEY, FLUE_OPERATION, FLUE_ORIGIN, MAX_TRACKED_FLUE_SPANS } from './constants';
1915
import type { SpanTracker } from './utils';
2016
import {
@@ -29,8 +25,6 @@ import type { FlueInstrumentation } from './types';
2925

3026
export type FlueOptions = GenAiOptions;
3127

32-
const SKIPPED_PROVIDERS = [OPENAI_INTEGRATION_NAME, ANTHROPIC_AI_INTEGRATION_NAME, GOOGLE_GENAI_INTEGRATION_NAME];
33-
3428
/**
3529
* Build the object to hand to `instrument()` from `@flue/runtime`.
3630
*
@@ -45,20 +39,13 @@ const SKIPPED_PROVIDERS = [OPENAI_INTEGRATION_NAME, ANTHROPIC_AI_INTEGRATION_NAM
4539
* which fall back to the current client's `dataCollection.genAI` settings and are read per event.
4640
*/
4741
export function createFlueInstrumentation(options: FlueOptions = {}): FlueInstrumentation {
48-
// Flue drives the providers through `@earendil-works/pi-ai`, which bundles the `openai`,
49-
// `@anthropic-ai/sdk` and `@google/genai` clients. Left alone they instrument the same call this
50-
// reports as a turn, emitting a second `gen_ai.chat` beside ours.
51-
//
52-
// Applied on first use rather than here, for two reasons. Constructing the object proves nothing
53-
// — if `instrument()` rejects it, suppressing the provider integrations would leave the app with
54-
// no `gen_ai.chat` spans at all. And the registry is reset per client (`_setupIntegrations`
55-
// clears it, and Cloudflare calls `init()` per request), so a one-shot call at module scope is
56-
// wiped by the next `init()` and every later request double-reports.
57-
const skipProviders = (): void => {
58-
if (!SKIPPED_PROVIDERS.every(provider => _INTERNAL_shouldSkipAiProviderWrapping(provider))) {
59-
_INTERNAL_skipAiProviderWrapping(SKIPPED_PROVIDERS);
60-
}
61-
};
42+
// Flue drives the providers through `@earendil-works/pi-ai`. The provider integrations are skipped
43+
// on first use rather than here, for two reasons. Constructing the object proves nothing: if
44+
// `instrument()` rejects it, suppressing the provider integrations would leave the app with no
45+
// `gen_ai.chat` spans at all. And the registry is reset per client (`_setupIntegrations` clears it,
46+
// and Cloudflare calls `init()` per request), so a one-shot call at module scope is wiped by the
47+
// next `init()` and every later request double-reports.
48+
const skipProviders = skipPiAiProviderIntegrations;
6249

6350
// Keyed by the agent operation's own id, which is what the observations carry. That keeps
6451
// concurrent runs apart and gives a delegated subagent its own span: Flue nests a second `agent`

‎packages/server-utils/src/ai/flue/types.ts‎

Lines changed: 5 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -6,21 +6,7 @@
66
* `FlueExecutionContext` as of `@flue/runtime` 2.x.
77
*/
88

9-
/** Token counts and Flue-computed costs on a settled turn. */
10-
export interface FlueUsage {
11-
input?: number;
12-
output?: number;
13-
cacheRead?: number;
14-
cacheWrite?: number;
15-
totalTokens?: number;
16-
cost?: {
17-
input?: number;
18-
output?: number;
19-
cacheRead?: number;
20-
cacheWrite?: number;
21-
total?: number;
22-
};
23-
}
9+
import type { PiAiUsage } from '../pi-ai/usage';
2410

2511
/** Mirrors `ModelRequestInfo`. */
2612
export interface FlueModelRequestInfo {
@@ -51,7 +37,8 @@ export interface FlueModelResponse {
5137
responseId?: string;
5238
responseModel?: string;
5339
output?: unknown;
54-
usage?: FlueUsage;
40+
/** pi-ai's usage, which Flue passes through with its own cost figures. */
41+
usage?: PiAiUsage;
5542
finishReason?: string;
5643
}
5744

@@ -88,6 +75,8 @@ export interface FlueObservation {
8875
request?: FlueModelRequest;
8976
args?: unknown;
9077
result?: unknown;
78+
/** The tool result as the model sees it, on successful `tool` events. */
79+
effectiveResult?: unknown;
9180
response?: FlueModelResponse;
9281
}
9382

‎packages/server-utils/src/ai/flue/utils.ts‎

Lines changed: 44 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
import type { LRUMap, Span } from '@sentry/core';
22
import {
33
captureException,
4+
isObjectLike,
45
SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN,
56
SPAN_STATUS_ERROR,
67
startInactiveSpan,
@@ -9,11 +10,6 @@ import {
910
} from '@sentry/core';
1011
import {
1112
GEN_AI_CONVERSATION_ID,
12-
GEN_AI_COST_CACHE_CREATION_INPUT_TOKENS,
13-
GEN_AI_COST_CACHE_READ_INPUT_TOKENS,
14-
GEN_AI_COST_INPUT_TOKENS,
15-
GEN_AI_COST_OUTPUT_TOKENS,
16-
GEN_AI_COST_TOTAL_TOKENS,
1713
GEN_AI_INPUT_MESSAGES,
1814
GEN_AI_OPERATION_NAME,
1915
GEN_AI_OUTPUT_MESSAGES,
@@ -30,17 +26,19 @@ import {
3026
GEN_AI_TOOL_CALL_RESULT,
3127
GEN_AI_TOOL_DEFINITIONS,
3228
GEN_AI_TOOL_NAME,
33-
GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS,
34-
GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS,
35-
GEN_AI_USAGE_INPUT_TOKENS,
36-
GEN_AI_USAGE_OUTPUT_TOKENS,
37-
GEN_AI_USAGE_TOTAL_TOKENS,
3829
SERVER_ADDRESS,
3930
SERVER_PORT,
4031
} from '@sentry/conventions/attributes';
4132
import { getGenAiSpanOp } from '../core/utils';
33+
import {
34+
piAiAssistantMessageToGenAiMessage,
35+
piAiContentToString,
36+
piAiFinishReason,
37+
piAiMessagesToGenAiMessages,
38+
} from '../pi-ai/messages';
39+
import { setPiAiUsageAttributes } from '../pi-ai/usage';
4240
import { FLUE_ORIGIN, MAX_TRACKED_FLUE_SPANS } from './constants';
43-
import type { FlueErrorInfo, FlueModelRequestInfo, FlueObservation, FlueUsage } from './types';
41+
import type { FlueErrorInfo, FlueModelRequestInfo, FlueObservation } from './types';
4442

4543
/**
4644
* Flue persists the incoming W3C `traceparent` at admission and replays it on the agent operation.
@@ -163,62 +161,32 @@ export function endTurnSpan(observation: FlueObservation, turnSpans: SpanTracker
163161

164162
setRequestAttributes(span, observation.request);
165163

166-
const { responseId, finishReason } = observation.response ?? {};
164+
const responseId = observation.response?.responseId;
167165
if (responseId) {
168166
span.setAttribute(GEN_AI_RESPONSE_ID, responseId);
169167
}
168+
const finishReason = piAiFinishReason(observation.response?.finishReason);
170169
if (finishReason) {
171170
// Serialized, not a raw array: the conventions declare this attribute's value type as `string`,
172171
// and that is what `ai/core`, Mastra, OpenAI and Vercel AI all write.
173172
span.setAttribute(GEN_AI_RESPONSE_FINISH_REASONS, stringify([finishReason]));
174173
}
175174

176-
const output = observation.response?.output;
177-
if (recordOutputs && output !== undefined) {
178-
span.setAttribute(GEN_AI_OUTPUT_MESSAGES, stringify(output));
175+
const output = recordOutputs
176+
? piAiAssistantMessageToGenAiMessage(observation.response?.output, finishReason)
177+
: undefined;
178+
if (output) {
179+
span.setAttribute(GEN_AI_OUTPUT_MESSAGES, stringify([output]));
179180
}
180181

181-
setUsageAttributes(span, observation.response?.usage, observation.isError);
182+
setPiAiUsageAttributes(span, observation.response?.usage, observation.isError);
182183

183184
if (observation.isError) {
184185
span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' });
185186
}
186187
span.end();
187188
}
188189

189-
/**
190-
* Flue reports token counts and its own computed costs on the same `usage` object, so both are set
191-
* here. The cost figures have no equivalent in the provider SDKs' own instrumentation.
192-
*/
193-
export function setUsageAttributes(span: Span, usage: FlueUsage | undefined, isError?: boolean): void {
194-
// A turn that failed before the provider billed anything reports every counter as 0. Writing
195-
// those is noise that reads as a real zero-cost call, so skip the block entirely.
196-
if (!usage || (isError && !usage.totalTokens)) {
197-
return;
198-
}
199-
200-
const attributes: Record<string, number> = {};
201-
const set = (key: string, value: number | undefined): void => {
202-
if (typeof value === 'number') {
203-
attributes[key] = value;
204-
}
205-
};
206-
207-
set(GEN_AI_USAGE_INPUT_TOKENS, usage.input);
208-
set(GEN_AI_USAGE_OUTPUT_TOKENS, usage.output);
209-
set(GEN_AI_USAGE_TOTAL_TOKENS, usage.totalTokens);
210-
set(GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS, usage.cacheRead);
211-
set(GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS, usage.cacheWrite);
212-
213-
set(GEN_AI_COST_INPUT_TOKENS, usage.cost?.input);
214-
set(GEN_AI_COST_OUTPUT_TOKENS, usage.cost?.output);
215-
set(GEN_AI_COST_TOTAL_TOKENS, usage.cost?.total);
216-
set(GEN_AI_COST_CACHE_READ_INPUT_TOKENS, usage.cost?.cacheRead);
217-
set(GEN_AI_COST_CACHE_CREATION_INPUT_TOKENS, usage.cost?.cacheWrite);
218-
219-
span.setAttributes(attributes);
220-
}
221-
222190
/**
223191
* Tool spans hang off the agent invocation rather than the turn, matching how Flue's own
224192
* OpenTelemetry adapter projects them: siblings of `chat`, correlated to model output by tool call
@@ -257,8 +225,11 @@ export function endToolSpan(observation: FlueObservation, toolSpans: SpanTracker
257225
}
258226
toolSpans.remove(toolCallId);
259227

260-
if (recordOutputs && observation.result !== undefined) {
261-
span.setAttribute(GEN_AI_TOOL_CALL_RESULT, stringify(observation.result));
228+
if (recordOutputs) {
229+
const result = toolResultForModel(observation);
230+
if (result !== undefined) {
231+
span.setAttribute(GEN_AI_TOOL_CALL_RESULT, result);
232+
}
262233
}
263234

264235
if (observation.isError) {
@@ -268,6 +239,25 @@ export function endToolSpan(observation: FlueObservation, toolSpans: SpanTracker
268239
span.end();
269240
}
270241

242+
/**
243+
* The result the model receives: Flue's `effectiveResult` when it has one, else the content blocks
244+
* of its harness-level `result` (`{ content, details }`, whose `details` payload is tool-specific).
245+
*/
246+
function toolResultForModel(observation: FlueObservation): string | undefined {
247+
const { effectiveResult } = observation;
248+
if (effectiveResult !== undefined) {
249+
if (typeof effectiveResult === 'string') {
250+
return effectiveResult;
251+
}
252+
return Array.isArray(effectiveResult) ? piAiContentToString(effectiveResult) : stringify(effectiveResult);
253+
}
254+
const { result } = observation;
255+
if (isObjectLike(result) && Array.isArray(result.content)) {
256+
return piAiContentToString(result.content);
257+
}
258+
return result === undefined ? undefined : stringify(result);
259+
}
260+
271261
/**
272262
* `turn_request` is the only event carrying the request's content — the settled `turn` reports
273263
* metadata alone — so input messages, system prompt and tool definitions are read from it.
@@ -283,8 +273,9 @@ export function recordRequestContent(observation: FlueObservation, turnSpans: Sp
283273
if (input.systemPrompt) {
284274
span.setAttribute(GEN_AI_SYSTEM_INSTRUCTIONS, input.systemPrompt);
285275
}
286-
if (input.messages) {
287-
span.setAttribute(GEN_AI_INPUT_MESSAGES, stringify(input.messages));
276+
const messages = piAiMessagesToGenAiMessages(input.messages);
277+
if (messages.length) {
278+
span.setAttribute(GEN_AI_INPUT_MESSAGES, stringify(messages));
288279
}
289280
if (input.tools?.length) {
290281
span.setAttribute(GEN_AI_TOOL_DEFINITIONS, stringify(input.tools));

0 commit comments

Comments
 (0)