From 5d647cdd6e4834bcbc3baad6aa0078774a0a042e Mon Sep 17 00:00:00 2001 From: Hector Hernandez <39923391+hectorhdzg@users.noreply.github.com> Date: Fri, 24 Apr 2026 17:10:22 -0700 Subject: [PATCH 1/4] Align GenAI instrumentations with A365 Node.js observability schema Sync OpenAI Agents and LangChain instrumentations with the A365 Node.js observability extensions (Agent365-nodejs commit b091a8b): - Add SpanKind per operation: SERVER for agent/invoke_agent, CLIENT for chat/execute_tool/handoff/generation/function/mcp_tools - Use structured message format (versioned envelope with MessagePart[]) for input/output messages via serializeMessages - Replace graph_node_id/graph_node_parent_id with gen_ai.agent.name and microsoft.a365.caller.agent.name attributes - Add gen_ai.conversation.id as separate attribute (split from session_id) - Add error.type attribute on error spans - Remove isContentRecordingEnabled option: content is now always recorded - Handoff spans set invoke_agent operation with caller/target agent names - Use safeSerializeToJson for tool arguments/results - Support both Responses API and Chat Completions token usage shapes - Update tests for new structured message format and attribute changes --- samples/src/langchainInstrumentation.ts | 1 - samples/src/openaiInstrumentation.ts | 1 - src/distro/distro.ts | 4 +- .../langchain/langchainTraceInstrumentor.ts | 24 +- .../instrumentations/langchain/tracer.ts | 64 ++- src/genai/instrumentations/langchain/utils.ts | 375 ++++++++++++++---- .../openai/openAIAgentsTraceInstrumentor.ts | 1 - .../openai/openAIAgentsTraceProcessor.ts | 144 ++++--- src/genai/instrumentations/openai/semconv.ts | 2 - src/genai/instrumentations/openai/utils.ts | 358 ++++++++++++++--- src/genai/semconv.ts | 6 + src/types.ts | 11 +- .../unit/genai/langchain/tracer.test.ts | 27 +- .../unit/genai/langchain/utils.test.ts | 21 +- .../openai/openAIAgentsTraceProcessor.test.ts | 30 +- test/internal/unit/genai/openai/utils.test.ts | 72 +--- test/internal/unit/main.test.ts | 5 - 17 files changed, 799 insertions(+), 347 deletions(-) diff --git a/samples/src/langchainInstrumentation.ts b/samples/src/langchainInstrumentation.ts index e835041..0df4e48 100644 --- a/samples/src/langchainInstrumentation.ts +++ b/samples/src/langchainInstrumentation.ts @@ -24,7 +24,6 @@ async function main(): Promise { instrumentationOptions: { langchain: { enabled: true, - isContentRecordingEnabled: true, }, }, }); diff --git a/samples/src/openaiInstrumentation.ts b/samples/src/openaiInstrumentation.ts index da9790e..00a2365 100644 --- a/samples/src/openaiInstrumentation.ts +++ b/samples/src/openaiInstrumentation.ts @@ -23,7 +23,6 @@ async function main(): Promise { instrumentationOptions: { openaiAgents: { enabled: true, - isContentRecordingEnabled: true, }, }, }); diff --git a/src/distro/distro.ts b/src/distro/distro.ts index 5d3099d..5f61e03 100644 --- a/src/distro/distro.ts +++ b/src/distro/distro.ts @@ -277,7 +277,7 @@ async function initializeOpenAIAgentsInstrumentation( } async function initializeLangChainInstrumentation( - options: LangChainInstrumentationConfig, + _options: LangChainInstrumentationConfig, ): Promise { try { const [{ LangChainTraceInstrumentor }, callbackManagerModule] = await Promise.all([ @@ -285,7 +285,7 @@ async function initializeLangChainInstrumentation( import("@langchain/core/callbacks/manager"), ]); if (isShutdown) return; - LangChainTraceInstrumentor.instrument(callbackManagerModule, options); + LangChainTraceInstrumentor.instrument(callbackManagerModule); } catch (error) { Logger.getInstance().warn( "[GenAI] Failed to initialize LangChain instrumentation. " + diff --git a/src/genai/instrumentations/langchain/langchainTraceInstrumentor.ts b/src/genai/instrumentations/langchain/langchainTraceInstrumentor.ts index 62f8ade..1264220 100644 --- a/src/genai/instrumentations/langchain/langchainTraceInstrumentor.ts +++ b/src/genai/instrumentations/langchain/langchainTraceInstrumentor.ts @@ -20,9 +20,8 @@ class LangChainTraceInstrumentorImpl extends InstrumentationBase ) { - args[0] = addTracerToHandlers(instrumentor.otelTracer, args[0], { - isContentRecordingEnabled: instrumentor.isContentRecordingEnabled, - }); + args[0] = addTracerToHandlers(instrumentor.otelTracer, args[0]); diag.debug("[LangChainTraceInstrumentor] _configureSync wrapped to add LangChainTracer"); return original.apply(this, args); }; @@ -153,9 +147,8 @@ export class LangChainTraceInstrumentor { */ static instrument( module: CallbackManagerModuleType, - options?: { isContentRecordingEnabled?: boolean }, ): void { - LangChainTraceInstrumentorImpl.getInstance(options).manuallyInstrumentImpl(module); + LangChainTraceInstrumentorImpl.getInstance().manuallyInstrumentImpl(module); } /** @@ -189,21 +182,20 @@ export class LangChainTraceInstrumentor { export function addTracerToHandlers( tracer: Tracer, handlers: CallbackManagerModule.Callbacks | undefined, - options?: { isContentRecordingEnabled?: boolean }, ): CallbackManagerModule.Callbacks { if (handlers == null) { - return [new LangChainTracer(tracer, options)]; + return [new LangChainTracer(tracer)]; } if (Array.isArray(handlers)) { if (!handlers.some((h) => h instanceof LangChainTracer)) { - handlers.push(new LangChainTracer(tracer, options)); + handlers.push(new LangChainTracer(tracer)); } return handlers; } if (!handlers.inheritableHandlers.some((h) => h instanceof LangChainTracer)) { - handlers.addHandler(new LangChainTracer(tracer, options), true); + handlers.addHandler(new LangChainTracer(tracer), true); } return handlers; } diff --git a/src/genai/instrumentations/langchain/tracer.ts b/src/genai/instrumentations/langchain/tracer.ts index 5b1ae07..1331f9a 100644 --- a/src/genai/instrumentations/langchain/tracer.ts +++ b/src/genai/instrumentations/langchain/tracer.ts @@ -6,7 +6,12 @@ import { context, trace, Span, SpanKind, SpanStatusCode, Tracer } from "@opentel import { BaseTracer, Run } from "@langchain/core/tracers/base"; import { isTracingSuppressed } from "@opentelemetry/core"; import { diag } from "@opentelemetry/api"; -import { ATTR_ERROR_MESSAGE, ATTR_GEN_AI_PROVIDER_NAME } from "../../index.js"; +import { + ATTR_ERROR_MESSAGE, + ATTR_ERROR_TYPE, + ATTR_GEN_AI_CALLER_AGENT_NAME, + ATTR_GEN_AI_PROVIDER_NAME, +} from "../../index.js"; import * as Utils from "./utils.js"; type RunWithSpan = { run: Run; span: Span; startTime: number; lastAccessTime: number }; @@ -26,23 +31,21 @@ type RunWithSpan = { run: Run; span: Span; startTime: number; lastAccessTime: nu * - Skips LangChain-internal runs (tagged `langsmith:hidden`, `Branch*`, or * unmapped run types) to avoid noisy traces. * - Guards against unbounded memory with a hard cap of {@link MAX_RUNS}. - * - Content-sensitive attributes (messages, tool args, system instructions) - * are only recorded when `isContentRecordingEnabled` is true. + * - Content attributes (messages, tool args) are always recorded + * (aligned with Python/.NET SDKs). */ export class LangChainTracer extends BaseTracer { /** Hard cap on concurrent tracked runs to prevent memory leaks. */ private static readonly MAX_RUNS = 10_000; private tracer: Tracer; - private isContentRecordingEnabled: boolean; /** Active runs keyed by LangChain run ID. */ private runs = new Map(); /** Maps each run ID → its parent run ID for parent-span-context lookup. */ private parentByRunId = new Map(); - constructor(tracer: Tracer, options?: { isContentRecordingEnabled?: boolean }) { + constructor(tracer: Tracer) { super(); this.tracer = tracer; - this.isContentRecordingEnabled = options?.isContentRecordingEnabled ?? false; } name = "OpenTelemetryLangChainTracer"; @@ -93,12 +96,16 @@ export class LangChainTracer extends BaseTracer { // Build span name: " " let spanName = run.name; + let kind: SpanKind = SpanKind.INTERNAL; if (operation === "invoke_agent") { spanName = `${operation} ${run.name}`; + kind = SpanKind.SERVER; } else if (operation === "execute_tool") { spanName = `${operation} ${run.name}`; + kind = SpanKind.CLIENT; } else if (operation === "chat") { spanName = `${operation} ${Utils.getModel(run) || run.name}`.trim(); + kind = SpanKind.CLIENT; } if (this.runs.size >= LangChainTracer.MAX_RUNS) { @@ -111,7 +118,7 @@ export class LangChainTracer extends BaseTracer { const span = this.tracer.startSpan( spanName, { - kind: SpanKind.INTERNAL, + kind, startTime, attributes: { [ATTR_GEN_AI_PROVIDER_NAME]: "langchain" }, }, @@ -123,8 +130,7 @@ export class LangChainTracer extends BaseTracer { /** * Called by LangChain when a run finishes. Sets status, enriches the span - * with GenAI attributes, and ends it. If content recording is enabled, - * message bodies, tool arguments, and system instructions are also attached. + * with GenAI attributes, and ends it. */ protected async _endTrace(run: Run) { if (isTracingSuppressed(context.active())) { @@ -162,6 +168,12 @@ export class LangChainTracer extends BaseTracer { if (run.error) { span.setStatus({ code: SpanStatusCode.ERROR }); span.setAttribute(ATTR_ERROR_MESSAGE, String(run.error)); + const errorType = + (run.error as { name?: string })?.name ?? + (run.error as { constructor?: { name?: string } })?.constructor?.name; + if (typeof errorType === "string" && errorType.length > 0) { + span.setAttribute(ATTR_ERROR_TYPE, errorType); + } } else { span.setStatus({ code: SpanStatusCode.OK }); } @@ -169,18 +181,22 @@ export class LangChainTracer extends BaseTracer { // Always-on attributes: operation type, agent info, model, provider, session, tokens Utils.setOperationTypeAttribute(operation, span); Utils.setAgentAttributes(run, span); + if (operation === "invoke_agent") { + const callerName = this.findCallerAgentName(run); + if (callerName) { + span.setAttribute(ATTR_GEN_AI_CALLER_AGENT_NAME, callerName); + } + } Utils.setModelAttribute(run, span); Utils.setProviderNameAttribute(run, span); Utils.setSessionIdAttribute(run, span); Utils.setTokenAttributes(run, span); - // Opt-in content attributes (may contain PII / large payloads) - if (this.isContentRecordingEnabled) { - Utils.setToolAttributes(run, span); - Utils.setInputMessagesAttribute(run, span); - Utils.setOutputMessagesAttribute(run, span); - Utils.setSystemInstructionsAttribute(run, span); - } + // Content attributes — always recorded (aligned with Python/.NET SDKs) + Utils.setToolAttributes(run, span); + Utils.setInputMessagesAttribute(run, span); + Utils.setOutputMessagesAttribute(run, span); + Utils.setSystemInstructionsAttribute(run, span); } catch (error) { diag.error( `[LangChainTracer] Error setting span attributes for run ${run.name}: ${error instanceof Error ? error.message : String(error)}`, @@ -209,4 +225,20 @@ export class LangChainTracer extends BaseTracer { } return undefined; } + + /** + * Walk up the parent run chain to find the nearest ancestor that is an + * invoke_agent run, returning its name as the caller agent name. + */ + private findCallerAgentName(run: Run): string | undefined { + let pid = run.parent_run_id; + while (pid) { + const entry = this.runs.get(pid); + if (entry && Utils.getOperationType(entry.run) === "invoke_agent") { + return entry.run.name; + } + pid = this.parentByRunId.get(pid); + } + return undefined; + } } diff --git a/src/genai/instrumentations/langchain/utils.ts b/src/genai/instrumentations/langchain/utils.ts index 6a3a045..4ec7ab8 100644 --- a/src/genai/instrumentations/langchain/utils.ts +++ b/src/genai/instrumentations/langchain/utils.ts @@ -6,6 +6,7 @@ import { Run } from "@langchain/core/tracers/base"; import { Span } from "@opentelemetry/api"; import { ATTR_GEN_AI_AGENT_NAME, + ATTR_GEN_AI_CONVERSATION_ID, ATTR_GEN_AI_INPUT_MESSAGES, ATTR_GEN_AI_OPERATION_NAME, ATTR_GEN_AI_OUTPUT_MESSAGES, @@ -24,6 +25,21 @@ import { GEN_AI_OPERATION_EXECUTE_TOOL, GEN_AI_OPERATION_INVOKE_AGENT, } from "../../index.js"; +import { + serializeMessages, + safeSerializeToJson, +} from "../../../a365/message-utils.js"; +import { + MessageRole, + A365_MESSAGE_SCHEMA_VERSION, +} from "../../../a365/contracts.js"; +import type { + ChatMessage, + OutputMessage, + InputMessages, + OutputMessages, + MessagePart, +} from "../../../a365/contracts.js"; // Type guards export function isString(value: unknown): value is string { @@ -71,20 +87,41 @@ export function setToolAttributes(run: Run, span: Span) { if (isString(run.name)) { span.setAttribute(ATTR_GEN_AI_TOOL_NAME, run.name); } - if (run.inputs) + if (run.inputs) { + const argsValue = run.inputs?.input ?? run.inputs; span.setAttribute( ATTR_GEN_AI_TOOL_CALL_ARGUMENTS, - JSON.stringify(run.inputs?.input ?? run.inputs), + safeSerializeToJson( + typeof argsValue === "object" + ? (argsValue as Record) + : String(argsValue), + "arguments", + ), ); - if (run.outputs?.output?.kwargs?.content) + } + + // Tool result: v0 uses output.kwargs.content, v1 returns output as a plain string or has content directly + const toolResult = + run.outputs?.output?.kwargs?.content ?? + (isString(run.outputs?.output) ? run.outputs.output : null) ?? + run.outputs?.output?.content; + if (toolResult != null) { span.setAttribute( ATTR_GEN_AI_TOOL_CALL_RESULT, - JSON.stringify(run.outputs?.output?.kwargs?.content), + safeSerializeToJson( + typeof toolResult === "object" + ? (toolResult as Record) + : String(toolResult), + "result", + ), ); + } + span.setAttribute(ATTR_GEN_AI_TOOL_TYPE, "extension"); - if (run.outputs?.output?.tool_call_id) - span.setAttribute(ATTR_GEN_AI_TOOL_CALL_ID, run.outputs?.output?.tool_call_id); + // Tool call ID: v0 uses output.tool_call_id, v1 may have it on inputs + const toolCallId = run.outputs?.output?.tool_call_id ?? run.inputs?.tool_call_id; + if (toolCallId) span.setAttribute(ATTR_GEN_AI_TOOL_CALL_ID, toolCallId); } export function setInputMessagesAttribute(run: Run, span: Span) { @@ -93,40 +130,56 @@ export function setInputMessagesAttribute(run: Run, span: Span) { return; } + // LangChain may provide messages as a direct array or as a single nested array. + // Normalize both shapes so agent/inference inputs are consistently processed. const preprocess = - getScopeType(run) === "inference" && messages.length > 0 ? messages[0] : messages; - const processed = preprocess - ?.map((msg: Record) => { - const content = extractMessageContent(msg); - if (!content) return null; - - const msgType = getMessageType(msg); - if (shouldIncludeInputMessage(msgType)) { - return content; - } - return null; - }) - .filter(Boolean); + getScopeType(run) !== "unknown" && messages.length > 0 && Array.isArray(messages[0]) + ? (messages[0] as unknown[]) + : messages; + const chatMessages: ChatMessage[] = []; + + for (const msg of preprocess) { + if (!msg || typeof msg !== "object") continue; + const msgObj = msg as Record; + const parts = buildPartsFromMessage(msgObj); + if (parts.length === 0) continue; + + const msgType = getMessageType(msgObj); + const role = mapLangChainRole(msgType); + chatMessages.push({ role, parts }); + } - if (processed.length > 0) { - span.setAttribute(ATTR_GEN_AI_INPUT_MESSAGES, JSON.stringify(processed)); + if (chatMessages.length > 0) { + const wrapper: InputMessages = { + version: A365_MESSAGE_SCHEMA_VERSION, + messages: chatMessages, + }; + span.setAttribute(ATTR_GEN_AI_INPUT_MESSAGES, serializeMessages(wrapper)); } } -// Helper: Extract message content from various formats -function extractMessageContent(msg: Record): string | null { - // Simple format: {role: "user", content} - if (isString(msg.content)) { - return msg.content; +// Helper: Extract string content from a message (used for fallback text extraction and system instructions) +function extractStringContent(msg: Record): string | null { + const raw = extractRawContent(msg); + return isString(raw) ? raw : null; +} + +// Helper: Extract raw content (string or content block array) from various message formats +function extractRawContent(msg: Record): string | unknown[] | null { + // Simple format: {role: "user", content: string | array} + if (msg.content !== undefined && msg.content !== null) { + if (isString(msg.content)) return msg.content; + if (Array.isArray(msg.content)) return msg.content; } // LangChain format: {lc_type: "human", lc_kwargs: {content}} if (msg.lc_kwargs && typeof msg.lc_kwargs === "object" && !Array.isArray(msg.lc_kwargs)) { const kwargs = msg.lc_kwargs as Record; if (isString(kwargs.content)) return kwargs.content; + if (Array.isArray(kwargs.content)) return kwargs.content; } - // New LangChain format: {lc: 1, type: "constructor", kwargs: {content}} + // LangChain v1 serialized class instance format: { lc: 1, type: "constructor", kwargs: {...} } if ( msg.lc === 1 && msg.type === "constructor" && @@ -136,24 +189,170 @@ function extractMessageContent(msg: Record): string | null { ) { const kwargs = msg.kwargs as Record; if (isString(kwargs.content)) return kwargs.content; + if (Array.isArray(kwargs.content)) return kwargs.content; } return null; } +// Helper: Map LangChain message type to MessageRole +function mapLangChainRole(msgType: string): MessageRole | string { + switch (msgType) { + case "user": + case "human": + return MessageRole.USER; + case "assistant": + case "ai": + return MessageRole.ASSISTANT; + case "system": + return MessageRole.SYSTEM; + case "tool": + return MessageRole.TOOL; + default: + return msgType; + } +} + +// Helper: Build MessagePart[] from a LangChain message +function buildPartsFromMessage(msg: Record): MessagePart[] { + const parts: MessagePart[] = []; + const rawContent = extractRawContent(msg); + + const addUnknownBlockPart = (blockType: string, block: Record) => { + try { + parts.push({ type: blockType, content: JSON.stringify(block) } as MessagePart); + } catch { + parts.push({ type: blockType, content: "[unserializable]" } as MessagePart); + } + }; + + const addPartFromContentBlock = (block: unknown) => { + if (!block || typeof block !== "object") return; + + const contentBlock = block as Record; + const blockType = contentBlock.type as string | undefined; + if (!blockType) return; + + if (blockType === "text" && isString(contentBlock.text)) { + parts.push({ type: "text", content: contentBlock.text }); + return; + } + + if (blockType === "reasoning" && isString(contentBlock.reasoning)) { + parts.push({ type: "reasoning", content: contentBlock.reasoning }); + return; + } + + if (blockType === "tool_call") { + parts.push({ + type: "tool_call", + name: String(contentBlock.name ?? ""), + id: contentBlock.id != null ? String(contentBlock.id) : undefined, + arguments: + contentBlock.args && typeof contentBlock.args === "object" + ? (contentBlock.args as Record) + : undefined, + }); + return; + } + + addUnknownBlockPart(blockType, contentBlock); + }; + + if (isString(rawContent)) { + parts.push({ type: "text", content: rawContent }); + } else if (Array.isArray(rawContent)) { + for (const block of rawContent) { + addPartFromContentBlock(block); + } + } + + // Extract tool_calls from the message (AI messages may have a separate tool_calls array) + // Deduplicate by ID to avoid duplicates when tool_calls appear in both content blocks and tool_calls array + const seenToolCallIds = new Set(); + for (const part of parts) { + if (part.type !== "tool_call") continue; + const partId = (part as Record).id; + if (isString(partId)) { + seenToolCallIds.add(partId); + } + } + + for (const toolCall of extractToolCalls(msg)) { + const toolCallId = (toolCall as Record).id; + if (isString(toolCallId) && seenToolCallIds.has(toolCallId)) { + continue; + } + if (isString(toolCallId)) { + seenToolCallIds.add(toolCallId); + } + parts.push(toolCall); + } + + // Fallback: if no parts were built, use text extraction + if (parts.length === 0) { + const textContent = extractStringContent(msg); + if (textContent) { + parts.push({ type: "text", content: textContent }); + } + } + + return parts; +} + +// Helper: Extract tool_calls from a LangChain message +function extractToolCalls(msg: Record): MessagePart[] { + const parts: MessagePart[] = []; + + // Standard format: message.tool_calls[] — check direct, lc_kwargs, and kwargs paths + const directToolCalls = + getNestedValue(msg, "tool_calls") ?? + getNestedValue(msg, "lc_kwargs", "tool_calls") ?? + getNestedValue(msg, "kwargs", "tool_calls"); + if (Array.isArray(directToolCalls)) { + for (const tc of directToolCalls) { + if (!tc || typeof tc !== "object") continue; + const call = tc as Record; + parts.push({ + type: "tool_call", + name: String(call.name ?? ""), + id: call.id != null ? String(call.id) : undefined, + arguments: + call.args && typeof call.args === "object" + ? (call.args as Record) + : undefined, + }); + } + } + + return parts; +} + +// Helper: Safely get a nested value from a message object +function getNestedValue(obj: Record, ...keys: string[]): unknown { + let current: unknown = obj; + for (const key of keys) { + if (!current || typeof current !== "object" || Array.isArray(current)) return undefined; + current = (current as Record)[key]; + } + return current; +} + // Helper: Determine message type function getMessageType(msg: Record): string { // Simple format if (isString(msg.role)) return msg.role; // LangChain old format if (isString(msg.lc_type)) return msg.lc_type; - if (isString(msg.type)) return msg.type; - // LangChain new format - check id array for message type + // Skip v1 constructor type marker — fall through to id array check + if (isString(msg.type) && msg.type !== "constructor") return msg.type; + // LangChain v1 format - check id array for message type (e.g., ["langchain_core", "messages", "HumanMessage"]) if (Array.isArray(msg.id)) { const lastId = msg.id[msg.id.length - 1]; if (isString(lastId)) { if (lastId.includes("Human")) return "human"; if (lastId.includes("AI")) return "ai"; if (lastId.includes("System")) return "system"; + if (lastId.includes("Tool")) return "tool"; } } return "unknown"; @@ -171,12 +370,6 @@ function getScopeType(run: Run): "agent" | "tool" | "inference" | "unknown" { return "unknown"; } -// Helper: Check if input message should be included based on scope and message type -function shouldIncludeInputMessage(msgType: string): boolean { - // For input messages: all scopes want user/human messages only - return msgType === "user" || msgType === "human"; -} - // Helper: Check if output message should be included based on scope and message type function shouldIncludeOutputMessage(scopeType: string, msgType: string): boolean { if (scopeType === "agent" || scopeType === "inference") { @@ -197,19 +390,25 @@ export function setOutputMessagesAttribute(run: Run, span: Span) { } const scopeType = getScopeType(run); - const messages: string[] = []; + const outputMessages: OutputMessage[] = []; + + // Helper: process a single message object into an OutputMessage + const processMessage = (msg: Record) => { + const msgType = getMessageType(msg); + if (!shouldIncludeOutputMessage(scopeType, msgType)) return; + + const parts = buildPartsFromMessage(msg); + if (parts.length === 0) return; + + const role = mapLangChainRole(msgType); + outputMessages.push({ role, parts }); + }; // Direct messages array (used in agent/chain outputs) if (Array.isArray(outputs.messages)) { - outputs.messages.forEach((msg: Record) => { - const content = extractMessageContent(msg); - if (!content) return; - - const msgType = getMessageType(msg); - if (shouldIncludeOutputMessage(scopeType, msgType)) { - messages.push(content); - } - }); + for (const msg of outputs.messages as Record[]) { + processMessage(msg); + } } // LangChain generations format (used in LLM/inference outputs) @@ -219,20 +418,14 @@ export function setOutputMessagesAttribute(run: Run, span: Span) { gen.forEach((item: Record) => { // Try message property if (item.message && typeof item.message === "object" && !Array.isArray(item.message)) { - const msg = item.message as Record; - const content = extractMessageContent(msg); - if (!content) { - return; - } - - const msgType = getMessageType(msg); - if (shouldIncludeOutputMessage(scopeType, msgType)) { - messages.push(content); - } + processMessage(item.message as Record); } // Try direct text property (for generation items) else if (isString(item.text) && scopeType === "inference") { - messages.push(item.text); + outputMessages.push({ + role: MessageRole.ASSISTANT, + parts: [{ type: "text", content: item.text }], + }); } }); } @@ -241,25 +434,26 @@ export function setOutputMessagesAttribute(run: Run, span: Span) { // Check for direct message object (some models return this) if (outputs.message && typeof outputs.message === "object" && !Array.isArray(outputs.message)) { - const msg = outputs.message as Record; - const content = extractMessageContent(msg); - if (content) { - const msgType = getMessageType(msg); - if (shouldIncludeOutputMessage(scopeType, msgType)) { - messages.push(content); - } - } + processMessage(outputs.message as Record); } - if (messages.length > 0) { - span.setAttribute(ATTR_GEN_AI_OUTPUT_MESSAGES, JSON.stringify(messages)); + if (outputMessages.length > 0) { + const wrapper: OutputMessages = { + version: A365_MESSAGE_SCHEMA_VERSION, + messages: outputMessages, + }; + span.setAttribute(ATTR_GEN_AI_OUTPUT_MESSAGES, serializeMessages(wrapper)); } } // Model - Helper to extract model name from run export function getModel(run: Run): string | undefined { return [ + // v1: response_metadata directly on message + run.outputs?.generations?.[0]?.[0]?.message?.response_metadata?.model_name, + // v0: response_metadata nested under kwargs run.outputs?.generations?.[0]?.[0]?.message?.kwargs?.response_metadata?.model_name, + // Metadata paths (both v0 and v1) run.extra?.metadata?.ls_model_name, run.extra?.invocation_params?.model, run.extra?.invocation_params?.model_name, @@ -284,11 +478,15 @@ export function setSessionIdAttribute(run: Run, span: Span): void { const metadata = run.extra?.metadata as Record | undefined; if (!metadata) return; - const sessionId = metadata.session_id ?? metadata.conversation_id ?? metadata.thread_id; - - if (typeof sessionId === "string" && sessionId.length > 0) { + const sessionId = metadata.session_id ?? metadata.thread_id; + if (isString(sessionId) && sessionId.length > 0) { span.setAttribute(ATTR_MICROSOFT_SESSION_ID, sessionId); } + + const conversationId = metadata.conversation_id; + if (isString(conversationId) && conversationId.length > 0) { + span.setAttribute(ATTR_GEN_AI_CONVERSATION_ID, conversationId); + } } // System instructions @@ -306,12 +504,20 @@ export function setSystemInstructionsAttribute(run: Run, span: Span) { : ""; if (prompts) return span.setAttribute(ATTR_GEN_AI_SYSTEM_INSTRUCTIONS, prompts); - const messages = Array.isArray(inputs.messages) ? inputs.messages : []; - const systemText = messages - .filter((m: Record) => m.lc_type === "system") - .map((m: Record) => - String((m.lc_kwargs as Record | undefined)?.content ?? "").trim(), - ) + // Check both flat and nested message arrays + const rawMessages = Array.isArray(inputs.messages) ? inputs.messages : []; + const flatMessages = + rawMessages.length > 0 && Array.isArray(rawMessages[0]) + ? (rawMessages[0] as unknown[]) + : rawMessages; + const systemText = flatMessages + .filter((m: unknown) => { + if (!m || typeof m !== "object") return false; + const msgType = getMessageType(m as Record); + return msgType === "system"; + }) + .map((m: unknown) => extractStringContent(m as Record) ?? "") + .map((s: string) => s.trim()) .filter(Boolean) .join("\n"); if (systemText) span.setAttribute(ATTR_GEN_AI_SYSTEM_INSTRUCTIONS, systemText); @@ -320,11 +526,16 @@ export function setSystemInstructionsAttribute(run: Run, span: Span) { // Tokens (input and output) export function setTokenAttributes(run: Run, span: Span) { // Try multiple paths to find usage metadata (LLM direct/kwargs/response_metadata, agent calls, and chain/model_request outputs) + // v1: usage_metadata is often on the last AI message in outputs.messages + const lastMsg = Array.isArray(run.outputs?.messages) + ? run.outputs.messages[run.outputs.messages.length - 1] + : undefined; const usage = run.outputs?.generations?.[0]?.[0]?.message?.usage_metadata || run.outputs?.generations?.[0]?.[0]?.message?.kwargs?.usage_metadata || + run.outputs?.generations?.[0]?.[0]?.message?.response_metadata?.tokenUsage || run.outputs?.generations?.[0]?.[0]?.message?.kwargs?.response_metadata?.tokenUsage || - run.outputs?.messages?.[1]?.usage_metadata || + lastMsg?.usage_metadata || run.outputs?.message?.response_metadata?.usage || run.outputs?.message?.response_metadata?.tokenUsage || run.outputs?.messages @@ -339,11 +550,15 @@ export function setTokenAttributes(run: Run, span: Span) { } const usageObj = usage as Record; - if (typeof usageObj.input_tokens === "number") { - span.setAttribute(ATTR_GEN_AI_USAGE_INPUT_TOKENS, usageObj.input_tokens); + // Support both usage_metadata shape (input_tokens/output_tokens) and + // tokenUsage shape (promptTokens/completionTokens) from LangChain OpenAI provider + const inputTokens = usageObj.input_tokens ?? usageObj.promptTokens; + const outputTokens = usageObj.output_tokens ?? usageObj.completionTokens; + if (typeof inputTokens === "number") { + span.setAttribute(ATTR_GEN_AI_USAGE_INPUT_TOKENS, inputTokens); } - if (typeof usageObj.output_tokens === "number") { - span.setAttribute(ATTR_GEN_AI_USAGE_OUTPUT_TOKENS, usageObj.output_tokens); + if (typeof outputTokens === "number") { + span.setAttribute(ATTR_GEN_AI_USAGE_OUTPUT_TOKENS, outputTokens); } } diff --git a/src/genai/instrumentations/openai/openAIAgentsTraceInstrumentor.ts b/src/genai/instrumentations/openai/openAIAgentsTraceInstrumentor.ts index 8ce25fa..a9eaea5 100644 --- a/src/genai/instrumentations/openai/openAIAgentsTraceInstrumentor.ts +++ b/src/genai/instrumentations/openai/openAIAgentsTraceInstrumentor.ts @@ -89,7 +89,6 @@ class OpenAIAgentsTraceInstrumentorImpl extends InstrumentationBase = new Map(); @@ -79,16 +89,10 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { tracer: OtelTracer, options?: { suppressInvokeAgentInput?: boolean; - isContentRecordingEnabled?: boolean; }, ) { this.tracer = tracer; this.suppressInvokeAgentInput = options?.suppressInvokeAgentInput ?? false; - this.isContentRecordingEnabled = options?.isContentRecordingEnabled ?? false; - } - - private isContentKey(key: string): boolean { - return Utils.CONTENT_KEYS.has(key); } private getNewKey(spanType: string, key: string): string | null { @@ -124,6 +128,13 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { return; } + const spanType = spanData?.type as string | undefined; + + // Skip span types we don't map to schema-defined operations. + if (!spanType || spanType === "custom" || spanType === "guardrail") { + return; + } + if (this.otelSpans.size >= OpenAIAgentsTraceProcessor.MAX_SPANS_IN_FLIGHT) { diag.warn( `[OpenAIAgentsTraceProcessor] Max spans in flight (${OpenAIAgentsTraceProcessor.MAX_SPANS_IN_FLIGHT}) reached, skipping span`, @@ -143,10 +154,18 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { const spanName = Utils.getSpanName(span); + // SpanKind per OTel client/server semantics + A365 schema: + const kind = OpenAIAgentsTraceProcessor.SERVER_SPAN_TYPES.has(spanType) + ? SpanKind.SERVER + : OpenAIAgentsTraceProcessor.CLIENT_SPAN_TYPES.has(spanType) + ? SpanKind.CLIENT + : undefined; + // Start OpenTelemetry span const otelSpan = this.tracer.startSpan( spanName, { + kind, startTime, attributes: { [ATTR_GEN_AI_OPERATION_NAME]: Utils.getOperationName(spanData), @@ -208,6 +227,14 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { const endTime = endedAt ? new Date(endedAt).getTime() : undefined; const status = Utils.getSpanStatus(span); otelSpan.setStatus(status); + if (span.error) { + const errData = (span.error as { data?: Record; name?: string }).data; + const errorType = + (typeof errData?.type === "string" && errData.type) || + (span.error as { name?: string }).name || + "error"; + otelSpan.setAttribute(ATTR_ERROR_TYPE, errorType); + } if (endTime) { otelSpan.end(endTime); } else { @@ -230,19 +257,18 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { private processSpanData(otelSpan: OtelSpan, data: SpanData, traceId: string): void { const type = data.type; - const contentRecording = this.isContentRecordingEnabled; switch (type) { case "response": - this.processResponseSpanData(otelSpan, data, contentRecording); + this.processResponseSpanData(otelSpan, data); break; case "generation": - this.processGenerationSpanData(otelSpan, data, traceId, contentRecording); + this.processGenerationSpanData(otelSpan, data, traceId); break; case "function": - this.processFunctionSpanData(otelSpan, data, traceId, contentRecording); + this.processFunctionSpanData(otelSpan, data, traceId); break; case "mcp_tools": @@ -259,11 +285,7 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { } } - private processResponseSpanData( - otelSpan: OtelSpan, - data: SpanData, - contentRecording: boolean, - ): void { + private processResponseSpanData(otelSpan: OtelSpan, data: SpanData): void { const responseData = data as Record; const responseObj = responseData._response || responseData.response; const inputObj = responseData._input || responseData.input; @@ -271,20 +293,22 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { if (responseObj) { const resp = responseObj as Record; - if (resp.output && contentRecording) { - if (typeof resp.output === "string") { - otelSpan.setAttribute(ATTR_GEN_AI_OUTPUT_MESSAGES, truncateValue(resp.output)); + // Store the output field as structured OutputMessages (always use versioned envelope) + if (resp.output != null) { + if (Array.isArray(resp.output)) { + const structured = Utils.buildStructuredOutputMessages( + resp.output as Array>, + ); + otelSpan.setAttribute( + ATTR_GEN_AI_OUTPUT_MESSAGES, + serializeMessages(structured), + ); } else { + // String or non-array object — wrap as raw content + const structured = Utils.wrapRawContentAsOutputMessages(resp.output); otelSpan.setAttribute( ATTR_GEN_AI_OUTPUT_MESSAGES, - truncateValue( - Utils.buildOutputMessages( - resp.output as Array<{ - role: string; - content: Array<{ type: string; text: string }>; - }>, - ), - ), + serializeMessages(structured), ); } } @@ -301,25 +325,31 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { otelSpan.updateName(`${GEN_AI_OPERATION_CHAT} ${modelName}`); } - if (inputObj && !this.suppressInvokeAgentInput && contentRecording) { + if (inputObj != null && !this.suppressInvokeAgentInput) { if (typeof inputObj === "string") { try { const parsed = JSON.parse(inputObj as string); if (Array.isArray(parsed)) { + const structured = Utils.buildStructuredInputMessages(parsed); otelSpan.setAttribute( ATTR_GEN_AI_INPUT_MESSAGES, - truncateValue(Utils.buildInputMessages(parsed)), + serializeMessages(structured), ); return; } } catch { - // Fall back to raw string + // If parsing fails, wrap raw string in versioned envelope } - otelSpan.setAttribute(ATTR_GEN_AI_INPUT_MESSAGES, truncateValue(inputObj as string)); + const wrappedInput = Utils.wrapRawContentAsInputMessages(inputObj); + otelSpan.setAttribute( + ATTR_GEN_AI_INPUT_MESSAGES, + serializeMessages(wrappedInput), + ); } else if (Array.isArray(inputObj)) { + const structured = Utils.buildStructuredInputMessages(inputObj); otelSpan.setAttribute( ATTR_GEN_AI_INPUT_MESSAGES, - truncateValue(Utils.buildInputMessages(inputObj)), + serializeMessages(structured), ); } } @@ -329,7 +359,6 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { otelSpan: OtelSpan, data: SpanData, traceId: string, - contentRecording: boolean, ): void { const attrs = Utils.getAttributesFromGenerationSpanData(data); Object.entries(attrs).forEach(([key, value]) => { @@ -337,7 +366,10 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { if (value !== null && value !== undefined && !shouldExcludeKey) { const newKey = this.getNewKey(data.type, key); const resolvedKey = newKey || key; - if (!this.isContentKey(resolvedKey) || contentRecording) { + if ( + resolvedKey !== ATTR_GEN_AI_INPUT_MESSAGES || + !this.suppressInvokeAgentInput + ) { otelSpan.setAttribute(resolvedKey, value as string | number | boolean); } } @@ -345,10 +377,9 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { this.stampCustomParent(otelSpan, traceId); - const operationName = attrs[ATTR_GEN_AI_OPERATION_NAME]; const modelName = attrs[ATTR_GEN_AI_REQUEST_MODEL]; - if (operationName && modelName) { - otelSpan.updateName(`${operationName} ${modelName}`); + if (typeof modelName === "string" && modelName.length > 0) { + otelSpan.updateName(`${GEN_AI_OPERATION_CHAT} ${modelName}`); } } @@ -356,7 +387,6 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { otelSpan: OtelSpan, data: SpanData, traceId: string, - contentRecording: boolean, ): void { const functionData = data as Record; const attrs = Utils.getAttributesFromFunctionSpanData(data); @@ -364,9 +394,7 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { if (value !== null && value !== undefined) { const newKey = this.getNewKey(data.type, key); const resolvedKey = newKey || key; - if (!this.isContentKey(resolvedKey) || contentRecording) { - otelSpan.setAttribute(resolvedKey, value as string | number | boolean); - } + otelSpan.setAttribute(resolvedKey, value as string | number | boolean); } }); otelSpan.setAttribute(ATTR_GEN_AI_TOOL_TYPE, "function"); @@ -396,9 +424,12 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { private processHandoffSpanData(_otelSpan: OtelSpan, data: SpanData, traceId: string): void { const handoffData = data as Record; - if (handoffData.to_agent && handoffData.from_agent) { - const key = `${handoffData.to_agent}:${traceId}`; - this.reverseHandoffsDict.set(key, handoffData.from_agent as string); + const fromAgent = handoffData.from_agent as string | undefined; + const toAgent = handoffData.to_agent as string | undefined; + + if (toAgent && fromAgent) { + const key = `${toAgent}:${traceId}`; + this.reverseHandoffsDict.set(key, fromAgent); while (this.reverseHandoffsDict.size > OpenAIAgentsTraceProcessor.MAX_HANDOFFS_IN_FLIGHT) { const firstKey = this.reverseHandoffsDict.keys().next().value; @@ -407,20 +438,29 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { } } } + + _otelSpan.setAttribute(ATTR_GEN_AI_OPERATION_NAME, GEN_AI_OPERATION_INVOKE_AGENT); + if (toAgent) { + _otelSpan.setAttribute(ATTR_GEN_AI_AGENT_NAME, toAgent); + _otelSpan.updateName(`${GEN_AI_OPERATION_INVOKE_AGENT} ${toAgent}`); + } + if (fromAgent) { + _otelSpan.setAttribute(ATTR_GEN_AI_CALLER_AGENT_NAME, fromAgent); + } } private processAgentSpanData(otelSpan: OtelSpan, data: SpanData, traceId: string): void { const agentData = data as Record; if (agentData.name) { - otelSpan.setAttribute(GEN_AI_GRAPH_NODE_ID, agentData.name as string); + otelSpan.setAttribute(ATTR_GEN_AI_AGENT_NAME, agentData.name as string); otelSpan.setAttribute(ATTR_GEN_AI_OPERATION_NAME, GEN_AI_OPERATION_INVOKE_AGENT); - // Lookup parent node from handoff + // Link back to the agent that handed off to this one (A2A caller semantics) const key = `${agentData.name}:${traceId}`; const parentNode = this.reverseHandoffsDict.get(key); if (parentNode) { this.reverseHandoffsDict.delete(key); - otelSpan.setAttribute(GEN_AI_GRAPH_NODE_PARENT_ID, parentNode); + otelSpan.setAttribute(ATTR_GEN_AI_CALLER_AGENT_NAME, parentNode); } otelSpan.updateName(`${GEN_AI_OPERATION_INVOKE_AGENT} ${agentData.name}`); diff --git a/src/genai/instrumentations/openai/semconv.ts b/src/genai/instrumentations/openai/semconv.ts index 74aea46..9e75f09 100644 --- a/src/genai/instrumentations/openai/semconv.ts +++ b/src/genai/instrumentations/openai/semconv.ts @@ -13,5 +13,3 @@ export const GEN_AI_SPAN_KIND_CHAT = "chat" as const; export const GEN_AI_REQUEST_CONTENT_KEY = "gen_ai.request.content" as const; export const GEN_AI_RESPONSE_CONTENT_KEY = "gen_ai.response.content" as const; export const GEN_AI_EXECUTION_PAYLOAD_KEY = "gen_ai.execution.payload" as const; -export const GEN_AI_GRAPH_NODE_ID = "graph_node_id" as const; -export const GEN_AI_GRAPH_NODE_PARENT_ID = "graph_node_parent_id" as const; diff --git a/src/genai/instrumentations/openai/utils.ts b/src/genai/instrumentations/openai/utils.ts index cf8503d..2a643cb 100644 --- a/src/genai/instrumentations/openai/utils.ts +++ b/src/genai/instrumentations/openai/utils.ts @@ -1,7 +1,6 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. // Vendored from microsoft/Agent365-nodejs packages/agents-a365-observability-extensions-openai -// Adapted: removed A365 observability imports, uses local semconv + truncateValue import { SpanStatusCode } from "@opentelemetry/api"; import type { Span as AgentsSpan, SpanData } from "@openai/agents-core"; @@ -16,7 +15,22 @@ import { GEN_AI_OPERATION_EXECUTE_TOOL, GEN_AI_OPERATION_INVOKE_AGENT, } from "../../index.js"; -import { truncateValue } from "../../utils.js"; +import { + serializeMessages, + safeSerializeToJson, +} from "../../../a365/message-utils.js"; +import { + MessageRole, + A365_MESSAGE_SCHEMA_VERSION, +} from "../../../a365/contracts.js"; +import type { + ChatMessage, + OutputMessage, + InputMessages, + OutputMessages, + MessagePart, +} from "../../../a365/contracts.js"; +import { MAX_SPAN_SIZE_BYTES } from "../../../a365/exporter/utils.js"; import { GEN_AI_EXECUTION_PAYLOAD_KEY, GEN_AI_REQUEST_CONTENT_KEY, @@ -28,13 +42,54 @@ import { } from "./semconv.js"; /** - * Safely stringify an object to JSON, truncating to the attribute limit. + * Locate and normalize usage counts across OpenAI API shapes: + * - Responses API: { input_tokens, output_tokens } + * - Chat Completions: { prompt_tokens, completion_tokens } + * Usage may live directly on the span data, on `.output`, or inside `.output[0]`. + */ +export function extractUsageTokens( + data: Record, +): { inputTokens?: number; outputTokens?: number } { + const candidates: Array | undefined> = []; + const direct = data.usage as Record | undefined; + candidates.push(direct); + const output = data.output as unknown; + if (output && typeof output === "object") { + if (Array.isArray(output)) { + const first = output[0]; + if (first && typeof first === "object") { + candidates.push( + (first as Record).usage as Record | undefined, + ); + } + } else { + candidates.push( + (output as Record).usage as Record | undefined, + ); + } + } + for (const usage of candidates) { + if (!usage) continue; + const inputTokens = usage.input_tokens ?? usage.prompt_tokens; + const outputTokens = usage.output_tokens ?? usage.completion_tokens; + if (typeof inputTokens === "number" || typeof outputTokens === "number") { + return { + inputTokens: typeof inputTokens === "number" ? inputTokens : undefined, + outputTokens: typeof outputTokens === "number" ? outputTokens : undefined, + }; + } + } + return {}; +} + +/** + * Safely stringify an object to JSON. */ export function safeJsonDumps(obj: unknown): string { try { - return truncateValue(JSON.stringify(obj)); + return JSON.stringify(obj); } catch { - return truncateValue(String(obj)); + return String(obj); } } @@ -100,6 +155,7 @@ export function getOperationName(spanData: SpanData | undefined): string { case "response": return GEN_AI_OPERATION_CHAT; case "handoff": + return GEN_AI_OPERATION_INVOKE_AGENT; case "custom": case "guardrail": default: @@ -141,21 +197,23 @@ export function getAttributesFromGenerationSpanData(data: SpanData): Record; - if (usage.input_tokens !== undefined) { - attributes[ATTR_GEN_AI_USAGE_INPUT_TOKENS] = usage.input_tokens; - } - if (usage.output_tokens !== undefined) { - attributes[ATTR_GEN_AI_USAGE_OUTPUT_TOKENS] = usage.output_tokens; - } + const genUsage = extractUsageTokens(genData); + if (genUsage.inputTokens !== undefined) { + attributes[ATTR_GEN_AI_USAGE_INPUT_TOKENS] = genUsage.inputTokens; + } + if (genUsage.outputTokens !== undefined) { + attributes[ATTR_GEN_AI_USAGE_OUTPUT_TOKENS] = genUsage.outputTokens; } // Map operation name for generation spans @@ -176,19 +234,22 @@ export function getAttributesFromFunctionSpanData(data: SpanData): Record) + : String(funcData.input), + "arguments", + ); } if (funcData.output !== undefined && funcData.output !== null) { - const output = + attributes[GEN_AI_RESPONSE_CONTENT_KEY] = safeSerializeToJson( typeof funcData.output === "object" - ? safeJsonDumps(funcData.output) - : truncateValue(String(funcData.output)); - attributes[GEN_AI_RESPONSE_CONTENT_KEY] = output; + ? (funcData.output as Record) + : String(funcData.output), + "result", + ); } return attributes; @@ -219,29 +280,17 @@ export function getAttributesFromResponse(response: unknown): Record; - if (usage.input_tokens !== undefined) { - attributes[ATTR_GEN_AI_USAGE_INPUT_TOKENS] = usage.input_tokens; - } - if (usage.output_tokens !== undefined) { - attributes[ATTR_GEN_AI_USAGE_OUTPUT_TOKENS] = usage.output_tokens; - } + const respUsage = extractUsageTokens(resp); + if (respUsage.inputTokens !== undefined) { + attributes[ATTR_GEN_AI_USAGE_INPUT_TOKENS] = respUsage.inputTokens; + } + if (respUsage.outputTokens !== undefined) { + attributes[ATTR_GEN_AI_USAGE_OUTPUT_TOKENS] = respUsage.outputTokens; } return attributes; } -/** Content-sensitive attribute keys that should only be recorded when content recording is enabled. */ -export const CONTENT_KEYS = new Set([ - "gen_ai.input.messages", - "gen_ai.output.messages", - "gen_ai.tool.call.arguments", - "gen_ai.tool.call.result", - GEN_AI_REQUEST_CONTENT_KEY, - GEN_AI_RESPONSE_CONTENT_KEY, -]); - /** Key remapping table: `${spanType}${originalKey}` → target semconv key */ export const KEY_MAPPINGS = new Map([ [`mcp_tools${GEN_AI_RESPONSE_CONTENT_KEY}`, "gen_ai.tool.call.result"], @@ -252,32 +301,221 @@ export const KEY_MAPPINGS = new Map([ [`generation${GEN_AI_REQUEST_CONTENT_KEY}`, "gen_ai.input.messages"], ]); +// --------------------------------------------------------------------------- +// Structured message builders (OTEL gen-ai message format) +// --------------------------------------------------------------------------- + +type OpenAIInputMessage = { role: string; content: string | unknown[] | unknown }; +type OpenAIOutputItem = { + role?: string; + content?: unknown[]; + type?: string; + text?: string; + [key: string]: unknown; +}; + /** - * Build a JSON string of input messages, extracting user-role content. + * Map an OpenAI role string to a MessageRole value. */ -export function buildInputMessages(arr: Array<{ role: string; content: string }>): string { - const userTexts = arr - .filter((m) => m && m.role === "user" && typeof m.content === "string") - .map((m) => m.content); - return JSON.stringify(userTexts.length ? userTexts : arr); +function mapOpenAIRole(role: string): MessageRole | string { + switch (role) { + case "user": + return MessageRole.USER; + case "assistant": + return MessageRole.ASSISTANT; + case "system": + return MessageRole.SYSTEM; + case "tool": + return MessageRole.TOOL; + default: + return role; + } +} + +function getModalityFromMimeType(mimeType: unknown): string { + return String(mimeType ?? "file").split("/")[0] || "file"; +} + +function mapGenericBlock( + blockType: string | undefined, + block: Record, +): MessagePart { + return { type: blockType ?? "unknown", content: safeJsonDumps(block) } as MessagePart; +} + +function parseToolCallArguments(args: unknown): Record | undefined { + if (typeof args === "string") { + try { + return JSON.parse(args) as Record; + } catch { + return { raw: args }; + } + } + + if (args && typeof args === "object") { + return args as Record; + } + + return undefined; +} + +function getToolCallId(block: Record): string | undefined { + if (block.call_id != null) return String(block.call_id); + if (block.id != null) return String(block.id); + return undefined; +} + +function wrapRawContentAsMessages( + raw: unknown, + role: MessageRole, +): InputMessages | OutputMessages { + const content = typeof raw === "string" ? raw : safeJsonDumps(raw); + return { + version: A365_MESSAGE_SCHEMA_VERSION, + messages: [{ role, parts: [{ type: "text", content }] }], + }; } /** - * Build a JSON string of output messages, extracting output_text content. + * Map an OpenAI input content block to a MessagePart. */ -export function buildOutputMessages( - arr: Array<{ role: string; content: Array<{ type: string; text: string }> }>, -): string { - const userTexts: string[] = []; - for (const { content } of arr) { - if (!Array.isArray(content)) { +function mapInputContentBlock(block: Record): MessagePart { + const blockType = block.type as string | undefined; + switch (blockType) { + case "input_text": + return { type: "text", content: String(block.text ?? "") }; + case "input_image": + return { type: "blob", modality: "image", ...stripBinaryFields(block) } as MessagePart; + case "input_file": + return { + type: "file" as string, + modality: getModalityFromMimeType(block.mime_type), + ...stripBinaryFields(block), + } as MessagePart; + default: + return mapGenericBlock(blockType, block); + } +} + +/** + * Strip large binary fields from a content block for telemetry. + */ +function stripBinaryFields(block: Record): Record { + const result: Record = {}; + for (const [key, value] of Object.entries(block)) { + if (key === "type") continue; + if (typeof value === "string" && value.length > MAX_SPAN_SIZE_BYTES) { + result[key] = "[truncated]"; + } else { + result[key] = value; + } + } + return result; +} + +/** + * Map an OpenAI output content block to a MessagePart. + */ +function mapOutputContentBlock(block: Record): MessagePart { + const blockType = block.type as string | undefined; + switch (blockType) { + case "output_text": + return { type: "text", content: String(block.text ?? "") }; + case "refusal": + return { type: "text", content: String(block.refusal ?? "") }; + case "tool_call": + case "function_call": { + const parsedArgs = parseToolCallArguments(block.arguments ?? block.args); + return { + type: "tool_call", + name: String(block.name ?? block.function ?? ""), + id: getToolCallId(block), + arguments: parsedArgs, + }; + } + case "reasoning": + return { type: "reasoning", content: String(block.text ?? block.content ?? "") }; + default: + return mapGenericBlock(blockType, block); + } +} + +/** + * Build structured InputMessages from an OpenAI _input message array. + * Includes all roles (system, user, assistant, tool). + */ +export function buildStructuredInputMessages( + arr: OpenAIInputMessage[], +): InputMessages { + const messages: ChatMessage[] = []; + + for (const msg of arr) { + if (!msg || typeof msg !== "object") continue; + + const role = mapOpenAIRole(msg.role ?? "user"); + let parts: MessagePart[]; + + if (typeof msg.content === "string") { + parts = [{ type: "text", content: msg.content }]; + } else if (Array.isArray(msg.content)) { + parts = (msg.content as Record[]).map(mapInputContentBlock); + } else { + parts = [{ type: "text", content: safeJsonDumps(msg.content) }]; + } + + messages.push({ role, parts }); + } + + return { version: A365_MESSAGE_SCHEMA_VERSION, messages }; +} + +/** + * Build structured OutputMessages from an OpenAI response.output array. + */ +export function buildStructuredOutputMessages( + arr: OpenAIOutputItem[], +): OutputMessages { + const messages: OutputMessage[] = []; + + for (const item of arr) { + if (!item || typeof item !== "object") continue; + + const role = mapOpenAIRole(item.role ?? "assistant"); + + // Items with a content array (standard response format) + if (Array.isArray(item.content)) { + const parts = (item.content as Record[]).map(mapOutputContentBlock); + messages.push({ role, parts }); continue; } - for (const { type, text } of content) { - if (type === "output_text" && typeof text === "string") { - userTexts.push(text); - } + + // Items that are themselves content blocks (e.g., type: 'message' with text) + if (item.type && typeof item.type === "string") { + const parts = [mapOutputContentBlock(item as Record)]; + messages.push({ role, parts }); + continue; } + + // Fallback: stringify the item + messages.push({ + role, + parts: [{ type: "text", content: safeJsonDumps(item) }], + }); } - return JSON.stringify(userTexts.length ? userTexts : arr); + + return { version: A365_MESSAGE_SCHEMA_VERSION, messages }; +} + +/** + * Wrap opaque raw content as InputMessages (for generation span data). + */ +export function wrapRawContentAsInputMessages(raw: unknown): InputMessages { + return wrapRawContentAsMessages(raw, MessageRole.USER) as InputMessages; +} + +/** + * Wrap opaque raw content as OutputMessages (for generation span data). + */ +export function wrapRawContentAsOutputMessages(raw: unknown): OutputMessages { + return wrapRawContentAsMessages(raw, MessageRole.ASSISTANT) as OutputMessages; } diff --git a/src/genai/semconv.ts b/src/genai/semconv.ts index 0e8fde9..13a16ce 100644 --- a/src/genai/semconv.ts +++ b/src/genai/semconv.ts @@ -47,6 +47,12 @@ export const ATTR_GEN_AI_TOOL_CALL_ARGUMENTS = "gen_ai.tool.call.arguments" as c export const ATTR_GEN_AI_TOOL_CALL_RESULT = "gen_ai.tool.call.result" as const; export const ATTR_GEN_AI_TOOL_TYPE = "gen_ai.tool.type" as const; +// GenAI agent-to-agent caller +export const ATTR_GEN_AI_CALLER_AGENT_NAME = "microsoft.a365.caller.agent.name" as const; + +// GenAI conversation +export const ATTR_GEN_AI_CONVERSATION_ID = "gen_ai.conversation.id" as const; + // Microsoft-specific (not in OTel semconv) export const ATTR_MICROSOFT_SESSION_ID = "microsoft.session.id" as const; export const ATTR_MICROSOFT_TENANT_ID = "microsoft.tenant.id" as const; diff --git a/src/types.ts b/src/types.ts index cc9a8f6..79dbe55 100644 --- a/src/types.ts +++ b/src/types.ts @@ -128,18 +128,11 @@ export interface OpenAIAgentsInstrumentationConfig extends InstrumentationConfig * @default false */ suppressInvokeAgentInput?: boolean; - /** - * Enable recording of message content (input/output messages, tool args, etc.) in spans. - * @default false - */ - isContentRecordingEnabled?: boolean; } /** Configuration for LangChain instrumentation. */ -export interface LangChainInstrumentationConfig extends InstrumentationConfig { - /** Enable recording of message content in spans. */ - isContentRecordingEnabled?: boolean; -} +// eslint-disable-next-line @typescript-eslint/no-empty-object-type +export interface LangChainInstrumentationConfig extends InstrumentationConfig {} /** * Statsbeat Features Configuration interface diff --git a/test/internal/unit/genai/langchain/tracer.test.ts b/test/internal/unit/genai/langchain/tracer.test.ts index c312949..25799a3 100644 --- a/test/internal/unit/genai/langchain/tracer.test.ts +++ b/test/internal/unit/genai/langchain/tracer.test.ts @@ -195,13 +195,13 @@ describe("LangChainTracer", () => { assert.strictEqual(attrs?.[ATTR_GEN_AI_PROVIDER_NAME], "langchain"); }); - it("sets span kind to INTERNAL", async () => { + it("sets span kind to CLIENT for LLM runs", async () => { const tracer = createMockTracer(); const lct = new LangChainTracer(tracer); const run = makeRun(); await lct.onRunCreate(run); const kind = (tracer.startSpan as ReturnType).mock.calls[0][1]?.kind; - assert.strictEqual(kind, SpanKind.INTERNAL); + assert.strictEqual(kind, SpanKind.CLIENT); }); }); @@ -246,28 +246,9 @@ describe("LangChainTracer", () => { ); }); - it("does not set content attributes when content recording is disabled", async () => { + it("sets content attributes (always recorded)", async () => { const tracer = createMockTracer(); - const lct = new LangChainTracer(tracer, { isContentRecordingEnabled: false }); - const run = makeRun({ - run_type: "tool", - name: "my_tool", - serialized: { name: "my_tool" }, - inputs: { input: "test" }, - }); - await lct.onRunCreate(run); - const span = tracer.lastSpan!; - await (lct as unknown as { _endTrace(run: Run): Promise })._endTrace(run); - const attrKeys = (span.setAttribute as ReturnType).mock.calls.map( - (c: unknown[]) => c[0], - ); - assert.ok(!attrKeys.includes("gen_ai.tool.call.arguments"), "should not set tool arguments"); - assert.ok(!attrKeys.includes("gen_ai.input.messages"), "should not set input messages"); - }); - - it("sets content attributes when content recording is enabled", async () => { - const tracer = createMockTracer(); - const lct = new LangChainTracer(tracer, { isContentRecordingEnabled: true }); + const lct = new LangChainTracer(tracer); const run = makeRun({ run_type: "tool", name: "my_tool", diff --git a/test/internal/unit/genai/langchain/utils.test.ts b/test/internal/unit/genai/langchain/utils.test.ts index 3394954..17bff74 100644 --- a/test/internal/unit/genai/langchain/utils.test.ts +++ b/test/internal/unit/genai/langchain/utils.test.ts @@ -34,6 +34,7 @@ import { ATTR_GEN_AI_USAGE_INPUT_TOKENS, ATTR_GEN_AI_USAGE_OUTPUT_TOKENS, ATTR_MICROSOFT_SESSION_ID, + ATTR_GEN_AI_CONVERSATION_ID, GEN_AI_OPERATION_CHAT, GEN_AI_OPERATION_EXECUTE_TOOL, GEN_AI_OPERATION_INVOKE_AGENT, @@ -227,9 +228,9 @@ describe("setInputMessagesAttribute", () => { const msgCall = calls.find((c: unknown[]) => c[0] === ATTR_GEN_AI_INPUT_MESSAGES); assert.ok(msgCall, "should set input messages"); const parsed = JSON.parse(msgCall![1] as string); - assert.ok(parsed.includes("Hello")); - // Assistant messages should be filtered out for input - assert.ok(!parsed.includes("Hi there")); + assert.ok(JSON.stringify(parsed).includes("Hello")); + // Both messages should be included in structured input + assert.ok(JSON.stringify(parsed).includes("Hi there")); }); it("extracts LangChain lc_kwargs format", () => { @@ -244,7 +245,7 @@ describe("setInputMessagesAttribute", () => { const calls = (span.setAttribute as ReturnType).mock.calls; const msgCall = calls.find((c: unknown[]) => c[0] === ATTR_GEN_AI_INPUT_MESSAGES); assert.ok(msgCall); - assert.ok(JSON.parse(msgCall![1] as string).includes("What is 2+2?")); + assert.ok(JSON.stringify(JSON.parse(msgCall![1] as string)).includes("What is 2+2?")); }); it("extracts messages using id array-based type detection", () => { @@ -268,7 +269,7 @@ describe("setInputMessagesAttribute", () => { const calls = (span.setAttribute as ReturnType).mock.calls; const msgCall = calls.find((c: unknown[]) => c[0] === ATTR_GEN_AI_INPUT_MESSAGES); assert.ok(msgCall); - assert.ok(JSON.parse(msgCall![1] as string).includes("Build this")); + assert.ok(JSON.stringify(JSON.parse(msgCall![1] as string)).includes("Build this")); }); it("does nothing when messages is not an array", () => { @@ -292,7 +293,7 @@ describe("setOutputMessagesAttribute", () => { const calls = (span.setAttribute as ReturnType).mock.calls; const msgCall = calls.find((c: unknown[]) => c[0] === ATTR_GEN_AI_OUTPUT_MESSAGES); assert.ok(msgCall); - assert.ok(JSON.parse(msgCall![1] as string).includes("Here is the answer")); + assert.ok(JSON.stringify(JSON.parse(msgCall![1] as string)).includes("Here is the answer")); }); it("extracts output from generations format", () => { @@ -313,7 +314,7 @@ describe("setOutputMessagesAttribute", () => { const calls = (span.setAttribute as ReturnType).mock.calls; const msgCall = calls.find((c: unknown[]) => c[0] === ATTR_GEN_AI_OUTPUT_MESSAGES); assert.ok(msgCall); - assert.ok(JSON.parse(msgCall![1] as string).includes("Generated text")); + assert.ok(JSON.stringify(JSON.parse(msgCall![1] as string)).includes("Generated text")); }); it("extracts output from single message object", () => { @@ -327,7 +328,7 @@ describe("setOutputMessagesAttribute", () => { const calls = (span.setAttribute as ReturnType).mock.calls; const msgCall = calls.find((c: unknown[]) => c[0] === ATTR_GEN_AI_OUTPUT_MESSAGES); assert.ok(msgCall); - assert.ok(JSON.parse(msgCall![1] as string).includes("Single response")); + assert.ok(JSON.stringify(JSON.parse(msgCall![1] as string)).includes("Single response")); }); it("does nothing when outputs is undefined", () => { @@ -420,13 +421,13 @@ describe("setSessionIdAttribute", () => { ); }); - it("falls back to conversation_id", () => { + it("sets conversation_id as separate attribute", () => { const span = makeSpan(); const run = makeRun({ extra: { metadata: { conversation_id: "conv-456" } } }); setSessionIdAttribute(run, span); assert.ok( (span.setAttribute as ReturnType).mock.calls.some( - (c: unknown[]) => c[0] === ATTR_MICROSOFT_SESSION_ID && c[1] === "conv-456", + (c: unknown[]) => c[0] === ATTR_GEN_AI_CONVERSATION_ID && c[1] === "conv-456", ), ); }); diff --git a/test/internal/unit/genai/openai/openAIAgentsTraceProcessor.test.ts b/test/internal/unit/genai/openai/openAIAgentsTraceProcessor.test.ts index 3acb532..5836680 100644 --- a/test/internal/unit/genai/openai/openAIAgentsTraceProcessor.test.ts +++ b/test/internal/unit/genai/openai/openAIAgentsTraceProcessor.test.ts @@ -79,9 +79,7 @@ describe("OpenAIAgentsTraceProcessor", () => { beforeEach(() => { mockSpan = makeMockOtelSpan(); tracer = makeMockTracer(mockSpan); - processor = new OpenAIAgentsTraceProcessor(tracer, { - isContentRecordingEnabled: true, - }); + processor = new OpenAIAgentsTraceProcessor(tracer); }); describe("start / shutdown", () => { @@ -155,7 +153,7 @@ describe("OpenAIAgentsTraceProcessor", () => { await processor.onSpanStart(span); await processor.onSpanEnd(span); - assert.strictEqual(mockSpan.attrs["graph_node_id"], "MyAgent"); + assert.strictEqual(mockSpan.attrs["gen_ai.agent.name"], "MyAgent"); assert.strictEqual(mockSpan.attrs["gen_ai.operation.name"], "invoke_agent"); }); }); @@ -241,32 +239,28 @@ describe("OpenAIAgentsTraceProcessor", () => { const agentOtelSpan = tracer.spans[tracer.spans.length - 1] as OtelSpan & { attrs: Record; }; - assert.strictEqual(agentOtelSpan.attrs["graph_node_parent_id"], "AgentA"); + assert.strictEqual(agentOtelSpan.attrs["microsoft.a365.caller.agent.name"], "AgentA"); }); }); describe("content recording", () => { - it("does not record content when disabled", async () => { - const noContentProcessor = new OpenAIAgentsTraceProcessor(tracer, { - isContentRecordingEnabled: false, - }); - + it("always records content", async () => { const span = makeAgentsSpan({ spanData: { type: "generation", model: "gpt-4o", - input: "secret input", - output: "secret output", + input: "some input", + output: "some output", }, }); - await noContentProcessor.onSpanStart(span); - await noContentProcessor.onSpanEnd(span); + await processor.onSpanStart(span); + await processor.onSpanEnd(span); - // Model should still be set + // Model should be set assert.strictEqual(mockSpan.attrs["gen_ai.request.model"], "gpt-4o"); - // Content should NOT be set - assert.strictEqual(mockSpan.attrs["gen_ai.input.messages"], undefined); - assert.strictEqual(mockSpan.attrs["gen_ai.output.messages"], undefined); + // Content should also be set (always recorded now) + assert.ok(mockSpan.attrs["gen_ai.input.messages"] !== undefined); + assert.ok(mockSpan.attrs["gen_ai.output.messages"] !== undefined); }); }); diff --git a/test/internal/unit/genai/openai/utils.test.ts b/test/internal/unit/genai/openai/utils.test.ts index 85807fb..356c651 100644 --- a/test/internal/unit/genai/openai/utils.test.ts +++ b/test/internal/unit/genai/openai/utils.test.ts @@ -13,9 +13,8 @@ import { getAttributesFromFunctionSpanData, getAttributesFromMCPListToolsSpanData, getAttributesFromResponse, - buildInputMessages, - buildOutputMessages, - CONTENT_KEYS, + buildStructuredInputMessages, + buildStructuredOutputMessages, KEY_MAPPINGS, } from "../../../../../src/genai/instrumentations/openai/utils.js"; import { @@ -167,8 +166,11 @@ describe("OpenAI Utils", () => { output: "world", } as unknown as SpanData; const attrs = getAttributesFromGenerationSpanData(data); - assert.strictEqual(attrs[GEN_AI_REQUEST_CONTENT_KEY], '"hello"'); - assert.strictEqual(attrs[GEN_AI_RESPONSE_CONTENT_KEY], '"world"'); + // Now serialized as structured messages + assert.ok(typeof attrs[GEN_AI_REQUEST_CONTENT_KEY] === "string"); + assert.ok((attrs[GEN_AI_REQUEST_CONTENT_KEY] as string).includes("hello")); + assert.ok(typeof attrs[GEN_AI_RESPONSE_CONTENT_KEY] === "string"); + assert.ok((attrs[GEN_AI_RESPONSE_CONTENT_KEY] as string).includes("world")); }); }); @@ -187,8 +189,9 @@ describe("OpenAI Utils", () => { output: "test output", } as unknown as SpanData; const attrs = getAttributesFromFunctionSpanData(data); - assert.strictEqual(attrs[GEN_AI_REQUEST_CONTENT_KEY], "test input"); - assert.strictEqual(attrs[GEN_AI_RESPONSE_CONTENT_KEY], "test output"); + // Now serialized via safeSerializeToJson + assert.ok((attrs[GEN_AI_REQUEST_CONTENT_KEY] as string).includes("test input")); + assert.ok((attrs[GEN_AI_RESPONSE_CONTENT_KEY] as string).includes("test output")); }); it("handles object input", () => { @@ -237,58 +240,25 @@ describe("OpenAI Utils", () => { }); }); - describe("buildInputMessages", () => { - it("extracts user role content", () => { + describe("buildStructuredInputMessages", () => { + it("wraps input messages in structured format", () => { const messages = [ - { role: "system", content: "You are helpful" }, { role: "user", content: "Hello" }, ]; - const result = buildInputMessages(messages); - assert.strictEqual(result, '["Hello"]'); - }); - - it("falls back to full array when no user messages", () => { - const messages = [{ role: "system", content: "System prompt" }]; - const result = buildInputMessages(messages); - assert.strictEqual(result, JSON.stringify(messages)); + const result = buildStructuredInputMessages(messages); + assert.ok(result.version); + assert.ok(result.messages.length > 0); }); }); - describe("buildOutputMessages", () => { - it("extracts output_text content", () => { + describe("buildStructuredOutputMessages", () => { + it("wraps output messages in structured format", () => { const messages = [ - { - role: "assistant", - content: [ - { type: "output_text", text: "Hello there!" }, - { type: "other", text: "ignored" }, - ], - }, + { role: "assistant", content: [{ type: "output_text", text: "Hello there!" }] }, ]; - const result = buildOutputMessages(messages); - assert.strictEqual(result, '["Hello there!"]'); - }); - - it("falls back to full array when no output_text", () => { - const messages = [ - { - role: "assistant", - content: [{ type: "tool_call", text: "" }], - }, - ]; - const result = buildOutputMessages(messages); - assert.strictEqual(result, JSON.stringify(messages)); - }); - }); - - describe("CONTENT_KEYS", () => { - it("contains expected content-sensitive keys", () => { - assert.ok(CONTENT_KEYS.has("gen_ai.input.messages")); - assert.ok(CONTENT_KEYS.has("gen_ai.output.messages")); - assert.ok(CONTENT_KEYS.has("gen_ai.tool.call.arguments")); - assert.ok(CONTENT_KEYS.has("gen_ai.tool.call.result")); - assert.ok(CONTENT_KEYS.has(GEN_AI_REQUEST_CONTENT_KEY)); - assert.ok(CONTENT_KEYS.has(GEN_AI_RESPONSE_CONTENT_KEY)); + const result = buildStructuredOutputMessages(messages); + assert.ok(result.version); + assert.ok(result.messages.length > 0); }); }); diff --git a/test/internal/unit/main.test.ts b/test/internal/unit/main.test.ts index caf74cd..ab562aa 100644 --- a/test/internal/unit/main.test.ts +++ b/test/internal/unit/main.test.ts @@ -1028,7 +1028,6 @@ describe("Main functions", () => { openaiAgents: { enabled: false }, langchain: { enabled: true, - isContentRecordingEnabled: true, }, }, }); @@ -1036,10 +1035,6 @@ describe("Main functions", () => { await vi.waitFor(() => { expect(instrumentSpy).toHaveBeenCalledWith( expect.any(Object), - expect.objectContaining({ - enabled: true, - isContentRecordingEnabled: true, - }), ); }); From 4b01f91ae0a166eb7f3dc6d2e98968ecc2030f44 Mon Sep 17 00:00:00 2001 From: Hector Hernandez <39923391+hectorhdzg@users.noreply.github.com> Date: Mon, 27 Apr 2026 15:31:48 -0700 Subject: [PATCH 2/4] Port upstream PRs #235, #237, #238: auth scope, payload chunking, token cache - PR #235: Update default auth scope to api://9b975845-388f-4429-889e-eab1ef63949c/.default - PR #237: Add payload chunking with maxPayloadBytes option (default 900KB), estimateSpanBytes/chunkBySize utilities, and per-chunk export with all-or-nothing semantics - PR #238: Restore AgenticTokenCache with authHandlerName parameter - Update functional tests to align with PR #75 schema changes (SpanKind, gen_ai.agent.name, content always recorded) --- src/a365/configuration/A365Configuration.ts | 2 +- src/a365/exporter/Agent365Exporter.ts | 116 ++++-- src/a365/exporter/Agent365ExporterOptions.ts | 7 +- src/a365/exporter/utils.ts | 110 ++++++ src/a365/hosting/agenticTokenCache.ts | 275 ++++++++++++++ src/a365/hosting/index.ts | 2 + src/a365/index.ts | 4 + src/index.ts | 4 + test/internal/functional/langchain.test.ts | 48 +-- test/internal/functional/openai.test.ts | 57 +-- .../unit/a365/a365Configuration.test.ts | 2 +- .../unit/a365/agent365Exporter.test.ts | 101 ++++- .../a365/hosting/agenticTokenCache.test.ts | 344 ++++++++++++++++++ 13 files changed, 951 insertions(+), 121 deletions(-) create mode 100644 src/a365/hosting/agenticTokenCache.ts create mode 100644 test/internal/unit/a365/hosting/agenticTokenCache.test.ts diff --git a/src/a365/configuration/A365Configuration.ts b/src/a365/configuration/A365Configuration.ts index 7f7c1a2..e36d003 100644 --- a/src/a365/configuration/A365Configuration.ts +++ b/src/a365/configuration/A365Configuration.ts @@ -36,7 +36,7 @@ export const A365_ENV_VARS = { LOG_LEVEL: "A365_OBSERVABILITY_LOG_LEVEL", } as const; -const DEFAULT_AUTH_SCOPE = "https://api.powerplatform.com/.default"; +const DEFAULT_AUTH_SCOPE = "api://9b975845-388f-4429-889e-eab1ef63949c/.default"; const VALID_CLUSTER_CATEGORIES: ReadonlySet = new Set([ "local", diff --git a/src/a365/exporter/Agent365Exporter.ts b/src/a365/exporter/Agent365Exporter.ts index 71fdd29..fe99923 100644 --- a/src/a365/exporter/Agent365Exporter.ts +++ b/src/a365/exporter/Agent365Exporter.ts @@ -17,6 +17,8 @@ import { statusName, resolveAgent365Endpoint, truncateSpan, + estimateSpanBytes, + chunkBySize, } from "./utils.js"; import { getA365Logger } from "../logging.js"; @@ -69,6 +71,13 @@ interface OTLPStatus { message?: string; } +interface MappedSpan { + span: OTLPSpan; + scopeKey: string; + scopeName: string; + scopeVersion?: string; +} + /** * Agent365 span exporter. * @@ -147,8 +156,20 @@ export class Agent365Exporter implements SpanExporter { const start = Date.now(); const { tenantId, agentId } = parseIdentityKey(identityKey); - const payload = this.buildExportRequest(spans); - const body = JSON.stringify(payload); + // Map, truncate, and chunk spans by estimated byte size + const mappedSpans = this.mapAndTruncateSpans(spans); + const resourceAttrs = this.getResourceAttributes(spans); + const chunks = chunkBySize( + mappedSpans, + (ms) => estimateSpanBytes(ms.span), + this.options.maxPayloadBytes, + ); + + if (chunks.length > 1) { + this.logger.info( + `[Agent365Exporter] Split ${spans.length} spans into ${chunks.length} chunks for ${tenantId}/${agentId}`, + ); + } const servicePrefix = this.options.useS2SEndpoint ? "/observabilityService" : "/observability"; const endpointPath = `${servicePrefix}/tenants/${encodeURIComponent(tenantId)}/otlp/agents/${encodeURIComponent(agentId)}/traces`; @@ -179,22 +200,38 @@ export class Agent365Exporter implements SpanExporter { } headers["authorization"] = `Bearer ${token}`; - const { ok, correlationId } = await this.postWithRetries(url, body, headers); - if (!ok) { - this.logExporterEvent(ExporterEventNames.EXPORT_GROUP, false, Date.now() - start, undefined, { - tenantId, - agentId, - correlationId, - }); - throw new Error(`Failed to export spans for ${tenantId}/${agentId}`); + // Send each chunk (all-or-nothing: fail on first chunk failure) + let lastCorrelationId = "unknown"; + for (let i = 0; i < chunks.length; i++) { + const chunk = chunks[i]; + const payload = this.buildEnvelope(chunk, resourceAttrs); + const body = JSON.stringify(payload); + + this.logger.info( + `[Agent365Exporter] Sending chunk ${i + 1} of ${chunks.length} (${chunk.length} spans)`, + ); + + const { ok, correlationId } = await this.postWithRetries(url, body, headers); + lastCorrelationId = correlationId; + + if (!ok) { + this.logExporterEvent( + ExporterEventNames.EXPORT_GROUP, + false, + Date.now() - start, + `chunk ${i + 1} of ${chunks.length} failed`, + { tenantId, agentId, correlationId }, + ); + throw new Error(`Failed to export spans (chunk ${i + 1} of ${chunks.length})`); + } } this.logExporterEvent( ExporterEventNames.EXPORT_GROUP, true, Date.now() - start, - "Spans exported successfully", - { tenantId, agentId, correlationId }, + `${chunks.length} chunk(s) exported successfully`, + { tenantId, agentId, correlationId: lastCorrelationId }, ); } @@ -262,34 +299,51 @@ export class Agent365Exporter implements SpanExporter { return { ok: false, correlationId: lastCorrelationId }; } - private buildExportRequest(spans: ReadableSpan[]): OTLPExportRequest { + private mapAndTruncateSpans(spans: ReadableSpan[]): MappedSpan[] { + return spans.map((sp) => { + const scope = sp.instrumentationScope; + const scopeName = scope?.name ?? "unknown"; + const scopeVersion = scope?.version ?? ""; + return { + span: truncateSpan(this.mapSpan(sp)), + scopeKey: `${scopeName}:${scopeVersion}`, + scopeName, + scopeVersion: scopeVersion || undefined, + }; + }); + } + + private getResourceAttributes(spans: ReadableSpan[]): Record { + if (spans.length > 0 && spans[0].resource?.attributes) { + return { ...spans[0].resource.attributes }; + } + return {}; + } + + private buildEnvelope( + mappedSpans: MappedSpan[], + resourceAttrs: Record, + ): OTLPExportRequest { const scopeMap = new Map(); - for (const sp of spans) { - const scope = sp.instrumentationScope; - const scopeKey = `${scope?.name ?? "unknown"}:${scope?.version ?? ""}`; - let existing = scopeMap.get(scopeKey); - if (!existing) { - existing = []; - scopeMap.set(scopeKey, existing); - } - existing.push(truncateSpan(this.mapSpan(sp))); + for (const ms of mappedSpans) { + const existing = scopeMap.get(ms.scopeKey) || []; + existing.push(ms.span); + scopeMap.set(ms.scopeKey, existing); } const scopeSpans: ScopeSpan[] = []; - for (const [scopeKey, mappedSpans] of scopeMap) { - const [name, version] = scopeKey.split(":"); + for (const [scopeKey, spans] of scopeMap) { + const representative = mappedSpans.find((ms) => ms.scopeKey === scopeKey)!; scopeSpans.push({ - scope: { name, version: version || undefined }, - spans: mappedSpans, + scope: { + name: representative.scopeName, + version: representative.scopeVersion, + }, + spans, }); } - let resourceAttrs: Record = {}; - if (spans.length > 0 && spans[0].resource?.attributes) { - resourceAttrs = { ...spans[0].resource.attributes }; - } - return { resourceSpans: [ { diff --git a/src/a365/exporter/Agent365ExporterOptions.ts b/src/a365/exporter/Agent365ExporterOptions.ts index 79514c4..02e625a 100644 --- a/src/a365/exporter/Agent365ExporterOptions.ts +++ b/src/a365/exporter/Agent365ExporterOptions.ts @@ -46,6 +46,9 @@ export interface Agent365ExporterOptions { /** Maximum number of spans per export batch. @default 512 */ maxExportBatchSize?: number; + + /** Maximum estimated payload size (bytes) per HTTP chunk. @default 900 * 1024 (900KB) */ + maxPayloadBytes?: number; } /** Resolved options with defaults applied. */ @@ -60,17 +63,19 @@ export class ResolvedExporterOptions { public readonly exporterTimeoutMilliseconds: number; public readonly httpRequestTimeoutMilliseconds: number; public readonly maxExportBatchSize: number; + public readonly maxPayloadBytes: number; constructor(options?: Agent365ExporterOptions) { this.clusterCategory = options?.clusterCategory ?? "prod"; this.tokenResolver = options?.tokenResolver; this.useS2SEndpoint = options?.useS2SEndpoint ?? false; this.domainOverride = options?.domainOverride; - this.authScopes = options?.authScopes ?? ["https://api.powerplatform.com/.default"]; + this.authScopes = options?.authScopes ?? ["api://9b975845-388f-4429-889e-eab1ef63949c/.default"]; this.maxQueueSize = options?.maxQueueSize ?? 2048; this.scheduledDelayMilliseconds = options?.scheduledDelayMilliseconds ?? 5000; this.exporterTimeoutMilliseconds = options?.exporterTimeoutMilliseconds ?? 90000; this.httpRequestTimeoutMilliseconds = options?.httpRequestTimeoutMilliseconds ?? 30000; this.maxExportBatchSize = options?.maxExportBatchSize ?? 512; + this.maxPayloadBytes = options?.maxPayloadBytes ?? 900 * 1024; } } diff --git a/src/a365/exporter/utils.ts b/src/a365/exporter/utils.ts index 5f38f71..df0a2eb 100644 --- a/src/a365/exporter/utils.ts +++ b/src/a365/exporter/utils.ts @@ -419,6 +419,116 @@ function runShrinkPhase( return nextSize; } +// ── Span size estimation and byte-level chunking ──────────────────────────── + +/** Overhead constant for OTLP JSON span fixed fields. @internal */ +const SPAN_BASE_OVERHEAD = 2000; + +/** Overhead per attribute in OTLP JSON format. @internal */ +const ATTR_OVERHEAD = 80; + +/** Overhead per event in OTLP JSON. @internal */ +const EVENT_OVERHEAD = 200; + +/** + * Estimate the serialized byte size of a single attribute value in OTLP/HTTP JSON. + */ +export function estimateValueBytes(value: unknown): number { + if (typeof value === "string") { + return 40 + Buffer.byteLength(value, "utf8"); + } + if (Array.isArray(value)) { + if (value.length === 0) return 60; + if (typeof value[0] === "string") { + let sum = 60; + for (const s of value) { + sum += 40 + Buffer.byteLength(String(s), "utf8"); + } + return sum; + } + return 60 + 50 * value.length; + } + return 40; // bool/int/float/null/other +} + +/** + * Heuristic estimator for the serialized size of an OTLP span in HTTP JSON. + * + * Uses generous constants tuned to over-estimate by ~25-50%, providing headroom + * for JSON serializer variance. + */ +export function estimateSpanBytes(span: { + name?: string; + attributes?: Record | null; + events?: Array<{ name: string; attributes?: Record | null }> | null; +}): number { + let total = SPAN_BASE_OVERHEAD; + + if (span.name) { + total += Buffer.byteLength(span.name, "utf8"); + } + + if (span.attributes) { + for (const [key, value] of Object.entries(span.attributes)) { + total += ATTR_OVERHEAD; + total += Buffer.byteLength(key, "utf8"); + total += estimateValueBytes(value); + } + } + + if (span.events) { + for (const ev of span.events) { + total += EVENT_OVERHEAD; + total += Buffer.byteLength(ev.name, "utf8"); + if (ev.attributes) { + for (const [key, value] of Object.entries(ev.attributes)) { + total += ATTR_OVERHEAD; + total += Buffer.byteLength(key, "utf8"); + total += estimateValueBytes(value); + } + } + } + } + + return total; +} + +/** + * Split items into sub-batches whose cumulative estimated size stays under maxChunkBytes. + * + * Invariants: + * - Input order is preserved across chunks. + * - Empty input produces empty output. + * - A single item whose size exceeds maxChunkBytes forms its own single-item chunk. + * - No chunk is ever empty. + */ +export function chunkBySize( + items: T[], + getSize: (item: T) => number, + maxChunkBytes: number, +): T[][] { + const chunks: T[][] = []; + let current: T[] = []; + let currentBytes = 0; + + for (const item of items) { + const itemBytes = getSize(item); + if (current.length > 0 && currentBytes + itemBytes > maxChunkBytes) { + chunks.push(current); + current = []; + currentBytes = 0; + } + current.push(item); + currentBytes += itemBytes; + } + + if (current.length > 0) { + chunks.push(current); + } + + return chunks; +} + /** * Truncate span attributes if the serialized span exceeds MAX_SPAN_SIZE_BYTES. * diff --git a/src/a365/hosting/agenticTokenCache.ts b/src/a365/hosting/agenticTokenCache.ts new file mode 100644 index 0000000..568ef9d --- /dev/null +++ b/src/a365/hosting/agenticTokenCache.ts @@ -0,0 +1,275 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +/** + * Adapted from microsoft/Agent365-nodejs agents-a365-observability-hosting/src/caching/AgenticTokenCache.ts + */ + +import { getA365Logger } from "../logging.js"; +import type { TurnContextLike } from "./types.js"; + +/** + * Minimal authorization shape required by AgenticTokenCache. + * + * Mirrors the `Authorization` interface from `@microsoft/agents-hosting` + * so the cache can be used without a direct dependency on that package. + */ +export interface AuthorizationLike { + exchangeToken( + turnContext: TurnContextLike, + authHandlerName: string, + options: { scopes: string[] }, + ): Promise<{ token?: string } | undefined>; +} + +interface CacheEntry { + scopes: string[]; + token?: string; + expiresOn?: number; + acquiredOn?: number; +} + +/** + * Cache for agentic authentication tokens used by observability services. + * + * @example + * ```typescript + * // Use the default singleton: + * import { AgenticTokenCacheInstance } from '@microsoft/opentelemetry'; + * + * // Or create an instance with custom scopes: + * const cache = new AgenticTokenCache({ authScopes: ['api://my-scope/.default'] }); + * ``` + */ +export class AgenticTokenCache { + private readonly _map = new Map(); + private readonly _defaultRefreshSkewMs = 60_000; + private readonly _defaultMaxTokenAgeMs = 3_600_000; + private readonly _maxCacheSize = 10_000; + private readonly _maxExpSeconds = 86_400; // 24 hours + private readonly _keyLocks = new Map>(); + private readonly _authScopes: string[]; + + constructor(options?: AgenticTokenCacheOptions) { + const envScopes = process.env.A365_OBSERVABILITY_SCOPES_OVERRIDE?.trim(); + if (envScopes) { + this._authScopes = envScopes.split(/\s+/).filter(Boolean); + } else { + this._authScopes = options?.authScopes ?? ["api://9b975845-388f-4429-889e-eab1ef63949c/.default"]; + } + } + + public static makeKey(agentId: string, tenantId: string): string { + return `${agentId}:${tenantId}`; + } + + public getObservabilityToken(agentId: string, tenantId: string): string | null { + const key = AgenticTokenCache.makeKey(agentId, tenantId); + const entry = this._map.get(key); + // Touch entry for LRU recency + if (entry) { + this._map.delete(key); + this._map.set(key, entry); + } + if (!entry) { + getA365Logger().error(`[AgenticTokenCache] No cache entry for ${key}`); + return null; + } + if (!entry.token) { + getA365Logger().error(`[AgenticTokenCache] No token cached for ${key}`); + return null; + } + if (this.isExpired(entry)) { + getA365Logger().error(`[AgenticTokenCache] Token expired for ${key}`); + return null; + } + return entry.token; + } + + public async refreshObservabilityToken( + agentId: string, + tenantId: string, + turnContext: TurnContextLike, + authorization: AuthorizationLike, + scopes?: string[], + authHandlerName: string = "agentic", + ): Promise { + const key = AgenticTokenCache.makeKey(agentId, tenantId); + if (!authorization) { + throw new Error("[AgenticTokenCache] Authorization not set"); + } + if (!turnContext) { + throw new Error("[AgenticTokenCache] TurnContext not set"); + } + return this.withKeyLock(key, async () => { + let entry = this._map.get(key); + if (!entry) { + const effectiveScopes = scopes && scopes.length > 0 ? [...scopes] : [...this._authScopes]; + if (!Array.isArray(effectiveScopes) || effectiveScopes.length === 0) { + getA365Logger().error("[AgenticTokenCache] No valid scopes"); + return; + } + entry = { scopes: effectiveScopes }; + if (this._map.size >= this._maxCacheSize) { + // Evict least-recently-used (first key in Map insertion order) + const lruKey = this._map.keys().next().value; + if (lruKey !== undefined) { + this._map.delete(lruKey); + } + } + this._map.set(key, entry); + } else { + // Touch for LRU recency + this._map.delete(key); + this._map.set(key, entry); + // Update scopes if caller provided new ones + if (scopes && scopes.length > 0) { + entry.scopes = [...scopes]; + } + } + if (!Array.isArray(entry.scopes) || entry.scopes.length === 0) { + getA365Logger().error("[AgenticTokenCache] Entry has invalid scopes"); + return; + } + + if (entry.token && !this.isExpired(entry)) { + return; + } + + const maxRetries = 2; + for (let attempt = 0; attempt <= maxRetries; attempt++) { + getA365Logger().info( + `[AgenticTokenCache] Exchanging token attempt ${attempt + 1}/${maxRetries + 1}`, + ); + try { + const tokenResponse = await authorization.exchangeToken(turnContext, authHandlerName, { + scopes: entry.scopes, + }); + if (!tokenResponse?.token) { + getA365Logger().error("[AgenticTokenCache] Undefined token returned"); + entry.token = undefined; + entry.expiresOn = undefined; + break; + } + entry.token = tokenResponse.token; + entry.acquiredOn = Date.now(); + const oboExp = this.decodeExp(entry.token); + if (oboExp) { + entry.expiresOn = oboExp * 1000; + } else { + getA365Logger().warn("[AgenticTokenCache] No exp claim, fallback TTL"); + } + getA365Logger().info("[AgenticTokenCache] Token cached"); + return; + } catch (e) { + const retriable = this.isRetriableError(e); + if (retriable && attempt < maxRetries) { + getA365Logger().warn( + `[AgenticTokenCache] Retriable failure attempt ${attempt + 1}`, + e instanceof Error ? e.message : String(e), + ); + await this.sleep(200 * (attempt + 1)); + continue; + } + getA365Logger().error( + "[AgenticTokenCache] Non-retriable failure", + e instanceof Error ? e.message : String(e), + ); + entry.token = undefined; + entry.expiresOn = undefined; + break; + } + } + }); + } + + public invalidateToken(agentId: string, tenantId: string): void { + const entry = this._map.get(AgenticTokenCache.makeKey(agentId, tenantId)); + if (entry) { + entry.token = undefined; + entry.expiresOn = undefined; + } + } + + public invalidateAll(): void { + this._map.clear(); + } + + private decodeExp(jwt: string): number | undefined { + try { + if (!jwt) return undefined; + const parts = jwt.split("."); + if (parts.length < 2) return undefined; + const payloadSegment = parts[1]; + const padded = payloadSegment + "=".repeat((4 - (payloadSegment.length % 4)) % 4); + const json = JSON.parse(Buffer.from(padded, "base64").toString("utf8")) as { + exp?: unknown; + }; + if (typeof json.exp !== "number") return undefined; + const maxExp = Math.floor(Date.now() / 1000) + this._maxExpSeconds; + return Math.min(json.exp, maxExp); + } catch { + return undefined; + } + } + + private isExpired(entry: CacheEntry): boolean { + const now = Date.now(); + if (entry.expiresOn) { + return now >= entry.expiresOn - this._defaultRefreshSkewMs; + } + if (entry.acquiredOn) { + return now >= entry.acquiredOn + this._defaultMaxTokenAgeMs; + } + return true; + } + + private isRetriableError(err: unknown): boolean { + const e = err as { code?: string; status?: number; message?: string } | undefined; + if (!e) return false; + const msg = (e.message || "").toLowerCase(); + if (msg.includes("timeout") || msg.includes("econnreset") || msg.includes("network")) + return true; + if (typeof e.status === "number") { + if (e.status === 408 || e.status === 429) return true; + if (e.status >= 500 && e.status < 600) return true; + } + return false; + } + + private sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); + } + + private withKeyLock(key: string, fn: () => Promise): Promise { + // Chain onto any existing promise for this key so that concurrent + // callers are serialised rather than racing after the same await. + const previous = this._keyLocks.get(key) ?? Promise.resolve(); + const next = previous + .catch(() => { + /* swallow */ + }) + .then(fn); + this._keyLocks.set(key, next); + // Clean up the lock when the chain settles and hasn't been extended. + next.finally(() => { + if (this._keyLocks.get(key) === next) { + this._keyLocks.delete(key); + } + }); + return next; + } +} + +/** + * Options for constructing an AgenticTokenCache instance. + */ +export interface AgenticTokenCacheOptions { + /** OAuth scopes for token exchange. Defaults to the A365 observability scope. */ + authScopes?: string[]; +} + +/** + * Default singleton instance of AgenticTokenCache using the default configuration. + */ +export const AgenticTokenCacheInstance = new AgenticTokenCache(); diff --git a/src/a365/hosting/index.ts b/src/a365/hosting/index.ts index 38bdc0a..8ce1653 100644 --- a/src/a365/hosting/index.ts +++ b/src/a365/hosting/index.ts @@ -19,6 +19,8 @@ export { } from "./outputLoggingMiddleware.js"; export { ObservabilityHostingManager } from "./observabilityHostingManager.js"; export type { ObservabilityHostingOptions } from "./observabilityHostingManager.js"; +export { AgenticTokenCache, AgenticTokenCacheInstance } from "./agenticTokenCache.js"; +export type { AuthorizationLike, AgenticTokenCacheOptions } from "./agenticTokenCache.js"; export { configureA365Hosting } from "./configureA365Hosting.js"; export type { HostingAdapterLike, diff --git a/src/a365/index.ts b/src/a365/index.ts index 4bb206f..2650d0a 100644 --- a/src/a365/index.ts +++ b/src/a365/index.ts @@ -99,10 +99,14 @@ export { A365_PARENT_SPAN_KEY, A365_AUTH_TOKEN_KEY, ObservabilityHostingManager, + AgenticTokenCache, + AgenticTokenCacheInstance, configureA365Hosting, } from "./hosting/index.js"; export type { ObservabilityHostingOptions, + AuthorizationLike, + AgenticTokenCacheOptions, HostingAdapterLike, TurnContextLike, ActivityLike, diff --git a/src/index.ts b/src/index.ts index 463352d..e67108c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -97,10 +97,14 @@ export { A365_PARENT_SPAN_KEY, A365_AUTH_TOKEN_KEY, ObservabilityHostingManager, + AgenticTokenCache, + AgenticTokenCacheInstance, configureA365Hosting, } from "./a365/index.js"; export type { ObservabilityHostingOptions, + AuthorizationLike, + AgenticTokenCacheOptions, HostingAdapterLike, TurnContextLike, ActivityLike, diff --git a/test/internal/functional/langchain.test.ts b/test/internal/functional/langchain.test.ts index d39c38e..594c2cd 100644 --- a/test/internal/functional/langchain.test.ts +++ b/test/internal/functional/langchain.test.ts @@ -38,14 +38,14 @@ let provider: BasicTracerProvider; let exporter: InMemorySpanExporter; let langchainTracer: LangChainTracer; -function setup(options?: { isContentRecordingEnabled?: boolean }) { +function setup() { exporter = new InMemorySpanExporter(); provider = new BasicTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)], }); const tracer = provider.getTracer("test-langchain", "1.0.0"); - langchainTracer = new LangChainTracer(tracer, options); + langchainTracer = new LangChainTracer(tracer); } function makeRun(overrides: Partial = {}): Run { @@ -108,7 +108,7 @@ describe("LangChain Instrumentation Functional Tests", () => { const span = spans[0]; assert.ok(span.name.includes("chat"), `span name "${span.name}" should contain "chat"`); - assert.strictEqual(span.kind, SpanKind.INTERNAL); + assert.strictEqual(span.kind, SpanKind.CLIENT); assert.strictEqual(span.status.code, SpanStatusCode.OK); assert.strictEqual(span.attributes[ATTR_GEN_AI_OPERATION_NAME], GEN_AI_OPERATION_CHAT); assert.strictEqual(span.attributes[ATTR_GEN_AI_REQUEST_MODEL], "gpt-4o"); @@ -120,7 +120,7 @@ describe("LangChain Instrumentation Functional Tests", () => { describe("agent with tool and LLM children", () => { it("produces parent-child spans matching the LangGraph execution flow", async () => { - setup({ isContentRecordingEnabled: true }); + setup(); // 1. Agent (parent) run const agentRun = makeRun({ @@ -245,43 +245,9 @@ describe("LangChain Instrumentation Functional Tests", () => { }); }); - describe("content recording gating", () => { - it("does not record message content when disabled", async () => { - setup({ isContentRecordingEnabled: false }); - - const run = makeRun({ - id: "llm-no-content", - name: "ChatOpenAI", - run_type: "llm", - inputs: { - messages: [[{ role: "user", content: "Secret question" }]], - }, - outputs: { - generations: [[{ text: "Secret answer" }]], - }, - }); - - await langchainTracer.onRunCreate(run); - await (langchainTracer as unknown as { _endTrace(r: Run): Promise })._endTrace(run); - - const spans = exporter.getFinishedSpans(); - assert.strictEqual(spans.length, 1); - - const span = spans[0]; - assert.strictEqual( - span.attributes[ATTR_GEN_AI_INPUT_MESSAGES], - undefined, - "should not record input messages", - ); - assert.strictEqual( - span.attributes[ATTR_GEN_AI_OUTPUT_MESSAGES], - undefined, - "should not record output messages", - ); - }); - - it("records message content when enabled", async () => { - setup({ isContentRecordingEnabled: true }); + describe("content recording", () => { + it("always records message content", async () => { + setup(); const run = makeRun({ id: "llm-with-content", diff --git a/test/internal/functional/openai.test.ts b/test/internal/functional/openai.test.ts index 98e23a0..c0cbab6 100644 --- a/test/internal/functional/openai.test.ts +++ b/test/internal/functional/openai.test.ts @@ -20,6 +20,8 @@ import { import type { Span as AgentsSpan, SpanData } from "@openai/agents-core"; import { OpenAIAgentsTraceProcessor } from "../../../src/genai/instrumentations/openai/openAIAgentsTraceProcessor.js"; import { + ATTR_GEN_AI_AGENT_NAME, + ATTR_GEN_AI_CALLER_AGENT_NAME, ATTR_GEN_AI_OPERATION_NAME, ATTR_GEN_AI_PROVIDER_NAME, ATTR_GEN_AI_REQUEST_MODEL, @@ -37,7 +39,6 @@ let exporter: InMemorySpanExporter; let processor: OpenAIAgentsTraceProcessor; function setup(options?: { - isContentRecordingEnabled?: boolean; suppressInvokeAgentInput?: boolean; }) { exporter = new InMemorySpanExporter(); @@ -76,7 +77,7 @@ afterEach(async () => { describe("OpenAI Agents Instrumentation Functional Tests", () => { describe("single generation (LLM call)", () => { it("produces a span with chat operation and model attributes", async () => { - setup({ isContentRecordingEnabled: true }); + setup(); const span = makeSpan({ spanId: "gen-1", @@ -110,7 +111,7 @@ describe("OpenAI Agents Instrumentation Functional Tests", () => { describe("response span", () => { it("extracts model and usage from response data", async () => { - setup({ isContentRecordingEnabled: true }); + setup(); const span = makeSpan({ spanId: "resp-1", @@ -141,7 +142,7 @@ describe("OpenAI Agents Instrumentation Functional Tests", () => { describe("agent with function tool and generation children", () => { it("produces parent-child spans matching the agent execution flow", async () => { - setup({ isContentRecordingEnabled: true }); + setup(); // 1. Agent span (root) const agentSpan = makeSpan({ @@ -206,7 +207,7 @@ describe("OpenAI Agents Instrumentation Functional Tests", () => { // Verify agent span assert.ok(agentOtel!.name.includes(GEN_AI_OPERATION_INVOKE_AGENT)); - assert.strictEqual(agentOtel!.attributes["graph_node_id"], "ResearchAgent"); + assert.strictEqual(agentOtel!.attributes[ATTR_GEN_AI_AGENT_NAME], "ResearchAgent"); // Verify tool span assert.ok(toolOtel!.name.includes(GEN_AI_OPERATION_EXECUTE_TOOL)); @@ -264,9 +265,9 @@ describe("OpenAI Agents Instrumentation Functional Tests", () => { await processor.onSpanEnd(agentB); const spans = exporter.getFinishedSpans(); - const agentBSpan = spans.find((s) => s.attributes["graph_node_id"] === "AgentB"); + const agentBSpan = spans.find((s) => s.attributes[ATTR_GEN_AI_AGENT_NAME] === "AgentB"); assert.ok(agentBSpan); - assert.strictEqual(agentBSpan!.attributes["graph_node_parent_id"], "AgentA"); + assert.strictEqual(agentBSpan!.attributes[ATTR_GEN_AI_CALLER_AGENT_NAME], "AgentA"); }); }); @@ -302,43 +303,9 @@ describe("OpenAI Agents Instrumentation Functional Tests", () => { }); }); - describe("content recording gating", () => { - it("does not record content when disabled", async () => { - setup({ isContentRecordingEnabled: false }); - - const span = makeSpan({ - spanId: "gen-no-content", - spanData: { - type: "generation", - model: "gpt-4o", - input: "secret input", - output: "secret output", - }, - }); - - await processor.onSpanStart(span); - await processor.onSpanEnd(span); - - const spans = exporter.getFinishedSpans(); - // Root span is also created since no parent - const genSpan = spans.find((s) => s.attributes[ATTR_GEN_AI_REQUEST_MODEL] === "gpt-4o"); - assert.ok(genSpan); - - // Content keys should be absent - assert.strictEqual( - genSpan!.attributes["gen_ai.input.messages"], - undefined, - "should not record input", - ); - assert.strictEqual( - genSpan!.attributes["gen_ai.output.messages"], - undefined, - "should not record output", - ); - }); - - it("records content when enabled", async () => { - setup({ isContentRecordingEnabled: true }); + describe("content recording", () => { + it("always records content", async () => { + setup(); const span = makeSpan({ spanId: "gen-with-content", @@ -385,7 +352,7 @@ describe("OpenAI Agents Instrumentation Functional Tests", () => { describe("suppressInvokeAgentInput", () => { it("suppresses input messages when flag is set", async () => { - setup({ isContentRecordingEnabled: true, suppressInvokeAgentInput: true }); + setup({ suppressInvokeAgentInput: true }); const span = makeSpan({ spanId: "resp-suppress", diff --git a/test/internal/unit/a365/a365Configuration.test.ts b/test/internal/unit/a365/a365Configuration.test.ts index e284c8c..c003097 100644 --- a/test/internal/unit/a365/a365Configuration.test.ts +++ b/test/internal/unit/a365/a365Configuration.test.ts @@ -27,7 +27,7 @@ describe("A365Configuration", () => { assert.strictEqual(config.enabled, false); assert.strictEqual(config.clusterCategory, "prod"); assert.strictEqual(config.domainOverride, undefined); - assert.deepStrictEqual(config.authScopes, ["https://api.powerplatform.com/.default"]); + assert.deepStrictEqual(config.authScopes, ["api://9b975845-388f-4429-889e-eab1ef63949c/.default"]); assert.strictEqual(config.tokenResolver, undefined); }); }); diff --git a/test/internal/unit/a365/agent365Exporter.test.ts b/test/internal/unit/a365/agent365Exporter.test.ts index b745df8..1ab7595 100644 --- a/test/internal/unit/a365/agent365Exporter.test.ts +++ b/test/internal/unit/a365/agent365Exporter.test.ts @@ -15,6 +15,9 @@ import { statusName, resolveAgent365Endpoint, truncateSpan, + estimateSpanBytes, + estimateValueBytes, + chunkBySize, MAX_SPAN_SIZE_BYTES, } from "../../../../src/a365/exporter/utils.js"; import { ResolvedExporterOptions } from "../../../../src/a365/exporter/Agent365ExporterOptions.js"; @@ -489,7 +492,7 @@ describe("Agent365Exporter", () => { infoLines.some( (line) => line.includes("[EVENT]: export-group succeeded in") && - line.includes("Spans exported successfully") && + line.includes("chunk(s) exported successfully") && line.includes(`"tenantId":"${TENANT_ID}"`) && line.includes(`"agentId":"${AGENT_ID}"`), ), @@ -1448,4 +1451,100 @@ describe("Exporter utils", () => { assert.ok(size <= MAX_SPAN_SIZE_BYTES); }); }); + + describe("estimateValueBytes", () => { + it("should estimate string values", () => { + const result = estimateValueBytes("hello"); + assert.ok(result >= 40 + 5); + }); + + it("should estimate empty arrays", () => { + assert.strictEqual(estimateValueBytes([]), 60); + }); + + it("should estimate string arrays", () => { + const result = estimateValueBytes(["a", "bb"]); + assert.ok(result > 60); + }); + + it("should estimate numeric arrays", () => { + const result = estimateValueBytes([1, 2, 3]); + assert.strictEqual(result, 60 + 50 * 3); + }); + + it("should estimate primitives", () => { + assert.strictEqual(estimateValueBytes(42), 40); + assert.strictEqual(estimateValueBytes(true), 40); + assert.strictEqual(estimateValueBytes(null), 40); + }); + }); + + describe("estimateSpanBytes", () => { + it("should include base overhead", () => { + const result = estimateSpanBytes({ name: "test", attributes: null }); + assert.ok(result >= 2000); + }); + + it("should account for attributes", () => { + const small = estimateSpanBytes({ name: "test", attributes: null }); + const large = estimateSpanBytes({ + name: "test", + attributes: { "key1": "value1", "key2": "value2" }, + }); + assert.ok(large > small); + }); + + it("should account for events", () => { + const noEvents = estimateSpanBytes({ name: "test", attributes: null, events: null }); + const withEvents = estimateSpanBytes({ + name: "test", + attributes: null, + events: [{ name: "event1", attributes: null }], + }); + assert.ok(withEvents > noEvents); + }); + }); + + describe("chunkBySize", () => { + it("should return empty array for empty input", () => { + const result = chunkBySize([], () => 100, 1000); + assert.deepStrictEqual(result, []); + }); + + it("should keep all items in one chunk when they fit", () => { + const items = [1, 2, 3]; + const result = chunkBySize(items, () => 100, 1000); + assert.strictEqual(result.length, 1); + assert.deepStrictEqual(result[0], [1, 2, 3]); + }); + + it("should split into multiple chunks when items exceed limit", () => { + const items = [1, 2, 3, 4, 5]; + const result = chunkBySize(items, () => 300, 500); + assert.ok(result.length > 1); + const flat = result.flat(); + assert.deepStrictEqual(flat, [1, 2, 3, 4, 5]); + }); + + it("should preserve order across chunks", () => { + const items = [10, 20, 30, 40]; + const result = chunkBySize(items, () => 400, 500); + const flat = result.flat(); + assert.deepStrictEqual(flat, [10, 20, 30, 40]); + }); + + it("should put oversized single item in its own chunk", () => { + const items = ["small", "HUGE"]; + const result = chunkBySize(items, (s) => (s === "HUGE" ? 2000 : 100), 500); + assert.ok(result.some((chunk) => chunk.length === 1 && chunk[0] === "HUGE")); + }); + + it("should never produce empty chunks", () => { + const items = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]; + const result = chunkBySize(items, () => 150, 500); + for (const chunk of result) { + assert.ok(chunk.length > 0); + } + }); + }); }); diff --git a/test/internal/unit/a365/hosting/agenticTokenCache.test.ts b/test/internal/unit/a365/hosting/agenticTokenCache.test.ts new file mode 100644 index 0000000..543a57c --- /dev/null +++ b/test/internal/unit/a365/hosting/agenticTokenCache.test.ts @@ -0,0 +1,344 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +import { describe, it, expect, vi, beforeEach } from "vitest"; + +import { + AgenticTokenCache, + type AuthorizationLike, +} from "../../../../../src/a365/hosting/agenticTokenCache.js"; +import type { TurnContextLike } from "../../../../../src/a365/hosting/types.js"; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +function makeTurnContext(): TurnContextLike { + return { + activity: { type: "message" }, + turnState: new Map(), + }; +} + +/** Build a minimal JWT with a given `exp` claim (seconds since epoch). */ +function makeJwtWithExp(exp: number): string { + const header = Buffer.from(JSON.stringify({ alg: "none" })).toString("base64url"); + const payload = Buffer.from(JSON.stringify({ exp })).toString("base64url"); + return `${header}.${payload}.sig`; +} + +function makeAuthMock( + tokenOrFn?: string | (() => Promise<{ token?: string } | undefined>), +): AuthorizationLike { + const impl = + typeof tokenOrFn === "function" + ? tokenOrFn + : async () => (tokenOrFn !== undefined ? { token: tokenOrFn } : undefined); + return { exchangeToken: vi.fn(impl) }; +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +describe("AgenticTokenCache", () => { + let cache: AgenticTokenCache; + + beforeEach(() => { + cache = new AgenticTokenCache(); + }); + + // ── Basic get / refresh ────────────────────────────────────────────────── + + it("returns null when no entry exists", () => { + expect(cache.getObservabilityToken("a", "t")).toBeNull(); + }); + + it("exchanges and caches token on first call", async () => { + const exp = Math.floor(Date.now() / 1000) + 3600; + const jwt = makeJwtWithExp(exp); + const auth = makeAuthMock(jwt); + + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth); + + expect(auth.exchangeToken).toHaveBeenCalledTimes(1); + expect(cache.getObservabilityToken("a", "t")).toBe(jwt); + }); + + it("does not re-exchange when token is still valid", async () => { + const exp = Math.floor(Date.now() / 1000) + 3600; + const jwt = makeJwtWithExp(exp); + const auth = makeAuthMock(jwt); + + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth); + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth); + + expect(auth.exchangeToken).toHaveBeenCalledTimes(1); + }); + + // ── Retry on retriable errors ──────────────────────────────────────────── + + it("retries on retriable error then succeeds", async () => { + const exp = Math.floor(Date.now() / 1000) + 3600; + const jwt = makeJwtWithExp(exp); + let callCount = 0; + const auth = makeAuthMock(async () => { + callCount++; + if (callCount === 1) throw Object.assign(new Error("timeout"), { status: 503 }); + return { token: jwt }; + }); + + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth); + + expect(auth.exchangeToken).toHaveBeenCalledTimes(2); + expect(cache.getObservabilityToken("a", "t")).toBe(jwt); + }); + + it("stops on non-retriable error and leaves token null", async () => { + const auth = makeAuthMock(async () => { + throw new Error("forbidden"); + }); + + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth); + + expect(cache.getObservabilityToken("a", "t")).toBeNull(); + }); + + // ── Expiry / refresh behaviour ─────────────────────────────────────────── + + it("treats near-expiry token as expired (skew refresh)", async () => { + // Token that expires in 30 seconds (within the 60 s skew window) + const exp = Math.floor(Date.now() / 1000) + 30; + const jwt = makeJwtWithExp(exp); + const freshJwt = makeJwtWithExp(Math.floor(Date.now() / 1000) + 7200); + let call = 0; + const auth = makeAuthMock(async () => { + call++; + return { token: call === 1 ? jwt : freshJwt }; + }); + + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth); + // Token is within skew, so getObservabilityToken returns null + expect(cache.getObservabilityToken("a", "t")).toBeNull(); + + // Refreshing again should exchange a new token + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth); + expect(auth.exchangeToken).toHaveBeenCalledTimes(2); + expect(cache.getObservabilityToken("a", "t")).toBe(freshJwt); + }); + + it("uses fallback TTL when JWT has no exp claim", async () => { + const header = Buffer.from(JSON.stringify({ alg: "none" })).toString("base64url"); + const payload = Buffer.from(JSON.stringify({ sub: "test" })).toString("base64url"); + const jwt = `${header}.${payload}.sig`; + const auth = makeAuthMock(jwt); + + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth); + // Token should still be cached (acquiredOn-based TTL is 1 hour) + expect(cache.getObservabilityToken("a", "t")).toBe(jwt); + }); + + it("caps JWT exp claim to 24 hours", async () => { + const farFuture = Math.floor(Date.now() / 1000) + 200_000; // ~55 hours + const jwt = makeJwtWithExp(farFuture); + const auth = makeAuthMock(jwt); + + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth); + expect(cache.getObservabilityToken("a", "t")).toBe(jwt); + }); + + // ── Invalidation ───────────────────────────────────────────────────────── + + it("invalidateToken clears a single entry", async () => { + const jwt = makeJwtWithExp(Math.floor(Date.now() / 1000) + 3600); + const auth = makeAuthMock(jwt); + + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth); + expect(cache.getObservabilityToken("a", "t")).toBe(jwt); + + cache.invalidateToken("a", "t"); + expect(cache.getObservabilityToken("a", "t")).toBeNull(); + }); + + it("invalidateAll clears every entry", async () => { + const jwt = makeJwtWithExp(Math.floor(Date.now() / 1000) + 3600); + const auth = makeAuthMock(jwt); + + await cache.refreshObservabilityToken("a1", "t1", makeTurnContext(), auth); + await cache.refreshObservabilityToken("a2", "t2", makeTurnContext(), auth); + + cache.invalidateAll(); + expect(cache.getObservabilityToken("a1", "t1")).toBeNull(); + expect(cache.getObservabilityToken("a2", "t2")).toBeNull(); + }); + + // ── Scopes handling ────────────────────────────────────────────────────── + + it("updates scopes on existing entry when caller provides new scopes", async () => { + const jwt = makeJwtWithExp(Math.floor(Date.now() / 1000) + 3600); + const auth = makeAuthMock(jwt); + + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth, ["scope1"]); + // Invalidate and refresh with different scopes + cache.invalidateToken("a", "t"); + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth, ["scope2"]); + + const lastCall = (auth.exchangeToken as ReturnType).mock.calls.at(-1); + expect(lastCall?.[2]).toEqual({ scopes: ["scope2"] }); + }); + + it("clones caller-provided scopes to prevent external mutation", async () => { + const jwt = makeJwtWithExp(Math.floor(Date.now() / 1000) + 3600); + const auth = makeAuthMock(jwt); + const scopes = ["original"]; + + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth, scopes); + scopes[0] = "mutated"; + + // Invalidate and refresh — should still use original scopes + cache.invalidateToken("a", "t"); + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth); + + const lastCall = (auth.exchangeToken as ReturnType).mock.calls.at(-1); + expect(lastCall?.[2]).toEqual({ scopes: ["original"] }); + }); + + it("passes authHandlerName to exchangeToken when provided", async () => { + const jwt = makeJwtWithExp(Math.floor(Date.now() / 1000) + 3600); + const auth = makeAuthMock(jwt); + + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth, undefined, "custom"); + + expect(auth.exchangeToken).toHaveBeenCalledWith(expect.anything(), "custom", expect.anything()); + }); + + it('defaults authHandlerName to "agentic"', async () => { + const jwt = makeJwtWithExp(Math.floor(Date.now() / 1000) + 3600); + const auth = makeAuthMock(jwt); + + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth); + + expect(auth.exchangeToken).toHaveBeenCalledWith( + expect.anything(), + "agentic", + expect.anything(), + ); + }); + + // ── LRU eviction ───────────────────────────────────────────────────────── + + it("evicts least-recently-used entry when cache exceeds max size", async () => { + // Use a small cache to test eviction + const smallCache = new (class extends AgenticTokenCache { + constructor() { + super(); + // Override private max via Object.defineProperty + Object.defineProperty(this, "_maxCacheSize", { value: 3 }); + } + })(); + + const jwt = makeJwtWithExp(Math.floor(Date.now() / 1000) + 3600); + const auth = makeAuthMock(jwt); + const ctx = makeTurnContext(); + + await smallCache.refreshObservabilityToken("a1", "t", ctx, auth); + await smallCache.refreshObservabilityToken("a2", "t", ctx, auth); + await smallCache.refreshObservabilityToken("a3", "t", ctx, auth); + + // Access a1 to make it recently-used (move it ahead of a2) + smallCache.getObservabilityToken("a1", "t"); + + // Adding a4 should evict a2 (the least recently used) + await smallCache.refreshObservabilityToken("a4", "t", ctx, auth); + + expect(smallCache.getObservabilityToken("a1", "t")).toBe(jwt); + expect(smallCache.getObservabilityToken("a2", "t")).toBeNull(); // evicted + expect(smallCache.getObservabilityToken("a3", "t")).toBe(jwt); + expect(smallCache.getObservabilityToken("a4", "t")).toBe(jwt); + }); + + // ── Per-key locking (serialisation) ────────────────────────────────────── + + it("serialises concurrent refreshes for the same key", async () => { + let concurrency = 0; + let maxConcurrency = 0; + const jwt = makeJwtWithExp(Math.floor(Date.now() / 1000) + 3600); + + const auth: AuthorizationLike = { + exchangeToken: vi.fn(async () => { + concurrency++; + maxConcurrency = Math.max(maxConcurrency, concurrency); + // Simulate async work + await new Promise((r) => setTimeout(r, 50)); + concurrency--; + return { token: jwt }; + }), + }; + + // Invalidate between calls so both attempts actually exchange + const p1 = cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth); + // Kick off a second concurrent refresh for the same key + cache.invalidateToken("a", "t"); + const p2 = cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth); + + await Promise.all([p1, p2]); + + // With proper serialisation, concurrency should never exceed 1 + expect(maxConcurrency).toBe(1); + }); + + it("allows concurrent refreshes for different keys", async () => { + let concurrency = 0; + let maxConcurrency = 0; + const jwt = makeJwtWithExp(Math.floor(Date.now() / 1000) + 3600); + + const auth: AuthorizationLike = { + exchangeToken: vi.fn(async () => { + concurrency++; + maxConcurrency = Math.max(maxConcurrency, concurrency); + await new Promise((r) => setTimeout(r, 50)); + concurrency--; + return { token: jwt }; + }), + }; + + const p1 = cache.refreshObservabilityToken("a1", "t", makeTurnContext(), auth); + const p2 = cache.refreshObservabilityToken("a2", "t", makeTurnContext(), auth); + + await Promise.all([p1, p2]); + + // Different keys should run concurrently + expect(maxConcurrency).toBe(2); + }); + + // ── Constructor / env override ─────────────────────────────────────────── + + it("uses custom authScopes from options", async () => { + const customCache = new AgenticTokenCache({ authScopes: ["custom://scope"] }); + const jwt = makeJwtWithExp(Math.floor(Date.now() / 1000) + 3600); + const auth = makeAuthMock(jwt); + + await customCache.refreshObservabilityToken("a", "t", makeTurnContext(), auth); + + expect(auth.exchangeToken).toHaveBeenCalledWith(expect.anything(), "agentic", { + scopes: ["custom://scope"], + }); + }); + + it("uses default scope when no options provided", async () => { + const jwt = makeJwtWithExp(Math.floor(Date.now() / 1000) + 3600); + const auth = makeAuthMock(jwt); + + await cache.refreshObservabilityToken("a", "t", makeTurnContext(), auth); + + expect(auth.exchangeToken).toHaveBeenCalledWith(expect.anything(), "agentic", { + scopes: ["api://9b975845-388f-4429-889e-eab1ef63949c/.default"], + }); + }); + + // ── Static helper ──────────────────────────────────────────────────────── + + it("makeKey produces agent:tenant format", () => { + expect(AgenticTokenCache.makeKey("agent1", "tenant1")).toBe("agent1:tenant1"); + }); +}); From 4279482fabb13615d51176f2bc8ca9a5db79a5b8 Mon Sep 17 00:00:00 2001 From: Hector Hernandez <39923391+hectorhdzg@users.noreply.github.com> Date: Mon, 27 Apr 2026 16:15:43 -0700 Subject: [PATCH 3/4] Port upstream PR #240: Filter A365 exports to only include genAI spans - Add gen_ai.operation.name check in partitionByIdentity() to filter non-genAI spans - Only export spans with known genAI operations (invoke_agent, execute_tool, output_messages, chat, Chat, TextCompletion, GenerateContent) - Add test for filtering spans with missing or unknown gen_ai.operation.name --- src/a365/exporter/utils.ts | 49 +++++++++++++++++-- .../unit/a365/agent365Exporter.test.ts | 33 +++++++++---- 2 files changed, 68 insertions(+), 14 deletions(-) diff --git a/src/a365/exporter/utils.ts b/src/a365/exporter/utils.ts index df0a2eb..12a15f3 100644 --- a/src/a365/exporter/utils.ts +++ b/src/a365/exporter/utils.ts @@ -9,8 +9,13 @@ import { ExporterEventNames } from "./ExporterEventNames.js"; import { ATTR_GEN_AI_AGENT_ID, ATTR_GEN_AI_INPUT_MESSAGES, + ATTR_GEN_AI_OPERATION_NAME, ATTR_GEN_AI_OUTPUT_MESSAGES, ATTR_MICROSOFT_TENANT_ID, + GEN_AI_OPERATION_CHAT, + GEN_AI_OPERATION_EXECUTE_TOOL, + GEN_AI_OPERATION_INVOKE_AGENT, + GEN_AI_OPERATION_OUTPUT_MESSAGES, } from "../../genai/semconv.js"; import { A365_MESSAGE_SCHEMA_VERSION } from "../contracts.js"; @@ -22,21 +27,44 @@ const MESSAGE_ATTR_KEYS: Set = new Set([ const MESSAGE_ROLE_SYSTEM = "system"; +/** + * Known genAI operation names produced by the SDK scopes and auto-instrumentation. + * Only spans whose gen_ai.operation.name matches one of these values are exported. + */ +const GEN_AI_OPERATION_NAMES: ReadonlySet = new Set([ + GEN_AI_OPERATION_INVOKE_AGENT, // 'invoke_agent' + GEN_AI_OPERATION_EXECUTE_TOOL, // 'execute_tool' + GEN_AI_OPERATION_OUTPUT_MESSAGES, // 'output_messages' + GEN_AI_OPERATION_CHAT, // 'chat' + "Chat", // InferenceOperationType.CHAT + "TextCompletion", // InferenceOperationType.TEXT_COMPLETION + "GenerateContent", // InferenceOperationType.GENERATE_CONTENT +]); + /** * Partition spans by (tenantId, agentId) identity pairs. - * Spans missing either attribute are skipped. + * Only genAI spans (those with a known gen_ai.operation.name) are included. */ export function partitionByIdentity(spans: ReadableSpan[]): Map { const groups = new Map(); - let skippedCount = 0; + + let nonGenAICount = 0; + let missingIdentityCount = 0; for (const span of spans) { const attrs = span.attributes || {}; + const operationName = asStr(attrs[ATTR_GEN_AI_OPERATION_NAME]); + + if (!operationName || !GEN_AI_OPERATION_NAMES.has(operationName)) { + nonGenAICount++; + continue; + } + const tenant = asStr(attrs[ATTR_MICROSOFT_TENANT_ID]); const agent = asStr(attrs[ATTR_GEN_AI_AGENT_ID]); if (!tenant || !agent) { - skippedCount++; + missingIdentityCount++; continue; } @@ -49,12 +77,23 @@ export function partitionByIdentity(spans: ReadableSpan[]): Map 0) { + if (nonGenAICount > 0) { Logger.getInstance().info( - `[${ExporterEventNames.EXPORT_PARTITION_SPAN_MISSING_IDENTITY}] ${skippedCount} spans skipped (missing tenant or agent ID)`, + `[Agent365Exporter] ${nonGenAICount} non-genAI spans filtered out`, ); } + if (missingIdentityCount > 0) { + Logger.getInstance().info( + `[${ExporterEventNames.EXPORT_PARTITION_SPAN_MISSING_IDENTITY}] ${missingIdentityCount} spans skipped (missing tenant or agent ID)`, + ); + } + + const skippedCount = nonGenAICount + missingIdentityCount; + Logger.getInstance().info( + `[Agent365Exporter] Partitioned into ${groups.size} identity groups (${skippedCount} spans skipped)`, + ); + return groups; } diff --git a/test/internal/unit/a365/agent365Exporter.test.ts b/test/internal/unit/a365/agent365Exporter.test.ts index 5ffafcc..9269cf5 100644 --- a/test/internal/unit/a365/agent365Exporter.test.ts +++ b/test/internal/unit/a365/agent365Exporter.test.ts @@ -42,6 +42,7 @@ function makeSpan(overrides: Partial = {}): ReadableSpan { attributes: { "microsoft.tenant.id": TENANT_ID, "gen_ai.agent.id": AGENT_ID, + "gen_ai.operation.name": "invoke_agent", }, events: [], links: [], @@ -72,6 +73,7 @@ async function exportAndGetPayload( attributes: { "microsoft.tenant.id": TENANT_ID, "gen_ai.agent.id": AGENT_ID, + "gen_ai.operation.name": "invoke_agent", ...attrs, }, }); @@ -146,6 +148,7 @@ describe("Agent365Exporter", () => { attributes: { "microsoft.tenant.id": TENANT_ID, "gen_ai.agent.id": AGENT_ID, + "gen_ai.operation.name": "invoke_agent", "gen_ai.caller.client_ip": "10.0.0.5", }, }); @@ -377,10 +380,10 @@ describe("Agent365Exporter", () => { }); const span1 = makeSpan({ - attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1" }, + attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1", "gen_ai.operation.name": "invoke_agent" }, }); const span2 = makeSpan({ - attributes: { "microsoft.tenant.id": "t2", "gen_ai.agent.id": "a2" }, + attributes: { "microsoft.tenant.id": "t2", "gen_ai.agent.id": "a2", "gen_ai.operation.name": "invoke_agent" }, }); await new Promise((resolve) => { @@ -741,9 +744,9 @@ describe("Exporter utils", () => { describe("partitionByIdentity", () => { it("should group spans by tenant:agent key", () => { const spans = [ - makeSpan({ attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1" } }), - makeSpan({ attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1" } }), - makeSpan({ attributes: { "microsoft.tenant.id": "t2", "gen_ai.agent.id": "a2" } }), + makeSpan({ attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1", "gen_ai.operation.name": "invoke_agent" } }), + makeSpan({ attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1", "gen_ai.operation.name": "chat" } }), + makeSpan({ attributes: { "microsoft.tenant.id": "t2", "gen_ai.agent.id": "a2", "gen_ai.operation.name": "execute_tool" } }), ]; const groups = partitionByIdentity(spans); assert.strictEqual(groups.size, 2); @@ -753,8 +756,8 @@ describe("Exporter utils", () => { it("should skip spans without identity attributes", () => { const spans = [ - makeSpan({ attributes: {} }), - makeSpan({ attributes: { "microsoft.tenant.id": "t1" } }), + makeSpan({ attributes: { "gen_ai.operation.name": "invoke_agent" } }), + makeSpan({ attributes: { "microsoft.tenant.id": "t1", "gen_ai.operation.name": "invoke_agent" } }), ]; const groups = partitionByIdentity(spans); assert.strictEqual(groups.size, 0); @@ -767,12 +770,24 @@ describe("Exporter utils", () => { it("should handle spans with same tenant but different agents", () => { const spans = [ - makeSpan({ attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1" } }), - makeSpan({ attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a2" } }), + makeSpan({ attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1", "gen_ai.operation.name": "invoke_agent" } }), + makeSpan({ attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a2", "gen_ai.operation.name": "invoke_agent" } }), ]; const groups = partitionByIdentity(spans); assert.strictEqual(groups.size, 2); }); + + it("should skip spans with missing or unknown gen_ai.operation.name", () => { + const spans = [ + makeSpan({ name: "genai-span", attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1", "gen_ai.operation.name": "invoke_agent" } }), + makeSpan({ name: "http-span", attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1" } }), + makeSpan({ name: "unknown-op", attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1", "gen_ai.operation.name": "some_random_op" } }), + ]; + const groups = partitionByIdentity(spans); + assert.strictEqual(groups.size, 1); + assert.strictEqual(groups.get("t1:a1")?.length, 1); + assert.strictEqual(groups.get("t1:a1")?.[0].name, "genai-span"); + }); }); describe("parseIdentityKey", () => { From 35d622a2f09c43a6c8ab1c525dfd1755d8475124 Mon Sep 17 00:00:00 2001 From: Hector Hernandez <39923391+hectorhdzg@users.noreply.github.com> Date: Mon, 27 Apr 2026 16:21:37 -0700 Subject: [PATCH 4/4] Fix prettier formatting --- src/a365/exporter/Agent365ExporterOptions.ts | 4 +- src/a365/exporter/utils.ts | 4 +- src/a365/hosting/agenticTokenCache.ts | 4 +- .../langchain/langchainTraceInstrumentor.ts | 4 +- src/genai/instrumentations/langchain/utils.ts | 14 +--- .../openai/openAIAgentsTraceProcessor.ts | 47 +++-------- src/genai/instrumentations/openai/utils.ts | 30 +++---- test/internal/functional/openai.test.ts | 4 +- .../unit/a365/a365Configuration.test.ts | 4 +- .../unit/a365/agent365Exporter.test.ts | 81 ++++++++++++++++--- test/internal/unit/genai/openai/utils.test.ts | 4 +- test/internal/unit/main.test.ts | 4 +- 12 files changed, 104 insertions(+), 100 deletions(-) diff --git a/src/a365/exporter/Agent365ExporterOptions.ts b/src/a365/exporter/Agent365ExporterOptions.ts index 02e625a..123fe8d 100644 --- a/src/a365/exporter/Agent365ExporterOptions.ts +++ b/src/a365/exporter/Agent365ExporterOptions.ts @@ -70,7 +70,9 @@ export class ResolvedExporterOptions { this.tokenResolver = options?.tokenResolver; this.useS2SEndpoint = options?.useS2SEndpoint ?? false; this.domainOverride = options?.domainOverride; - this.authScopes = options?.authScopes ?? ["api://9b975845-388f-4429-889e-eab1ef63949c/.default"]; + this.authScopes = options?.authScopes ?? [ + "api://9b975845-388f-4429-889e-eab1ef63949c/.default", + ]; this.maxQueueSize = options?.maxQueueSize ?? 2048; this.scheduledDelayMilliseconds = options?.scheduledDelayMilliseconds ?? 5000; this.exporterTimeoutMilliseconds = options?.exporterTimeoutMilliseconds ?? 90000; diff --git a/src/a365/exporter/utils.ts b/src/a365/exporter/utils.ts index 12a15f3..01d94db 100644 --- a/src/a365/exporter/utils.ts +++ b/src/a365/exporter/utils.ts @@ -78,9 +78,7 @@ export function partitionByIdentity(spans: ReadableSpan[]): Map 0) { - Logger.getInstance().info( - `[Agent365Exporter] ${nonGenAICount} non-genAI spans filtered out`, - ); + Logger.getInstance().info(`[Agent365Exporter] ${nonGenAICount} non-genAI spans filtered out`); } if (missingIdentityCount > 0) { diff --git a/src/a365/hosting/agenticTokenCache.ts b/src/a365/hosting/agenticTokenCache.ts index 568ef9d..5c3739a 100644 --- a/src/a365/hosting/agenticTokenCache.ts +++ b/src/a365/hosting/agenticTokenCache.ts @@ -55,7 +55,9 @@ export class AgenticTokenCache { if (envScopes) { this._authScopes = envScopes.split(/\s+/).filter(Boolean); } else { - this._authScopes = options?.authScopes ?? ["api://9b975845-388f-4429-889e-eab1ef63949c/.default"]; + this._authScopes = options?.authScopes ?? [ + "api://9b975845-388f-4429-889e-eab1ef63949c/.default", + ]; } } diff --git a/src/genai/instrumentations/langchain/langchainTraceInstrumentor.ts b/src/genai/instrumentations/langchain/langchainTraceInstrumentor.ts index 1264220..71acaec 100644 --- a/src/genai/instrumentations/langchain/langchainTraceInstrumentor.ts +++ b/src/genai/instrumentations/langchain/langchainTraceInstrumentor.ts @@ -145,9 +145,7 @@ export class LangChainTraceInstrumentor { * @param module The CallbackManager module to instrument * @param options Optional configuration options */ - static instrument( - module: CallbackManagerModuleType, - ): void { + static instrument(module: CallbackManagerModuleType): void { LangChainTraceInstrumentorImpl.getInstance().manuallyInstrumentImpl(module); } diff --git a/src/genai/instrumentations/langchain/utils.ts b/src/genai/instrumentations/langchain/utils.ts index 4ec7ab8..44a9a3e 100644 --- a/src/genai/instrumentations/langchain/utils.ts +++ b/src/genai/instrumentations/langchain/utils.ts @@ -25,14 +25,8 @@ import { GEN_AI_OPERATION_EXECUTE_TOOL, GEN_AI_OPERATION_INVOKE_AGENT, } from "../../index.js"; -import { - serializeMessages, - safeSerializeToJson, -} from "../../../a365/message-utils.js"; -import { - MessageRole, - A365_MESSAGE_SCHEMA_VERSION, -} from "../../../a365/contracts.js"; +import { serializeMessages, safeSerializeToJson } from "../../../a365/message-utils.js"; +import { MessageRole, A365_MESSAGE_SCHEMA_VERSION } from "../../../a365/contracts.js"; import type { ChatMessage, OutputMessage, @@ -92,9 +86,7 @@ export function setToolAttributes(run: Run, span: Span) { span.setAttribute( ATTR_GEN_AI_TOOL_CALL_ARGUMENTS, safeSerializeToJson( - typeof argsValue === "object" - ? (argsValue as Record) - : String(argsValue), + typeof argsValue === "object" ? (argsValue as Record) : String(argsValue), "arguments", ), ); diff --git a/src/genai/instrumentations/openai/openAIAgentsTraceProcessor.ts b/src/genai/instrumentations/openai/openAIAgentsTraceProcessor.ts index 98ac601..ee436c9 100644 --- a/src/genai/instrumentations/openai/openAIAgentsTraceProcessor.ts +++ b/src/genai/instrumentations/openai/openAIAgentsTraceProcessor.ts @@ -35,10 +35,7 @@ import { } from "../../index.js"; import { serializeMessages } from "../../../a365/message-utils.js"; -import { - GEN_AI_RESPONSE_CONTENT_KEY, - GEN_AI_EXECUTION_PAYLOAD_KEY, -} from "./semconv.js"; +import { GEN_AI_RESPONSE_CONTENT_KEY, GEN_AI_EXECUTION_PAYLOAD_KEY } from "./semconv.js"; import * as Utils from "./utils.js"; type ContextToken = unknown; @@ -299,17 +296,11 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { const structured = Utils.buildStructuredOutputMessages( resp.output as Array>, ); - otelSpan.setAttribute( - ATTR_GEN_AI_OUTPUT_MESSAGES, - serializeMessages(structured), - ); + otelSpan.setAttribute(ATTR_GEN_AI_OUTPUT_MESSAGES, serializeMessages(structured)); } else { // String or non-array object — wrap as raw content const structured = Utils.wrapRawContentAsOutputMessages(resp.output); - otelSpan.setAttribute( - ATTR_GEN_AI_OUTPUT_MESSAGES, - serializeMessages(structured), - ); + otelSpan.setAttribute(ATTR_GEN_AI_OUTPUT_MESSAGES, serializeMessages(structured)); } } @@ -331,45 +322,29 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { const parsed = JSON.parse(inputObj as string); if (Array.isArray(parsed)) { const structured = Utils.buildStructuredInputMessages(parsed); - otelSpan.setAttribute( - ATTR_GEN_AI_INPUT_MESSAGES, - serializeMessages(structured), - ); + otelSpan.setAttribute(ATTR_GEN_AI_INPUT_MESSAGES, serializeMessages(structured)); return; } } catch { // If parsing fails, wrap raw string in versioned envelope } const wrappedInput = Utils.wrapRawContentAsInputMessages(inputObj); - otelSpan.setAttribute( - ATTR_GEN_AI_INPUT_MESSAGES, - serializeMessages(wrappedInput), - ); + otelSpan.setAttribute(ATTR_GEN_AI_INPUT_MESSAGES, serializeMessages(wrappedInput)); } else if (Array.isArray(inputObj)) { const structured = Utils.buildStructuredInputMessages(inputObj); - otelSpan.setAttribute( - ATTR_GEN_AI_INPUT_MESSAGES, - serializeMessages(structured), - ); + otelSpan.setAttribute(ATTR_GEN_AI_INPUT_MESSAGES, serializeMessages(structured)); } } } - private processGenerationSpanData( - otelSpan: OtelSpan, - data: SpanData, - traceId: string, - ): void { + private processGenerationSpanData(otelSpan: OtelSpan, data: SpanData, traceId: string): void { const attrs = Utils.getAttributesFromGenerationSpanData(data); Object.entries(attrs).forEach(([key, value]) => { const shouldExcludeKey = key === GEN_AI_EXECUTION_PAYLOAD_KEY; if (value !== null && value !== undefined && !shouldExcludeKey) { const newKey = this.getNewKey(data.type, key); const resolvedKey = newKey || key; - if ( - resolvedKey !== ATTR_GEN_AI_INPUT_MESSAGES || - !this.suppressInvokeAgentInput - ) { + if (resolvedKey !== ATTR_GEN_AI_INPUT_MESSAGES || !this.suppressInvokeAgentInput) { otelSpan.setAttribute(resolvedKey, value as string | number | boolean); } } @@ -383,11 +358,7 @@ export class OpenAIAgentsTraceProcessor implements TracingProcessor { } } - private processFunctionSpanData( - otelSpan: OtelSpan, - data: SpanData, - traceId: string, - ): void { + private processFunctionSpanData(otelSpan: OtelSpan, data: SpanData, traceId: string): void { const functionData = data as Record; const attrs = Utils.getAttributesFromFunctionSpanData(data); Object.entries(attrs).forEach(([key, value]) => { diff --git a/src/genai/instrumentations/openai/utils.ts b/src/genai/instrumentations/openai/utils.ts index 2a643cb..bd6cc28 100644 --- a/src/genai/instrumentations/openai/utils.ts +++ b/src/genai/instrumentations/openai/utils.ts @@ -15,14 +15,8 @@ import { GEN_AI_OPERATION_EXECUTE_TOOL, GEN_AI_OPERATION_INVOKE_AGENT, } from "../../index.js"; -import { - serializeMessages, - safeSerializeToJson, -} from "../../../a365/message-utils.js"; -import { - MessageRole, - A365_MESSAGE_SCHEMA_VERSION, -} from "../../../a365/contracts.js"; +import { serializeMessages, safeSerializeToJson } from "../../../a365/message-utils.js"; +import { MessageRole, A365_MESSAGE_SCHEMA_VERSION } from "../../../a365/contracts.js"; import type { ChatMessage, OutputMessage, @@ -47,9 +41,10 @@ import { * - Chat Completions: { prompt_tokens, completion_tokens } * Usage may live directly on the span data, on `.output`, or inside `.output[0]`. */ -export function extractUsageTokens( - data: Record, -): { inputTokens?: number; outputTokens?: number } { +export function extractUsageTokens(data: Record): { + inputTokens?: number; + outputTokens?: number; +} { const candidates: Array | undefined> = []; const direct = data.usage as Record | undefined; candidates.push(direct); @@ -365,10 +360,7 @@ function getToolCallId(block: Record): string | undefined { return undefined; } -function wrapRawContentAsMessages( - raw: unknown, - role: MessageRole, -): InputMessages | OutputMessages { +function wrapRawContentAsMessages(raw: unknown, role: MessageRole): InputMessages | OutputMessages { const content = typeof raw === "string" ? raw : safeJsonDumps(raw); return { version: A365_MESSAGE_SCHEMA_VERSION, @@ -444,9 +436,7 @@ function mapOutputContentBlock(block: Record): MessagePart { * Build structured InputMessages from an OpenAI _input message array. * Includes all roles (system, user, assistant, tool). */ -export function buildStructuredInputMessages( - arr: OpenAIInputMessage[], -): InputMessages { +export function buildStructuredInputMessages(arr: OpenAIInputMessage[]): InputMessages { const messages: ChatMessage[] = []; for (const msg of arr) { @@ -472,9 +462,7 @@ export function buildStructuredInputMessages( /** * Build structured OutputMessages from an OpenAI response.output array. */ -export function buildStructuredOutputMessages( - arr: OpenAIOutputItem[], -): OutputMessages { +export function buildStructuredOutputMessages(arr: OpenAIOutputItem[]): OutputMessages { const messages: OutputMessage[] = []; for (const item of arr) { diff --git a/test/internal/functional/openai.test.ts b/test/internal/functional/openai.test.ts index c0cbab6..edb453e 100644 --- a/test/internal/functional/openai.test.ts +++ b/test/internal/functional/openai.test.ts @@ -38,9 +38,7 @@ let provider: BasicTracerProvider; let exporter: InMemorySpanExporter; let processor: OpenAIAgentsTraceProcessor; -function setup(options?: { - suppressInvokeAgentInput?: boolean; -}) { +function setup(options?: { suppressInvokeAgentInput?: boolean }) { exporter = new InMemorySpanExporter(); provider = new BasicTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)], diff --git a/test/internal/unit/a365/a365Configuration.test.ts b/test/internal/unit/a365/a365Configuration.test.ts index c003097..a47d562 100644 --- a/test/internal/unit/a365/a365Configuration.test.ts +++ b/test/internal/unit/a365/a365Configuration.test.ts @@ -27,7 +27,9 @@ describe("A365Configuration", () => { assert.strictEqual(config.enabled, false); assert.strictEqual(config.clusterCategory, "prod"); assert.strictEqual(config.domainOverride, undefined); - assert.deepStrictEqual(config.authScopes, ["api://9b975845-388f-4429-889e-eab1ef63949c/.default"]); + assert.deepStrictEqual(config.authScopes, [ + "api://9b975845-388f-4429-889e-eab1ef63949c/.default", + ]); assert.strictEqual(config.tokenResolver, undefined); }); }); diff --git a/test/internal/unit/a365/agent365Exporter.test.ts b/test/internal/unit/a365/agent365Exporter.test.ts index 9269cf5..7eea154 100644 --- a/test/internal/unit/a365/agent365Exporter.test.ts +++ b/test/internal/unit/a365/agent365Exporter.test.ts @@ -380,10 +380,18 @@ describe("Agent365Exporter", () => { }); const span1 = makeSpan({ - attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1", "gen_ai.operation.name": "invoke_agent" }, + attributes: { + "microsoft.tenant.id": "t1", + "gen_ai.agent.id": "a1", + "gen_ai.operation.name": "invoke_agent", + }, }); const span2 = makeSpan({ - attributes: { "microsoft.tenant.id": "t2", "gen_ai.agent.id": "a2", "gen_ai.operation.name": "invoke_agent" }, + attributes: { + "microsoft.tenant.id": "t2", + "gen_ai.agent.id": "a2", + "gen_ai.operation.name": "invoke_agent", + }, }); await new Promise((resolve) => { @@ -744,9 +752,27 @@ describe("Exporter utils", () => { describe("partitionByIdentity", () => { it("should group spans by tenant:agent key", () => { const spans = [ - makeSpan({ attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1", "gen_ai.operation.name": "invoke_agent" } }), - makeSpan({ attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1", "gen_ai.operation.name": "chat" } }), - makeSpan({ attributes: { "microsoft.tenant.id": "t2", "gen_ai.agent.id": "a2", "gen_ai.operation.name": "execute_tool" } }), + makeSpan({ + attributes: { + "microsoft.tenant.id": "t1", + "gen_ai.agent.id": "a1", + "gen_ai.operation.name": "invoke_agent", + }, + }), + makeSpan({ + attributes: { + "microsoft.tenant.id": "t1", + "gen_ai.agent.id": "a1", + "gen_ai.operation.name": "chat", + }, + }), + makeSpan({ + attributes: { + "microsoft.tenant.id": "t2", + "gen_ai.agent.id": "a2", + "gen_ai.operation.name": "execute_tool", + }, + }), ]; const groups = partitionByIdentity(spans); assert.strictEqual(groups.size, 2); @@ -757,7 +783,9 @@ describe("Exporter utils", () => { it("should skip spans without identity attributes", () => { const spans = [ makeSpan({ attributes: { "gen_ai.operation.name": "invoke_agent" } }), - makeSpan({ attributes: { "microsoft.tenant.id": "t1", "gen_ai.operation.name": "invoke_agent" } }), + makeSpan({ + attributes: { "microsoft.tenant.id": "t1", "gen_ai.operation.name": "invoke_agent" }, + }), ]; const groups = partitionByIdentity(spans); assert.strictEqual(groups.size, 0); @@ -770,8 +798,20 @@ describe("Exporter utils", () => { it("should handle spans with same tenant but different agents", () => { const spans = [ - makeSpan({ attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1", "gen_ai.operation.name": "invoke_agent" } }), - makeSpan({ attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a2", "gen_ai.operation.name": "invoke_agent" } }), + makeSpan({ + attributes: { + "microsoft.tenant.id": "t1", + "gen_ai.agent.id": "a1", + "gen_ai.operation.name": "invoke_agent", + }, + }), + makeSpan({ + attributes: { + "microsoft.tenant.id": "t1", + "gen_ai.agent.id": "a2", + "gen_ai.operation.name": "invoke_agent", + }, + }), ]; const groups = partitionByIdentity(spans); assert.strictEqual(groups.size, 2); @@ -779,9 +819,26 @@ describe("Exporter utils", () => { it("should skip spans with missing or unknown gen_ai.operation.name", () => { const spans = [ - makeSpan({ name: "genai-span", attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1", "gen_ai.operation.name": "invoke_agent" } }), - makeSpan({ name: "http-span", attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1" } }), - makeSpan({ name: "unknown-op", attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1", "gen_ai.operation.name": "some_random_op" } }), + makeSpan({ + name: "genai-span", + attributes: { + "microsoft.tenant.id": "t1", + "gen_ai.agent.id": "a1", + "gen_ai.operation.name": "invoke_agent", + }, + }), + makeSpan({ + name: "http-span", + attributes: { "microsoft.tenant.id": "t1", "gen_ai.agent.id": "a1" }, + }), + makeSpan({ + name: "unknown-op", + attributes: { + "microsoft.tenant.id": "t1", + "gen_ai.agent.id": "a1", + "gen_ai.operation.name": "some_random_op", + }, + }), ]; const groups = partitionByIdentity(spans); assert.strictEqual(groups.size, 1); @@ -1539,7 +1596,7 @@ describe("Exporter utils", () => { const small = estimateSpanBytes({ name: "test", attributes: null }); const large = estimateSpanBytes({ name: "test", - attributes: { "key1": "value1", "key2": "value2" }, + attributes: { key1: "value1", key2: "value2" }, }); assert.ok(large > small); }); diff --git a/test/internal/unit/genai/openai/utils.test.ts b/test/internal/unit/genai/openai/utils.test.ts index 356c651..3b6989b 100644 --- a/test/internal/unit/genai/openai/utils.test.ts +++ b/test/internal/unit/genai/openai/utils.test.ts @@ -242,9 +242,7 @@ describe("OpenAI Utils", () => { describe("buildStructuredInputMessages", () => { it("wraps input messages in structured format", () => { - const messages = [ - { role: "user", content: "Hello" }, - ]; + const messages = [{ role: "user", content: "Hello" }]; const result = buildStructuredInputMessages(messages); assert.ok(result.version); assert.ok(result.messages.length > 0); diff --git a/test/internal/unit/main.test.ts b/test/internal/unit/main.test.ts index dda679f..0fa5d7f 100644 --- a/test/internal/unit/main.test.ts +++ b/test/internal/unit/main.test.ts @@ -1185,9 +1185,7 @@ describe("Main functions", () => { }); await vi.waitFor(() => { - expect(instrumentSpy).toHaveBeenCalledWith( - expect.any(Object), - ); + expect(instrumentSpy).toHaveBeenCalledWith(expect.any(Object)); }); await shutdownMicrosoftOpenTelemetry();