ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(observability): add turn links to session traces

Browse source
This commit is contained in:
starptech 2026-07-08 21:12:27 +02:00
commit a2e640aef9
20 changed files with 544 additions and 191 deletions
Download patch file Download diff file Expand all files Collapse all files

4
packages/llm/src/route/transport/websocket.ts
View file

@ -1,4 +1,4 @@
import { Cause, Context, Effect, Layer, Queue, Stream } from "effect"
import { Cause, Clock, Context, Effect, Layer, Queue, Stream } from "effect"
import { Headers } from "effect/unstable/http"
import { LLMError, TransportReason } from "../../schema"
import { LLMWebSocketTelemetry } from "../../telemetry/websocket"
@ -256,6 +256,8 @@ export const json = <Body, Message>(input: JsonInput<Body, Message>): JsonTransp
webSocket.open({ url: prepared.url, headers: prepared.headers }),
(connection) => connection.close,
)
const requestIssued = yield* LLMWebSocketTelemetry.RequestIssued
if (requestIssued) yield* requestIssued(yield* Clock.currentTimeNanos)
yield* connection.sendText(prepared.message)
const firstChunk = yield* LLMWebSocketTelemetry.ResponseChunkReceived
return connection.messages.pipe(

48
packages/llm/src/telemetry.ts
View file

@ -40,11 +40,11 @@ import {
} from "./semconv"
import { LLMError, LLMEvent, type LLMRequest, type Usage } from "./schema"
import { RequestIssued, ResponseChunkReceived } from "./telemetry/http"
import { CurrentModelSpan } from "./telemetry/context"
export { RequestIssued, ResponseChunkReceived } from "./telemetry/http"
type TelemetryState = {
ended: boolean
requestIssued?: bigint
firstChunkReceived?: bigint
usage?: Usage
@ -102,7 +102,7 @@ export function stream(request: LLMRequest, source: Stream.Stream<LLMEvent, LLME
kind: "client",
attributes,
})
const state: TelemetryState = { ended: false }
const state: TelemetryState = {}
yield* Effect.addFinalizer((exit) => observe(finalize(span, state, exit)))
return observeStream({
span,
@ -131,10 +131,7 @@ function observeStream(input: {
if (errorType) input.span.attribute(ATTR_ERROR_TYPE, errorType)
return true
}
return Stream.concat(
input.stream,
Stream.fromEffect(Effect.sync(() => (input.state.ended = true))).pipe(Stream.drain),
).pipe(
return input.stream.pipe(
Stream.onStart(
observe(
Effect.sync(() => {
@ -149,9 +146,14 @@ function observeStream(input: {
if (input.state.terminal) return
if ("usage" in event && event.usage !== undefined) input.state.usage = event.usage
if (LLMEvent.is.finish(event)) {
if (!terminate(Exit.void)) return
const type = event.reason === "error" ? "provider_error" : undefined
if (!terminate(type ? Exit.fail(new Error(type)) : Exit.void, type)) return
const attributes = finishAttributes(event)
for (const [key, value] of Object.entries(attributes)) input.span.attribute(key, value)
if (type) {
input.span.attribute(ATTR_OPENCODE_ERROR_SOURCE, "provider")
input.span.attribute(ATTR_OPENCODE_ERROR_STAGE, "response")
}
}
if (LLMEvent.is.providerError(event)) {
const type = errorType(event)
@ -174,20 +176,24 @@ function observeStream(input: {
}),
),
),
Stream.onEnd(Effect.sync(() => (input.state.ended = true))),
Stream.provideService(ParentSpan, input.span),
Stream.provideService(CurrentModelSpan, input.span),
Stream.provideService(RequestIssued, (time) =>
Effect.sync(() => {
input.state.requestIssued ??= time
recordFirstChunk(input.span, input.state)
}),
observe(
Effect.sync(() => {
input.state.requestIssued ??= time
recordFirstChunk(input.span, input.state)
}),
),
),
Stream.provideService(
ResponseChunkReceived,
Effect.gen(function* () {
input.state.firstChunkReceived ??= yield* Clock.currentTimeNanos
recordFirstChunk(input.span, input.state)
}),
observe(
Effect.gen(function* () {
input.state.firstChunkReceived ??= yield* Clock.currentTimeNanos
recordFirstChunk(input.span, input.state)
}),
),
),
Stream.provideService(References.TracerEnabled, false),
)
@ -196,13 +202,9 @@ function observeStream(input: {
function finalize(span: Span, state: TelemetryState, scopeExit: Exit.Exit<unknown, unknown>) {
return Effect.gen(function* () {
if (!state.terminal) {
const type = state.ended
? "incomplete_response"
: Exit.isFailure(scopeExit) && Cause.hasInterruptsOnly(scopeExit.cause)
? "canceled"
: Exit.isFailure(scopeExit)
? causeErrorType(scopeExit.cause)
: "incomplete_response"
const type = Exit.isFailure(scopeExit) && Cause.hasInterruptsOnly(scopeExit.cause)
? "canceled"
: "incomplete_response"
span.attribute(ATTR_ERROR_TYPE, type)
state.terminal = Exit.fail(new Error(type))
}

6
packages/llm/src/telemetry/context.ts Normal file
View file

@ -0,0 +1,6 @@
import { Context } from "effect"
import type { Span } from "effect/Tracer"
export const CurrentModelSpan = Context.Reference<Span | undefined>("@opencode/LLM/Telemetry/CurrentModelSpan", {
defaultValue: () => undefined,
})

33
packages/llm/src/telemetry/http.ts
View file

@ -16,6 +16,7 @@ import {
ATTR_URL_SCHEME,
} from "../semconv"
import { LLMError } from "../schema"
import { CurrentModelSpan } from "./context"
export const RequestIssued = Context.Reference<((time: bigint) => Effect.Effect<void>) | undefined>(
"@opencode/LLM/Telemetry/RequestIssued",
@ -42,8 +43,8 @@ const observe = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
export const stream = <A, R>(request: HttpClientRequest.HttpClientRequest, source: Stream.Stream<A, LLMError, R>) =>
Stream.unwrap(
Effect.gen(function* () {
const parent = Option.getOrUndefined(yield* Effect.serviceOption(ParentSpan))
if (parent?._tag !== "Span") return source
const parent = yield* CurrentModelSpan
if (!parent) return source
const url = URL.canParse(request.url) ? new URL(request.url) : undefined
const port = url?.port
? Number(url.port)
@ -68,7 +69,7 @@ export const stream = <A, R>(request: HttpClientRequest.HttpClientRequest, sourc
: {}),
},
})
const state: State = { ended: false, responseReceived: false }
const state: State = { responseReceived: false }
yield* Effect.addFinalizer((exit) => observe(finalize(span, state, exit)))
return observeStream(span, state, source)
}).pipe(
@ -81,7 +82,6 @@ export const stream = <A, R>(request: HttpClientRequest.HttpClientRequest, sourc
)
type State = {
ended: boolean
responseReceived: boolean
terminal?: Exit.Exit<void, unknown>
}
@ -92,7 +92,7 @@ function observeStream<A, R>(span: Span, state: State, source: Stream.Stream<A,
state.terminal = exit
if (type) span.attribute(ATTR_ERROR_TYPE, type)
}
return Stream.concat(source, Stream.fromEffect(Effect.sync(() => (state.ended = true))).pipe(Stream.drain)).pipe(
return source.pipe(
Stream.onStart(
observe(
Effect.gen(function* () {
@ -130,10 +130,12 @@ function observeStream<A, R>(span: Span, state: State, source: Stream.Stream<A,
),
Stream.provideService(ParentSpan, span),
Stream.provideService(ResponseReceived, (status) =>
Effect.sync(() => {
state.responseReceived = true
span.attribute(ATTR_HTTP_RESPONSE_STATUS_CODE, status)
}),
observe(
Effect.sync(() => {
state.responseReceived = true
span.attribute(ATTR_HTTP_RESPONSE_STATUS_CODE, status)
}),
),
),
Stream.provideService(HttpClient.TracerDisabledWhen, () => true),
Stream.provideService(HttpClient.TracerPropagationEnabled, false),
@ -144,11 +146,9 @@ function observeStream<A, R>(span: Span, state: State, source: Stream.Stream<A,
function finalize(span: Span, state: State, scopeExit: Exit.Exit<unknown, unknown>) {
return Effect.gen(function* () {
if (!state.terminal) {
state.terminal = state.ended
? Exit.void
: Exit.isFailure(scopeExit)
? Exit.failCause(scopeExit.cause)
: Exit.void
state.terminal = Exit.isFailure(scopeExit) && Cause.hasInterruptsOnly(scopeExit.cause)
? Exit.failCause(scopeExit.cause)
: Exit.void
}
span.end(yield* Clock.currentTimeNanos, state.terminal)
})
@ -158,10 +158,7 @@ export function safeUrl(url: URL) {
const safe = new URL(url)
safe.username = ""
safe.password = ""
for (const key of safe.searchParams.keys()) {
if (/(?:api[-_]?key|token|secret|signature|credential|password|authorization|auth)/i.test(key))
safe.searchParams.set(key, "[REDACTED]")
}
safe.search = ""
safe.hash = ""
return safe.toString()
}

31
packages/llm/src/telemetry/websocket.ts
View file

@ -15,9 +15,10 @@ import {
ATTR_URL_SCHEME,
} from "../semconv"
import { LLMError } from "../schema"
import { RequestIssued, safeUrl } from "./http"
import { safeUrl } from "./http"
import { CurrentModelSpan } from "./context"
export { ResponseChunkReceived } from "./http"
export { RequestIssued, ResponseChunkReceived } from "./http"
const observe = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
Effect.catchCauseIf(
@ -29,8 +30,8 @@ const observe = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
export const stream = <A, R>(urlValue: string, source: Stream.Stream<A, LLMError, R>) =>
Stream.unwrap(
Effect.gen(function* () {
const parent = Option.getOrUndefined(yield* Effect.serviceOption(ParentSpan))
if (parent?._tag !== "Span") return source
const parent = yield* CurrentModelSpan
if (!parent) return source
const url = URL.canParse(urlValue) ? new URL(urlValue) : undefined
const port = url?.port
? Number(url.port)
@ -56,7 +57,7 @@ export const stream = <A, R>(urlValue: string, source: Stream.Stream<A, LLMError
: {}),
},
})
const state: State = { ended: false }
const state: State = {}
yield* Effect.addFinalizer((exit) => observe(finalize(span, state, exit)))
return observeStream(span, state, source)
}).pipe(
@ -68,7 +69,7 @@ export const stream = <A, R>(urlValue: string, source: Stream.Stream<A, LLMError
),
)
type State = { ended: boolean; terminal?: Exit.Exit<void, unknown> }
type State = { terminal?: Exit.Exit<void, unknown> }
function observeStream<A, R>(span: Span, state: State, source: Stream.Stream<A, LLMError, R>) {
const terminate = (exit: Exit.Exit<void, unknown>, type?: string) => {
@ -76,15 +77,7 @@ function observeStream<A, R>(span: Span, state: State, source: Stream.Stream<A,
state.terminal = exit
if (type) span.attribute(ATTR_ERROR_TYPE, type)
}
return Stream.concat(source, Stream.fromEffect(Effect.sync(() => (state.ended = true))).pipe(Stream.drain)).pipe(
Stream.onStart(
observe(
Effect.gen(function* () {
const requestIssued = yield* RequestIssued
if (requestIssued) yield* requestIssued(yield* Clock.currentTimeNanos)
}),
),
),
return source.pipe(
Stream.tapCause((cause) =>
observe(
Effect.gen(function* () {
@ -115,11 +108,9 @@ function observeStream<A, R>(span: Span, state: State, source: Stream.Stream<A,
function finalize(span: Span, state: State, scopeExit: Exit.Exit<unknown, unknown>) {
return Effect.gen(function* () {
if (!state.terminal) {
state.terminal = state.ended
? Exit.void
: Exit.isFailure(scopeExit)
? Exit.failCause(scopeExit.cause)
: Exit.void
state.terminal = Exit.isFailure(scopeExit) && Cause.hasInterruptsOnly(scopeExit.cause)
? Exit.failCause(scopeExit.cause)
: Exit.void
}
span.end(yield* Clock.currentTimeNanos, state.terminal)
})

60
packages/llm/test/telemetry.test.ts
View file

@ -86,7 +86,12 @@ describe("GenAI telemetry", () => {
completion_tokens_details: { reasoning_tokens: 1 },
}
const model = OpenAIChat.route
.with({ endpoint: { baseURL: "https://api.openai.test/v1", query: { api_key: "secret-key" } } })
.with({
endpoint: {
baseURL: "https://api.openai.test/v1",
query: { api_key: "secret-key", key: "short-key", sig: "signed-value" },
},
})
.model({ id: "gpt-4o-mini" })
let traceparent: string | undefined
const request = LLM.request({
@ -168,6 +173,8 @@ describe("GenAI telemetry", () => {
expect(http?.attributes.get(ATTR_SERVER_PORT)).toBe(443)
expect(http?.attributes.get(ATTR_URL_FULL)).toStartWith("https://api.openai.test/")
expect(http?.attributes.get(ATTR_URL_FULL)).not.toContain("secret-key")
expect(http?.attributes.get(ATTR_URL_FULL)).not.toContain("short-key")
expect(http?.attributes.get(ATTR_URL_FULL)).not.toContain("signed-value")
expect(ancestorNames(http)).toContain("chat gpt-4o-mini")
expect(
span?.status._tag === "Ended" && http?.status._tag === "Ended"
@ -254,6 +261,31 @@ describe("GenAI telemetry", () => {
}),
)
it.effect("does not create transport spans beneath a disabled model span", () =>
Effect.gen(function* () {
const spans: Tracer.NativeSpan[] = []
const tracer = Tracer.make({
span(options) {
const span = new Tracer.NativeSpan(options)
spans.push(span)
return span
},
})
const model = OpenAIChat.route
.with({ endpoint: { baseURL: "https://api.openai.test/v1" } })
.model({ id: "disabled-transport-model" })
yield* Effect.useSpan("parent", () =>
LLMClient.generate(LLM.request({ model, prompt: "secret" })).pipe(
Effect.provide(fixedResponse(sseEvents(deltaChunk({}, "stop")))),
Effect.withTracerEnabled(false),
),
).pipe(Effect.provideService(Tracer.Tracer, tracer))
expect(spans.map((span) => span.name)).toEqual(["parent"])
}),
)
it.effect("measures first-chunk latency from request issuance", () =>
Effect.gen(function* () {
const spans: Tracer.NativeSpan[] = []
@ -371,6 +403,32 @@ describe("GenAI telemetry", () => {
}),
)
it.effect("marks error finish reasons as provider failures", () =>
Effect.gen(function* () {
const spans: Tracer.NativeSpan[] = []
const tracer = Tracer.make({
span(options) {
const span = new Tracer.NativeSpan(options)
spans.push(span)
return span
},
})
const model = OpenAIChat.route
.with({ endpoint: { baseURL: "https://api.openai.test/v1" } })
.model({ id: "error-finish-model" })
yield* instrument(
LLM.request({ model, prompt: "secret" }),
Stream.succeed(LLMEvent.finish({ reason: "error" })),
).pipe(Stream.runDrain, Effect.provideService(Tracer.Tracer, tracer))
const span = spans.find((span) => span.name === "chat error-finish-model")
expect(span?.attributes.get(ATTR_ERROR_TYPE)).toBe("provider_error")
expect(span?.attributes.get(ATTR_OPENCODE_ERROR_SOURCE)).toBe("provider")
expect(span?.status._tag === "Ended" && span.status.exit._tag).toBe("Failure")
}),
)
it.live("does not mutate provider request headers", () =>
Effect.acquireUseRelease(
Effect.sync(() => {