From d7182b69e4fafed8b2ab9bcdb074eed939e493a9 Mon Sep 17 00:00:00 2001 From: Sanchit Rk <0xsanchit@gmail.com> Date: Tue, 13 Oct 2026 15:23:19 +0530 Subject: [PATCH 1/6] feat: export OpenTelemetry traces and metrics over OTLP Every CLI command can export the harness's spans and metrics through Effect's built-in OTLP exporter, configured only by the standard OTEL_* environment variables. Without an endpoint nothing is exported, and a configuration that fails to build runs the command without export. - harness: GenAI attributes on the LLM, tool and agent spans; a failed provider request now ends its span with an error status. - harness: metrics for token usage, request/tool/agent duration, time to first token, time per output token, cost, retries, sandbox provisioning and tool output truncation, plus Effect's fiber runtime metrics. - cli: per-signal http/protobuf (default) or http/json, OTLP exporters default to otlp as the spec says, and resource attributes for the instance, release channel and command. - E2E test against a fake provider and collector, snapshotted to __artifacts__/telemetry.json. Co-Authored-By: Claude Opus 5.5 --- packages/codework/package.json | 1 + packages/codework/src/framework/otel.ts | 92 +++++ packages/codework/src/framework/runtime.ts | 22 +- .../test/__artifacts__/telemetry.json | 287 +++++++++++++ packages/codework/test/telemetry.test.ts | 382 ++++++++++++++++++ packages/harness/src/runner/llm.ts | 114 +++++- packages/harness/src/runner/loop.ts | 97 ++++- packages/harness/src/sandbox/control.ts | 48 ++- packages/harness/src/telemetry/telemetry.ts | 105 +++++ pnpm-lock.yaml | 3 + 10 files changed, 1117 insertions(+), 34 deletions(-) create mode 100644 packages/codework/src/framework/otel.ts create mode 100644 packages/codework/test/__artifacts__/telemetry.json create mode 100644 packages/codework/test/telemetry.test.ts create mode 100644 packages/harness/src/telemetry/telemetry.ts diff --git a/packages/codework/package.json b/packages/codework/package.json index 6ec1e59b..e32f4f28 100644 --- a/packages/codework/package.json +++ b/packages/codework/package.json @@ -51,6 +51,7 @@ "@effect/platform-node": "4.0.1", "effect": "4.0.1", "jsonc-parser": "3.3.1", + "uuidv7": "^1.2.1", "zod": "4.6.5" }, "devDependencies": { diff --git a/packages/codework/src/framework/otel.ts b/packages/codework/src/framework/otel.ts new file mode 100644 index 00000000..f51b22f5 --- /dev/null +++ b/packages/codework/src/framework/otel.ts @@ -0,0 +1,92 @@ +import { Config, ConfigProvider, Effect, Layer, Metric, Option, Schema } from "effect"; +import { FetchHttpClient, type HttpClient } from "effect/http"; +import { OtlpExporter, OtlpMetrics, OtlpSerialization, OtlpTracer } from "effect/observability"; +import { uuidv7 } from "uuidv7"; + +/** + * OTLP export of the harness's spans and metrics, configured by the standard + * `OTEL_*` environment variables alone. Without an endpoint every layer is + * empty; a configuration that fails to build runs the command without export. + */ + +// The OTel spec defaults `OTEL_{TRACES,METRICS}_EXPORTER` to `otlp`; Effect reads an +// unset list as none, so an endpoint alone would export nothing. +const exporterDefaults = ConfigProvider.layerAdd( + ConfigProvider.fromUnknown({ OTEL_TRACES_EXPORTER: ["otlp"], OTEL_METRICS_EXPORTER: ["otlp"] }), +); + +// Effect's fiber metrics hook every fiber start and end, so they follow metrics export. +const metricsEnabled = Config.all({ + disabled: Config.Boolean("OTEL_SDK_DISABLED").pipe(Config.withDefault(false)), + endpoint: Config.String("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT").pipe( + Config.orElse(() => Config.String("OTEL_EXPORTER_OTLP_ENDPOINT")), + Config.option, + ), + exporters: Config.Array(Schema.String, "OTEL_METRICS_EXPORTER"), +}).pipe( + Config.map( + ({ disabled, endpoint, exporters }) => + !disabled && Option.isSome(endpoint) && exporters.some((exporter) => exporter.trim().toLowerCase() === "otlp"), + ), +); + +const fiberMetrics = Layer.unwrap( + Effect.gen(function* () { + return (yield* metricsEnabled) ? Metric.enableRuntimeMetricsLayer : Layer.empty; + }), +); + +/** + * Serialize each signal by its `OTEL_EXPORTER_OTLP_*PROTOCOL`, defaulting to the + * spec's `http/protobuf`. Effect has no gRPC exporter, so `grpc` exports nothing + * rather than sending HTTP to a gRPC port. + */ +const serialized = ( + signal: "TRACES" | "METRICS", + self: Layer.Layer, +): Layer.Layer => + Layer.unwrap( + Effect.gen(function* () { + const protocol = yield* Config.String(`OTEL_EXPORTER_OTLP_${signal}_PROTOCOL`).pipe( + Config.orElse(() => Config.String("OTEL_EXPORTER_OTLP_PROTOCOL")), + Config.withDefault("http/protobuf"), + ); + switch (protocol.trim().toLowerCase()) { + case "http/json": + return Layer.provide(self, OtlpSerialization.layerJson); + case "http/protobuf": + return Layer.provide(self, OtlpSerialization.layerProtobuf); + default: + return OtlpExporter.layerFlusher; + } + }), + ).pipe(Layer.orDie); + +/** The release channel a version was published to, from its prerelease tag. */ +const channel = (version: string) => + version.includes("-canary.") ? "canary" : version.includes("-dev.") ? "dev" : "latest"; + +const instanceId = uuidv7(); + +export const layer = (options: { readonly version: string; readonly command: string }) => { + const resource = { + serviceName: "codework", + serviceVersion: options.version, + attributes: { + "service.instance.id": instanceId, + "deployment.environment.name": channel(options.version), + "codework.command": options.command, + }, + }; + return Layer.mergeAll( + serialized("TRACES", OtlpTracer.layerFromConfig({ resource })), + serialized("METRICS", OtlpMetrics.layerFromConfig({ resource })), + fiberMetrics, + ).pipe( + Layer.provide(FetchHttpClient.layer), + Layer.provide(exporterDefaults), + Layer.catchCause(() => Layer.empty), + ); +}; + +export * as Otel from "./otel.ts"; diff --git a/packages/codework/src/framework/runtime.ts b/packages/codework/src/framework/runtime.ts index 370a05c7..4d25a3b3 100644 --- a/packages/codework/src/framework/runtime.ts +++ b/packages/codework/src/framework/runtime.ts @@ -1,6 +1,7 @@ import { Effect } from "effect"; import { Command } from "effect/cli"; import type { CommandError } from "../cli/error.ts"; +import { Otel } from "./otel.ts"; import { Spec } from "./spec.ts"; /** @@ -72,21 +73,34 @@ export function handlers( } export function run(root: Spec.Any, handlers: ReadonlyArray, options: { readonly version: string }) { - return Command.run(provide(root, handlers), options); + return Command.run(provide(root, handlers, { version: options.version, path: [] }), options); } -function provide(node: Spec.Any, handlers: ReadonlyArray): ProvidedCommand { +/** Each handler runs inside its own OTLP export, named by its command path below the root. */ +function provide( + node: Spec.Any, + handlers: ReadonlyArray, + context: { readonly version: string; readonly path: ReadonlyArray }, +): ProvidedCommand { const found = handlers.find((entry) => entry.spec === node.spec); + const command = context.path.length === 0 ? node.name : context.path.join(" "); const spec = found ? node.spec.pipe( Command.withHandler((input) => - Effect.promise(found.load).pipe(Effect.flatMap((module) => module.default(input))), + Effect.promise(found.load).pipe( + Effect.flatMap((module) => module.default(input)), + Effect.provide(Otel.layer({ version: context.version, command })), + ), ), ) : node.spec; const children = Object.values(node.commands); if (children.length === 0) return spec as ProvidedCommand; - return spec.pipe(Command.withSubcommands(children.map((child) => provide(child, handlers)))) as ProvidedCommand; + return spec.pipe( + Command.withSubcommands( + children.map((child) => provide(child, handlers, { ...context, path: [...context.path, child.name] })), + ), + ) as ProvidedCommand; } export * as Runtime from "./runtime.ts"; diff --git a/packages/codework/test/__artifacts__/telemetry.json b/packages/codework/test/__artifacts__/telemetry.json new file mode 100644 index 00000000..95e527fb --- /dev/null +++ b/packages/codework/test/__artifacts__/telemetry.json @@ -0,0 +1,287 @@ +{ + "resource": { + "codework.command": "run", + "deployment.environment.name": "latest", + "service.instance.id": "", + "service.name": "codework", + "service.version": "" + }, + "spans": [ + { + "name": "Loop.run", + "kind": "internal", + "parent": null, + "error": false, + "attributes": { + "gen_ai.operation.name": "invoke_agent" + } + }, + { + "name": "LLM.run", + "kind": "client", + "parent": "Loop.run", + "error": true, + "attributes": { + "error.type": "Runner.ProviderUnavailableError", + "gen_ai.operation.name": "chat", + "gen_ai.provider.name": "otel", + "gen_ai.request.model": "otel-model", + "gen_ai.response.finish_reasons": [ + "error" + ], + "gen_ai.response.model": "otel-model", + "gen_ai.usage.cache_creation.input_tokens": 0, + "gen_ai.usage.cache_read.input_tokens": 0, + "gen_ai.usage.input_tokens": 0, + "gen_ai.usage.output_tokens": 0 + } + }, + { + "name": "LLM.run", + "kind": "client", + "parent": "Loop.run", + "error": false, + "attributes": { + "gen_ai.operation.name": "chat", + "gen_ai.provider.name": "otel", + "gen_ai.request.model": "otel-model", + "gen_ai.response.finish_reasons": [ + "toolUse" + ], + "gen_ai.response.id": "", + "gen_ai.response.model": "otel-model", + "gen_ai.usage.cache_creation.input_tokens": 0, + "gen_ai.usage.cache_read.input_tokens": 0, + "gen_ai.usage.input_tokens": 120, + "gen_ai.usage.output_tokens": 8, + "gen_ai.usage.reasoning_tokens": 0 + } + }, + { + "name": "Loop.handleTool", + "kind": "internal", + "parent": "Loop.run", + "error": false, + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.call.id": "", + "gen_ai.tool.name": "read" + } + }, + { + "name": "LLM.run", + "kind": "client", + "parent": "Loop.run", + "error": false, + "attributes": { + "gen_ai.operation.name": "chat", + "gen_ai.provider.name": "otel", + "gen_ai.request.model": "otel-model", + "gen_ai.response.finish_reasons": [ + "stop" + ], + "gen_ai.response.id": "", + "gen_ai.response.model": "otel-model", + "gen_ai.usage.cache_creation.input_tokens": 0, + "gen_ai.usage.cache_read.input_tokens": 0, + "gen_ai.usage.input_tokens": 120, + "gen_ai.usage.output_tokens": 8, + "gen_ai.usage.reasoning_tokens": 0 + } + } + ], + "metrics": { + "child_fiber_failures": { + "unit": "1", + "type": "sum", + "series": [ + { + "attributes": {} + } + ] + }, + "child_fiber_successes": { + "unit": "1", + "type": "sum", + "series": [ + { + "attributes": {} + } + ] + }, + "child_fibers_active": { + "unit": "1", + "type": "gauge", + "series": [ + { + "attributes": {} + } + ] + }, + "child_fibers_started": { + "unit": "1", + "type": "sum", + "series": [ + { + "attributes": {} + } + ] + }, + "codework.llm.cost": { + "unit": "USD", + "type": "sum", + "series": [ + { + "attributes": { + "gen_ai.operation.name": "chat", + "gen_ai.provider.name": "otel", + "gen_ai.request.model": "otel-model", + "unit": "USD" + }, + "value": 0.00096 + } + ] + }, + "codework.llm.time_per_output_token": { + "unit": "s", + "type": "histogram", + "series": [ + { + "attributes": { + "gen_ai.operation.name": "chat", + "gen_ai.provider.name": "otel", + "gen_ai.request.model": "otel-model", + "unit": "s" + }, + "count": 2 + } + ] + }, + "codework.llm.time_to_first_token": { + "unit": "s", + "type": "histogram", + "series": [ + { + "attributes": { + "gen_ai.operation.name": "chat", + "gen_ai.provider.name": "otel", + "gen_ai.request.model": "otel-model", + "unit": "s" + }, + "count": 2 + } + ] + }, + "codework.tool.output.truncated": { + "unit": "{call}", + "type": "sum", + "series": [ + { + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "read", + "unit": "{call}" + }, + "value": 1 + } + ] + }, + "codework.turn.retries": { + "unit": "{retry}", + "type": "sum", + "series": [ + { + "attributes": { + "gen_ai.provider.name": "otel", + "gen_ai.request.model": "otel-model", + "unit": "{retry}" + }, + "value": 1 + } + ] + }, + "gen_ai.client.operation.duration": { + "unit": "s", + "type": "histogram", + "series": [ + { + "attributes": { + "error.type": "Runner.ProviderUnavailableError", + "gen_ai.operation.name": "chat", + "gen_ai.provider.name": "otel", + "gen_ai.request.model": "otel-model", + "unit": "s" + }, + "count": 1 + }, + { + "attributes": { + "gen_ai.operation.name": "chat", + "gen_ai.provider.name": "otel", + "gen_ai.request.model": "otel-model", + "unit": "s" + }, + "count": 2 + } + ] + }, + "gen_ai.client.token.usage": { + "unit": "{token}", + "type": "histogram", + "series": [ + { + "attributes": { + "gen_ai.operation.name": "chat", + "gen_ai.provider.name": "otel", + "gen_ai.request.model": "otel-model", + "gen_ai.token.type": "input", + "unit": "{token}" + }, + "count": 2, + "value": 240 + }, + { + "attributes": { + "gen_ai.operation.name": "chat", + "gen_ai.provider.name": "otel", + "gen_ai.request.model": "otel-model", + "gen_ai.token.type": "output", + "unit": "{token}" + }, + "count": 2, + "value": 16 + } + ] + }, + "gen_ai.execute_tool.duration": { + "unit": "s", + "type": "histogram", + "series": [ + { + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "read", + "outcome": "ok", + "unit": "s" + }, + "count": 1 + } + ] + }, + "gen_ai.invoke_agent.duration": { + "unit": "s", + "type": "histogram", + "series": [ + { + "attributes": { + "gen_ai.operation.name": "invoke_agent", + "gen_ai.provider.name": "otel", + "gen_ai.request.model": "otel-model", + "unit": "s" + }, + "count": 1 + } + ] + } + } +} diff --git a/packages/codework/test/telemetry.test.ts b/packages/codework/test/telemetry.test.ts new file mode 100644 index 00000000..00a71112 --- /dev/null +++ b/packages/codework/test/telemetry.test.ts @@ -0,0 +1,382 @@ +/* @effect-diagnostics nodeBuiltinImport:off -- this suite spawns the CLI and serves a fake provider and collector. */ +/* @effect-diagnostics globalTimers:off -- the subprocess deadline runs outside any Effect. */ +import { spawn } from "node:child_process"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { createServer, type IncomingMessage } from "node:http"; +import type { AddressInfo } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { afterAll, describe, expect, it } from "vite-plus/test"; + +const cli = fileURLToPath(new URL("../src/index.ts", import.meta.url)); +const models = fileURLToPath(new URL("../../../models.gen.json", import.meta.url)); +const temps: string[] = []; + +afterAll(() => { + for (const dir of temps) rmSync(dir, { recursive: true, force: true }); +}); + +interface OtlpValue { + readonly stringValue?: string; + readonly intValue?: number | string; + readonly doubleValue?: number; + readonly boolValue?: boolean; + readonly arrayValue?: { readonly values?: ReadonlyArray }; +} +interface OtlpAttribute { + readonly key: string; + readonly value: OtlpValue; +} +interface OtlpSpan { + readonly spanId: string; + readonly parentSpanId?: string; + readonly name: string; + readonly kind: number; + readonly startTimeUnixNano: string; + readonly attributes?: ReadonlyArray; + readonly events?: ReadonlyArray; + readonly status?: { readonly code?: number }; +} +interface OtlpDataPoint { + readonly attributes?: ReadonlyArray; + readonly asDouble?: number; + readonly asInt?: number | string; + readonly count?: number | string; + readonly sum?: number; +} +interface OtlpMetric { + readonly name: string; + readonly unit?: string; + readonly sum?: { readonly dataPoints: ReadonlyArray }; + readonly gauge?: { readonly dataPoints: ReadonlyArray }; + readonly histogram?: { readonly dataPoints: ReadonlyArray }; +} +interface Resource { + readonly resource?: { readonly attributes?: ReadonlyArray }; +} +interface Export { + readonly path: string; + readonly header: string | undefined; + readonly body: { + readonly resourceSpans?: ReadonlyArray< + Resource & { readonly scopeSpans: ReadonlyArray<{ readonly spans: ReadonlyArray }> } + >; + readonly resourceMetrics?: ReadonlyArray< + Resource & { readonly scopeMetrics: ReadonlyArray<{ readonly metrics: ReadonlyArray }> } + >; + }; +} + +const read = (request: IncomingMessage) => + new Promise((resolve, reject) => { + const chunks: Buffer[] = []; + request.on("data", (chunk: Buffer) => chunks.push(chunk)); + request.on("end", () => resolve(Buffer.concat(chunks).toString("utf8"))); + request.on("error", reject); + }); + +const sse = (chunks: ReadonlyArray) => + `${chunks.map((chunk) => `data: ${JSON.stringify(chunk)}\n\n`).join("")}data: [DONE]\n\n`; + +const chunk = (choice: object | undefined, usage?: object) => ({ + id: "chatcmpl-otel", + object: "chat.completion.chunk", + created: 1, + model: "otel-model", + choices: choice === undefined ? [] : [{ index: 0, ...choice }], + ...(usage === undefined ? {} : { usage }), +}); + +/** + * The first successful request answers with one `read` call; the request that + * carries its result answers with text. Both report usage, so every GenAI metric records. + */ +const completion = (body: { readonly messages: ReadonlyArray<{ readonly role: string }> }) => { + const usage = { prompt_tokens: 120, completion_tokens: 8, total_tokens: 128 }; + if (body.messages.some((message) => message.role === "tool")) { + return sse([ + chunk({ delta: { role: "assistant", content: "The note says hello." }, finish_reason: null }), + chunk({ delta: {}, finish_reason: "stop" }), + chunk(undefined, usage), + ]); + } + return sse([ + chunk({ + delta: { + role: "assistant", + tool_calls: [ + { + index: 0, + id: "call_otel", + type: "function", + function: { name: "read", arguments: '{"path":"note.txt"}' }, + }, + ], + }, + finish_reason: null, + }), + chunk({ delta: {}, finish_reason: "tool_calls" }), + chunk(undefined, usage), + ]); +}; + +/** + * One local server is both the OpenAI-compatible provider and the OTLP/HTTP + * collector. The provider's first answer is a 503, so the turn retries once. + */ +const serve = async () => { + const exports: Export[] = []; + let completions = 0; + const server = createServer(async (request, response) => { + const body = await read(request); + const path = request.url ?? ""; + if (path === "/v1/chat/completions") { + completions += 1; + if (completions === 1) { + response.writeHead(503, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: { message: "overloaded", type: "server_error" } })); + return; + } + response.writeHead(200, { "content-type": "text/event-stream" }); + response.end(completion(JSON.parse(body))); + return; + } + const header = request.headers["x-otel-test"]; + exports.push({ path, header: typeof header === "string" ? header : undefined, body: JSON.parse(body) }); + response.writeHead(200, { "content-type": "application/json" }); + response.end("{}"); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const { port } = server.address() as AddressInfo; + return { url: `http://127.0.0.1:${port}`, exports, close: () => new Promise((resolve) => server.close(resolve)) }; +}; + +/** Spawn asynchronously: the fake provider answers from this process's event loop. */ +const runCli = (args: ReadonlyArray, options: { readonly cwd: string; readonly env: NodeJS.ProcessEnv }) => + new Promise<{ readonly status: number | null; readonly stdout: string; readonly stderr: string }>( + (resolve, reject) => { + const child = spawn(process.execPath, ["--conditions=development", cli, ...args], { + cwd: options.cwd, + env: options.env, + }); + let stdout = ""; + let stderr = ""; + child.stdout.on("data", (data: Buffer) => { + stdout += data.toString("utf8"); + }); + child.stderr.on("data", (data: Buffer) => { + stderr += data.toString("utf8"); + }); + // A hung export or shutdown must not outlive the test and its collector. + const deadline = setTimeout(() => child.kill("SIGKILL"), 45_000); + child.on("error", reject); + child.on("close", (status) => { + clearTimeout(deadline); + resolve({ status, stdout, stderr }); + }); + }, + ); + +const value = (input: OtlpValue): unknown => { + if (input.stringValue !== undefined) return input.stringValue; + if (input.intValue !== undefined) return Number(input.intValue); + if (input.doubleValue !== undefined) return input.doubleValue; + if (input.boolValue !== undefined) return input.boolValue; + return (input.arrayValue?.values ?? []).map(value); +}; + +/** Ids and versions change per run; everything else in the artifact is fixed by the fake provider. */ +const volatile = new Set(["gen_ai.tool.call.id", "gen_ai.response.id", "service.instance.id", "service.version"]); + +const attributes = (input: ReadonlyArray | undefined) => + Object.fromEntries( + [...(input ?? [])] + .sort((left, right) => left.key.localeCompare(right.key)) + .map((attribute) => [attribute.key, volatile.has(attribute.key) ? "" : value(attribute.value)]), + ); + +const SPAN_KIND = ["unspecified", "internal", "server", "client", "producer", "consumer"]; + +/** + * The GenAI spans in start order, each under its nearest GenAI ancestor. Every + * other harness span is left out: their count follows the database and the + * sandbox, not the telemetry contract. + */ +const genAiSpans = (exports: ReadonlyArray) => { + const spans = exports.flatMap((entry) => + (entry.body.resourceSpans ?? []).flatMap((resource) => resource.scopeSpans.flatMap((scope) => scope.spans)), + ); + const byId = new Map(spans.map((span) => [span.spanId, span])); + const isGenAi = (span: OtlpSpan) => span.attributes?.some((attribute) => attribute.key === "gen_ai.operation.name"); + const ancestor = (span: OtlpSpan): string | undefined => { + for (let parent = byId.get(span.parentSpanId ?? ""); parent; parent = byId.get(parent.parentSpanId ?? "")) { + if (isGenAi(parent)) return parent.name; + } + return undefined; + }; + return spans + .filter(isGenAi) + .sort((left, right) => (BigInt(left.startTimeUnixNano) < BigInt(right.startTimeUnixNano) ? -1 : 1)) + .map((span) => ({ + name: span.name, + kind: SPAN_KIND[span.kind] ?? span.kind, + parent: ancestor(span) ?? null, + error: span.status?.code === 2, + attributes: attributes(span.attributes), + })); +}; + +// Counts and sums that the fake provider fixes; every other recorded value follows timing. +const exact = new Set([ + "gen_ai.client.token.usage", + "codework.llm.cost", + "codework.turn.retries", + "codework.tool.output.truncated", +]); +const runtime = (name: string) => name.startsWith("child_fiber"); + +const recorded = (name: string, point: OtlpDataPoint) => { + if (runtime(name)) return {}; + const count = point.count === undefined ? {} : { count: Number(point.count) }; + if (!exact.has(name)) return count; + const total = point.sum ?? point.asDouble ?? (point.asInt === undefined ? undefined : Number(point.asInt)); + return { ...count, value: total === undefined ? null : Number(total.toPrecision(12)) }; +}; + +/** + * Each metric's unit and type, and each series it exported: its full attribute + * set with its count, plus its value where the run fixes it. Metrics are + * cumulative, so the last export of a series is its total. + */ +const metrics = (exports: ReadonlyArray) => { + const found = new Map< + string, + { readonly unit: string | undefined; readonly type: string; readonly series: Map } + >(); + for (const entry of exports) { + for (const resource of entry.body.resourceMetrics ?? []) { + for (const scope of resource.scopeMetrics) { + for (const metric of scope.metrics) { + const type = metric.histogram ? "histogram" : metric.gauge ? "gauge" : "sum"; + const current = found.get(metric.name) ?? { unit: metric.unit, type, series: new Map() }; + for (const point of (metric.histogram ?? metric.gauge ?? metric.sum)?.dataPoints ?? []) { + const labels = attributes(point.attributes); + current.series.set(JSON.stringify(labels), { attributes: labels, ...recorded(metric.name, point) }); + } + found.set(metric.name, current); + } + } + } + } + return Object.fromEntries( + [...found.entries()] + .sort(([left], [right]) => left.localeCompare(right)) + .map(([name, metric]) => [ + name, + { + unit: metric.unit, + type: metric.type, + series: [...metric.series.entries()] + .sort(([left], [right]) => left.localeCompare(right)) + .map(([, series]) => series), + }, + ]), + ); +}; + +/** Every exported resource, traces and metrics alike, must describe the same process. */ +const resources = (exports: ReadonlyArray) => + exports.flatMap((entry) => [...(entry.body.resourceSpans ?? []), ...(entry.body.resourceMetrics ?? [])]); + +describe("OTLP export", () => { + it("exports GenAI spans and metrics for a run through the standard OTEL_* environment", async () => { + const collector = await serve(); + try { + const home = mkdtempSync(join(tmpdir(), "codework-otel-")); + temps.push(home); + const project = join(home, "project"); + const settings = join(home, "settings"); + mkdirSync(project); + mkdirSync(settings); + // Longer than `read`'s line limit, so its result is truncated. + writeFileSync( + join(project, "note.txt"), + Array.from({ length: 2500 }, (_, index) => `hello ${index}`).join("\n"), + ); + writeFileSync( + join(settings, "settings.jsonc"), + JSON.stringify({ + retry: { baseDelayMs: 10 }, + model: { provider: "otel", id: "otel-model" }, + models: [ + { + id: "otel-model", + name: "otel-model", + provider: { id: "otel", name: "Otel", source: "config", env: [] }, + baseUrl: `${collector.url}/v1`, + reasoning: false, + input: ["text"], + cost: { input: 3, output: 15, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 32768, + maxTokens: 8192, + protocol: "openai-compatible", + }, + ], + }), + ); + + const env: NodeJS.ProcessEnv = { + ...process.env, + HOME: home, + CODEWORK_MODELS_FILE: models, + OTEL_EXPORTER_OTLP_ENDPOINT: collector.url, + OTEL_EXPORTER_OTLP_HEADERS: "x-otel-test=present", + // The collector reads JSON; protobuf is the default otherwise. + OTEL_EXPORTER_OTLP_PROTOCOL: "http/json", + }; + const configured = new Set([ + "OTEL_EXPORTER_OTLP_ENDPOINT", + "OTEL_EXPORTER_OTLP_HEADERS", + "OTEL_EXPORTER_OTLP_PROTOCOL", + ]); + for (const key of Object.keys(env)) { + if (key.startsWith("OTEL_") && !configured.has(key)) delete env[key]; + } + + const result = await runCli( + ["--home", home, "--database", ":memory:", "--user-config-dir", settings, "run", "What does note.txt say?"], + { cwd: project, env }, + ); + expect(result.status, result.stderr).toBe(0); + expect(result.stdout).toContain("The note says hello."); + + expect(collector.exports.length).toBeGreaterThan(0); + expect(new Set(collector.exports.map((entry) => entry.path))).toEqual(new Set(["/v1/traces", "/v1/metrics"])); + expect(collector.exports.every((entry) => entry.header === "present")).toBe(true); + + const exported = resources(collector.exports).map((entry) => JSON.stringify(entry.resource)); + expect(new Set(exported).size, "every export carries one resource").toBe(1); + + // No prompt, file content, or model output may leave in any span, attribute, or event. + const traces = JSON.stringify(collector.exports.filter((entry) => entry.path === "/v1/traces")); + for (const content of ["What does note.txt say?", "hello 1", "The note says hello."]) { + expect(traces).not.toContain(content); + } + + await expect( + `${JSON.stringify( + { + resource: attributes(resources(collector.exports)[0]?.resource?.attributes), + spans: genAiSpans(collector.exports), + metrics: metrics(collector.exports), + }, + null, + "\t", + )}\n`, + ).toMatchFileSnapshot("./__artifacts__/telemetry.json"); + } finally { + await collector.close(); + } + }, 60_000); +}); diff --git a/packages/harness/src/runner/llm.ts b/packages/harness/src/runner/llm.ts index 2db82f17..b73e0e8e 100644 --- a/packages/harness/src/runner/llm.ts +++ b/packages/harness/src/runner/llm.ts @@ -11,13 +11,14 @@ import { stream as aikitStream, llm, type Event as AikitEvent, type Message, typ import * as AikitFailure from "@codeworksh/aikit/failure"; import { getGitHubCopilotApiKey, JsonGitHubCopilotAuthStorage } from "@codeworksh/aikit/oauth/github/copilot"; import { getOpenAICodexApiKey, JsonOpenAICodexAuthStorage } from "@codeworksh/aikit/oauth/openai/codex"; -import { Duration, Effect, Exit, Fiber, Scope, Stream } from "effect"; +import { Cause, Data, Duration, Effect, Exit, Fiber, Schema, Scope, Stream } from "effect"; import { ModelCatalog } from "../model/catalog.ts"; import type { SessionSchema } from "../session/schema.ts"; import { merge } from "../settings/merge.ts"; import { resolveOverrides, resolveRequest } from "../settings/resolve.ts"; import { type Block, type DeclaredModel, keyName } from "../settings/schema.ts"; import type { State } from "../state/state.ts"; +import { Telemetry } from "../telemetry/telemetry.ts"; import { Attribution } from "./attribution.ts"; import { LLMEventPublisher } from "./event.ts"; import { Runner } from "./run.ts"; @@ -217,12 +218,108 @@ export const openWith = (options: OpenOptions = {}): Open => export const open: Open = openWith(); +const isProviderError = Schema.is(Runner.ProviderError); + +/** `error.type` for a request: the provider's failure reason, so a terminal and a thrown failure agree. */ +const requestErrorType = (input: RequestInput, exit: Exit.Exit) => { + if (Exit.isFailure(exit)) { + const error = Cause.squash(exit.cause); + return isProviderError(error) ? error.reason._tag : Telemetry.errorType(exit.cause); + } + if (exit.value.outcome === "ended") return undefined; + if (exit.value.reason === "aborted") return "aborted"; + return providerError(input, messageFailure(exit.value.message)).reason._tag; +}; + +/** Span attributes and GenAI metrics for one settled provider request. */ +const observe = Effect.fnUntraced(function* ( + input: RequestInput, + timing: { readonly start: number; readonly firstToken: number | undefined }, + exit: Exit.Exit, +) { + const now = yield* Effect.clockWith((clock) => clock.currentTimeMillis); + const attributes = { + "gen_ai.operation.name": "chat", + "gen_ai.provider.name": input.provider, + "gen_ai.request.model": input.model, + }; + const errorType = requestErrorType(input, exit); + yield* Telemetry.record( + Telemetry.operationDuration, + (now - timing.start) / 1000, + errorType === undefined ? attributes : { ...attributes, "error.type": errorType }, + ); + if (timing.firstToken !== undefined) { + yield* Telemetry.record(Telemetry.timeToFirstToken, (timing.firstToken - timing.start) / 1000, attributes); + } + if (errorType !== undefined) yield* Effect.annotateCurrentSpan("error.type", errorType); + if (Exit.isFailure(exit)) return; + + const { message, outcome } = exit.value; + const { usage } = message; + // aikit's `input` excludes cached tokens; the conventions count them as input. + const inputTokens = usage.input + usage.cacheRead + usage.cacheWrite; + // A request that failed before the provider reported usage carries zeros, which are not observations. + if (inputTokens > 0 || usage.output > 0) { + yield* Telemetry.record(Telemetry.tokenUsage, inputTokens, { ...attributes, "gen_ai.token.type": "input" }); + yield* Telemetry.record(Telemetry.tokenUsage, usage.output, { ...attributes, "gen_ai.token.type": "output" }); + } + if (usage.cost.total > 0) yield* Telemetry.record(Telemetry.cost, usage.cost.total, attributes); + // The interval after the first token covers every token but that one. + if (outcome === "ended" && timing.firstToken !== undefined && usage.output > 1) { + yield* Telemetry.record( + Telemetry.timePerOutputToken, + (now - timing.firstToken) / 1000 / (usage.output - 1), + attributes, + ); + } + yield* Effect.annotateCurrentSpan({ + "gen_ai.response.model": message.responseModel ?? message.model, + ...(message.responseId === undefined ? {} : { "gen_ai.response.id": message.responseId }), + "gen_ai.response.finish_reasons": [exit.value.reason], + "gen_ai.usage.input_tokens": inputTokens, + "gen_ai.usage.output_tokens": usage.output, + "gen_ai.usage.cache_read.input_tokens": usage.cacheRead, + "gen_ai.usage.cache_creation.input_tokens": usage.cacheWrite, + ...(usage.reasoning === undefined ? {} : { "gen_ai.usage.reasoning_tokens": usage.reasoning }), + }); +}); + +// aikit pushes `start` and block starts before any content, so the first token is the first delta. +const isToken = (event: AikitEvent.LLMMessageEvent) => + event.type === "text.delta" || event.type === "thinking.delta" || event.type === "toolcall.delta"; + +/** + * A failed terminal is the request's result, not an error. It leaves the + * traced request as a failure so the span records an error status, and is + * recovered just outside it. + */ +class FailedTerminal extends Data.TaggedError("LLM.FailedTerminal")<{ + readonly terminal: LLMEventPublisher.Terminal; +}> {} + /** * Build a provider request from an opener. Tests use this with deterministic * aikit event streams; production uses `open` above. */ export const make = (openStream: Open): Request => { - const request = Effect.fn("LLM.run")(function* (input: RequestInput) { + const request = Effect.fn("LLM.run", { kind: "client" })(function* (input: RequestInput) { + yield* Effect.annotateCurrentSpan({ + "gen_ai.operation.name": "chat", + "gen_ai.provider.name": input.provider, + "gen_ai.request.model": input.model, + }); + const start = yield* Effect.clockWith((clock) => clock.currentTimeMillis); + let firstToken: number | undefined; + const markToken = (event: AikitEvent.LLMMessageEvent) => + firstToken !== undefined || !isToken(event) + ? Effect.void + : Effect.clockWith((clock) => + Effect.sync(() => { + firstToken = clock.currentTimeMillisUnsafe(); + }), + ); + /* * Give the transport signal its own closeable scope. Closing it aborts the * signal immediately without closing this request's outer scope, which must @@ -233,11 +330,11 @@ export const make = (openStream: Open): Request => { const signal = yield* Effect.abortSignal.pipe(Scope.provide(signalScope)); const abort = Scope.close(signalScope, Exit.void); - return yield* Effect.uninterruptibleMask((restore) => + const settled = Effect.uninterruptibleMask((restore) => Effect.gen(function* () { const iterable = yield* restore(openStream(input, signal)).pipe(Effect.onInterrupt(() => abort)); const consume = Stream.fromAsyncIterable(iterable, (cause) => providerErrorFromUnknown(input, cause)).pipe( - Stream.runForEach(input.publisher.publish), + Stream.runForEach((event) => Effect.andThen(markToken(event), input.publisher.publish(event))), Effect.andThen(input.publisher.terminal), Effect.interruptible, ); @@ -266,9 +363,16 @@ export const make = (openStream: Open): Request => { return completed.value; }), ); + const terminal = yield* settled.pipe(Effect.onExit((exit) => observe(input, { start, firstToken }, exit))); + if (terminal.outcome === "failed") return yield* new FailedTerminal({ terminal }); + return terminal; }); - return (input) => request(input).pipe(Effect.scoped); + return (input) => + request(input).pipe( + Effect.catchTag("LLM.FailedTerminal", (failed) => Effect.succeed(failed.terminal)), + Effect.scoped, + ); }; export const run = make(open); diff --git a/packages/harness/src/runner/loop.ts b/packages/harness/src/runner/loop.ts index c8845f38..603e39ee 100644 --- a/packages/harness/src/runner/loop.ts +++ b/packages/harness/src/runner/loop.ts @@ -16,6 +16,7 @@ import { SessionMessageSchema } from "../session/message/schema.ts"; import type { SessionSchema } from "../session/schema.ts"; import { Session } from "../session/session.ts"; import { State } from "../state/state.ts"; +import { Telemetry } from "../telemetry/telemetry.ts"; import { errorMessage } from "../util/error.ts"; import { LLMEventPublisher } from "./event.ts"; import { LLM } from "./llm.ts"; @@ -66,6 +67,48 @@ const errorPart = ( time: { ...call.time, end: endTime(call, now) }, }); +const agentAttributes = (snapshot: State.Snapshot) => ({ + "gen_ai.operation.name": "invoke_agent", + "gen_ai.provider.name": snapshot.provider, + "gen_ai.request.model": snapshot.model, +}); + +// Built-in tools report truncation as `details.truncation` (read, find, grep, ls) or `details.truncated` (bash). +const isTruncated = (details: unknown) => { + if (typeof details !== "object" || details === null) return false; + if ("truncated" in details && details.truncated === true) return true; + return ( + "truncation" in details && + typeof details.truncation === "object" && + details.truncation !== null && + "truncated" in details.truncation && + details.truncation.truncated === true + ); +}; + +const observeTool = ( + snapshot: State.Snapshot, + call: Message.ToolCallPendingPart, + seconds: number, + exit: Exit.Exit, +) => + Effect.gen(function* () { + const outcome = Exit.isFailure(exit) + ? Cause.hasInterruptsOnly(exit.cause) + ? "interrupted" + : "error" + : exit.value.status === "completed" + ? "ok" + : "error"; + // The model names the tool; an unregistered name would open a new series per hallucination. + const name = snapshot.tools.defs.some((def) => def.name === call.name) ? call.name : "_OTHER"; + const attributes = { "gen_ai.operation.name": "execute_tool", "gen_ai.tool.name": name }; + yield* Telemetry.record(Telemetry.toolDuration, seconds, { ...attributes, outcome }); + if (Exit.isSuccess(exit) && isTruncated(exit.value.result.details)) { + yield* Telemetry.record(Telemetry.truncated, 1, attributes); + } + }); + export const layer = (options: Options = {}) => Layer.effect( Runner.Service, @@ -171,6 +214,11 @@ export const layer = (options: Options = {}) => const settled = new Set(); const handle = Effect.fn("Loop.handleTool")(function* (call: Message.ToolCallPendingPart) { + yield* Effect.annotateCurrentSpan({ + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": call.name, + "gen_ai.tool.call.id": call.callID, + }); yield* events.publish(EventList.ToolStarted, { timestamp: yield* DateTime.now, sessionId: snapshot.sessionId, @@ -180,7 +228,7 @@ export const layer = (options: Options = {}) => label: snapshot.tools.defs.find((def) => def.name === call.name)?.label, arguments: call.arguments, }); - const handled = yield* snapshot.tools + const execute = snapshot.tools .handle(call, { sessionId: snapshot.sessionId, messageId, @@ -205,6 +253,9 @@ export const layer = (options: Options = {}) => ), ), ); + const handled = yield* Telemetry.timed(execute, (seconds, exit) => + observeTool(snapshot, call, seconds, exit), + ); // The durable settlement and the in-memory completion marker form one // protected step. Otherwise an interrupt between them could make the // cleanup publish a second terminal transition for the same call. @@ -317,6 +368,10 @@ export const layer = (options: Options = {}) => if (delayMs === undefined) return Effect.fail(error); return Effect.gen(function* () { retries += 1; + yield* Telemetry.record(Telemetry.retries, 1, { + "gen_ai.provider.name": snapshot.provider, + "gen_ai.request.model": snapshot.model, + }); yield* Effect.uninterruptible(settleDraft({ _tag: "error", message: error.message })); yield* events.publish(EventList.RetryScheduled, { timestamp: yield* DateTime.now, @@ -399,20 +454,38 @@ export const layer = (options: Options = {}) => const hasSteer = yield* inputs.hasPending(input.sessionId, "steer"); const hasFollowUp = hasSteer ? false : yield* inputs.hasPending(input.sessionId, "followUp"); if (!input.force && !hasSteer && !hasFollowUp) return; + yield* Effect.annotateCurrentSpan("gen_ai.operation.name", "invoke_agent"); let promotion: Promotion = hasSteer ? "steer" : hasFollowUp ? "followUp" : undefined; - let shouldRun = true; - while (shouldRun) { - const snapshot = yield* state.snapshot(input.sessionId); - let needsContinuation = true; - while (needsContinuation) { - const result = yield* runTurn(promotion, snapshot); - promotion = "steer"; - needsContinuation = result.needsContinuation || (yield* inputs.hasPending(input.sessionId, "steer")); + // A follow-up re-reads State, so the run is attributed to the model it last used. + let latest: State.Snapshot | undefined; + const drain = Effect.gen(function* () { + let shouldRun = true; + while (shouldRun) { + const snapshot = yield* state.snapshot(input.sessionId); + latest = snapshot; + let needsContinuation = true; + while (needsContinuation) { + const result = yield* runTurn(promotion, snapshot); + promotion = "steer"; + needsContinuation = + result.needsContinuation || (yield* inputs.hasPending(input.sessionId, "steer")); + } + shouldRun = yield* inputs.hasPending(input.sessionId, "followUp"); + promotion = shouldRun ? "followUp" : undefined; } - shouldRun = yield* inputs.hasPending(input.sessionId, "followUp"); - promotion = shouldRun ? "followUp" : undefined; - } + }); + yield* Telemetry.timed(drain, (seconds, exit) => + latest === undefined + ? Effect.void + : Telemetry.record( + Telemetry.agentDuration, + seconds, + Exit.isSuccess(exit) + ? agentAttributes(latest) + : { ...agentAttributes(latest), "error.type": Telemetry.errorType(exit.cause) }, + ), + ); }); return Runner.Service.of({ run }); diff --git a/packages/harness/src/sandbox/control.ts b/packages/harness/src/sandbox/control.ts index 1409af21..8f7e423d 100644 --- a/packages/harness/src/sandbox/control.ts +++ b/packages/harness/src/sandbox/control.ts @@ -1,7 +1,20 @@ -import { Cause, Context, DateTime, type Duration, Effect, Layer, LayerMap, Option, Schema, Semaphore } from "effect"; +import { + Cause, + Context, + DateTime, + type Duration, + Effect, + Exit, + Layer, + LayerMap, + Option, + Schema, + Semaphore, +} from "effect"; import { SqlClient } from "effect/sql"; import { SandboxInstanceRow } from "../db/schema.sql.ts"; import { Space } from "../space/space.ts"; +import { Telemetry } from "../telemetry/telemetry.ts"; import { SandboxDriver } from "./driver.ts"; import { providerError, @@ -126,6 +139,19 @@ const isSandboxProviderError = Schema.is(SandboxProviderError); const parseJson = (driver: SandboxDriver.Name, operation: string, value: string) => decodeUnknownJson(value).pipe(Effect.mapError((cause) => providerError({ driver, operation, cause }))); +/** Time a driver's provider-side create or wake, the cold start a remote sandbox pays. */ +const provision = (driver: SandboxDriver.Name, operation: "create" | "wake", self: Effect.Effect) => + Telemetry.timed(self, (seconds, exit) => { + const attributes = { "codework.sandbox.driver": driver, "codework.sandbox.operation": operation }; + return Telemetry.record( + Telemetry.provisionDuration, + seconds, + Exit.isSuccess(exit) + ? { ...attributes, outcome: "ok" } + : { ...attributes, outcome: "error", "error.type": Telemetry.errorType(exit.cause) }, + ); + }); + export const make = Effect.fn("Sandbox.Controller.make")(function* (options: Options) { const sql = yield* SqlClient.SqlClient; const registry = yield* SandboxDriverRegistry.Registry; @@ -413,9 +439,9 @@ export const make = Effect.fn("Sandbox.Controller.make")(function* (options: Opt metadata: input.metadata, }); - const provisioned = yield* driver - .create({ instanceId: id, config }) - .pipe(Effect.catch((error) => persistProviderFailure(id, "faulted", error))); + const provisioned = yield* provision(driver.name, "create", driver.create({ instanceId: id, config })).pipe( + Effect.catch((error) => persistProviderFailure(id, "faulted", error)), + ); const finalize = Effect.gen(function* () { const encoded = yield* registry.encodeRuntimeConfig(driver.name, provisioned.runtimeConfig); const runtimeConfig = yield* encodeUnknownJson(encoded).pipe( @@ -706,15 +732,11 @@ export const make = Effect.fn("Sandbox.Controller.make")(function* (options: Opt }); } if (refCount(id) === 0) yield* transports.invalidate(id); - const observed = yield* attached.driver - .wake(attached.input) - .pipe( - Effect.catch((error): Effect.Effect => - providerErrorIsNotFound(error) - ? markUnavailable(id, error) - : persistProviderFailure(id, "faulted", error), - ), - ); + const observed = yield* provision(attached.driver.name, "wake", attached.driver.wake(attached.input)).pipe( + Effect.catch((error): Effect.Effect => + providerErrorIsNotFound(error) ? markUnavailable(id, error) : persistProviderFailure(id, "faulted", error), + ), + ); yield* updateObservation({ id, status: observed.status, diff --git a/packages/harness/src/telemetry/telemetry.ts b/packages/harness/src/telemetry/telemetry.ts new file mode 100644 index 00000000..bcbf0a8d --- /dev/null +++ b/packages/harness/src/telemetry/telemetry.ts @@ -0,0 +1,105 @@ +/* + * @file Harness metrics and the GenAI span vocabulary. + * + * Effect owns the instruments, the spans, and the export; this module only names + * them. Names follow the OpenTelemetry GenAI conventions where one exists, and + * Effect's OTLP exporter reads a metric's unit from its `unit` attribute. + */ + +import { Cause, Effect, type Exit, Metric } from "effect"; + +// Bucket boundaries are the GenAI conventions' advice for each instrument. +const durationBuckets = Metric.boundariesFromIterable([ + 0.01, 0.02, 0.04, 0.08, 0.16, 0.32, 0.64, 1.28, 2.56, 5.12, 10.24, 20.48, 40.96, 81.92, +]); + +export const tokenUsage = Metric.histogram("gen_ai.client.token.usage", { + description: "Tokens used per provider request", + attributes: { unit: "{token}" }, + boundaries: Metric.boundariesFromIterable([ + 1, 4, 16, 64, 256, 1024, 4096, 16384, 65536, 262144, 1048576, 4194304, 16777216, 67108864, + ]), +}); + +export const operationDuration = Metric.histogram("gen_ai.client.operation.duration", { + description: "Provider request duration", + attributes: { unit: "s" }, + boundaries: durationBuckets, +}); + +// The `gen_ai.server.*` timings are a model server's own measure; a client also times the transport. +export const timeToFirstToken = Metric.histogram("codework.llm.time_to_first_token", { + description: "Time from request start to the first streamed content", + attributes: { unit: "s" }, + boundaries: Metric.boundariesFromIterable([ + 0.001, 0.005, 0.01, 0.02, 0.04, 0.06, 0.08, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10, + ]), +}); + +export const timePerOutputToken = Metric.histogram("codework.llm.time_per_output_token", { + description: "Generation time after the first token, per later output token", + attributes: { unit: "s" }, + boundaries: Metric.boundariesFromIterable([0.01, 0.025, 0.05, 0.075, 0.1, 0.15, 0.2, 0.3, 0.4, 0.5, 0.75, 1, 2.5]), +}); + +export const toolDuration = Metric.histogram("gen_ai.execute_tool.duration", { + description: "Tool execution duration", + attributes: { unit: "s" }, + boundaries: durationBuckets, +}); + +export const agentDuration = Metric.histogram("gen_ai.invoke_agent.duration", { + description: "Session run duration, across every turn it drains", + attributes: { unit: "s" }, + boundaries: durationBuckets, +}); + +export const cost = Metric.counter("codework.llm.cost", { + description: "Provider request cost from the model catalog's prices", + attributes: { unit: "USD" }, + incremental: true, +}); + +export const retries = Metric.counter("codework.turn.retries", { + description: "Provider requests retried within a turn", + attributes: { unit: "{retry}" }, + incremental: true, +}); + +export const provisionDuration = Metric.histogram("codework.sandbox.provision.duration", { + description: "Sandbox driver create and wake duration", + attributes: { unit: "s" }, + boundaries: durationBuckets, +}); + +export const truncated = Metric.counter("codework.tool.output.truncated", { + description: "Tool results whose output was truncated before the model saw it", + attributes: { unit: "{call}" }, + incremental: true, +}); + +export type Attributes = Readonly>; + +export const record = (metric: Metric.Metric, value: Input, attributes: Attributes) => + Metric.update(Metric.withAttributes(metric, attributes), value); + +/** `error.type` for a failed exit: the error's own tag, or `interrupted`. */ +export const errorType = (cause: Cause.Cause): string => { + if (Cause.hasInterruptsOnly(cause)) return "interrupted"; + const error = Cause.squash(cause); + return typeof error === "object" && error !== null && "_tag" in error && typeof error._tag === "string" + ? error._tag + : "_OTHER"; +}; + +/** Run `self`, then hand its duration in seconds and its exit to `observe`. */ +export const timed = ( + self: Effect.Effect, + observe: (seconds: number, exit: Exit.Exit) => Effect.Effect, +): Effect.Effect => + Effect.clockWith((clock) => { + const start = clock.currentTimeMillisUnsafe(); + return self.pipe(Effect.onExit((exit) => observe((clock.currentTimeMillisUnsafe() - start) / 1000, exit))); + }); + +export * as Telemetry from "./telemetry.ts"; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 8cef4678..54a34021 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -468,6 +468,9 @@ importers: jsonc-parser: specifier: 3.3.1 version: 3.3.1 + uuidv7: + specifier: ^1.2.1 + version: 1.2.1 zod: specifier: 4.6.5 version: 4.6.5 From 1d449df6d48f6e4aee9637ae58cb98874f7f06dc Mon Sep 17 00:00:00 2001 From: Sanchit Rk <0xsanchit@gmail.com> Date: Tue, 13 Oct 2026 15:32:03 +0530 Subject: [PATCH 2/6] docs: changelog for OpenTelemetry export Also adds the blank line the formatter expects before the Internal heading. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 605f5bee..0d7bfac9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,9 +11,15 @@ This file is the canonical source for unreleased changes and published release n ## [Unreleased] +### Added + +- `@codeworksh/cli`: Added OpenTelemetry export. Set `OTEL_EXPORTER_OTLP_ENDPOINT` and every command sends its traces and metrics over OTLP/HTTP, as protobuf by default or JSON with `OTEL_EXPORTER_OTLP_PROTOCOL=http/json`. The standard `OTEL_*` variables configure it (headers, per-signal endpoints, `OTEL_SDK_DISABLED`, `OTEL_SERVICE_NAME`, `OTEL_RESOURCE_ATTRIBUTES`); there is no settings key. Without an endpoint nothing is exported, and a bad configuration never stops a command. +- `@codeworksh/harness`: Spans for model requests, tool calls and agent runs carry the OpenTelemetry GenAI attributes (provider, model, response id, finish reason, token counts), and a failed model request ends its span with an error. Metrics cover token usage, request, tool and agent duration, time to first token, time per output token, cost, turn retries, sandbox create and wake duration, and truncated tool output, alongside Effect's fiber runtime metrics. Prompts, file contents and model output are never exported. + ### Fixed - `@codeworksh/plugin`: Fixed tools whose parameters are `Schema.Struct({})` being rejected by OpenAI and Anthropic. Their JSON Schema is now `{ "type": "object", "properties": {} }` instead of `{ "not": { "type": "null" } }`. + ### Internal - The `codework tui` E2E artifact now records the CLI version as ``, so it no longer goes stale on every version bump. From 46c77ec76bdd2ed868ed6b8a84906cf95884a9c8 Mon Sep 17 00:00:00 2001 From: Sanchit Rk <0xsanchit@gmail.com> Date: Tue, 13 Oct 2026 16:12:53 +0530 Subject: [PATCH 3/6] refactor: name telemetry metrics under codework.* The GenAI semantic conventions are still in development and have renamed their metrics more than once, so metrics now use codework's own names and short labels; spans keep the gen_ai.* attributes trace views read. - codework.llm.{duration,ttft,tokens,cost}, codework.agent.duration, codework.tool.duration, codework.sandbox.{provision,exec}.duration and codework.sandbox.active, plus Effect's fiber runtime metrics. - Spans: gen_ai.usage.cache_write/reasoning.output_tokens names, conversation id, stream flag, time to first chunk, agent name. - Drop the retries and truncation counters and per-signal protocol vars. - Test drives a bash call so the sandbox exec metric is covered. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 2 +- packages/codework/src/framework/otel.ts | 51 +++--- .../test/__artifacts__/telemetry.json | 145 +++++++----------- packages/codework/test/telemetry.test.ts | 35 +++-- packages/harness/src/runner/llm.ts | 49 +++--- packages/harness/src/runner/loop.ts | 69 ++------- packages/harness/src/sandbox/control.ts | 90 ++++++++--- packages/harness/src/telemetry/telemetry.ts | 84 +++++----- 8 files changed, 247 insertions(+), 278 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0d7bfac9..ea1263a7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,7 +14,7 @@ This file is the canonical source for unreleased changes and published release n ### Added - `@codeworksh/cli`: Added OpenTelemetry export. Set `OTEL_EXPORTER_OTLP_ENDPOINT` and every command sends its traces and metrics over OTLP/HTTP, as protobuf by default or JSON with `OTEL_EXPORTER_OTLP_PROTOCOL=http/json`. The standard `OTEL_*` variables configure it (headers, per-signal endpoints, `OTEL_SDK_DISABLED`, `OTEL_SERVICE_NAME`, `OTEL_RESOURCE_ATTRIBUTES`); there is no settings key. Without an endpoint nothing is exported, and a bad configuration never stops a command. -- `@codeworksh/harness`: Spans for model requests, tool calls and agent runs carry the OpenTelemetry GenAI attributes (provider, model, response id, finish reason, token counts), and a failed model request ends its span with an error. Metrics cover token usage, request, tool and agent duration, time to first token, time per output token, cost, turn retries, sandbox create and wake duration, and truncated tool output, alongside Effect's fiber runtime metrics. Prompts, file contents and model output are never exported. +- `@codeworksh/harness`: Spans for model requests, tool calls and agent runs carry the OpenTelemetry GenAI attributes (provider, model, session, response id, finish reason, token counts, time to first token), and a failed model request ends its span with an error. Metrics: `codework.llm.duration`, `codework.llm.ttft`, `codework.llm.tokens` (by `type`: input, output, cache_read, cache_write, reasoning), `codework.llm.cost`, `codework.agent.duration`, `codework.tool.duration` (by `tool`), `codework.sandbox.provision.duration`, `codework.sandbox.exec.duration` and `codework.sandbox.active` (by `driver`), plus Effect's fiber runtime metrics (`child_fibers_*`). Prompts, file contents and model output are never exported. ### Fixed diff --git a/packages/codework/src/framework/otel.ts b/packages/codework/src/framework/otel.ts index f51b22f5..17d0fbdd 100644 --- a/packages/codework/src/framework/otel.ts +++ b/packages/codework/src/framework/otel.ts @@ -1,6 +1,6 @@ import { Config, ConfigProvider, Effect, Layer, Metric, Option, Schema } from "effect"; -import { FetchHttpClient, type HttpClient } from "effect/http"; -import { OtlpExporter, OtlpMetrics, OtlpSerialization, OtlpTracer } from "effect/observability"; +import { FetchHttpClient } from "effect/http"; +import { OtlpMetrics, OtlpSerialization, OtlpTracer } from "effect/observability"; import { uuidv7 } from "uuidv7"; /** @@ -36,31 +36,32 @@ const fiberMetrics = Layer.unwrap( }), ); +interface OtlpResource { + readonly serviceName: string; + readonly serviceVersion: string; + readonly attributes: Record; +} + /** - * Serialize each signal by its `OTEL_EXPORTER_OTLP_*PROTOCOL`, defaulting to the - * spec's `http/protobuf`. Effect has no gRPC exporter, so `grpc` exports nothing + * Serialize by `OTEL_EXPORTER_OTLP_PROTOCOL`, defaulting to the spec's + * `http/protobuf`. Effect has no gRPC exporter, so `grpc` exports nothing * rather than sending HTTP to a gRPC port. */ -const serialized = ( - signal: "TRACES" | "METRICS", - self: Layer.Layer, -): Layer.Layer => +const exporters = (resource: OtlpResource) => Layer.unwrap( Effect.gen(function* () { - const protocol = yield* Config.String(`OTEL_EXPORTER_OTLP_${signal}_PROTOCOL`).pipe( - Config.orElse(() => Config.String("OTEL_EXPORTER_OTLP_PROTOCOL")), - Config.withDefault("http/protobuf"), - ); - switch (protocol.trim().toLowerCase()) { - case "http/json": - return Layer.provide(self, OtlpSerialization.layerJson); - case "http/protobuf": - return Layer.provide(self, OtlpSerialization.layerProtobuf); - default: - return OtlpExporter.layerFlusher; - } + const protocol = yield* Config.String("OTEL_EXPORTER_OTLP_PROTOCOL").pipe(Config.withDefault("http/protobuf")); + const format = { + "http/json": OtlpSerialization.layerJson, + "http/protobuf": OtlpSerialization.layerProtobuf, + }[protocol.trim().toLowerCase()]; + if (format === undefined) return Layer.empty; + return Layer.mergeAll( + OtlpTracer.layerFromConfig({ resource }), + OtlpMetrics.layerFromConfig({ resource }), + ).pipe(Layer.provide(format)); }), - ).pipe(Layer.orDie); + ); /** The release channel a version was published to, from its prerelease tag. */ const channel = (version: string) => @@ -69,7 +70,7 @@ const channel = (version: string) => const instanceId = uuidv7(); export const layer = (options: { readonly version: string; readonly command: string }) => { - const resource = { + const resource: OtlpResource = { serviceName: "codework", serviceVersion: options.version, attributes: { @@ -78,11 +79,7 @@ export const layer = (options: { readonly version: string; readonly command: str "codework.command": options.command, }, }; - return Layer.mergeAll( - serialized("TRACES", OtlpTracer.layerFromConfig({ resource })), - serialized("METRICS", OtlpMetrics.layerFromConfig({ resource })), - fiberMetrics, - ).pipe( + return Layer.mergeAll(exporters(resource), fiberMetrics).pipe( Layer.provide(FetchHttpClient.layer), Layer.provide(exporterDefaults), Layer.catchCause(() => Layer.empty), diff --git a/packages/codework/test/__artifacts__/telemetry.json b/packages/codework/test/__artifacts__/telemetry.json index 95e527fb..1b3e65e1 100644 --- a/packages/codework/test/__artifacts__/telemetry.json +++ b/packages/codework/test/__artifacts__/telemetry.json @@ -13,6 +13,8 @@ "parent": null, "error": false, "attributes": { + "gen_ai.agent.name": "codework", + "gen_ai.conversation.id": "", "gen_ai.operation.name": "invoke_agent" } }, @@ -23,15 +25,17 @@ "error": true, "attributes": { "error.type": "Runner.ProviderUnavailableError", + "gen_ai.conversation.id": "", "gen_ai.operation.name": "chat", "gen_ai.provider.name": "otel", "gen_ai.request.model": "otel-model", + "gen_ai.request.stream": true, "gen_ai.response.finish_reasons": [ "error" ], "gen_ai.response.model": "otel-model", - "gen_ai.usage.cache_creation.input_tokens": 0, "gen_ai.usage.cache_read.input_tokens": 0, + "gen_ai.usage.cache_write.input_tokens": 0, "gen_ai.usage.input_tokens": 0, "gen_ai.usage.output_tokens": 0 } @@ -42,19 +46,22 @@ "parent": "Loop.run", "error": false, "attributes": { + "gen_ai.conversation.id": "", "gen_ai.operation.name": "chat", "gen_ai.provider.name": "otel", "gen_ai.request.model": "otel-model", + "gen_ai.request.stream": true, "gen_ai.response.finish_reasons": [ "toolUse" ], "gen_ai.response.id": "", "gen_ai.response.model": "otel-model", - "gen_ai.usage.cache_creation.input_tokens": 0, + "gen_ai.response.time_to_first_chunk": "", "gen_ai.usage.cache_read.input_tokens": 0, + "gen_ai.usage.cache_write.input_tokens": 0, "gen_ai.usage.input_tokens": 120, "gen_ai.usage.output_tokens": 8, - "gen_ai.usage.reasoning_tokens": 0 + "gen_ai.usage.reasoning.output_tokens": 0 } }, { @@ -65,7 +72,7 @@ "attributes": { "gen_ai.operation.name": "execute_tool", "gen_ai.tool.call.id": "", - "gen_ai.tool.name": "read" + "gen_ai.tool.name": "bash" } }, { @@ -74,19 +81,22 @@ "parent": "Loop.run", "error": false, "attributes": { + "gen_ai.conversation.id": "", "gen_ai.operation.name": "chat", "gen_ai.provider.name": "otel", "gen_ai.request.model": "otel-model", + "gen_ai.request.stream": true, "gen_ai.response.finish_reasons": [ "stop" ], "gen_ai.response.id": "", "gen_ai.response.model": "otel-model", - "gen_ai.usage.cache_creation.input_tokens": 0, + "gen_ai.response.time_to_first_chunk": "", "gen_ai.usage.cache_read.input_tokens": 0, + "gen_ai.usage.cache_write.input_tokens": 0, "gen_ai.usage.input_tokens": 120, "gen_ai.usage.output_tokens": 8, - "gen_ai.usage.reasoning_tokens": 0 + "gen_ai.usage.reasoning.output_tokens": 0 } } ], @@ -127,156 +137,113 @@ } ] }, - "codework.llm.cost": { - "unit": "USD", - "type": "sum", - "series": [ - { - "attributes": { - "gen_ai.operation.name": "chat", - "gen_ai.provider.name": "otel", - "gen_ai.request.model": "otel-model", - "unit": "USD" - }, - "value": 0.00096 - } - ] - }, - "codework.llm.time_per_output_token": { + "codework.agent.duration": { "unit": "s", "type": "histogram", "series": [ { "attributes": { - "gen_ai.operation.name": "chat", - "gen_ai.provider.name": "otel", - "gen_ai.request.model": "otel-model", "unit": "s" }, - "count": 2 - } - ] - }, - "codework.llm.time_to_first_token": { - "unit": "s", - "type": "histogram", - "series": [ - { - "attributes": { - "gen_ai.operation.name": "chat", - "gen_ai.provider.name": "otel", - "gen_ai.request.model": "otel-model", - "unit": "s" - }, - "count": 2 - } - ] - }, - "codework.tool.output.truncated": { - "unit": "{call}", - "type": "sum", - "series": [ - { - "attributes": { - "gen_ai.operation.name": "execute_tool", - "gen_ai.tool.name": "read", - "unit": "{call}" - }, - "value": 1 + "count": 1 } ] }, - "codework.turn.retries": { - "unit": "{retry}", + "codework.llm.cost": { + "unit": "USD", "type": "sum", "series": [ { "attributes": { - "gen_ai.provider.name": "otel", - "gen_ai.request.model": "otel-model", - "unit": "{retry}" + "model": "otel-model", + "provider": "otel", + "unit": "USD" }, - "value": 1 + "value": 0.00096 } ] }, - "gen_ai.client.operation.duration": { + "codework.llm.duration": { "unit": "s", "type": "histogram", "series": [ { "attributes": { - "error.type": "Runner.ProviderUnavailableError", - "gen_ai.operation.name": "chat", - "gen_ai.provider.name": "otel", - "gen_ai.request.model": "otel-model", + "error": "Runner.ProviderUnavailableError", + "model": "otel-model", + "provider": "otel", "unit": "s" }, "count": 1 }, { "attributes": { - "gen_ai.operation.name": "chat", - "gen_ai.provider.name": "otel", - "gen_ai.request.model": "otel-model", + "model": "otel-model", + "provider": "otel", "unit": "s" }, "count": 2 } ] }, - "gen_ai.client.token.usage": { + "codework.llm.tokens": { "unit": "{token}", - "type": "histogram", + "type": "sum", "series": [ { "attributes": { - "gen_ai.operation.name": "chat", - "gen_ai.provider.name": "otel", - "gen_ai.request.model": "otel-model", - "gen_ai.token.type": "input", + "model": "otel-model", + "provider": "otel", + "type": "input", "unit": "{token}" }, - "count": 2, "value": 240 }, { "attributes": { - "gen_ai.operation.name": "chat", - "gen_ai.provider.name": "otel", - "gen_ai.request.model": "otel-model", - "gen_ai.token.type": "output", + "model": "otel-model", + "provider": "otel", + "type": "output", "unit": "{token}" }, - "count": 2, "value": 16 } ] }, - "gen_ai.execute_tool.duration": { + "codework.llm.ttft": { + "unit": "s", + "type": "histogram", + "series": [ + { + "attributes": { + "model": "otel-model", + "provider": "otel", + "unit": "s" + }, + "count": 2 + } + ] + }, + "codework.sandbox.exec.duration": { "unit": "s", "type": "histogram", "series": [ { "attributes": { - "gen_ai.operation.name": "execute_tool", - "gen_ai.tool.name": "read", - "outcome": "ok", + "driver": "local", "unit": "s" }, "count": 1 } ] }, - "gen_ai.invoke_agent.duration": { + "codework.tool.duration": { "unit": "s", "type": "histogram", "series": [ { "attributes": { - "gen_ai.operation.name": "invoke_agent", - "gen_ai.provider.name": "otel", - "gen_ai.request.model": "otel-model", + "tool": "bash", "unit": "s" }, "count": 1 diff --git a/packages/codework/test/telemetry.test.ts b/packages/codework/test/telemetry.test.ts index 00a71112..079e71eb 100644 --- a/packages/codework/test/telemetry.test.ts +++ b/packages/codework/test/telemetry.test.ts @@ -89,7 +89,7 @@ const chunk = (choice: object | undefined, usage?: object) => ({ }); /** - * The first successful request answers with one `read` call; the request that + * The first successful request answers with one `bash` call; the request that * carries its result answers with text. Both report usage, so every GenAI metric records. */ const completion = (body: { readonly messages: ReadonlyArray<{ readonly role: string }> }) => { @@ -110,7 +110,7 @@ const completion = (body: { readonly messages: ReadonlyArray<{ readonly role: st index: 0, id: "call_otel", type: "function", - function: { name: "read", arguments: '{"path":"note.txt"}' }, + function: { name: "bash", arguments: '{"command":"cat note.txt"}' }, }, ], }, @@ -186,8 +186,15 @@ const value = (input: OtlpValue): unknown => { return (input.arrayValue?.values ?? []).map(value); }; -/** Ids and versions change per run; everything else in the artifact is fixed by the fake provider. */ -const volatile = new Set(["gen_ai.tool.call.id", "gen_ai.response.id", "service.instance.id", "service.version"]); +/** Ids, versions and timings change per run; everything else in the artifact is fixed by the fake provider. */ +const volatile = new Set([ + "gen_ai.conversation.id", + "gen_ai.tool.call.id", + "gen_ai.response.id", + "gen_ai.response.time_to_first_chunk", + "service.instance.id", + "service.version", +]); const attributes = (input: ReadonlyArray | undefined) => Object.fromEntries( @@ -228,12 +235,7 @@ const genAiSpans = (exports: ReadonlyArray) => { }; // Counts and sums that the fake provider fixes; every other recorded value follows timing. -const exact = new Set([ - "gen_ai.client.token.usage", - "codework.llm.cost", - "codework.turn.retries", - "codework.tool.output.truncated", -]); +const exact = new Set(["codework.llm.tokens", "codework.llm.cost"]); const runtime = (name: string) => name.startsWith("child_fiber"); const recorded = (name: string, point: OtlpDataPoint) => { @@ -299,11 +301,7 @@ describe("OTLP export", () => { const settings = join(home, "settings"); mkdirSync(project); mkdirSync(settings); - // Longer than `read`'s line limit, so its result is truncated. - writeFileSync( - join(project, "note.txt"), - Array.from({ length: 2500 }, (_, index) => `hello ${index}`).join("\n"), - ); + writeFileSync(join(project, "note.txt"), "hello from the note\n"); writeFileSync( join(settings, "settings.jsonc"), JSON.stringify({ @@ -360,7 +358,12 @@ describe("OTLP export", () => { // No prompt, file content, or model output may leave in any span, attribute, or event. const traces = JSON.stringify(collector.exports.filter((entry) => entry.path === "/v1/traces")); - for (const content of ["What does note.txt say?", "hello 1", "The note says hello."]) { + for (const content of [ + "What does note.txt say?", + "cat note.txt", + "hello from the note", + "The note says hello.", + ]) { expect(traces).not.toContain(content); } diff --git a/packages/harness/src/runner/llm.ts b/packages/harness/src/runner/llm.ts index b73e0e8e..f3d1309d 100644 --- a/packages/harness/src/runner/llm.ts +++ b/packages/harness/src/runner/llm.ts @@ -231,48 +231,43 @@ const requestErrorType = (input: RequestInput, exit: Exit.Exit, ) { const now = yield* Effect.clockWith((clock) => clock.currentTimeMillis); - const attributes = { - "gen_ai.operation.name": "chat", - "gen_ai.provider.name": input.provider, - "gen_ai.request.model": input.model, - }; + const labels = { provider: input.provider, model: input.model }; const errorType = requestErrorType(input, exit); yield* Telemetry.record( - Telemetry.operationDuration, + Telemetry.llmDuration, (now - timing.start) / 1000, - errorType === undefined ? attributes : { ...attributes, "error.type": errorType }, + errorType === undefined ? labels : { ...labels, error: errorType }, ); if (timing.firstToken !== undefined) { - yield* Telemetry.record(Telemetry.timeToFirstToken, (timing.firstToken - timing.start) / 1000, attributes); + const ttft = (timing.firstToken - timing.start) / 1000; + yield* Telemetry.record(Telemetry.llmTtft, ttft, labels); + yield* Effect.annotateCurrentSpan("gen_ai.response.time_to_first_chunk", ttft); } if (errorType !== undefined) yield* Effect.annotateCurrentSpan("error.type", errorType); if (Exit.isFailure(exit)) return; - const { message, outcome } = exit.value; + const { message } = exit.value; const { usage } = message; - // aikit's `input` excludes cached tokens; the conventions count them as input. + // aikit's `input` excludes cached tokens; input counts them, as the GenAI conventions do. const inputTokens = usage.input + usage.cacheRead + usage.cacheWrite; - // A request that failed before the provider reported usage carries zeros, which are not observations. - if (inputTokens > 0 || usage.output > 0) { - yield* Telemetry.record(Telemetry.tokenUsage, inputTokens, { ...attributes, "gen_ai.token.type": "input" }); - yield* Telemetry.record(Telemetry.tokenUsage, usage.output, { ...attributes, "gen_ai.token.type": "output" }); - } - if (usage.cost.total > 0) yield* Telemetry.record(Telemetry.cost, usage.cost.total, attributes); - // The interval after the first token covers every token but that one. - if (outcome === "ended" && timing.firstToken !== undefined && usage.output > 1) { - yield* Telemetry.record( - Telemetry.timePerOutputToken, - (now - timing.firstToken) / 1000 / (usage.output - 1), - attributes, - ); + const tokens = { + input: inputTokens, + output: usage.output, + cache_read: usage.cacheRead, + cache_write: usage.cacheWrite, + reasoning: usage.reasoning ?? 0, + }; + for (const [type, count] of Object.entries(tokens)) { + if (count > 0) yield* Telemetry.record(Telemetry.llmTokens, count, { ...labels, type }); } + if (usage.cost.total > 0) yield* Telemetry.record(Telemetry.llmCost, usage.cost.total, labels); yield* Effect.annotateCurrentSpan({ "gen_ai.response.model": message.responseModel ?? message.model, ...(message.responseId === undefined ? {} : { "gen_ai.response.id": message.responseId }), @@ -280,8 +275,8 @@ const observe = Effect.fnUntraced(function* ( "gen_ai.usage.input_tokens": inputTokens, "gen_ai.usage.output_tokens": usage.output, "gen_ai.usage.cache_read.input_tokens": usage.cacheRead, - "gen_ai.usage.cache_creation.input_tokens": usage.cacheWrite, - ...(usage.reasoning === undefined ? {} : { "gen_ai.usage.reasoning_tokens": usage.reasoning }), + "gen_ai.usage.cache_write.input_tokens": usage.cacheWrite, + ...(usage.reasoning === undefined ? {} : { "gen_ai.usage.reasoning.output_tokens": usage.reasoning }), }); }); @@ -308,6 +303,8 @@ export const make = (openStream: Open): Request => { "gen_ai.operation.name": "chat", "gen_ai.provider.name": input.provider, "gen_ai.request.model": input.model, + "gen_ai.conversation.id": input.sessionId, + "gen_ai.request.stream": true, }); const start = yield* Effect.clockWith((clock) => clock.currentTimeMillis); let firstToken: number | undefined; diff --git a/packages/harness/src/runner/loop.ts b/packages/harness/src/runner/loop.ts index 603e39ee..584dea93 100644 --- a/packages/harness/src/runner/loop.ts +++ b/packages/harness/src/runner/loop.ts @@ -67,47 +67,21 @@ const errorPart = ( time: { ...call.time, end: endTime(call, now) }, }); -const agentAttributes = (snapshot: State.Snapshot) => ({ - "gen_ai.operation.name": "invoke_agent", - "gen_ai.provider.name": snapshot.provider, - "gen_ai.request.model": snapshot.model, -}); - -// Built-in tools report truncation as `details.truncation` (read, find, grep, ls) or `details.truncated` (bash). -const isTruncated = (details: unknown) => { - if (typeof details !== "object" || details === null) return false; - if ("truncated" in details && details.truncated === true) return true; - return ( - "truncation" in details && - typeof details.truncation === "object" && - details.truncation !== null && - "truncated" in details.truncation && - details.truncation.truncated === true - ); -}; - const observeTool = ( snapshot: State.Snapshot, call: Message.ToolCallPendingPart, seconds: number, exit: Exit.Exit, -) => - Effect.gen(function* () { - const outcome = Exit.isFailure(exit) - ? Cause.hasInterruptsOnly(exit.cause) - ? "interrupted" - : "error" - : exit.value.status === "completed" - ? "ok" - : "error"; - // The model names the tool; an unregistered name would open a new series per hallucination. - const name = snapshot.tools.defs.some((def) => def.name === call.name) ? call.name : "_OTHER"; - const attributes = { "gen_ai.operation.name": "execute_tool", "gen_ai.tool.name": name }; - yield* Telemetry.record(Telemetry.toolDuration, seconds, { ...attributes, outcome }); - if (Exit.isSuccess(exit) && isTruncated(exit.value.result.details)) { - yield* Telemetry.record(Telemetry.truncated, 1, attributes); - } - }); +) => { + // The model names the tool; an unregistered name would open a new series per hallucination. + const tool = snapshot.tools.defs.some((def) => def.name === call.name) ? call.name : "_OTHER"; + const labels = Exit.isFailure(exit) + ? { tool, error: Telemetry.errorType(exit.cause) } + : exit.value.status === "completed" + ? { tool } + : { tool, error: exit.value.status }; + return Telemetry.record(Telemetry.toolDuration, seconds, labels); +}; export const layer = (options: Options = {}) => Layer.effect( @@ -368,10 +342,6 @@ export const layer = (options: Options = {}) => if (delayMs === undefined) return Effect.fail(error); return Effect.gen(function* () { retries += 1; - yield* Telemetry.record(Telemetry.retries, 1, { - "gen_ai.provider.name": snapshot.provider, - "gen_ai.request.model": snapshot.model, - }); yield* Effect.uninterruptible(settleDraft({ _tag: "error", message: error.message })); yield* events.publish(EventList.RetryScheduled, { timestamp: yield* DateTime.now, @@ -454,16 +424,17 @@ export const layer = (options: Options = {}) => const hasSteer = yield* inputs.hasPending(input.sessionId, "steer"); const hasFollowUp = hasSteer ? false : yield* inputs.hasPending(input.sessionId, "followUp"); if (!input.force && !hasSteer && !hasFollowUp) return; - yield* Effect.annotateCurrentSpan("gen_ai.operation.name", "invoke_agent"); + yield* Effect.annotateCurrentSpan({ + "gen_ai.operation.name": "invoke_agent", + "gen_ai.agent.name": "codework", + "gen_ai.conversation.id": input.sessionId, + }); let promotion: Promotion = hasSteer ? "steer" : hasFollowUp ? "followUp" : undefined; - // A follow-up re-reads State, so the run is attributed to the model it last used. - let latest: State.Snapshot | undefined; const drain = Effect.gen(function* () { let shouldRun = true; while (shouldRun) { const snapshot = yield* state.snapshot(input.sessionId); - latest = snapshot; let needsContinuation = true; while (needsContinuation) { const result = yield* runTurn(promotion, snapshot); @@ -476,15 +447,7 @@ export const layer = (options: Options = {}) => } }); yield* Telemetry.timed(drain, (seconds, exit) => - latest === undefined - ? Effect.void - : Telemetry.record( - Telemetry.agentDuration, - seconds, - Exit.isSuccess(exit) - ? agentAttributes(latest) - : { ...agentAttributes(latest), "error.type": Telemetry.errorType(exit.cause) }, - ), + Telemetry.record(Telemetry.agentDuration, seconds, Telemetry.withError({}, exit)), ); }); diff --git a/packages/harness/src/sandbox/control.ts b/packages/harness/src/sandbox/control.ts index 8f7e423d..09b4cf3d 100644 --- a/packages/harness/src/sandbox/control.ts +++ b/packages/harness/src/sandbox/control.ts @@ -4,12 +4,12 @@ import { DateTime, type Duration, Effect, - Exit, Layer, LayerMap, Option, Schema, Semaphore, + Stream, } from "effect"; import { SqlClient } from "effect/sql"; import { SandboxInstanceRow } from "../db/schema.sql.ts"; @@ -46,7 +46,7 @@ import { SandboxInstance } from "./instance.ts"; import { SandboxIO } from "./io.ts"; import { SandboxDriverRegistry } from "./registry.ts"; import { HostExe } from "./shell/host.ts"; -import { withCwd as shellWithCwd } from "./shell/shell.ts"; +import { type ISandboxExe, withCwd as shellWithCwd } from "./shell/shell.ts"; import { SandboxStore } from "./store.ts"; export interface CreateInput { @@ -141,16 +141,43 @@ const parseJson = (driver: SandboxDriver.Name, operation: string, value: string) /** Time a driver's provider-side create or wake, the cold start a remote sandbox pays. */ const provision = (driver: SandboxDriver.Name, operation: "create" | "wake", self: Effect.Effect) => - Telemetry.timed(self, (seconds, exit) => { - const attributes = { "codework.sandbox.driver": driver, "codework.sandbox.operation": operation }; - return Telemetry.record( - Telemetry.provisionDuration, - seconds, - Exit.isSuccess(exit) - ? { ...attributes, outcome: "ok" } - : { ...attributes, outcome: "error", "error.type": Telemetry.errorType(exit.cause) }, + Telemetry.timed(self, (seconds, exit) => + Telemetry.record(Telemetry.sandboxProvisionDuration, seconds, Telemetry.withError({ driver, operation }, exit)), + ); + +/** Time every command a mount's shell runs, labelled by the driver that runs it. */ +const timedShell = (shell: ISandboxExe, driver: SandboxDriver.Name): ISandboxExe => { + const time = (self: Effect.Effect) => + Telemetry.timed(self, (seconds, exit) => + Telemetry.record(Telemetry.sandboxExecDuration, seconds, Telemetry.withError({ driver }, exit)), ); - }); + const stream = shell.stream; + return { + exec: (command, options) => time(shell.exec(command, options)), + execArgv: (argv, options) => time(shell.execArgv(argv, options)), + ...(stream === undefined + ? {} + : { + stream: (command, options) => + Stream.unwrap( + Effect.clockWith((clock) => { + const start = clock.currentTimeMillisUnsafe(); + return Effect.succeed( + stream(command, options).pipe( + Stream.onExit((exit) => + Telemetry.record( + Telemetry.sandboxExecDuration, + (clock.currentTimeMillisUnsafe() - start) / 1000, + Telemetry.withError({ driver }, exit), + ), + ), + ), + ); + }), + ), + }), + }; +}; export const make = Effect.fn("Sandbox.Controller.make")(function* (options: Options) { const sql = yield* SqlClient.SqlClient; @@ -167,29 +194,40 @@ export const make = Effect.fn("Sandbox.Controller.make")(function* (options: Opt interface Lease { held: boolean; mounted: boolean; + driver: SandboxDriver.Name | undefined; } + // A sandbox counts as active from its first lease to its last release. + const activeChanged = (lease: Lease, delta: 1 | -1) => + lease.driver === undefined + ? Effect.void + : Telemetry.adjust(Telemetry.sandboxActive, delta, { driver: lease.driver }); + const acquireLease = (id: SandboxInstance.ID, lease: Lease) => - gate.withPermits(1)( - Effect.sync(() => { - const wasIdle = refCount(id) === 0; - refs.set(id, refCount(id) + 1); - lease.held = true; - return wasIdle; - }), - ); + gate + .withPermits(1)( + Effect.sync(() => { + const wasIdle = refCount(id) === 0; + refs.set(id, refCount(id) + 1); + lease.held = true; + return wasIdle; + }), + ) + .pipe(Effect.tap((wasIdle) => (wasIdle ? activeChanged(lease, 1) : Effect.void))); const releaseLease = (id: SandboxInstance.ID, lease: Lease) => Effect.gen(function* () { if (!lease.held) return; - yield* gate.withPermits(1)( + const idle = yield* gate.withPermits(1)( Effect.sync(() => { const next = Math.max(0, refCount(id) - 1); if (next === 0) refs.delete(id); else refs.set(id, next); lease.held = false; + return next === 0; }), ); + if (idle) yield* activeChanged(lease, -1); if (!lease.mounted) return; const now = yield* Effect.clockWith((clock) => clock.currentTimeMillis); yield* sql` @@ -557,7 +595,7 @@ export const make = Effect.fn("Sandbox.Controller.make")(function* (options: Opt Effect.gen(function* () { let identity: SandboxIO.Identity; let invalidateBeforeAttach = false; - const lease: Lease = reserved ?? { held: false, mounted: false }; + const lease: Lease = reserved ?? { held: false, mounted: false, driver: undefined }; if (id === SandboxInstance.ID.local) { identity = { @@ -586,6 +624,7 @@ export const make = Effect.fn("Sandbox.Controller.make")(function* (options: Opt } invalidateBeforeAttach = row.status !== "online"; const attached = yield* runtime(row, "mount"); + lease.driver = attached.driver.name; identity = { id, driver: attached.driver.name, @@ -617,7 +656,10 @@ export const make = Effect.fn("Sandbox.Controller.make")(function* (options: Opt return idle; }), ); - if (reserved === undefined) yield* Effect.addFinalizer(() => releaseLease(id, lease)); + if (reserved === undefined) { + yield* Effect.addFinalizer(() => releaseLease(id, lease)); + if (wasIdle) yield* activeChanged(lease, 1); + } if (wasIdle && invalidateBeforeAttach) yield* transports.invalidate(id); } @@ -668,7 +710,7 @@ export const make = Effect.fn("Sandbox.Controller.make")(function* (options: Opt return Layer.mergeAll( Layer.succeed(SandboxIO.Current, identity), Layer.succeed(SandboxIO.FileSystem, SandboxFileSystem.withCwd(fs, identity.cwd)), - Layer.succeed(SandboxIO.Shell, shellWithCwd(shell, identity.cwd)), + Layer.succeed(SandboxIO.Shell, timedShell(shellWithCwd(shell, identity.cwd), identity.driver)), ).pipe(SandboxIO.withMutation); }), ); @@ -894,7 +936,7 @@ export const make = Effect.fn("Sandbox.Controller.make")(function* (options: Opt const createAndMount: Interface["createAndMount"] = (input, mountOptions) => { const id = input.instanceId ?? SandboxInstance.ID.create(); - const lease: Lease = { held: false, mounted: false }; + const lease: Lease = { held: false, mounted: false, driver: input.driver.name }; return Layer.unwrap( Effect.gen(function* () { // Reserve before the provider call. Once create inserts the diff --git a/packages/harness/src/telemetry/telemetry.ts b/packages/harness/src/telemetry/telemetry.ts index bcbf0a8d..79062fe6 100644 --- a/packages/harness/src/telemetry/telemetry.ts +++ b/packages/harness/src/telemetry/telemetry.ts @@ -1,81 +1,73 @@ /* - * @file Harness metrics and the GenAI span vocabulary. + * @file Harness metrics. * * Effect owns the instruments, the spans, and the export; this module only names - * them. Names follow the OpenTelemetry GenAI conventions where one exists, and - * Effect's OTLP exporter reads a metric's unit from its `unit` attribute. + * them. Metrics use codework's own names and short labels, so a moving draft + * convention never renames a dashboard. Effect's OTLP exporter reads a metric's + * unit from its `unit` attribute. */ -import { Cause, Effect, type Exit, Metric } from "effect"; +import { Cause, Effect, Exit, Metric } from "effect"; -// Bucket boundaries are the GenAI conventions' advice for each instrument. const durationBuckets = Metric.boundariesFromIterable([ 0.01, 0.02, 0.04, 0.08, 0.16, 0.32, 0.64, 1.28, 2.56, 5.12, 10.24, 20.48, 40.96, 81.92, ]); -export const tokenUsage = Metric.histogram("gen_ai.client.token.usage", { - description: "Tokens used per provider request", - attributes: { unit: "{token}" }, - boundaries: Metric.boundariesFromIterable([ - 1, 4, 16, 64, 256, 1024, 4096, 16384, 65536, 262144, 1048576, 4194304, 16777216, 67108864, - ]), -}); - -export const operationDuration = Metric.histogram("gen_ai.client.operation.duration", { +export const llmDuration = Metric.histogram("codework.llm.duration", { description: "Provider request duration", attributes: { unit: "s" }, boundaries: durationBuckets, }); -// The `gen_ai.server.*` timings are a model server's own measure; a client also times the transport. -export const timeToFirstToken = Metric.histogram("codework.llm.time_to_first_token", { - description: "Time from request start to the first streamed content", +export const llmTtft = Metric.histogram("codework.llm.ttft", { + description: "Time to first token: from request start to the first streamed content", attributes: { unit: "s" }, boundaries: Metric.boundariesFromIterable([ 0.001, 0.005, 0.01, 0.02, 0.04, 0.06, 0.08, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10, ]), }); -export const timePerOutputToken = Metric.histogram("codework.llm.time_per_output_token", { - description: "Generation time after the first token, per later output token", - attributes: { unit: "s" }, - boundaries: Metric.boundariesFromIterable([0.01, 0.025, 0.05, 0.075, 0.1, 0.15, 0.2, 0.3, 0.4, 0.5, 0.75, 1, 2.5]), +export const llmTokens = Metric.counter("codework.llm.tokens", { + description: "Tokens used, by type; cache and reasoning tokens are part of input and output", + attributes: { unit: "{token}" }, + incremental: true, }); -export const toolDuration = Metric.histogram("gen_ai.execute_tool.duration", { - description: "Tool execution duration", - attributes: { unit: "s" }, - boundaries: durationBuckets, +export const llmCost = Metric.counter("codework.llm.cost", { + description: "Provider request cost from the model catalog's prices", + attributes: { unit: "USD" }, + incremental: true, }); -export const agentDuration = Metric.histogram("gen_ai.invoke_agent.duration", { +export const agentDuration = Metric.histogram("codework.agent.duration", { description: "Session run duration, across every turn it drains", attributes: { unit: "s" }, - boundaries: durationBuckets, + boundaries: Metric.boundariesFromIterable([ + 0.1, 0.2, 0.4, 0.8, 1.6, 3.2, 6.4, 12.8, 25.6, 51.2, 102.4, 204.8, 409.6, + ]), }); -export const cost = Metric.counter("codework.llm.cost", { - description: "Provider request cost from the model catalog's prices", - attributes: { unit: "USD" }, - incremental: true, +export const toolDuration = Metric.histogram("codework.tool.duration", { + description: "Tool execution duration", + attributes: { unit: "s" }, + boundaries: durationBuckets, }); -export const retries = Metric.counter("codework.turn.retries", { - description: "Provider requests retried within a turn", - attributes: { unit: "{retry}" }, - incremental: true, +export const sandboxProvisionDuration = Metric.histogram("codework.sandbox.provision.duration", { + description: "Sandbox driver create and wake duration", + attributes: { unit: "s" }, + boundaries: durationBuckets, }); -export const provisionDuration = Metric.histogram("codework.sandbox.provision.duration", { - description: "Sandbox driver create and wake duration", +export const sandboxExecDuration = Metric.histogram("codework.sandbox.exec.duration", { + description: "Sandbox shell command duration", attributes: { unit: "s" }, boundaries: durationBuckets, }); -export const truncated = Metric.counter("codework.tool.output.truncated", { - description: "Tool results whose output was truncated before the model saw it", - attributes: { unit: "{call}" }, - incremental: true, +export const sandboxActive = Metric.gauge("codework.sandbox.active", { + description: "Managed and remote sandboxes with at least one mount", + attributes: { unit: "{sandbox}" }, }); export type Attributes = Readonly>; @@ -83,7 +75,11 @@ export type Attributes = Readonly>; export const record = (metric: Metric.Metric, value: Input, attributes: Attributes) => Metric.update(Metric.withAttributes(metric, attributes), value); -/** `error.type` for a failed exit: the error's own tag, or `interrupted`. */ +/** Add `delta` to a gauge rather than setting it. */ +export const adjust = (metric: Metric.Metric, delta: number, attributes: Attributes) => + Metric.modify(Metric.withAttributes(metric, attributes), delta); + +/** The `error` label for a failed exit: the error's own tag, or `interrupted`. */ export const errorType = (cause: Cause.Cause): string => { if (Cause.hasInterruptsOnly(cause)) return "interrupted"; const error = Cause.squash(cause); @@ -92,6 +88,10 @@ export const errorType = (cause: Cause.Cause): string => { : "_OTHER"; }; +/** `attributes`, plus `error` when the exit failed. */ +export const withError = (attributes: Attributes, exit: Exit.Exit): Attributes => + Exit.isSuccess(exit) ? attributes : { ...attributes, error: errorType(exit.cause) }; + /** Run `self`, then hand its duration in seconds and its exit to `observe`. */ export const timed = ( self: Effect.Effect, From 40e75ffb541c0574345eaf3eef55a725b14b0d9d Mon Sep 17 00:00:00 2001 From: Sanchit Rk <0xsanchit@gmail.com> Date: Tue, 13 Oct 2026 16:59:09 +0530 Subject: [PATCH 4/6] feat: export only high-level spans, CODEWORK_TRACE_LEVEL for the rest A single run produced ~500 spans, a third of them Effect's own sql.execute. Spans now default to Debug and export starts at Info. The command, agent loop, model and tool calls, and sandbox create/wake are marked Info, so a run exports about a dozen spans. CODEWORK_TRACE_LEVEL=debug exports every span again. Effect drops a span's children with it, so every span on the path from a command to a model or tool call is Info, Loop.settleTools included. The E2E test pins the exported span tree. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 2 +- packages/codework/src/cli/cmd/handlers/acp.ts | 2 +- packages/codework/src/cli/cmd/handlers/run.ts | 2 +- .../codework/src/cli/cmd/handlers/serve.ts | 2 +- packages/codework/src/cli/cmd/handlers/tui.ts | 2 +- packages/codework/src/framework/otel.ts | 24 +++++++++++++++++-- .../test/__artifacts__/telemetry.json | 9 +++++++ packages/codework/test/telemetry.test.ts | 16 +++++++++++++ packages/harness/src/effect/sandbox.ts | 4 ++-- packages/harness/src/runner/llm.ts | 2 +- packages/harness/src/runner/loop.ts | 15 ++++++++---- packages/harness/src/sandbox/control.ts | 4 ++-- 12 files changed, 67 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ea1263a7..9ea18420 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,7 +13,7 @@ This file is the canonical source for unreleased changes and published release n ### Added -- `@codeworksh/cli`: Added OpenTelemetry export. Set `OTEL_EXPORTER_OTLP_ENDPOINT` and every command sends its traces and metrics over OTLP/HTTP, as protobuf by default or JSON with `OTEL_EXPORTER_OTLP_PROTOCOL=http/json`. The standard `OTEL_*` variables configure it (headers, per-signal endpoints, `OTEL_SDK_DISABLED`, `OTEL_SERVICE_NAME`, `OTEL_RESOURCE_ATTRIBUTES`); there is no settings key. Without an endpoint nothing is exported, and a bad configuration never stops a command. +- `@codeworksh/cli`: Added OpenTelemetry export. Set `OTEL_EXPORTER_OTLP_ENDPOINT` and every command sends its traces and metrics over OTLP/HTTP, as protobuf by default or JSON with `OTEL_EXPORTER_OTLP_PROTOCOL=http/json`. The standard `OTEL_*` variables configure it (headers, per-signal endpoints, `OTEL_SDK_DISABLED`, `OTEL_SERVICE_NAME`, `OTEL_RESOURCE_ATTRIBUTES`); there is no settings key. Without an endpoint nothing is exported, and a bad configuration never stops a command. Only the high-level spans are exported (command, agent loop, model and tool calls, sandbox create and wake); set `CODEWORK_TRACE_LEVEL=debug` to export every span, including database queries. - `@codeworksh/harness`: Spans for model requests, tool calls and agent runs carry the OpenTelemetry GenAI attributes (provider, model, session, response id, finish reason, token counts, time to first token), and a failed model request ends its span with an error. Metrics: `codework.llm.duration`, `codework.llm.ttft`, `codework.llm.tokens` (by `type`: input, output, cache_read, cache_write, reasoning), `codework.llm.cost`, `codework.agent.duration`, `codework.tool.duration` (by `tool`), `codework.sandbox.provision.duration`, `codework.sandbox.exec.duration` and `codework.sandbox.active` (by `driver`), plus Effect's fiber runtime metrics (`child_fibers_*`). Prompts, file contents and model output are never exported. ### Fixed diff --git a/packages/codework/src/cli/cmd/handlers/acp.ts b/packages/codework/src/cli/cmd/handlers/acp.ts index d31aa861..4ab355e4 100644 --- a/packages/codework/src/cli/cmd/handlers/acp.ts +++ b/packages/codework/src/cli/cmd/handlers/acp.ts @@ -7,7 +7,7 @@ import { Cmd } from "../cmd.ts"; export default Runtime.handler( Cmd.commands.acp, - Effect.fn("CLI.acp")(function* () { + Effect.fn("CLI.acp", { level: "Info" })(function* () { const shared = yield* Cmd.spec; return yield* Server.serve(harnessOptions(shared)).pipe(Effect.catch(reportFailure)); }), diff --git a/packages/codework/src/cli/cmd/handlers/run.ts b/packages/codework/src/cli/cmd/handlers/run.ts index 6225c20f..f4f917e9 100644 --- a/packages/codework/src/cli/cmd/handlers/run.ts +++ b/packages/codework/src/cli/cmd/handlers/run.ts @@ -77,7 +77,7 @@ const selectSandbox = Effect.fn("CLI.selectSandbox")(function* (input: Sandbox.S export default Runtime.handler( Cmd.commands.run, - Effect.fn("CLI.run")(function* ({ + Effect.fn("CLI.run", { level: "Info" })(function* ({ prompt, session, cwd, diff --git a/packages/codework/src/cli/cmd/handlers/serve.ts b/packages/codework/src/cli/cmd/handlers/serve.ts index abca6911..fb014213 100644 --- a/packages/codework/src/cli/cmd/handlers/serve.ts +++ b/packages/codework/src/cli/cmd/handlers/serve.ts @@ -16,7 +16,7 @@ const report = (error: unknown) => writeError(renderError(error)).pipe(Effect.an export default Runtime.handler( Cmd.commands.serve, - Effect.fn("CLI.serve")(function* ({ host, port, service }) { + Effect.fn("CLI.serve", { level: "Info" })(function* ({ host, port, service }) { const shared = yield* Cmd.spec; const harness = { ...harnessOptions(shared), watchModels: true }; const serve = Layer.launch(Server.layer({ host, port, harness })); diff --git a/packages/codework/src/cli/cmd/handlers/tui.ts b/packages/codework/src/cli/cmd/handlers/tui.ts index fbce2615..3c6ecfb5 100644 --- a/packages/codework/src/cli/cmd/handlers/tui.ts +++ b/packages/codework/src/cli/cmd/handlers/tui.ts @@ -73,7 +73,7 @@ const run = (binary: string, args: ReadonlyArray, env: Record Layer.empty), diff --git a/packages/codework/test/__artifacts__/telemetry.json b/packages/codework/test/__artifacts__/telemetry.json index 1b3e65e1..0d571ab1 100644 --- a/packages/codework/test/__artifacts__/telemetry.json +++ b/packages/codework/test/__artifacts__/telemetry.json @@ -6,6 +6,15 @@ "service.name": "codework", "service.version": "" }, + "exported": [ + "(root) > CLI.run", + "CLI.run > Loop.run", + "Loop.run > Loop.runTurn", + "Loop.runTurn > Loop.runTurnAttempt", + "Loop.runTurnAttempt > LLM.run", + "Loop.runTurnAttempt > Loop.settleTools", + "Loop.settleTools > Loop.handleTool" + ], "spans": [ { "name": "Loop.run", diff --git a/packages/codework/test/telemetry.test.ts b/packages/codework/test/telemetry.test.ts index 079e71eb..dda2513b 100644 --- a/packages/codework/test/telemetry.test.ts +++ b/packages/codework/test/telemetry.test.ts @@ -287,6 +287,20 @@ const metrics = (exports: ReadonlyArray) => { ); }; +/** + * Every exported span name, with its parent's. By default only the spans marked + * `Info` leave the process: Effect's own `sql.*` and the harness internals stay out. + */ +const exportedSpans = (exports: ReadonlyArray) => { + const spans = exports.flatMap((entry) => + (entry.body.resourceSpans ?? []).flatMap((resource) => resource.scopeSpans.flatMap((scope) => scope.spans)), + ); + const byId = new Map(spans.map((span) => [span.spanId, span])); + return [ + ...new Set(spans.map((span) => `${byId.get(span.parentSpanId ?? "")?.name ?? "(root)"} > ${span.name}`)), + ].sort(); +}; + /** Every exported resource, traces and metrics alike, must describe the same process. */ const resources = (exports: ReadonlyArray) => exports.flatMap((entry) => [...(entry.body.resourceSpans ?? []), ...(entry.body.resourceMetrics ?? [])]); @@ -341,6 +355,7 @@ describe("OTLP export", () => { for (const key of Object.keys(env)) { if (key.startsWith("OTEL_") && !configured.has(key)) delete env[key]; } + delete env.CODEWORK_TRACE_LEVEL; const result = await runCli( ["--home", home, "--database", ":memory:", "--user-config-dir", settings, "run", "What does note.txt say?"], @@ -371,6 +386,7 @@ describe("OTLP export", () => { `${JSON.stringify( { resource: attributes(resources(collector.exports)[0]?.resource?.attributes), + exported: exportedSpans(collector.exports), spans: genAiSpans(collector.exports), metrics: metrics(collector.exports), }, diff --git a/packages/harness/src/effect/sandbox.ts b/packages/harness/src/effect/sandbox.ts index c965dc23..72fc77c0 100644 --- a/packages/harness/src/effect/sandbox.ts +++ b/packages/harness/src/effect/sandbox.ts @@ -47,7 +47,7 @@ const runtimeOverrides = (driver: SandboxDriver.Name, value: unknown) => { return Effect.succeed(value as Readonly>); }; -export const create = Effect.fn("Sandbox.create")(function* (input: CreateInput) { +export const create = Effect.fn("Sandbox.create", { level: "Info" })(function* (input: CreateInput) { const controller = yield* SandboxController.Controller; const registry = yield* SandboxDriverRegistry.Registry; const driverName = name(input.driver); @@ -133,7 +133,7 @@ export const refresh = Effect.fn("Sandbox.refresh")(function* (id: SandboxInstan return yield* controller.refresh(id); }); -export const wake = Effect.fn("Sandbox.wake")(function* (id: SandboxInstance.ID) { +export const wake = Effect.fn("Sandbox.wake", { level: "Info" })(function* (id: SandboxInstance.ID) { const controller = yield* SandboxController.Controller; return yield* controller.wake(id); }); diff --git a/packages/harness/src/runner/llm.ts b/packages/harness/src/runner/llm.ts index f3d1309d..a67c7ecb 100644 --- a/packages/harness/src/runner/llm.ts +++ b/packages/harness/src/runner/llm.ts @@ -298,7 +298,7 @@ class FailedTerminal extends Data.TaggedError("LLM.FailedTerminal")<{ * aikit event streams; production uses `open` above. */ export const make = (openStream: Open): Request => { - const request = Effect.fn("LLM.run", { kind: "client" })(function* (input: RequestInput) { + const request = Effect.fn("LLM.run", { kind: "client", level: "Info" })(function* (input: RequestInput) { yield* Effect.annotateCurrentSpan({ "gen_ai.operation.name": "chat", "gen_ai.provider.name": input.provider, diff --git a/packages/harness/src/runner/loop.ts b/packages/harness/src/runner/loop.ts index 584dea93..9ada716b 100644 --- a/packages/harness/src/runner/loop.ts +++ b/packages/harness/src/runner/loop.ts @@ -133,7 +133,7 @@ export const layer = (options: Options = {}) => yield* publishFailure(draft, { _tag: "interrupted" }); }); - const settleTools = Effect.fn("Loop.settleTools")(function* ( + const settleTools = Effect.fn("Loop.settleTools", { level: "Info" })(function* ( snapshot: State.Snapshot, message: Message.AssistantMessage, reason: "stop" | "length" | "toolUse", @@ -187,7 +187,9 @@ export const layer = (options: Options = {}) => } const settled = new Set(); - const handle = Effect.fn("Loop.handleTool")(function* (call: Message.ToolCallPendingPart) { + const handle = Effect.fn("Loop.handleTool", { level: "Info" })(function* ( + call: Message.ToolCallPendingPart, + ) { yield* Effect.annotateCurrentSpan({ "gen_ai.operation.name": "execute_tool", "gen_ai.tool.name": call.name, @@ -262,7 +264,7 @@ export const layer = (options: Options = {}) => return interruptedCause; }); - const runTurnAttempt = Effect.fn("Loop.runTurnAttempt")(function* ( + const runTurnAttempt = Effect.fn("Loop.runTurnAttempt", { level: "Info" })(function* ( promotion: Promotion, snapshot: State.Snapshot, ) { @@ -411,11 +413,14 @@ export const layer = (options: Options = {}) => ); }); - const runTurn = Effect.fn("Loop.runTurn")(function* (promotion: Promotion, snapshot: State.Snapshot) { + const runTurn = Effect.fn("Loop.runTurn", { level: "Info" })(function* ( + promotion: Promotion, + snapshot: State.Snapshot, + ) { return yield* runTurnAttempt(promotion, snapshot); }); - const run = Effect.fn("Loop.run")(function* (input: { + const run = Effect.fn("Loop.run", { level: "Info" })(function* (input: { readonly sessionId: SessionSchema.ID; readonly force: boolean; }) { diff --git a/packages/harness/src/sandbox/control.ts b/packages/harness/src/sandbox/control.ts index 09b4cf3d..afa228f3 100644 --- a/packages/harness/src/sandbox/control.ts +++ b/packages/harness/src/sandbox/control.ts @@ -458,7 +458,7 @@ export const make = Effect.fn("Sandbox.Controller.make")(function* (options: Opt ); }); - const create: Interface["create"] = Effect.fn("Sandbox.Controller.create")(function* (input) { + const create: Interface["create"] = Effect.fn("Sandbox.Controller.create", { level: "Info" })(function* (input) { const driver = yield* registry.get(input.driver.name); const config = yield* registry.decodeCreateConfig(input.driver.name, input.config); const id = input.instanceId ?? SandboxInstance.ID.create(); @@ -756,7 +756,7 @@ export const make = Effect.fn("Sandbox.Controller.make")(function* (options: Opt return yield* reload(id); }); - const wake: Interface["wake"] = Effect.fn("Sandbox.Controller.wake")(function* (id) { + const wake: Interface["wake"] = Effect.fn("Sandbox.Controller.wake", { level: "Info" })(function* (id) { if (id === SandboxInstance.ID.local) return hostInfo(); const row = yield* requireRow(id); if (row.status === "removed") { From a0a05f82584bc2fb8d83b16206851c496def0c27 Mon Sep 17 00:00:00 2001 From: Sanchit Rk <0xsanchit@gmail.com> Date: Tue, 13 Oct 2026 17:17:50 +0530 Subject: [PATCH 5/6] fix(harness): probe instruction files with exists, not stat Most instruction candidates are missing, and a stat on a missing path fails its span even when the caller treats absence as normal: every debug trace carried ~54 SandboxFileSystem.stat errors per run. Candidates are now probed with exists, which answers absence without failing, and only the files that exist are stat'ed. Not-found stays silent and other failures still warn on every driver. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + .../src/plugin/builtin/prompt/instruction.ts | 10 ++++++++-- packages/harness/test/plugin.instruction.test.ts | 14 ++++++++------ 3 files changed, 17 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 37a7867e..8842ba3c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,7 @@ This file is the canonical source for unreleased changes and published release n ### Internal +- Instruction-file discovery probes candidates with `exists` and only stats the files that exist, so a missing `AGENTS.md` or `CLAUDE.md` no longer records a failed span in debug traces. - The `codework tui` E2E artifact now records the CLI version as ``, so it no longer goes stale on every version bump. - Tests load provider and sandbox credentials from the root `.env.local` instead of one file per package. - Added a live E2E test that runs `codework run` on a memory, Daytona and Vercel sandbox and checks the sandbox is stopped on exit. diff --git a/packages/harness/src/plugin/builtin/prompt/instruction.ts b/packages/harness/src/plugin/builtin/prompt/instruction.ts index 5b0d68ad..4555cfe0 100644 --- a/packages/harness/src/plugin/builtin/prompt/instruction.ts +++ b/packages/harness/src/plugin/builtin/prompt/instruction.ts @@ -47,13 +47,19 @@ const ancestors = (directory: string): string[] => { /** * Which of `paths` are files. Absence is the expected answer and stays silent; any other failure * (permissions, a remote blip) is logged so a dropped file is not a mystery, then counts as absent. + * + * `exists` answers absence without failing, so only the few paths that exist are stat'ed. A stat + * on a missing path fails its span, and most candidates are missing: probing them all with `stat` + * filled every trace with errors that were never errors. */ const present = (fs: SandboxFileSystem.Interface, paths: ReadonlyArray) => Effect.forEach( paths, (path) => - fs.stat(path).pipe( - Effect.map((stat) => stat.isFile), + fs.exists(path).pipe( + Effect.flatMap((exists) => + exists ? fs.stat(path).pipe(Effect.map((stat) => stat.isFile)) : Effect.succeed(false), + ), Effect.catch((error) => SandboxFileSystem.isNotFoundError(error.cause) ? Effect.succeed(false) diff --git a/packages/harness/test/plugin.instruction.test.ts b/packages/harness/test/plugin.instruction.test.ts index 8ec6aaac..1b003174 100644 --- a/packages/harness/test/plugin.instruction.test.ts +++ b/packages/harness/test/plugin.instruction.test.ts @@ -60,8 +60,8 @@ const prompt = async (input: { root: string; directory: string; hostCwd: string; }; /** - * A filesystem over `files`. Paths in `slow` answer `stat` and `readFile` late, so a probe that took - * completion order for candidate order would be caught; `failing` paths fail `stat` with EIO. + * A filesystem over `files`. Paths in `slow` answer `exists` and `readFile` late, so a probe that took + * completion order for candidate order would be caught; `failing` paths fail `exists` with EIO. */ const instrumented = ( files: Readonly>, @@ -80,16 +80,18 @@ const instrumented = ( readBytes: unused, writeFile: unused, stat: async (path) => { + if (!(path in files)) throw Object.assign(new Error(`ENOENT: stat '${path}'`), { code: "ENOENT" }); + return { isFile: true, isDirectory: false }; + }, + readdir: unused, + exists: async (path) => { stats.calls += 1; stats.peak = Math.max(stats.peak, ++stats.inFlight); await settle(path); stats.inFlight -= 1; if (failing.has(path)) throw Object.assign(new Error(`EIO: i/o error, stat '${path}'`), { code: "EIO" }); - if (!(path in files)) throw Object.assign(new Error(`ENOENT: stat '${path}'`), { code: "ENOENT" }); - return { isFile: true, isDirectory: false }; + return path in files; }, - readdir: unused, - exists: unused, mkdir: unused, rm: unused, realpath: unused, From 29282902b3d7da0c77951a3eb726660ea9665153 Mon Sep 17 00:00:00 2001 From: Sanchit Rk <0xsanchit@gmail.com> Date: Tue, 13 Oct 2026 17:39:57 +0530 Subject: [PATCH 6/6] fix: address telemetry review findings - Export sandbox provisioning: CLI.selectSandbox and Sandbox.resolve are Info, and the server RPCs that provision a sandbox start their own Info trace, since RpcServer's Debug request span would drop them. - Record tokens and cost of an interrupted model call from the terminal the consumer drains. - E2E covers CODEWORK_TRACE_LEVEL=debug; the instruction test asserts stat only runs for existing files. - Drop comments that restated the code. Co-Authored-By: Claude Opus 5.5 --- packages/codework/src/cli/cmd/handlers/run.ts | 2 +- packages/codework/src/framework/otel.ts | 24 +-- packages/codework/src/framework/runtime.ts | 1 - packages/codework/src/server/handlers.ts | 13 +- .../test/__artifacts__/telemetry.json | 2 + packages/codework/test/telemetry.test.ts | 150 ++++++++++-------- packages/harness/src/effect/sandbox.ts | 2 +- .../src/plugin/builtin/prompt/instruction.ts | 4 - packages/harness/src/runner/llm.ts | 20 ++- packages/harness/src/sandbox/control.ts | 2 - packages/harness/src/telemetry/telemetry.ts | 14 +- .../harness/test/plugin.instruction.test.ts | 5 +- 12 files changed, 122 insertions(+), 117 deletions(-) diff --git a/packages/codework/src/cli/cmd/handlers/run.ts b/packages/codework/src/cli/cmd/handlers/run.ts index e8a444bf..4bedd908 100644 --- a/packages/codework/src/cli/cmd/handlers/run.ts +++ b/packages/codework/src/cli/cmd/handlers/run.ts @@ -66,7 +66,7 @@ const awaitMessage = Effect.fn("CLI.awaitMessage")(function* (ended: Queue.Queue } }); -const selectSandbox = Effect.fn("CLI.selectSandbox")(function* (input: Sandbox.Selection) { +const selectSandbox = Effect.fn("CLI.selectSandbox", { level: "Info" })(function* (input: Sandbox.Selection) { const selection = yield* Sandbox.resolve(input); if (selection.created && selection.info !== undefined) { const id = selection.info.id; diff --git a/packages/codework/src/framework/otel.ts b/packages/codework/src/framework/otel.ts index 34311e7a..1d386f2d 100644 --- a/packages/codework/src/framework/otel.ts +++ b/packages/codework/src/framework/otel.ts @@ -3,11 +3,8 @@ import { FetchHttpClient } from "effect/http"; import { OtlpMetrics, OtlpSerialization, OtlpTracer } from "effect/observability"; import { uuidv7 } from "uuidv7"; -/** - * OTLP export of the harness's spans and metrics, configured by the standard - * `OTEL_*` environment variables alone. Without an endpoint every layer is - * empty; a configuration that fails to build runs the command without export. - */ +// Configured by the standard `OTEL_*` variables alone. A configuration that fails +// to build runs the command without export rather than failing it. // The OTel spec defaults `OTEL_{TRACES,METRICS}_EXPORTER` to `otlp`; Effect reads an // unset list as none, so an endpoint alone would export nothing. @@ -36,13 +33,9 @@ const fiberMetrics = Layer.unwrap( }), ); -/** - * Export only the spans marked `level: "Info"`: the command, the agent loop, model - * and tool calls, and sandbox provisioning. Every other span, Effect's own `sql.*` - * among them, defaults to `Debug`; `CODEWORK_TRACE_LEVEL=debug` exports those too. - * Effect drops a span's children with it, so every span on the path from a command - * to a model call must be `Info`. - */ +// Only spans marked `level: "Info"` are exported; the rest, Effect's own `sql.*` +// among them, default to `Debug`. Effect drops a span's children with it, so every +// span on the path from a command to a model or tool call must be `Info`. const levels = { all: "All", trace: "Trace", debug: "Debug", info: "Info", warn: "Warn", error: "Error" } as const; const traceLevels = Layer.unwrap( @@ -62,11 +55,7 @@ interface OtlpResource { readonly attributes: Record; } -/** - * Serialize by `OTEL_EXPORTER_OTLP_PROTOCOL`, defaulting to the spec's - * `http/protobuf`. Effect has no gRPC exporter, so `grpc` exports nothing - * rather than sending HTTP to a gRPC port. - */ +// Effect has no gRPC exporter: `grpc` exports nothing rather than sending HTTP to a gRPC port. const exporters = (resource: OtlpResource) => Layer.unwrap( Effect.gen(function* () { @@ -83,7 +72,6 @@ const exporters = (resource: OtlpResource) => }), ); -/** The release channel a version was published to, from its prerelease tag. */ const channel = (version: string) => version.includes("-canary.") ? "canary" : version.includes("-dev.") ? "dev" : "latest"; diff --git a/packages/codework/src/framework/runtime.ts b/packages/codework/src/framework/runtime.ts index 4d25a3b3..1b8c84da 100644 --- a/packages/codework/src/framework/runtime.ts +++ b/packages/codework/src/framework/runtime.ts @@ -76,7 +76,6 @@ export function run(root: Spec.Any, handlers: ReadonlyArray, option return Command.run(provide(root, handlers, { version: options.version, path: [] }), options); } -/** Each handler runs inside its own OTLP export, named by its command path below the root. */ function provide( node: Spec.Any, handlers: ReadonlyArray, diff --git a/packages/codework/src/server/handlers.ts b/packages/codework/src/server/handlers.ts index d4dd4e40..7a61e826 100644 --- a/packages/codework/src/server/handlers.ts +++ b/packages/codework/src/server/handlers.ts @@ -40,6 +40,11 @@ const sessionInfo = Effect.fn("Server.sessionInfo")(function* (sessionId: Sessio return toSessionInfo(yield* found.value.info); }); +// RpcServer's request span has no level, so it is Debug and drops every Info span under it. +// Provisioning a sandbox is worth exporting, so it starts a trace of its own. +const exported = (name: string, self: Effect.Effect) => + self.pipe(Effect.withSpan(name, { level: "Info", root: true })); + const releaseOnFailure = (selection: Effect.Success> | undefined) => selection?.created && selection.info !== undefined ? Sandbox.stop(selection.info.id).pipe(Effect.catchCause(Effect.logError)) @@ -61,7 +66,8 @@ export const layer = Contract.Api.toLayer( "auth.refresh": ({ provider }) => oauth.refresh(provider), "auth.logout": ({ provider }) => oauth.logout(provider), "session.create": Effect.fnUntraced(function* ({ title, directory, hostDir, sandbox, runtime }) { - const selection = sandbox === undefined ? undefined : yield* Sandbox.resolve(sandbox); + const selection = + sandbox === undefined ? undefined : yield* exported("Server.session.create", Sandbox.resolve(sandbox)); const selected = selection?.info; const handle = yield* Session.create({ ...(title === undefined ? {} : { title }), @@ -93,7 +99,8 @@ export const layer = Contract.Api.toLayer( return toSessionInfo(yield* handle.info); }), "session.relink": Effect.fnUntraced(function* ({ sessionId, sandbox, directory }) { - const selection = sandbox === undefined ? undefined : yield* Sandbox.resolve(sandbox); + const selection = + sandbox === undefined ? undefined : yield* exported("Server.session.relink", Sandbox.resolve(sandbox)); const selected = selection?.info; const handle = yield* Session.relink({ sessionId, @@ -132,7 +139,7 @@ export const layer = Contract.Api.toLayer( return (yield* Sandbox.list()).map(toSandboxInfo); }), "sandbox.create": Effect.fnUntraced(function* ({ driver }) { - return toSandboxInfo(yield* Sandbox.create({ driver })); + return toSandboxInfo(yield* exported("Server.sandbox.create", Sandbox.create({ driver }))); }), "sandbox.register": Effect.fnUntraced(function* ({ driver, providerResourceId }) { return toSandboxInfo(yield* Sandbox.register({ driver, providerResourceId })); diff --git a/packages/codework/test/__artifacts__/telemetry.json b/packages/codework/test/__artifacts__/telemetry.json index 0d571ab1..d028be08 100644 --- a/packages/codework/test/__artifacts__/telemetry.json +++ b/packages/codework/test/__artifacts__/telemetry.json @@ -8,7 +8,9 @@ }, "exported": [ "(root) > CLI.run", + "CLI.run > CLI.selectSandbox", "CLI.run > Loop.run", + "CLI.selectSandbox > Sandbox.resolve", "Loop.run > Loop.runTurn", "Loop.runTurn > Loop.runTurnAttempt", "Loop.runTurnAttempt > LLM.run", diff --git a/packages/codework/test/telemetry.test.ts b/packages/codework/test/telemetry.test.ts index dda2513b..fe0fe044 100644 --- a/packages/codework/test/telemetry.test.ts +++ b/packages/codework/test/telemetry.test.ts @@ -1,7 +1,7 @@ /* @effect-diagnostics nodeBuiltinImport:off -- this suite spawns the CLI and serves a fake provider and collector. */ /* @effect-diagnostics globalTimers:off -- the subprocess deadline runs outside any Effect. */ import { spawn } from "node:child_process"; -import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { createServer, type IncomingMessage } from "node:http"; import type { AddressInfo } from "node:net"; import { tmpdir } from "node:os"; @@ -301,68 +301,81 @@ const exportedSpans = (exports: ReadonlyArray) => { ].sort(); }; -/** Every exported resource, traces and metrics alike, must describe the same process. */ const resources = (exports: ReadonlyArray) => exports.flatMap((entry) => [...(entry.body.resourceSpans ?? []), ...(entry.body.resourceMetrics ?? [])]); +/** Run the CLI once against `collector` as its provider and OTLP endpoint, with no other `OTEL_*` set. */ +const exportRun = async (collector: Awaited>, traceLevel?: string) => { + const home = mkdtempSync(join(tmpdir(), "codework-otel-")); + temps.push(home); + const project = join(home, "project"); + const settings = join(home, "settings"); + mkdirSync(project); + mkdirSync(settings); + writeFileSync(join(project, "note.txt"), "hello from the note\n"); + writeFileSync( + join(settings, "settings.jsonc"), + JSON.stringify({ + retry: { baseDelayMs: 10 }, + model: { provider: "otel", id: "otel-model" }, + models: [ + { + id: "otel-model", + name: "otel-model", + provider: { id: "otel", name: "Otel", source: "config", env: [] }, + baseUrl: `${collector.url}/v1`, + reasoning: false, + input: ["text"], + cost: { input: 3, output: 15, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 32768, + maxTokens: 8192, + protocol: "openai-compatible", + }, + ], + }), + ); + + const env: NodeJS.ProcessEnv = { + ...process.env, + HOME: home, + CODEWORK_MODELS_FILE: models, + OTEL_EXPORTER_OTLP_ENDPOINT: collector.url, + OTEL_EXPORTER_OTLP_HEADERS: "x-otel-test=present", + // The collector reads JSON; protobuf is the default otherwise. + OTEL_EXPORTER_OTLP_PROTOCOL: "http/json", + }; + const configured = new Set([ + "OTEL_EXPORTER_OTLP_ENDPOINT", + "OTEL_EXPORTER_OTLP_HEADERS", + "OTEL_EXPORTER_OTLP_PROTOCOL", + ]); + for (const key of Object.keys(env)) { + if (key.startsWith("OTEL_") && !configured.has(key)) delete env[key]; + } + delete env.CODEWORK_TRACE_LEVEL; + if (traceLevel !== undefined) env.CODEWORK_TRACE_LEVEL = traceLevel; + + const result = await runCli( + ["--home", home, "--database", ":memory:", "--user-config-dir", settings, "run", "What does note.txt say?"], + { cwd: project, env }, + ); + expect(result.status, result.stderr).toBe(0); + expect(result.stdout).toContain("The note says hello."); + + // No prompt, command, file content, or model output may leave in any span, attribute, or event. + const traces = JSON.stringify(collector.exports.filter((entry) => entry.path === "/v1/traces")); + for (const content of ["What does note.txt say?", "cat note.txt", "hello from the note", "The note says hello."]) { + expect(traces).not.toContain(content); + } +}; + +const artifact = fileURLToPath(new URL("./__artifacts__/telemetry.json", import.meta.url)); + describe("OTLP export", () => { it("exports GenAI spans and metrics for a run through the standard OTEL_* environment", async () => { const collector = await serve(); try { - const home = mkdtempSync(join(tmpdir(), "codework-otel-")); - temps.push(home); - const project = join(home, "project"); - const settings = join(home, "settings"); - mkdirSync(project); - mkdirSync(settings); - writeFileSync(join(project, "note.txt"), "hello from the note\n"); - writeFileSync( - join(settings, "settings.jsonc"), - JSON.stringify({ - retry: { baseDelayMs: 10 }, - model: { provider: "otel", id: "otel-model" }, - models: [ - { - id: "otel-model", - name: "otel-model", - provider: { id: "otel", name: "Otel", source: "config", env: [] }, - baseUrl: `${collector.url}/v1`, - reasoning: false, - input: ["text"], - cost: { input: 3, output: 15, cacheRead: 0, cacheWrite: 0 }, - contextWindow: 32768, - maxTokens: 8192, - protocol: "openai-compatible", - }, - ], - }), - ); - - const env: NodeJS.ProcessEnv = { - ...process.env, - HOME: home, - CODEWORK_MODELS_FILE: models, - OTEL_EXPORTER_OTLP_ENDPOINT: collector.url, - OTEL_EXPORTER_OTLP_HEADERS: "x-otel-test=present", - // The collector reads JSON; protobuf is the default otherwise. - OTEL_EXPORTER_OTLP_PROTOCOL: "http/json", - }; - const configured = new Set([ - "OTEL_EXPORTER_OTLP_ENDPOINT", - "OTEL_EXPORTER_OTLP_HEADERS", - "OTEL_EXPORTER_OTLP_PROTOCOL", - ]); - for (const key of Object.keys(env)) { - if (key.startsWith("OTEL_") && !configured.has(key)) delete env[key]; - } - delete env.CODEWORK_TRACE_LEVEL; - - const result = await runCli( - ["--home", home, "--database", ":memory:", "--user-config-dir", settings, "run", "What does note.txt say?"], - { cwd: project, env }, - ); - expect(result.status, result.stderr).toBe(0); - expect(result.stdout).toContain("The note says hello."); + await exportRun(collector); expect(collector.exports.length).toBeGreaterThan(0); expect(new Set(collector.exports.map((entry) => entry.path))).toEqual(new Set(["/v1/traces", "/v1/metrics"])); @@ -371,17 +384,6 @@ describe("OTLP export", () => { const exported = resources(collector.exports).map((entry) => JSON.stringify(entry.resource)); expect(new Set(exported).size, "every export carries one resource").toBe(1); - // No prompt, file content, or model output may leave in any span, attribute, or event. - const traces = JSON.stringify(collector.exports.filter((entry) => entry.path === "/v1/traces")); - for (const content of [ - "What does note.txt say?", - "cat note.txt", - "hello from the note", - "The note says hello.", - ]) { - expect(traces).not.toContain(content); - } - await expect( `${JSON.stringify( { @@ -398,4 +400,18 @@ describe("OTLP export", () => { await collector.close(); } }, 60_000); + + it("exports every span, Effect's own sql.* included, with CODEWORK_TRACE_LEVEL=debug", async () => { + const collector = await serve(); + try { + await exportRun(collector, "debug"); + const spans = exportedSpans(collector.exports); + expect(spans.some((pair) => pair.endsWith("> sql.execute"))).toBe(true); + // Debug adds spans; it never loses the ones exported by default. + const defaults: ReadonlyArray = JSON.parse(readFileSync(artifact, "utf8")).exported; + expect(spans).toEqual(expect.arrayContaining([...defaults])); + } finally { + await collector.close(); + } + }, 60_000); }); diff --git a/packages/harness/src/effect/sandbox.ts b/packages/harness/src/effect/sandbox.ts index 72fc77c0..e76c4ebc 100644 --- a/packages/harness/src/effect/sandbox.ts +++ b/packages/harness/src/effect/sandbox.ts @@ -74,7 +74,7 @@ export const Selection = Schema.Union([ ]); export type Selection = typeof Selection.Type; -export const resolve = Effect.fn("Sandbox.resolve")(function* (input: Selection) { +export const resolve = Effect.fn("Sandbox.resolve", { level: "Info" })(function* (input: Selection) { if ("id" in input) { const found = yield* get(input.id); if (Option.isNone(found)) return yield* new SandboxNotFoundError({ id: input.id }); diff --git a/packages/harness/src/plugin/builtin/prompt/instruction.ts b/packages/harness/src/plugin/builtin/prompt/instruction.ts index 4555cfe0..740e73cf 100644 --- a/packages/harness/src/plugin/builtin/prompt/instruction.ts +++ b/packages/harness/src/plugin/builtin/prompt/instruction.ts @@ -47,10 +47,6 @@ const ancestors = (directory: string): string[] => { /** * Which of `paths` are files. Absence is the expected answer and stays silent; any other failure * (permissions, a remote blip) is logged so a dropped file is not a mystery, then counts as absent. - * - * `exists` answers absence without failing, so only the few paths that exist are stat'ed. A stat - * on a missing path fails its span, and most candidates are missing: probing them all with `stat` - * filled every trace with errors that were never errors. */ const present = (fs: SandboxFileSystem.Interface, paths: ReadonlyArray) => Effect.forEach( diff --git a/packages/harness/src/runner/llm.ts b/packages/harness/src/runner/llm.ts index a67c7ecb..18a4cae6 100644 --- a/packages/harness/src/runner/llm.ts +++ b/packages/harness/src/runner/llm.ts @@ -231,11 +231,12 @@ const requestErrorType = (input: RequestInput, exit: Exit.Exit, + // An interrupted request still drains aikit's terminal, which carries the usage it spent. + drained: LLMEventPublisher.Terminal | undefined, ) { const now = yield* Effect.clockWith((clock) => clock.currentTimeMillis); const labels = { provider: input.provider, model: input.model }; @@ -251,9 +252,10 @@ const observe = Effect.fnUntraced(function* ( yield* Effect.annotateCurrentSpan("gen_ai.response.time_to_first_chunk", ttft); } if (errorType !== undefined) yield* Effect.annotateCurrentSpan("error.type", errorType); - if (Exit.isFailure(exit)) return; + const terminal = Exit.isSuccess(exit) ? exit.value : drained; + if (terminal === undefined) return; - const { message } = exit.value; + const { message } = terminal; const { usage } = message; // aikit's `input` excludes cached tokens; input counts them, as the GenAI conventions do. const inputTokens = usage.input + usage.cacheRead + usage.cacheWrite; @@ -271,7 +273,7 @@ const observe = Effect.fnUntraced(function* ( yield* Effect.annotateCurrentSpan({ "gen_ai.response.model": message.responseModel ?? message.model, ...(message.responseId === undefined ? {} : { "gen_ai.response.id": message.responseId }), - "gen_ai.response.finish_reasons": [exit.value.reason], + "gen_ai.response.finish_reasons": [terminal.reason], "gen_ai.usage.input_tokens": inputTokens, "gen_ai.usage.output_tokens": usage.output, "gen_ai.usage.cache_read.input_tokens": usage.cacheRead, @@ -308,6 +310,7 @@ export const make = (openStream: Open): Request => { }); const start = yield* Effect.clockWith((clock) => clock.currentTimeMillis); let firstToken: number | undefined; + let drained: LLMEventPublisher.Terminal | undefined; const markToken = (event: AikitEvent.LLMMessageEvent) => firstToken !== undefined || !isToken(event) ? Effect.void @@ -350,8 +353,9 @@ export const make = (openStream: Open): Request => { ); if (Exit.isFailure(awaited)) { - const drained = yield* Fiber.await(consumer); - if (Exit.isFailure(drained)) return yield* Effect.failCause(drained.cause); + const consumed = yield* Fiber.await(consumer); + if (Exit.isFailure(consumed)) return yield* Effect.failCause(consumed.cause); + drained = consumed.value; return yield* Effect.failCause(awaited.cause); } @@ -360,7 +364,9 @@ export const make = (openStream: Open): Request => { return completed.value; }), ); - const terminal = yield* settled.pipe(Effect.onExit((exit) => observe(input, { start, firstToken }, exit))); + const terminal = yield* settled.pipe( + Effect.onExit((exit) => observe(input, { start, firstToken }, exit, drained)), + ); if (terminal.outcome === "failed") return yield* new FailedTerminal({ terminal }); return terminal; }); diff --git a/packages/harness/src/sandbox/control.ts b/packages/harness/src/sandbox/control.ts index afa228f3..39280f31 100644 --- a/packages/harness/src/sandbox/control.ts +++ b/packages/harness/src/sandbox/control.ts @@ -139,13 +139,11 @@ const isSandboxProviderError = Schema.is(SandboxProviderError); const parseJson = (driver: SandboxDriver.Name, operation: string, value: string) => decodeUnknownJson(value).pipe(Effect.mapError((cause) => providerError({ driver, operation, cause }))); -/** Time a driver's provider-side create or wake, the cold start a remote sandbox pays. */ const provision = (driver: SandboxDriver.Name, operation: "create" | "wake", self: Effect.Effect) => Telemetry.timed(self, (seconds, exit) => Telemetry.record(Telemetry.sandboxProvisionDuration, seconds, Telemetry.withError({ driver, operation }, exit)), ); -/** Time every command a mount's shell runs, labelled by the driver that runs it. */ const timedShell = (shell: ISandboxExe, driver: SandboxDriver.Name): ISandboxExe => { const time = (self: Effect.Effect) => Telemetry.timed(self, (seconds, exit) => diff --git a/packages/harness/src/telemetry/telemetry.ts b/packages/harness/src/telemetry/telemetry.ts index 79062fe6..a80d0798 100644 --- a/packages/harness/src/telemetry/telemetry.ts +++ b/packages/harness/src/telemetry/telemetry.ts @@ -1,11 +1,5 @@ -/* - * @file Harness metrics. - * - * Effect owns the instruments, the spans, and the export; this module only names - * them. Metrics use codework's own names and short labels, so a moving draft - * convention never renames a dashboard. Effect's OTLP exporter reads a metric's - * unit from its `unit` attribute. - */ +// Metrics use codework's own names, not the GenAI conventions': those are still a +// draft and keep renaming. Effect's OTLP exporter reads a unit from the `unit` attribute. import { Cause, Effect, Exit, Metric } from "effect"; @@ -75,11 +69,9 @@ export type Attributes = Readonly>; export const record = (metric: Metric.Metric, value: Input, attributes: Attributes) => Metric.update(Metric.withAttributes(metric, attributes), value); -/** Add `delta` to a gauge rather than setting it. */ export const adjust = (metric: Metric.Metric, delta: number, attributes: Attributes) => Metric.modify(Metric.withAttributes(metric, attributes), delta); -/** The `error` label for a failed exit: the error's own tag, or `interrupted`. */ export const errorType = (cause: Cause.Cause): string => { if (Cause.hasInterruptsOnly(cause)) return "interrupted"; const error = Cause.squash(cause); @@ -88,11 +80,9 @@ export const errorType = (cause: Cause.Cause): string => { : "_OTHER"; }; -/** `attributes`, plus `error` when the exit failed. */ export const withError = (attributes: Attributes, exit: Exit.Exit): Attributes => Exit.isSuccess(exit) ? attributes : { ...attributes, error: errorType(exit.cause) }; -/** Run `self`, then hand its duration in seconds and its exit to `observe`. */ export const timed = ( self: Effect.Effect, observe: (seconds: number, exit: Exit.Exit) => Effect.Effect, diff --git a/packages/harness/test/plugin.instruction.test.ts b/packages/harness/test/plugin.instruction.test.ts index 1b003174..ee159bb2 100644 --- a/packages/harness/test/plugin.instruction.test.ts +++ b/packages/harness/test/plugin.instruction.test.ts @@ -68,7 +68,7 @@ const instrumented = ( slow: ReadonlySet, failing: ReadonlySet, ) => { - const stats = { calls: 0, inFlight: 0, peak: 0 }; + const stats = { calls: 0, inFlight: 0, peak: 0, stats: 0 }; const unused = () => Promise.reject(new Error("unused")); const settle = (path: string) => new Promise((resolve) => setTimeout(resolve, slow.has(path) ? 30 : 1)); const fs = SandboxFileSystem.fromProvider({ @@ -80,6 +80,7 @@ const instrumented = ( readBytes: unused, writeFile: unused, stat: async (path) => { + stats.stats += 1; if (!(path in files)) throw Object.assign(new Error(`ENOENT: stat '${path}'`), { code: "ENOENT" }); return { isFile: true, isDirectory: false }; }, @@ -209,6 +210,8 @@ describe("codework.prompt.instruction", () => { // Seven directories, six names each: all in flight at once instead of 42 serial round trips. expect(stats.calls).toBe(42); expect(stats.peak).toBe(42); + // Only the four files that exist are stat'ed; a miss never reaches `stat`. + expect(stats.stats).toBe(4); expect(found.map((file) => `${file.path}=${file.content}`)).toEqual([ "/AGENTS.md=root", "/a/b/AGENTS.MD=b: wins",