diff --git a/ts/docs/architecture/telemetry/opentelemetry.md b/ts/docs/architecture/telemetry/opentelemetry.md index 06e524214f..7821326a85 100644 --- a/ts/docs/architecture/telemetry/opentelemetry.md +++ b/ts/docs/architecture/telemetry/opentelemetry.md @@ -141,6 +141,29 @@ restartable or replace host globals. Each subprocess extracts RPC context, creates child spans with its own provider, and exports independently. Telemetry payloads do not pass through the dispatcher. +Browser-shared RPC code reads active TypeAgent metadata through the +`@typeagent/telemetry/traceContext` subpath. It does not import the Node-only +telemetry composition root. + +The trusted dispatcher RPC channel extends the same trace from a client host +into agent-server request processing: + +```text +client rpc.client + -> agent-server rpc.server + -> typeagent.request + -> typeagent.action + -> agent-server rpc.client + -> agent subprocess rpc.server +``` + +Dispatcher RPC propagation remains opt-in. The agent-server client and server +composition roots enable it for their TypeAgent-owned channel. The shared +agent-server dispatcher also enables `telemetry.joinActiveTrace`, which captures +the RPC SERVER context when a command is submitted and restores it when the +queued `typeagent.request` begins. Embedded dispatcher hosts retain the default +`joinActiveTrace: false`, so installing a provider does not implicitly join +their requests to an unrelated active span. ## Signals @@ -289,6 +312,9 @@ OTel owns the canonical trace ID. Preserve the existing caller value as Send TypeAgent correlation values as explicit, allowlisted RPC metadata, not broad W3C baggage. V1 does not inject them into generic HTTP propagation. +The process-backed agent boundary also carries bounded `agentName` and +`actionName` values already known at dispatch time. It never adds action +parameters or RPC-internal context identifiers. Accept remote context only on designated RPC channels. Enforce OTel parsing, size limits, and correlation-field length and character rules. Ignore malformed @@ -840,6 +866,10 @@ Always end manual spans in `finally`. | **5. Metrics** | **Operational visibility** | Token and duration instruments, bounded attributes, in-memory tests, optional Grafana queries | Readers verify values; partner metrics join the host provider; POC can inspect export | | **6. Hardening** | **Production-ready v1** | Provider/no-provider, partner wiring, privacy, compatibility, overhead, queue/disk/export/shutdown failures | Partner and operational tests meet ownership, reliability, privacy, and performance requirements | +Phase 3 is complete. A real three-process OTLP smoke test verifies the literal +client, agent-server, request, action, and agent-subprocess chain above with one +trace ID, exact parent span IDs, and distinct process resources. + Phases normally proceed in order. Phase 4 depends on Phase 3 so the POC shows the Functional MVP. Metrics implementation may start after Phase 0, but the primary sequence integrates it after the POC. diff --git a/ts/packages/agentRpc/README.md b/ts/packages/agentRpc/README.md index b41467a3f6..c0ec959262 100644 --- a/ts/packages/agentRpc/README.md +++ b/ts/packages/agentRpc/README.md @@ -91,7 +91,9 @@ if (rpc) { Every `invoke` creates one `CLIENT` span and one `SERVER` span. One-way `send` notifications are not traced. The request may carry a versioned metadata envelope with bounded W3C `traceparent`/`tracestate` values and the allowlisted TypeAgent -`traceId`, `sessionId`, and `activationId` correlation fields. +`agentName`, `actionName`, `traceId`, `sessionId`, and `activationId` metadata +fields. Values that fail bounds, character validation, or telemetry secret +filtering are omitted. Outbound metadata and inbound trust are separate opt-ins. Enable each only for an approved destination or transport: @@ -116,6 +118,11 @@ composition roots must deliberately thread these options to TypeAgent-owned IPC channels; adding the envelope type alone does not activate cross-process parenting. +Code shared with browser RPC consumers must import active TypeAgent trace +metadata from `@typeagent/telemetry/traceContext`. The main +`@typeagent/telemetry` entry point is Node-only because it includes provider and +exporter lifecycle support. + Cancellation continues to use each application protocol's existing mechanism. For example, agent actions send `cancelAction`, abort the server handler, and cause the original invoke to reject with `AbortError`. The RPC SERVER and CLIENT diff --git a/ts/packages/agentRpc/package.json b/ts/packages/agentRpc/package.json index 58279767f8..2a7cc742e4 100644 --- a/ts/packages/agentRpc/package.json +++ b/ts/packages/agentRpc/package.json @@ -32,6 +32,7 @@ "@opentelemetry/api": "1.9.0", "@typeagent/agent-sdk": "workspace:*", "@typeagent/common-utils": "workspace:*", + "@typeagent/telemetry": "workspace:*", "debug": "^4.4.0" }, "devDependencies": { diff --git a/ts/packages/agentRpc/src/client.ts b/ts/packages/agentRpc/src/client.ts index 3010d73ac8..3811f3b9b3 100644 --- a/ts/packages/agentRpc/src/client.ts +++ b/ts/packages/agentRpc/src/client.ts @@ -30,11 +30,21 @@ import { AgentInvokeFunctions, ContextParams, } from "./types.js"; -import { createRpc } from "./rpc.js"; +import { + createRpc, + type RpcCorrelationFields, + type RpcInvocation, + type RpcOptions, +} from "./rpc.js"; import { ChannelProvider } from "./common.js"; import { getObjectProperty, uint8ArrayToBase64 } from "@typeagent/common-utils"; import { AgentInterfaceFunctionName } from "./server.js"; import { randomUUID } from "crypto"; +import { getActiveTypeAgentSpanAttributes } from "@typeagent/telemetry/traceContext"; + +export type AgentRpcOptions = { + trustedContextPropagation?: boolean; +}; /** * Race a promise against an AbortSignal. If the signal fires before the @@ -147,37 +157,47 @@ function getOptionsFunctions(options?: any): string[] | undefined { return funcs.length > 0 ? funcs : undefined; } -function createOptionsRpc(channelProvider: ChannelProvider, name: string) { +function createOptionsRpc( + channelProvider: ChannelProvider, + name: string, + options?: AgentRpcOptions, +) { const channel = channelProvider.createChannel(`options:${name}`); const optionsMap = createObjectMap(); return { optionsMap, - rpc: createRpc(name, channel, { - callback: async (param: { - id: number; - name: string; - args: any[]; - }) => { - const options: any = optionsMap.get(param.id); - let thisObject: any = undefined; - let fn: (...args: any[]) => any; - const name = param.name; - if (name === "") { - fn = options; - } else { - const names = param.name.split("."); - if (names.length === 1) { - thisObject = options; - fn = options[name]; + rpc: createRpc( + name, + channel, + { + callback: async (param: { + id: number; + name: string; + args: any[]; + }) => { + const options: any = optionsMap.get(param.id); + let thisObject: any = undefined; + let fn: (...args: any[]) => any; + const name = param.name; + if (name === "") { + fn = options; } else { - const funcName = names.pop(); - thisObject = getObjectProperty(options, name); - fn = thisObject[funcName!]; + const names = param.name.split("."); + if (names.length === 1) { + thisObject = options; + fn = options[name]; + } else { + const funcName = names.pop(); + thisObject = getObjectProperty(options, name); + fn = thisObject[funcName!]; + } } - } - return fn.call(thisObject, ...param.args); + return fn.call(thisObject, ...param.args); + }, }, - }), + undefined, + getTrustedRpcOptions(options), + ), }; } @@ -185,6 +205,7 @@ export async function createAgentRpcClient( name: string, channelProvider: ChannelProvider, agentInterface: AgentInterfaceFunctionName[], + options?: AgentRpcOptions, ) { const channel = channelProvider.createChannel(`agent:${name}`); const contextMap = createObjectMap>(); @@ -211,16 +232,16 @@ export async function createAgentRpcClient( const actionContextMap = createObjectMap>(); let optionsRpc: ReturnType | undefined; - function getOptionsCallBack(options?: any) { - const functions = getOptionsFunctions(options); + function getOptionsCallBack(initOptions?: any) { + const functions = getOptionsFunctions(initOptions); if (functions === undefined) { return undefined; } if (optionsRpc === undefined) { - optionsRpc = createOptionsRpc(channelProvider, name); + optionsRpc = createOptionsRpc(channelProvider, name, options); } return { - id: optionsRpc.optionsMap.getId(options), + id: optionsRpc.optionsMap.getId(initOptions), functions, }; } @@ -295,6 +316,7 @@ export async function createAgentRpcClient( param.name, channelProvider, param.agentInterface, + options, ), ); } catch (e: any) { @@ -570,12 +592,21 @@ export async function createAgentRpcClient( }, }; + const rpcOptions = getTrustedRpcOptions(options, (invocation) => + getAgentRpcCorrelation(name, invocation), + ); const rpc = createRpc< AgentInvokeFunctions, AgentCallFunctions, AgentContextInvokeFunctions, AgentContextCallFunctions - >(name, channel, agentContextInvokeHandlers, agentContextCallHandlers); + >( + name, + channel, + agentContextInvokeHandlers, + agentContextCallHandlers, + rpcOptions, + ); // The shim needs to implement all the APIs regardless whether the actual agent // has that API. We remove remove it the one that is not necessary below. @@ -843,3 +874,65 @@ export async function createAgentRpcClient( return result; } + +function getTrustedRpcOptions( + options: AgentRpcOptions | undefined, + getCorrelationFields?: ( + invocation: RpcInvocation, + ) => RpcCorrelationFields | undefined, +): RpcOptions | undefined { + return options?.trustedContextPropagation === true + ? { + tracing: { + propagateContext: true, + trustRemoteContext: true, + ...(getCorrelationFields === undefined + ? undefined + : { getCorrelationFields }), + }, + } + : undefined; +} + +function getAgentRpcCorrelation( + agentName: string, + invocation: RpcInvocation, +): RpcCorrelationFields { + const active = getActiveTypeAgentSpanAttributes(); + return { + agentName, + ...(active?.actionName === undefined + ? getInvocationActionName(invocation) + : { actionName: active.actionName }), + ...(active?.traceId === undefined + ? undefined + : { traceId: active.traceId }), + ...(active?.sessionId === undefined + ? undefined + : { sessionId: active.sessionId }), + ...(active?.activationId === undefined + ? undefined + : { activationId: active.activationId }), + }; +} + +function getInvocationActionName( + invocation: RpcInvocation, +): { actionName: string } | undefined { + if ( + invocation.method !== "executeAction" && + invocation.method !== "validateWildcardMatch" + ) { + return undefined; + } + const param = invocation.args[0]; + if (param === null || typeof param !== "object") { + return undefined; + } + const action = (param as { action?: unknown }).action; + if (action === null || typeof action !== "object") { + return undefined; + } + const actionName = (action as { actionName?: unknown }).actionName; + return typeof actionName === "string" ? { actionName } : undefined; +} diff --git a/ts/packages/agentRpc/src/rpc.ts b/ts/packages/agentRpc/src/rpc.ts index a584cb1bd4..9d4053e5d7 100644 --- a/ts/packages/agentRpc/src/rpc.ts +++ b/ts/packages/agentRpc/src/rpc.ts @@ -3,6 +3,7 @@ import { context, + createContextKey, isSpanContextValid, propagation, ROOT_CONTEXT, @@ -13,6 +14,7 @@ import { type Span, } from "@opentelemetry/api"; import registerDebug from "debug"; +import { filterSecrets } from "@typeagent/common-utils"; import { RpcChannel } from "./common.js"; @@ -45,6 +47,8 @@ type RpcReturn< export const RPC_METADATA_VERSION = 1; export type RpcCorrelationFields = { + agentName?: string; + actionName?: string; traceId?: string; sessionId?: string; activationId?: string; @@ -100,6 +104,9 @@ const MAX_RPC_METHOD_LENGTH = 256; const MAX_TRACESTATE_MEMBERS = 32; const CORRELATION_VALUE_PATTERN = /^[A-Za-z0-9._:@/-]+$/; const RPC_METHOD_PATTERN = /^[A-Za-z0-9._:/-]+$/; +const ACTIVE_RPC_CORRELATION = createContextKey( + "typeagent.rpc.correlation-fields", +); const TRACEPARENT_PATTERN = /^([0-9a-f]{2})-[0-9a-f]{32}-[0-9a-f]{16}-[0-9a-f]{2}(.*)$/; const SIMPLE_TRACESTATE_KEY_PATTERN = /^[a-z][a-z0-9_*/-]{0,255}$/; @@ -236,26 +243,31 @@ export function createRpc< serverInvokes.set(message.callId, serverInvoke); const handler = invokeHandlers?.[message.name]; - const handlerResult = - handler === undefined - ? Promise.resolve({ - kind: "error" as const, - error: new Error( - `No invoke handler ${message.name}`, - ), - }) - : Promise.resolve() - .then(() => handler(...message.args)) - .then( - (result) => ({ - kind: "result" as const, - result, - }), - (error) => ({ - kind: "error" as const, - error, - }), - ); + const handlerResult = context.with( + context + .active() + .setValue(ACTIVE_RPC_CORRELATION, remote.correlation), + () => + handler === undefined + ? Promise.resolve({ + kind: "error" as const, + error: new Error( + `No invoke handler ${message.name}`, + ), + }) + : Promise.resolve() + .then(() => handler(...message.args)) + .then( + (result) => ({ + kind: "result" as const, + result, + }), + (error) => ({ + kind: "error" as const, + error, + }), + ), + ); try { const result = await Promise.race([ @@ -540,6 +552,12 @@ function setCorrelationAttributes( span: Span, correlation: RpcCorrelationFields | undefined, ): void { + if (correlation?.agentName !== undefined) { + span.setAttribute("typeagent.agent.name", correlation.agentName); + } + if (correlation?.actionName !== undefined) { + span.setAttribute("typeagent.action.name", correlation.actionName); + } if (correlation?.traceId !== undefined) { span.setAttribute("typeagent.trace.id", correlation.traceId); } @@ -560,13 +578,29 @@ function getOutboundCorrelation( } try { return validateCorrelationFields( - tracingOptions.getCorrelationFields?.(invocation), + mergeCorrelationFields( + context.active().getValue(ACTIVE_RPC_CORRELATION), + tracingOptions.getCorrelationFields?.(invocation), + ), ); } catch { return undefined; } } +function mergeCorrelationFields( + inherited: unknown, + supplied: RpcCorrelationFields | undefined, +): RpcCorrelationFields | undefined { + if (inherited === null || typeof inherited !== "object") { + return supplied; + } + return { + ...(inherited as RpcCorrelationFields), + ...supplied, + }; +} + function createMetadataEnvelope( correlation: RpcCorrelationFields | undefined, ): RpcMetadataEnvelope | undefined { @@ -684,6 +718,12 @@ function validateCorrelationFields( } const source = value as RpcCorrelationFields; const correlation: RpcCorrelationFields = {}; + if (isValidCorrelationValue(source.agentName)) { + correlation.agentName = source.agentName; + } + if (isValidCorrelationValue(source.actionName)) { + correlation.actionName = source.actionName; + } if (isValidCorrelationValue(source.traceId)) { correlation.traceId = source.traceId; } @@ -697,12 +737,15 @@ function validateCorrelationFields( } function isValidCorrelationValue(value: unknown): value is string { - return ( - typeof value === "string" && - value.length > 0 && - value.length <= MAX_CORRELATION_LENGTH && - CORRELATION_VALUE_PATTERN.test(value) - ); + if ( + typeof value !== "string" || + value.length === 0 || + value.length > MAX_CORRELATION_LENGTH || + !CORRELATION_VALUE_PATTERN.test(value) + ) { + return false; + } + return filterSecrets(value) === value; } function validateTraceparent(value: unknown): value is string { diff --git a/ts/packages/agentRpc/src/server.ts b/ts/packages/agentRpc/src/server.ts index 76ed6b029a..68c2b25093 100644 --- a/ts/packages/agentRpc/src/server.ts +++ b/ts/packages/agentRpc/src/server.ts @@ -31,7 +31,7 @@ import { ContextParams, OptionsFunctionCallBack, } from "./types.js"; -import { createRpc } from "./rpc.js"; +import { createRpc, type RpcOptions } from "./rpc.js"; import { ChannelProvider, RpcChannel } from "./common.js"; import { base64ToUint8Array, @@ -40,11 +40,38 @@ import { setObjectProperty, } from "@typeagent/common-utils"; -function createOptionsRpc(channelProvider: ChannelProvider, name: string) { +export type AgentRpcServerOptions = { + trustedContextPropagation?: boolean; +}; + +function getTrustedRpcOptions( + options: AgentRpcServerOptions | undefined, +): RpcOptions | undefined { + return options?.trustedContextPropagation === true + ? { + tracing: { + propagateContext: true, + trustRemoteContext: true, + }, + } + : undefined; +} + +function createOptionsRpc( + channelProvider: ChannelProvider, + name: string, + options?: AgentRpcServerOptions, +) { const optionsChannel: RpcChannel = channelProvider.createChannel( `options:${name}`, ); - return createRpc(name, optionsChannel); + return createRpc( + name, + optionsChannel, + undefined, + undefined, + getTrustedRpcOptions(options), + ); } function populateOptionsFunctions( @@ -68,6 +95,7 @@ export function createAgentRpcServer( name: string, agent: AppAgent, channelProvider: ChannelProvider, + options?: AgentRpcServerOptions, ) { const channelName = `agent:${name}`; const channel = channelProvider.createChannel(channelName); @@ -98,7 +126,11 @@ export function createAgentRpcServer( ); } if (optionsRpc === undefined) { - optionsRpc = createOptionsRpc(channelProvider, name); + optionsRpc = createOptionsRpc( + channelProvider, + name, + options, + ); } populateOptionsFunctions( optionsRpc, @@ -321,12 +353,13 @@ export function createAgentRpcServer( }, }; + const rpcOptions = getTrustedRpcOptions(options); const rpc = createRpc< AgentContextInvokeFunctions, AgentContextCallFunctions, AgentInvokeFunctions, AgentCallFunctions - >(name, channel, agentInvokeHandlers, agentCallHandlers); + >(name, channel, agentInvokeHandlers, agentCallHandlers, rpcOptions); function getStorage(contextId: number, session: boolean): Storage { const tokenCachePersistence: TokenCachePersistence = { @@ -572,6 +605,7 @@ export function createAgentRpcServer( name, agent, channelProvider, + options, ); // Trigger the addDynamicAgent on the client side const p = rpc.invoke("addDynamicAgent", { diff --git a/ts/packages/agentRpc/test/rpc.spec.ts b/ts/packages/agentRpc/test/rpc.spec.ts index d294c8dfac..d514f633f2 100644 --- a/ts/packages/agentRpc/test/rpc.spec.ts +++ b/ts/packages/agentRpc/test/rpc.spec.ts @@ -421,6 +421,8 @@ describe("createRpc OpenTelemetry propagation", () => { traceId: "legacy-trace", sessionId: "session-1", activationId: "activation-1", + agentName: "player", + actionName: "play", }), }, }, @@ -444,6 +446,8 @@ describe("createRpc OpenTelemetry propagation", () => { "typeagent.trace.id": "legacy-trace", "typeagent.session.id": "session-1", "typeagent.activation.id": "activation-1", + "typeagent.agent.name": "player", + "typeagent.action.name": "play", }); expect(client.sent[0].metadata).toMatchObject({ version: RPC_METADATA_VERSION, @@ -451,6 +455,8 @@ describe("createRpc OpenTelemetry propagation", () => { traceId: "legacy-trace", sessionId: "session-1", activationId: "activation-1", + agentName: "player", + actionName: "play", }, }); }); @@ -598,8 +604,10 @@ describe("createRpc OpenTelemetry propagation", () => { getCorrelationFields: () => ({ sessionId: "session-valid", - traceId: "contains spaces", + traceId: "sk-secret-shaped-identifier", activationId: "x".repeat(257), + agentName: "agent valid", + actionName: "action valid", userText: "must-not-propagate", }) as any, }, diff --git a/ts/packages/agentServer/client/src/agentServerClient.ts b/ts/packages/agentServer/client/src/agentServerClient.ts index 8046a50887..bf898d2a1d 100644 --- a/ts/packages/agentServer/client/src/agentServerClient.ts +++ b/ts/packages/agentServer/client/src/agentServerClient.ts @@ -283,6 +283,9 @@ export function createAgentServerConnection( getDispatcherChannelName(conversationId), ), result.connectionId, + { + trustedContextPropagation: true, + }, ); createClientIORpcServer( diff --git a/ts/packages/agentServer/server/src/connectionHandler.ts b/ts/packages/agentServer/server/src/connectionHandler.ts index e297590140..0ddc0cf171 100644 --- a/ts/packages/agentServer/server/src/connectionHandler.ts +++ b/ts/packages/agentServer/server/src/connectionHandler.ts @@ -319,6 +319,9 @@ export function createAgentServerConnectionHandler( createDispatcherRpcServer( result.dispatcher, dispatcherChannel, + { + trustedContextPropagation: true, + }, ); } catch (e) { channelProvider.deleteChannel( diff --git a/ts/packages/agentServer/server/src/server.ts b/ts/packages/agentServer/server/src/server.ts index 8c9b40a033..beb7e46fb3 100644 --- a/ts/packages/agentServer/server/src/server.ts +++ b/ts/packages/agentServer/server/src/server.ts @@ -351,6 +351,7 @@ async function main() { developerMode, traceId, telemetry: { + joinActiveTrace: true, structuredLogs: telemetryConfig.structuredLogs === true, }, indexingServiceRegistry: await getIndexingServiceRegistry( diff --git a/ts/packages/dispatcher/dispatcher/src/otel/actionSpan.ts b/ts/packages/dispatcher/dispatcher/src/otel/actionSpan.ts index 09a477c009..644f6c1e8c 100644 --- a/ts/packages/dispatcher/dispatcher/src/otel/actionSpan.ts +++ b/ts/packages/dispatcher/dispatcher/src/otel/actionSpan.ts @@ -2,6 +2,7 @@ // Licensed under the MIT License. import { + context, SpanStatusCode, trace, type Span, @@ -75,21 +76,29 @@ export async function wrapActionSpan( otel.TYPEAGENT_SPAN_NAMES.ACTION, async (span) => { otel.setTypeAgentSpanAttributes(span, attributes); - try { - return await body(span); - } catch (error) { - const isAbort = - error !== null && - typeof error === "object" && - (error as { name?: unknown }).name === "AbortError"; - const name = isAbort ? "AbortError" : "ActionError"; - const message = isAbort ? "cancelled" : "action failed"; - span.recordException({ name, message }); - span.setStatus({ code: SpanStatusCode.ERROR, message }); - throw error; - } finally { - span.end(); - } + return context.with( + otel.setActiveTypeAgentSpanAttributes( + context.active(), + attributes, + ), + async () => { + try { + return await body(span); + } catch (error) { + const isAbort = + error !== null && + typeof error === "object" && + (error as { name?: unknown }).name === "AbortError"; + const name = isAbort ? "AbortError" : "ActionError"; + const message = isAbort ? "cancelled" : "action failed"; + span.recordException({ name, message }); + span.setStatus({ code: SpanStatusCode.ERROR, message }); + throw error; + } finally { + span.end(); + } + }, + ); }, ); } diff --git a/ts/packages/dispatcher/dispatcher/test/otelActionSpan.spec.ts b/ts/packages/dispatcher/dispatcher/test/otelActionSpan.spec.ts index ed9cbbad01..da4c90346e 100644 --- a/ts/packages/dispatcher/dispatcher/test/otelActionSpan.spec.ts +++ b/ts/packages/dispatcher/dispatcher/test/otelActionSpan.spec.ts @@ -2,6 +2,7 @@ // Licensed under the MIT License. import { SpanStatusCode, trace } from "@opentelemetry/api"; +import { otel } from "@typeagent/telemetry"; import { createInMemorySpanManager, type CapturedSpan, @@ -105,15 +106,18 @@ describe("wrapActionSpan", () => { it("keeps the action span active through asynchronous work", async () => { let activeSpanId: string | undefined; + let activeAttributes: otel.TypeAgentSpanAttributes | undefined; await wrapActionSpan(ATTRIBUTES, async () => { await Promise.resolve(); activeSpanId = trace.getActiveSpan()?.spanContext().spanId; + activeAttributes = otel.getActiveTypeAgentSpanAttributes(); }); expect(activeSpanId).toBe( getOnlySpan(manager, "typeagent.action").spanContext().spanId, ); + expect(activeAttributes).toEqual(ATTRIBUTES); }); it("records bounded setup and typed-result failures", async () => { diff --git a/ts/packages/dispatcher/dispatcher/test/otelDispatcherContext.spec.ts b/ts/packages/dispatcher/dispatcher/test/otelDispatcherContext.spec.ts new file mode 100644 index 0000000000..f809bd3cc5 --- /dev/null +++ b/ts/packages/dispatcher/dispatcher/test/otelDispatcherContext.spec.ts @@ -0,0 +1,77 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +import { context, createContextKey, type Context } from "@opentelemetry/api"; +import { + createInMemorySpanManager, + type InMemorySpanManager, +} from "@typeagent/telemetry/testing/inMemorySpanManager"; +import { createDispatcherFromContext } from "../src/dispatcher.js"; +import type { CommandHandlerContext } from "../src/context/commandHandlerContext.js"; + +interface CapturedQueueInput { + traceContext?: Context; +} + +function createTestDispatcher(joinActiveTrace: boolean): { + dispatcher: ReturnType; + getCapturedInput(): CapturedQueueInput | undefined; +} { + let capturedInput: CapturedQueueInput | undefined; + const commandContext = { + telemetryOptions: { joinActiveTrace }, + requestQueue: { + submit(input: CapturedQueueInput) { + capturedInput = input; + return { + requestId: "request-1", + originatorConnectionId: "connection-1", + text: "test", + submittedAt: Date.now(), + state: "queued", + completion: Promise.resolve(undefined), + }; + }, + }, + } as unknown as CommandHandlerContext; + return { + dispatcher: createDispatcherFromContext(commandContext, "connection-1"), + getCapturedInput: () => capturedInput, + }; +} + +describe("dispatcher queued trace context", () => { + let spanManager: InMemorySpanManager; + + beforeEach(() => { + spanManager = createInMemorySpanManager(); + }); + + afterEach(async () => { + await spanManager.shutdown(); + }); + + it("captures the active context when the host opts into trace joining", async () => { + const key = createContextKey("dispatcher-rpc-parent"); + const parentContext = context.active().setValue(key, "rpc-server"); + const test = createTestDispatcher(true); + + await context.with(parentContext, () => + test.dispatcher.submitCommand("test"), + ); + + expect(test.getCapturedInput()?.traceContext).toBe(parentContext); + }); + + it("keeps embedded dispatcher requests independent by default", async () => { + const key = createContextKey("embedded-host-parent"); + const parentContext = context.active().setValue(key, "host"); + const test = createTestDispatcher(false); + + await context.with(parentContext, () => + test.dispatcher.submitCommand("test"), + ); + + expect(test.getCapturedInput()?.traceContext).toBeUndefined(); + }); +}); diff --git a/ts/packages/dispatcher/nodeProviders/README.md b/ts/packages/dispatcher/nodeProviders/README.md index a86dd27edc..ae9103b2f6 100644 --- a/ts/packages/dispatcher/nodeProviders/README.md +++ b/ts/packages/dispatcher/nodeProviders/README.md @@ -2,6 +2,17 @@ Node implementation of various dispatcher providers. +## Process-backed agents + +Separate-process agents start through an IPC control-channel handshake. Process +creation resolves only after the child reports its supported agent interface. +If the child exits first, startup rejects with the exit code and signal instead +of leaving the provider waiting indefinitely. + +The process transport explicitly enables trusted RPC trace propagation. The +parent agent server and child agent process retain separate OpenTelemetry +providers and export their own spans. + ## Trademarks This project may contain trademarks or logos for projects, products, or services. Authorized use of Microsoft diff --git a/ts/packages/dispatcher/nodeProviders/package.json b/ts/packages/dispatcher/nodeProviders/package.json index a102c45385..29aecb13e5 100644 --- a/ts/packages/dispatcher/nodeProviders/package.json +++ b/ts/packages/dispatcher/nodeProviders/package.json @@ -40,6 +40,8 @@ "debug": "^4.4.0" }, "devDependencies": { + "@opentelemetry/api": "1.9.0", + "@typeagent/dispatcher-rpc": "workspace:*", "@typeagent/dispatcher-types": "workspace:*", "@types/debug": "^4.1.12", "@types/jest": "^29.5.7", diff --git a/ts/packages/dispatcher/nodeProviders/src/agentProvider/process/agentProcess.ts b/ts/packages/dispatcher/nodeProviders/src/agentProvider/process/agentProcess.ts index 9f5ba91ecb..b602855b1b 100644 --- a/ts/packages/dispatcher/nodeProviders/src/agentProvider/process/agentProcess.ts +++ b/ts/packages/dispatcher/nodeProviders/src/agentProvider/process/agentProcess.ts @@ -111,6 +111,7 @@ async function startAgentProcess(): Promise { agentName, agent, channelProvider, + { trustedContextPropagation: true }, ); const controlChannel = channelProvider.createChannel< diff --git a/ts/packages/dispatcher/nodeProviders/src/agentProvider/process/agentProcessShim.ts b/ts/packages/dispatcher/nodeProviders/src/agentProvider/process/agentProcessShim.ts index 3645a18c31..6185af5946 100644 --- a/ts/packages/dispatcher/nodeProviders/src/agentProvider/process/agentProcessShim.ts +++ b/ts/packages/dispatcher/nodeProviders/src/agentProvider/process/agentProcessShim.ts @@ -153,7 +153,20 @@ export async function createAgentProcess( >("control"); const agentInterface = await new Promise( (resolve, reject) => { + const onExit = ( + exitCode: number | null, + signal: NodeJS.Signals | null, + ) => { + reject( + new Error( + `Agent process '${agentName}' exited before startup completed ` + + `(code=${exitCode ?? "null"}, signal=${signal ?? "null"})`, + ), + ); + }; + agentProcess.once("exit", onExit); channel.once("message", (message: any) => { + agentProcess.off("exit", onExit); if (Array.isArray(message)) { resolve(message); } else { @@ -176,6 +189,7 @@ export async function createAgentProcess( agentName, channelProvider, agentInterface, + { trustedContextPropagation: true }, ), // `count` is a HOLDER refcount owned by the caller (the npm provider's // load/unload). A freshly created process has no holders yet; the diff --git a/ts/packages/dispatcher/nodeProviders/test/agentProcessTelemetry.spec.ts b/ts/packages/dispatcher/nodeProviders/test/agentProcessTelemetry.spec.ts new file mode 100644 index 0000000000..06c1e08f32 --- /dev/null +++ b/ts/packages/dispatcher/nodeProviders/test/agentProcessTelemetry.spec.ts @@ -0,0 +1,885 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +import { + context, + propagation, + SpanKind, + SpanStatusCode, + trace, + type Context, + type Span, +} from "@opentelemetry/api"; +import type { + ActionContext, + SessionContext, + TypeAgentAction, +} from "@typeagent/agent-sdk"; +import { otel } from "@typeagent/telemetry"; +import { createChannelProvider } from "@typeagent/agent-rpc/channel"; +import { createDispatcherRpcServer } from "@typeagent/dispatcher-rpc/dispatcher/server"; +import { + createDispatcherFromContext, + type CommandHandlerContext, +} from "agent-dispatcher/internal"; +import { fork, type ChildProcess } from "node:child_process"; +import { once } from "node:events"; +import { mkdtempSync, rmSync } from "node:fs"; +import { createServer, type IncomingMessage } from "node:http"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; + +import { createAgentProcess } from "../src/agentProvider/process/agentProcessShim.js"; + +interface ProtobufField { + readonly number: number; + readonly wireType: number; + readonly value: bigint | Buffer; +} + +interface ExportedSpan { + readonly name: string; + readonly traceId: string; + readonly spanId: string; + readonly parentSpanId: string; + readonly kind: number; + readonly statusCode: number; + readonly attributes: ReadonlyMap; + readonly processName: string; +} + +const ACTION_NAMES = ["succeed", "fail", "cancel"] as const; +const AGENT_NAME = "telemetry-fixture"; +const SESSION_ID = "session-rpc-test"; +const ACTIVATION_ID = "activation-rpc-test"; +const LEGACY_TRACE_ID = "legacy-rpc-test"; + +describe("agent subprocess OpenTelemetry propagation", () => { + afterEach(() => { + trace.disable(); + context.disable(); + propagation.disable(); + }); + + it("preserves trace and parent continuity for success, failure, and cancellation", async () => { + const payloads: Buffer[] = []; + const receiver = createServer(async (request, response) => { + expect(request.url).toBe("/v1/traces"); + payloads.push(await readRequestBody(request)); + response.writeHead(200, { + "content-type": "application/x-protobuf", + }); + response.end(); + }); + receiver.listen(0, "127.0.0.1"); + await once(receiver, "listening"); + + const address = receiver.address(); + if (address === null || typeof address === "string") { + throw new Error("Expected a TCP receiver address"); + } + const endpoint = `http://127.0.0.1:${address.port}/v1/traces`; + const telemetryEnvironment = configureTelemetryEnvironment(endpoint); + + const coordinator = otel.createTelemetryCoordinator(); + let agentProcess: + | Awaited> + | undefined; + try { + await coordinator.init({ + config: { + traces: { + otlp: { endpoint }, + }, + }, + serviceName: "typeagent-agent-server-test", + processName: "agent-server-test", + }); + agentProcess = await createAgentProcess( + AGENT_NAME, + new URL("./fixtures/telemetryAgent.js", import.meta.url).href, + ); + + await runOptionsCallback(agentProcess.appAgent); + await runAction(agentProcess.appAgent, "succeed"); + await expect( + runAction(agentProcess.appAgent, "fail"), + ).rejects.toThrow("fixture failure"); + await expect( + runAction(agentProcess.appAgent, "cancel"), + ).rejects.toMatchObject({ name: "AbortError" }); + await new Promise((resolve) => setTimeout(resolve, 50)); + + await agentProcess.close?.(); + agentProcess = undefined; + await coordinator.shutdown(); + } finally { + await agentProcess?.close?.(); + await coordinator.shutdown(); + restoreTelemetryEnvironment(telemetryEnvironment); + receiver.close(); + await once(receiver, "close"); + } + + const spans = payloads.flatMap(decodeSpans); + assertOptionsCallbackChain(spans); + for (const actionName of ACTION_NAMES) { + assertActionRpcChain(spans, actionName); + } + }); + + it("preserves the full client, dispatcher, request, action, and subprocess chain", async () => { + const payloads: Buffer[] = []; + const receiver = createServer(async (request, response) => { + expect(request.url).toBe("/v1/traces"); + payloads.push(await readRequestBody(request)); + response.writeHead(200, { + "content-type": "application/x-protobuf", + }); + response.end(); + }); + receiver.listen(0, "127.0.0.1"); + await once(receiver, "listening"); + + const address = receiver.address(); + if (address === null || typeof address === "string") { + throw new Error("Expected a TCP receiver address"); + } + const endpoint = `http://127.0.0.1:${address.port}/v1/traces`; + const telemetryEnvironment = configureTelemetryEnvironment(endpoint); + + const coordinator = otel.createTelemetryCoordinator(); + let agentProcess: + | Awaited> + | undefined; + let clientProcess: ChildProcess | undefined; + let clientExit: Promise | undefined; + try { + await coordinator.init({ + config: { + traces: { + otlp: { endpoint }, + }, + }, + serviceName: "typeagent-agent-server-test", + processName: "agent-server-test", + }); + agentProcess = await createAgentProcess( + AGENT_NAME, + new URL("./fixtures/telemetryAgent.js", import.meta.url).href, + ); + + let resolveExecution!: () => void; + let rejectExecution!: (error: unknown) => void; + const execution = new Promise((resolve, reject) => { + resolveExecution = resolve; + rejectExecution = reject; + }); + const commandContext = { + telemetryOptions: { joinActiveTrace: true }, + requestQueue: { + submit(input: { + readonly text: string; + readonly traceContext?: Context; + }) { + const traceContext = input.traceContext; + if (traceContext === undefined) { + throw new Error( + "Expected dispatcher to capture the RPC context", + ); + } + setImmediate(() => { + void context + .with(traceContext, () => + runQueuedRequest(agentProcess!.appAgent), + ) + .then(resolveExecution, rejectExecution); + }); + return { + requestId: "full-trace-request", + originatorConnectionId: "client-test", + text: input.text, + submittedAt: Date.now(), + state: "queued", + completion: execution.then(() => undefined), + }; + }, + }, + } as unknown as CommandHandlerContext; + const dispatcher = createDispatcherFromContext( + commandContext, + "client-test", + ); + + clientProcess = fork( + fileURLToPath( + new URL( + "./fixtures/dispatcherRpcClient.js", + import.meta.url, + ), + ), + [], + { + env: { ...process.env }, + stdio: ["ignore", "inherit", "inherit", "ipc"], + windowsHide: true, + } as Parameters[2] & { windowsHide: boolean }, + ); + clientExit = observeChildExit(clientProcess); + const channelProvider = createChannelProvider( + "dispatcher-telemetry-server", + clientProcess, + ); + createDispatcherRpcServer( + dispatcher, + channelProvider.createChannel("dispatcher"), + { + trustedContextPropagation: true, + }, + ); + const controlChannel = + channelProvider.createChannel("control"); + const submitted = waitForChannelMessage( + controlChannel, + "submitted", + ); + controlChannel.send("run"); + + await Promise.race([ + submitted, + clientExit.then(({ code, signal }) => { + throw new Error( + `Dispatcher RPC client exited before submission ` + + `(code=${code ?? "null"}, signal=${signal ?? "null"})`, + ); + }), + ]); + await execution; + await agentProcess.close?.(); + agentProcess = undefined; + controlChannel.send("shutdown"); + await clientExit; + clientProcess = undefined; + clientExit = undefined; + await coordinator.shutdown(); + } finally { + await agentProcess?.close?.(); + if (clientProcess !== undefined) { + if ( + clientProcess.exitCode === null && + clientProcess.signalCode === null + ) { + clientProcess.kill(); + } + await clientExit?.catch(() => {}); + } + await coordinator.shutdown(); + restoreTelemetryEnvironment(telemetryEnvironment); + receiver.close(); + await once(receiver, "close"); + } + + assertFullProcessChain(payloads.flatMap(decodeSpans)); + }); +}); + +async function runQueuedRequest( + appAgent: Awaited>["appAgent"], +): Promise { + const executeAction = appAgent.executeAction; + if (executeAction === undefined) { + throw new Error("Fixture agent must implement executeAction"); + } + const attributes = { + agentName: AGENT_NAME, + actionName: "succeed", + sessionId: SESSION_ID, + activationId: ACTIVATION_ID, + traceId: LEGACY_TRACE_ID, + }; + const tracer = trace.getTracer( + otel.INSTRUMENTATION_SCOPE_NAME, + otel.INSTRUMENTATION_SCOPE_VERSION, + ); + await tracer.startActiveSpan( + otel.TYPEAGENT_SPAN_NAMES.REQUEST, + async (requestSpan) => { + otel.setTypeAgentSpanAttributes(requestSpan, attributes); + try { + await tracer.startActiveSpan( + otel.TYPEAGENT_SPAN_NAMES.ACTION, + async (actionSpan) => { + otel.setTypeAgentSpanAttributes(actionSpan, attributes); + try { + await context.with( + otel.setActiveTypeAgentSpanAttributes( + context.active(), + attributes, + ), + () => + executeAction( + { + schemaName: AGENT_NAME, + actionName: "succeed", + parameters: {}, + }, + createActionContext( + new AbortController().signal, + ), + ), + ); + } finally { + actionSpan.end(); + } + }, + ); + } finally { + requestSpan.end(); + } + }, + ); +} + +function waitForChannelMessage( + channel: { + on(event: "message", callback: (message: string) => void): void; + off(event: "message", callback: (message: string) => void): void; + }, + expected: string, +): Promise { + return new Promise((resolve, reject) => { + const listener = (message: string) => { + if (message === expected) { + channel.off("message", listener); + resolve(); + } else if (message.startsWith("error:")) { + channel.off("message", listener); + reject(new Error(message.slice("error:".length))); + } + }; + channel.on("message", listener); + }); +} + +interface ChildExit { + readonly code: number | null; + readonly signal: NodeJS.Signals | null; +} + +function observeChildExit(child: ChildProcess): Promise { + return new Promise((resolve, reject) => { + const onError = (error: Error) => { + child.off("exit", onExit); + reject(error); + }; + const onExit = (code: number | null, signal: NodeJS.Signals | null) => { + child.off("error", onError); + resolve({ code, signal }); + }; + child.once("error", onError); + child.once("exit", onExit); + }); +} + +function assertFullProcessChain(spans: readonly ExportedSpan[]): void { + const clientRpc = spans.find( + (span) => + span.name === "typeagent.rpc.invoke" && + span.kind === SpanKind.CLIENT && + span.processName === "cli-test" && + span.attributes.get("rpc.method") === "submitCommand", + ); + const dispatcherRpc = spans.find( + (span) => + span.name === "typeagent.rpc.invoke" && + span.kind === SpanKind.SERVER && + span.processName === "agent-server-test" && + span.attributes.get("rpc.method") === "submitCommand", + ); + const request = spans.find( + (span) => + span.name === otel.TYPEAGENT_SPAN_NAMES.REQUEST && + span.processName === "agent-server-test", + ); + const action = spans.find( + (span) => + span.name === otel.TYPEAGENT_SPAN_NAMES.ACTION && + span.processName === "agent-server-test" && + span.attributes.get("typeagent.action.name") === "succeed", + ); + const agentClientRpc = spans.find( + (span) => + span.name === "typeagent.rpc.invoke" && + span.kind === SpanKind.CLIENT && + span.processName === "agent-server-test" && + span.attributes.get("rpc.method") === "executeAction", + ); + const agentServerRpc = spans.find( + (span) => + span.name === "typeagent.rpc.invoke" && + span.kind === SpanKind.SERVER && + span.processName === `agent-${AGENT_NAME}` && + span.attributes.get("rpc.method") === "executeAction", + ); + const chain = [ + clientRpc, + dispatcherRpc, + request, + action, + agentClientRpc, + agentServerRpc, + ]; + for (const span of chain) { + expect(span).toBeDefined(); + } + for (const span of chain.slice(1)) { + expect(span!.traceId).toBe(clientRpc!.traceId); + } + expect(dispatcherRpc!.parentSpanId).toBe(clientRpc!.spanId); + expect(request!.parentSpanId).toBe(dispatcherRpc!.spanId); + expect(action!.parentSpanId).toBe(request!.spanId); + expect(agentClientRpc!.parentSpanId).toBe(action!.spanId); + expect(agentServerRpc!.parentSpanId).toBe(agentClientRpc!.spanId); + expect(new Set(chain.map((span) => span!.processName))).toEqual( + new Set(["cli-test", "agent-server-test", `agent-${AGENT_NAME}`]), + ); +} + +async function runAction( + appAgent: Awaited>["appAgent"], + actionName: (typeof ACTION_NAMES)[number], +): Promise { + const executeAction = appAgent.executeAction; + if (executeAction === undefined) { + throw new Error("Fixture agent must implement executeAction"); + } + const abortController = new AbortController(); + const action = { + schemaName: AGENT_NAME, + actionName, + parameters: { privateValue: "must-not-be-telemetry" }, + } as TypeAgentAction; + const actionContext = createActionContext(abortController.signal); + const tracer = trace.getTracer( + otel.INSTRUMENTATION_SCOPE_NAME, + otel.INSTRUMENTATION_SCOPE_VERSION, + ); + + await tracer.startActiveSpan( + otel.TYPEAGENT_SPAN_NAMES.ACTION, + { + kind: SpanKind.INTERNAL, + }, + async (span: Span) => { + const attributes = { + agentName: AGENT_NAME, + actionName, + sessionId: SESSION_ID, + activationId: ACTIVATION_ID, + traceId: LEGACY_TRACE_ID, + }; + otel.setTypeAgentSpanAttributes(span, attributes); + try { + await context.with( + otel.setActiveTypeAgentSpanAttributes( + context.active(), + attributes, + ), + async () => { + const result = executeAction(action, actionContext); + if (actionName === "cancel") { + setTimeout(() => abortController.abort(), 25); + } + await result; + }, + ); + } catch (error) { + const cancelled = + error !== null && + typeof error === "object" && + (error as { name?: unknown }).name === "AbortError"; + span.setStatus({ + code: SpanStatusCode.ERROR, + message: cancelled ? "cancelled" : "action failed", + }); + throw error; + } finally { + span.end(); + } + }, + ); +} + +async function runOptionsCallback( + appAgent: Awaited>["appAgent"], +): Promise { + const initializeAgentContext = appAgent.initializeAgentContext; + if (initializeAgentContext === undefined) { + throw new Error("Fixture agent must implement initializeAgentContext"); + } + const attributes = { + agentName: AGENT_NAME, + actionName: "initialize", + sessionId: SESSION_ID, + activationId: ACTIVATION_ID, + traceId: LEGACY_TRACE_ID, + }; + const tracer = trace.getTracer( + otel.INSTRUMENTATION_SCOPE_NAME, + otel.INSTRUMENTATION_SCOPE_VERSION, + ); + await tracer.startActiveSpan( + otel.TYPEAGENT_SPAN_NAMES.ACTION, + async (span) => { + otel.setTypeAgentSpanAttributes(span, attributes); + try { + await context.with( + otel.setActiveTypeAgentSpanAttributes( + context.active(), + attributes, + ), + () => + initializeAgentContext({ + options: { + callback: async () => {}, + }, + }), + ); + } finally { + span.end(); + } + }, + ); +} + +function createActionContext(signal: AbortSignal): ActionContext { + const sessionContext = { + agentContext: undefined, + sessionContextId: "rpc-internal-context-must-not-be-telemetry", + } as unknown as SessionContext; + return { + streamingContext: undefined, + activityContext: undefined, + actionIO: { + setDisplay: () => {}, + appendDisplay: () => {}, + takeAction: () => {}, + appendDiagnosticData: () => {}, + }, + sessionContext, + abortSignal: signal, + isFromReasoningLoop: false, + queueToggleTransientAgent: async () => {}, + }; +} + +function assertActionRpcChain( + spans: readonly ExportedSpan[], + actionName: string, +): void { + const matching = spans.filter( + (span) => span.attributes.get("typeagent.action.name") === actionName, + ); + const action = matching.find( + (span) => + span.name === otel.TYPEAGENT_SPAN_NAMES.ACTION && + span.processName === "agent-server-test", + ); + const client = matching.find( + (span) => + span.name === "typeagent.rpc.invoke" && + span.kind === SpanKind.CLIENT && + span.processName === "agent-server-test", + ); + const server = matching.find( + (span) => + span.name === "typeagent.rpc.invoke" && + span.kind === SpanKind.SERVER && + span.processName === `agent-${AGENT_NAME}`, + ); + expect(action).toBeDefined(); + expect(client).toBeDefined(); + expect(server).toBeDefined(); + expect(client!.traceId).toBe(action!.traceId); + expect(client!.parentSpanId).toBe(action!.spanId); + expect(server!.traceId).toBe(action!.traceId); + expect(server!.parentSpanId).toBe(client!.spanId); + + for (const span of [client!, server!]) { + expect(span.attributes.get("typeagent.agent.name")).toBe(AGENT_NAME); + expect(span.attributes.get("typeagent.session.id")).toBe(SESSION_ID); + expect(span.attributes.get("typeagent.activation.id")).toBe( + ACTIVATION_ID, + ); + expect(span.attributes.get("typeagent.trace.id")).toBe(LEGACY_TRACE_ID); + expect( + [...span.attributes.keys()].some( + (key) => + key.includes("parameter") || + key.includes("context.id") || + key.includes("private"), + ), + ).toBe(false); + expect([...span.attributes.values()]).not.toContain( + "must-not-be-telemetry", + ); + expect([...span.attributes.values()]).not.toContain( + "rpc-internal-context-must-not-be-telemetry", + ); + } + if (actionName !== "succeed") { + expect(client!.statusCode).toBe(SpanStatusCode.ERROR); + expect(server!.statusCode).toBe(SpanStatusCode.ERROR); + } +} + +function assertOptionsCallbackChain(spans: readonly ExportedSpan[]): void { + const matching = spans.filter( + (span) => span.attributes.get("typeagent.action.name") === "initialize", + ); + const action = matching.find( + (span) => + span.name === otel.TYPEAGENT_SPAN_NAMES.ACTION && + span.processName === "agent-server-test", + ); + const mainClient = matching.find( + (span) => + span.attributes.get("rpc.method") === "initializeAgentContext" && + span.kind === SpanKind.CLIENT, + ); + const mainServer = matching.find( + (span) => + span.attributes.get("rpc.method") === "initializeAgentContext" && + span.kind === SpanKind.SERVER, + ); + const callbackClient = matching.find( + (span) => + span.attributes.get("rpc.method") === "callback" && + span.kind === SpanKind.CLIENT, + ); + const callbackServer = matching.find( + (span) => + span.attributes.get("rpc.method") === "callback" && + span.kind === SpanKind.SERVER, + ); + for (const span of [ + action, + mainClient, + mainServer, + callbackClient, + callbackServer, + ]) { + expect(span).toBeDefined(); + expect(span!.traceId).toBe(action!.traceId); + } + expect(mainClient!.parentSpanId).toBe(action!.spanId); + expect(mainServer!.parentSpanId).toBe(mainClient!.spanId); + expect(callbackClient!.parentSpanId).toBe(mainServer!.spanId); + expect(callbackServer!.parentSpanId).toBe(callbackClient!.spanId); +} + +function readVarint( + buffer: Buffer, + start: number, +): { value: bigint; next: number } { + let value = 0n; + let shift = 0n; + let offset = start; + while (offset < buffer.length) { + const byte = buffer[offset++]!; + value |= BigInt(byte & 0x7f) << shift; + if ((byte & 0x80) === 0) { + return { value, next: offset }; + } + shift += 7n; + } + throw new Error("Truncated protobuf varint"); +} + +function parseMessage(buffer: Buffer): ProtobufField[] { + const fields: ProtobufField[] = []; + let offset = 0; + while (offset < buffer.length) { + const tag = readVarint(buffer, offset); + offset = tag.next; + const number = Number(tag.value >> 3n); + const wireType = Number(tag.value & 7n); + if (wireType === 0) { + const item = readVarint(buffer, offset); + fields.push({ number, wireType, value: item.value }); + offset = item.next; + } else if (wireType === 1) { + fields.push({ + number, + wireType, + value: buffer.subarray(offset, offset + 8), + }); + offset += 8; + } else if (wireType === 2) { + const length = readVarint(buffer, offset); + offset = length.next; + const end = offset + Number(length.value); + fields.push({ + number, + wireType, + value: buffer.subarray(offset, end), + }); + offset = end; + } else if (wireType === 5) { + fields.push({ + number, + wireType, + value: buffer.subarray(offset, offset + 4), + }); + offset += 4; + } else { + throw new Error(`Unsupported protobuf wire type ${wireType}`); + } + } + return fields; +} + +function getBuffers(fields: ProtobufField[], number: number): Buffer[] { + return fields + .filter( + (field) => field.number === number && Buffer.isBuffer(field.value), + ) + .map((field) => field.value as Buffer); +} + +function getVarint(fields: ProtobufField[], number: number): number { + const field = fields.find( + (candidate) => + candidate.number === number && typeof candidate.value === "bigint", + ); + return field === undefined ? 0 : Number(field.value); +} + +function getString(fields: ProtobufField[], number: number): string { + return getBuffers(fields, number)[0]?.toString("utf8") ?? ""; +} + +function decodeSpans(request: Buffer): ExportedSpan[] { + const spans: ExportedSpan[] = []; + for (const resourceSpans of getBuffers(parseMessage(request), 1)) { + const resourceFields = parseMessage(resourceSpans); + const resource = getBuffers(resourceFields, 1)[0]; + const resourceAttributes = + resource === undefined + ? new Map() + : decodeAttributes(parseMessage(resource), 1); + const processName = + resourceAttributes.get("typeagent.process.name") ?? ""; + for (const scopeSpans of getBuffers(resourceFields, 2)) { + for (const encodedSpan of getBuffers(parseMessage(scopeSpans), 2)) { + const fields = parseMessage(encodedSpan); + const status = getBuffers(fields, 15)[0]; + spans.push({ + name: getString(fields, 5), + traceId: getBuffers(fields, 1)[0]?.toString("hex") ?? "", + spanId: getBuffers(fields, 2)[0]?.toString("hex") ?? "", + parentSpanId: + getBuffers(fields, 4)[0]?.toString("hex") ?? "", + kind: getVarint(fields, 6) - 1, + statusCode: + status === undefined + ? 0 + : getVarint(parseMessage(status), 3), + attributes: decodeAttributes(fields, 9), + processName, + }); + } + } + } + return spans; +} + +function decodeAttributes( + fields: ProtobufField[], + fieldNumber: number, +): Map { + const attributes = new Map(); + for (const keyValue of getBuffers(fields, fieldNumber)) { + const keyValueFields = parseMessage(keyValue); + const key = getString(keyValueFields, 1); + const anyValue = getBuffers(keyValueFields, 2)[0]; + if (key !== "" && anyValue !== undefined) { + attributes.set(key, getString(parseMessage(anyValue), 1)); + } + } + return attributes; +} + +async function readRequestBody(request: IncomingMessage): Promise { + const chunks: Buffer[] = []; + for await (const chunk of request) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + } + return Buffer.concat(chunks); +} + +function captureEnv( + names: readonly string[], +): ReadonlyMap { + return new Map(names.map((name) => [name, process.env[name]])); +} + +function restoreEnv(values: ReadonlyMap): void { + for (const [name, value] of values) { + if (value === undefined) { + delete process.env[name]; + } else { + process.env[name] = value; + } + } +} + +const TELEMETRY_ENVIRONMENT_NAMES = [ + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", + "OTEL_TRACES_EXPORTER", + "OTEL_METRICS_EXPORTER", + "OTEL_LOGS_EXPORTER", + "OTEL_TRACES_SAMPLER", + "OTEL_TRACES_SAMPLER_ARG", + "TYPEAGENT_CONFIG_DIR", + "TYPEAGENT_CONFIG_DEFAULTS", + "TYPEAGENT_CONFIG_LOCAL", + "TYPEAGENT_DOTENV", + "TYPEAGENT_OTEL_LOG_FILE", + "TYPEAGENT_OTEL_DEBUG_BRIDGE", + "DEBUG", +] as const; + +interface TelemetryTestEnvironment { + readonly configDir: string; + readonly previousEnv: ReadonlyMap; +} + +function configureTelemetryEnvironment( + endpoint: string, +): TelemetryTestEnvironment { + const configDir = mkdtempSync(join(tmpdir(), "typeagent-otel-test-")); + const previousEnv = captureEnv(TELEMETRY_ENVIRONMENT_NAMES); + process.env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT = endpoint; + process.env.OTEL_TRACES_EXPORTER = "otlp"; + process.env.OTEL_METRICS_EXPORTER = "none"; + process.env.OTEL_LOGS_EXPORTER = "none"; + process.env.OTEL_TRACES_SAMPLER = "always_on"; + process.env.TYPEAGENT_CONFIG_DIR = configDir; + process.env.TYPEAGENT_OTEL_DEBUG_BRIDGE = "false"; + delete process.env.OTEL_TRACES_SAMPLER_ARG; + delete process.env.TYPEAGENT_CONFIG_DEFAULTS; + delete process.env.TYPEAGENT_CONFIG_LOCAL; + delete process.env.TYPEAGENT_DOTENV; + delete process.env.TYPEAGENT_OTEL_LOG_FILE; + delete process.env.DEBUG; + return { configDir, previousEnv }; +} + +function restoreTelemetryEnvironment( + environment: TelemetryTestEnvironment, +): void { + restoreEnv(environment.previousEnv); + rmSync(environment.configDir, { recursive: true, force: true }); +} diff --git a/ts/packages/dispatcher/nodeProviders/test/fixtures/dispatcherRpcClient.ts b/ts/packages/dispatcher/nodeProviders/test/fixtures/dispatcherRpcClient.ts new file mode 100644 index 0000000000..bde764e780 --- /dev/null +++ b/ts/packages/dispatcher/nodeProviders/test/fixtures/dispatcherRpcClient.ts @@ -0,0 +1,59 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +import { createChannelProvider } from "@typeagent/agent-rpc/channel"; +import { createDispatcherRpcClient } from "@typeagent/dispatcher-rpc/dispatcher/client"; +import { otel } from "@typeagent/telemetry"; + +function isIpcProcess( + value: NodeJS.Process, +): value is NodeJS.Process & { send: (message: unknown) => boolean } { + return typeof value.send === "function"; +} + +if (!isIpcProcess(process)) { + throw new Error("Dispatcher RPC client fixture requires an IPC channel"); +} + +const channelProvider = createChannelProvider( + "dispatcher-telemetry-client", + process, +); +const controlChannel = channelProvider.createChannel("control"); +const dispatcherChannel = channelProvider.createChannel("dispatcher"); + +controlChannel.on("message", (message) => { + if (message === "run") { + void run().catch((error) => { + const detail = + error instanceof Error ? error.message : String(error); + controlChannel.send(`error:${detail}`); + }); + } else if (message === "shutdown") { + void shutdown(); + } +}); + +async function run(): Promise { + await otel.initTelemetry({ + serviceName: "typeagent-client-test", + processName: "cli-test", + }); + const { dispatcher } = createDispatcherRpcClient( + dispatcherChannel, + undefined, + { + trustedContextPropagation: true, + }, + ); + const result = await dispatcher.submitCommand("run telemetry fixture"); + if (!result.ok) { + throw new Error(`Dispatcher submission failed: ${result.error}`); + } + controlChannel.send("submitted"); +} + +async function shutdown(): Promise { + await otel.shutdownTelemetry(); + process.disconnect(); +} diff --git a/ts/packages/dispatcher/nodeProviders/test/fixtures/telemetryAgent.ts b/ts/packages/dispatcher/nodeProviders/test/fixtures/telemetryAgent.ts new file mode 100644 index 0000000000..62880864a7 --- /dev/null +++ b/ts/packages/dispatcher/nodeProviders/test/fixtures/telemetryAgent.ts @@ -0,0 +1,61 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +import type { + ActionContext, + AppAgent, + AppAgentInitSettings, + TypeAgentAction, +} from "@typeagent/agent-sdk"; +import { createActionResult } from "@typeagent/agent-sdk/helpers/action"; + +export function instantiate(): AppAgent { + return { + async initializeAgentContext(settings?: AppAgentInitSettings) { + const callback = (settings?.options as { callback?: unknown }) + ?.callback; + if (typeof callback !== "function") { + throw new Error("Expected an initialization callback"); + } + await callback(); + }, + async executeAction(action: TypeAgentAction, context: ActionContext) { + switch (action.actionName) { + case "succeed": + return createActionResult("success"); + case "fail": + throw new Error("fixture failure"); + case "cancel": + await waitForCancellation(context.abortSignal); + return createActionResult("unexpected completion"); + default: + throw new Error( + `Unknown fixture action: ${action.actionName}`, + ); + } + }, + }; +} + +function waitForCancellation(signal: AbortSignal | undefined): Promise { + if (signal?.aborted === true) { + return Promise.reject( + signal.reason ?? + new DOMException("The operation was aborted.", "AbortError"), + ); + } + return new Promise((_resolve, reject) => { + signal?.addEventListener( + "abort", + () => + reject( + signal.reason ?? + new DOMException( + "The operation was aborted.", + "AbortError", + ), + ), + { once: true }, + ); + }); +} diff --git a/ts/packages/dispatcher/rpc/src/dispatcherClient.ts b/ts/packages/dispatcher/rpc/src/dispatcherClient.ts index a3af9c1716..25029a8eff 100644 --- a/ts/packages/dispatcher/rpc/src/dispatcherClient.ts +++ b/ts/packages/dispatcher/rpc/src/dispatcherClient.ts @@ -43,6 +43,14 @@ export interface DispatcherRpcClient { notifyRequestCancelled(requestId: string, reason: QueueCancelReason): void; } +export interface DispatcherRpcOptions { + /** + * Propagate the active OTel context across this TypeAgent-owned channel. + * Enable only when the destination is the trusted agent-server endpoint. + */ + trustedContextPropagation?: boolean; +} + type PendingEntry = { resolve: (value: CommandResult | undefined) => void; reject: (err: unknown) => void; @@ -55,10 +63,20 @@ type SettledEntry = export function createDispatcherRpcClient( channel: RpcChannel, connectionId?: ConnectionId, + options?: DispatcherRpcOptions, ): DispatcherRpcClient { const rpc = createRpc( "dispatcher", channel, + undefined, + undefined, + options?.trustedContextPropagation === true + ? { + tracing: { + propagateContext: true, + }, + } + : undefined, ); // requestId → pending submitCommand awaiter (set on submit, resolved by diff --git a/ts/packages/dispatcher/rpc/src/dispatcherServer.ts b/ts/packages/dispatcher/rpc/src/dispatcherServer.ts index ca99211961..75f43f9f74 100644 --- a/ts/packages/dispatcher/rpc/src/dispatcherServer.ts +++ b/ts/packages/dispatcher/rpc/src/dispatcherServer.ts @@ -9,6 +9,7 @@ import type { DispatcherInvokeFunctions, WireSubmitResult, } from "./dispatcherTypes.js"; +import type { DispatcherRpcOptions } from "./dispatcherClient.js"; /** * Drop the in-process-only `completion` promise from a `SubmitResult` so it @@ -26,6 +27,7 @@ function toWire(result: SubmitResult): WireSubmitResult { export function createDispatcherRpcServer( dispatcher: Dispatcher, channel: RpcChannel, + options?: DispatcherRpcOptions, ) { const dispatcherCallHandler: DispatcherCallFunctions = { cancelCommandByClientId(...args) { @@ -107,5 +109,12 @@ export function createDispatcherRpcServer( channel, dispatcherInvokeHandler, dispatcherCallHandler, + options?.trustedContextPropagation === true + ? { + tracing: { + trustRemoteContext: true, + }, + } + : undefined, ); } diff --git a/ts/packages/telemetry/package.json b/ts/packages/telemetry/package.json index e1a927911f..0b851be1e9 100644 --- a/ts/packages/telemetry/package.json +++ b/ts/packages/telemetry/package.json @@ -18,6 +18,10 @@ "./testing/inMemorySpanManager": { "types": "./dist/otel/testing/inMemorySpanManager.d.ts", "node": "./dist/otel/testing/inMemorySpanManager.js" + }, + "./traceContext": { + "types": "./dist/otel/traceContract.d.ts", + "default": "./dist/otel/traceContract.js" } }, "files": [ diff --git a/ts/packages/telemetry/src/otel/index.ts b/ts/packages/telemetry/src/otel/index.ts index 7b83fa34a2..86967e4d37 100644 --- a/ts/packages/telemetry/src/otel/index.ts +++ b/ts/packages/telemetry/src/otel/index.ts @@ -77,6 +77,8 @@ export { export { TYPEAGENT_SPAN_NAMES, TYPEAGENT_SPAN_ATTRIBUTES, + getActiveTypeAgentSpanAttributes, + setActiveTypeAgentSpanAttributes, setTypeAgentSpanAttributes, type TypeAgentSpanName, type TypeAgentSpanAttributeKey, diff --git a/ts/packages/telemetry/src/otel/traceContract.ts b/ts/packages/telemetry/src/otel/traceContract.ts index 8f0406319a..5f320f0e30 100644 --- a/ts/packages/telemetry/src/otel/traceContract.ts +++ b/ts/packages/telemetry/src/otel/traceContract.ts @@ -1,7 +1,12 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. -import type { Span } from "@opentelemetry/api"; +import { + context, + createContextKey, + type Context, + type Span, +} from "@opentelemetry/api"; import { redactText, type RedactionOptions } from "./redaction.js"; /** @@ -94,6 +99,25 @@ export interface TypeAgentSpanAttributes { readonly traceId?: string; } +const ACTIVE_TYPEAGENT_ATTRIBUTES = createContextKey( + "typeagent.active-span-attributes", +); + +export function getActiveTypeAgentSpanAttributes(): + | TypeAgentSpanAttributes + | undefined { + return context.active().getValue(ACTIVE_TYPEAGENT_ATTRIBUTES) as + | TypeAgentSpanAttributes + | undefined; +} + +export function setActiveTypeAgentSpanAttributes( + activeContext: Context, + attributes: TypeAgentSpanAttributes, +): Context { + return activeContext.setValue(ACTIVE_TYPEAGENT_ATTRIBUTES, attributes); +} + const ATTRIBUTE_KEY_FOR_FIELD: { readonly [K in keyof TypeAgentSpanAttributes]-?: TypeAgentSpanAttributeKey; } = { diff --git a/ts/pnpm-lock.yaml b/ts/pnpm-lock.yaml index 9b6741018c..f925bca834 100644 --- a/ts/pnpm-lock.yaml +++ b/ts/pnpm-lock.yaml @@ -1467,6 +1467,9 @@ importers: '@typeagent/common-utils': specifier: workspace:* version: link:../utils/commonUtils + '@typeagent/telemetry': + specifier: workspace:* + version: link:../telemetry debug: specifier: ^4.4.0 version: 4.4.1(supports-color@8.1.1) @@ -5067,6 +5070,12 @@ importers: specifier: ^4.4.0 version: 4.4.3(supports-color@8.1.1) devDependencies: + '@opentelemetry/api': + specifier: 1.9.0 + version: 1.9.0 + '@typeagent/dispatcher-rpc': + specifier: workspace:* + version: link:../rpc '@typeagent/dispatcher-types': specifier: workspace:* version: link:../types