mirror of
https://github.com/anomalyco/opencode.git
synced 2026-11-11 16:52:16 +03:00
feat(core): trace each request and session turn on its own, with session, tool and model details (#60974)
This commit is contained in:
committed by
GitHub
Unverified
parent
6f2ecd2d41
commit
cec637385c
@@ -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()
|
||||
|
||||
@@ -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<Interface["drain"]>[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<SessionSchema.ID, void, never>()
|
||||
|
||||
const drain = Effect.fn("SessionRunner.drain")(function* (input: Parameters<Interface["drain"]>[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)),
|
||||
})
|
||||
}),
|
||||
)
|
||||
|
||||
|
||||
@@ -590,8 +590,6 @@ export const createLLMEventPublisher = (bus: Pick<Bus.Interface, "publish">, 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<Bus.Interface, "publish">, 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<void> => publish(event),
|
||||
progress,
|
||||
toolExecution,
|
||||
flush,
|
||||
|
||||
@@ -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" &&
|
||||
|
||||
@@ -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 })),
|
||||
|
||||
@@ -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<LLMClientService>) =>
|
||||
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")),
|
||||
)
|
||||
})
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -95,6 +95,8 @@ export const start = Effect.fn("ServerProcess.start")(function* <E, R>(
|
||||
),
|
||||
)
|
||||
|
||||
// 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* <E, R>(
|
||||
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(
|
||||
|
||||
@@ -1,4 +1,9 @@
|
||||
import { Context, Effect, Scope, Tracer } from "effect"
|
||||
import { Context, Effect, Tracer } from "effect"
|
||||
|
||||
export const withoutParentSpan = <A, E>(effect: Effect.Effect<A, E, Scope.Scope>) =>
|
||||
effect.pipe(Effect.updateContext((context: Context.Context<Scope.Scope>) => Context.omit(Tracer.ParentSpan)(context)))
|
||||
// ParentSpan is optional, so removing it never takes away a requirement of the effect.
|
||||
export const withoutParentSpan = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
|
||||
effect.pipe(
|
||||
Effect.updateContext(
|
||||
(context: Context.Context<R>) => Context.omit(Tracer.ParentSpan)(context) as Context.Context<R>,
|
||||
),
|
||||
)
|
||||
|
||||
@@ -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<never, never>({
|
||||
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)
|
||||
})
|
||||
|
||||
@@ -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(
|
||||
|
||||
Reference in New Issue
Block a user