diff --git a/bun.lock b/bun.lock index eaf91dd7e..a500e71f2 100644 --- a/bun.lock +++ b/bun.lock @@ -89,9 +89,14 @@ "@oh-my-pi/pi-wire": "catalog:", "@oh-my-pi/snapcompact": "catalog:", "@opentelemetry/api": "catalog:", + "@opentelemetry/api-logs": "catalog:", "@opentelemetry/context-async-hooks": "catalog:", + "@opentelemetry/exporter-logs-otlp-proto": "catalog:", + "@opentelemetry/exporter-metrics-otlp-proto": "catalog:", "@opentelemetry/exporter-trace-otlp-proto": "catalog:", "@opentelemetry/resources": "catalog:", + "@opentelemetry/sdk-logs": "catalog:", + "@opentelemetry/sdk-metrics": "catalog:", "@opentelemetry/sdk-trace-base": "catalog:", "@opentelemetry/sdk-trace-node": "catalog:", "@puppeteer/browsers": "catalog:", @@ -377,11 +382,16 @@ "@oh-my-pi/pi-wire": "17.0.1", "@oh-my-pi/snapcompact": "17.0.1", "@opentelemetry/api": "^1.9.1", - "@opentelemetry/context-async-hooks": "^2.7.1", + "@opentelemetry/api-logs": "^0.220.0", + "@opentelemetry/context-async-hooks": "^2.9.0", + "@opentelemetry/exporter-logs-otlp-proto": "^0.220.0", + "@opentelemetry/exporter-metrics-otlp-proto": "^0.220.0", "@opentelemetry/exporter-trace-otlp-proto": "^0.220.0", - "@opentelemetry/resources": "^2.7.1", - "@opentelemetry/sdk-trace-base": "^2.7.1", - "@opentelemetry/sdk-trace-node": "^2.7.1", + "@opentelemetry/resources": "^2.9.0", + "@opentelemetry/sdk-logs": "^0.220.0", + "@opentelemetry/sdk-metrics": "^2.9.0", + "@opentelemetry/sdk-trace-base": "^2.9.0", + "@opentelemetry/sdk-trace-node": "^2.9.0", "@puppeteer/browsers": "^3.0.6", "@tailwindcss/node": "^4.3.2", "@tailwindcss/vite": "^4.3.2", @@ -800,6 +810,12 @@ "@opentelemetry/core": ["@opentelemetry/core@2.9.0", "", { "dependencies": { "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.0.0 <1.10.0" } }, "sha512-m2nckMT80NnmjTYSPjJQObBJ+8dgkoajEOUbznL8AHZ3T3yHRk2P7gI1PhEBc1+lOnrYE9UWrWHqJDsmqjmNbw=="], + "@opentelemetry/exporter-logs-otlp-proto": ["@opentelemetry/exporter-logs-otlp-proto@0.220.0", "", { "dependencies": { "@opentelemetry/otlp-exporter-base": "0.220.0", "@opentelemetry/otlp-transformer": "0.220.0", "@opentelemetry/sdk-logs": "0.220.0" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-8LZAxdJ0ENDAFwr4j0oY35mHBltiSzvlhdQAPGiC7p9VnxtuSq4SW1gfBAdW6t6hiQG6OwUl8w7KHaOdJPKHWg=="], + + "@opentelemetry/exporter-metrics-otlp-http": ["@opentelemetry/exporter-metrics-otlp-http@0.220.0", "", { "dependencies": { "@opentelemetry/core": "2.9.0", "@opentelemetry/otlp-exporter-base": "0.220.0", "@opentelemetry/otlp-transformer": "0.220.0", "@opentelemetry/resources": "2.9.0", "@opentelemetry/sdk-metrics": "2.9.0" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-Yqt3RBw/bRVncaE9qIIhk4WfjbAQqXuP9FgAaU+IKPndnLEp/cUqZlSC324+bpmduRz7DoTjig8Ub0PeILWXUA=="], + + "@opentelemetry/exporter-metrics-otlp-proto": ["@opentelemetry/exporter-metrics-otlp-proto@0.220.0", "", { "dependencies": { "@opentelemetry/core": "2.9.0", "@opentelemetry/exporter-metrics-otlp-http": "0.220.0", "@opentelemetry/otlp-exporter-base": "0.220.0", "@opentelemetry/otlp-transformer": "0.220.0", "@opentelemetry/resources": "2.9.0", "@opentelemetry/sdk-metrics": "2.9.0" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-lyO+IQBdSvqHN/ZOW/OzrSWemtfD+HgWngn+HBNLhjy0YrCQQTz0OE/kSekH2Pl340dn9DWzhqHdz5Eftr+HLA=="], + "@opentelemetry/exporter-trace-otlp-proto": ["@opentelemetry/exporter-trace-otlp-proto@0.220.0", "", { "dependencies": { "@opentelemetry/core": "2.9.0", "@opentelemetry/otlp-exporter-base": "0.220.0", "@opentelemetry/otlp-transformer": "0.220.0", "@opentelemetry/resources": "2.9.0", "@opentelemetry/sdk-trace": "2.9.0" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-voTAD8XgJxlK7zLkXh8EzMB09zrQr3tyY/BsnDTlDiQU/UdK58MZ63A3mUjdEDrxMjCVmBHU3WQJhRmQe+Dvzg=="], "@opentelemetry/otlp-exporter-base": ["@opentelemetry/otlp-exporter-base@0.220.0", "", { "dependencies": { "@opentelemetry/core": "2.9.0", "@opentelemetry/otlp-transformer": "0.220.0" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-CXYo8UD5Mn9YbgebO2EL4wejtA+gxLmLiu6HCk2KH2BR7XhFN6/6p1UlCb23DYCjeYkndevLHuejCCN1yx4+OQ=="], diff --git a/package.json b/package.json index 6c1cbb930..21163bf70 100644 --- a/package.json +++ b/package.json @@ -39,11 +39,16 @@ "@oh-my-pi/pi-wire": "17.0.1", "@oh-my-pi/snapcompact": "17.0.1", "@opentelemetry/api": "^1.9.1", - "@opentelemetry/context-async-hooks": "^2.7.1", + "@opentelemetry/api-logs": "^0.220.0", + "@opentelemetry/context-async-hooks": "^2.9.0", + "@opentelemetry/exporter-logs-otlp-proto": "^0.220.0", + "@opentelemetry/exporter-metrics-otlp-proto": "^0.220.0", "@opentelemetry/exporter-trace-otlp-proto": "^0.220.0", - "@opentelemetry/resources": "^2.7.1", - "@opentelemetry/sdk-trace-base": "^2.7.1", - "@opentelemetry/sdk-trace-node": "^2.7.1", + "@opentelemetry/resources": "^2.9.0", + "@opentelemetry/sdk-logs": "^0.220.0", + "@opentelemetry/sdk-metrics": "^2.9.0", + "@opentelemetry/sdk-trace-base": "^2.9.0", + "@opentelemetry/sdk-trace-node": "^2.9.0", "@puppeteer/browsers": "^3.0.6", "@tailwindcss/node": "^4.3.2", "@tailwindcss/vite": "^4.3.2", diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 58093b9f9..3abc94925 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -10,6 +10,7 @@ - Added native Warp CLI-agent events for rich session status, tool approvals, and completion notifications ([#5592](https://github.com/can1357/oh-my-pi/pull/5592) by [@metaphorics](https://github.com/metaphorics)). - Added Codex (ChatGPT subscription) support to `generate_image`. The tool now resolves a connected `openai-codex` OAuth credential and drives OpenAI's hosted `image_generation` tool through the ChatGPT backend (`chatgpt.com/backend-api/codex/responses`, `chatgpt-account-id` header) **independent of the active chat model** — so image generation works on a ChatGPT/Codex subscription with no metered `OPENAI_API_KEY`, even when the active model is Claude/Gemini/etc. A new `providers.image: "openai-codex"` option forces it; `auto` now auto-detects a connected subscription (priority: active GPT image tool > Codex subscription > Antigravity > xAI > OpenRouter > Gemini), and the `openai` preference falls back to it when no `OPENAI_API_KEY`/active GPT model is present. - Added an optional `provider` parameter to `generate_image` (`auto` | `openai` | `openai-codex` | `antigravity` | `xai` | `gemini` | `openrouter`) that overrides the `providers.image` setting **for a single request** — so "generate this using gemini / codex / xai" routes per-call without changing the global setting. Absent → the `providers.image` setting applies, unchanged; the named provider uses the same resolution semantics (falls back to auto-detect if it has no credentials). File: `tools/image-gen.ts` (`imageProviderSchema`, `findImageApiKey` `preference` arg). +- Added OpenTelemetry log and metric export alongside the existing trace export. When `OTEL_EXPORTER_OTLP_LOGS_ENDPOINT` (or the shared `OTEL_EXPORTER_OTLP_ENDPOINT`) is set, `omp` registers a `LoggerProvider` and forwards every centralized-logger event as an OTLP log record (severity + attributes + active span context for log↔trace correlation, min level via `OTEL_LOG_LEVEL`, plus a structured `agent run completed` summary event). When `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` (or the shared endpoint) is set, it registers a `MeterProvider` with a `PeriodicExportingMetricReader` and records GenAI-semconv `gen_ai.client.token.usage` plus `pi.omp.agent.*` counters/histograms (runs, steps, chat/tool calls by name+status+finish reason, latencies, estimated cost, errors) from the agent run summary and per-chat usage hooks. Each signal honors its own `OTEL_*_EXPORTER=none` kill switch, the global `OTEL_SDK_DISABLED`, and declines non-`http/protobuf` protocols independently ([#4604](https://github.com/can1357/oh-my-pi/issues/4604)). - `retry.fallbackChains` wildcards now support id-prefixed targets and keys: a chain entry like `"openrouter/google/*"` re-prefixes the failing model's bare id (`google-antigravity/gemini-x` → `openrouter/google/gemini-x`), a plain `"provider/*"` entry falling back *from* an aggregator strips the vendor prefix when the target provider only knows the bare id (`openrouter/google/x` → `google-vertex/x`), and an id-prefixed key (`"openrouter/google/*"`) scopes a chain to that provider's ids under the prefix. ### Changed diff --git a/packages/coding-agent/package.json b/packages/coding-agent/package.json index 6c7f8f1d5..8ff91928e 100644 --- a/packages/coding-agent/package.json +++ b/packages/coding-agent/package.json @@ -64,9 +64,14 @@ "@oh-my-pi/pi-wire": "catalog:", "@oh-my-pi/snapcompact": "catalog:", "@opentelemetry/api": "catalog:", + "@opentelemetry/api-logs": "catalog:", "@opentelemetry/context-async-hooks": "catalog:", + "@opentelemetry/exporter-logs-otlp-proto": "catalog:", + "@opentelemetry/exporter-metrics-otlp-proto": "catalog:", "@opentelemetry/exporter-trace-otlp-proto": "catalog:", "@opentelemetry/resources": "catalog:", + "@opentelemetry/sdk-logs": "catalog:", + "@opentelemetry/sdk-metrics": "catalog:", "@opentelemetry/sdk-trace-base": "catalog:", "@opentelemetry/sdk-trace-node": "catalog:", "@puppeteer/browsers": "catalog:", diff --git a/packages/coding-agent/src/main.ts b/packages/coding-agent/src/main.ts index 134948162..e47d64e6e 100644 --- a/packages/coding-agent/src/main.ts +++ b/packages/coding-agent/src/main.ts @@ -77,7 +77,7 @@ import { executeBuiltinSlashCommand } from "./slash-commands/builtin-registry"; import { shouldShowStartupSplash } from "./startup-splash"; import { discoverTitleSystemPromptFile, resolvePromptInput } from "./system-prompt"; import { createPersistedSubagentReviverFactory } from "./task/persisted-revive"; -import { initTelemetryExport, isTelemetryExportEnabled } from "./telemetry-export"; +import { createTelemetryExportConfig, initTelemetryExport, isTelemetryExportEnabled } from "./telemetry-export"; import { concreteThinkingLevel, parseConfiguredThinkingLevel } from "./thinking"; import type { LspStartupServerInfo } from "./tools"; import { @@ -1351,15 +1351,13 @@ export async function runRootCommand( sessionOptions.hasUI = isInteractive || mode === "rpc-ui"; sessionOptions.settings = settingsInstance; - // OTEL: register the global OTLP trace exporter when an OTLP endpoint is - // configured via env, then switch on the agent loop's telemetry so its - // GenAI spans (invoke_agent / chat / execute_tool) are actually emitted. - // Both are no-ops when OTEL_EXPORTER_OTLP_ENDPOINT is unset. An empty config - // is enough to enable telemetry — content capture is governed by the - // standard OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT env var. + // OTEL: register global OTLP exporters when an endpoint is configured via + // env, then switch on the agent loop's telemetry hooks so traces, run-level + // metrics, and structured logs have source events to export. Content capture + // remains governed by OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT. await logger.time("initTelemetryExport", initTelemetryExport); if (isTelemetryExportEnabled()) { - sessionOptions.telemetry = {}; + sessionOptions.telemetry = createTelemetryExportConfig(sessionOptions.telemetry); } // Handle CLI --api-key as runtime override (not persisted) diff --git a/packages/coding-agent/src/telemetry-export.ts b/packages/coding-agent/src/telemetry-export.ts index 234d43cf3..0dad0ecc7 100644 --- a/packages/coding-agent/src/telemetry-export.ts +++ b/packages/coding-agent/src/telemetry-export.ts @@ -1,144 +1,500 @@ /** - * OTLP trace export bootstrap. + * OTLP telemetry export bootstrap. * * oh-my-pi's agent core (`@oh-my-pi/pi-agent-core`) emits OpenTelemetry GenAI - * spans through the global `@opentelemetry/api` tracer, but only when a - * TracerProvider is registered in the process — otherwise the API returns a - * no-op tracer and the spans are silently dropped. The shipped CLI never - * registered one, so headless / embedded hosts (e.g. an ACP harness that - * spawns `omp` as a child process) had no way to collect omp's internal traces. + * spans through the global `@opentelemetry/api` tracer, and exposes run-level + * callbacks for metrics/log pipelines. This module registers the OTLP/proto + * trace, log, and metric SDK providers when the standard `OTEL_*` endpoint env + * vars are set so `omp` can be observed by any OTLP collector without vendor + * coupling. * - * This module registers a NodeTracerProvider with an OTLP/proto exporter when - * the standard `OTEL_EXPORTER_OTLP_ENDPOINT` (or `..._TRACES_ENDPOINT`) env var - * is set, following the zero-code OTEL env contract: the exporter reads its - * endpoint, headers, and timeout from `OTEL_EXPORTER_OTLP_*` itself. The - * consuming process configures the destination entirely through env; omp stays - * provider-agnostic and ships no vendor coupling. Only the `http/protobuf` - * transport is supported — an `OTEL_EXPORTER_OTLP*_PROTOCOL` of `grpc` or - * `http/json` declines rather than misrouting spans. - * - * The OTLP/proto exporter on the 2.x line is used deliberately: the 1.x line - * deadlocks under Bun — its `req.on('close')` handler fires a spurious failure - * after the success path. `exporter-trace-otlp-proto@0.218` paired with - * `sdk-trace-base@2.7` exports cleanly on Bun. + * Only the `http/protobuf` transport is supported — an + * `OTEL_EXPORTER_OTLP*_PROTOCOL` of `grpc` or `http/json` declines rather than + * misrouting protobuf payloads. The exporter line is pinned to the 0.218/2.7 + * family validated under Bun; the 1.x OTLP line deadlocks when its + * `req.on("close")` handler fires after a successful export. */ +import type { + AgentRunCoverage, + AgentRunSummary, + AgentTelemetryConfig, + AgentTelemetryWarning, + ChatUsageEvent, + ToolStatus, +} from "@oh-my-pi/pi-agent-core"; import { logger, postmortem } from "@oh-my-pi/pi-utils"; -import type * as TraceNode from "@opentelemetry/sdk-trace-node"; +import { + type Attributes, + type AttributeValue, + type Counter, + context, + type Histogram, + type Meter, + metrics, +} from "@opentelemetry/api"; +import { type LogAttributes, logs, type Logger as OtelLogger, SeverityNumber } from "@opentelemetry/api-logs"; +import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks"; +import { OTLPLogExporter } from "@opentelemetry/exporter-logs-otlp-proto"; +import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-proto"; +import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-proto"; +import { resourceFromAttributes } from "@opentelemetry/resources"; +import { BatchLogRecordProcessor, LoggerProvider } from "@opentelemetry/sdk-logs"; +import { MeterProvider, PeriodicExportingMetricReader } from "@opentelemetry/sdk-metrics"; +import { BatchSpanProcessor } from "@opentelemetry/sdk-trace-base"; +import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node"; /** * Periodic flush interval. A long-lived `omp` process (the ACP server is * spawned once and reused across many turns) would otherwise hold finished - * spans until the batch window elapses or the process exits. + * telemetry until a batch window elapses or the process exits. */ const FLUSH_INTERVAL_MS = 30_000; -let provider: TraceNode.NodeTracerProvider | undefined; +const SERVICE_NAME = "oh-my-pi"; + +type TelemetrySignal = "trace" | "log" | "metric"; +type OtelLogLevel = "none" | logger.LogLevel; + +interface SignalConfig { + readonly trace: boolean; + readonly log: boolean; + readonly metric: boolean; +} + +const LOG_SEVERITY: Record = { + error: SeverityNumber.ERROR, + warn: SeverityNumber.WARN, + info: SeverityNumber.INFO, + debug: SeverityNumber.DEBUG, +}; + +const LOG_LEVEL_WEIGHT: Record = { + error: 0, + warn: 1, + info: 2, + debug: 3, +}; + +const TOOL_STATUSES = ["ok", "error", "skipped", "blocked", "timeout", "aborted"] satisfies readonly ToolStatus[]; + +let traceProvider: NodeTracerProvider | undefined; +let logProvider: LoggerProvider | undefined; +let meterProvider: MeterProvider | undefined; +let metricRecorder: AgentMetricRecorder | undefined; +let otelLogger: OtelLogger | undefined; +let unregisterLogSink: (() => void) | undefined; let initPromise: Promise | undefined; /** - * Whether {@link initTelemetryExport} registered a real provider. The CLI uses - * this to decide whether to switch on the agent loop's telemetry config — there - * is no point emitting spans into a no-op tracer. + * Whether {@link initTelemetryExport} registered any real OTLP signal provider. + * The CLI uses this to decide whether to switch on the agent loop's telemetry + * hooks; metrics and structured logs need those callbacks even when traces are + * disabled. */ export function isTelemetryExportEnabled(): boolean { - return provider !== undefined; + if (traceProvider) return true; + if (logProvider) return true; + if (meterProvider) return true; + return false; } /** - * Register the global TracerProvider + OTLP exporter when an OTLP endpoint is - * configured via env. Idempotent, and a no-op when no endpoint is set (or when - * the OTEL kill-switches are engaged), so it is safe to call unconditionally at - * startup. + * Merge OTLP metrics/log hooks into an existing agent telemetry config. + * + * The caller still owns content-capture policy, cost estimation, and custom + * attributes. This only appends host-level metrics/log forwarding for the + * providers registered by {@link initTelemetryExport}. + */ +export function createTelemetryExportConfig( + config: AgentTelemetryConfig | undefined, +): AgentTelemetryConfig | undefined { + if (!isTelemetryExportEnabled()) return config; + return { + ...config, + onChatUsage: async event => { + await config?.onChatUsage?.(event); + metricRecorder?.recordChatUsage(event); + }, + onRunEnd: (summary, coverage) => { + config?.onRunEnd?.(summary, coverage); + metricRecorder?.recordRun(summary, coverage); + emitRunSummaryLog(summary, coverage); + }, + onTelemetryWarning: warning => { + config?.onTelemetryWarning?.(warning); + emitTelemetryWarningLog(warning); + }, + }; +} + +/** + * Register global trace/log/meter providers when OTLP endpoints are configured + * through env. Idempotent, and a no-op when no signal has an endpoint (or when + * the OTEL kill-switches are engaged), so startup can call it unconditionally. */ export async function initTelemetryExport(): Promise { - if (provider) return; + if (isTelemetryExportEnabled()) return; if (initPromise) return initPromise; - // The OTEL env contract parses booleans and enum lists case-insensitively, so - // OTEL_SDK_DISABLED=TRUE and OTEL_TRACES_EXPORTER=None must also disable export. if (process.env.OTEL_SDK_DISABLED?.trim().toLowerCase() === "true") return; - if (tracesExporterDisabled(process.env.OTEL_TRACES_EXPORTER)) return; - const endpoint = process.env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT ?? process.env.OTEL_EXPORTER_OTLP_ENDPOINT; - if (!endpoint) return; + const signalConfig = resolveSignalConfig(); + if (!signalConfig.trace && !signalConfig.log && !signalConfig.metric) return; - // We only ship the http/protobuf transport (the line validated on Bun). The - // OTEL contract lets OTEL_EXPORTER_OTLP*_PROTOCOL select grpc / http/json; - // rather than silently send protobuf-over-HTTP to a grpc :4317 port and lose - // every span, decline when an unsupported protocol is requested. - const protocol = (process.env.OTEL_EXPORTER_OTLP_TRACES_PROTOCOL ?? process.env.OTEL_EXPORTER_OTLP_PROTOCOL) - ?.trim() - .toLowerCase(); - if (protocol && protocol !== "http/protobuf") { - logger.warn( - `OTEL trace export disabled: OTEL_EXPORTER_OTLP_PROTOCOL=${protocol} is unsupported (only http/protobuf)`, - ); - return; - } - - initPromise = registerProvider(); + initPromise = registerProviders(signalConfig); return initPromise; } -async function registerProvider(): Promise { - const [ - { AsyncLocalStorageContextManager }, - { OTLPTraceExporter }, - { resourceFromAttributes }, - { BatchSpanProcessor }, - { NodeTracerProvider }, - ] = await Promise.all([ - import("@opentelemetry/context-async-hooks"), - import("@opentelemetry/exporter-trace-otlp-proto"), - import("@opentelemetry/resources"), - import("@opentelemetry/sdk-trace-base"), - import("@opentelemetry/sdk-trace-node"), - ]); - - // The exporter reads endpoint/headers/timeout from OTEL_EXPORTER_OTLP_* itself, - // so there is nothing to thread through here. - const exporter = new OTLPTraceExporter(); - const tracerProvider = new NodeTracerProvider({ - resource: resourceFromAttributes({ - "service.name": process.env.OTEL_SERVICE_NAME ?? "oh-my-pi", - }), - spanProcessors: [new BatchSpanProcessor(exporter)], +async function registerProviders(signalConfig: SignalConfig): Promise { + const resource = resourceFromAttributes({ + "service.name": process.env.OTEL_SERVICE_NAME ?? SERVICE_NAME, }); - // register() installs the global tracer provider and the W3C trace-context + - // baggage propagators; the explicit AsyncLocalStorage context manager keeps - // parent/child span linkage working under Bun. - tracerProvider.register({ contextManager: new AsyncLocalStorageContextManager().enable() }); - provider = tracerProvider; + + if (signalConfig.trace) { + const exporter = new OTLPTraceExporter(); + traceProvider = new NodeTracerProvider({ + resource, + spanProcessors: [new BatchSpanProcessor(exporter)], + }); + traceProvider.register({ contextManager: new AsyncLocalStorageContextManager().enable() }); + } + + if (signalConfig.metric) { + const exporter = new OTLPMetricExporter(); + meterProvider = new MeterProvider({ + resource, + readers: [new PeriodicExportingMetricReader({ exporter })], + }); + metrics.setGlobalMeterProvider(meterProvider); + metricRecorder = new AgentMetricRecorder(metrics.getMeter("@oh-my-pi/pi-coding-agent")); + } + + if (signalConfig.log) { + const exporter = new OTLPLogExporter(); + logProvider = new LoggerProvider({ + resource, + processors: [new BatchLogRecordProcessor({ exporter })], + }); + logs.setGlobalLoggerProvider(logProvider); + otelLogger = logProvider.getLogger("@oh-my-pi/pi-coding-agent"); + unregisterLogSink = logger.registerLogSink(event => { + emitOtelLog( + event.level, + event.message, + logAttributesFromContext(event.context), + "pi.omp.log", + event.timestamp, + ); + }); + } const flushTimer = setInterval(() => { - provider?.forceFlush().catch(() => {}); + flushTelemetryExport().catch(() => {}); }, FLUSH_INTERVAL_MS); flushTimer.unref(); - // Shut down through postmortem rather than a bare signal listener. postmortem - // owns SIGINT/SIGTERM/SIGHUP/exit and quit(), and awaits registered cleanups - // before calling process.exit — so the batch processor's final OTLP export - // completes instead of being cut off mid-flight on the shutdown path. - postmortem.register("otel-trace-export", async () => { + postmortem.register("otel-export", async () => { clearInterval(flushTimer); - await provider?.shutdown(); + unregisterLogSink?.(); + unregisterLogSink = undefined; + const shutdowns: Promise[] = []; + if (traceProvider) shutdowns.push(traceProvider.shutdown()); + if (logProvider) shutdowns.push(logProvider.shutdown()); + if (meterProvider) shutdowns.push(meterProvider.shutdown()); + await Promise.all(shutdowns); }); } -/** - * Parse the `OTEL_TRACES_EXPORTER` selection. The value is a case-insensitive, - * comma-separated list; the literal `none` disables span export entirely. - */ -function tracesExporterDisabled(raw: string | undefined): boolean { - if (!raw) return false; - return raw.split(",").some(entry => entry.trim().toLowerCase() === "none"); +function resolveSignalConfig(): SignalConfig { + const signalConfig: SignalConfig = { + trace: signalEnabled( + "trace", + process.env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT ?? process.env.OTEL_EXPORTER_OTLP_ENDPOINT, + process.env.OTEL_TRACES_EXPORTER, + process.env.OTEL_EXPORTER_OTLP_TRACES_PROTOCOL ?? process.env.OTEL_EXPORTER_OTLP_PROTOCOL, + ), + log: signalEnabled( + "log", + process.env.OTEL_EXPORTER_OTLP_LOGS_ENDPOINT ?? process.env.OTEL_EXPORTER_OTLP_ENDPOINT, + process.env.OTEL_LOGS_EXPORTER, + process.env.OTEL_EXPORTER_OTLP_LOGS_PROTOCOL ?? process.env.OTEL_EXPORTER_OTLP_PROTOCOL, + ), + metric: signalEnabled( + "metric", + process.env.OTEL_EXPORTER_OTLP_METRICS_ENDPOINT ?? process.env.OTEL_EXPORTER_OTLP_ENDPOINT, + process.env.OTEL_METRICS_EXPORTER, + process.env.OTEL_EXPORTER_OTLP_METRICS_PROTOCOL ?? process.env.OTEL_EXPORTER_OTLP_PROTOCOL, + ), + }; + return signalConfig; +} + +function signalEnabled( + signal: TelemetrySignal, + endpoint: string | undefined, + exporterSelection: string | undefined, + protocolSelection: string | undefined, +): boolean { + if (exporterSelection) { + for (const entry of exporterSelection.split(",")) { + if (entry.trim().toLowerCase() === "none") return false; + } + } + if (!endpoint) return false; + + const protocol = protocolSelection?.trim().toLowerCase(); + if (protocol && protocol !== "http/protobuf") { + logger.warn(`OTEL ${signal} export disabled: OTEL_EXPORTER_OTLP_PROTOCOL=${protocol} is unsupported`, { + supported: "http/protobuf", + }); + return false; + } + return true; +} + +class AgentMetricRecorder { + readonly #tokenUsage: Histogram; + readonly #chatCostUsd: Counter; + readonly #runs: Counter; + readonly #steps: Counter; + readonly #chatCalls: Counter; + readonly #chatDurationMs: Histogram; + readonly #toolCalls: Counter; + readonly #toolDurationMs: Histogram; + readonly #errors: Counter; + + constructor(meter: Meter) { + this.#tokenUsage = meter.createHistogram("gen_ai.client.token.usage", { + description: "Token usage reported by GenAI chat calls.", + unit: "{token}", + }); + this.#chatCostUsd = meter.createCounter("pi.omp.agent.chat.cost.estimated_usd", { + description: "Estimated USD cost for completed chat calls.", + unit: "USD", + }); + this.#runs = meter.createCounter("pi.omp.agent.runs", { + description: "Completed agent runs.", + unit: "{run}", + }); + this.#steps = meter.createCounter("pi.omp.agent.steps", { + description: "Agent loop steps completed inside a run.", + unit: "{step}", + }); + this.#chatCalls = meter.createCounter("pi.omp.agent.chat.calls", { + description: "Chat calls completed inside agent runs.", + unit: "{call}", + }); + this.#chatDurationMs = meter.createHistogram("pi.omp.agent.chat.duration", { + description: "Total chat latency observed in an agent run.", + unit: "ms", + }); + this.#toolCalls = meter.createCounter("pi.omp.agent.tool.calls", { + description: "Tool calls completed inside agent runs.", + unit: "{call}", + }); + this.#toolDurationMs = meter.createHistogram("pi.omp.agent.tool.duration", { + description: "Total tool latency observed in an agent run.", + unit: "ms", + }); + this.#errors = meter.createCounter("pi.omp.agent.errors", { + description: "Errors observed in chat and tool execution.", + unit: "{error}", + }); + } + + recordChatUsage(event: ChatUsageEvent): void { + const baseAttrs = metricAttributes({ + "gen_ai.operation.name": "chat", + "gen_ai.provider.name": event.provider, + "gen_ai.request.model": event.model, + "gen_ai.response.service_tier": event.serviceTier, + "pi.gen_ai.agent.id": event.agent?.id, + "pi.gen_ai.agent.name": event.agent?.name, + }); + + this.#recordToken(event.usage.inputTokens, baseAttrs, "input"); + this.#recordToken(event.usage.outputTokens, baseAttrs, "output"); + this.#recordToken(event.usage.totalTokens, baseAttrs, "total"); + this.#recordToken(event.usage.cachedInputTokens, baseAttrs, "cache_read_input"); + this.#recordToken(event.usage.cacheWriteTokens, baseAttrs, "cache_write_input"); + this.#recordToken(event.usage.reasoningOutputTokens, baseAttrs, "reasoning_output"); + + if (event.cost && "usd" in event.cost && event.cost.usd > 0) { + this.#chatCostUsd.add(event.cost.usd, baseAttrs); + } + } + + recordRun(summary: AgentRunSummary, coverage: AgentRunCoverage): void { + const runAttrs = metricAttributes({ + "pi.omp.agent.models_used.count": coverage.modelsUsed.length, + "pi.omp.agent.providers_used.count": coverage.providersUsed.length, + "pi.omp.agent.tools_available.count": coverage.toolsAvailable.length, + "pi.omp.agent.tools_invoked.count": coverage.toolsInvoked.length, + "pi.omp.agent.tools_unused.count": coverage.toolsUnused.length, + }); + + this.#runs.add(1, runAttrs); + if (summary.stepCount > 0) this.#steps.add(summary.stepCount, runAttrs); + if (summary.chats.totalLatencyMs > 0) this.#chatDurationMs.record(summary.chats.totalLatencyMs, runAttrs); + + for (const reason in summary.chats.byStopReason) { + const count = summary.chats.byStopReason[reason]; + if (count > 0) + this.#chatCalls.add(count, metricAttributes({ ...runAttrs, "gen_ai.response.finish_reason": reason })); + } + for (const toolName in summary.tools.byName) { + const counters = summary.tools.byName[toolName]; + const toolAttrs = metricAttributes({ ...runAttrs, "gen_ai.tool.name": toolName }); + if (counters.totalLatencyMs > 0) this.#toolDurationMs.record(counters.totalLatencyMs, toolAttrs); + for (const status of TOOL_STATUSES) { + const count = counters[status]; + if (count > 0) this.#toolCalls.add(count, metricAttributes({ ...toolAttrs, "pi.omp.tool.status": status })); + } + } + for (const errorType in summary.errors.byType) { + const count = summary.errors.byType[errorType]; + if (count > 0) this.#errors.add(count, metricAttributes({ ...runAttrs, "error.type": errorType })); + } + } + + #recordToken(value: number | undefined, baseAttrs: Attributes, tokenType: string): void { + if (!value || value <= 0) return; + this.#tokenUsage.record(value, metricAttributes({ ...baseAttrs, "gen_ai.token.type": tokenType })); + } +} + +function metricAttributes(fields: Readonly>): Attributes { + const out: Attributes = {}; + for (const key in fields) { + const value = fields[key]; + if (value === undefined || value === null) continue; + if (typeof value === "string" || typeof value === "number" || typeof value === "boolean") { + out[key] = value; + continue; + } + const text = String(value); + if (text.length > 0) out[key] = text; + } + return out; +} + +function emitRunSummaryLog(summary: AgentRunSummary, coverage: AgentRunCoverage): void { + emitOtelLog( + "info", + "agent run completed", + { + "pi.omp.agent.step_count": summary.stepCount, + "pi.omp.agent.chats.total": summary.chats.total, + "pi.omp.agent.chats.total_latency_ms": summary.chats.totalLatencyMs, + "pi.omp.agent.tools.total": summary.tools.total, + "pi.omp.agent.tools.ok": summary.tools.ok, + "pi.omp.agent.tools.error": summary.tools.error, + "pi.omp.agent.tools.skipped": summary.tools.skipped, + "pi.omp.agent.tools.blocked": summary.tools.blocked, + "pi.omp.agent.tools.timeout": summary.tools.timeout, + "pi.omp.agent.tools.aborted": summary.tools.aborted, + "pi.omp.agent.tools.total_latency_ms": summary.tools.totalLatencyMs, + "pi.omp.agent.usage.input_tokens": summary.usage.inputTokens, + "pi.omp.agent.usage.output_tokens": summary.usage.outputTokens, + "pi.omp.agent.usage.cached_input_tokens": summary.usage.cachedInputTokens, + "pi.omp.agent.usage.cache_write_tokens": summary.usage.cacheWriteTokens, + "pi.omp.agent.usage.reasoning_output_tokens": summary.usage.reasoningOutputTokens, + "pi.omp.agent.usage.total_tokens": summary.usage.totalTokens, + "pi.omp.agent.cost.estimated_usd": summary.cost.estimatedUsd, + "pi.omp.agent.cost.unavailable_reasons": summary.cost.unavailableReasons.join(","), + "pi.omp.agent.errors.total": summary.errors.total, + "pi.omp.agent.coverage.tools_available": coverage.toolsAvailable.join(","), + "pi.omp.agent.coverage.tools_invoked": coverage.toolsInvoked.join(","), + "pi.omp.agent.coverage.tools_unused": coverage.toolsUnused.join(","), + "pi.omp.agent.coverage.models_used": coverage.modelsUsed.join(","), + "pi.omp.agent.coverage.providers_used": coverage.providersUsed.join(","), + }, + "pi.omp.agent.run.completed", + ); +} + +function emitTelemetryWarningLog(warning: AgentTelemetryWarning): void { + const attrs = logAttributesFromContext({ + code: warning.code, + error: warning.error, + }); + emitOtelLog("warn", warning.message, attrs, "pi.omp.telemetry.warning"); +} + +function emitOtelLog( + level: logger.LogLevel, + body: string, + attributes: LogAttributes, + eventName: string, + timestamp = new Date(), +): void { + if (!otelLogger) return; + const minLevel = parseOtelLogLevel(process.env.OTEL_LOG_LEVEL); + if (minLevel === "none") return; + if (LOG_LEVEL_WEIGHT[level] > LOG_LEVEL_WEIGHT[minLevel]) return; + otelLogger.emit({ + eventName, + timestamp, + observedTimestamp: new Date(), + severityNumber: LOG_SEVERITY[level], + severityText: level.toUpperCase(), + body, + attributes, + context: context.active(), + }); +} + +function parseOtelLogLevel(raw: string | undefined): OtelLogLevel { + if (!raw) return "info"; + switch (raw.trim().toLowerCase()) { + case "none": + return "none"; + case "error": + return "error"; + case "warn": + case "warning": + return "warn"; + case "debug": + return "debug"; + default: + return "info"; + } +} + +function logAttributesFromContext(input: Record | undefined): LogAttributes { + const out: LogAttributes = { "process.pid": process.pid }; + if (!input) return out; + for (const key in input) { + const attr = logAttributeValue(input[key]); + if (attr !== undefined) out[key] = attr; + } + return out; +} + +function logAttributeValue(value: unknown): AttributeValue | undefined { + if (value === undefined || value === null) return undefined; + if (typeof value === "string" || typeof value === "number" || typeof value === "boolean") return value; + if (value instanceof Error) { + return `${value.name}: ${value.message}`; + } + try { + const text = JSON.stringify(value); + if (text && text.length > 0) return text; + } catch { + return String(value); + } + return String(value); } /** - * Flush any buffered spans to the exporter. No-op when export is disabled. + * Flush buffered spans, log records, and metrics. No-op when export is disabled. * Hosts embedding the agent can call this at natural boundaries (e.g. the end - * of a turn) so traces surface promptly rather than on the batch interval. + * of a turn) so telemetry surfaces promptly rather than on the batch interval. */ export async function flushTelemetryExport(): Promise { - await provider?.forceFlush(); + const flushes: Promise[] = []; + if (traceProvider) flushes.push(traceProvider.forceFlush()); + if (logProvider) flushes.push(logProvider.forceFlush()); + if (meterProvider) flushes.push(meterProvider.forceFlush()); + await Promise.all(flushes); } diff --git a/packages/coding-agent/test/otel-signals-probe.ts b/packages/coding-agent/test/otel-signals-probe.ts new file mode 100644 index 000000000..d9e9d6950 --- /dev/null +++ b/packages/coding-agent/test/otel-signals-probe.ts @@ -0,0 +1,209 @@ +/** + * Positive-path probe for the OTLP log + metric exporters, run as a subprocess + * by telemetry-export.test.ts. Keeping it out-of-process means the global + * LoggerProvider / MeterProvider singletons that initTelemetryExport() registers + * never leak into the test runner. + * + * Stands up a loopback OTLP/proto receiver, points the standard env vars at it, + * registers the providers, drives a log record through the bridged + * `@oh-my-pi/pi-utils` logger and metric instruments through the agent + * telemetry hooks, flushes, and exits 0 only if the receiver got a non-empty + * protobuf POST at both /v1/logs and /v1/metrics. + */ + +import type { AgentRunCoverage, AgentRunSummary, ChatUsageEvent } from "@oh-my-pi/pi-agent-core"; +import { emptyAgentRunCoverage, emptyAgentRunSummary } from "@oh-my-pi/pi-agent-core"; +import { + createTelemetryExportConfig, + flushTelemetryExport, + initTelemetryExport, + isTelemetryExportEnabled, +} from "@oh-my-pi/pi-coding-agent/telemetry-export"; +import { logger } from "@oh-my-pi/pi-utils"; + +const seen = new Set(); +const metricPayloads: Uint8Array[] = []; + +interface ProtobufField { + readonly number: number; + readonly bytes?: Uint8Array; +} + +function readVarint(bytes: Uint8Array, offset: number): [number, number] { + let value = 0; + let shift = 0; + while (offset < bytes.length) { + const byte = bytes[offset++]; + value += (byte & 0x7f) * 2 ** shift; + if ((byte & 0x80) === 0) return [value, offset]; + shift += 7; + } + throw new Error("Truncated protobuf varint"); +} + +function protobufFields(bytes: Uint8Array): ProtobufField[] { + const fields: ProtobufField[] = []; + for (let offset = 0; offset < bytes.length; ) { + const [tag, nextOffset] = readVarint(bytes, offset); + offset = nextOffset; + const wireType = tag & 7; + const number = tag >>> 3; + if (wireType === 0) { + [, offset] = readVarint(bytes, offset); + fields.push({ number }); + } else if (wireType === 1) { + offset += 8; + fields.push({ number }); + } else if (wireType === 2) { + const [length, valueOffset] = readVarint(bytes, offset); + offset = valueOffset; + const end = offset + length; + if (end > bytes.length) throw new Error("Truncated protobuf field"); + fields.push({ number, bytes: bytes.slice(offset, end) }); + offset = end; + } else if (wireType === 5) { + offset += 4; + fields.push({ number }); + } else { + throw new Error(`Unsupported protobuf wire type ${wireType}`); + } + } + return fields; +} + +function pointCountForMetric(bytes: Uint8Array, metricName: string): number | undefined { + const fields = protobufFields(bytes); + const isMetric = fields.some( + field => field.number === 1 && field.bytes && new TextDecoder().decode(field.bytes) === metricName, + ); + if (isMetric) { + const aggregation = fields.find(field => field.number === 7 || field.number === 9)?.bytes; + if (!aggregation) return undefined; + return protobufFields(aggregation).filter(field => field.number === 1).length; + } + for (const field of fields) { + if (!field.bytes) continue; + try { + const count = pointCountForMetric(field.bytes, metricName); + if (count !== undefined) return count; + } catch { + // This length-delimited field is a scalar string or bytes value, not a nested message. + } + } + return undefined; +} + +function assertSingleMetricPoint(metricName: string): void { + const counts = metricPayloads.map(payload => pointCountForMetric(payload, metricName)); + if (!counts.includes(1)) { + throw new Error(`${metricName} expected one dimensioned point, got ${counts.join(",")}`); + } +} + + +const server = Bun.serve({ + port: 0, + async fetch(req) { + const path = new URL(req.url).pathname; + if (req.method === "POST" && req.headers.get("content-type")?.startsWith("application/x-protobuf")) { + const body = await req.arrayBuffer(); + if (path.endsWith("/v1/metrics")) metricPayloads.push(new Uint8Array(body)); + if (body.byteLength > 0) { + if (path.endsWith("/v1/logs")) seen.add("logs"); + if (path.endsWith("/v1/metrics")) seen.add("metrics"); + } + } + return new Response('{"partialSuccess":{}}', { + status: 200, + headers: { "content-type": "application/json" }, + }); + }, +}); + +const base = `http://localhost:${server.port}`; +process.env.OTEL_EXPORTER_OTLP_LOGS_ENDPOINT = `${base}/v1/logs`; +process.env.OTEL_EXPORTER_OTLP_METRICS_ENDPOINT = `${base}/v1/metrics`; +process.env.OTEL_SERVICE_NAME = "oh-my-pi-signals-probe"; +// Force a short metric export interval so the periodic reader flushes fast. +process.env.OTEL_METRIC_EXPORT_INTERVAL = "500"; + +await initTelemetryExport(); +if (!isTelemetryExportEnabled()) { + console.error("PROBE: providers did not register"); + await server.stop(true); + process.exit(2); +} + +const config = createTelemetryExportConfig(undefined); +if (!config) { + console.error("PROBE: export config not produced"); + await server.stop(true); + process.exit(2); +} + +// Bridged utility logger -> OTel log record. +logger.error("probe error", { code: "probe" }); + +// Metric instruments via the agent telemetry hooks. +const usage: ChatUsageEvent = { + span: undefined as never, + agent: { id: "main", name: "Main" }, + conversationId: "probe-session", + stepNumber: 0, + model: "claude-haiku-4-5", + provider: "anthropic", + serviceTier: undefined, + usage: { + inputTokens: 1000, + outputTokens: 200, + totalTokens: 1200, + cachedInputTokens: 0, + cacheWriteTokens: 0, + reasoningOutputTokens: 0, + }, + cost: { usd: 0.01 }, + attributes: undefined, + headers: undefined, +}; +await config.onChatUsage?.(usage); + +const summary: AgentRunSummary = { + ...emptyAgentRunSummary(), + chats: { total: 1, byStopReason: { end_turn: 1 }, totalLatencyMs: 1500 }, + tools: { + total: 1, + ok: 1, + error: 0, + skipped: 0, + blocked: 0, + timeout: 0, + aborted: 0, + totalLatencyMs: 42, + byName: { + read: { total: 1, ok: 1, error: 0, skipped: 0, blocked: 0, timeout: 0, aborted: 0, totalLatencyMs: 42 }, + }, + }, + stepCount: 1, +}; +const coverage: AgentRunCoverage = { + ...emptyAgentRunCoverage(), + toolsAvailable: ["read", "write"], + toolsInvoked: ["read"], + toolsUnused: ["write"], + modelsUsed: ["claude-haiku-4-5"], + providersUsed: ["anthropic"], +}; +config.onRunEnd?.(summary, coverage); + +await flushTelemetryExport(); +// The metric reader exports on its own interval; wait one cycle then flush. +await Bun.sleep(700); +await flushTelemetryExport(); +assertSingleMetricPoint("pi.omp.agent.chat.calls"); +assertSingleMetricPoint("pi.omp.agent.tool.calls"); +assertSingleMetricPoint("pi.omp.agent.tool.duration"); +await server.stop(true); + +const ok = seen.has("logs") && seen.has("metrics"); +console.log(ok ? "PROBE: RECEIVED" : `PROBE: MISSING ${["logs", "metrics"].filter(s => !seen.has(s)).join(",")}`); +process.exit(ok ? 0 : 1); diff --git a/packages/coding-agent/test/telemetry-export.test.ts b/packages/coding-agent/test/telemetry-export.test.ts index 2ec90acdb..cdc924a5e 100644 --- a/packages/coding-agent/test/telemetry-export.test.ts +++ b/packages/coding-agent/test/telemetry-export.test.ts @@ -11,10 +11,16 @@ import { initTelemetryExport, isTelemetryExportEnabled } from "@oh-my-pi/pi-codi const OTEL_KEYS = [ "OTEL_EXPORTER_OTLP_ENDPOINT", "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", + "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT", + "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", "OTEL_EXPORTER_OTLP_PROTOCOL", "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL", + "OTEL_EXPORTER_OTLP_LOGS_PROTOCOL", + "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL", "OTEL_SDK_DISABLED", "OTEL_TRACES_EXPORTER", + "OTEL_LOGS_EXPORTER", + "OTEL_METRICS_EXPORTER", ] as const; let saved: Record; @@ -45,8 +51,8 @@ describe("initTelemetryExport gating", () => { expect(isTelemetryExportEnabled()).toBe(false); }); - it("stays disabled when OTEL_TRACES_EXPORTER=none even with an endpoint", async () => { - process.env.OTEL_EXPORTER_OTLP_ENDPOINT = "http://localhost:4318"; + it("stays disabled when OTEL_TRACES_EXPORTER=none and only the traces endpoint is set", async () => { + process.env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT = "http://localhost:4318"; process.env.OTEL_TRACES_EXPORTER = "none"; await initTelemetryExport(); expect(isTelemetryExportEnabled()).toBe(false); @@ -71,12 +77,23 @@ describe("initTelemetryExport gating", () => { delete process.env.OTEL_SDK_DISABLED; process.env.OTEL_TRACES_EXPORTER = "otlp,None"; + process.env.OTEL_LOGS_EXPORTER = "none"; + process.env.OTEL_METRICS_EXPORTER = "none"; + await initTelemetryExport(); + expect(isTelemetryExportEnabled()).toBe(false); + }); + + it("stays disabled when every signal exporter is set to none", async () => { + process.env.OTEL_EXPORTER_OTLP_ENDPOINT = "http://localhost:4318"; + process.env.OTEL_TRACES_EXPORTER = "none"; + process.env.OTEL_LOGS_EXPORTER = "none"; + process.env.OTEL_METRICS_EXPORTER = "none"; await initTelemetryExport(); expect(isTelemetryExportEnabled()).toBe(false); }); }); -describe("initTelemetryExport export path", () => { +describe("initTelemetryExport signals export path", () => { it("registers a provider and exports spans to an OTLP/proto receiver", async () => { // Run in a subprocess: initTelemetryExport() registers a process-global // provider, so exercising the positive path in-process would leak that @@ -88,4 +105,15 @@ describe("initTelemetryExport export path", () => { expect(stdout).toContain("PROBE: RECEIVED"); expect(code).toBe(0); }, 20_000); + + it("exports log records and metrics to OTLP/proto receivers", async () => { + // Same subprocess isolation as the trace probe: the logs/metrics probe + // drives the bridged logger and the agent telemetry metric hooks, then + // asserts protobuf POSTs landed at both /v1/logs and /v1/metrics. + const probe = fileURLToPath(new URL("./otel-signals-probe.ts", import.meta.url)); + const proc = Bun.spawn(["bun", probe], { stdout: "pipe", stderr: "pipe" }); + const [code, stdout] = await Promise.all([proc.exited, new Response(proc.stdout).text()]); + expect(stdout).toContain("PROBE: RECEIVED"); + expect(code).toBe(0); + }, 20_000); }); diff --git a/packages/utils/CHANGELOG.md b/packages/utils/CHANGELOG.md index e6cf17de3..d44a7416d 100644 --- a/packages/utils/CHANGELOG.md +++ b/packages/utils/CHANGELOG.md @@ -2,6 +2,9 @@ ## [Unreleased] +### Added + +- Added a structured log sink API to the centralized logger (`registerLogSink`, `LogEvent`, `LogLevel`) so out-of-band consumers (e.g. OpenTelemetry log export) receive every `error`/`warn`/`info`/`debug` event after the local transport path runs, without disturbing existing file/console logging ([#4604](https://github.com/can1357/oh-my-pi/issues/4604)). ### Fixed - Fixed fatal cleanup failing to reach `process.exit()` when terminal stderr is revoked, and isolated rotating log files/audit state per process to prevent concurrent OMP instances from racing compression and rotation ([#5716](https://github.com/can1357/oh-my-pi/issues/5716)). diff --git a/packages/utils/src/logger.ts b/packages/utils/src/logger.ts index 17641a298..3d0e7933f 100644 --- a/packages/utils/src/logger.ts +++ b/packages/utils/src/logger.ts @@ -17,6 +17,42 @@ import winston from "winston"; import DailyRotateFile from "winston-daily-rotate-file"; import { getLogsDir } from "./dirs"; import { drainModuleLoadEvents } from "./timing-buffer"; +/** Severity names accepted by the centralized logger. */ +export type LogLevel = "error" | "warn" | "info" | "debug"; + +/** Structured log event forwarded to out-of-band sinks such as OpenTelemetry. */ +export interface LogEvent { + readonly level: LogLevel; + readonly message: string; + readonly context: Record | undefined; + readonly timestamp: Date; +} + +/** Receives each structured log event after the local transport path runs. */ +export type LogSink = (event: LogEvent) => void; + +const logSinks = new Set(); + +/** Register an out-of-band log sink and return a disposer. */ +export function registerLogSink(sink: LogSink): () => void { + logSinks.add(sink); + return () => { + logSinks.delete(sink); + }; +} + +function emitToSinks(level: LogLevel, message: string, context: Record | undefined): void { + if (logSinks.size === 0) return; + const event: LogEvent = { level, message, context, timestamp: new Date() }; + for (const sink of logSinks) { + try { + sink(event); + } catch { + // Sinks are side channels; they must never break local logging. + } + } +} + const PROCESS_LOG_PATTERN = /^omp\.\d{4}-\d{2}-\d{2}\.(\d+)\.log(?:\.\d+)?$/; const PROCESS_AUDIT_PATTERN = /^\.omp\.(\d+)-audit\.json$/; @@ -212,6 +248,7 @@ export function error(message: string, context?: Record): void } catch { // Silently ignore logging failures } + emitToSinks("error", message, context); } /** @@ -225,6 +262,7 @@ export function warn(message: string, context?: Record): void { } catch { // Silently ignore logging failures } + emitToSinks("warn", message, context); } /** @@ -238,6 +276,7 @@ export function info(message: string, context?: Record): void { } catch { // Silently ignore logging failures } + emitToSinks("info", message, context); } /** @@ -251,6 +290,7 @@ export function debug(message: string, context?: Record): void } catch { // Silently ignore logging failures } + emitToSinks("debug", message, context); } /**