diff --git a/packages/agent/docs/telemetry-schema.md b/packages/agent/docs/telemetry-schema.md index 4454cbb4d..de1984972 100644 --- a/packages/agent/docs/telemetry-schema.md +++ b/packages/agent/docs/telemetry-schema.md @@ -18,6 +18,9 @@ One logical request to an AI provider | Name | Type | Required | Values | Notes | Description | |---|---|---:|---|---|---| +| `diffex.invocation.id` | `string` | no | | high cardinality | Owning in-process agent invocation id | +| `diffex.turn.id` | `string` | no | | high cardinality | Owning in-process turn id | +| `diffex.request.id` | `string` | no | | high cardinality | Logical provider request id | | `diffex.ai.operation` | `string` | yes | stream, fetch_deferred, cancel_deferred, generate_images | | Logical provider operation | | `diffex.ai.provider` | `string` | yes | | | Selected provider id | | `diffex.ai.model` | `string` | yes | | | Requested model id | @@ -50,6 +53,97 @@ All end attributes are optional completion enrichment. No declared span events. +## Active agent runtime schema + +Schema version: 1 + +### `diffex.agent.run` + +One in-process agent prompt or continuation + +- Parents: root or caller-owned external span +- Default status: `ok` +- Error when: The invocation fails or is aborted + +#### Start attributes + +| Name | Type | Required | Values | Notes | Description | +|---|---|---:|---|---|---| +| `diffex.invocation.id` | `string` | yes | | high cardinality | Invocation-local run id | +| `diffex.session.id` | `string` | no | | high cardinality | Caller-provided session id | +| `diffex.agent.run_kind` | `string` | yes | prompt, continue | | Agent admission path | + +#### End attributes + +All end attributes are optional completion enrichment. + +| Name | Type | Values | Notes | Description | +|---|---|---|---|---| +| `diffex.agent.outcome` | `string` | completed, failed, aborted | | Terminal invocation outcome | +| `diffex.agent.turn_count` | `number` | | | Number of admitted turns | + +#### Events + +No declared span events. + +### `diffex.agent.turn` + +One assistant response and its tool batch in the active agent runtime + +- Parents: `diffex.agent.run` +- Default status: `ok` +- Error when: The turn fails or is aborted + +#### Start attributes + +| Name | Type | Required | Values | Notes | Description | +|---|---|---:|---|---|---| +| `diffex.invocation.id` | `string` | yes | | high cardinality | Owning agent invocation id | +| `diffex.turn.id` | `string` | yes | | high cardinality | Invocation-local turn id | +| `diffex.turn.index` | `number` | yes | | | Zero-based turn index | + +#### End attributes + +All end attributes are optional completion enrichment. + +| Name | Type | Values | Notes | Description | +|---|---|---|---|---| +| `diffex.agent.outcome` | `string` | completed, failed, aborted | | Terminal turn outcome | + +#### Events + +No declared span events. + +### `diffex.agent.tool` + +One tool call admitted by the active agent runtime + +- Parents: `diffex.agent.turn` +- Default status: `ok` +- Error when: The finalized tool result is an error + +#### Start attributes + +| Name | Type | Required | Values | Notes | Description | +|---|---|---:|---|---|---| +| `diffex.invocation.id` | `string` | yes | | high cardinality | Owning agent invocation id | +| `diffex.turn.id` | `string` | yes | | high cardinality | Owning turn id | +| `diffex.tool.call_id` | `string` | yes | | high cardinality | Provider-assigned tool call id | +| `diffex.tool.name` | `string` | yes | | | Requested tool name | + +#### End attributes + +All end attributes are optional completion enrichment. + +| Name | Type | Values | Notes | Description | +|---|---|---|---|---| +| `diffex.tool.is_error` | `boolean` | | | Whether the finalized tool result is an error | +| `diffex.tool.outcome` | `string` | completed, failed, aborted | | Terminal tool-call outcome | + +#### Events + +No declared span events. + ## Harness schema Schema version: 1 diff --git a/packages/agent/scripts/generate-telemetry-docs.ts b/packages/agent/scripts/generate-telemetry-docs.ts index 52c69726c..c05b6edbe 100644 --- a/packages/agent/scripts/generate-telemetry-docs.ts +++ b/packages/agent/scripts/generate-telemetry-docs.ts @@ -2,7 +2,11 @@ import { mkdirSync, readFileSync, writeFileSync } from "node:fs"; import { dirname, resolve } from "node:path"; import { pathToFileURL } from "node:url"; import type { TelemetryAttributeDefinition, TelemetrySchemaDefinition } from "@earendil-works/diffex-telemetry"; -import { AI_TELEMETRY_SCHEMA, HARNESS_TELEMETRY_SCHEMA } from "../src/harness/telemetry.ts"; +import { + AGENT_RUNTIME_TELEMETRY_SCHEMA, + AI_TELEMETRY_SCHEMA, + HARNESS_TELEMETRY_SCHEMA, +} from "../src/harness/telemetry.ts"; function escapeCell(value: string): string { return value.replaceAll("|", "\\|").replaceAll("\n", " "); @@ -88,6 +92,7 @@ export function renderAgentTelemetrySchemaMarkdown(): string { "", "", ...renderSchema(AI_TELEMETRY_SCHEMA, "AI request schema"), + ...renderSchema(AGENT_RUNTIME_TELEMETRY_SCHEMA, "Active agent runtime schema"), ...renderSchema(HARNESS_TELEMETRY_SCHEMA, "Harness schema"), ]; return `${lines.join("\n").trimEnd()}\n`; diff --git a/packages/agent/src/agent-loop.ts b/packages/agent/src/agent-loop.ts index b89f97f9d..c218cd7e6 100644 --- a/packages/agent/src/agent-loop.ts +++ b/packages/agent/src/agent-loop.ts @@ -8,8 +8,16 @@ import { type Context, EventStream, type ToolResultMessage, + uuidv7, validateToolArguments, } from "@earendil-works/diffex-ai"; +import { NOOP_TELEMETRY_CONTEXT } from "@earendil-works/diffex-telemetry"; +import { + type AgentRuntimeTelemetrySpan, + type AiTelemetrySpan, + startAgentRuntimeSpan, + startAiSpan, +} from "./harness/telemetry.ts"; import { getDefaultStreamFn } from "./stream-fn.ts"; import type { AgentContext, @@ -100,21 +108,48 @@ export async function runAgentLoop( signal: AbortSignal | undefined, streamFn: StreamFn, ): Promise { - const newMessages: AgentMessage[] = [...prompts]; - const currentContext: AgentContext = { - ...context, - messages: [...context.messages, ...prompts], - }; + const invocationId = config.invocationId ?? uuidv7(); + return startAgentRuntimeSpan( + config.telemetryContext ?? NOOP_TELEMETRY_CONTEXT, + "diffex.agent.run", + { + "diffex.invocation.id": invocationId, + ...(config.sessionId === undefined ? {} : { "diffex.session.id": config.sessionId }), + "diffex.agent.run_kind": "prompt", + }, + async (span) => { + const newMessages: AgentMessage[] = [...prompts]; + const currentContext: AgentContext = { + ...context, + messages: [...context.messages, ...prompts], + }; + const instrumentedConfig = { ...config, telemetryContext: span, invocationId }; - await emit({ type: "agent_start" }); - await emit({ type: "turn_start" }); - for (const prompt of prompts) { - await emit({ type: "message_start", message: prompt }); - await emit({ type: "message_end", message: prompt }); - } + try { + await emit({ type: "agent_start" }); + await emit({ type: "turn_start" }); + for (const prompt of prompts) { + await emit({ type: "message_start", message: prompt }); + await emit({ type: "message_end", message: prompt }); + } - await runLoop(currentContext, newMessages, config, signal, emit, streamFn ?? getDefaultStreamFn()); - return newMessages; + const turnCount = await runLoop( + currentContext, + newMessages, + instrumentedConfig, + signal, + emit, + streamFn ?? getDefaultStreamFn(), + ); + settleAgentRunSpan(span, newMessages, turnCount); + return newMessages; + } catch (error) { + span.setAttributes({ "diffex.agent.outcome": signal?.aborted ? "aborted" : "failed" }); + span.setStatus({ status: "error" }); + throw error; + } + }, + ); } export async function runAgentLoopContinue( @@ -132,14 +167,66 @@ export async function runAgentLoopContinue( throw new Error("Cannot continue from message role: assistant"); } - const newMessages: AgentMessage[] = []; - const currentContext: AgentContext = { ...context }; + const invocationId = config.invocationId ?? uuidv7(); + return startAgentRuntimeSpan( + config.telemetryContext ?? NOOP_TELEMETRY_CONTEXT, + "diffex.agent.run", + { + "diffex.invocation.id": invocationId, + ...(config.sessionId === undefined ? {} : { "diffex.session.id": config.sessionId }), + "diffex.agent.run_kind": "continue", + }, + async (span) => { + const newMessages: AgentMessage[] = []; + const currentContext: AgentContext = { ...context }; + const instrumentedConfig = { ...config, telemetryContext: span, invocationId }; - await emit({ type: "agent_start" }); - await emit({ type: "turn_start" }); + try { + await emit({ type: "agent_start" }); + await emit({ type: "turn_start" }); + const turnCount = await runLoop( + currentContext, + newMessages, + instrumentedConfig, + signal, + emit, + streamFn ?? getDefaultStreamFn(), + ); + settleAgentRunSpan(span, newMessages, turnCount); + return newMessages; + } catch (error) { + span.setAttributes({ "diffex.agent.outcome": signal?.aborted ? "aborted" : "failed" }); + span.setStatus({ status: "error" }); + throw error; + } + }, + ); +} - await runLoop(currentContext, newMessages, config, signal, emit, streamFn ?? getDefaultStreamFn()); - return newMessages; +function getAgentOutcome(messages: AgentMessage[]): "completed" | "failed" | "aborted" { + for (let index = messages.length - 1; index >= 0; index--) { + const message = messages[index]; + if (message.role !== "assistant") continue; + if (message.stopReason === "aborted") return "aborted"; + if (message.stopReason === "error") return "failed"; + return "completed"; + } + return "completed"; +} + +function settleAgentRunSpan( + span: AgentRuntimeTelemetrySpan<"diffex.agent.run">, + messages: AgentMessage[], + turnCount: number, +): void { + const outcome = getAgentOutcome(messages); + span.setAttributes({ + "diffex.agent.outcome": outcome, + "diffex.agent.turn_count": turnCount, + }); + if (outcome !== "completed") { + span.setStatus({ status: "error" }); + } } function createAgentStream(): EventStream { @@ -159,10 +246,11 @@ async function runLoop( signal: AbortSignal | undefined, emit: AgentEventSink, streamFunction: StreamFn, -): Promise { +): Promise { let currentContext = initialContext; let config = initialConfig; let firstTurn = true; + let turnIndex = 0; // Check for steering messages at start (user may have typed while waiting) let pendingMessages: AgentMessage[] = (await config.getSteeringMessages?.()) || []; @@ -172,91 +260,36 @@ async function runLoop( // Inner loop: process tool calls and steering messages while (hasMoreToolCalls || pendingMessages.length > 0) { - if (!firstTurn) { - await emit({ type: "turn_start" }); - } else { - firstTurn = false; - } - - // Process pending messages (inject before next assistant response) - if (pendingMessages.length > 0) { - for (const message of pendingMessages) { - await emit({ type: "message_start", message }); - await emit({ type: "message_end", message }); - currentContext.messages.push(message); - newMessages.push(message); - } - pendingMessages = []; - } - - // Stream assistant response - const message = await streamAssistantResponse(currentContext, config, signal, emit, streamFunction); - newMessages.push(message); - - if (message.stopReason === "error" || message.stopReason === "aborted") { - await emit({ type: "turn_end", message, toolResults: [] }); - await emit({ type: "agent_end", messages: newMessages }); - return; - } - - // Check for tool calls - const toolCalls = message.content.filter((c) => c.type === "toolCall"); - - const toolResults: ToolResultMessage[] = []; - hasMoreToolCalls = false; - if (toolCalls.length > 0) { - // A "length" stop means the output was cut off by the token limit, so - // every tool call in the message may carry truncated arguments. Fail - // them all instead of executing potentially borked calls. - const executedToolBatch = - message.stopReason === "length" - ? await failToolCallsFromTruncatedMessage(toolCalls, emit) - : await executeToolCalls(currentContext, message, config, signal, emit); - toolResults.push(...executedToolBatch.messages); - hasMoreToolCalls = !executedToolBatch.terminate; - - for (const result of toolResults) { - currentContext.messages.push(result); - newMessages.push(result); - } - } - - await emit({ type: "turn_end", message, toolResults }); - - const nextTurnContext = { - message, - toolResults, - context: currentContext, - newMessages, - }; - const nextTurnSnapshot = await config.prepareNextTurn?.(nextTurnContext); - if (nextTurnSnapshot) { - currentContext = nextTurnSnapshot.context ?? currentContext; - config = { - ...config, - model: nextTurnSnapshot.model ?? config.model, - reasoning: - nextTurnSnapshot.thinkingLevel === undefined - ? config.reasoning - : nextTurnSnapshot.thinkingLevel === "off" - ? undefined - : nextTurnSnapshot.thinkingLevel, - }; - } - - if ( - await config.shouldStopAfterTurn?.({ - message, - toolResults, - context: currentContext, - newMessages, - }) - ) { - await emit({ type: "agent_end", messages: newMessages }); - return; - } - - pendingMessages = (await config.getSteeringMessages?.()) || []; + const turnId = uuidv7(); + const runTelemetryContext = config.telemetryContext ?? NOOP_TELEMETRY_CONTEXT; + const result = await startAgentRuntimeSpan( + runTelemetryContext, + "diffex.agent.turn", + { + "diffex.invocation.id": config.invocationId ?? uuidv7(), + "diffex.turn.id": turnId, + "diffex.turn.index": turnIndex, + }, + (span) => + runTurn({ + currentContext, + newMessages, + config: { ...config, telemetryContext: span, turnId }, + signal, + emit, + streamFunction, + pendingMessages, + firstTurn, + span, + }), + ); + turnIndex++; + firstTurn = false; + currentContext = result.currentContext; + config = { ...result.config, telemetryContext: runTelemetryContext, turnId: undefined }; + pendingMessages = result.pendingMessages; + hasMoreToolCalls = result.hasMoreToolCalls; + if (result.ended) return turnIndex; } // Agent would stop here. Check for follow-up messages. @@ -272,6 +305,115 @@ async function runLoop( } await emit({ type: "agent_end", messages: newMessages }); + return turnIndex; +} + +interface RunTurnOptions { + currentContext: AgentContext; + newMessages: AgentMessage[]; + config: AgentLoopConfig; + signal: AbortSignal | undefined; + emit: AgentEventSink; + streamFunction: StreamFn; + pendingMessages: AgentMessage[]; + firstTurn: boolean; + span: AgentRuntimeTelemetrySpan<"diffex.agent.turn">; +} + +interface RunTurnResult { + currentContext: AgentContext; + config: AgentLoopConfig; + pendingMessages: AgentMessage[]; + hasMoreToolCalls: boolean; + ended: boolean; +} + +async function runTurn(options: RunTurnOptions): Promise { + let { currentContext, config, pendingMessages } = options; + if (!options.firstTurn) { + await options.emit({ type: "turn_start" }); + } + + if (pendingMessages.length > 0) { + for (const message of pendingMessages) { + await options.emit({ type: "message_start", message }); + await options.emit({ type: "message_end", message }); + currentContext.messages.push(message); + options.newMessages.push(message); + } + pendingMessages = []; + } + + const message = await streamAssistantResponse( + currentContext, + config, + options.signal, + options.emit, + options.streamFunction, + ); + options.newMessages.push(message); + const outcome = + message.stopReason === "aborted" ? "aborted" : message.stopReason === "error" ? "failed" : "completed"; + options.span.setAttributes({ "diffex.agent.outcome": outcome }); + if (outcome !== "completed") { + options.span.setStatus({ status: "error" }); + await options.emit({ type: "turn_end", message, toolResults: [] }); + await options.emit({ type: "agent_end", messages: options.newMessages }); + return { currentContext, config, pendingMessages, hasMoreToolCalls: false, ended: true }; + } + + const toolCalls = message.content.filter((content) => content.type === "toolCall"); + const toolResults: ToolResultMessage[] = []; + let hasMoreToolCalls = false; + if (toolCalls.length > 0) { + const executedToolBatch = + message.stopReason === "length" + ? await failToolCallsFromTruncatedMessage(toolCalls, config, options.signal, options.emit) + : await executeToolCalls(currentContext, message, config, options.signal, options.emit); + toolResults.push(...executedToolBatch.messages); + hasMoreToolCalls = !executedToolBatch.terminate; + for (const result of toolResults) { + currentContext.messages.push(result); + options.newMessages.push(result); + } + } + + await options.emit({ type: "turn_end", message, toolResults }); + const nextTurnContext = { + message, + toolResults, + context: currentContext, + newMessages: options.newMessages, + }; + const nextTurnSnapshot = await config.prepareNextTurn?.(nextTurnContext); + if (nextTurnSnapshot) { + currentContext = nextTurnSnapshot.context ?? currentContext; + config = { + ...config, + model: nextTurnSnapshot.model ?? config.model, + reasoning: + nextTurnSnapshot.thinkingLevel === undefined + ? config.reasoning + : nextTurnSnapshot.thinkingLevel === "off" + ? undefined + : nextTurnSnapshot.thinkingLevel, + }; + } + + if ( + await config.shouldStopAfterTurn?.({ + message, + toolResults, + context: currentContext, + newMessages: options.newMessages, + }) + ) { + await options.emit({ type: "agent_end", messages: options.newMessages }); + return { currentContext, config, pendingMessages, hasMoreToolCalls, ended: true }; + } + + pendingMessages = (await config.getSteeringMessages?.()) || []; + return { currentContext, config, pendingMessages, hasMoreToolCalls, ended: false }; } /** @@ -304,71 +446,124 @@ async function streamAssistantResponse( // Resolve API key (important for expiring tokens) const resolvedApiKey = (config.getApiKey ? await config.getApiKey(config.model.provider) : undefined) || config.apiKey; - - const response = await streamFunction(config.model, llmContext, { - ...config, - apiKey: resolvedApiKey, - signal, - }); - - let partialMessage: AssistantMessage | null = null; - let addedPartial = false; - - for await (const event of response) { - switch (event.type) { - case "start": - partialMessage = event.partial; - context.messages.push(partialMessage); - addedPartial = true; - await emit({ type: "message_start", message: { ...partialMessage } }); - break; - - case "text_start": - case "text_delta": - case "text_end": - case "thinking_start": - case "thinking_delta": - case "thinking_end": - case "toolcall_start": - case "toolcall_delta": - case "toolcall_end": - if (partialMessage) { - partialMessage = event.partial; - context.messages[context.messages.length - 1] = partialMessage; - await emit({ - type: "message_update", - assistantMessageEvent: event, - message: { ...partialMessage }, - }); + const requestId = uuidv7(); + return startAiSpan( + config.telemetryContext ?? NOOP_TELEMETRY_CONTEXT, + "diffex.ai.request", + { + ...(config.invocationId === undefined ? {} : { "diffex.invocation.id": config.invocationId }), + ...(config.turnId === undefined ? {} : { "diffex.turn.id": config.turnId }), + "diffex.request.id": requestId, + "diffex.ai.operation": "stream", + "diffex.ai.provider": config.model.provider, + "diffex.ai.model": config.model.id, + "diffex.ai.api": config.model.api, + "diffex.ai.streaming": true, + }, + async (span) => { + const startedAt = performance.now(); + let firstChunkAt: number | undefined; + let chunkCount = 0; + const response = await streamFunction(config.model, llmContext, { + ...config, + apiKey: resolvedApiKey, + signal, + telemetryContext: span, + }); + + let partialMessage: AssistantMessage | null = null; + let addedPartial = false; + + for await (const event of response) { + chunkCount++; + firstChunkAt ??= performance.now(); + switch (event.type) { + case "start": + partialMessage = event.partial; + context.messages.push(partialMessage); + addedPartial = true; + await emit({ type: "message_start", message: { ...partialMessage } }); + break; + + case "text_start": + case "text_delta": + case "text_end": + case "thinking_start": + case "thinking_delta": + case "thinking_end": + case "toolcall_start": + case "toolcall_delta": + case "toolcall_end": + if (partialMessage) { + partialMessage = event.partial; + context.messages[context.messages.length - 1] = partialMessage; + await emit({ + type: "message_update", + assistantMessageEvent: event, + message: { ...partialMessage }, + }); + } + break; + + case "done": + case "error": { + const finalMessage = await response.result(); + if (addedPartial) { + context.messages[context.messages.length - 1] = finalMessage; + } else { + context.messages.push(finalMessage); + } + if (!addedPartial) { + await emit({ type: "message_start", message: { ...finalMessage } }); + } + await emit({ type: "message_end", message: finalMessage }); + settleAiRequestSpan(span, finalMessage, chunkCount, startedAt, firstChunkAt); + return finalMessage; + } } - break; + } - case "done": - case "error": { - const finalMessage = await response.result(); - if (addedPartial) { - context.messages[context.messages.length - 1] = finalMessage; - } else { - context.messages.push(finalMessage); - } - if (!addedPartial) { - await emit({ type: "message_start", message: { ...finalMessage } }); - } - await emit({ type: "message_end", message: finalMessage }); - return finalMessage; + const finalMessage = await response.result(); + if (addedPartial) { + context.messages[context.messages.length - 1] = finalMessage; + } else { + context.messages.push(finalMessage); + await emit({ type: "message_start", message: { ...finalMessage } }); } - } - } + await emit({ type: "message_end", message: finalMessage }); + settleAiRequestSpan(span, finalMessage, chunkCount, startedAt, firstChunkAt); + return finalMessage; + }, + ); +} - const finalMessage = await response.result(); - if (addedPartial) { - context.messages[context.messages.length - 1] = finalMessage; - } else { - context.messages.push(finalMessage); - await emit({ type: "message_start", message: { ...finalMessage } }); +function settleAiRequestSpan( + span: AiTelemetrySpan<"diffex.ai.request">, + message: AssistantMessage, + chunkCount: number, + startedAt: number, + firstChunkAt: number | undefined, +): void { + const stopReason = + message.stopReason === "pending" ? undefined : message.stopReason === "toolUse" ? "tool_use" : message.stopReason; + span.setAttributes({ + ...(message.responseModel === undefined ? {} : { "diffex.ai.response.model": message.responseModel }), + ...(message.responseId === undefined ? {} : { "diffex.ai.response.id": message.responseId }), + ...(stopReason === undefined ? {} : { "diffex.ai.response.stop_reason": stopReason }), + "diffex.ai.usage.input_tokens": message.usage.input, + "diffex.ai.usage.output_tokens": message.usage.output, + "diffex.ai.usage.cache_read_tokens": message.usage.cacheRead, + "diffex.ai.usage.cache_write_tokens": message.usage.cacheWrite, + ...(message.usage.reasoning === undefined ? {} : { "diffex.ai.usage.reasoning_tokens": message.usage.reasoning }), + "diffex.ai.usage.total_tokens": message.usage.totalTokens, + "diffex.ai.usage.cost": message.usage.cost.total, + "diffex.ai.stream.chunk_count": chunkCount, + ...(firstChunkAt === undefined ? {} : { "diffex.ai.stream.time_to_first_chunk_ms": firstChunkAt - startedAt }), + ...(message.stopReason === "error" ? { "diffex.ai.error.type": "AssistantMessageError" } : {}), + }); + if (message.stopReason === "error" || message.stopReason === "aborted") { + span.setStatus({ status: "error" }); } - await emit({ type: "message_end", message: finalMessage }); - return finalMessage; } /** @@ -380,6 +575,8 @@ async function streamAssistantResponse( */ async function failToolCallsFromTruncatedMessage( toolCalls: AgentToolCall[], + config: AgentLoopConfig, + signal: AbortSignal | undefined, emit: AgentEventSink, ): Promise { const messages: ToolResultMessage[] = []; @@ -390,13 +587,13 @@ async function failToolCallsFromTruncatedMessage( toolName: toolCall.name, args: toolCall.arguments, }); - const finalized: FinalizedToolCallOutcome = { + const finalized = await withToolSpan(config, signal, toolCall, async () => ({ toolCall, result: createErrorToolResult( `Tool call "${toolCall.name}" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments.`, ), isError: true, - }; + })); await emitToolExecutionEnd(finalized, emit); const toolResultMessage = createToolResultMessage(finalized); await emitToolResultMessage(toolResultMessage, emit); @@ -449,25 +646,18 @@ async function executeToolCallsSequential( args: toolCall.arguments, }); - const preparation = await prepareToolCall(currentContext, assistantMessage, toolCall, config, signal); - let finalized: FinalizedToolCallOutcome; - if (preparation.kind === "immediate") { - finalized = { - toolCall, - result: preparation.result, - isError: preparation.isError, - }; - } else { + const finalized = await withToolSpan(config, signal, toolCall, async () => { + const preparation = await prepareToolCall(currentContext, assistantMessage, toolCall, config, signal); + if (preparation.kind === "immediate") { + return { + toolCall, + result: preparation.result, + isError: preparation.isError, + }; + } const executed = await executePreparedToolCall(preparation, signal, emit); - finalized = await finalizeExecutedToolCall( - currentContext, - assistantMessage, - preparation, - executed, - config, - signal, - ); - } + return finalizeExecutedToolCall(currentContext, assistantMessage, preparation, executed, config, signal); + }); await emitToolExecutionEnd(finalized, emit); const toolResultMessage = createToolResultMessage(finalized); @@ -506,11 +696,11 @@ async function executeToolCallsParallel( const preparation = await prepareToolCall(currentContext, assistantMessage, toolCall, config, signal); if (preparation.kind === "immediate") { - const finalized = { + const finalized = await withToolSpan(config, signal, toolCall, async () => ({ toolCall, result: preparation.result, isError: preparation.isError, - } satisfies FinalizedToolCallOutcome; + })); await emitToolExecutionEnd(finalized, emit); finalizedCalls.push(finalized); if (signal?.aborted) { @@ -520,15 +710,10 @@ async function executeToolCallsParallel( } finalizedCalls.push(async () => { - const executed = await executePreparedToolCall(preparation, signal, emit); - const finalized = await finalizeExecutedToolCall( - currentContext, - assistantMessage, - preparation, - executed, - config, - signal, - ); + const finalized = await withToolSpan(config, signal, toolCall, async () => { + const executed = await executePreparedToolCall(preparation, signal, emit); + return finalizeExecutedToolCall(currentContext, assistantMessage, preparation, executed, config, signal); + }); await emitToolExecutionEnd(finalized, emit); return finalized; }); @@ -579,6 +764,34 @@ type FinalizedToolCallOutcome = { type FinalizedToolCallEntry = FinalizedToolCallOutcome | (() => Promise); +async function withToolSpan( + config: AgentLoopConfig, + signal: AbortSignal | undefined, + toolCall: AgentToolCall, + run: () => Promise, +): Promise { + return startAgentRuntimeSpan( + config.telemetryContext ?? NOOP_TELEMETRY_CONTEXT, + "diffex.agent.tool", + { + "diffex.invocation.id": config.invocationId ?? uuidv7(), + "diffex.turn.id": config.turnId ?? uuidv7(), + "diffex.tool.call_id": toolCall.id, + "diffex.tool.name": toolCall.name, + }, + async (span) => { + const finalized = await run(); + const outcome = finalized.isError ? (signal?.aborted ? "aborted" : "failed") : "completed"; + span.setAttributes({ + "diffex.tool.is_error": finalized.isError, + "diffex.tool.outcome": outcome, + }); + if (finalized.isError) span.setStatus({ status: "error" }); + return finalized; + }, + ); +} + function shouldTerminateToolBatch(finalizedCalls: FinalizedToolCallOutcome[]): boolean { return finalizedCalls.length > 0 && finalizedCalls.every((finalized) => finalized.result.terminate === true); } diff --git a/packages/agent/src/agent.ts b/packages/agent/src/agent.ts index e76e07462..19d2b99fe 100644 --- a/packages/agent/src/agent.ts +++ b/packages/agent/src/agent.ts @@ -7,6 +7,7 @@ import type { ThinkingBudgets, Transport, } from "@earendil-works/diffex-ai"; +import { NOOP_TELEMETRY_CONTEXT, type TelemetryContext } from "@earendil-works/diffex-telemetry"; import { runAgentLoop, runAgentLoopContinue } from "./agent-loop.ts"; import { getDefaultStreamFn } from "./stream-fn.ts"; import type { @@ -120,6 +121,8 @@ export interface AgentOptions { transport?: Transport; maxRetryDelayMs?: number; toolExecution?: ToolExecutionMode; + /** Explicit parent context for agent runtime telemetry. */ + telemetryContext?: TelemetryContext; } class PendingMessageQueue { @@ -212,6 +215,8 @@ export class Agent { public maxRetryDelayMs?: number; /** Tool execution strategy for assistant messages that contain multiple tool calls. */ public toolExecution: ToolExecutionMode; + /** Explicit parent context used by future prompt and continuation admissions. */ + public telemetryContext: TelemetryContext; constructor(options: AgentOptions) { // Older compiled consumers may omit options or streamFn even though the current API requires them. @@ -235,6 +240,7 @@ export class Agent { this.transport = runtimeOptions.transport ?? "auto"; this.maxRetryDelayMs = runtimeOptions.maxRetryDelayMs; this.toolExecution = runtimeOptions.toolExecution ?? "parallel"; + this.telemetryContext = runtimeOptions.telemetryContext ?? NOOP_TELEMETRY_CONTEXT; } /** @@ -455,6 +461,7 @@ export class Agent { thinkingBudgets: this.thinkingBudgets, maxRetryDelayMs: this.maxRetryDelayMs, toolExecution: this.toolExecution, + telemetryContext: this.telemetryContext, beforeToolCall: this.beforeToolCall, afterToolCall: this.afterToolCall, shouldStopAfterTurn: shouldStopAfterTurn diff --git a/packages/agent/src/harness/telemetry.ts b/packages/agent/src/harness/telemetry.ts index 85ac0a316..692fb4225 100644 --- a/packages/agent/src/harness/telemetry.ts +++ b/packages/agent/src/harness/telemetry.ts @@ -46,6 +46,24 @@ export const AI_TELEMETRY_SCHEMA = { description: "One logical request to an AI provider", parents: { kind: "any" }, startAttributes: { + "diffex.invocation.id": { + type: "string", + required: false, + cardinality: "high", + description: "Owning in-process agent invocation id", + }, + "diffex.turn.id": { + type: "string", + required: false, + cardinality: "high", + description: "Owning in-process turn id", + }, + "diffex.request.id": { + type: "string", + required: false, + cardinality: "high", + description: "Logical provider request id", + }, "diffex.ai.operation": { type: "string", required: true, @@ -144,6 +162,148 @@ export function startAiSpan callback(span as AiTelemetrySpan)); } +export const AGENT_RUNTIME_TELEMETRY_SCHEMA = { + version: 1, + spans: { + "diffex.agent.run": { + description: "One in-process agent prompt or continuation", + parents: { kind: "root_or_external" }, + startAttributes: { + "diffex.invocation.id": { + type: "string", + required: true, + cardinality: "high", + description: "Invocation-local run id", + }, + "diffex.session.id": { + type: "string", + required: false, + cardinality: "high", + description: "Caller-provided session id", + }, + "diffex.agent.run_kind": { + type: "string", + required: true, + values: ["prompt", "continue"], + description: "Agent admission path", + }, + }, + endAttributes: { + "diffex.agent.outcome": { + type: "string", + values: ["completed", "failed", "aborted"], + description: "Terminal invocation outcome", + }, + "diffex.agent.turn_count": { + type: "number", + description: "Number of admitted turns", + }, + }, + status: { default: "ok", errorWhen: "The invocation fails or is aborted" }, + }, + "diffex.agent.turn": { + description: "One assistant response and its tool batch in the active agent runtime", + parents: { kind: "spans", spans: ["diffex.agent.run"] }, + startAttributes: { + "diffex.invocation.id": { + type: "string", + required: true, + cardinality: "high", + description: "Owning agent invocation id", + }, + "diffex.turn.id": { + type: "string", + required: true, + cardinality: "high", + description: "Invocation-local turn id", + }, + "diffex.turn.index": { + type: "number", + required: true, + description: "Zero-based turn index", + }, + }, + endAttributes: { + "diffex.agent.outcome": { + type: "string", + values: ["completed", "failed", "aborted"], + description: "Terminal turn outcome", + }, + }, + status: { default: "ok", errorWhen: "The turn fails or is aborted" }, + }, + "diffex.agent.tool": { + description: "One tool call admitted by the active agent runtime", + parents: { kind: "spans", spans: ["diffex.agent.turn"] }, + startAttributes: { + "diffex.invocation.id": { + type: "string", + required: true, + cardinality: "high", + description: "Owning agent invocation id", + }, + "diffex.turn.id": { + type: "string", + required: true, + cardinality: "high", + description: "Owning turn id", + }, + "diffex.tool.call_id": { + type: "string", + required: true, + cardinality: "high", + description: "Provider-assigned tool call id", + }, + "diffex.tool.name": { + type: "string", + required: true, + description: "Requested tool name", + }, + }, + endAttributes: { + "diffex.tool.is_error": { + type: "boolean", + description: "Whether the finalized tool result is an error", + }, + "diffex.tool.outcome": { + type: "string", + values: ["completed", "failed", "aborted"], + description: "Terminal tool-call outcome", + }, + }, + status: { default: "ok", errorWhen: "The finalized tool result is an error" }, + }, + }, +} as const satisfies TelemetrySchemaDefinition; + +export type AgentRuntimeSpanName = TelemetrySchemaSpanName; +export type AgentRuntimeSpanStartAttributes = TelemetrySchemaSpanStartAttributes< + typeof AGENT_RUNTIME_TELEMETRY_SCHEMA, + Name +>; +export type AgentRuntimeSpanEndAttributes = TelemetrySchemaSpanEndAttributes< + typeof AGENT_RUNTIME_TELEMETRY_SCHEMA, + Name +>; +export type AgentRuntimeTelemetrySpan = SchemaTelemetrySpan< + typeof AGENT_RUNTIME_TELEMETRY_SCHEMA, + Name +>; +export type AgentRuntimeSpan = TelemetrySchemaSpanUnion; + +export function startAgentRuntimeSpan< + Name extends AgentRuntimeSpanName, + const Attributes extends AgentRuntimeSpanStartAttributes, + Result, +>( + telemetryContext: TelemetryContext, + name: Name, + attributes: ExactTelemetryAttributes, Attributes>, + callback: (span: AgentRuntimeTelemetrySpan) => Result | Promise, +): Promise { + return telemetryContext.startSpan({ name, attributes }, (span) => callback(span as AgentRuntimeTelemetrySpan)); +} + const HOOK_NAMES = [ "before_run", "before_resume", @@ -576,8 +736,12 @@ export const HARNESS_TELEMETRY_SCHEMA = { }, } as const satisfies TelemetrySchemaDefinition; -/** Combined typed span vocabulary for agent-owned AI-request and harness telemetry. */ -export const AGENT_TELEMETRY_SCHEMAS = [AI_TELEMETRY_SCHEMA, HARNESS_TELEMETRY_SCHEMA] as const; +/** Combined typed span vocabulary for agent-owned AI-request, runtime, and harness telemetry. */ +export const AGENT_TELEMETRY_SCHEMAS = [ + AI_TELEMETRY_SCHEMA, + AGENT_RUNTIME_TELEMETRY_SCHEMA, + HARNESS_TELEMETRY_SCHEMA, +] as const; export type HarnessSpanName = TelemetrySchemaSpanName; export type HarnessSpanStartAttributes = TelemetrySchemaSpanStartAttributes< diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index 99bedec76..53b409e03 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -82,6 +82,11 @@ export * from "./harness/session/search.ts"; export * from "./harness/skills.ts"; export * from "./harness/system-prompt.ts"; export type { + AgentRuntimeSpan, + AgentRuntimeSpanEndAttributes, + AgentRuntimeSpanName, + AgentRuntimeSpanStartAttributes, + AgentRuntimeTelemetrySpan, AiSpan, AiSpanAttributes, AiSpanEndAttributes, @@ -100,9 +105,11 @@ export type { HarnessTelemetrySpan, } from "./harness/telemetry.ts"; export { + AGENT_RUNTIME_TELEMETRY_SCHEMA, AGENT_TELEMETRY_SCHEMAS, AI_TELEMETRY_SCHEMA, HARNESS_TELEMETRY_SCHEMA, + startAgentRuntimeSpan, startAiSpan, startHarnessSpan, } from "./harness/telemetry.ts"; diff --git a/packages/agent/src/types.ts b/packages/agent/src/types.ts index 87ebcf46e..02beacf21 100644 --- a/packages/agent/src/types.ts +++ b/packages/agent/src/types.ts @@ -13,6 +13,7 @@ import type { ToolResultMessage, Usage, } from "@earendil-works/diffex-ai"; +import type { TelemetryContext } from "@earendil-works/diffex-telemetry"; import type { Static, TSchema } from "typebox"; /** @@ -148,6 +149,12 @@ export interface PrepareNextTurnContext extends ShouldStopAfterTurnContext {} export interface AgentLoopConfig extends SimpleStreamOptions { model: Model; + /** Explicit parent context for this in-process agent invocation. */ + telemetryContext?: TelemetryContext; + /** Stable invocation id supplied by a higher-level admission boundary. */ + invocationId?: string; + /** Stable turn id supplied by the active loop turn. */ + turnId?: string; /** * Converts AgentMessage[] to LLM-compatible Message[] before each LLM call. diff --git a/packages/agent/test/agent.test.ts b/packages/agent/test/agent.test.ts index 0a2544af6..d8c9a522a 100644 --- a/packages/agent/test/agent.test.ts +++ b/packages/agent/test/agent.test.ts @@ -11,6 +11,7 @@ import { type AgentEvent, type AgentTool, type AgentToolUpdateCallback, + InMemoryTelemetryContext, type StreamFn, setDefaultStreamFn, } from "../src/index.ts"; @@ -812,4 +813,103 @@ describe("Agent", () => { await agent.prompt("hello again"); expect(receivedSessionId).toBe("session-def"); }); + + it("records nested runtime telemetry without changing tool execution", async () => { + const telemetryContext = new InMemoryTelemetryContext(); + const schema = Type.Object({}); + const tool: AgentTool = { + name: "finish", + label: "Finish", + description: "Completes the run", + parameters: schema, + execute: async () => ({ + content: [{ type: "text", text: "done" }], + details: {}, + terminate: true, + }), + }; + const agent = new Agent({ + initialState: { tools: [tool] }, + telemetryContext, + sessionId: "session-telemetry", + streamFn: (_model, _context, options) => { + expect(options?.telemetryContext).toBeDefined(); + const stream = new MockAssistantStream(); + queueMicrotask(() => { + stream.push({ + type: "done", + reason: "toolUse", + message: createAssistantToolUseMessage([ + { type: "toolCall", id: "call-telemetry", name: "finish", arguments: {} }, + ]), + }); + }); + return stream; + }, + }); + + await agent.prompt("run the tool"); + + const spans = telemetryContext.getSpans(); + expect(spans.map((span) => span.name)).toEqual([ + "diffex.agent.run", + "diffex.agent.turn", + "diffex.ai.request", + "diffex.agent.tool", + ]); + const [runSpan, turnSpan, requestSpan, toolSpan] = spans; + expect(runSpan.parentId).toBeNull(); + expect(turnSpan.parentId).toBe(runSpan.id); + expect(requestSpan.parentId).toBe(turnSpan.id); + expect(toolSpan.parentId).toBe(turnSpan.id); + expect(runSpan.attributes).toMatchObject({ + "diffex.session.id": "session-telemetry", + "diffex.agent.outcome": "completed", + "diffex.agent.turn_count": 1, + }); + expect(requestSpan.attributes["diffex.request.id"]).toEqual(expect.any(String)); + expect(toolSpan.attributes).toMatchObject({ + "diffex.tool.call_id": "call-telemetry", + "diffex.tool.outcome": "completed", + "diffex.tool.is_error": false, + }); + expect(spans.every((span) => span.settled)).toBe(true); + }); + + it("marks returned provider failures without changing prompt settlement", async () => { + const telemetryContext = new InMemoryTelemetryContext(); + const agent = new Agent({ + telemetryContext, + streamFn: () => { + const stream = new MockAssistantStream(); + queueMicrotask(() => { + stream.push({ + type: "error", + reason: "error", + error: { + ...createAssistantMessage(""), + stopReason: "error", + errorMessage: "provider failed", + }, + }); + }); + return stream; + }, + }); + + await expect(agent.prompt("hello")).resolves.toBeUndefined(); + + const spans = telemetryContext.getSpans(); + expect(spans.find((span) => span.name === "diffex.agent.run")).toMatchObject({ + attributes: { "diffex.agent.outcome": "failed", "diffex.agent.turn_count": 1 }, + status: { status: "error" }, + }); + expect(spans.find((span) => span.name === "diffex.ai.request")).toMatchObject({ + attributes: { + "diffex.ai.response.stop_reason": "error", + "diffex.ai.error.type": "AssistantMessageError", + }, + status: { status: "error" }, + }); + }); }); diff --git a/packages/agent/test/harness/telemetry.test.ts b/packages/agent/test/harness/telemetry.test.ts index 8bd117872..affee1402 100644 --- a/packages/agent/test/harness/telemetry.test.ts +++ b/packages/agent/test/harness/telemetry.test.ts @@ -8,6 +8,7 @@ import { import { describe, expect, expectTypeOf, it } from "vitest"; import { renderAgentTelemetrySchemaMarkdown } from "../../scripts/generate-telemetry-docs.ts"; import { + AGENT_RUNTIME_TELEMETRY_SCHEMA, AGENT_TELEMETRY_SCHEMAS, AI_TELEMETRY_SCHEMA, type AiSpanEndAttributes, @@ -20,10 +21,20 @@ import { } from "../../src/harness/telemetry.ts"; describe("agent telemetry schemas", () => { - it("serializes both schemas and generates the checked-in reference", () => { + it("serializes the schemas and generates the checked-in reference", () => { expect(() => JSON.stringify(AI_TELEMETRY_SCHEMA)).not.toThrow(); + expect(() => JSON.stringify(AGENT_RUNTIME_TELEMETRY_SCHEMA)).not.toThrow(); expect(() => JSON.stringify(HARNESS_TELEMETRY_SCHEMA)).not.toThrow(); - expect(AGENT_TELEMETRY_SCHEMAS).toEqual([AI_TELEMETRY_SCHEMA, HARNESS_TELEMETRY_SCHEMA]); + expect(AGENT_TELEMETRY_SCHEMAS).toEqual([ + AI_TELEMETRY_SCHEMA, + AGENT_RUNTIME_TELEMETRY_SCHEMA, + HARNESS_TELEMETRY_SCHEMA, + ]); + expect(Object.keys(AGENT_RUNTIME_TELEMETRY_SCHEMA.spans)).toEqual([ + "diffex.agent.run", + "diffex.agent.turn", + "diffex.agent.tool", + ]); expect(Object.keys(HARNESS_TELEMETRY_SCHEMA.spans)).toEqual([ "diffex.harness.run", "diffex.harness.compaction", diff --git a/packages/coding-agent/src/core/agent-session-runtime.ts b/packages/coding-agent/src/core/agent-session-runtime.ts index 5d606d035..d6cc5d65c 100644 --- a/packages/coding-agent/src/core/agent-session-runtime.ts +++ b/packages/coding-agent/src/core/agent-session-runtime.ts @@ -1,5 +1,6 @@ import { copyFileSync, existsSync, mkdirSync } from "node:fs"; import { basename, join, resolve } from "node:path"; +import type { TelemetryContext } from "@earendil-works/diffex-agent-core"; import { resolvePath } from "../utils/paths.ts"; import type { AgentSession } from "./agent-session.ts"; import type { AgentSessionRuntimeDiagnostic, AgentSessionServices } from "./agent-session-services.ts"; @@ -38,6 +39,7 @@ export type CreateAgentSessionRuntimeFactory = (options: { sessionManager: SessionManager; sessionStartEvent?: SessionStartEvent; projectTrustContext?: ProjectTrustContext; + telemetryContext?: TelemetryContext; }) => Promise; /** @@ -90,7 +92,11 @@ export class AgentSessionRuntime { ) { this._session = _session; this._services = _services; - this.createRuntime = createRuntime; + this.createRuntime = (options) => + createRuntime({ + ...options, + telemetryContext: options.telemetryContext ?? _services.telemetryContext, + }); this._diagnostics = _diagnostics; this._modelFallbackMessage = _modelFallbackMessage; } @@ -449,6 +455,7 @@ export async function createAgentSessionRuntime( agentDir: string; sessionManager: SessionManager; sessionStartEvent?: SessionStartEvent; + telemetryContext?: TelemetryContext; }, ): Promise { assertSessionCwdExists(options.sessionManager, options.cwd); diff --git a/packages/coding-agent/src/core/agent-session-services.ts b/packages/coding-agent/src/core/agent-session-services.ts index 5876eaefa..7ae2e2083 100644 --- a/packages/coding-agent/src/core/agent-session-services.ts +++ b/packages/coding-agent/src/core/agent-session-services.ts @@ -1,5 +1,5 @@ import { join } from "node:path"; -import type { ThinkingLevel } from "@earendil-works/diffex-agent-core"; +import { NOOP_TELEMETRY_CONTEXT, type TelemetryContext, type ThinkingLevel } from "@earendil-works/diffex-agent-core"; import type { Model } from "@earendil-works/diffex-ai"; import type { HarnessStateStore, HarnessVersionBundle } from "@earendil-works/diffex-harness-state"; import { getAgentDir } from "../config.ts"; @@ -45,6 +45,9 @@ export interface CreateAgentSessionServicesOptions { extensionFlagValues?: Map; resourceLoaderOptions?: Omit; resourceLoaderReloadOptions?: ResourceLoaderReloadOptions; + harnessStateStore?: HarnessStateStore; + harnessVersionBundle?: HarnessVersionBundle; + telemetryContext?: TelemetryContext; } /** @@ -81,6 +84,7 @@ export interface AgentSessionServices { resourceLoader: ResourceLoader; harnessStateStore: HarnessStateStore; harnessVersionBundle: HarnessVersionBundle | undefined; + telemetryContext: TelemetryContext; diagnostics: AgentSessionRuntimeDiagnostic[]; } @@ -150,7 +154,9 @@ export async function createAgentSessionServices( signal: options.modelRuntimeSignal, })); const settingsManager = options.settingsManager ?? SettingsManager.create(cwd, agentDir); - const harnessState = await loadGlobalHarnessState(agentDir); + const harnessState = options.harnessStateStore + ? { store: options.harnessStateStore, bundle: options.harnessVersionBundle, warning: undefined } + : await loadGlobalHarnessState(agentDir); const resourceLoader = new DefaultResourceLoader({ ...(options.resourceLoaderOptions ?? {}), cwd, @@ -204,6 +210,7 @@ export async function createAgentSessionServices( resourceLoader, harnessStateStore: harnessState.store, harnessVersionBundle: harnessState.bundle, + telemetryContext: options.telemetryContext ?? NOOP_TELEMETRY_CONTEXT, diagnostics, }; } @@ -226,6 +233,7 @@ export async function createAgentSessionFromServices( resourceLoader: options.services.resourceLoader, harnessStateStore: options.services.harnessStateStore, harnessVersionBundle: options.services.harnessVersionBundle, + telemetryContext: options.services.telemetryContext, sessionManager: options.sessionManager, model: options.model, thinkingLevel: options.thinkingLevel, diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 984843d8b..5a785b304 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -15,14 +15,17 @@ import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; import { basename, dirname } from "node:path"; -import type { - Agent, - AgentEvent, - AgentMessage, - AgentState, - AgentTool, - PrepareNextTurnContext, - ThinkingLevel, +import { + type Agent, + type AgentEvent, + type AgentMessage, + type AgentState, + type AgentTool, + NOOP_TELEMETRY_CONTEXT, + type PrepareNextTurnContext, + type TelemetryContext, + type ThinkingLevel, + uuidv7, } from "@earendil-works/diffex-agent-core"; import { contentText } from "@earendil-works/diffex-ai"; import type { @@ -114,6 +117,7 @@ import { createSyntheticSourceInfo, type SourceInfo } from "./source-info.ts"; import type { SubagentController } from "./subagents/controller.ts"; import type { SubagentBatchResult, SubagentDeliveryBatch, SubagentSummary } from "./subagents/types.ts"; import { type BuildSystemPromptOptions, buildSystemPrompt } from "./system-prompt.ts"; +import { startCodingAgentSpan } from "./telemetry.ts"; import { type BashOperations, createLocalBashOperations } from "./tools/bash.ts"; import { createAllToolDefinitions } from "./tools/index.ts"; import { createToolDefinitionFromAgentTool } from "./tools/tool-definition-wrapper.ts"; @@ -250,6 +254,8 @@ export interface AgentSessionConfig { harnessStateStore?: HarnessStateStore; /** Harness revision activated for this session. */ harnessVersionBundle?: HarnessVersionBundle; + /** Parent context for passive runtime telemetry. */ + telemetryContext?: TelemetryContext; } export interface ExtensionBindings { @@ -405,6 +411,7 @@ export class AgentSession { private readonly _subagentController: SubagentController | undefined; private readonly _harnessStateStore: HarnessStateStore | undefined; private _harnessVersionBundle: HarnessVersionBundle | undefined; + private readonly _telemetryContext: TelemetryContext; private _harnessSwitchInProgress = false; private _subagentWaitAbortController: AbortController | undefined; private _agentRunAborted = false; @@ -428,6 +435,7 @@ export class AgentSession { this._subagentController = config.subagentController; this._harnessStateStore = config.harnessStateStore; this._harnessVersionBundle = config.harnessVersionBundle; + this._telemetryContext = config.telemetryContext ?? NOOP_TELEMETRY_CONTEXT; this._armInitialCurrentPlanRestoration(); @@ -448,6 +456,18 @@ export class AgentSession { return this._modelRuntime; } + get telemetryContext(): TelemetryContext { + return this._telemetryContext; + } + + get harnessStateStore(): HarnessStateStore | undefined { + return this._harnessStateStore; + } + + get harnessVersionBundle(): HarnessVersionBundle | undefined { + return this._harnessVersionBundle; + } + private async _getRequiredRequestAuth(model: Model): Promise<{ model: Model; apiKey?: string; @@ -1329,6 +1349,44 @@ export class AgentSession { // Prompting // ========================================================================= + private _traceOperation( + kind: "prompt" | "compaction" | "navigation" | "abort" | "queue" | "bash" | "export", + callback: (telemetryContext: TelemetryContext) => Result | Promise, + ): Promise { + return startCodingAgentSpan( + this.agent.telemetryContext, + "diffex.coding_agent.operation", + { + "diffex.operation.id": uuidv7(), + "diffex.session.id": this.sessionId, + "diffex.operation.kind": kind, + }, + async (span) => { + const previousAgentContext = this.agent.telemetryContext; + const bindsAgentContext = kind === "prompt" && !this.isStreaming; + if (bindsAgentContext) this.agent.telemetryContext = span; + try { + const result = await callback(span); + const aborted = bindsAgentContext && this._agentRunAborted; + span.setAttributes({ "diffex.operation.outcome": aborted ? "aborted" : "completed" }); + if (aborted) span.setStatus({ status: "error" }); + return result; + } catch (error) { + const aborted = + error instanceof Error && + (error.name === "AbortError" || error.message.toLowerCase().includes("cancelled")); + span.setAttributes({ "diffex.operation.outcome": aborted ? "aborted" : "failed" }); + span.setStatus({ status: "error" }); + throw error; + } finally { + if (bindsAgentContext && this.agent.telemetryContext === span) { + this.agent.telemetryContext = previousAgentContext; + } + } + }, + ); + } + private async _runAgentPrompt(messages: AgentMessage | AgentMessage[]): Promise { this._agentRunAborted = false; this._isAgentRunActive = true; @@ -1488,6 +1546,10 @@ export class AgentSession { * @throws Error if no model selected or no API key available (when not streaming) */ async prompt(text: string, options?: PromptOptions): Promise { + return this._traceOperation("prompt", () => this._prompt(text, options)); + } + + private async _prompt(text: string, options?: PromptOptions): Promise { if (this._harnessSwitchInProgress) { throw new Error("Cannot submit a prompt while a harness version switch is in progress."); } @@ -1754,6 +1816,10 @@ export class AgentSession { * Internal: Queue a steering message (already expanded, no extension command check). */ private async _queueSteer(text: string, images?: ImageContent[]): Promise { + await this._traceOperation("queue", () => this._enqueueSteer(text, images)); + } + + private async _enqueueSteer(text: string, images?: ImageContent[]): Promise { this._steeringMessages.push(text); this._emitQueueUpdate(); const content: (TextContent | ImageContent)[] = [{ type: "text", text }]; @@ -1771,6 +1837,10 @@ export class AgentSession { * Internal: Queue a follow-up message (already expanded, no extension command check). */ private async _queueFollowUp(text: string, images?: ImageContent[]): Promise { + await this._traceOperation("queue", () => this._enqueueFollowUp(text, images)); + } + + private async _enqueueFollowUp(text: string, images?: ImageContent[]): Promise { this._followUpMessages.push(text); this._emitQueueUpdate(); const content: (TextContent | ImageContent)[] = [{ type: "text", text }]; @@ -1925,6 +1995,10 @@ export class AgentSession { * Abort current operation and wait for agent to become idle. */ async abort(): Promise { + return this._traceOperation("abort", () => this._abort()); + } + + private async _abort(): Promise { this.abortRetry(); if (this._isAgentRunActive) this._agentRunAborted = true; this._subagentWaitAbortController?.abort(new Error("Subagent coordination cancelled")); @@ -2167,10 +2241,12 @@ export class AgentSession { * @param customInstructions Optional instructions for the compaction summary */ compact(customInstructions?: string): Promise { - return this._runOwnedOperation(() => this._compact(customInstructions)); + return this._runOwnedOperation(() => + this._traceOperation("compaction", (telemetryContext) => this._compact(customInstructions, telemetryContext)), + ); } - private async _compact(customInstructions?: string): Promise { + private async _compact(customInstructions?: string, telemetryContext?: TelemetryContext): Promise { await this.abort(); if (this._disposePromise) { throw new Error("Compaction cancelled"); @@ -2248,7 +2324,8 @@ export class AgentSession { this.agent.streamFunction, env, this.settingsManager.getRetrySettings(), - this._summarizationRetryCallbacks({ source: "compaction", reason: "manual" }), + this._summarizationRetryCallbacks({ source: "compaction", reason: "manual" }, telemetryContext), + telemetryContext, ); summary = result.summary; firstKeptEntryId = result.firstKeptEntryId; @@ -2474,10 +2551,18 @@ export class AgentSession { */ private _runAutoCompaction(reason: "overflow" | "threshold", willRetry: boolean): Promise { if (this._disposePromise) return Promise.resolve(false); - return this._runOwnedOperation(() => this._runAutoCompactionOperation(reason, willRetry)); + return this._runOwnedOperation(() => + this._traceOperation("compaction", (telemetryContext) => + this._runAutoCompactionOperation(reason, willRetry, telemetryContext), + ), + ); } - private async _runAutoCompactionOperation(reason: "overflow" | "threshold", willRetry: boolean): Promise { + private async _runAutoCompactionOperation( + reason: "overflow" | "threshold", + willRetry: boolean, + telemetryContext: TelemetryContext = this._telemetryContext, + ): Promise { const settings = this.settingsManager.getCompactionSettings(); let started = false; @@ -2557,7 +2642,8 @@ export class AgentSession { this.agent.streamFunction, env, this.settingsManager.getRetrySettings(), - this._summarizationRetryCallbacks({ source: "compaction", reason }), + this._summarizationRetryCallbacks({ source: "compaction", reason }, telemetryContext), + telemetryContext, ); summary = compactResult.summary; firstKeptEntryId = compactResult.firstKeptEntryId; @@ -2997,6 +3083,7 @@ export class AgentSession { this._cwd, this.sessionManager, new ModelRegistry(this._modelRuntime), + this._telemetryContext, ); if (this._extensionRunnerRef) { this._extensionRunnerRef.current = this._extensionRunner; @@ -3064,9 +3151,24 @@ export class AgentSession { */ private _summarizationRetryCallbacks( source: { source: "branchSummary" } | { source: "compaction"; reason: "manual" | "threshold" | "overflow" }, + telemetryContext: TelemetryContext = this.agent.telemetryContext, ): RetryCallbacks { return { onRetryScheduled: (attempt, maxAttempts, delayMs, errorMessage) => { + void startCodingAgentSpan( + telemetryContext, + "diffex.coding_agent.retry", + { + "diffex.session.id": this.sessionId, + "diffex.retry.kind": source.source === "branchSummary" ? "branch_summary" : "compaction", + "diffex.retry.attempt": attempt, + "diffex.retry.max_attempts": maxAttempts, + "diffex.retry.delay_ms": delayMs, + }, + (span) => { + span.setAttributes({ "diffex.retry.outcome": "scheduled" }); + }, + ); this._emit({ type: "summarization_retry_scheduled", attempt, @@ -3124,18 +3226,40 @@ export class AgentSession { // Wait with exponential backoff (abortable) this._retryAbortController = new AbortController(); try { - await sleep(delayMs, this._retryAbortController.signal); - } catch { - // Aborted during sleep - emit end event so UI can clean up - const attempt = this._retryAttempt; - this._retryAttempt = 0; - this._emit({ - type: "auto_retry_end", - success: false, - attempt, - finalError: "Retry cancelled", - }); - return false; + const resumed = await startCodingAgentSpan( + this.agent.telemetryContext, + "diffex.coding_agent.retry", + { + "diffex.session.id": this.sessionId, + "diffex.retry.kind": "provider", + "diffex.retry.attempt": this._retryAttempt, + "diffex.retry.max_attempts": settings.maxRetries, + "diffex.retry.delay_ms": delayMs, + }, + async (span) => { + try { + await sleep(delayMs, this._retryAbortController?.signal); + span.setAttributes({ "diffex.retry.outcome": "resumed" }); + return true; + } catch { + span.setAttributes({ "diffex.retry.outcome": "cancelled" }); + span.setStatus({ status: "error" }); + return false; + } + }, + ); + if (!resumed) { + // Aborted during sleep - emit end event so UI can clean up + const attempt = this._retryAttempt; + this._retryAttempt = 0; + this._emit({ + type: "auto_retry_end", + success: false, + attempt, + finalError: "Retry cancelled", + }); + return false; + } } finally { this._retryAbortController = undefined; } @@ -3185,7 +3309,25 @@ export class AgentSession { onChunk?: (chunk: string) => void, options?: { excludeFromContext?: boolean; id?: string; operations?: BashOperations }, ): Promise { - return this._runOwnedOperation(() => this._executeBash(command, onChunk, options)); + return this._runOwnedOperation(() => + this._traceOperation("bash", () => + startCodingAgentSpan( + this.agent.telemetryContext, + "diffex.coding_agent.terminal", + { "diffex.session.id": this.sessionId }, + async (span) => { + const result = await this._executeBash(command, onChunk, options); + const outcome = result.cancelled ? "cancelled" : result.exitCode === 0 ? "completed" : "failed"; + span.setAttributes({ + "diffex.terminal.exit_code": result.exitCode, + "diffex.terminal.outcome": outcome, + }); + if (outcome !== "completed") span.setStatus({ status: "error" }); + return result; + }, + ), + ), + ); } private async _executeBash( @@ -3322,12 +3464,17 @@ export class AgentSession { targetId: string, options: { summarize?: boolean; customInstructions?: string; replaceInstructions?: boolean; label?: string } = {}, ): Promise<{ editorText?: string; cancelled: boolean; aborted?: boolean; summaryEntry?: BranchSummaryEntry }> { - return this._runOwnedOperation(() => this._navigateTree(targetId, options)); + return this._runOwnedOperation(() => + this._traceOperation("navigation", (telemetryContext) => + this._navigateTree(targetId, options, telemetryContext), + ), + ); } private async _navigateTree( targetId: string, options: { summarize?: boolean; customInstructions?: string; replaceInstructions?: boolean; label?: string }, + telemetryContext: TelemetryContext, ): Promise<{ editorText?: string; cancelled: boolean; aborted?: boolean; summaryEntry?: BranchSummaryEntry }> { if (this.isStreaming) { throw new Error("Wait for the current response to finish before navigating the session tree."); @@ -3428,7 +3575,8 @@ export class AgentSession { reserveTokens: branchSummarySettings.reserveTokens, streamFn: this.agent.streamFunction, retry: this.settingsManager.getRetrySettings(), - callbacks: this._summarizationRetryCallbacks({ source: "branchSummary" }), + callbacks: this._summarizationRetryCallbacks({ source: "branchSummary" }, telemetryContext), + telemetryContext, }); if (result.aborted) { return { cancelled: true, aborted: true }; @@ -3646,6 +3794,10 @@ export class AgentSession { * @returns Path to exported file */ async exportToHtml(outputPath?: string): Promise { + return this._traceOperation("export", () => this._exportToHtml(outputPath)); + } + + private async _exportToHtml(outputPath?: string): Promise { const configuredThemeName = this.settingsManager.getTheme(); const themeName = configuredThemeName && getThemeByName(configuredThemeName) ? configuredThemeName : undefined; @@ -3656,11 +3808,30 @@ export class AgentSession { cwd: this.sessionManager.getCwd(), }); - return await exportSessionToHtml(this.sessionManager, this.state, { + const filePath = await exportSessionToHtml(this.sessionManager, this.state, { outputPath, themeName, toolRenderer, }); + this._recordArtifact("html", filePath); + return filePath; + } + + private _recordArtifact(kind: "html" | "jsonl", filePath: string): void { + void startCodingAgentSpan( + this.agent.telemetryContext, + "diffex.coding_agent.artifact", + { + "diffex.session.id": this.sessionId, + "diffex.artifact.kind": kind, + }, + (span) => { + span.setAttributes({ + "diffex.artifact.path": filePath, + "diffex.artifact.outcome": "written", + }); + }, + ); } /** @@ -3699,6 +3870,7 @@ export class AgentSession { } writeFileSync(filePath, `${lines.join("\n")}\n`); + this._recordArtifact("jsonl", filePath); return filePath; } diff --git a/packages/coding-agent/src/core/compaction/branch-summarization.ts b/packages/coding-agent/src/core/compaction/branch-summarization.ts index e51380855..cef01afa0 100644 --- a/packages/coding-agent/src/core/compaction/branch-summarization.ts +++ b/packages/coding-agent/src/core/compaction/branch-summarization.ts @@ -5,7 +5,7 @@ * a summary of the branch being left so context isn't lost. */ -import type { AgentMessage, StreamFn } from "@earendil-works/diffex-agent-core"; +import type { AgentMessage, StreamFn, TelemetryContext } from "@earendil-works/diffex-agent-core"; import type { RetryCallbacks, RetryPolicy } from "@earendil-works/diffex-ai"; import { contentText } from "@earendil-works/diffex-ai"; import type { Model, SimpleStreamOptions, Usage } from "@earendil-works/diffex-ai/compat"; @@ -87,6 +87,8 @@ export interface GenerateBranchSummaryOptions { retry?: RetryPolicy; /** Optional callbacks for retry reporting (e.g. TUI retry indicators). */ callbacks?: RetryCallbacks; + /** Explicit parent for provider-request telemetry. */ + telemetryContext?: TelemetryContext; } // ============================================================================ @@ -306,6 +308,7 @@ export async function generateBranchSummary( streamFn, retry, callbacks, + telemetryContext, } = options; // Token budget = context window minus reserved space for prompt + response @@ -347,7 +350,7 @@ export async function generateBranchSummary( // without running through agent state/events. Retried via completeSummarization // so transient stream drops reuse the configured retry policy. const context = { systemPrompt: SUMMARIZATION_SYSTEM_PROMPT, messages: summarizationMessages }; - const requestOptions: SimpleStreamOptions = { apiKey, headers, env, signal, maxTokens: 2048 }; + const requestOptions: SimpleStreamOptions = { apiKey, headers, env, signal, maxTokens: 2048, telemetryContext }; const response = await completeSummarization(model, context, requestOptions, streamFn, retry, callbacks); // Check if aborted or errored diff --git a/packages/coding-agent/src/core/compaction/compaction.ts b/packages/coding-agent/src/core/compaction/compaction.ts index d44b69e3e..ee30daee3 100644 --- a/packages/coding-agent/src/core/compaction/compaction.ts +++ b/packages/coding-agent/src/core/compaction/compaction.ts @@ -5,7 +5,15 @@ * and after compaction the session is reloaded. */ -import type { AgentMessage, StreamFn, ThinkingLevel } from "@earendil-works/diffex-agent-core"; +import { + type AgentMessage, + type AiTelemetrySpan, + NOOP_TELEMETRY_CONTEXT, + type StreamFn, + startAiSpan, + type TelemetryContext, + type ThinkingLevel, +} from "@earendil-works/diffex-agent-core"; import { contentText, type RetryCallbacks, @@ -550,8 +558,9 @@ function createSummarizationOptions( env: Record | undefined, signal: AbortSignal | undefined, thinkingLevel: ThinkingLevel | undefined, + telemetryContext: TelemetryContext | undefined, ): SimpleStreamOptions { - const options: SimpleStreamOptions = { maxTokens, signal, apiKey, headers, env }; + const options: SimpleStreamOptions = { maxTokens, signal, apiKey, headers, env, telemetryContext }; if (model.reasoning && thinkingLevel && thinkingLevel !== "off") { options.reasoning = thinkingLevel; } @@ -580,12 +589,50 @@ export async function completeSummarization( sessionId: uuidv7(), }; const produce = async (): Promise => - streamFn - ? (await streamFn(model, context, requestOptions)).result() - : completeSimple(model, context, requestOptions); + startAiSpan( + requestOptions.telemetryContext ?? NOOP_TELEMETRY_CONTEXT, + "diffex.ai.request", + { + "diffex.request.id": uuidv7(), + "diffex.ai.operation": "stream", + "diffex.ai.provider": model.provider, + "diffex.ai.model": model.id, + "diffex.ai.api": model.api, + "diffex.ai.streaming": true, + }, + async (span) => { + const telemetryOptions = { ...requestOptions, telemetryContext: span }; + const response = streamFn + ? await (await streamFn(model, context, telemetryOptions)).result() + : await completeSimple(model, context, telemetryOptions); + settleSummarizationRequestSpan(span, response); + return response; + }, + ); return retryAssistantCall(produce, retry, requestOptions.signal, callbacks); } +function settleSummarizationRequestSpan(span: AiTelemetrySpan<"diffex.ai.request">, message: AssistantMessage): void { + const stopReason = + message.stopReason === "pending" ? undefined : message.stopReason === "toolUse" ? "tool_use" : message.stopReason; + span.setAttributes({ + ...(message.responseModel === undefined ? {} : { "diffex.ai.response.model": message.responseModel }), + ...(message.responseId === undefined ? {} : { "diffex.ai.response.id": message.responseId }), + ...(stopReason === undefined ? {} : { "diffex.ai.response.stop_reason": stopReason }), + "diffex.ai.usage.input_tokens": message.usage.input, + "diffex.ai.usage.output_tokens": message.usage.output, + "diffex.ai.usage.cache_read_tokens": message.usage.cacheRead, + "diffex.ai.usage.cache_write_tokens": message.usage.cacheWrite, + ...(message.usage.reasoning === undefined ? {} : { "diffex.ai.usage.reasoning_tokens": message.usage.reasoning }), + "diffex.ai.usage.total_tokens": message.usage.totalTokens, + "diffex.ai.usage.cost": message.usage.cost.total, + ...(message.stopReason === "error" ? { "diffex.ai.error.type": "AssistantMessageError" } : {}), + }); + if (message.stopReason === "error" || message.stopReason === "aborted") { + span.setStatus({ status: "error" }); + } +} + /** * Generate a summary of the conversation using the LLM. * If previousSummary is provided, uses the update prompt to merge. @@ -604,6 +651,7 @@ export async function generateSummary( env?: Record, retry?: RetryPolicy, callbacks?: RetryCallbacks, + telemetryContext?: TelemetryContext, ): Promise { return ( await generateSummaryWithUsage( @@ -620,6 +668,7 @@ export async function generateSummary( env, retry, callbacks, + telemetryContext, ) ).text; } @@ -639,6 +688,7 @@ export async function generateSummaryWithUsage( env?: Record, retry?: RetryPolicy, callbacks?: RetryCallbacks, + telemetryContext?: TelemetryContext, ): Promise<{ text: string; usage: Usage }> { const maxTokens = Math.min( Math.floor(0.8 * reserveTokens), @@ -671,7 +721,16 @@ export async function generateSummaryWithUsage( }, ]; - const completionOptions = createSummarizationOptions(model, maxTokens, apiKey, headers, env, signal, thinkingLevel); + const completionOptions = createSummarizationOptions( + model, + maxTokens, + apiKey, + headers, + env, + signal, + thinkingLevel, + telemetryContext, + ); const response = await completeSummarization( model, @@ -832,6 +891,7 @@ export async function compact( env?: Record, retry?: RetryPolicy, callbacks?: RetryCallbacks, + telemetryContext?: TelemetryContext, ): Promise { const { firstKeptEntryId, @@ -866,6 +926,7 @@ export async function compact( env, retry, callbacks, + telemetryContext, ); historyText = historyResult.text; historyUsage = historyResult.usage; @@ -882,6 +943,7 @@ export async function compact( streamFn, retry, callbacks, + telemetryContext, ); // Merge into single summary summary = `${historyText}\n\n---\n\n**Turn Context (split turn):**\n\n${turnPrefixResult.text}`; @@ -902,6 +964,7 @@ export async function compact( env, retry, callbacks, + telemetryContext, ); summary = result.text; summaryUsage = result.usage; @@ -939,6 +1002,7 @@ async function generateTurnPrefixSummary( streamFn?: StreamFn, retry?: RetryPolicy, callbacks?: RetryCallbacks, + telemetryContext?: TelemetryContext, ): Promise<{ text: string; usage: Usage }> { const maxTokens = Math.min( Math.floor(0.5 * reserveTokens), @@ -958,7 +1022,7 @@ async function generateTurnPrefixSummary( const response = await completeSummarization( model, { systemPrompt: SUMMARIZATION_SYSTEM_PROMPT, messages: summarizationMessages }, - createSummarizationOptions(model, maxTokens, apiKey, headers, env, signal, thinkingLevel), + createSummarizationOptions(model, maxTokens, apiKey, headers, env, signal, thinkingLevel, telemetryContext), streamFn, retry, callbacks, diff --git a/packages/coding-agent/src/core/extensions/runner.ts b/packages/coding-agent/src/core/extensions/runner.ts index c04e56158..12fa67de3 100644 --- a/packages/coding-agent/src/core/extensions/runner.ts +++ b/packages/coding-agent/src/core/extensions/runner.ts @@ -2,7 +2,7 @@ * Extension runner - executes extensions and manages their lifecycle. */ -import type { AgentMessage } from "@earendil-works/diffex-agent-core"; +import { type AgentMessage, NOOP_TELEMETRY_CONTEXT, type TelemetryContext } from "@earendil-works/diffex-agent-core"; import type { ImageContent, Model, Provider, ProviderHeaders } from "@earendil-works/diffex-ai"; import type { KeyId } from "@earendil-works/diffex-tui"; import { type Theme, theme } from "../../modes/interactive/theme/theme.ts"; @@ -12,6 +12,7 @@ import type { ModelRegistry } from "../model-registry.ts"; import type { ScopedModel } from "../model-resolver.ts"; import type { SessionManager } from "../session-manager.ts"; import type { BuildSystemPromptOptions } from "../system-prompt.ts"; +import { startCodingAgentSpan } from "../telemetry.ts"; import type { BeforeAgentStartEvent, BeforeAgentStartEventResult, @@ -204,6 +205,7 @@ export async function emitProjectTrustEvent( extensionsResult: LoadExtensionsResult, event: ProjectTrustEvent, ctx: ProjectTrustContext, + telemetryContext: TelemetryContext = NOOP_TELEMETRY_CONTEXT, ): Promise<{ result?: ProjectTrustEventResult; errors: ExtensionError[] }> { const errors: ExtensionError[] = []; for (const ext of extensionsResult.extensions) { @@ -214,7 +216,25 @@ export async function emitProjectTrustEvent( for (const handler of handlers) { try { - const handlerResult = (await handler(event, ctx)) as ProjectTrustEventResult; + const handlerResult = (await startCodingAgentSpan( + telemetryContext, + "diffex.coding_agent.extension_hook", + { + "diffex.extension.hook": event.type, + "diffex.extension.path": ext.path, + }, + async (span) => { + try { + const result = await handler(event, ctx); + span.setAttributes({ "diffex.extension.outcome": "completed" }); + return result; + } catch (error) { + span.setAttributes({ "diffex.extension.outcome": "failed" }); + span.setStatus({ status: "error" }); + throw error; + } + }, + )) as ProjectTrustEventResult; if (handlerResult.trusted === "undecided") { continue; } @@ -295,6 +315,7 @@ export class ExtensionRunner { private shortcutDiagnostics: ResourceDiagnostic[] = []; private commandDiagnostics: ResourceDiagnostic[] = []; private staleMessage: string | undefined; + private readonly telemetryContext: TelemetryContext; constructor( extensions: Extension[], @@ -302,6 +323,7 @@ export class ExtensionRunner { cwd: string, sessionManager: SessionManager, modelRegistry: ModelRegistry, + telemetryContext: TelemetryContext = NOOP_TELEMETRY_CONTEXT, ) { this.extensions = extensions; this.runtime = runtime; @@ -309,6 +331,7 @@ export class ExtensionRunner { this.cwd = cwd; this.sessionManager = sessionManager; this.modelRegistry = modelRegistry; + this.telemetryContext = telemetryContext; } bindCore( @@ -798,6 +821,33 @@ export class ExtensionRunner { ); } + private invokeExtensionHook( + extensionPath: string, + hook: string, + callback: () => Result | Promise, + resolveOutcome?: (result: Result) => "completed" | "blocked", + ): Promise { + return startCodingAgentSpan( + this.telemetryContext, + "diffex.coding_agent.extension_hook", + { + "diffex.extension.hook": hook, + "diffex.extension.path": extensionPath, + }, + async (span) => { + try { + const result = await callback(); + span.setAttributes({ "diffex.extension.outcome": resolveOutcome?.(result) ?? "completed" }); + return result; + } catch (error) { + span.setAttributes({ "diffex.extension.outcome": "failed" }); + span.setStatus({ status: "error" }); + throw error; + } + }, + ); + } + async emit(event: TEvent): Promise> { const ctx = this.createContext(); let result: SessionBeforeEventResult | undefined; @@ -808,7 +858,15 @@ export class ExtensionRunner { for (const handler of handlers) { try { - const handlerResult = await handler(event, ctx); + const handlerResult = await this.invokeExtensionHook( + ext.path, + event.type, + () => handler(event, ctx), + (value) => + this.isSessionBeforeEvent(event) && (value as SessionBeforeEventResult | undefined)?.cancel + ? "blocked" + : "completed", + ); if (this.isSessionBeforeEvent(event) && handlerResult) { result = handlerResult as SessionBeforeEventResult; @@ -844,7 +902,9 @@ export class ExtensionRunner { for (const handler of handlers) { try { const currentEvent: MessageEndEvent = { ...event, message: currentMessage }; - const handlerResult = (await handler(currentEvent, ctx)) as MessageEndEventResult | undefined; + const handlerResult = (await this.invokeExtensionHook(ext.path, "message_end", () => + handler(currentEvent, ctx), + )) as MessageEndEventResult | undefined; if (!handlerResult?.message) continue; if (handlerResult.message.role !== currentMessage.role) { @@ -885,7 +945,9 @@ export class ExtensionRunner { for (const handler of handlers) { try { - const handlerResult = (await handler(currentEvent, ctx)) as ToolResultEventResult | undefined; + const handlerResult = (await this.invokeExtensionHook(ext.path, "tool_result", () => + handler(currentEvent, ctx), + )) as ToolResultEventResult | undefined; if (!handlerResult) continue; if (handlerResult.content !== undefined) { @@ -938,7 +1000,12 @@ export class ExtensionRunner { if (!handlers || handlers.length === 0) continue; for (const handler of handlers) { - const handlerResult = await handler(event, ctx); + const handlerResult = await this.invokeExtensionHook( + ext.path, + "tool_call", + () => handler(event, ctx), + (value) => ((value as ToolCallEventResult | undefined)?.block ? "blocked" : "completed"), + ); if (handlerResult) { result = handlerResult as ToolCallEventResult; @@ -961,7 +1028,7 @@ export class ExtensionRunner { for (const handler of handlers) { try { - const handlerResult = await handler(event, ctx); + const handlerResult = await this.invokeExtensionHook(ext.path, "user_bash", () => handler(event, ctx)); if (handlerResult) { return handlerResult as UserBashEventResult; } @@ -992,7 +1059,7 @@ export class ExtensionRunner { for (const handler of handlers) { try { const event: ContextEvent = { type: "context", messages: currentMessages }; - const handlerResult = await handler(event, ctx); + const handlerResult = await this.invokeExtensionHook(ext.path, "context", () => handler(event, ctx)); if (handlerResult && (handlerResult as ContextEventResult).messages) { currentMessages = (handlerResult as ContextEventResult).messages!; @@ -1027,7 +1094,9 @@ export class ExtensionRunner { type: "before_provider_request", payload: currentPayload, }; - const handlerResult = await handler(event, ctx); + const handlerResult = await this.invokeExtensionHook(ext.path, "before_provider_request", () => + handler(event, ctx), + ); if (handlerResult !== undefined) { currentPayload = handlerResult; } @@ -1061,7 +1130,7 @@ export class ExtensionRunner { type: "before_provider_headers", headers, }; - await handler(event, ctx); + await this.invokeExtensionHook(ext.path, "before_provider_headers", () => handler(event, ctx)); } catch (err) { const message = err instanceof Error ? err.message : String(err); const stack = err instanceof Error ? err.stack : undefined; @@ -1109,7 +1178,9 @@ export class ExtensionRunner { systemPrompt: currentSystemPrompt, systemPromptOptions, }; - const handlerResult = await handler(event, ctx); + const handlerResult = await this.invokeExtensionHook(ext.path, "before_agent_start", () => + handler(event, ctx), + ); if (handlerResult) { const result = handlerResult as BeforeAgentStartEventResult; @@ -1164,7 +1235,9 @@ export class ExtensionRunner { for (const handler of handlers) { try { const event: ResourcesDiscoverEvent = { type: "resources_discover", cwd, reason }; - const handlerResult = await handler(event, ctx); + const handlerResult = await this.invokeExtensionHook(ext.path, "resources_discover", () => + handler(event, ctx), + ); const result = handlerResult as ResourcesDiscoverResult | undefined; if (result?.skillPaths?.length) { @@ -1213,7 +1286,9 @@ export class ExtensionRunner { source, streamingBehavior, }; - const result = (await handler(event, ctx)) as InputEventResult | undefined; + const result = (await this.invokeExtensionHook(ext.path, "input", () => handler(event, ctx))) as + | InputEventResult + | undefined; if (result?.action === "handled") return result; if (result?.action === "transform") { currentText = result.text; diff --git a/packages/coding-agent/src/core/index.ts b/packages/coding-agent/src/core/index.ts index 924932e67..4517a6e2b 100644 --- a/packages/coding-agent/src/core/index.ts +++ b/packages/coding-agent/src/core/index.ts @@ -78,3 +78,12 @@ export { type WorkingIndicatorOptions, } from "./extensions/index.ts"; export { createSyntheticSourceInfo } from "./source-info.ts"; +export { + CODING_AGENT_TELEMETRY_SCHEMA, + type CodingAgentSpan, + type CodingAgentSpanEndAttributes, + type CodingAgentSpanName, + type CodingAgentSpanStartAttributes, + type CodingAgentTelemetrySpan, + startCodingAgentSpan, +} from "./telemetry.ts"; diff --git a/packages/coding-agent/src/core/project-trust.ts b/packages/coding-agent/src/core/project-trust.ts index 858e03cc1..a6fa9532f 100644 --- a/packages/coding-agent/src/core/project-trust.ts +++ b/packages/coding-agent/src/core/project-trust.ts @@ -1,3 +1,4 @@ +import type { TelemetryContext } from "@earendil-works/diffex-agent-core"; import { CONFIG_DIR_NAME } from "../config.ts"; import { emitProjectTrustEvent } from "./extensions/runner.ts"; import type { LoadExtensionsResult, ProjectTrustContext } from "./extensions/types.ts"; @@ -19,6 +20,7 @@ export interface ResolveProjectTrustedOptions { extensionsResult?: LoadExtensionsResult; projectTrustContext: ProjectTrustContext; onExtensionError?: (message: string) => void; + telemetryContext?: TelemetryContext; } function formatProjectTrustPrompt(cwd: string): string { @@ -56,6 +58,7 @@ export async function resolveProjectTrusted(options: ResolveProjectTrustedOption options.extensionsResult, { type: "project_trust", cwd: options.cwd }, options.projectTrustContext, + options.telemetryContext, ); for (const error of errors) { options.onExtensionError?.(`Extension "${error.extensionPath}" project_trust error: ${error.error}`); diff --git a/packages/coding-agent/src/core/sdk.ts b/packages/coding-agent/src/core/sdk.ts index 07593b147..c0a66ab80 100644 --- a/packages/coding-agent/src/core/sdk.ts +++ b/packages/coding-agent/src/core/sdk.ts @@ -1,5 +1,12 @@ import { join } from "node:path"; -import { Agent, type AgentMessage, setDefaultStreamFn, type ThinkingLevel } from "@earendil-works/diffex-agent-core"; +import { + Agent, + type AgentMessage, + NOOP_TELEMETRY_CONTEXT, + setDefaultStreamFn, + type TelemetryContext, + type ThinkingLevel, +} from "@earendil-works/diffex-agent-core"; import { clampThinkingLevel, type Message, type Model, streamSimple } from "@earendil-works/diffex-ai/compat"; import type { HarnessStateStore, HarnessVersionBundle } from "@earendil-works/diffex-harness-state"; import { getAgentDir } from "../config.ts"; @@ -102,6 +109,8 @@ export interface CreateAgentSessionOptions { harnessStateStore?: HarnessStateStore; /** Activated harness revision from the preloaded store. */ harnessVersionBundle?: HarnessVersionBundle; + /** Optional parent for passive runtime telemetry. */ + telemetryContext?: TelemetryContext; } /** Result from createAgentSession */ @@ -182,10 +191,15 @@ function assertNoReservedSubagentTools(customTools: readonly ToolDefinition[] | } } -function createRootSubagentController(maxConcurrent: number, agentDir: string): SubagentController { +function createRootSubagentController( + maxConcurrent: number, + agentDir: string, + telemetryContext: TelemetryContext, +): SubagentController { const sessionManagerFactory = new RootLocalEphemeralSubagentSessionManagerFactory(); return new SubagentController({ maxConcurrent, + telemetryContext, captureSession: (parent, childName, origin) => { const config = captureSubagentSessionConfig(parent, agentDir); return (signal) => @@ -241,9 +255,12 @@ export async function createAgentSession(options: CreateAgentSessionOptions = {} ? { store: options.harnessStateStore, bundle: options.harnessVersionBundle, warning: undefined } : await loadGlobalHarnessState(agentDir); const subagentOptions = normalizeSubagentOptions(options.subagents); + const telemetryContext = options.telemetryContext ?? NOOP_TELEMETRY_CONTEXT; assertNoReservedSubagentTools(options.customTools); const subagentController = - subagentOptions === false ? undefined : createRootSubagentController(subagentOptions.maxConcurrent, agentDir); + subagentOptions === false + ? undefined + : createRootSubagentController(subagentOptions.maxConcurrent, agentDir, telemetryContext); const customTools = [...(options.customTools ?? []), ...(subagentController?.createToolDefinitions() ?? [])]; let resourceLoader = options.resourceLoader; @@ -368,6 +385,7 @@ export async function createAgentSession(options: CreateAgentSessionOptions = {} const extensionRunnerRef: { current?: ExtensionRunner } = {}; agent = new Agent({ + telemetryContext, initialState: { systemPrompt: "", model, @@ -466,6 +484,7 @@ export async function createAgentSession(options: CreateAgentSessionOptions = {} subagentController, harnessStateStore: loadedHarnessState.store, harnessVersionBundle: loadedHarnessState.bundle, + telemetryContext, }); subagentController?.bindParent(session); const extensionsResult = resourceLoader.getExtensions(); diff --git a/packages/coding-agent/src/core/subagents/controller.ts b/packages/coding-agent/src/core/subagents/controller.ts index 28c6263e0..8058aaa78 100644 --- a/packages/coding-agent/src/core/subagents/controller.ts +++ b/packages/coding-agent/src/core/subagents/controller.ts @@ -1,6 +1,8 @@ +import { NOOP_TELEMETRY_CONTEXT, type TelemetryContext, uuidv7 } from "@earendil-works/diffex-agent-core"; import type { AssistantMessage } from "@earendil-works/diffex-ai"; import type { AgentSession } from "../agent-session.ts"; import type { ToolDefinition } from "../extensions/types.ts"; +import { startCodingAgentSpan } from "../telemetry.ts"; import { createAgentActionTool, createAgentsTool, createSpawnAgentTool } from "../tools/subagents.ts"; import type { CaptureSubagentSession, @@ -80,6 +82,7 @@ function getAssistantText(message: AssistantMessage | undefined): string | undef } interface SubagentRecord { + id: string; name: string; task: string; status: SubagentStatus; @@ -98,6 +101,7 @@ interface SubagentRecord { cleanupFailures?: unknown[]; coordinatedRun?: CoordinatedRun; lastRetryCycleId?: number; + telemetryContext?: TelemetryContext; } function isTerminalStatus(status: SubagentStatus): status is SubagentTerminalStatus { @@ -143,6 +147,7 @@ const MAX_TIMER_MS = 2_147_483_647; export class SubagentController { readonly #maxConcurrent: number; readonly #captureSession: CaptureSubagentSession | undefined; + readonly #telemetryContext: TelemetryContext; readonly #records = new Map(); readonly #listeners = new Set(); readonly #waiters = new Set(); @@ -159,12 +164,17 @@ export class SubagentController { #resetCleanupPromise: Promise | undefined; #shutdownPromise: Promise | undefined; - constructor(options: { maxConcurrent: number; captureSession?: CaptureSubagentSession }) { + constructor(options: { + maxConcurrent: number; + captureSession?: CaptureSubagentSession; + telemetryContext?: TelemetryContext; + }) { if (!Number.isInteger(options.maxConcurrent) || options.maxConcurrent <= 0) { throw new RangeError("maxConcurrent must be a positive integer"); } this.#maxConcurrent = options.maxConcurrent; this.#captureSession = options.captureSession; + this.#telemetryContext = options.telemetryContext ?? NOOP_TELEMETRY_CONTEXT; } createToolDefinitions(): ToolDefinition[] { @@ -176,7 +186,7 @@ export class SubagentController { if (!createSession) { throw new Error("Subagent session capture is unavailable"); } - this.spawn(name, task, createSession, origin); + this.spawn(name, task, createSession, origin, parent.agent.telemetryContext); }), createAgentActionTool((request) => { const parent = this.#getBoundParent(); @@ -217,6 +227,7 @@ export class SubagentController { const timestamp = Date.now(); const record: SubagentRecord = { + id: uuidv7(), name, task, status: "queued", @@ -233,13 +244,32 @@ export class SubagentController { })); } - spawn(name: string, task: string, createSession: CreateSubagentSession, origin?: SubagentSpawnOrigin): void { + spawn( + name: string, + task: string, + createSession: CreateSubagentSession, + origin?: SubagentSpawnOrigin, + telemetryContext: TelemetryContext = this.#telemetryContext, + ): void { this.reserve(name, task); const record = this.#records.get(name)!; record.createSession = createSession; + record.telemetryContext = telemetryContext; if (origin?.type === "agent_tool_call") { record.coordinatedRun = this.#createCoordinatedRun("initial"); } + void startCodingAgentSpan( + telemetryContext, + "diffex.coding_agent.subagent", + { + "diffex.subagent.id": record.id, + "diffex.subagent.name": record.name, + "diffex.subagent.action": "spawn", + }, + (span) => { + span.setAttributes({ "diffex.subagent.outcome": "queued" }); + }, + ); this.#scheduleRun(record, task); } @@ -247,11 +277,13 @@ export class SubagentController { if (message.trim().length === 0) { return Promise.reject(new Error("Subagent message must be non-empty")); } - return this.#serializeAction(name, () => this.#message(name, message, wake, coordinated)); + return this.#serializeAction(name, () => + this.#traceAction(name, "message", () => this.#message(name, message, wake, coordinated)), + ); } interrupt(name: string): Promise { - return this.#serializeAction(name, () => this.#interrupt(name)); + return this.#serializeAction(name, () => this.#traceAction(name, "interrupt", () => this.#interrupt(name))); } attachSession(name: string, session: AgentSession): void { @@ -321,6 +353,32 @@ export class SubagentController { } async waitForBatch(batch: SubagentDeliveryBatch, options: SubagentBatchWaitOptions = {}): Promise { + const batchTelemetryContext = batch.attempts + .map((attempt) => this.#records.get(attempt.name)?.telemetryContext) + .find((context) => context !== undefined); + return startCodingAgentSpan( + batchTelemetryContext ?? this.#telemetryContext, + "diffex.coding_agent.subagent", + { + "diffex.subagent.id": `batch:${batch.id}`, + "diffex.subagent.name": "batch", + "diffex.subagent.action": "wait", + }, + async (span) => { + try { + await this.#waitForBatch(batch, options); + span.setAttributes({ "diffex.subagent.outcome": "completed" }); + } catch (error) { + const interrupted = options.signal?.aborted === true; + span.setAttributes({ "diffex.subagent.outcome": interrupted ? "interrupted" : "failed" }); + span.setStatus({ status: "error" }); + throw error; + } + }, + ); + } + + async #waitForBatch(batch: SubagentDeliveryBatch, options: SubagentBatchWaitOptions): Promise { const notified = new Set(); let snapshot = this.snapshot(); while (true) { @@ -867,6 +925,43 @@ export class SubagentController { return result; } + #traceAction( + name: string, + action: "message" | "interrupt", + callback: () => Promise, + ): Promise { + const record = this.#getRecordForAction(name); + return startCodingAgentSpan( + record.telemetryContext ?? this.#telemetryContext, + "diffex.coding_agent.subagent", + { + "diffex.subagent.id": record.id, + "diffex.subagent.name": record.name, + "diffex.subagent.action": action, + }, + async (span) => { + try { + const result = await callback(); + span.setAttributes({ + "diffex.subagent.outcome": + result.status === "failed" + ? "failed" + : result.status === "interrupted" + ? "interrupted" + : result.status === "queued" + ? "queued" + : "completed", + }); + return result; + } catch (error) { + span.setAttributes({ "diffex.subagent.outcome": "failed" }); + span.setStatus({ status: "error" }); + throw error; + } + }, + ); + } + async #message(name: string, message: string, wake: boolean, coordinated: boolean): Promise { const record = this.#getRecordForAction(name); while (record.status === "running" && record.session?.isIdle) { @@ -960,6 +1055,34 @@ export class SubagentController { } async #runPrompt(record: SubagentRecord, prompt: string, runController: AbortController): Promise { + const action = record.coordinatedRun?.attempt === "retry" ? "retry" : "run"; + await startCodingAgentSpan( + record.telemetryContext ?? this.#telemetryContext, + "diffex.coding_agent.subagent", + { + "diffex.subagent.id": record.id, + "diffex.subagent.name": record.name, + "diffex.subagent.action": action, + }, + async (span) => { + await this.#runPromptInner(record, prompt, runController); + const outcome = + record.status === "failed" + ? "failed" + : record.status === "interrupted" + ? "interrupted" + : record.status === "queued" + ? "queued" + : "completed"; + span.setAttributes({ "diffex.subagent.outcome": outcome }); + if (outcome === "failed" || outcome === "interrupted") { + span.setStatus({ status: "error" }); + } + }, + ); + } + + async #runPromptInner(record: SubagentRecord, prompt: string, runController: AbortController): Promise { let permit: SubagentPermit | undefined; let session = record.session; const assistantMessages: AssistantMessage[] = []; diff --git a/packages/coding-agent/src/core/subagents/services.ts b/packages/coding-agent/src/core/subagents/services.ts index 1a5355895..2a6a64ac4 100644 --- a/packages/coding-agent/src/core/subagents/services.ts +++ b/packages/coding-agent/src/core/subagents/services.ts @@ -1,5 +1,6 @@ -import type { ThinkingLevel } from "@earendil-works/diffex-agent-core"; +import type { TelemetryContext, ThinkingLevel } from "@earendil-works/diffex-agent-core"; import type { Model } from "@earendil-works/diffex-ai"; +import type { HarnessStateStore, HarnessVersionBundle } from "@earendil-works/diffex-harness-state"; import type { AgentSession } from "../agent-session.ts"; import { type AgentSessionServices, @@ -21,6 +22,9 @@ export interface CreateSubagentServicesOptions { agentDir: string; modelRuntime: ModelRuntime; settingsManager: SettingsManager; + harnessStateStore?: HarnessStateStore; + harnessVersionBundle?: HarnessVersionBundle; + telemetryContext?: TelemetryContext; } export interface SubagentSessionConfig { @@ -33,6 +37,9 @@ export interface SubagentSessionConfig { readonly tools: readonly ToolName[]; readonly parentSession: string; readonly sourceSessionManager: SessionManager; + readonly harnessStateStore?: HarnessStateStore; + readonly harnessVersionBundle?: HarnessVersionBundle; + readonly telemetryContext?: TelemetryContext; } export interface CreateSubagentSessionOptions { @@ -56,6 +63,9 @@ export function createSubagentServices(options: CreateSubagentServicesOptions): agentDir: options.agentDir, modelRuntime: options.modelRuntime, settingsManager: options.settingsManager, + harnessStateStore: options.harnessStateStore, + harnessVersionBundle: options.harnessVersionBundle, + telemetryContext: options.telemetryContext, resourceLoaderOptions: { noExtensions: true, noPromptTemplates: true, @@ -92,6 +102,9 @@ export function captureSubagentSessionConfig(parent: AgentSession, agentDir: str tools, parentSession: parent.sessionId, sourceSessionManager: parent.sessionManager, + harnessStateStore: parent.harnessStateStore, + harnessVersionBundle: parent.harnessVersionBundle, + telemetryContext: parent.agent.telemetryContext, }); } @@ -105,6 +118,9 @@ export async function createSubagentSession( agentDir: config.agentDir, modelRuntime: config.modelRuntime, settingsManager: config.settingsManager, + harnessStateStore: config.harnessStateStore, + harnessVersionBundle: config.harnessVersionBundle, + telemetryContext: config.telemetryContext, }); options.signal?.throwIfAborted(); diff --git a/packages/coding-agent/src/core/telemetry.ts b/packages/coding-agent/src/core/telemetry.ts index 14442104d..2c79c4405 100644 --- a/packages/coding-agent/src/core/telemetry.ts +++ b/packages/coding-agent/src/core/telemetry.ts @@ -1,3 +1,13 @@ +import type { + ExactTelemetryAttributes, + SchemaTelemetrySpan, + TelemetryContext, + TelemetrySchemaDefinition, + TelemetrySchemaSpanEndAttributes, + TelemetrySchemaSpanName, + TelemetrySchemaSpanStartAttributes, + TelemetrySchemaSpanUnion, +} from "@earendil-works/diffex-agent-core"; import type { SettingsManager } from "./settings-manager.ts"; function isTruthyEnvFlag(value: string | undefined): boolean { @@ -11,3 +21,221 @@ export function isInstallTelemetryEnabled( ): boolean { return telemetryEnv !== undefined ? isTruthyEnvFlag(telemetryEnv) : settingsManager.getEnableInstallTelemetry(); } + +export const CODING_AGENT_TELEMETRY_SCHEMA = { + version: 1, + spans: { + "diffex.coding_agent.operation": { + description: "One user-visible coding-agent session operation", + parents: { kind: "any" }, + startAttributes: { + "diffex.operation.id": { + type: "string", + required: true, + cardinality: "high", + description: "Operation-local stable id", + }, + "diffex.session.id": { + type: "string", + required: true, + cardinality: "high", + description: "Owning session id", + }, + "diffex.operation.kind": { + type: "string", + required: true, + values: ["prompt", "compaction", "navigation", "abort", "queue", "bash", "export"], + description: "Session operation kind", + }, + }, + endAttributes: { + "diffex.operation.outcome": { + type: "string", + values: ["completed", "queued", "handled", "declined", "failed", "aborted"], + description: "Terminal operation outcome", + }, + }, + status: { default: "ok", errorWhen: "The operation fails or is aborted" }, + }, + "diffex.coding_agent.retry": { + description: "One coding-agent retry decision and delay", + parents: { kind: "any" }, + startAttributes: { + "diffex.session.id": { + type: "string", + required: true, + cardinality: "high", + description: "Owning session id", + }, + "diffex.retry.kind": { + type: "string", + required: true, + values: ["provider", "compaction", "branch_summary"], + description: "Retrying operation kind", + }, + "diffex.retry.attempt": { + type: "number", + required: true, + description: "One-based retry attempt", + }, + "diffex.retry.max_attempts": { + type: "number", + required: false, + description: "Configured retry limit", + }, + "diffex.retry.delay_ms": { + type: "number", + required: false, + description: "Scheduled delay before retry", + }, + }, + endAttributes: { + "diffex.retry.outcome": { + type: "string", + values: ["scheduled", "resumed", "cancelled", "failed"], + description: "Retry transition outcome", + }, + }, + status: { default: "ok", errorWhen: "The retry transition fails" }, + }, + "diffex.coding_agent.extension_hook": { + description: "One extension hook invocation", + parents: { kind: "any" }, + startAttributes: { + "diffex.extension.hook": { + type: "string", + required: true, + description: "Extension hook name", + }, + "diffex.extension.path": { + type: "string", + required: true, + cardinality: "high", + sensitive: true, + description: "Loaded extension path", + }, + }, + endAttributes: { + "diffex.extension.outcome": { + type: "string", + values: ["completed", "blocked", "failed"], + description: "Terminal hook outcome", + }, + }, + status: { default: "ok", errorWhen: "The extension hook throws" }, + }, + "diffex.coding_agent.subagent": { + description: "One child-agent lifecycle action", + parents: { kind: "any" }, + startAttributes: { + "diffex.subagent.id": { + type: "string", + required: true, + cardinality: "high", + description: "Stable child-agent id", + }, + "diffex.subagent.name": { + type: "string", + required: true, + cardinality: "high", + description: "Caller-assigned child-agent name", + }, + "diffex.subagent.action": { + type: "string", + required: true, + values: ["spawn", "run", "wait", "retry", "interrupt", "message"], + description: "Child-agent lifecycle action", + }, + }, + endAttributes: { + "diffex.subagent.outcome": { + type: "string", + values: ["completed", "failed", "interrupted", "timed_out", "queued"], + description: "Terminal child-agent action outcome", + }, + }, + status: { default: "ok", errorWhen: "The child-agent action fails or is interrupted" }, + }, + "diffex.coding_agent.terminal": { + description: "One user-invoked terminal execution", + parents: { kind: "any" }, + startAttributes: { + "diffex.session.id": { + type: "string", + required: true, + cardinality: "high", + description: "Owning session id", + }, + }, + endAttributes: { + "diffex.terminal.exit_code": { type: "number", description: "Process exit code" }, + "diffex.terminal.outcome": { + type: "string", + values: ["completed", "failed", "cancelled"], + description: "Terminal execution outcome", + }, + }, + status: { default: "ok", errorWhen: "The terminal execution fails or is cancelled" }, + }, + "diffex.coding_agent.artifact": { + description: "One session artifact written to disk", + parents: { kind: "any" }, + startAttributes: { + "diffex.session.id": { + type: "string", + required: true, + cardinality: "high", + description: "Owning session id", + }, + "diffex.artifact.kind": { + type: "string", + required: true, + values: ["html", "jsonl"], + description: "Written artifact format", + }, + }, + endAttributes: { + "diffex.artifact.path": { + type: "string", + cardinality: "high", + sensitive: true, + description: "Resolved artifact path", + }, + "diffex.artifact.outcome": { + type: "string", + values: ["written", "failed"], + description: "Artifact write outcome", + }, + }, + status: { default: "ok", errorWhen: "The artifact write fails" }, + }, + }, +} as const satisfies TelemetrySchemaDefinition; + +export type CodingAgentSpanName = TelemetrySchemaSpanName; +export type CodingAgentSpanStartAttributes = TelemetrySchemaSpanStartAttributes< + typeof CODING_AGENT_TELEMETRY_SCHEMA, + Name +>; +export type CodingAgentSpanEndAttributes = TelemetrySchemaSpanEndAttributes< + typeof CODING_AGENT_TELEMETRY_SCHEMA, + Name +>; +export type CodingAgentTelemetrySpan = SchemaTelemetrySpan< + typeof CODING_AGENT_TELEMETRY_SCHEMA, + Name +>; +export type CodingAgentSpan = TelemetrySchemaSpanUnion; + +export function startCodingAgentSpan< + Name extends CodingAgentSpanName, + const Attributes extends CodingAgentSpanStartAttributes, + Result, +>( + telemetryContext: TelemetryContext, + name: Name, + attributes: ExactTelemetryAttributes, Attributes>, + callback: (span: CodingAgentTelemetrySpan) => Result | Promise, +): Promise { + return telemetryContext.startSpan({ name, attributes }, (span) => callback(span as CodingAgentTelemetrySpan)); +} diff --git a/packages/coding-agent/src/index.ts b/packages/coding-agent/src/index.ts index 858c940bc..645ddb15a 100644 --- a/packages/coding-agent/src/index.ts +++ b/packages/coding-agent/src/index.ts @@ -276,6 +276,15 @@ export { } from "./core/skills.ts"; export { createSyntheticSourceInfo } from "./core/source-info.ts"; export type { SubagentStatus, SubagentSummary } from "./core/subagents/types.ts"; +export { + CODING_AGENT_TELEMETRY_SCHEMA, + type CodingAgentSpan, + type CodingAgentSpanEndAttributes, + type CodingAgentSpanName, + type CodingAgentSpanStartAttributes, + type CodingAgentTelemetrySpan, + startCodingAgentSpan, +} from "./core/telemetry.ts"; export { type EditDiffResult, generateDiffString, generateUnifiedPatch } from "./core/tools/edit-diff.ts"; // Tools export { diff --git a/packages/coding-agent/src/main.ts b/packages/coding-agent/src/main.ts index 38621e22b..d571fd12d 100644 --- a/packages/coding-agent/src/main.ts +++ b/packages/coding-agent/src/main.ts @@ -718,6 +718,7 @@ export async function main(args: string[], options?: MainOptions) { sessionManager, sessionStartEvent, projectTrustContext, + telemetryContext, }) => { const isInitialRuntime = sessionStartEvent === undefined; const projectTrustDiagnostics: AgentSessionRuntimeDiagnostic[] = []; @@ -736,6 +737,7 @@ export async function main(args: string[], options?: MainOptions) { agentDir, settingsManager: runtimeSettingsManager, modelRuntimeSignal: AbortSignal.timeout(15_000), + telemetryContext, extensionFlagValues: parsed.unknownFlags, resourceLoaderReloadOptions: shouldResolveProjectTrust ? { @@ -755,6 +757,7 @@ export async function main(args: string[], options?: MainOptions) { hasUI: isInitialRuntime && trustPromptMode === "interactive", }), onExtensionError: (message) => projectTrustDiagnostics.push({ type: "warning", message }), + telemetryContext, }); projectTrustByCwd.set(cwd, trusted); return trusted; diff --git a/packages/coding-agent/test/suite/harness.ts b/packages/coding-agent/test/suite/harness.ts index 4596f1ce8..2523ae3dd 100644 --- a/packages/coding-agent/test/suite/harness.ts +++ b/packages/coding-agent/test/suite/harness.ts @@ -6,7 +6,7 @@ import { createInMemoryModelRegistry, createModelRegistry, getModelRuntime } fro import { existsSync, mkdirSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import type { AgentMessage, AgentTool } from "@earendil-works/diffex-agent-core"; +import type { AgentMessage, AgentTool, TelemetryContext } from "@earendil-works/diffex-agent-core"; import { Agent } from "@earendil-works/diffex-agent-core"; import type { FauxModelDefinition, @@ -77,6 +77,7 @@ export interface HarnessOptions { sessionStartEvent?: SessionStartEvent; harnessStateStore?: HarnessStateStore; harnessVersionBundle?: HarnessVersionBundle; + telemetryContext?: TelemetryContext; } export interface Harness { @@ -146,6 +147,7 @@ export async function createHarness(options: HarnessOptions = {}): Promise (withConfiguredAuth ? "faux-key" : undefined), streamFn: streamSimple, initialState: { @@ -199,6 +201,7 @@ export async function createHarness(options: HarnessOptions = {}): Promise { + while (true) { + const snapshot = controller.snapshot(); + if (snapshot.agents.find((agent) => agent.name === name)?.status === status) return; + const change = await controller.waitForChange(snapshot.revision, { timeoutMs: 1_000 }); + if (change.reason !== "changed") throw new Error(`Timed out waiting for ${name} to become ${status}`); + } +} + +describe("coding-agent runtime telemetry", () => { + const harnesses: Harness[] = []; + const controllers: SubagentController[] = []; + + afterEach(async () => { + while (controllers.length > 0) await controllers.pop()?.shutdown(); + while (harnesses.length > 0) await harnesses.pop()?.cleanup(); + }); + + it("nests retried agent runs under one prompt operation", async () => { + const telemetry = new InMemoryTelemetryContext(); + const harness = await createHarness({ + telemetryContext: telemetry, + settings: { retry: { enabled: true, maxRetries: 2, baseDelayMs: 1 } }, + extensionFactories: [ + (pi) => { + pi.on("before_agent_start", async () => ({})); + }, + ], + }); + harnesses.push(harness); + harness.setResponses([ + fauxAssistantMessage("", { stopReason: "error", errorMessage: "overloaded_error" }), + fauxAssistantMessage("recovered"), + ]); + + await harness.session.prompt("retry once"); + + const spans = telemetry.getSpans(); + const operation = spans.find( + (span) => + span.name === "diffex.coding_agent.operation" && span.attributes["diffex.operation.kind"] === "prompt", + ); + expect(operation).toMatchObject({ + parentId: null, + attributes: { "diffex.operation.outcome": "completed" }, + }); + expect(operation?.attributes["diffex.operation.id"]).toEqual(expect.any(String)); + expect(spans.filter((span) => span.name === "diffex.agent.run").map((span) => span.parentId)).toEqual([ + operation?.id, + operation?.id, + ]); + expect(spans.find((span) => span.name === "diffex.coding_agent.retry")).toMatchObject({ + parentId: operation?.id, + attributes: { + "diffex.retry.attempt": 1, + "diffex.retry.outcome": "resumed", + }, + }); + expect(spans.find((span) => span.name === "diffex.coding_agent.extension_hook")).toMatchObject({ + attributes: { + "diffex.extension.hook": "before_agent_start", + "diffex.extension.outcome": "completed", + }, + }); + }); + + it("records provider-backed compaction at dispatch", async () => { + const telemetry = new InMemoryTelemetryContext(); + const harness = await createHarness({ + telemetryContext: telemetry, + settings: { compaction: { keepRecentTokens: 1 } }, + }); + harnesses.push(harness); + harness.setResponses([ + fauxAssistantMessage("one"), + fauxAssistantMessage("two"), + fauxAssistantMessage("captured summary"), + fauxAssistantMessage("captured turn prefix"), + ]); + await harness.session.prompt("first"); + await harness.session.prompt("second"); + + await harness.session.compact(); + + const spans = telemetry.getSpans(); + const operation = spans.find( + (span) => + span.name === "diffex.coding_agent.operation" && span.attributes["diffex.operation.kind"] === "compaction", + ); + expect(operation).toMatchObject({ attributes: { "diffex.operation.outcome": "completed" } }); + expect(spans.find((span) => span.name === "diffex.ai.request" && span.parentId === operation?.id)).toMatchObject({ + attributes: { + "diffex.ai.response.stop_reason": "stop", + "diffex.ai.streaming": true, + }, + }); + }); + + it("marks an interrupted prompt and agent run as aborted", async () => { + const telemetry = new InMemoryTelemetryContext(); + const harness = await createHarness({ telemetryContext: telemetry }); + harnesses.push(harness); + harness.setResponses([fauxAssistantMessage("x".repeat(20_000))]); + const sawUpdate = new Promise((resolve) => { + const unsubscribe = harness.session.subscribe((event) => { + if (event.type !== "message_update") return; + unsubscribe(); + resolve(); + }); + }); + + const prompt = harness.session.prompt("interrupt me"); + await sawUpdate; + await harness.session.abort(); + await prompt; + + const spans = telemetry.getSpans(); + expect( + spans.find( + (span) => + span.name === "diffex.coding_agent.operation" && span.attributes["diffex.operation.kind"] === "prompt", + ), + ).toMatchObject({ + status: { status: "error" }, + attributes: { "diffex.operation.outcome": "aborted" }, + }); + expect(spans.find((span) => span.name === "diffex.agent.run")).toMatchObject({ + status: { status: "error" }, + attributes: { "diffex.agent.outcome": "aborted" }, + }); + }); + + it("records a completed child-agent run with a stable id", async () => { + const telemetry = new InMemoryTelemetryContext(); + const child = await createHarness({ telemetryContext: telemetry }); + harnesses.push(child); + child.setResponses([fauxAssistantMessage("child complete")]); + const controller = new SubagentController({ maxConcurrent: 1, telemetryContext: telemetry }); + controllers.push(controller); + + controller.spawn("worker", "focused task", async () => child.session); + await waitForStatus(controller, "worker", "idle"); + + const run = telemetry + .getSpans() + .find( + (span) => + span.name === "diffex.coding_agent.subagent" && span.attributes["diffex.subagent.action"] === "run", + ); + expect(run).toMatchObject({ + attributes: { + "diffex.subagent.name": "worker", + "diffex.subagent.outcome": "completed", + }, + }); + expect(run?.attributes["diffex.subagent.id"]).toEqual(expect.any(String)); + }); + + it("captures the parent harness snapshot and telemetry context for child sessions", async () => { + const telemetry = new InMemoryTelemetryContext(); + const bootstrap = await createHarness(); + harnesses.push(bootstrap); + const store = new HarnessStateStore(bootstrap.tempDir); + const version = `sha256:${"1".repeat(64)}`; + const bundle: HarnessVersionBundle = { + state: { memory: "captured memory" }, + manifest: { schemaVersion: 1, version, base: null, artifacts: [] }, + }; + const harness = await createHarness({ + telemetryContext: telemetry, + harnessStateStore: store, + harnessVersionBundle: bundle, + }); + harnesses.push(harness); + + const captured = captureSubagentSessionConfig(harness.session, harness.tempDir); + + expect(captured.harnessStateStore).toBe(store); + expect(captured.harnessVersionBundle).toBe(bundle); + expect(captured.telemetryContext).toBe(telemetry); + }); + + it("records terminal and artifact facts without command or content payloads", async () => { + const telemetry = new InMemoryTelemetryContext(); + const harness = await createHarness({ telemetryContext: telemetry }); + harnesses.push(harness); + const operations: BashOperations = { + exec: async () => ({ exitCode: 0 }), + }; + + await harness.session.executeBash("private command", undefined, { operations }); + harness.session.exportToJsonl(`${harness.tempDir}/session-export.jsonl`); + await Promise.resolve(); + + const spans = telemetry.getSpans(); + expect(spans.find((span) => span.name === "diffex.coding_agent.terminal")).toMatchObject({ + attributes: { + "diffex.terminal.exit_code": 0, + "diffex.terminal.outcome": "completed", + }, + }); + expect(spans.find((span) => span.name === "diffex.coding_agent.artifact")).toMatchObject({ + attributes: { + "diffex.artifact.kind": "jsonl", + "diffex.artifact.outcome": "written", + }, + }); + expect(JSON.stringify(spans)).not.toContain("private command"); + }); +});