From cec637385cdc2cbb2daa579c902c504dd265b27f Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 31 Oct 2026 15:38:10 -0700 Subject: [PATCH] feat(core): trace each request and session turn on its own, with session, tool and model details (#60974) --- packages/core/src/credential.ts | 3 +- packages/core/src/session/runner/llm.ts | 9 +++- .../src/session/runner/publish-llm-event.ts | 8 +--- packages/core/src/session/runner/step.ts | 21 +++++++++- packages/core/src/tool.ts | 5 +++ .../core/test/session-runner-recorded.test.ts | 32 +++++++++++++-- packages/server/src/fetch.ts | 4 +- packages/server/src/process.ts | 3 ++ packages/server/src/request-tracing.ts | 11 +++-- packages/server/test/request-tracing.test.ts | 41 ++++++++++++++++++- packages/util/src/fs-util.ts | 2 +- 11 files changed, 118 insertions(+), 21 deletions(-) diff --git a/packages/core/src/credential.ts b/packages/core/src/credential.ts index f1ca9148ab3..5dc6b7b1eaf 100644 --- a/packages/core/src/credential.ts +++ b/packages/core/src/credential.ts @@ -112,14 +112,13 @@ const layer = Layer.effect( }), ), ), - all: Effect.fn("Credential.all")(() => + all: () => db .select() .from(CredentialTable) .orderBy(asc(CredentialTable.active), asc(CredentialTable.time_created), asc(CredentialTable.id)) .all() .pipe(Effect.orDie, Effect.map(storedRows)), - ), list: Effect.fn("Credential.list")((integrationID) => db .select() diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 8ec498e1d78..7e26333edab 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -40,6 +40,8 @@ const CONTINUE_AFTER_INCOMPLETE_STREAM = const CONTINUE_AFTER_OUTPUT_LIMIT = "Your last response hit the output token limit (stop reason: length). Do not apologize, recap, or repeat yourself. Break the remaining work into smaller pieces." +type DrainInput = Parameters[0] + const layer = Layer.effect( Service, Effect.gen(function* () { @@ -58,7 +60,8 @@ const layer = Layer.effect( // Title generation starts once input is visible and must not delay model execution. const titles = yield* FiberMap.make() - const drain = Effect.fn("SessionRunner.drain")(function* (input: Parameters[0]) { + // Each drain is one turn and starts its own trace, whatever fiber woke it. + const drain = Effect.fn("SessionRunner.drain", { root: true })(function* (input: DrainInput) { const sessionID = input.sessionID let force = input.force let continuing = input.continuation !== undefined @@ -431,7 +434,9 @@ const layer = Layer.effect( } }) - return Service.of({ drain }) + return Service.of({ + drain: (input) => drain(input).pipe(Effect.annotateSpans("session.id", input.sessionID)), + }) }), ) diff --git a/packages/core/src/session/runner/publish-llm-event.ts b/packages/core/src/session/runner/publish-llm-event.ts index a4fabb1f0db..d51fd3b6b78 100644 --- a/packages/core/src/session/runner/publish-llm-event.ts +++ b/packages/core/src/session/runner/publish-llm-event.ts @@ -590,8 +590,6 @@ export const createLLMEventPublisher = (bus: Pick, inp } }, Effect.uninterruptible) - const publishTraced = Effect.fn("SessionRunner.publishLLMEvent")(publish) - const progress = Effect.fnUntraced(function* (id: string, update: Tool.Metadata) { const tool = tools.get(id) if (!tool?.called || tool.settled) return yield* Effect.die(new Error(`Tool progress outside running call: ${id}`)) @@ -625,11 +623,7 @@ export const createLLMEventPublisher = (bus: Pick, inp }) return { - publish: (event: LLMEvent) => { - if (event.type === "text-delta" || event.type === "reasoning-delta" || event.type === "tool-input-delta") - return publish(event) - return publishTraced(event) - }, + publish: (event: LLMEvent): Effect.Effect => publish(event), progress, toolExecution, flush, diff --git a/packages/core/src/session/runner/step.ts b/packages/core/src/session/runner/step.ts index bb1c279c816..cd7d995104d 100644 --- a/packages/core/src/session/runner/step.ts +++ b/packages/core/src/session/runner/step.ts @@ -74,6 +74,12 @@ export const make = Effect.gen(function* () { // The start snapshot only has to exist before local tools run, which cannot happen before Step.Started, // so it is captured while the provider request is in flight instead of delaying it. const pendingStartSnapshot = yield* snapshots.capture().pipe(Effect.forkScoped) + const started = yield* Clock.currentTimeMillis + let firstOutputAt: number | undefined + yield* Effect.annotateCurrentSpan({ + "gen_ai.provider.name": input.model.ref.providerID, + "gen_ai.request.model": input.model.ref.id, + }) const publisher = createLLMEventPublisher(bus, { sessionID: input.sessionID, assistantMessageID: input.assistantMessageID, @@ -81,7 +87,7 @@ export const make = Effect.gen(function* () { model: input.model.ref, providerMetadataKey: input.model.model.route.providerMetadataKey ?? input.model.model.provider, pendingSnapshot: Fiber.join(pendingStartSnapshot), - started: yield* Clock.currentTimeMillis, + started, }) const toolRuns: Array<{ readonly call: ToolCall @@ -119,6 +125,8 @@ export const make = Effect.gen(function* () { return } yield* publisher.publish(event) + if (firstOutputAt === undefined && publisher.record().outputStarted) + firstOutputAt = yield* Clock.currentTimeMillis if (event.type !== "tool-call" || event.providerExecuted) return toolRuns.push({ call: event, @@ -184,6 +192,17 @@ export const make = Effect.gen(function* () { : undefined const llmFailure = streamFailure instanceof AIError ? streamFailure : unknownFinish const llmError = llmFailure && !recorded.providerFailed ? toSessionError(llmFailure) : undefined + const tokens = recorded.finish?.usage + yield* Effect.annotateCurrentSpan({ + ...(firstOutputAt === undefined ? {} : { "opencode.llm.first_output_ms": firstOutputAt - started }), + ...(recorded.finish === undefined ? {} : { "gen_ai.response.finish_reasons": [recorded.finish.finish] }), + ...(tokens?.inputTokens === undefined ? {} : { "gen_ai.usage.input_tokens": tokens.inputTokens }), + ...(tokens?.outputTokens === undefined ? {} : { "gen_ai.usage.output_tokens": tokens.outputTokens }), + ...(tokens?.cacheReadInputTokens === undefined + ? {} + : { "gen_ai.usage.cache_read.input_tokens": tokens.cacheReadInputTokens }), + ...(llmFailure === undefined ? {} : { "error.type": llmFailure.reason._tag }), + }) if ( input.recoverContinuation && llmFailure?.reason._tag === "Transport" && diff --git a/packages/core/src/tool.ts b/packages/core/src/tool.ts index 156a05bcef1..8ad9879880c 100644 --- a/packages/core/src/tool.ts +++ b/packages/core/src/tool.ts @@ -134,6 +134,11 @@ const layer = Layer.effect( input: unknown, context: Tool.Context, ) { + yield* Effect.annotateCurrentSpan({ + "gen_ai.tool.name": name, + "gen_ai.tool.call.id": context.id, + "session.id": context.sessionID, + }) const execution = yield* execute(tool, input, context).pipe( Effect.map((value) => ({ value })), Effect.catchTag("Tool.Error", (failure) => Effect.succeed({ failure })), diff --git a/packages/core/test/session-runner-recorded.test.ts b/packages/core/test/session-runner-recorded.test.ts index dd390694d2f..27f87304067 100644 --- a/packages/core/test/session-runner-recorded.test.ts +++ b/packages/core/test/session-runner-recorded.test.ts @@ -39,7 +39,7 @@ import { OptimizePlugin } from "@opencode/core/plugin/optimize" import { IdentityPlugin } from "@opencode/core/plugin/identity" import { describe, expect } from "bun:test" import { eq } from "drizzle-orm" -import { Effect, Layer } from "effect" +import { Effect, Layer, Option, Tracer } from "effect" import path from "node:path" import { testEffect } from "./lib/effect" import { LocationServiceMap } from "@opencode/core/location-service-map" @@ -167,7 +167,15 @@ const testLayer = (llmClient: Layer.Layer) => SessionExecution.node.replace(execution(llmClient)), ], ) -const it = testEffect(testLayer(client)) +const spans: Tracer.NativeSpan[] = [] +const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options) + spans.push(span) + return span + }, +}) +const it = testEffect(testLayer(client).pipe(Layer.provide(Layer.succeed(Tracer.Tracer, tracer)))) const sessionID = Session.ID.make("ses_runner_recorded") describe("SessionRunnerLLM recorded", () => { @@ -243,6 +251,24 @@ describe("SessionRunnerLLM recorded", () => { "session.step.streamed.1", "session.step.ended.1", ]) - }), + + // The turn is its own trace, every span in it names the session, and the attempt names the model and tokens. + const drain = spans.find((span) => span.name === "SessionRunner.drain") + expect(drain).toBeDefined() + expect(Option.isNone(drain!.parent)).toBe(true) + const turn = spans.filter((span) => span.traceId === drain!.traceId) + expect(turn.filter((span) => span.attributes.get("session.id") !== sessionID).map((span) => span.name)).toEqual( + [], + ) + const attempt = turn.find((span) => span.name === "SessionStep.attempt") + expect(Object.fromEntries(attempt?.attributes ?? [])).toMatchObject({ + "gen_ai.provider.name": "openai", + "gen_ai.request.model": "gpt-4o-mini", + "gen_ai.response.finish_reasons": ["stop"], + "gen_ai.usage.input_tokens": 465, + "gen_ai.usage.output_tokens": 2, + }) + expect(attempt?.attributes.get("opencode.llm.first_output_ms")).toBeNumber() + }).pipe(Effect.withSpan("fixture.request")), ) }) diff --git a/packages/server/src/fetch.ts b/packages/server/src/fetch.ts index e6aba4da568..3e4d3768c79 100644 --- a/packages/server/src/fetch.ts +++ b/packages/server/src/fetch.ts @@ -5,6 +5,7 @@ import { HttpEffect, HttpMiddleware, HttpRouter, HttpServer } from "effect/http" import { SessionRestart } from "@opencode/core/session/execution/restart" import type { LayerNode } from "@opencode/util/effect/layer-node" import { isAllowedCorsOrigin } from "./cors" +import { withoutParentSpan } from "./request-tracing" import { createRoutes } from "./routes" import type { ServerOptions } from "./options" @@ -39,9 +40,10 @@ export interface BootOptions { * next boot, and the sweep is a no-op when nothing is suspended. */ export const make = Effect.fn("ServerFetch.make")(function* (options: ServerOptions = {}, boot: BootOptions = {}) { + // Handlers capture the build context; without the caller's span, each request is its own trace. const context = yield* Layer.build( createRoutes(options, () => [], boot.overrides ?? []).pipe(Layer.provide(HttpServer.layerServices)), - ) + ).pipe(withoutParentSpan) // Forked so the returned handler is never delayed; resumed drains are already // logged and durably recorded by the execution layer. yield* Effect.forkDetach(Context.get(context, SessionRestart.Service).resumeSuspendedSessions) diff --git a/packages/server/src/process.ts b/packages/server/src/process.ts index 9058d497228..71aff3ecb6a 100644 --- a/packages/server/src/process.ts +++ b/packages/server/src/process.ts @@ -95,6 +95,8 @@ export const start = Effect.fn("ServerProcess.start")(function* ( ), ) + // Built without the startup span: route handlers and background fibers capture this context, and would + // otherwise nest every request and session turn for the life of the process under one trace. const boot = Effect.gen(function* () { const context = yield* Layer.buildWithScope( createRoutes( @@ -124,6 +126,7 @@ export const start = Effect.fn("ServerProcess.start")(function* ( updated: (version: string) => bus.publish(InstallationEvent.Updated, { version }).pipe(Effect.asVoid), } }).pipe( + withoutParentSpan, Effect.catchCause((cause) => { if (!lifecycle || Cause.hasInterruptsOnly(cause)) return Effect.failCause(cause) return status.fail.pipe( diff --git a/packages/server/src/request-tracing.ts b/packages/server/src/request-tracing.ts index b351d287695..9c2da3f739c 100644 --- a/packages/server/src/request-tracing.ts +++ b/packages/server/src/request-tracing.ts @@ -1,4 +1,9 @@ -import { Context, Effect, Scope, Tracer } from "effect" +import { Context, Effect, Tracer } from "effect" -export const withoutParentSpan = (effect: Effect.Effect) => - effect.pipe(Effect.updateContext((context: Context.Context) => Context.omit(Tracer.ParentSpan)(context))) +// ParentSpan is optional, so removing it never takes away a requirement of the effect. +export const withoutParentSpan = (effect: Effect.Effect) => + effect.pipe( + Effect.updateContext( + (context: Context.Context) => Context.omit(Tracer.ParentSpan)(context) as Context.Context, + ), + ) diff --git a/packages/server/test/request-tracing.test.ts b/packages/server/test/request-tracing.test.ts index 3854a11b5e5..622922c3ebf 100644 --- a/packages/server/test/request-tracing.test.ts +++ b/packages/server/test/request-tracing.test.ts @@ -1,6 +1,8 @@ import { expect, test } from "bun:test" import { Effect, Option, Scope, Tracer } from "effect" -import { HttpMiddleware, HttpServerRequest, HttpServerResponse } from "effect/http" +import { HttpMiddleware, HttpServer, HttpServerRequest, HttpServerResponse } from "effect/http" +import { it } from "../../core/test/lib/effect" +import { ServerProcess } from "../src/process" import { withoutParentSpan } from "../src/request-tracing" test("requests ignore ambient parents and continue inbound trace context", async () => { @@ -46,3 +48,40 @@ test("requests ignore ambient parents and continue inbound trace context", async expect(requests[2]?.traceId).toBe("11111111111111111111111111111111") expect(Option.getOrUndefined(requests[2]?.parent ?? Option.none())?.spanId).toBe("2222222222222222") }) + +it.live("handler spans nest under their request, not under server startup", () => { + const spans: Tracer.NativeSpan[] = [] + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options) + spans.push(span) + return span + }, + }) + return Effect.gen(function* () { + const server = yield* ServerProcess.start({ + hostname: "127.0.0.1", + port: 0, + password: "secret", + app: { version: "test" }, + database: { path: ":memory:" }, + fs: { filewatcher: false }, + }) + const response = yield* Effect.promise(() => + fetch(new URL("/api/session", HttpServer.formatAddress(server.address)), { + method: "POST", + headers: { authorization: `Basic ${btoa("opencode:secret")}`, "content-type": "application/json" }, + body: JSON.stringify({ title: "Traced" }), + }), + ) + expect(response.status).toBe(200) + + const startup = spans.find((span) => span.name === "ServerProcess.start") + const request = spans.find((span) => span.kind === "server" && span.attributes.get("http.route") === "/api/session") + const create = spans.find((span) => span.name === "Session.create") + expect(request).toBeDefined() + expect(Option.isNone(request!.parent)).toBe(true) + expect(create?.traceId).toBe(request!.traceId) + expect(create?.traceId).not.toBe(startup?.traceId) + }).pipe(Effect.withTracer(tracer), Effect.scoped) +}) diff --git a/packages/util/src/fs-util.ts b/packages/util/src/fs-util.ts index f814d38b0a1..a8496d98733 100644 --- a/packages/util/src/fs-util.ts +++ b/packages/util/src/fs-util.ts @@ -78,7 +78,7 @@ export namespace FSUtil { return info?.type === "File" }) - const resolve = Effect.fn("FileSystem.resolve")(function* (input: string) { + const resolve = Effect.fnUntraced(function* (input: string) { const resolved = path.resolve(windowsPath(input)) // Sandboxes deny reads outside granted paths; canonicalize what we can see. return yield* fs.realPath(resolved).pipe(