diff --git a/package.json b/package.json index ac03e3d..8f41982 100644 --- a/package.json +++ b/package.json @@ -114,7 +114,8 @@ "@lavamoat/preinstall-always-fail": false, "@langfuse/otel>@opentelemetry/exporter-trace-otlp-http>@opentelemetry/otlp-transformer>protobufjs": false, "@opentelemetry/sdk-node>@opentelemetry/exporter-logs-otlp-grpc>@grpc/grpc-js>@grpc/proto-loader>protobufjs": false, - "eslint-plugin-import-x>unrs-resolver": false + "eslint-plugin-import-x>unrs-resolver": false, + "@langfuse/otel>@opentelemetry/otlp-exporter-base>@opentelemetry/otlp-transformer>protobufjs": false } } } diff --git a/src/telemetry/message-handler.test.ts b/src/telemetry/message-handler.test.ts index 47004ca..fb37ce7 100644 --- a/src/telemetry/message-handler.test.ts +++ b/src/telemetry/message-handler.test.ts @@ -109,7 +109,11 @@ describe('createMessageHandler', () => { }); expect(tracingMocks.traceSpan).toHaveBeenCalledWith( - expect.objectContaining({ sessionId: 'session-1', userId: 'user-1' }), + expect.objectContaining({ + sessionId: 'session-1', + userId: 'user-1', + traceName: 'agent-runner', + }), expect.any(Function), ); expect(tracingMocks.createSessionSpan).toHaveBeenCalledWith( @@ -661,7 +665,7 @@ describe('createMessageHandler', () => { expect(tracingMocks.createSessionSpan).not.toHaveBeenCalled(); }); - it('propagates custom traceName to createSessionSpan', () => { + it('propagates custom traceName to the root observation and trace context', () => { const handler = createHandler({ traceName: 'my-eval-trace' }); handler.handleMessage({ type: 'init', sessionId: 'session-1' }); @@ -671,6 +675,10 @@ describe('createMessageHandler', () => { expect.any(Object), false, ); + expect(tracingMocks.traceSpan).toHaveBeenCalledWith( + expect.objectContaining({ traceName: 'my-eval-trace' }), + expect.any(Function), + ); }); it('propagates metadata, tags, and version to traceSpan', () => { @@ -738,6 +746,7 @@ describe('createMessageHandler', () => { it('propagates trace attributes on generation spans', () => { const handler = createHandler({ + traceName: 'my-eval-trace', traceTags: ['eval'], traceVersion: '2.0.0', }); @@ -747,6 +756,7 @@ describe('createMessageHandler', () => { const generationTraceCall = tracingMocks.traceSpan.mock.calls[1]; expect(generationTraceCall?.[0]).toStrictEqual( expect.objectContaining({ + traceName: 'my-eval-trace', tags: ['eval'], version: '2.0.0', }), diff --git a/src/telemetry/message-handler.ts b/src/telemetry/message-handler.ts index 61eca3b..30175d3 100644 --- a/src/telemetry/message-handler.ts +++ b/src/telemetry/message-handler.ts @@ -46,7 +46,7 @@ export type MessageHandlerConfig = { */ initialSessionId: string | undefined; /** - * Name for the root Langfuse trace observation. + * Name for the Langfuse trace and its root observation. */ traceName?: string | undefined; /** @@ -166,6 +166,7 @@ export function createMessageHandler( config: MessageHandlerConfig, ): MessageHandler { const scrubCredential = createCredentialScrubber(); + const traceName = config.traceName ?? 'agent-runner'; /** * Applies credential redaction followed by the caller's redactor. * @@ -244,13 +245,14 @@ export function createMessageHandler( { sessionId: state.langfuseSessionId, userId: config.userId, + traceName, metadata, tags: config.traceTags, version: config.traceVersion, }, () => { const session = createSessionSpan( - config.traceName ?? 'agent-runner', + traceName, config.redact ? config.prompt : scrubText(config.prompt), { model: message.model ?? config.model, @@ -284,6 +286,7 @@ export function createMessageHandler( { sessionId: state.langfuseSessionId, userId: config.userId, + traceName, metadata: config.traceMetadata, tags: config.traceTags, version: config.traceVersion, diff --git a/src/telemetry/trace-name.test.ts b/src/telemetry/trace-name.test.ts new file mode 100644 index 0000000..a45784e --- /dev/null +++ b/src/telemetry/trace-name.test.ts @@ -0,0 +1,104 @@ +import { LangfuseSpanProcessor } from '@langfuse/otel'; +import { NodeSDK } from '@opentelemetry/sdk-node'; +import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'; + +import { createMessageHandler } from './message-handler.js'; +import { setLangfuseProcessor } from './tracing.js'; + +type CapturedSpan = { + name: string; + traceId: string; + attributes: Record; +}; + +describe('Langfuse trace-name attributes', () => { + const spans: CapturedSpan[] = []; + const processor = new LangfuseSpanProcessor({ + exporter: { + export: (batch, callback): void => { + for (const span of batch) { + spans.push({ + name: span.name, + traceId: span.spanContext().traceId, + attributes: { ...span.attributes }, + }); + } + callback({ code: 0 }); + }, + shutdown: async (): Promise => undefined, + }, + }); + const sdk = new NodeSDK({ + autoDetectResources: false, + spanProcessors: [processor], + metricReaders: [], + logRecordProcessors: [], + }); + + beforeAll(() => { + sdk.start(); + setLangfuseProcessor(processor); + }); + + beforeEach(() => { + spans.length = 0; + }); + + afterAll(async () => { + setLangfuseProcessor(undefined); + await sdk.shutdown(); + }); + + it('emits a trace name on every observation without leaking it between runs in one session', async () => { + for (const traceName of ['pr-visual-validation.plan', undefined]) { + const handler = createMessageHandler({ + prompt: 'Inspect a fixture.', + model: 'fixture-model', + maxTurns: 1, + redact: false, + userId: 'fixture-user', + initialSessionId: 'shared-workflow-session', + traceName, + }); + handler.handleMessage({ type: 'init', sessionId: 'sdk-session' }); + handler.handleMessage({ + type: 'generation', + model: 'fixture-model', + text: '', + toolCalls: [ + { id: 'read-1', name: 'Read', input: { file_path: '/fixture.txt' } }, + ], + usage: { inputTokens: 1, outputTokens: 1 }, + stopReason: 'tool_use', + }); + handler.handleMessage({ + type: 'tool_result', + toolUseId: 'read-1', + content: 'Fixture contents.', + isError: false, + }); + handler.handleMessage({ type: 'result', success: true, result: 'Done.' }); + handler.finalizeSessionSpan(); + await processor.forceFlush(); + + const runSpans = spans.filter( + (span) => span.traceId === handler.getState().traceId, + ); + const expectedName = traceName ?? 'agent-runner'; + expect(runSpans.map((span) => span.name)).toStrictEqual([ + 'fixture-model', + 'Read: fixture.txt', + expectedName, + ]); + for (const span of runSpans) { + expect(span.attributes).toMatchObject({ + 'langfuse.trace.name': expectedName, + 'session.id': 'shared-workflow-session', + }); + } + } + + expect(spans).toHaveLength(6); + expect(new Set(spans.map((span) => span.traceId)).size).toBe(2); + }); +}); diff --git a/src/telemetry/tracing.test.ts b/src/telemetry/tracing.test.ts index 71c9bdd..0f41ae9 100644 --- a/src/telemetry/tracing.test.ts +++ b/src/telemetry/tracing.test.ts @@ -101,7 +101,7 @@ describe('tracing helpers', () => { expect(callback).toHaveBeenCalledOnce(); }); - it('traceSpan propagates metadata, tags, and version when provided', () => { + it('traceSpan propagates trace name, metadata, tags, and version when provided', () => { setLangfuseProcessor({ forceFlush: vi.fn(async () => undefined) }); const callback = vi.fn(); @@ -109,6 +109,7 @@ describe('tracing helpers', () => { { sessionId: 'session-1', userId: 'user-1', + traceName: 'my-eval-trace', metadata: { repo: 'metamask' }, tags: ['eval', 'ci'], version: '1.0.0', @@ -120,6 +121,7 @@ describe('tracing helpers', () => { { sessionId: 'session-1', userId: 'user-1', + traceName: 'my-eval-trace', metadata: { repo: 'metamask' }, tags: ['eval', 'ci'], version: '1.0.0', diff --git a/src/telemetry/tracing.ts b/src/telemetry/tracing.ts index 8f1aaa2..8f6c0f8 100644 --- a/src/telemetry/tracing.ts +++ b/src/telemetry/tracing.ts @@ -50,6 +50,8 @@ export type TraceAttributes = { sessionId: string | undefined; /** User identifier for trace attribution. */ userId: string; + /** Trace name shared by all observations in the run. */ + traceName?: string | undefined; /** Arbitrary metadata attached to the trace. */ metadata?: Record | undefined; /** Tags for trace filtering and categorisation. */ @@ -103,6 +105,9 @@ export function traceSpan(attrs: TraceAttributes, fn: () => void): void { if (attrs.version !== undefined) { propagated.version = attrs.version; } + if (attrs.traceName !== undefined) { + propagated.traceName = attrs.traceName; + } propagateAttributes(propagated, fn); } catch { // Fire-and-forget tracing should not affect agent execution. diff --git a/src/types.ts b/src/types.ts index 35acb98..426e17a 100644 --- a/src/types.ts +++ b/src/types.ts @@ -422,7 +422,7 @@ export type AgentRunnerConfig< * Per-run telemetry attributes propagated to Langfuse traces. */ export type AgentRunTelemetryAttributes = { - /** Name for the root Langfuse trace. */ + /** Name for the Langfuse trace and its root observation. Defaults to `agent-runner`. */ traceName?: string; /** User identifier propagated to traces. */ userId?: string;