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..7618055 --- /dev/null +++ b/src/v2/handlers/session.ts @@ -0,0 +1,391 @@ +import { SeverityNumber } from "@opentelemetry/api-logs" +import { SpanStatusCode, trace } 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 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, + [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": agentType, + "session.is_subagent": isSubagent, + ...(promptText + ? { + [INPUT_VALUE]: safePrompt, + [INPUT_MIME_TYPE]: MimeType.TEXT, + [LLM_INPUT_MESSAGES]: JSON.stringify([{ role: "user", content: safePrompt }]), + } + : {}), + model, + ...ctx.commonAttrs, + }, + }, + // 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()) +} + +/** 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) { + // 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`, + { + 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, + }, + }, + parentContext, + ) + 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) +} + +/** 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)) 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. */ +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) + 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) { + 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) + endSessionSpan(sessionID, ctx) + + 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..bbc2328 --- /dev/null +++ b/src/v2/handlers/tool.ts @@ -0,0 +1,109 @@ +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" +/** 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) { + 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 }) + // 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. */ +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) + 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/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..9918c2f --- /dev/null +++ b/src/v2/index.ts @@ -0,0 +1,443 @@ +import { logs } from "@opentelemetry/api-logs" +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" +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, + 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 + pendingSubagentSpans: Map + 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(), + pendingSubagentSpans: new Map(), + 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, + pendingSubagentSpans: tracing.pendingSubagentSpans, + 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 promptText: string = event?.prompt?.text ?? "" + // 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) + }), + ) + + // 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..e69516c --- /dev/null +++ b/src/v2/types.ts @@ -0,0 +1,111 @@ +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 + pendingSubagentSpans: 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..b1efda2 --- /dev/null +++ b/src/v2/util.ts @@ -0,0 +1,132 @@ +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() + // 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) + 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 +} + +/** + * 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", + ) + }) +})