From a9cfc95e4f35fc7381d3d02b68072d13f55a2ac8 Mon Sep 17 00:00:00 2001 From: Alex Hawat <1254687+alexhawat@users.noreply.github.com> Date: Wed, 23 Sep 2026 15:30:26 +0200 Subject: [PATCH 1/2] feat: add OpenCode V2 support Add V2 support alongside V1 in one package: V1 keeps the named OtelPlugin export; V2 uses a new default export (id devtheops.otel, setup) built on the V2 granular event stream (session.step.*, session.tool.*, session.usage.*, session.execution.*, session.retry.scheduled). Shared changes: - config.ts: additive redactSecrets/redactValues options - headers.ts: use node:child_process instead of the Bun global Refs #128 --- README.md | 18 ++ src/config.ts | 24 ++ src/headers.ts | 37 +-- src/index.ts | 4 + src/redact.ts | 37 +++ src/v2/handlers/chat-headers.ts | 28 ++ src/v2/handlers/session.ts | 349 +++++++++++++++++++++++++ src/v2/handlers/tool.ts | 104 ++++++++ src/v2/handlers/usage.ts | 202 +++++++++++++++ src/v2/index.ts | 443 ++++++++++++++++++++++++++++++++ src/v2/types.ts | 110 ++++++++ src/v2/util.ts | 129 ++++++++++ src/v2/v2.ts | 40 +++ tests/redact.test.ts | 34 +++ 14 files changed, 1542 insertions(+), 17 deletions(-) create mode 100644 src/redact.ts create mode 100644 src/v2/handlers/chat-headers.ts create mode 100644 src/v2/handlers/session.ts create mode 100644 src/v2/handlers/tool.ts create mode 100644 src/v2/handlers/usage.ts create mode 100644 src/v2/index.ts create mode 100644 src/v2/types.ts create mode 100644 src/v2/util.ts create mode 100644 src/v2/v2.ts create mode 100644 tests/redact.test.ts diff --git a/README.md b/README.md index 36ae0a7..3af37f2 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,7 @@ An [opencode](https://opencode.ai) plugin that exports telemetry via OpenTelemetry (OTLP over gRPC or HTTP/protobuf), mirroring the same signals as [Claude Code's monitoring](https://code.claude.com/docs/en/monitoring-usage). +- [OpenCode V2 support](#opencode-v2-support) - [What it instruments](#what-it-instruments) - [Metrics](#metrics) - [Log events](#log-events) @@ -29,6 +30,23 @@ An [opencode](https://opencode.ai) plugin that exports telemetry via OpenTelemet - [Local development](#local-development) - [GitHub Discord notifications](#github-discord-notifications) +## OpenCode V2 support + +This plugin supports **OpenCode V1 and V2 from one package**: + +- **V1** uses the named `OtelPlugin` export. +- **V2** uses the default export (`id: devtheops.otel`, `setup()`), which reads V2's granular + event stream — `session.step.*`, `session.tool.*`, `session.usage.*`, `session.execution.*`, + and `session.retry.scheduled`. + +V2 support covers session, LLM-step and tool spans, token/cost/cache metrics, the retry +counter, execution-failure handling, and `model.request` trace-context injection. The V1 +handlers for `message.updated`, `message.part.updated`, `permission.*`, `command.executed`, +and `session.diff` have no V2 equivalent and are not ported. + +V2 additionally supports optional prompt capture (`capturePromptInLogs`) and best-effort +secret redaction (`redactSecrets`, `redactValues`) — see the plugin options below. + ## What it instruments ### Metrics diff --git a/src/config.ts b/src/config.ts index 6c235d5..ea322af 100644 --- a/src/config.ts +++ b/src/config.ts @@ -29,6 +29,8 @@ export type PluginConfig = { disabledMetrics: Set disabledTraces: Set tracePropagationProviders: Set + redactSecrets: boolean + redactValues: string[] } export function parseAttributePairs(raw: string | undefined): Record { @@ -73,6 +75,8 @@ export type OtelPluginOptions = { disabledMetrics?: string[] disabledTraces?: string[] tracePropagationProviders?: string[] + redactSecrets?: boolean + redactValues?: string[] } const VALID_PROTOCOLS = new Set(["grpc", "http/protobuf", "http/json"]) @@ -212,9 +216,29 @@ export function loadConfig(options: OtelPluginOptions = {}): PluginConfig { disabledMetrics, disabledTraces, tracePropagationProviders, + redactSecrets: pickBoolean(resolvedOptions.redactSecrets) ?? !hasNonEmptyEnv("OPENCODE_NO_REDACT"), + redactValues: collectRedactValues(resolvedOptions.redactValues), } } +/** + * Exact values to mask verbatim: any configured `redactValues` plus the values of + * secret-looking environment variables (e.g. `LOGFIRE_TOKEN`, `*_API_KEY`). Short values + * are ignored so common words are not over-redacted. + */ +function collectRedactValues(configured: string[] | undefined): string[] { + const values = new Set() + for (const value of configured ?? []) { + if (typeof value === "string" && value.length >= 6) values.add(value) + } + const secretEnv = /(TOKEN|SECRET|PASSWORD|PASSWD|API_?KEY|ACCESS_?KEY|PRIVATE_?KEY|CLIENT_?SECRET|CREDENTIAL)/i + for (const [key, value] of Object.entries(process.env)) { + if (!value || value.length < 6) continue + if (secretEnv.test(key)) values.add(value) + } + return [...values] +} + export function resolveHelperPath( helper: string | undefined, directory: string | undefined, diff --git a/src/headers.ts b/src/headers.ts index 1a4c3ff..58d6d0f 100644 --- a/src/headers.ts +++ b/src/headers.ts @@ -1,4 +1,5 @@ import { createRequire } from "module" +import { execFile } from "node:child_process" import { ExportResultCode, type ExportResult } from "@opentelemetry/core" import type { PushMetricExporter, ResourceMetrics } from "@opentelemetry/sdk-metrics" import type { SpanExporter, ReadableSpan } from "@opentelemetry/sdk-trace-base" @@ -89,24 +90,26 @@ export class DynamicHeaders { } private async runHelper(): Promise { - const proc = Bun.spawn([this.helper!], { - stdout: "pipe", - stderr: "pipe", - timeout: this.helperTimeoutMs, - killSignal: "SIGTERM", + const stdout = await new Promise((resolve, reject) => { + execFile( + this.helper!, + [], + { timeout: this.helperTimeoutMs, killSignal: "SIGTERM", maxBuffer: 1024 * 1024 }, + (error, out, err) => { + if (error) { + const signal = (error as NodeJS.ErrnoException & { signal?: string | null }).signal + if (signal) { + reject(new Error(`OTLP headers helper was terminated by ${signal}`)) + return + } + const detail = (err ?? "").trim() || error.message + reject(new Error(`OTLP headers helper failed: ${detail}`)) + return + } + resolve(out) + }, + ) }) - const [stdout, stderr, exitCode] = await Promise.all([ - new Response(proc.stdout).text(), - new Response(proc.stderr).text(), - proc.exited, - ]) - if (proc.signalCode) { - throw new Error(`OTLP headers helper was terminated by ${proc.signalCode}`) - } - if (exitCode !== 0) { - const detail = stderr.trim() || `exit code ${exitCode}` - throw new Error(`OTLP headers helper failed: ${detail}`) - } const parsed = JSON.parse(stdout) as unknown if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) { throw new Error("OTLP headers helper must return a JSON object") diff --git a/src/index.ts b/src/index.ts index 194270d..c713a3f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -364,3 +364,7 @@ export const OtelPlugin: Plugin = async ({ project, client, directory, worktree }), } } + +// OpenCode V2 entrypoint. V2 reads the default export's `id` and `setup()`; V1 keeps using +// the named `OtelPlugin` export above. See src/v2/index.ts. +export { default } from "./v2/index.ts" diff --git a/src/redact.ts b/src/redact.ts new file mode 100644 index 0000000..13ae01e --- /dev/null +++ b/src/redact.ts @@ -0,0 +1,37 @@ +// Masks credential-shaped substrings in captured text. Intentionally conservative: +// it targets known token formats and secret-looking key/value pairs, and leaves normal +// prompt/tool text untouched. + +const REPLACEMENT = "[REDACTED]" + +/** `[pattern, replacement]` pairs applied in order. */ +const RULES: ReadonlyArray = [ + // Authorization headers / bearer tokens. + [/(authorization\s*[:=]\s*)(?:bearer\s+)?[^\s"',;]+/gi, `$1${REPLACEMENT}`], + [/\bbearer\s+[A-Za-z0-9._\-]+/gi, `Bearer ${REPLACEMENT}`], + // Known token prefixes. + [/\bsk-[A-Za-z0-9_\-]{16,}\b/g, REPLACEMENT], // OpenAI-style + [/\bpylf_v\d+_[A-Za-z0-9._\-]+/g, REPLACEMENT], // Logfire + [/\bgh[pousr]_[A-Za-z0-9]{20,}\b/g, REPLACEMENT], // GitHub + [/\bgithub_pat_[A-Za-z0-9_]{20,}\b/g, REPLACEMENT], // GitHub fine-grained PAT + [/\bAKIA[0-9A-Z]{16}\b/g, REPLACEMENT], // AWS access key id + [/\bxox[baprs]-[A-Za-z0-9-]{10,}\b/g, REPLACEMENT], // Slack + [/\beyJ[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}/g, REPLACEMENT], // JWT + // Secret-looking key/value pairs (ENV=..., "password": "...", api_key: ...). + [ + /([A-Za-z0-9_]*(?:TOKEN|SECRET|PASSWORD|PASSWD|API[_-]?KEY|ACCESS[_-]?KEY|PRIVATE[_-]?KEY|CLIENT[_-]?SECRET)[A-Za-z0-9_]*\s*[:=]\s*)(["']?)([^\s"',;]+)\2/gi, + `$1$2${REPLACEMENT}$2`, + ], +] + +/** Returns `text` with credential-shaped substrings replaced by `[REDACTED]`. */ +export function redactSecrets(text: string, literals: readonly string[] = []): string { + if (!text) return text + let out = text + // Exact known values first (handles opaque tokens with no recognisable shape). + for (const literal of literals) { + if (literal && literal.length >= 6) out = out.split(literal).join(REPLACEMENT) + } + for (const [pattern, replacement] of RULES) out = out.replace(pattern, replacement) + return out +} diff --git a/src/v2/handlers/chat-headers.ts b/src/v2/handlers/chat-headers.ts new file mode 100644 index 0000000..5e49d9f --- /dev/null +++ b/src/v2/handlers/chat-headers.ts @@ -0,0 +1,28 @@ +import type { HandlerContext } from "../types.ts" +import { injectTraceContext } from "../../trace-context.ts" + +/** Injects the matching LLM step span context for explicitly enabled providers. */ +export function handleModelRequest( + event: { + sessionID: string + agent: string + model: { id: string; providerID: string } + headers: Record + }, + ctx: HandlerContext, +): void { + const providerID = event.model.providerID + if (!ctx.tracePropagationProviders.has(providerID) && !ctx.tracePropagationProviders.has("*")) return + + const request = ctx.llmRequestContexts + .get(event.sessionID) + ?.findLast( + (candidate) => + candidate.agent === event.agent && + candidate.modelID === event.model.id && + candidate.providerID === providerID, + ) + if (!request) return + + injectTraceContext(request.spanContext, event.headers) +} diff --git a/src/v2/handlers/session.ts b/src/v2/handlers/session.ts new file mode 100644 index 0000000..6a46f44 --- /dev/null +++ b/src/v2/handlers/session.ts @@ -0,0 +1,349 @@ +import { SeverityNumber } from "@opentelemetry/api-logs" +import { SpanStatusCode } from "@opentelemetry/api" +import { + AGENT_NAME, + INPUT_MIME_TYPE, + INPUT_VALUE, + LLM_INPUT_MESSAGES, + MimeType, + OpenInferenceSpanKind, + SemanticConventions, + SESSION_ID, +} from "@arizeai/openinference-semantic-conventions" +import { + agentAttrs, + getSessionAgentMeta, + isMetricEnabled, + isTraceEnabled, + resolveSessionTraceContext, + setBoundedMap, +} from "../util.ts" +import type { HandlerContext, SessionAgentType } from "../types.ts" +import { errorSummary, modelRef, type V2Error, type V2Model } from "../v2.ts" + +const OPENINFERENCE_SPAN_KIND = SemanticConventions.OPENINFERENCE_SPAN_KIND + +/** + * Creates the session totals entry on first sight. V2 does not always emit `session.created` + * (e.g. for sessions that predate the plugin), so totals are created lazily from whichever + * event arrives first. The session counter and `session.created` log are emitted exactly once. + */ +export function ensureSessionTotals( + sessionID: string, + agent: string, + agentType: SessionAgentType, + createdAt: number, + ctx: HandlerContext, +): boolean { + const existing = ctx.sessionTotals.get(sessionID) + if (existing) { + if (agent && agent !== "unknown" && (existing.agent !== agent || existing.agentType !== agentType)) { + setBoundedMap(ctx.sessionTotals, sessionID, { ...existing, agent, agentType }) + } + return false + } + if (isMetricEnabled("session.count", ctx)) { + ctx.instruments.sessionCounter.add(1, { + ...ctx.commonAttrs, + "session.id": sessionID, + is_subagent: agentType === "subagent", + }) + } + setBoundedMap(ctx.sessionTotals, sessionID, { + startMs: createdAt, + tokens: 0, + cost: 0, + messages: 0, + agent, + agentType, + }) + const prevMeta = ctx.sessionMeta.get(sessionID) + setBoundedMap(ctx.sessionMeta, sessionID, { + agent: agent !== "unknown" ? agent : (prevMeta?.agent ?? "unknown"), + model: prevMeta?.model ?? "unknown", + }) + ctx.emitLog({ + severityNumber: SeverityNumber.INFO, + severityText: "INFO", + timestamp: createdAt, + observedTimestamp: Date.now(), + body: "session.created", + attributes: { + "event.name": "session.created", + "session.id": sessionID, + is_subagent: agentType === "subagent", + ...agentAttrs(agent, agentType), + ...ctx.commonAttrs, + }, + }) + return true +} + +/** Starts or refreshes the root run span for a single user turn, keyed by the user message ID. */ +export function handleRunStarted( + runID: string, + sessionID: string, + agent: string, + promptText: string, + model: string, + startTime: number, + ctx: HandlerContext, +) { + ctx.activeRuns.set(sessionID, runID) + const safePrompt = ctx.redact(promptText) + if (!isTraceEnabled("session", ctx)) return + const existing = ctx.runSpans.get(runID) + if (existing) { + existing.setAttributes({ + [AGENT_NAME]: agent, + ...(promptText + ? { + [INPUT_VALUE]: safePrompt, + [INPUT_MIME_TYPE]: MimeType.TEXT, + [LLM_INPUT_MESSAGES]: JSON.stringify([{ role: "user", content: safePrompt }]), + } + : {}), + model, + }) + return + } + + const runSpan = ctx.tracer.startSpan( + `${ctx.tracePrefix}session`, + { + startTime, + attributes: { + [OPENINFERENCE_SPAN_KIND]: OpenInferenceSpanKind.AGENT, + [SESSION_ID]: sessionID, + [AGENT_NAME]: agent, + "agent.type": "primary", + "session.is_subagent": false, + ...(promptText + ? { + [INPUT_VALUE]: safePrompt, + [INPUT_MIME_TYPE]: MimeType.TEXT, + [LLM_INPUT_MESSAGES]: JSON.stringify([{ role: "user", content: safePrompt }]), + } + : {}), + model, + ...ctx.commonAttrs, + }, + }, + ctx.rootContext(), + ) + ctx.runSpans.set(runID, runSpan) + setBoundedMap(ctx.runSpanContexts, runID, runSpan.spanContext()) +} + +/** Records a session's creation, starting a subagent session span when it has a parent. */ +export function handleSessionCreated( + data: { sessionID: string; parentID?: string; agent?: string; model?: V2Model; title?: string }, + createdAt: number, + ctx: HandlerContext, +) { + const sessionID = data.sessionID + const isSubagent = !!data.parentID + const agentType: SessionAgentType = isSubagent ? "subagent" : "primary" + const agent = data.agent ?? "unknown" + ensureSessionTotals(sessionID, agent, agentType, createdAt, ctx) + setBoundedMap(ctx.sessionMeta, sessionID, { + agent, + model: data.model ? modelRef(data.model) : (ctx.sessionMeta.get(sessionID)?.model ?? "unknown"), + }) + + if (isTraceEnabled("session", ctx) && data.parentID) { + const sessionSpan = ctx.tracer.startSpan( + `${ctx.tracePrefix}session`, + { + startTime: createdAt, + attributes: { + [OPENINFERENCE_SPAN_KIND]: OpenInferenceSpanKind.AGENT, + [SESSION_ID]: sessionID, + [AGENT_NAME]: agent, + "agent.type": agentType, + "session.is_subagent": isSubagent, + ...(data.model ? { model: modelRef(data.model) } : {}), + ...ctx.commonAttrs, + }, + }, + resolveSessionTraceContext(data.parentID, ctx), + ) + ctx.sessionSpans.set(sessionID, sessionSpan) + setBoundedMap(ctx.sessionSpanContexts, sessionID, sessionSpan.spanContext()) + } + + return ctx.log("info", "otel: session.created", { sessionID, createdAt, isSubagent }) +} + +function sweepSession(sessionID: string, ctx: HandlerContext) { + for (const [key, tool] of ctx.pendingToolSpans) { + if (tool.sessionID === sessionID) { + tool.span?.setStatus({ code: SpanStatusCode.ERROR, message: "session ended before tool completed" }) + tool.span?.end() + ctx.pendingToolSpans.delete(key) + } + } + const prefix = `${sessionID}:` + for (const [key, step] of ctx.stepSpans) { + if (key.startsWith(prefix)) { + step.span.setStatus({ code: SpanStatusCode.ERROR, message: "session ended before step completed" }) + step.span.end() + ctx.stepSpans.delete(key) + } + } + for (const key of ctx.llmRequestContexts.keys()) { + if (key.startsWith(prefix)) ctx.llmRequestContexts.delete(key) + } +} + +/** Ends the active run span for a turn, stamping the cumulative session totals. */ +function endRunSpan(sessionID: string, ctx: HandlerContext, error?: string) { + const totals = ctx.sessionTotals.get(sessionID) + const runID = ctx.activeRuns.get(sessionID) + if (runID) ctx.activeRuns.delete(sessionID) + const runSpan = runID ? ctx.runSpans.get(runID) : undefined + if (!runSpan) return + if (totals) { + runSpan.setAttributes({ + [AGENT_NAME]: totals.agent, + "agent.type": totals.agentType, + "session.total_tokens": totals.tokens, + "session.total_cost_usd": totals.cost, + "session.total_messages": totals.messages, + }) + } + if (error) { + runSpan.setStatus({ code: SpanStatusCode.ERROR, message: error }) + runSpan.setAttribute("error", error) + } else { + runSpan.setStatus({ code: SpanStatusCode.OK }) + } + runSpan.end() + if (runID) ctx.runSpans.delete(runID) +} + +/** V2 emits this per turn; ensures totals exist and starts a run span if the prompt hook missed it. */ +export function handleExecutionStarted(data: { sessionID: string }, ctx: HandlerContext) { + const sessionID = data.sessionID + ensureSessionTotals(sessionID, "unknown", "primary", Date.now(), ctx) + if (!ctx.activeRuns.get(sessionID)) { + const agent = ctx.sessionTotals.get(sessionID)?.agent ?? "unknown" + handleRunStarted(`${sessionID}:exec:${Date.now()}`, sessionID, agent, "", "unknown", Date.now(), ctx) + } +} + +/** V2 emits this when a turn completes; ends the run span. */ +export function handleExecutionEnded( + data: { sessionID: string; reason?: string }, + ctx: HandlerContext, + error?: V2Error, +) { + const sessionID = data.sessionID + const totals = ctx.sessionTotals.get(sessionID) + const { agentName, agentType } = getSessionAgentMeta(sessionID, ctx) + endRunSpan(sessionID, ctx, error ? errorSummary(error) : undefined) + + const attrs = { ...ctx.commonAttrs, "session.id": sessionID } + if (totals) { + if (isMetricEnabled("session.duration", ctx)) { + ctx.instruments.sessionDurationHistogram.record(Date.now() - totals.startMs, attrs) + } + if (isMetricEnabled("session.token.total", ctx)) { + ctx.instruments.sessionTokenGauge.record(totals.tokens, attrs) + } + if (isMetricEnabled("session.cost.total", ctx)) { + ctx.instruments.sessionCostGauge.record(totals.cost, attrs) + } + } + + ctx.emitLog({ + severityNumber: error ? SeverityNumber.ERROR : SeverityNumber.INFO, + severityText: error ? "ERROR" : "INFO", + timestamp: Date.now(), + observedTimestamp: Date.now(), + body: error ? "session.error" : "session.execution.succeeded", + attributes: { + "event.name": error ? "session.error" : "session.execution.succeeded", + "session.id": sessionID, + ...(error ? { error: errorSummary(error) } : {}), + total_tokens: totals?.tokens ?? 0, + total_cost_usd: totals?.cost ?? 0, + total_messages: totals?.messages ?? 0, + ...agentAttrs(agentName, agentType), + ...ctx.commonAttrs, + }, + }) + ctx.log(error ? "error" : "debug", error ? "otel: session.execution.failed" : "otel: session.execution.succeeded", { + sessionID, + reason: data.reason, + ...(error ? { error: errorSummary(error) } : {}), + }) +} + +/** Session idle: end any remaining spans and clear per-session bookkeeping. */ +export function handleSessionIdle(sessionID: string, ctx: HandlerContext) { + const totals = ctx.sessionTotals.get(sessionID) + const { agentName, agentType } = getSessionAgentMeta(sessionID, ctx) + ctx.sessionTotals.delete(sessionID) + sweepSession(sessionID, ctx) + + const attrs = { ...ctx.commonAttrs, "session.id": sessionID } + if (totals) { + if (isMetricEnabled("session.duration", ctx)) { + ctx.instruments.sessionDurationHistogram.record(Date.now() - totals.startMs, attrs) + } + if (isMetricEnabled("session.token.total", ctx)) { + ctx.instruments.sessionTokenGauge.record(totals.tokens, attrs) + } + if (isMetricEnabled("session.cost.total", ctx)) { + ctx.instruments.sessionCostGauge.record(totals.cost, attrs) + } + } + endRunSpan(sessionID, ctx) + const sessionSpan = ctx.sessionSpans.get(sessionID) + if (sessionSpan) { + sessionSpan.setStatus({ code: SpanStatusCode.OK }) + sessionSpan.end() + ctx.sessionSpans.delete(sessionID) + } + + ctx.emitLog({ + severityNumber: SeverityNumber.INFO, + severityText: "INFO", + timestamp: Date.now(), + observedTimestamp: Date.now(), + body: "session.idle", + attributes: { + "event.name": "session.idle", + "session.id": sessionID, + total_tokens: totals?.tokens ?? 0, + total_cost_usd: totals?.cost ?? 0, + total_messages: totals?.messages ?? 0, + ...agentAttrs(agentName, agentType), + ...ctx.commonAttrs, + }, + }) + ctx.log("debug", "otel: session.idle", { sessionID }) +} + +/** Emits a `session.error` log event and ends the run span with error status. */ +export function handleExecutionFailed(data: { sessionID: string; error?: V2Error }, ctx: HandlerContext) { + handleExecutionEnded({ sessionID: data.sessionID }, ctx, data.error) +} + +/** Increments the retry counter when the session enters a retry state. */ +export function handleSessionStatus( + data: { sessionID: string; status?: { type: string; attempt?: number; message?: string } }, + ctx: HandlerContext, +) { + if (data.status?.type !== "retry") return + const { sessionID, status } = data + if (isMetricEnabled("retry.count", ctx)) { + ctx.instruments.retryCounter.add(1, { ...ctx.commonAttrs, "session.id": sessionID }) + ctx.log("debug", "otel: retry counter incremented", { + sessionID, + attempt: status.attempt, + retryMessage: status.message, + }) + } +} diff --git a/src/v2/handlers/tool.ts b/src/v2/handlers/tool.ts new file mode 100644 index 0000000..926a1f8 --- /dev/null +++ b/src/v2/handlers/tool.ts @@ -0,0 +1,104 @@ +import { SpanStatusCode } from "@opentelemetry/api" +import { + INPUT_MIME_TYPE, + INPUT_VALUE, + MimeType, + OpenInferenceSpanKind, + OUTPUT_MIME_TYPE, + OUTPUT_VALUE, + SemanticConventions, + SESSION_ID, +} from "@arizeai/openinference-semantic-conventions" +import { isMetricEnabled, isTraceEnabled, resolveSessionTraceContext, setBoundedMap } from "../util.ts" +import type { HandlerContext } from "../types.ts" +import { errorSummary, type V2Error, type V2ToolContent } from "../v2.ts" + +const OPENINFERENCE_SPAN_KIND = SemanticConventions.OPENINFERENCE_SPAN_KIND +const TOOL_NAME = "tool.name" + +/** Remembers the tool name for a call id, since V2 carries it only on `session.tool.input.started`. */ +export function handleToolInputStarted(data: { sessionID: string; id: string; name?: string }, ctx: HandlerContext) { + if (data.name) setBoundedMap(ctx.pendingToolNames, data.id, data.name) +} + +/** Starts the tool span when the tool is called. */ +export function handleToolCalled( + data: { sessionID: string; id: string; input?: Record; assistantMessageID?: string }, + ctx: HandlerContext, +) { + const tool = ctx.pendingToolNames.get(data.id) ?? "unknown" + if (!isTraceEnabled("tool", ctx)) return + const span = ctx.tracer.startSpan( + `${ctx.tracePrefix}tool.${tool}`, + { + startTime: Date.now(), + attributes: { + [OPENINFERENCE_SPAN_KIND]: OpenInferenceSpanKind.TOOL, + [SESSION_ID]: data.sessionID, + [TOOL_NAME]: tool, + [INPUT_VALUE]: ctx.redact(JSON.stringify(data.input ?? {})), + [INPUT_MIME_TYPE]: MimeType.JSON, + ...ctx.commonAttrs, + }, + }, + resolveSessionTraceContext(data.sessionID, ctx, { + assistantMessageID: data.assistantMessageID, + }), + ) + ctx.pendingToolSpans.set(data.id, { tool, sessionID: data.sessionID, startMs: Date.now(), span }) +} + +/** Ends the tool span successfully and records its duration. */ +export function handleToolSuccess( + data: { sessionID: string; id: string; content?: V2ToolContent[]; metadata?: Record }, + ctx: HandlerContext, +) { + finishTool(data.sessionID, data.id, ctx, data.content, undefined) +} + +/** Ends the tool span with an error and records its duration. */ +export function handleToolFailed( + data: { sessionID: string; id: string; content?: V2ToolContent[]; error?: V2Error }, + ctx: HandlerContext, +) { + finishTool(data.sessionID, data.id, ctx, data.content, data.error) +} + +function finishTool( + sessionID: string, + id: string, + ctx: HandlerContext, + content: V2ToolContent[] | undefined, + error: V2Error | undefined, +) { + const pending = ctx.pendingToolSpans.get(id) + const tool = pending?.tool ?? ctx.pendingToolNames.get(id) ?? "unknown" + const startMs = pending?.startMs ?? Date.now() + const duration = Date.now() - startMs + + const output = content?.map((part) => (part.type === "text" ? part.text : part.uri)).join("\n") + if (pending?.span) { + if (error) { + pending.span.setStatus({ code: SpanStatusCode.ERROR, message: errorSummary(error) }) + pending.span.setAttribute("error", errorSummary(error)) + } else { + pending.span.setStatus({ code: SpanStatusCode.OK }) + } + if (output) { + pending.span.setAttributes({ [OUTPUT_VALUE]: ctx.redact(output), [OUTPUT_MIME_TYPE]: MimeType.TEXT }) + } + pending.span.end() + } + + if (isMetricEnabled("tool.duration", ctx)) { + ctx.instruments.toolDurationHistogram.record(duration, { + ...ctx.commonAttrs, + "session.id": sessionID, + tool, + }) + } + + ctx.pendingToolSpans.delete(id) + ctx.pendingToolNames.delete(id) + ctx.log("debug", error ? "otel: tool.failed" : "otel: tool.success", { sessionID, tool, duration }) +} diff --git a/src/v2/handlers/usage.ts b/src/v2/handlers/usage.ts new file mode 100644 index 0000000..48e23b1 --- /dev/null +++ b/src/v2/handlers/usage.ts @@ -0,0 +1,202 @@ +import { SeverityNumber } from "@opentelemetry/api-logs" +import { SpanStatusCode, trace } from "@opentelemetry/api" +import { + INPUT_MIME_TYPE, + INPUT_VALUE, + LLM_COST_TOTAL, + LLM_MODEL_NAME, + LLM_PROVIDER, + LLM_SYSTEM, + LLM_TOKEN_COUNT_COMPLETION, + LLM_TOKEN_COUNT_COMPLETION_DETAILS_REASONING, + LLM_TOKEN_COUNT_PROMPT, + LLM_TOKEN_COUNT_PROMPT_DETAILS_CACHE_READ, + LLM_TOKEN_COUNT_PROMPT_DETAILS_CACHE_WRITE, + LLM_TOKEN_COUNT_TOTAL, + MimeType, + OpenInferenceSpanKind, + SemanticConventions, + SESSION_ID, +} from "@arizeai/openinference-semantic-conventions" +import { agentAttrs, genAiProviderName, isMetricEnabled, isTraceEnabled, resolveSessionTraceContext, setBoundedMap } from "../util.ts" +import type { HandlerContext } from "../types.ts" +import { errorSummary, modelRef, totalTokens, type V2Error, type V2Model, type V2Tokens } from "../v2.ts" + +const OPENINFERENCE_SPAN_KIND = SemanticConventions.OPENINFERENCE_SPAN_KIND +const LLM_FINISH_REASON = "llm.finish_reason" + +const stepKey = (sessionID: string, assistantMessageID: string) => `${sessionID}:${assistantMessageID}` + +/** Applies the cumulative per-session usage snapshot emitted by V2 `session.usage.updated`. */ +export function handleUsageUpdated( + data: { sessionID: string; cost?: number; tokens?: V2Tokens }, + ctx: HandlerContext, +) { + const existing = ctx.sessionTotals.get(data.sessionID) + if (!existing) return + setBoundedMap(ctx.sessionTotals, data.sessionID, { + ...existing, + tokens: totalTokens(data.tokens), + cost: data.cost ?? 0, + }) +} + +/** Starts the LLM step span for an assistant message and records its request context for header injection. */ +export function handleStepStarted( + data: { sessionID: string; assistantMessageID: string; agent?: string; model?: V2Model; started?: number }, + ctx: HandlerContext, +) { + const agent = data.agent ?? "unknown" + const modelID = data.model?.id ?? "unknown" + const providerID = data.model?.providerID ?? "unknown" + const started = data.started ?? Date.now() + const runID = ctx.activeRuns.get(data.sessionID) + if (runID) setBoundedMap(ctx.assistantRuns, data.assistantMessageID, runID) + setBoundedMap(ctx.sessionMeta, data.sessionID, { + agent, + model: modelRef(data.model), + }) + // Enrich the turn's root span with the model once the first step resolves it. + const runSpan = runID ? ctx.runSpans.get(runID) : undefined + if (runSpan) runSpan.setAttribute("model", modelRef(data.model)) + // Propagate the resolved agent into session totals so turn-end logs/spans carry it. + const totals = ctx.sessionTotals.get(data.sessionID) + if (totals && agent !== "unknown" && totals.agent !== agent) { + setBoundedMap(ctx.sessionTotals, data.sessionID, { ...totals, agent }) + } + + // Emit the user_prompt log once per turn, now that agent/model are resolved, linked to + // the turn's run span so it shares the trace. + if (runID && !ctx.promptEmitted.has(runID)) { + ctx.promptEmitted.add(runID) + const promptText = ctx.runPrompts.get(runID) ?? "" + const promptContext = runSpan ? trace.setSpan(ctx.rootContext(), runSpan) : undefined + ctx.emitLog( + { + severityNumber: SeverityNumber.INFO, + severityText: "INFO", + timestamp: Date.now(), + observedTimestamp: Date.now(), + body: "user_prompt", + attributes: { + "event.name": "user_prompt", + "session.id": data.sessionID, + ...agentAttrs(agent, "primary"), + prompt_length: promptText.length, + ...(ctx.capturePromptInLogs ? { prompt: ctx.redact(promptText) } : {}), + model: modelRef(data.model), + ...ctx.commonAttrs, + }, + }, + promptContext, + ) + } + + if (!isTraceEnabled("llm", ctx)) return + const span = ctx.tracer.startSpan( + `${ctx.tracePrefix}llm`, + { + startTime: started, + attributes: { + [OPENINFERENCE_SPAN_KIND]: OpenInferenceSpanKind.LLM, + [SESSION_ID]: data.sessionID, + [LLM_SYSTEM]: providerID, + [LLM_PROVIDER]: providerID, + "gen_ai.provider.name": genAiProviderName(providerID), + [LLM_MODEL_NAME]: modelID, + agent, + ...ctx.commonAttrs, + }, + }, + resolveSessionTraceContext(data.sessionID, ctx, { assistantMessageID: data.assistantMessageID }), + ) + ctx.stepSpans.set(stepKey(data.sessionID, data.assistantMessageID), { + span, + sessionID: data.sessionID, + agent, + modelID, + providerID, + started, + }) + const existing = ctx.llmRequestContexts.get(data.sessionID) ?? [] + ctx.llmRequestContexts.set( + data.sessionID, + [ + ...existing.slice(-9), + { messageID: data.assistantMessageID, agent, modelID, providerID, spanContext: span.spanContext() }, + ], + ) +} + +/** Ends the LLM step span and records token, cost, cache, and message metrics. */ +export function handleStepEnded( + data: { + sessionID: string + assistantMessageID: string + finish?: string + cost?: number + tokens?: V2Tokens + files?: string[] + }, + ctx: HandlerContext, + error?: V2Error, +) { + const key = stepKey(data.sessionID, data.assistantMessageID) + const step = ctx.stepSpans.get(key) + const tokens = data.tokens + const total = totalTokens(tokens) + const attrs = { ...ctx.commonAttrs, "session.id": data.sessionID, agent: step?.agent ?? "unknown" } + + if (step) { + step.span.setAttributes({ + ...(tokens + ? { + [LLM_TOKEN_COUNT_PROMPT]: tokens.input, + [LLM_TOKEN_COUNT_COMPLETION]: tokens.output, + [LLM_TOKEN_COUNT_COMPLETION_DETAILS_REASONING]: tokens.reasoning, + [LLM_TOKEN_COUNT_PROMPT_DETAILS_CACHE_READ]: tokens.cache.read, + [LLM_TOKEN_COUNT_PROMPT_DETAILS_CACHE_WRITE]: tokens.cache.write, + [LLM_TOKEN_COUNT_TOTAL]: total, + } + : {}), + [LLM_FINISH_REASON]: error ? "error" : (data.finish ?? "stop"), + ...(typeof data.cost === "number" ? { [LLM_COST_TOTAL]: data.cost } : {}), + ...(data.files ? { "session.files_changed": data.files.length } : {}), + }) + if (error) { + step.span.setStatus({ code: SpanStatusCode.ERROR, message: errorSummary(error) }) + } else { + step.span.setStatus({ code: SpanStatusCode.OK }) + } + step.span.end() + ctx.stepSpans.delete(key) + } + + if (isMetricEnabled("token.usage", ctx)) { + ctx.instruments.tokenCounter.add(total, { + ...attrs, + model: step ? modelRef({ id: step.modelID, providerID: step.providerID }) : "unknown", + }) + } + if (tokens && isMetricEnabled("cache.count", ctx)) { + ctx.instruments.cacheCounter.add(tokens.cache.read + tokens.cache.write, attrs) + } + if (typeof data.cost === "number" && isMetricEnabled("cost.usage", ctx)) { + ctx.instruments.costCounter.add(data.cost, attrs) + } + if (isMetricEnabled("model.usage", ctx) && step) { + ctx.instruments.modelUsageCounter.add(1, { + ...attrs, + model: step.modelID, + provider: step.providerID, + }) + } + if (isMetricEnabled("message.count", ctx)) { + ctx.instruments.messageCounter.add(1, attrs) + } + + const existing = ctx.sessionTotals.get(data.sessionID) + if (existing) { + setBoundedMap(ctx.sessionTotals, data.sessionID, { ...existing, messages: existing.messages + 1 }) + } +} diff --git a/src/v2/index.ts b/src/v2/index.ts new file mode 100644 index 0000000..1b41e9a --- /dev/null +++ b/src/v2/index.ts @@ -0,0 +1,443 @@ +import { logs } from "@opentelemetry/api-logs" +import { diag, DiagLogLevel, ROOT_CONTEXT, trace } from "@opentelemetry/api" +import { appendFileSync, mkdirSync, statSync, writeFileSync } from "node:fs" +import { homedir } from "node:os" +import { dirname, join } from "node:path" +import { LEVELS, type HandlerContext, type Level } from "./types.ts" +import { loadConfig, parseAttributePairs, resolveHelperPath, type OtelPluginOptions } from "../config.ts" +import { probeEndpoint } from "../probe.ts" +import { createInstruments, forceFlushOtel, setupOtel, type OtelProviders } from "../otel.ts" +import { remoteParentContext } from "../trace-context.ts" +import { + handleExecutionEnded, + handleExecutionFailed, + handleExecutionStarted, + handleRunStarted, + handleSessionCreated, + handleSessionIdle, + handleSessionStatus, +} from "./handlers/session.ts" +import { handleStepEnded, handleStepStarted, handleUsageUpdated } from "./handlers/usage.ts" +import { + handleToolCalled, + handleToolFailed, + handleToolInputStarted, + handleToolSuccess, +} from "./handlers/tool.ts" +import { handleModelRequest } from "./handlers/chat-headers.ts" +import { setBoundedMap } from "../util.ts" +import { redactSecrets } from "../redact.ts" +import type { V2Event } from "./v2.ts" + +const PLUGIN_VERSION = "2.0.0-v2-port" + +/** Append-only diagnostics for debugging the live plugin (server console output is not captured). */ +const DIAG_LOG = + process.env["OPENCODE_OTEL_DIAG_LOG"] ?? + join(homedir(), ".local", "share", "opencode", "otel-plugin-diag.log") +function diagLog(message: string) { + try { + mkdirSync(dirname(DIAG_LOG), { recursive: true }) + if (((statSync(DIAG_LOG, { throwIfNoEntry: false })?.size) ?? 0) > 1_000_000) writeFileSync(DIAG_LOG, "") + appendFileSync(DIAG_LOG, `${new Date().toISOString()} ${message}\n`) + } catch { + /* ignore */ + } +} + +/** Resolves `{env:VAR}` placeholders from the process environment as a fallback to V2 config interpolation. */ +function resolveEnv(value: string | undefined): string | undefined { + if (!value) return value + return value.replace(/\{env:([A-Za-z_][A-Za-z0-9_]*)\}/g, (_match, name: string) => process.env[name] ?? "") +} + +// V2 loads a global plugin once per location, and evaluates the module per instance, so +// module-level state is not shared. The OTel SDK's global providers may only be registered +// once per process, so the instance is stored on `globalThis` and reused. Providers are +// flushed (never shut down) on release: shutting a global provider down poisons the OTel +// registry for the rest of the process, which would silently drop all later telemetry. +const SHARED_KEY = "__opencode_otel_shared_v1__" + +type SharedOtel = { + providers: OtelProviders + instruments: ReturnType + logger: ReturnType + tracer: ReturnType + refs: number +} + +type SharedState = { promise: Promise | null; instance: SharedOtel | null } + +function sharedState(): SharedState { + const g = globalThis as unknown as Record + if (!g[SHARED_KEY]) g[SHARED_KEY] = { promise: null, instance: null } + return g[SHARED_KEY]! +} + +// Tracing bookkeeping is keyed by session/message id and must be shared across every plugin +// instance in the process, otherwise a run span created by one instance's hook is invisible +// to the instance handling the matching event, and child spans become separate root traces. +const TRACING_KEY = "__opencode_otel_tracing_v1__" + +type TracingState = { + seenEvents: Set + runPrompts: Map + promptEmitted: Set + pendingToolSpans: HandlerContext["pendingToolSpans"] + pendingToolNames: HandlerContext["pendingToolNames"] + sessionTotals: HandlerContext["sessionTotals"] + sessionMeta: HandlerContext["sessionMeta"] + runSpans: HandlerContext["runSpans"] + runSpanContexts: HandlerContext["runSpanContexts"] + activeRuns: HandlerContext["activeRuns"] + assistantRuns: HandlerContext["assistantRuns"] + sessionSpans: HandlerContext["sessionSpans"] + sessionSpanContexts: HandlerContext["sessionSpanContexts"] + stepSpans: HandlerContext["stepSpans"] + llmRequestContexts: HandlerContext["llmRequestContexts"] +} + +function emptyTracing(): TracingState { + return { + seenEvents: new Set(), + runPrompts: new Map(), + promptEmitted: new Set(), + pendingToolSpans: new Map(), + pendingToolNames: new Map(), + sessionTotals: new Map(), + sessionMeta: new Map(), + runSpans: new Map(), + runSpanContexts: new Map(), + activeRuns: new Map(), + assistantRuns: new Map(), + sessionSpans: new Map(), + sessionSpanContexts: new Map(), + stepSpans: new Map(), + llmRequestContexts: new Map(), + } +} + +function tracingState(): TracingState { + const g = globalThis as unknown as Record + const existing = g[TRACING_KEY] + if (existing) { + // Backfill fields added since the object was first created, so a plugin reload + // against a long-lived process cannot leave new fields undefined. + const defaults = emptyTracing() + for (const key of Object.keys(defaults) as (keyof TracingState)[]) { + if (existing[key] === undefined) (existing as Record)[key] = defaults[key] + } + return existing + } + const created = emptyTracing() + g[TRACING_KEY] = created + return created +} + +async function acquireOtel(input: { + config: ReturnType + otlpHeaders: string | undefined + helper: string | undefined +}): Promise { + const state = sharedState() + if (!state.promise) { + state.promise = (async () => { + diag.setLogger( + { + error: (...args: unknown[]) => diagLog(`OTEL ERROR ${args.map(String).join(" ")}`), + warn: (...args: unknown[]) => diagLog(`OTEL WARN ${args.map(String).join(" ")}`), + info: () => {}, + debug: () => {}, + verbose: () => {}, + }, + DiagLogLevel.WARN, + ) + const providers = await setupOtel( + input.config.endpoint, + input.config.protocol, + input.config.metricsInterval, + input.config.logsInterval, + PLUGIN_VERSION, + input.otlpHeaders, + input.helper, + ) + const instance: SharedOtel = { + providers, + instruments: createInstruments(input.config.metricPrefix), + logger: logs.getLogger("com.opencode"), + tracer: trace.getTracer("com.opencode"), + refs: 0, + } + state.instance = instance + diagLog(`otel shared instance initialized endpoint=${input.config.endpoint} protocol=${input.config.protocol}`) + return instance + })() + } + const instance = await state.promise + instance.refs++ + diagLog(`otel shared instance acquired refs=${instance.refs}`) + return instance +} + +async function releaseOtel() { + const instance = sharedState().instance + if (!instance) return + instance.refs-- + diagLog(`otel shared instance released refs=${instance.refs}`) + if (instance.refs <= 0) { + await forceFlushOtel(instance.providers).catch(() => {}) + } +} + +/** + * V2 port of @devtheops/opencode-plugin-otel (scoped v0.1). + * + * Emits session lifecycle, usage/token/cost, LLM step, tool, retry, and execution + * telemetry from the V2 granular event stream. V1-only signals (message parts, + * permission prompts, command execution, session diffs) are out of scope; see README. + */ +export default { + id: "devtheops.otel", + async setup(ctx: any) { + const options = (ctx?.options ?? {}) as OtelPluginOptions + const config = loadConfig(options) + const directory: string | undefined = ctx?.location?.directory + const worktree: string | undefined = + ctx?.location?.project?.canonical ?? ctx?.location?.project?.directory + const otlpHeadersHelper = resolveHelperPath(config.otlpHeadersHelper, directory, worktree) + + const minLevel: Level = "info" + const log: HandlerContext["log"] = async (level, message, extra) => { + diagLog(`log ${level}: ${message} ${extra ? JSON.stringify(extra) : ""}`) + if (LEVELS[level] < LEVELS[minLevel]) return + const line = `[opencode-plugin-otel] ${level}: ${message}` + if (level === "error") console.error(line, extra ?? "") + else if (level === "warn") console.warn(line, extra ?? "") + else console.log(line, extra ?? "") + } + + if (!config.enabled) { + await log("info", "telemetry disabled (set OPENCODE_ENABLE_TELEMETRY to enable)") + return + } + + const probe = await probeEndpoint(config.endpoint) + if (!probe.ok) { + await log("warn", "OTLP endpoint unreachable — exports may fail", { + endpoint: config.endpoint, + error: probe.error, + }) + } + + const shared = await acquireOtel({ + config, + otlpHeaders: resolveEnv(config.otlpHeaders), + helper: otlpHeadersHelper, + }) + const { instruments, logger, tracer, providers } = shared + + await log("info", "starting up", { + version: PLUGIN_VERSION, + endpoint: config.endpoint, + protocol: config.protocol, + metricsInterval: config.metricsInterval, + logsInterval: config.logsInterval, + metricPrefix: config.metricPrefix, + headersHelperSet: !!config.otlpHeadersHelper, + redactSecrets: config.redactSecrets, + redactValueCount: config.redactValues.length, + }) + + const emitLog: HandlerContext["emitLog"] = (record, context) => { + if (!config.logsEnabled) return + logger.emit(context ? { ...record, context } : record) + } + const remoteContext = remoteParentContext(config.traceparent, config.tracestate) + const rootContext = remoteContext ? () => remoteContext : () => ROOT_CONTEXT + + const commonAttrs = { + ...parseAttributePairs(config.spanAttributes), + "project.id": ctx?.location?.project?.id ?? "unknown", + } as const + + const tracing = tracingState() + const hctx: HandlerContext = { + log, + emitLog, + instruments, + commonAttrs, + pendingToolSpans: tracing.pendingToolSpans, + pendingToolNames: tracing.pendingToolNames, + sessionTotals: tracing.sessionTotals, + sessionMeta: tracing.sessionMeta, + disabledMetrics: config.disabledMetrics, + disabledTraces: config.disabledTraces, + tracer, + tracePrefix: config.metricPrefix, + rootContext, + runSpans: tracing.runSpans, + runSpanContexts: tracing.runSpanContexts, + activeRuns: tracing.activeRuns, + assistantRuns: tracing.assistantRuns, + sessionSpans: tracing.sessionSpans, + sessionSpanContexts: tracing.sessionSpanContexts, + stepSpans: tracing.stepSpans, + llmRequestContexts: tracing.llmRequestContexts, + tracePropagationProviders: config.tracePropagationProviders, + capturePromptInLogs: config.capturePromptInLogs, + redact: config.redactSecrets + ? (text: string) => redactSecrets(text, config.redactValues) + : (text: string) => text, + runPrompts: tracing.runPrompts, + promptEmitted: tracing.promptEmitted, + } + + const safe = ( + name: string, + fn: (...args: T) => Promise | void, + ): ((...args: T) => Promise) => + async (...args: T) => { + try { + await fn(...args) + } catch (err) { + await log("error", `otel: unhandled error in ${name}`, { + error: err instanceof Error ? err.message : String(err), + stack: err instanceof Error ? err.stack : undefined, + }) + } + } + + // Prompt admission: start the root run span for the user turn. + await ctx.session.hook( + "prompt", + safe("prompt", async (event: any) => { + diagLog(`hook prompt session=${event?.sessionID} msg=${event?.messageID}`) + const sessionID = event?.sessionID + let meta = hctx.sessionMeta.get(sessionID) + if (!meta || meta.agent === "unknown" || meta.model === "unknown") { + try { + const info: any = await ctx.session.get({ sessionID }) + const s: any = info?.data ?? info + meta = { + agent: s?.agent ?? meta?.agent ?? "unknown", + model: s?.model ? `${s.model.providerID}/${s.model.id}` : (meta?.model ?? "unknown"), + } + diagLog( + `prompt enrich session=${sessionID} rawKeys=${Object.keys(info ?? {}).join(",")} agent=${meta.agent} model=${meta.model}`, + ) + setBoundedMap(hctx.sessionMeta, sessionID, meta) + } catch (err) { + diagLog(`prompt enrich failed session=${sessionID}: ${err instanceof Error ? err.message : String(err)}`) + } + } + const agent = meta?.agent ?? hctx.sessionTotals.get(sessionID)?.agent ?? "unknown" + const model = meta?.model ?? "unknown" + const promptText: string = event?.prompt?.text ?? "" + handleRunStarted(event.messageID, sessionID, agent, promptText, model, Date.now(), hctx) + // The user_prompt log is emitted on the first step, where agent/model are resolved. + setBoundedMap(hctx.runPrompts, event.messageID, promptText) + }), + ) + + // Model request: inject W3C trace context for enabled providers. + await ctx.session.hook( + "model.request", + safe("model.request", (event: any) => { + handleModelRequest( + { + sessionID: event.sessionID, + agent: event.agent, + model: event.model, + headers: event.headers, + }, + hctx, + ) + }), + ) + + const controller = new AbortController() + void (async () => { + try { + for await (const event of ctx.event.subscribe({ signal: controller.signal }) as AsyncIterable) { + // V2 delivers each event to every plugin instance; process it exactly once. + const eventID = event.id + if (eventID) { + if (tracing.seenEvents.has(eventID)) continue + tracing.seenEvents.add(eventID) + if (tracing.seenEvents.size > 5000) { + const oldest = tracing.seenEvents.values().next().value + if (oldest) tracing.seenEvents.delete(oldest) + } + } + await safe(`event:${event.type}`, async () => { + switch (event.type) { + case "session.created": + await handleSessionCreated(event.data as any, event.created, hctx) + break + case "session.idle": + handleSessionIdle((event.data as any).sessionID, hctx) + await forceFlushOtel(providers) + break + case "session.status": + handleSessionStatus(event.data as any, hctx) + break + case "session.usage.updated": + handleUsageUpdated(event.data as any, hctx) + break + case "session.step.started": + handleStepStarted(event.data as any, hctx) + break + case "session.step.ended": + handleStepEnded(event.data as any, hctx) + break + case "session.step.failed": + handleStepEnded(event.data as any, hctx, (event.data as any).error) + await forceFlushOtel(providers) + break + case "session.tool.input.started": + handleToolInputStarted(event.data as any, hctx) + break + case "session.tool.called": + handleToolCalled(event.data as any, hctx) + break + case "session.tool.success": + handleToolSuccess(event.data as any, hctx) + break + case "session.tool.failed": + handleToolFailed(event.data as any, hctx) + break + case "session.execution.started": + handleExecutionStarted(event.data as any, hctx) + break + case "session.execution.succeeded": + handleExecutionEnded(event.data as any, hctx) + await forceFlushOtel(providers) + break + case "session.execution.failed": + handleExecutionFailed(event.data as any, hctx) + await forceFlushOtel(providers) + break + case "session.execution.interrupted": + handleExecutionEnded(event.data as any, hctx) + await forceFlushOtel(providers) + break + } + })() + } + } catch (err) { + if (!controller.signal.aborted) { + await log("error", "otel: event subscription ended", { + error: err instanceof Error ? err.message : String(err), + }) + } + } + })() + + await log("info", "plugin ready", { version: PLUGIN_VERSION }) + + return async () => { + controller.abort() + await releaseOtel() + } + }, +} diff --git a/src/v2/types.ts b/src/v2/types.ts new file mode 100644 index 0000000..8d0be87 --- /dev/null +++ b/src/v2/types.ts @@ -0,0 +1,110 @@ +import type { Context, Counter, Gauge, Histogram, Span, SpanContext, Tracer } from "@opentelemetry/api" +import type { LogRecord } from "@opentelemetry/api-logs" + +/** Numeric priority map for log levels; higher value = higher severity. */ +export const LEVELS = { debug: 0, info: 1, warn: 2, error: 3 } as const + +/** Union of supported log level names. */ +export type Level = keyof typeof LEVELS + +/** Maximum number of entries kept in bounded maps. */ +export const MAX_PENDING = 500 + +/** Structured logger forwarded to the opencode plugin logger. */ +export type PluginLogger = ( + level: Level, + message: string, + extra?: Record, +) => Promise + +/** OTel attributes common to every emitted span, log, and metric. */ +export type CommonAttrs = Readonly> + +/** In-flight tool execution tracked between `called` and `success`/`failed`. */ +export type PendingToolSpan = { + tool: string + sessionID: string + startMs: number + span?: Span +} + +/** Session role emitted by opencode: either the primary/root agent or a spawned subagent. */ +export type SessionAgentType = "primary" | "subagent" + +/** Accumulated per-session totals used for gauge snapshots on session idle. */ +export type SessionTotals = { + startMs: number + tokens: number + cost: number + messages: number + agent: string + agentType: SessionAgentType +} + +/** Live LLM request span metadata used by the outbound header hook. */ +export type LlmRequestContext = { + messageID: string + agent: string + modelID: string + providerID: string + spanContext: SpanContext +} + +/** In-flight model step span keyed by `${sessionID}:${assistantMessageID}`. */ +export type StepSpan = { + span: Span + sessionID: string + agent: string + modelID: string + providerID: string + started: number +} + +/** OTel metric instruments created once at plugin startup and shared via `HandlerContext`. */ +export type Instruments = { + sessionCounter: Counter + tokenCounter: Counter + costCounter: Counter + linesCounter: Counter + linesTotalGauge: Gauge + commitCounter: Counter + toolDurationHistogram: Histogram + cacheCounter: Counter + sessionDurationHistogram: Histogram + messageCounter: Counter + sessionTokenGauge: Histogram + sessionCostGauge: Histogram + modelUsageCounter: Counter + retryCounter: Counter + subtaskCounter: Counter +} + +/** Shared context threaded through every V2 event handler. */ +export type HandlerContext = { + log: PluginLogger + emitLog: (record: LogRecord, context?: Context) => void + instruments: Instruments + commonAttrs: CommonAttrs + pendingToolSpans: Map + pendingToolNames: Map + sessionTotals: Map + sessionMeta: Map + disabledMetrics: Set + disabledTraces: Set + tracer: Tracer + tracePrefix: string + rootContext: () => Context + runSpans: Map + runSpanContexts: Map + activeRuns: Map + assistantRuns: Map + sessionSpans: Map + sessionSpanContexts: Map + stepSpans: Map + llmRequestContexts: Map + tracePropagationProviders: Set + capturePromptInLogs: boolean + redact: (text: string) => string + runPrompts: Map + promptEmitted: Set +} diff --git a/src/v2/util.ts b/src/v2/util.ts new file mode 100644 index 0000000..27958a0 --- /dev/null +++ b/src/v2/util.ts @@ -0,0 +1,129 @@ +import { trace } from "@opentelemetry/api" +import { MAX_PENDING } from "./types.ts" +import type { HandlerContext, SessionAgentType } from "./types.ts" + +const GEN_AI_PROVIDER_NAMES: Readonly> = { + "amazon-bedrock": "aws.bedrock", + azure: "azure.ai.openai", + "azure-cognitive-services": "azure.ai.openai", + google: "gcp.gemini", + "google-vertex": "gcp.vertex_ai", + "google-vertex-anthropic": "gcp.vertex_ai", + mistral: "mistral_ai", + xai: "x_ai", +} + +/** Returns a human-readable summary string from an opencode error object. */ +export function errorSummary(err: { name: string; data?: unknown } | undefined): string { + if (!err) return "unknown" + if (err.data && typeof err.data === "object" && "message" in err.data) { + return `${err.name}: ${(err.data as { message: string }).message}` + } + return err.name +} + +/** Returns the canonical OTel GenAI provider name, preserving unknown provider IDs. */ +export function genAiProviderName(providerID: string): string { + return GEN_AI_PROVIDER_NAMES[providerID] ?? providerID +} + +/** + * Inserts a key/value pair into `map`, evicting the oldest entry first when the map + * has reached `MAX_PENDING` capacity to prevent unbounded memory growth. + */ +export function setBoundedMap(map: Map, key: K, value: V) { + if (!map.has(key) && map.size >= MAX_PENDING) { + const [firstKey] = map.keys() + if (firstKey !== undefined) map.delete(firstKey) + } + map.set(key, value) +} + +/** Resolves a root-run context from the live span first, then from the retained ended span context. */ +export function resolveRunTraceContext(runID: string, ctx: Pick) { + const baseCtx = ctx.rootContext() + const runSpan = ctx.runSpans.get(runID) + if (runSpan) return trace.setSpan(baseCtx, runSpan) + const runSpanContext = ctx.runSpanContexts.get(runID) + return runSpanContext ? trace.setSpanContext(baseCtx, runSpanContext) : baseCtx +} + +/** Resolves the best available trace parent for a session event or message/tool child span. */ +export function resolveSessionTraceContext( + sessionID: string, + ctx: HandlerContext, + input?: { assistantMessageID?: string; runID?: string }, +) { + const baseCtx = ctx.rootContext() + const sessionSpan = ctx.sessionSpans.get(sessionID) + if (sessionSpan) return trace.setSpan(baseCtx, sessionSpan) + const sessionSpanContext = ctx.sessionSpanContexts.get(sessionID) + if (sessionSpanContext) return trace.setSpanContext(baseCtx, sessionSpanContext) + if (input?.runID) return resolveRunTraceContext(input.runID, ctx) + const assistantRunID = input?.assistantMessageID + ? ctx.assistantRuns.get(input.assistantMessageID) + : undefined + if (assistantRunID) return resolveRunTraceContext(assistantRunID, ctx) + const activeRunID = ctx.activeRuns.get(sessionID) + return activeRunID ? resolveRunTraceContext(activeRunID, ctx) : baseCtx +} + +/** + * Returns `true` if the metric name (without prefix) is not in the disabled set. + * The `name` should be the suffix after the metric prefix, e.g. `"session.count"`. + */ +export function isMetricEnabled(name: string, ctx: { disabledMetrics: Set }): boolean { + return !ctx.disabledMetrics.has(name) +} + +/** + * Returns `true` if the trace type is not in the disabled set. + * Valid names are `"session"`, `"llm"`, and `"tool"`. + */ +export function isTraceEnabled(name: string, ctx: { disabledTraces: Set }): boolean { + return !ctx.disabledTraces.has(name) +} + +/** + * Accumulates token and cost totals for a session, and increments the message count. + * Uses `setBoundedMap` to produce a new object rather than mutating in-place. + * No-ops silently if the session was not previously registered via `handleSessionCreated`. + */ +export function accumulateSessionTotals( + sessionID: string, + tokens: number, + cost: number, + ctx: HandlerContext, +) { + const existing = ctx.sessionTotals.get(sessionID) + if (!existing) return + setBoundedMap(ctx.sessionTotals, sessionID, { + startMs: existing.startMs, + tokens: existing.tokens + tokens, + cost: existing.cost + cost, + messages: existing.messages + 1, + agent: existing.agent, + agentType: existing.agentType, + }) +} + +/** Returns the current session-scoped agent name/type, defaulting to `unknown` when unavailable. */ +export function getSessionAgentMeta( + sessionID: string, + ctx: Pick, +): { agentName: string; agentType: SessionAgentType | "unknown" } { + const totals = ctx.sessionTotals.get(sessionID) + return { + agentName: totals?.agent ?? "unknown", + agentType: totals?.agentType ?? "unknown", + } +} + +/** Builds a consistent agent attribute set for OTLP logs, metrics, and spans. */ +export function agentAttrs(agentName: string, agentType: SessionAgentType | "unknown") { + return { + agent: agentName, + "agent.name": agentName, + "agent.type": agentType, + } as const +} diff --git a/src/v2/v2.ts b/src/v2/v2.ts new file mode 100644 index 0000000..8902cd5 --- /dev/null +++ b/src/v2/v2.ts @@ -0,0 +1,40 @@ +// Minimal local typings for the V2 server event contract consumed by this plugin. +// Shapes are taken from @opencode/schema (v2.0.x) and verified against live payloads. + +export type V2Tokens = { + input: number + output: number + reasoning: number + cache: { read: number; write: number } +} + +export type V2Error = { type: string; message: string; status?: number } + +export type V2Model = { id: string; providerID: string; variant?: string } + +export type V2ToolContent = { type: "text"; text: string } | { type: "file"; uri: string; mime: string; name?: string } + +/** A single decoded V2 server event, as yielded by `ctx.event.subscribe()`. */ +export type V2Event = { + id?: string + type: string + created: number + data: Record +} + +/** Total billed tokens for a usage sample (excludes cache reads/writes). */ +export function totalTokens(tokens: V2Tokens | undefined): number { + if (!tokens) return 0 + return (tokens.input ?? 0) + (tokens.output ?? 0) + (tokens.reasoning ?? 0) +} + +/** Formats a V2 model ref as `provider/id`. */ +export function modelRef(model: V2Model | undefined): string { + return model ? `${model.providerID}/${model.id}` : "unknown" +} + +/** Human-readable summary of a V2 error payload. */ +export function errorSummary(error: V2Error | undefined): string { + if (!error) return "unknown" + return `${error.type}: ${error.message}` +} diff --git a/tests/redact.test.ts b/tests/redact.test.ts new file mode 100644 index 0000000..c876ff5 --- /dev/null +++ b/tests/redact.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, test } from "bun:test" +import { redactSecrets } from "../src/redact.ts" + +describe("redactSecrets", () => { + test("leaves ordinary text untouched", () => { + const text = "Refactor the parser and add tests for the new branch." + expect(redactSecrets(text)).toBe(text) + }) + + test("masks known token prefixes", () => { + expect(redactSecrets("token sk-abcdefghijklmnop123456")).toBe("token [REDACTED]") + expect(redactSecrets("pylf_v2_eu_abcdef0123456789")).toBe("[REDACTED]") + expect(redactSecrets("ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789")).toBe("[REDACTED]") + expect(redactSecrets("AKIAIOSFODNN7EXAMPLE")).toBe("[REDACTED]") + }) + + test("masks authorization headers and JWTs", () => { + expect(redactSecrets("Authorization: Bearer abcdef123456")).toBe("Authorization: [REDACTED]") + expect( + redactSecrets("eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTYifQ.dozjgNryP4J3jVmNHl0w5N_XgL0n3I9PlFUP0THsR8U"), + ).toBe("[REDACTED]") + }) + + test("masks secret-looking key/value pairs", () => { + expect(redactSecrets("OPENAI_API_KEY=sk-proj-abcdefghijklmnop")).toBe("OPENAI_API_KEY=[REDACTED]") + expect(redactSecrets("password: hunter2")).toBe("password: [REDACTED]") + }) + + test("masks exact configured values even without a recognisable shape", () => { + expect(redactSecrets("my token is XADwrfwe2323ef32r23r2 ok", ["XADwrfwe2323ef32r23r2"])).toBe( + "my token is [REDACTED] ok", + ) + }) +}) From 6808031382386a0cba76343032c88b4c3baeb3ef Mon Sep 17 00:00:00 2001 From: Alex Hawat <1254687+alexhawat@users.noreply.github.com> Date: Wed, 23 Sep 2026 18:40:28 +0200 Subject: [PATCH 2/2] fix(v2): nest subagent traces under the parent dispatch tool span Subagent sessions produced orphan root run spans and dangling parents: the run span was created on the prompt hook (before session.created) and always used the root context. The run span is now created on session.execution.started, nests under the subagent session span, and is marked is_subagent; the subagent session span parents to the parent's opencode.tool.subagent span; session spans end on execution end. --- src/v2/handlers/session.ts | 74 +++++++++++++++++++++++++++++--------- src/v2/handlers/tool.ts | 5 +++ src/v2/index.ts | 14 ++++---- src/v2/types.ts | 1 + src/v2/util.ts | 13 ++++--- 5 files changed, 79 insertions(+), 28 deletions(-) diff --git a/src/v2/handlers/session.ts b/src/v2/handlers/session.ts index 6a46f44..7618055 100644 --- a/src/v2/handlers/session.ts +++ b/src/v2/handlers/session.ts @@ -1,5 +1,5 @@ import { SeverityNumber } from "@opentelemetry/api-logs" -import { SpanStatusCode } from "@opentelemetry/api" +import { SpanStatusCode, trace } from "@opentelemetry/api" import { AGENT_NAME, INPUT_MIME_TYPE, @@ -92,10 +92,15 @@ export function handleRunStarted( ctx.activeRuns.set(sessionID, runID) const safePrompt = ctx.redact(promptText) if (!isTraceEnabled("session", ctx)) return + const totals = ctx.sessionTotals.get(sessionID) + const agentType: SessionAgentType | "unknown" = totals?.agentType ?? "primary" + const isSubagent = agentType === "subagent" const existing = ctx.runSpans.get(runID) if (existing) { existing.setAttributes({ [AGENT_NAME]: agent, + "agent.type": agentType, + "session.is_subagent": isSubagent, ...(promptText ? { [INPUT_VALUE]: safePrompt, @@ -116,8 +121,8 @@ export function handleRunStarted( [OPENINFERENCE_SPAN_KIND]: OpenInferenceSpanKind.AGENT, [SESSION_ID]: sessionID, [AGENT_NAME]: agent, - "agent.type": "primary", - "session.is_subagent": false, + "agent.type": agentType, + "session.is_subagent": isSubagent, ...(promptText ? { [INPUT_VALUE]: safePrompt, @@ -129,7 +134,11 @@ export function handleRunStarted( ...ctx.commonAttrs, }, }, - ctx.rootContext(), + // Subagent turns nest under their session span (itself under the parent's dispatch + // tool span); primary turns resolve to the root context. + ctx.sessionSpans.get(sessionID) + ? trace.setSpan(ctx.rootContext(), ctx.sessionSpans.get(sessionID)!) + : ctx.rootContext(), ) ctx.runSpans.set(runID, runSpan) setBoundedMap(ctx.runSpanContexts, runID, runSpan.spanContext()) @@ -152,6 +161,12 @@ export function handleSessionCreated( }) if (isTraceEnabled("session", ctx) && data.parentID) { + // Nest under the parent's subagent-dispatch tool span when available, so the whole + // subagent subtree hangs off `opencode.tool.subagent`. + const dispatchSpan = ctx.pendingSubagentSpans.get(data.parentID) + const parentContext = dispatchSpan + ? trace.setSpan(ctx.rootContext(), dispatchSpan) + : resolveSessionTraceContext(data.parentID, ctx) const sessionSpan = ctx.tracer.startSpan( `${ctx.tracePrefix}session`, { @@ -166,7 +181,7 @@ export function handleSessionCreated( ...ctx.commonAttrs, }, }, - resolveSessionTraceContext(data.parentID, ctx), + parentContext, ) ctx.sessionSpans.set(sessionID, sessionSpan) setBoundedMap(ctx.sessionSpanContexts, sessionID, sessionSpan.spanContext()) @@ -222,14 +237,42 @@ function endRunSpan(sessionID: string, ctx: HandlerContext, error?: string) { if (runID) ctx.runSpans.delete(runID) } -/** V2 emits this per turn; ensures totals exist and starts a run span if the prompt hook missed it. */ +/** Ends a subagent session span (created on `session.created`) so it exports at turn end. */ +function endSessionSpan(sessionID: string, ctx: HandlerContext, error?: string) { + const sessionSpan = ctx.sessionSpans.get(sessionID) + if (!sessionSpan) return + const totals = ctx.sessionTotals.get(sessionID) + if (totals) { + sessionSpan.setAttributes({ + [AGENT_NAME]: totals.agent, + "agent.type": totals.agentType, + "session.total_tokens": totals.tokens, + "session.total_cost_usd": totals.cost, + "session.total_messages": totals.messages, + }) + } + if (error) { + sessionSpan.setStatus({ code: SpanStatusCode.ERROR, message: error }) + sessionSpan.setAttribute("error", error) + } else { + sessionSpan.setStatus({ code: SpanStatusCode.OK }) + } + sessionSpan.end() + ctx.sessionSpans.delete(sessionID) +} + +/** V2 emits this per turn; creates the run span (ordered after `session.created`). */ export function handleExecutionStarted(data: { sessionID: string }, ctx: HandlerContext) { const sessionID = data.sessionID ensureSessionTotals(sessionID, "unknown", "primary", Date.now(), ctx) - if (!ctx.activeRuns.get(sessionID)) { - const agent = ctx.sessionTotals.get(sessionID)?.agent ?? "unknown" - handleRunStarted(`${sessionID}:exec:${Date.now()}`, sessionID, agent, "", "unknown", Date.now(), ctx) - } + if (ctx.activeRuns.get(sessionID)) return + const meta = ctx.sessionMeta.get(sessionID) + const totals = ctx.sessionTotals.get(sessionID) + const agent = meta?.agent ?? totals?.agent ?? "unknown" + const model = meta?.model ?? "unknown" + const promptText = ctx.runPrompts.get(sessionID) ?? "" + // runID = sessionID: one run span per turn, ended (and replaced) on execution end. + handleRunStarted(sessionID, sessionID, agent, promptText, model, Date.now(), ctx) } /** V2 emits this when a turn completes; ends the run span. */ @@ -242,6 +285,10 @@ export function handleExecutionEnded( const totals = ctx.sessionTotals.get(sessionID) const { agentName, agentType } = getSessionAgentMeta(sessionID, ctx) endRunSpan(sessionID, ctx, error ? errorSummary(error) : undefined) + endSessionSpan(sessionID, ctx, error ? errorSummary(error) : undefined) + // Clear per-turn state so the next turn starts a fresh run span and prompt log. + ctx.promptEmitted.delete(sessionID) + ctx.runPrompts.delete(sessionID) const attrs = { ...ctx.commonAttrs, "session.id": sessionID } if (totals) { @@ -300,12 +347,7 @@ export function handleSessionIdle(sessionID: string, ctx: HandlerContext) { } } endRunSpan(sessionID, ctx) - const sessionSpan = ctx.sessionSpans.get(sessionID) - if (sessionSpan) { - sessionSpan.setStatus({ code: SpanStatusCode.OK }) - sessionSpan.end() - ctx.sessionSpans.delete(sessionID) - } + endSessionSpan(sessionID, ctx) ctx.emitLog({ severityNumber: SeverityNumber.INFO, diff --git a/src/v2/handlers/tool.ts b/src/v2/handlers/tool.ts index 926a1f8..bbc2328 100644 --- a/src/v2/handlers/tool.ts +++ b/src/v2/handlers/tool.ts @@ -15,6 +15,8 @@ import { errorSummary, type V2Error, type V2ToolContent } from "../v2.ts" const OPENINFERENCE_SPAN_KIND = SemanticConventions.OPENINFERENCE_SPAN_KIND const TOOL_NAME = "tool.name" +/** Tools that spawn a child (subagent) session; their span parents the subagent's tree. */ +const SUBAGENT_TOOL = /^(subagent|task)$/i /** Remembers the tool name for a call id, since V2 carries it only on `session.tool.input.started`. */ export function handleToolInputStarted(data: { sessionID: string; id: string; name?: string }, ctx: HandlerContext) { @@ -46,6 +48,8 @@ export function handleToolCalled( }), ) ctx.pendingToolSpans.set(data.id, { tool, sessionID: data.sessionID, startMs: Date.now(), span }) + // Let a subagent session created during this call nest under the dispatch span. + if (SUBAGENT_TOOL.test(tool)) setBoundedMap(ctx.pendingSubagentSpans, data.sessionID, span) } /** Ends the tool span successfully and records its duration. */ @@ -100,5 +104,6 @@ function finishTool( ctx.pendingToolSpans.delete(id) ctx.pendingToolNames.delete(id) + if (SUBAGENT_TOOL.test(tool)) ctx.pendingSubagentSpans.delete(sessionID) ctx.log("debug", error ? "otel: tool.failed" : "otel: tool.success", { sessionID, tool, duration }) } diff --git a/src/v2/index.ts b/src/v2/index.ts index 1b41e9a..9918c2f 100644 --- a/src/v2/index.ts +++ b/src/v2/index.ts @@ -1,5 +1,5 @@ import { logs } from "@opentelemetry/api-logs" -import { diag, DiagLogLevel, ROOT_CONTEXT, trace } from "@opentelemetry/api" +import { diag, DiagLogLevel, ROOT_CONTEXT, trace, type Span } from "@opentelemetry/api" import { appendFileSync, mkdirSync, statSync, writeFileSync } from "node:fs" import { homedir } from "node:os" import { dirname, join } from "node:path" @@ -12,7 +12,6 @@ import { handleExecutionEnded, handleExecutionFailed, handleExecutionStarted, - handleRunStarted, handleSessionCreated, handleSessionIdle, handleSessionStatus, @@ -81,6 +80,7 @@ const TRACING_KEY = "__opencode_otel_tracing_v1__" type TracingState = { seenEvents: Set + pendingSubagentSpans: Map runPrompts: Map promptEmitted: Set pendingToolSpans: HandlerContext["pendingToolSpans"] @@ -100,6 +100,7 @@ type TracingState = { function emptyTracing(): TracingState { return { seenEvents: new Set(), + pendingSubagentSpans: new Map(), runPrompts: new Map(), promptEmitted: new Set(), pendingToolSpans: new Map(), @@ -268,6 +269,7 @@ export default { commonAttrs, pendingToolSpans: tracing.pendingToolSpans, pendingToolNames: tracing.pendingToolNames, + pendingSubagentSpans: tracing.pendingSubagentSpans, sessionTotals: tracing.sessionTotals, sessionMeta: tracing.sessionMeta, disabledMetrics: config.disabledMetrics, @@ -330,12 +332,10 @@ export default { diagLog(`prompt enrich failed session=${sessionID}: ${err instanceof Error ? err.message : String(err)}`) } } - const agent = meta?.agent ?? hctx.sessionTotals.get(sessionID)?.agent ?? "unknown" - const model = meta?.model ?? "unknown" const promptText: string = event?.prompt?.text ?? "" - handleRunStarted(event.messageID, sessionID, agent, promptText, model, Date.now(), hctx) - // The user_prompt log is emitted on the first step, where agent/model are resolved. - setBoundedMap(hctx.runPrompts, event.messageID, promptText) + // The run span is created on `session.execution.started` — an event ordered after + // `session.created` — so a subagent's run span can nest under its session span. + setBoundedMap(hctx.runPrompts, sessionID, promptText) }), ) diff --git a/src/v2/types.ts b/src/v2/types.ts index 8d0be87..e69516c 100644 --- a/src/v2/types.ts +++ b/src/v2/types.ts @@ -87,6 +87,7 @@ export type HandlerContext = { commonAttrs: CommonAttrs pendingToolSpans: Map pendingToolNames: Map + pendingSubagentSpans: Map sessionTotals: Map sessionMeta: Map disabledMetrics: Set diff --git a/src/v2/util.ts b/src/v2/util.ts index 27958a0..b1efda2 100644 --- a/src/v2/util.ts +++ b/src/v2/util.ts @@ -55,17 +55,20 @@ export function resolveSessionTraceContext( input?: { assistantMessageID?: string; runID?: string }, ) { const baseCtx = ctx.rootContext() - const sessionSpan = ctx.sessionSpans.get(sessionID) - if (sessionSpan) return trace.setSpan(baseCtx, sessionSpan) - const sessionSpanContext = ctx.sessionSpanContexts.get(sessionID) - if (sessionSpanContext) return trace.setSpanContext(baseCtx, sessionSpanContext) + // Prefer the active turn run span so LLM/tool spans nest under it; the run span itself is + // nested under the subagent session span by handleRunStarted. if (input?.runID) return resolveRunTraceContext(input.runID, ctx) const assistantRunID = input?.assistantMessageID ? ctx.assistantRuns.get(input.assistantMessageID) : undefined if (assistantRunID) return resolveRunTraceContext(assistantRunID, ctx) const activeRunID = ctx.activeRuns.get(sessionID) - return activeRunID ? resolveRunTraceContext(activeRunID, ctx) : baseCtx + if (activeRunID) return resolveRunTraceContext(activeRunID, ctx) + const sessionSpan = ctx.sessionSpans.get(sessionID) + if (sessionSpan) return trace.setSpan(baseCtx, sessionSpan) + const sessionSpanContext = ctx.sessionSpanContexts.get(sessionID) + if (sessionSpanContext) return trace.setSpanContext(baseCtx, sessionSpanContext) + return baseCtx } /**