mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-19 16:55:02 +02:00
fix(core): Trace verification helpers and provider attempts (#40936)
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
967591041e
commit
f73dc337f8
|
|
@ -0,0 +1,149 @@
|
|||
import type * as AiImport from 'ai';
|
||||
|
||||
import type { BuiltTelemetry } from '../../types/telemetry';
|
||||
import { AgentRuntime } from '../loop/agent-runtime';
|
||||
import {
|
||||
createObservationLogObserveFn,
|
||||
createObservationLogReflectFn,
|
||||
} from '../memory/observation-log-defaults';
|
||||
import { AgentEventBus } from '../state/event-bus';
|
||||
|
||||
// The provider stub keeps the transport it is given, so each test can call the
|
||||
// model's fetch and see whether a request span is recorded.
|
||||
vi.mock('@ai-sdk/anthropic', () => ({
|
||||
createAnthropic: (opts?: { fetch?: typeof globalThis.fetch }) => (model: string) => ({
|
||||
provider: 'anthropic',
|
||||
modelId: model,
|
||||
specificationVersion: 'v3',
|
||||
supportedUrls: {},
|
||||
fetch: opts?.fetch,
|
||||
}),
|
||||
}));
|
||||
|
||||
const { mockGenerateText } = vi.hoisted(() => ({
|
||||
mockGenerateText: vi.fn<(call: Record<string, unknown>) => Promise<Record<string, unknown>>>(),
|
||||
}));
|
||||
|
||||
vi.mock('ai', async () => {
|
||||
const actual = await vi.importActual<typeof AiImport>('ai');
|
||||
return {
|
||||
...actual,
|
||||
generateText: async (call: Record<string, unknown>) => {
|
||||
const result = await mockGenerateText(call);
|
||||
return { ...result, finalStep: { providerMetadata: undefined } };
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
const MODEL = 'anthropic/claude-sonnet-4-5';
|
||||
|
||||
function tracedTelemetry() {
|
||||
const spans: Array<{ name: string; attributes: Record<string, unknown> }> = [];
|
||||
const telemetry: BuiltTelemetry = {
|
||||
enabled: true,
|
||||
traceModelRequests: true,
|
||||
functionId: 'wiring-test',
|
||||
recordInputs: false,
|
||||
recordOutputs: false,
|
||||
integrations: [],
|
||||
tracer: {
|
||||
startSpan(name: string, options: { attributes: Record<string, unknown> }) {
|
||||
spans.push({ name, attributes: options.attributes });
|
||||
return { setAttributes() {}, setStatus() {}, end() {} };
|
||||
},
|
||||
},
|
||||
};
|
||||
return { spans, telemetry };
|
||||
}
|
||||
|
||||
/** Calls the transport of the model passed to the last generateText call. */
|
||||
async function sendThroughLastModel() {
|
||||
const call = mockGenerateText.mock.calls.at(-1)?.[0];
|
||||
const model = call?.model as { fetch?: typeof globalThis.fetch } | undefined;
|
||||
if (!model?.fetch) throw new Error('model has no transport');
|
||||
await model.fetch('https://api.anthropic.test/v1/messages', { method: 'POST' });
|
||||
}
|
||||
|
||||
const requestSpans = (spans: Array<{ name: string }>) =>
|
||||
spans.filter((span) => span.name === 'model.provider.request');
|
||||
|
||||
describe('model request tracing wiring', () => {
|
||||
beforeEach(() => {
|
||||
mockGenerateText.mockReset();
|
||||
vi.stubGlobal(
|
||||
'fetch',
|
||||
vi.fn(async () => new Response('{}')),
|
||||
);
|
||||
});
|
||||
afterEach(() => vi.unstubAllGlobals());
|
||||
|
||||
it('an agent run sends model requests through the traced transport', async () => {
|
||||
mockGenerateText.mockResolvedValue({
|
||||
finishReason: 'stop',
|
||||
usage: { inputTokens: 1, outputTokens: 1, totalTokens: 2 },
|
||||
response: { messages: [{ role: 'assistant', content: [{ type: 'text', text: 'OK' }] }] },
|
||||
responseMessages: [{ role: 'assistant', content: [{ type: 'text', text: 'OK' }] }],
|
||||
toolCalls: [],
|
||||
});
|
||||
const modelFetch = vi.fn(async () => new Response('{}'));
|
||||
const { spans, telemetry } = tracedTelemetry();
|
||||
const runtime = new AgentRuntime({
|
||||
name: 'wiring-test',
|
||||
model: MODEL,
|
||||
instructions: 'test',
|
||||
eventBus: new AgentEventBus(),
|
||||
telemetry,
|
||||
modelFetch,
|
||||
});
|
||||
|
||||
await runtime.generate('hello');
|
||||
await sendThroughLastModel();
|
||||
|
||||
expect(modelFetch).toHaveBeenCalledTimes(1);
|
||||
expect(requestSpans(spans)).toEqual([
|
||||
expect.objectContaining({
|
||||
attributes: expect.objectContaining({
|
||||
'gen_ai.system': 'anthropic',
|
||||
'gen_ai.request.model': 'claude-sonnet-4-5',
|
||||
}),
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it('the memory observer sends model requests through the traced transport', async () => {
|
||||
mockGenerateText.mockResolvedValue({ text: '' });
|
||||
const { spans, telemetry } = tracedTelemetry();
|
||||
|
||||
await createObservationLogObserveFn(MODEL)({
|
||||
observationScopeId: 'thread-1',
|
||||
now: new Date('2026-05-12T14:30:00.000Z'),
|
||||
deltaMessages: [],
|
||||
transcript: '',
|
||||
transcriptTokenCount: 0,
|
||||
observationLogTail: [],
|
||||
renderedObservationLogTail: null,
|
||||
telemetry,
|
||||
});
|
||||
await sendThroughLastModel();
|
||||
|
||||
expect(requestSpans(spans)).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('the memory reflector sends model requests through the traced transport', async () => {
|
||||
mockGenerateText.mockResolvedValue({ text: '{"drop":[],"merge":[]}' });
|
||||
const { spans, telemetry } = tracedTelemetry();
|
||||
|
||||
await createObservationLogReflectFn(MODEL)({
|
||||
observationScopeId: 'thread-1',
|
||||
now: new Date('2026-05-12T14:30:00.000Z'),
|
||||
activeObservationLog: [],
|
||||
renderedObservationLog: '',
|
||||
tokenCount: 10,
|
||||
tokenBudget: 12_000,
|
||||
telemetry,
|
||||
});
|
||||
await sendThroughLastModel();
|
||||
|
||||
expect(requestSpans(spans)).toHaveLength(1);
|
||||
});
|
||||
});
|
||||
|
|
@ -51,6 +51,13 @@ async function runTitle(
|
|||
threadId: THREAD_ID,
|
||||
resourceId: RESOURCE_ID,
|
||||
titleConfig: { model: MODEL_ID },
|
||||
telemetry: {
|
||||
enabled: true,
|
||||
recordInputs: false,
|
||||
recordOutputs: false,
|
||||
integrations: [],
|
||||
functionId: 'helper-title',
|
||||
},
|
||||
agentModel: MODEL_ID,
|
||||
turnDelta: [
|
||||
{
|
||||
|
|
@ -83,6 +90,11 @@ describe('generateThreadTitle side-call cost', () => {
|
|||
|
||||
const reports = await runTitle(memory);
|
||||
|
||||
expect(mockGenerateText).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
telemetry: expect.objectContaining({ isEnabled: true, functionId: 'helper-title' }),
|
||||
}),
|
||||
);
|
||||
expect(reports).toHaveLength(1);
|
||||
const report = reports[0];
|
||||
expect(report.task).toBe('title');
|
||||
|
|
|
|||
|
|
@ -773,6 +773,7 @@ export class AgentRuntime {
|
|||
threadId: options.persistence.threadId,
|
||||
resourceId: options.persistence.resourceId,
|
||||
titleConfig: this.config.titleGeneration,
|
||||
telemetry: this.telemetry.resolve(options),
|
||||
agentModel: this.config.model,
|
||||
modelFetch: this.config.modelFetch,
|
||||
turnDelta: list.turnDelta(),
|
||||
|
|
|
|||
|
|
@ -85,7 +85,11 @@ export class RuntimeContextBuilder {
|
|||
): StaticLoopContext {
|
||||
const ai = loadAi();
|
||||
const aiProviderTools = toAiSdkProviderTools(this.config.providerTools);
|
||||
const model = createModel(this.config.model, this.config.modelFetch);
|
||||
const model = createModel(
|
||||
this.config.model,
|
||||
this.config.modelFetch,
|
||||
this.config.telemetry ?? execOptions?.telemetry,
|
||||
);
|
||||
const outputSchema = this.config.structuredOutput;
|
||||
const isRawJsonSchemaOutput = outputSchema !== undefined && !isZodSchema(outputSchema);
|
||||
const providerOptions = this.relaxStrictJsonSchemaIfNeeded(
|
||||
|
|
|
|||
|
|
@ -169,7 +169,7 @@ export function createObservationLogObserveFn(
|
|||
): ObservationLogObserveFn {
|
||||
return async (input) => {
|
||||
const { text, usage, finalStep } = await loadAi().generateText({
|
||||
model: createModel(model),
|
||||
model: createModel(model, undefined, input.telemetry),
|
||||
instructions: options.observerPrompt ?? DEFAULT_OBSERVATION_LOG_OBSERVER_PROMPT,
|
||||
prompt: buildObservationLogObserverPrompt(input),
|
||||
maxRetries: options.maxRetries ?? DEFAULT_OBSERVATION_LOG_OBSERVER_MAX_RETRIES,
|
||||
|
|
@ -337,7 +337,7 @@ export function createObservationLogReflectFn(
|
|||
})),
|
||||
);
|
||||
const { text, usage, finalStep } = await loadAi().generateText({
|
||||
model: createModel(model),
|
||||
model: createModel(model, undefined, input.telemetry),
|
||||
instructions: options.reflectorPrompt ?? DEFAULT_OBSERVATION_LOG_REFLECTOR_PROMPT,
|
||||
prompt: buildObservationLogReflectorPrompt({ ...input, renderedObservationLog }),
|
||||
...buildAiSdkTelemetry(input.telemetry, { functionSuffix: 'memory-reflector' }),
|
||||
|
|
|
|||
|
|
@ -161,7 +161,7 @@ export async function generateTitleFromMessage(
|
|||
export async function generateTitleAndEmojiFromMessage(
|
||||
model: LanguageModel,
|
||||
userMessage: string,
|
||||
opts?: { instructions?: string; executionCounter?: AgentExecutionCounter },
|
||||
opts?: GenerateTitleFromMessageOptions,
|
||||
): Promise<{ title: string | null; emoji?: string; usage?: TokenUsage } | null> {
|
||||
const trimmed = userMessage.trim();
|
||||
if (!trimmed) return null;
|
||||
|
|
@ -180,6 +180,7 @@ ${trimmed}
|
|||
model,
|
||||
instructions: opts?.instructions ?? DEFAULT_TITLE_AND_EMOJI_INSTRUCTIONS,
|
||||
messages: [{ role: 'user', content: wrappedMessage }],
|
||||
...buildAiSdkTelemetry(opts?.telemetry, { fallbackFunctionId: 'title-generation' }),
|
||||
});
|
||||
incrementTokenCountFromUsage(opts?.executionCounter, result.usage);
|
||||
const usage = toTokenUsage(result.usage, result.finalStep.providerMetadata);
|
||||
|
|
@ -231,6 +232,7 @@ export async function generateThreadTitle(opts: {
|
|||
agentModel: ModelConfig;
|
||||
/** Proxy/SSRF-aware `fetch` for the title model call (see {@link AgentRuntimeConfig.modelFetch}). */
|
||||
modelFetch?: FetchFn;
|
||||
telemetry?: BuiltTelemetry;
|
||||
/** Messages from the current turn, used to find the first user message. */
|
||||
turnDelta: AgentDbMessage[];
|
||||
executionCounter?: AgentExecutionCounter;
|
||||
|
|
@ -256,9 +258,10 @@ export async function generateThreadTitle(opts: {
|
|||
if (!userText) return;
|
||||
|
||||
const titleModelId = opts.titleConfig.model ?? opts.agentModel;
|
||||
const titleModel = createModel(titleModelId, opts.modelFetch);
|
||||
const titleModel = createModel(titleModelId, opts.modelFetch, opts.telemetry);
|
||||
const generated = await generateTitleAndEmojiFromMessage(titleModel, userText, {
|
||||
instructions: opts.titleConfig.instructions,
|
||||
telemetry: opts.telemetry,
|
||||
executionCounter: opts.executionCounter,
|
||||
});
|
||||
if (!generated) return;
|
||||
|
|
|
|||
|
|
@ -20,6 +20,8 @@ import {
|
|||
type ProviderId,
|
||||
type ProviderCredentials,
|
||||
} from './provider-credentials';
|
||||
import type { BuiltTelemetry } from '../../types/telemetry';
|
||||
import { traceModelRequests } from '../telemetry/model-request-tracing';
|
||||
import type { ModelConfig } from '../../types/sdk/agent';
|
||||
import { getModelIdString } from '../../utils/model';
|
||||
|
||||
|
|
@ -493,7 +495,11 @@ const SUPPORTED_PROVIDERS = Object.keys(LANGUAGE_PROVIDERS).join(', ');
|
|||
* Provider packages are loaded dynamically via require() so only the
|
||||
* provider needed at runtime must be installed.
|
||||
*/
|
||||
export function createModel(config: ModelConfig, fetch?: FetchFn): LanguageModel {
|
||||
export function createModel(
|
||||
config: ModelConfig,
|
||||
fetch?: FetchFn,
|
||||
telemetry?: BuiltTelemetry,
|
||||
): LanguageModel {
|
||||
if (isLanguageModel(config)) {
|
||||
return config;
|
||||
}
|
||||
|
|
@ -543,7 +549,10 @@ export function createModel(config: ModelConfig, fetch?: FetchFn): LanguageModel
|
|||
}
|
||||
|
||||
// Caller-injected transport wins; fall back to the ambient env-proxy resolver.
|
||||
const resolvedFetch = fetch ?? getProxyFetch();
|
||||
const transport = fetch ?? getProxyFetch();
|
||||
const resolvedFetch = telemetry?.traceModelRequests
|
||||
? traceModelRequests(transport ?? globalFetch, telemetry, provider, modelName)
|
||||
: transport;
|
||||
// Type cast: the registry guarantees the schema and builder are aligned per provider.
|
||||
return (entry.build as EntryBuilder<typeof provider>)(parsed.data, modelName, resolvedFetch);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,225 @@
|
|||
import type { BuiltTelemetry } from '../../../types/telemetry';
|
||||
import { traceModelRequests } from '../model-request-tracing';
|
||||
|
||||
function fixture() {
|
||||
const spans: Array<{
|
||||
name: string;
|
||||
attributes: Record<string, unknown>;
|
||||
status?: { code: number };
|
||||
ended: boolean;
|
||||
}> = [];
|
||||
const tracer = {
|
||||
startSpan(name: string, options: { attributes: Record<string, unknown> }) {
|
||||
const record = {
|
||||
name,
|
||||
attributes: options.attributes,
|
||||
ended: false,
|
||||
status: undefined as { code: number } | undefined,
|
||||
};
|
||||
spans.push(record);
|
||||
return {
|
||||
setAttributes(attributes: Record<string, unknown>) {
|
||||
Object.assign(record.attributes, attributes);
|
||||
},
|
||||
setStatus(status: { code: number }) {
|
||||
record.status = status;
|
||||
},
|
||||
end() {
|
||||
record.ended = true;
|
||||
},
|
||||
};
|
||||
},
|
||||
};
|
||||
const telemetry: BuiltTelemetry = {
|
||||
enabled: true,
|
||||
traceModelRequests: true,
|
||||
recordInputs: false,
|
||||
recordOutputs: false,
|
||||
integrations: [],
|
||||
isLangSmith: true,
|
||||
tracer,
|
||||
metadata: { thread_id: 'thread', evalRole: 'helper' },
|
||||
};
|
||||
return { spans, telemetry };
|
||||
}
|
||||
|
||||
it('records each HTTP attempt, with no payload or duplicate usage', async () => {
|
||||
const { spans, telemetry } = fixture();
|
||||
const delegate = vi
|
||||
.fn<typeof fetch>()
|
||||
.mockResolvedValueOnce(new Response('private error', { status: 500 }))
|
||||
.mockResolvedValueOnce(new Response('private output'));
|
||||
const fetcher = traceModelRequests(delegate, telemetry, 'custom', 'test-model');
|
||||
const init = {
|
||||
method: 'POST',
|
||||
headers: { Authorization: 'private key' },
|
||||
body: 'private prompt',
|
||||
};
|
||||
expect((await fetcher('https://private.url', init)).status).toBe(500);
|
||||
expect((await fetcher('https://private.url', init)).status).toBe(200);
|
||||
expect(delegate).toHaveBeenNthCalledWith(1, 'https://private.url', init);
|
||||
expect(spans.map((span) => span.status?.code)).toEqual([2, 1]);
|
||||
expect(spans.every((span) => span.ended)).toBe(true);
|
||||
expect(spans[0].attributes).toMatchObject({
|
||||
'langsmith.traceable': 'true',
|
||||
'langsmith.metadata.thread_id': 'thread',
|
||||
'langsmith.metadata.evalRole': 'helper',
|
||||
'gen_ai.system': 'custom',
|
||||
'gen_ai.request.model': 'test-model',
|
||||
'http.response.status_code': 500,
|
||||
});
|
||||
expect(JSON.stringify(spans)).not.toMatch(/private|usage|tokens/);
|
||||
});
|
||||
|
||||
it('preserves a transport exception without recording its message', async () => {
|
||||
const { spans, telemetry } = fixture();
|
||||
const failure = new Error('private error');
|
||||
const delegate = vi.fn<typeof fetch>().mockRejectedValue(failure);
|
||||
await expect(traceModelRequests(delegate, telemetry, 'custom', 'model')('url')).rejects.toBe(
|
||||
failure,
|
||||
);
|
||||
expect(spans[0]).toMatchObject({ ended: true, status: { code: 2 } });
|
||||
expect(JSON.stringify(spans)).not.toContain('private');
|
||||
});
|
||||
|
||||
it('keeps generic OpenTelemetry spans free of LangSmith attributes', async () => {
|
||||
const { spans, telemetry } = fixture();
|
||||
const delegate = vi.fn<typeof fetch>().mockResolvedValue(new Response('ok'));
|
||||
await traceModelRequests(
|
||||
delegate,
|
||||
{ ...telemetry, isLangSmith: false },
|
||||
'custom',
|
||||
'model',
|
||||
)('url');
|
||||
expect(spans).toEqual([
|
||||
{
|
||||
name: 'model.provider.request',
|
||||
attributes: {
|
||||
'gen_ai.system': 'custom',
|
||||
'gen_ai.request.model': 'model',
|
||||
'http.response.status_code': 200,
|
||||
},
|
||||
status: { code: 1 },
|
||||
ended: true,
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('records LangSmith request spans without optional metadata', async () => {
|
||||
const { spans, telemetry } = fixture();
|
||||
const delegate = vi.fn<typeof fetch>().mockResolvedValue(new Response('ok'));
|
||||
await traceModelRequests(
|
||||
delegate,
|
||||
{ ...telemetry, metadata: undefined },
|
||||
'custom',
|
||||
'model',
|
||||
)('url');
|
||||
expect(spans[0]).toEqual({
|
||||
name: 'model.provider.request',
|
||||
attributes: {
|
||||
'gen_ai.system': 'custom',
|
||||
'gen_ai.request.model': 'model',
|
||||
'langsmith.traceable': 'true',
|
||||
'langsmith.span.kind': 'chain',
|
||||
'http.response.status_code': 200,
|
||||
},
|
||||
status: { code: 1 },
|
||||
ended: true,
|
||||
});
|
||||
});
|
||||
|
||||
it('preserves metadata keys that already have the LangSmith prefix', async () => {
|
||||
const { spans, telemetry } = fixture();
|
||||
const delegate = vi.fn<typeof fetch>().mockResolvedValue(new Response('ok'));
|
||||
const metadata = { 'langsmith.metadata.thread_id': 'thread', evalRole: 'helper' };
|
||||
await traceModelRequests(delegate, { ...telemetry, metadata }, 'custom', 'model')('url');
|
||||
expect(spans[0].attributes).toMatchObject({
|
||||
'langsmith.metadata.thread_id': 'thread',
|
||||
'langsmith.metadata.evalRole': 'helper',
|
||||
});
|
||||
expect(Object.keys(spans[0].attributes)).not.toContain(
|
||||
'langsmith.metadata.langsmith.metadata.thread_id',
|
||||
);
|
||||
});
|
||||
|
||||
it('keeps the delegate when tracing is disabled or not requested', () => {
|
||||
const { telemetry } = fixture();
|
||||
const delegate = vi.fn<typeof fetch>();
|
||||
expect(traceModelRequests(delegate, undefined, 'custom', 'model')).toBe(delegate);
|
||||
expect(
|
||||
traceModelRequests(delegate, { ...telemetry, traceModelRequests: false }, 'custom', 'model'),
|
||||
).toBe(delegate);
|
||||
expect(traceModelRequests(delegate, { ...telemetry, enabled: false }, 'custom', 'model')).toBe(
|
||||
delegate,
|
||||
);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ name: 'missing', tracer: undefined },
|
||||
{ name: 'null', tracer: null },
|
||||
{ name: 'empty', tracer: {} },
|
||||
{ name: 'invalid', tracer: { startSpan: 'not callable' } },
|
||||
])('keeps the delegate when the tracer is $name', ({ tracer }) => {
|
||||
const { telemetry } = fixture();
|
||||
const delegate = vi.fn<typeof fetch>();
|
||||
expect(traceModelRequests(delegate, { ...telemetry, tracer }, 'custom', 'model')).toBe(delegate);
|
||||
});
|
||||
|
||||
it('does not change provider responses when telemetry fails', async () => {
|
||||
const { telemetry } = fixture();
|
||||
const response = new Response('ok');
|
||||
const delegate = vi.fn<typeof fetch>().mockResolvedValue(response);
|
||||
const broken = {
|
||||
...telemetry,
|
||||
tracer: {
|
||||
startSpan() {
|
||||
throw new Error('tracer failed');
|
||||
},
|
||||
},
|
||||
};
|
||||
await expect(traceModelRequests(delegate, broken, 'custom', 'model')('url')).resolves.toBe(
|
||||
response,
|
||||
);
|
||||
expect(delegate).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it.each(['setAttributes', 'setStatus', 'end'] as const)(
|
||||
'preserves the provider response when span.%s throws',
|
||||
async (method) => {
|
||||
const { telemetry } = fixture();
|
||||
const span = { setAttributes: vi.fn(), setStatus: vi.fn(), end: vi.fn() };
|
||||
span[method].mockImplementation(() => {
|
||||
throw new Error('span failed');
|
||||
});
|
||||
const tracer = { startSpan: vi.fn().mockReturnValue(span) };
|
||||
const response = new Response('private output');
|
||||
const delegate = vi.fn<typeof fetch>().mockResolvedValue(response);
|
||||
await expect(
|
||||
traceModelRequests(delegate, { ...telemetry, tracer }, 'custom', 'model')('url'),
|
||||
).resolves.toBe(response);
|
||||
expect(span[method]).toHaveBeenCalledOnce();
|
||||
expect(span.end).toHaveBeenCalledOnce();
|
||||
expect(delegate).toHaveBeenCalledOnce();
|
||||
},
|
||||
);
|
||||
|
||||
it.each(['setStatus', 'end'] as const)(
|
||||
'preserves the transport error when span.%s throws',
|
||||
async (method) => {
|
||||
const { telemetry } = fixture();
|
||||
const span = { setAttributes: vi.fn(), setStatus: vi.fn(), end: vi.fn() };
|
||||
span[method].mockImplementation(() => {
|
||||
throw new Error('span failed');
|
||||
});
|
||||
const tracer = { startSpan: vi.fn().mockReturnValue(span) };
|
||||
const failure = new Error('private transport error');
|
||||
const delegate = vi.fn<typeof fetch>().mockRejectedValue(failure);
|
||||
await expect(
|
||||
traceModelRequests(delegate, { ...telemetry, tracer }, 'custom', 'model')('url'),
|
||||
).rejects.toBe(failure);
|
||||
expect(span[method]).toHaveBeenCalledOnce();
|
||||
expect(span.end).toHaveBeenCalledOnce();
|
||||
expect(span.setAttributes).not.toHaveBeenCalled();
|
||||
expect(delegate).toHaveBeenCalledOnce();
|
||||
},
|
||||
);
|
||||
|
|
@ -0,0 +1,66 @@
|
|||
import type { Span, Tracer } from '@opentelemetry/api';
|
||||
|
||||
import type { AttributeValue, BuiltTelemetry } from '../../types/telemetry';
|
||||
|
||||
function isTracer(value: unknown): value is Tracer {
|
||||
return (
|
||||
value !== null &&
|
||||
typeof value === 'object' &&
|
||||
typeof Reflect.get(value, 'startSpan') === 'function'
|
||||
);
|
||||
}
|
||||
|
||||
/** Record headers-time per HTTP attempt. Generation spans alone own token usage. */
|
||||
export function traceModelRequests(
|
||||
delegate: typeof globalThis.fetch,
|
||||
telemetry: BuiltTelemetry | undefined,
|
||||
provider: string,
|
||||
model: string,
|
||||
): typeof globalThis.fetch {
|
||||
if (!telemetry?.enabled || !telemetry.traceModelRequests || !isTracer(telemetry.tracer))
|
||||
return delegate;
|
||||
const tracer = telemetry.tracer;
|
||||
return async (input, init) => {
|
||||
let span: Span | undefined;
|
||||
try {
|
||||
const attributes: Record<string, AttributeValue> = {
|
||||
'gen_ai.system': provider,
|
||||
'gen_ai.request.model': model,
|
||||
};
|
||||
if (telemetry.isLangSmith) {
|
||||
attributes['langsmith.traceable'] = 'true';
|
||||
attributes['langsmith.span.kind'] = 'chain';
|
||||
for (const [key, value] of Object.entries(telemetry.metadata ?? {})) {
|
||||
attributes[key.startsWith('langsmith.metadata.') ? key : `langsmith.metadata.${key}`] =
|
||||
value;
|
||||
}
|
||||
}
|
||||
span = tracer.startSpan('model.provider.request', { attributes });
|
||||
} catch {
|
||||
/* Telemetry must not prevent a provider request. */
|
||||
}
|
||||
try {
|
||||
const response = await delegate(input, init);
|
||||
try {
|
||||
span?.setAttributes({ 'http.response.status_code': response.status });
|
||||
span?.setStatus({ code: response.ok ? 1 : 2 });
|
||||
} catch {
|
||||
/* Preserve the provider response. */
|
||||
}
|
||||
return response;
|
||||
} catch (error) {
|
||||
try {
|
||||
span?.setStatus({ code: 2 });
|
||||
} catch {
|
||||
/* No error bodies in spans. */
|
||||
}
|
||||
throw error;
|
||||
} finally {
|
||||
try {
|
||||
span?.end();
|
||||
} catch {
|
||||
/* Export errors do not change model results. */
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
|
@ -28,6 +28,8 @@ export interface OpaqueTracerProvider {
|
|||
|
||||
export interface BuiltTelemetry {
|
||||
readonly enabled: boolean;
|
||||
/** Opt in to metadata-only HTTP attempt spans; token usage stays on generation spans. */
|
||||
readonly traceModelRequests?: boolean;
|
||||
readonly functionId?: string;
|
||||
readonly metadata?: Record<string, AttributeValue>;
|
||||
readonly recordInputs: boolean;
|
||||
|
|
|
|||
|
|
@ -332,3 +332,25 @@ search:
|
|||
- html
|
||||
- json # required for Instance AI
|
||||
```
|
||||
|
||||
## Verification helper and provider request traces
|
||||
|
||||
When product tracing is enabled, the workflow destructiveness classifier and
|
||||
simulation-fixture generator reuse the current product trace and provider.
|
||||
Their spans carry `evalRole=helper` and the helper name in `agent_role`. Their
|
||||
prompt and output recording is disabled. Calls outside an active product trace
|
||||
keep their existing behavior.
|
||||
|
||||
Product model calls also emit `model.provider.request` spans per HTTP attempt.
|
||||
These report provider/model identity, HTTP status and time to response headers.
|
||||
They preserve the configured transport and parent span. They contain no request
|
||||
URL, headers, body, response body, token usage, or exception text. HTTP success
|
||||
does not mean a stream completed: the parent generation span reports stream
|
||||
errors, tokens and timing. Generation spans own usage, so request spans must not
|
||||
be summed into generation/token totals. Prebuilt external LanguageModel objects
|
||||
bypass the SDK transport and must supply their own request tracing.
|
||||
|
||||
The agents SDK option `BuiltTelemetry.traceModelRequests` is opt-in. Instance AI
|
||||
enables it for product calls; external harnesses with their own fetch tracing
|
||||
leave it unset to avoid duplicate attempt spans. The existing trace enablement,
|
||||
redaction and provider lifetime rules still apply.
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ import { executeTool } from '../../__tests__/tool-test-utils';
|
|||
import { createToolRegistry } from '../../tool-registry';
|
||||
import { createAskUserTool } from '../../tools/shared/ask-user.tool';
|
||||
import {
|
||||
getCurrentHelperTelemetry,
|
||||
buildAgentTraceInputs,
|
||||
canEmitTraceOnlyChildRun,
|
||||
createInstanceAiTraceContext,
|
||||
|
|
@ -2191,6 +2192,35 @@ describe('createInstanceAiTraceContext', () => {
|
|||
).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
it('shares product telemetry with helpers only inside an active trace', async () => {
|
||||
expect(getCurrentHelperTelemetry('verification-helper')).toBeUndefined();
|
||||
const tracing = await createInstanceAiTraceContext({
|
||||
threadId: 'helper-thread',
|
||||
messageId: 'message',
|
||||
runId: 'run',
|
||||
userId: 'user',
|
||||
input: {},
|
||||
});
|
||||
expect(tracing).toBeDefined();
|
||||
await startForegroundActor(tracing!);
|
||||
await tracing!.withActiveSpan(tracing!.orchestratorRun, async () => {
|
||||
const helper = getCurrentHelperTelemetry('verification-helper');
|
||||
expect(helper).toMatchObject({
|
||||
enabled: true,
|
||||
rootAnchored: false,
|
||||
recordInputs: false,
|
||||
recordOutputs: false,
|
||||
metadata: {
|
||||
thread_id: 'helper-thread',
|
||||
evalRole: 'helper',
|
||||
agent_role: 'verification-helper',
|
||||
},
|
||||
});
|
||||
expect(helper?.provider).toBe(agentsMock.getProvider());
|
||||
});
|
||||
expect(getCurrentHelperTelemetry('verification-helper')).toBeUndefined();
|
||||
});
|
||||
|
||||
it('runs ad-hoc span callbacks when product span creation fails', async () => {
|
||||
const tracing = await createInstanceAiTraceContext({
|
||||
threadId: 'thread-span-start-failure',
|
||||
|
|
|
|||
|
|
@ -639,6 +639,25 @@ function getCurrentProductTrace():
|
|||
return productTraceStorage.getStore();
|
||||
}
|
||||
|
||||
/** Reuse the active product provider and parent for synchronous verification helpers. */
|
||||
export function getCurrentHelperTelemetry(name: string): BuiltTelemetry | undefined {
|
||||
const current = getCurrentProductTrace();
|
||||
if (!current || current.runtime.shutdown || !current.runtime.telemetry.enabled) return undefined;
|
||||
return {
|
||||
...current.runtime.telemetry,
|
||||
functionId: name,
|
||||
metadata: toTelemetryMetadata(current.currentRun.metadata, {
|
||||
evalRole: 'helper',
|
||||
agent_role: name,
|
||||
}),
|
||||
recordInputs: false,
|
||||
recordOutputs: false,
|
||||
runtimeRootSpanEnabled: true,
|
||||
rootAnchored: false,
|
||||
traceModelRequests: true,
|
||||
};
|
||||
}
|
||||
|
||||
function getActiveOtelContextWithSpan(expectedTraceId?: string): OtelContext | undefined {
|
||||
const activeContext = otelContext.active();
|
||||
const activeSpanContext = otelTrace.getSpan(activeContext)?.spanContext();
|
||||
|
|
@ -1801,6 +1820,7 @@ function createTelemetryFactory(options: {
|
|||
if (options.baseTelemetry) {
|
||||
return {
|
||||
...options.baseTelemetry,
|
||||
traceModelRequests: true,
|
||||
functionId,
|
||||
metadata,
|
||||
recordInputs: true,
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ type MockAgentInstance = {
|
|||
instructions: Mock;
|
||||
thinking: Mock;
|
||||
configuration: Mock;
|
||||
telemetry: Mock;
|
||||
};
|
||||
|
||||
const mockAgentInstances: MockAgentInstance[] = [];
|
||||
|
|
@ -15,6 +16,7 @@ vi.mock('@n8n/agents', () => ({
|
|||
this.instructions = vi.fn().mockReturnThis();
|
||||
this.thinking = vi.fn().mockReturnThis();
|
||||
this.configuration = vi.fn().mockReturnThis();
|
||||
this.telemetry = vi.fn().mockReturnThis();
|
||||
mockAgentInstances.push(this);
|
||||
}),
|
||||
Tool: vi.fn(),
|
||||
|
|
@ -70,6 +72,18 @@ describe('eval agent model config', () => {
|
|||
process.env = ORIGINAL_ENV;
|
||||
});
|
||||
|
||||
it('attaches caller telemetry to helper agents', () => {
|
||||
process.env.N8N_AI_ANTHROPIC_KEY = 'test';
|
||||
const telemetry = {
|
||||
enabled: true,
|
||||
recordInputs: false,
|
||||
recordOutputs: false,
|
||||
integrations: [],
|
||||
};
|
||||
createEvalAgent('helper', { instructions: 'test', telemetry });
|
||||
expect(mockAgentInstances[0]?.telemetry).toHaveBeenCalledWith(telemetry);
|
||||
});
|
||||
|
||||
it('keeps the legacy Anthropic key fallback for eval models', () => {
|
||||
process.env.N8N_AI_ANTHROPIC_KEY = 'legacy-anthropic-key';
|
||||
|
||||
|
|
|
|||
|
|
@ -1,11 +1,14 @@
|
|||
import type { MockedFunction } from 'vitest';
|
||||
import { z } from 'zod';
|
||||
|
||||
vi.mock('../../tracing/langsmith-tracing', () => ({ getCurrentHelperTelemetry: vi.fn() }));
|
||||
|
||||
vi.mock('../eval-agents', async () => {
|
||||
const actual: object = await vi.importActual('../eval-agents');
|
||||
return { ...actual, createEvalAgent: vi.fn(), extractText: vi.fn() };
|
||||
});
|
||||
|
||||
import { getCurrentHelperTelemetry } from '../../tracing/langsmith-tracing';
|
||||
import { createEvalAgent, extractText } from '../eval-agents';
|
||||
import { generateValidatedJson } from '../generate-validated-json';
|
||||
|
||||
|
|
@ -65,6 +68,18 @@ describe('generateValidatedJson', () => {
|
|||
if (!result.ok) expect(result.issues?.length).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
it('still calls the model, without telemetry, when tracing fails', async () => {
|
||||
vi.mocked(getCurrentHelperTelemetry).mockImplementationOnce(() => {
|
||||
throw new Error('tracing unavailable');
|
||||
});
|
||||
setupAgentMock('{"ok": true}');
|
||||
expect(await generate()).toEqual({ ok: true, data: { ok: true } });
|
||||
expect(mockCreateEvalAgent).toHaveBeenCalledWith(
|
||||
'test-agent',
|
||||
expect.objectContaining({ telemetry: undefined }),
|
||||
);
|
||||
});
|
||||
|
||||
it('reports generation_failed when the LLM call throws', async () => {
|
||||
const generateMock = vi.fn().mockRejectedValue(new Error('boom'));
|
||||
mockCreateEvalAgent.mockReturnValue({ generate: generateMock } as unknown as ReturnType<
|
||||
|
|
@ -115,3 +130,15 @@ describe('generateValidatedJson', () => {
|
|||
);
|
||||
});
|
||||
});
|
||||
|
||||
it('passes the current product trace to verification helpers', async () => {
|
||||
const telemetry = { enabled: true, recordInputs: false, recordOutputs: false, integrations: [] };
|
||||
vi.mocked(getCurrentHelperTelemetry).mockReturnValueOnce(telemetry);
|
||||
setupAgentMock('{"ok": true}');
|
||||
await generate();
|
||||
expect(getCurrentHelperTelemetry).toHaveBeenCalledWith('test-agent');
|
||||
expect(mockCreateEvalAgent).toHaveBeenCalledWith(
|
||||
'test-agent',
|
||||
expect.objectContaining({ telemetry }),
|
||||
);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import {
|
|||
Agent,
|
||||
Tool,
|
||||
type AnthropicThinkingEffort,
|
||||
type BuiltTelemetry,
|
||||
type ExecutionOptions,
|
||||
type GenerateResult,
|
||||
type ModelConfig,
|
||||
|
|
@ -244,6 +245,7 @@ export function createEvalAgent(
|
|||
cache?: boolean;
|
||||
/** Host-resolved model used when no eval model API key is configured in the environment. */
|
||||
fallbackModelConfig?: ModelConfig;
|
||||
telemetry?: BuiltTelemetry;
|
||||
/** Use `N8N_INSTANCE_AI_EVAL_JUDGE_MODEL` / `_JUDGE_EFFORT` instead of the default eval settings. */
|
||||
judge?: boolean;
|
||||
},
|
||||
|
|
@ -253,6 +255,7 @@ export function createEvalAgent(
|
|||
options.fallbackModelConfig,
|
||||
);
|
||||
const agent = new Agent(name).model(model);
|
||||
if (options.telemetry) agent.telemetry(options.telemetry);
|
||||
|
||||
if (options.cache) {
|
||||
agent.instructions(options.instructions, CACHE_PROVIDER_OPTS);
|
||||
|
|
|
|||
|
|
@ -33,13 +33,25 @@ export interface GenerateValidatedJsonOptions<T> {
|
|||
fallbackModelConfig?: ModelConfig;
|
||||
}
|
||||
|
||||
/** Tracing must never block a model call: when it fails, the call runs without spans. */
|
||||
async function helperTelemetry(agentName: string) {
|
||||
try {
|
||||
const { getCurrentHelperTelemetry } = await import('../tracing/langsmith-tracing.js');
|
||||
return getCurrentHelperTelemetry(agentName);
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
export async function generateValidatedJson<T>(
|
||||
agentName: string,
|
||||
options: GenerateValidatedJsonOptions<T>,
|
||||
): Promise<ValidatedJsonResult<T>> {
|
||||
const telemetry = await helperTelemetry(agentName);
|
||||
let text: string;
|
||||
try {
|
||||
const llm = createEvalAgent(agentName, {
|
||||
telemetry,
|
||||
model: options.model,
|
||||
instructions: options.instructions,
|
||||
fallbackModelConfig: options.fallbackModelConfig,
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user