diff --git a/CHANGELOG.md b/CHANGELOG.md index 0a31a53c..8842ba3c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,11 @@ 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. 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 - `@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" } }`. @@ -20,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/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/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 0574fc16..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; @@ -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 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; + }), +); + +// 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( + Effect.gen(function* () { + const name = yield* Config.String("CODEWORK_TRACE_LEVEL").pipe(Config.withDefault("info")); + const minimum = levels[name.trim().toLowerCase() as keyof typeof levels] ?? "Info"; + return Layer.mergeAll( + Layer.succeed(References.CurrentTraceLevel, "Debug"), + Layer.succeed(References.MinimumTraceLevel, minimum), + ); + }), +); + +interface OtlpResource { + readonly serviceName: string; + readonly serviceVersion: string; + readonly attributes: Record; +} + +// 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* () { + 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)); + }), + ); + +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: OtlpResource = { + serviceName: "codework", + serviceVersion: options.version, + attributes: { + "service.instance.id": instanceId, + "deployment.environment.name": channel(options.version), + "codework.command": options.command, + }, + }; + return Layer.mergeAll(exporters(resource), fiberMetrics, traceLevels).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..1b8c84da 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,33 @@ 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 { +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/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 new file mode 100644 index 00000000..d028be08 --- /dev/null +++ b/packages/codework/test/__artifacts__/telemetry.json @@ -0,0 +1,265 @@ +{ + "resource": { + "codework.command": "run", + "deployment.environment.name": "latest", + "service.instance.id": "", + "service.name": "codework", + "service.version": "" + }, + "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", + "Loop.runTurnAttempt > Loop.settleTools", + "Loop.settleTools > Loop.handleTool" + ], + "spans": [ + { + "name": "Loop.run", + "kind": "internal", + "parent": null, + "error": false, + "attributes": { + "gen_ai.agent.name": "codework", + "gen_ai.conversation.id": "", + "gen_ai.operation.name": "invoke_agent" + } + }, + { + "name": "LLM.run", + "kind": "client", + "parent": "Loop.run", + "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_read.input_tokens": 0, + "gen_ai.usage.cache_write.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.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.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.output_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": "bash" + } + }, + { + "name": "LLM.run", + "kind": "client", + "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.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.output_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.agent.duration": { + "unit": "s", + "type": "histogram", + "series": [ + { + "attributes": { + "unit": "s" + }, + "count": 1 + } + ] + }, + "codework.llm.cost": { + "unit": "USD", + "type": "sum", + "series": [ + { + "attributes": { + "model": "otel-model", + "provider": "otel", + "unit": "USD" + }, + "value": 0.00096 + } + ] + }, + "codework.llm.duration": { + "unit": "s", + "type": "histogram", + "series": [ + { + "attributes": { + "error": "Runner.ProviderUnavailableError", + "model": "otel-model", + "provider": "otel", + "unit": "s" + }, + "count": 1 + }, + { + "attributes": { + "model": "otel-model", + "provider": "otel", + "unit": "s" + }, + "count": 2 + } + ] + }, + "codework.llm.tokens": { + "unit": "{token}", + "type": "sum", + "series": [ + { + "attributes": { + "model": "otel-model", + "provider": "otel", + "type": "input", + "unit": "{token}" + }, + "value": 240 + }, + { + "attributes": { + "model": "otel-model", + "provider": "otel", + "type": "output", + "unit": "{token}" + }, + "value": 16 + } + ] + }, + "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": { + "driver": "local", + "unit": "s" + }, + "count": 1 + } + ] + }, + "codework.tool.duration": { + "unit": "s", + "type": "histogram", + "series": [ + { + "attributes": { + "tool": "bash", + "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..fe0fe044 --- /dev/null +++ b/packages/codework/test/telemetry.test.ts @@ -0,0 +1,417 @@ +/* @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, readFileSync, 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 `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 }> }) => { + 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: "bash", arguments: '{"command":"cat 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, 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( + [...(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(["codework.llm.tokens", "codework.llm.cost"]); +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 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(); +}; + +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 { + 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"])); + 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); + + await expect( + `${JSON.stringify( + { + resource: attributes(resources(collector.exports)[0]?.resource?.attributes), + exported: exportedSpans(collector.exports), + spans: genAiSpans(collector.exports), + metrics: metrics(collector.exports), + }, + null, + "\t", + )}\n`, + ).toMatchFileSnapshot("./__artifacts__/telemetry.json"); + } finally { + 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 c965dc23..e76c4ebc 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); @@ -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 }); @@ -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/plugin/builtin/prompt/instruction.ts b/packages/harness/src/plugin/builtin/prompt/instruction.ts index 5b0d68ad..740e73cf 100644 --- a/packages/harness/src/plugin/builtin/prompt/instruction.ts +++ b/packages/harness/src/plugin/builtin/prompt/instruction.ts @@ -52,8 +52,10 @@ 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/src/runner/llm.ts b/packages/harness/src/runner/llm.ts index 2db82f17..18a4cae6 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; +}; + +const observe = Effect.fnUntraced(function* ( + input: RequestInput, + timing: { readonly start: number; readonly firstToken: number | undefined }, + 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 }; + const errorType = requestErrorType(input, exit); + yield* Telemetry.record( + Telemetry.llmDuration, + (now - timing.start) / 1000, + errorType === undefined ? labels : { ...labels, error: errorType }, + ); + if (timing.firstToken !== undefined) { + 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); + const terminal = Exit.isSuccess(exit) ? exit.value : drained; + if (terminal === undefined) return; + + 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; + 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 }), + "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, + "gen_ai.usage.cache_write.input_tokens": usage.cacheWrite, + ...(usage.reasoning === undefined ? {} : { "gen_ai.usage.reasoning.output_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", level: "Info" })(function* (input: RequestInput) { + yield* Effect.annotateCurrentSpan({ + "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; + let drained: LLMEventPublisher.Terminal | 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, ); @@ -256,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); } @@ -266,9 +364,18 @@ export const make = (openStream: Open): Request => { return completed.value; }), ); + 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; }); - 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..9ada716b 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,22 @@ const errorPart = ( time: { ...call.time, end: endTime(call, now) }, }); +const observeTool = ( + snapshot: State.Snapshot, + call: Message.ToolCallPendingPart, + seconds: number, + exit: Exit.Exit, +) => { + // 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( Runner.Service, @@ -116,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", @@ -170,7 +187,14 @@ 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, + "gen_ai.tool.call.id": call.callID, + }); yield* events.publish(EventList.ToolStarted, { timestamp: yield* DateTime.now, sessionId: snapshot.sessionId, @@ -180,7 +204,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 +229,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. @@ -237,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, ) { @@ -386,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; }) { @@ -399,20 +429,31 @@ 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", + "gen_ai.agent.name": "codework", + "gen_ai.conversation.id": input.sessionId, + }); 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")); + const drain = Effect.gen(function* () { + 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")); + } + 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) => + Telemetry.record(Telemetry.agentDuration, seconds, Telemetry.withError({}, exit)), + ); }); return Runner.Service.of({ run }); diff --git a/packages/harness/src/sandbox/control.ts b/packages/harness/src/sandbox/control.ts index 1409af21..39280f31 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, + Layer, + LayerMap, + Option, + Schema, + Semaphore, + Stream, +} 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, @@ -33,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 { @@ -126,6 +139,44 @@ const isSandboxProviderError = Schema.is(SandboxProviderError); const parseJson = (driver: SandboxDriver.Name, operation: string, value: string) => decodeUnknownJson(value).pipe(Effect.mapError((cause) => providerError({ driver, operation, cause }))); +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)), + ); + +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; const registry = yield* SandboxDriverRegistry.Registry; @@ -141,29 +192,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` @@ -394,7 +456,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(); @@ -413,9 +475,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( @@ -531,7 +593,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 = { @@ -560,6 +622,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, @@ -591,7 +654,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); } @@ -642,7 +708,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); }), ); @@ -688,7 +754,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") { @@ -706,15 +772,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, @@ -872,7 +934,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 new file mode 100644 index 00000000..a80d0798 --- /dev/null +++ b/packages/harness/src/telemetry/telemetry.ts @@ -0,0 +1,95 @@ +// 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"; + +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 llmDuration = Metric.histogram("codework.llm.duration", { + description: "Provider request duration", + attributes: { unit: "s" }, + boundaries: durationBuckets, +}); + +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 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 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("codework.agent.duration", { + description: "Session run duration, across every turn it drains", + attributes: { unit: "s" }, + 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 toolDuration = Metric.histogram("codework.tool.duration", { + description: "Tool execution duration", + attributes: { unit: "s" }, + boundaries: durationBuckets, +}); + +export const sandboxProvisionDuration = Metric.histogram("codework.sandbox.provision.duration", { + description: "Sandbox driver create and wake duration", + attributes: { unit: "s" }, + boundaries: durationBuckets, +}); + +export const sandboxExecDuration = Metric.histogram("codework.sandbox.exec.duration", { + description: "Sandbox shell command duration", + attributes: { unit: "s" }, + boundaries: durationBuckets, +}); + +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>; + +export const record = (metric: Metric.Metric, value: Input, attributes: Attributes) => + Metric.update(Metric.withAttributes(metric, attributes), value); + +export const adjust = (metric: Metric.Metric, delta: number, attributes: Attributes) => + Metric.modify(Metric.withAttributes(metric, attributes), delta); + +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"; +}; + +export const withError = (attributes: Attributes, exit: Exit.Exit): Attributes => + Exit.isSuccess(exit) ? attributes : { ...attributes, error: errorType(exit.cause) }; + +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/packages/harness/test/plugin.instruction.test.ts b/packages/harness/test/plugin.instruction.test.ts index 8ec6aaac..ee159bb2 100644 --- a/packages/harness/test/plugin.instruction.test.ts +++ b/packages/harness/test/plugin.instruction.test.ts @@ -60,15 +60,15 @@ 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>, 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,16 +80,19 @@ 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 }; + }, + 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, @@ -207,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", 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