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

feat(observability): add v2 genai tracing

Browse source
This commit is contained in:
starptech 2026-07-08 18:02:31 +02:00
commit 0271872c09
42 changed files with 3150 additions and 373 deletions
Download patch file Download diff file Expand all files Collapse all files

19
packages/core/src/observability.ts
View file

@ -2,16 +2,23 @@ export * as Observability from "./observability"
import { NodeFileSystem } from "@effect/platform-node"
import { LayerNode } from "./effect/layer-node"
import { Effect, Layer, Logger, References } from "effect"
import { FetchHttpClient } from "effect/unstable/http"
import { Cause, Effect, Layer, Logger, References } from "effect"
import { FetchHttpClient, HttpClient } from "effect/unstable/http"
import { OtlpSerialization } from "effect/unstable/observability"
import { Logging } from "./observability/logging"
import { Otlp } from "./observability/otlp"
const references = Layer.mergeAll(
Layer.succeed(References.MinimumLogLevel, Logging.minimumLogLevel()),
Layer.succeed(References.TracerEnabled, false),
Layer.succeed(HttpClient.TracerDisabledWhen, () => true),
Layer.succeed(HttpClient.TracerPropagationEnabled, false),
)
const local = Logger.layer(Logging.loggers(), { mergeWithExisting: false }).pipe(
Layer.provide(NodeFileSystem.layer),
Layer.orDie,
Layer.merge(Layer.succeed(References.MinimumLogLevel, Logging.minimumLogLevel())),
Layer.merge(references),
)
export const layer = Layer.unwrap(
@ -21,10 +28,10 @@ export const layer = Layer.unwrap(
Layer.provide(OtlpSerialization.layerJson),
Layer.provide(FetchHttpClient.layer),
Layer.orDie,
Layer.merge(Layer.succeed(References.MinimumLogLevel, Logging.minimumLogLevel())),
Layer.merge(references),
)
return Layer.merge(logs, yield* Effect.promise(Otlp.tracingLayer))
return Layer.merge(logs, yield* Otlp.tracingLayer)
}),
).pipe(Layer.catchCause(() => local))
).pipe(Layer.catchCause((cause) => (Cause.hasInterrupts(cause) ? Layer.effectDiscard(Effect.failCause(cause)) : local)))
export const node = LayerNode.make({ name: "observability", layer, deps: [] })

267
packages/core/src/observability/agent.ts Normal file
View file

@ -0,0 +1,267 @@
export * as AgentTelemetry from "./agent"
import type { LLMEvent, Usage } from "@opencode-ai/llm"
import { Cause, Clock, Context, Effect, Exit } from "effect"
import { ParentSpan, type Span } from "effect/Tracer"
import {
ATTR_ERROR_TYPE,
ATTR_GEN_AI_AGENT_NAME,
ATTR_GEN_AI_CONVERSATION_COMPACTED,
ATTR_GEN_AI_CONVERSATION_ID,
ATTR_GEN_AI_OPERATION_NAME,
ATTR_GEN_AI_TOOL_CALL_ID,
ATTR_GEN_AI_TOOL_NAME,
ATTR_GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS,
ATTR_GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS,
ATTR_GEN_AI_USAGE_INPUT_TOKENS,
ATTR_GEN_AI_USAGE_OUTPUT_TOKENS,
ATTR_GEN_AI_USAGE_REASONING_OUTPUT_TOKENS,
ATTR_OPENCODE_AGENT_STEP_INDEX,
ATTR_OPENCODE_AGENT_STEP_TRIGGER,
ATTR_OPENCODE_COMPACTION_REASON,
ATTR_OPENCODE_ERROR_SOURCE,
ATTR_OPENCODE_ERROR_STAGE,
ATTR_OPENCODE_RETRY_ATTEMPT,
ATTR_OPENCODE_RETRY_DELAY_MS,
ATTR_OPENCODE_RETRY_DELAY_SOURCE,
ATTR_OPENCODE_RETRY_DECISION,
ATTR_OPENCODE_RETRY_MAX_ATTEMPTS,
ATTR_OPENCODE_SESSION_INPUT_COUNT,
ATTR_OPENCODE_SESSION_INPUT_DELIVERY,
ATTR_OPENCODE_SESSION_PARENT_ID,
ATTR_OPENCODE_TOOL_OUTCOME,
EVENT_OPENCODE_COMPACTION_COMPLETED,
EVENT_OPENCODE_COMPACTION_FAILED,
EVENT_OPENCODE_COMPACTION_STARTED,
EVENT_OPENCODE_PROVIDER_TOOL_CALLED,
EVENT_OPENCODE_PROVIDER_TOOL_COMPLETED,
EVENT_OPENCODE_RETRY_SCHEDULED,
EVENT_OPENCODE_RETRY_STOPPED,
EVENT_OPENCODE_SESSION_INPUT_PROMOTED,
GEN_AI_OPERATION_NAME_VALUE_INVOKE_AGENT,
} from "./semconv"
import { SessionTelemetry } from "./session"
export type ModelCallTrigger = "input" | "tool_result" | "retry" | "compaction" | "resume"
export type RetryDecision = "exhausted" | "non_retryable" | "output_started" | "step_limit"
export const Current = Context.Reference<Span | undefined>("@opencode/AgentTelemetry/Current", {
defaultValue: () => undefined,
})
const FailureState = Context.Reference<{ stage?: string } | undefined>("@opencode/AgentTelemetry/FailureState", {
defaultValue: () => undefined,
})
const observe = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
Effect.catchCauseIf(
effect,
(cause) => !Cause.hasInterrupts(cause),
() => Effect.void,
)
export const invoke = <A, E, R>(
input: { readonly sessionID: string; readonly errorType: (cause: unknown) => string },
effect: Effect.Effect<A, E, R>,
) =>
Effect.gen(function* () {
const traceParent = yield* SessionTelemetry.TraceParent
const failureState: { stage?: string } = {}
const span = yield* Effect.makeSpan("invoke_agent", {
kind: "internal",
parent: traceParent ?? undefined,
root: traceParent === null,
attributes: {
[ATTR_GEN_AI_OPERATION_NAME]: GEN_AI_OPERATION_NAME_VALUE_INVOKE_AGENT,
[ATTR_GEN_AI_CONVERSATION_ID]: input.sessionID,
},
}).pipe(
Effect.withTracerEnabled(true),
Effect.catchCause(() => Effect.succeed(undefined)),
)
if (!span) return yield* effect
return yield* effect.pipe(
Effect.onExit((exit) =>
observe(
Effect.gen(function* () {
if (Exit.isSuccess(exit)) {
span.end(yield* Clock.currentTimeNanos, exit)
return
}
const canceled = Cause.hasInterruptsOnly(exit.cause)
const type = canceled
? "canceled"
: yield* classify(() => input.errorType(Cause.squash(exit.cause)))
const source = canceled
? "cancellation"
: type.startsWith("provider.")
? "provider"
: type.startsWith("permission.")
? "permission"
: type.startsWith("tool.")
? "tool"
: "session"
span.attribute(ATTR_ERROR_TYPE, type)
span.attribute(ATTR_OPENCODE_ERROR_SOURCE, source)
if (failureState.stage) span.attribute(ATTR_OPENCODE_ERROR_STAGE, failureState.stage)
span.end(yield* Clock.currentTimeNanos, Exit.fail(new Error(type)))
}),
),
),
Effect.withParentSpan(span, { captureStackTrace: false }),
Effect.provideService(Current, span),
Effect.provideService(FailureState, failureState),
Effect.withTracerEnabled(false),
)
})
export const identify = (input: { readonly agent: string; readonly parentSessionID?: string }) =>
withCurrent((span) => {
span.attribute(ATTR_GEN_AI_AGENT_NAME, input.agent)
if (input.parentSessionID) span.attribute(ATTR_OPENCODE_SESSION_PARENT_ID, input.parentSessionID)
})
export const stage = <A, E, R>(name: string, effect: Effect.Effect<A, E, R>) =>
Effect.gen(function* () {
const state = yield* FailureState
return yield* effect.pipe(
Effect.onExit((exit) =>
Effect.sync(() => {
if (Exit.isFailure(exit) && !Cause.hasInterruptsOnly(exit.cause) && state && !state.stage) state.stage = name
}),
),
)
})
export const resetStage = Effect.gen(function* () {
const state = yield* FailureState
if (state) state.stage = undefined
})
export const inputPromoted = (input: { readonly delivery: "steer" | "queue"; readonly count: number }) =>
event(EVENT_OPENCODE_SESSION_INPUT_PROMOTED, {
[ATTR_OPENCODE_SESSION_INPUT_DELIVERY]: input.delivery,
[ATTR_OPENCODE_SESSION_INPUT_COUNT]: input.count,
})
export const compactionStarted = (reason: "manual" | "automatic") =>
event(EVENT_OPENCODE_COMPACTION_STARTED, { [ATTR_OPENCODE_COMPACTION_REASON]: reason })
export const compactionCompleted = (reason: "manual" | "automatic") =>
event(EVENT_OPENCODE_COMPACTION_COMPLETED, { [ATTR_OPENCODE_COMPACTION_REASON]: reason })
export const compactionFailed = (reason: "manual" | "automatic") =>
event(EVENT_OPENCODE_COMPACTION_FAILED, { [ATTR_OPENCODE_COMPACTION_REASON]: reason })
export const modelCall = (input: {
readonly sessionID: string
readonly agent: string
readonly step: number
readonly trigger: ModelCallTrigger
readonly retryAttempt?: number
readonly delivery?: "steer" | "queue"
readonly compacted: boolean
}) => {
let usage: Usage | undefined
const observeEvent = (value: LLMEvent) =>
Effect.gen(function* () {
if (value.type === "tool-call" && value.providerExecuted)
yield* event(EVENT_OPENCODE_PROVIDER_TOOL_CALLED, {
[ATTR_GEN_AI_TOOL_CALL_ID]: value.id,
[ATTR_GEN_AI_TOOL_NAME]: value.name,
})
if (value.type === "tool-result" && value.providerExecuted)
yield* event(EVENT_OPENCODE_PROVIDER_TOOL_COMPLETED, {
[ATTR_GEN_AI_TOOL_CALL_ID]: value.id,
[ATTR_GEN_AI_TOOL_NAME]: value.name,
[ATTR_OPENCODE_TOOL_OUTCOME]: value.result.type === "error" ? "error" : "completed",
})
if ("usage" in value && value.usage !== undefined) usage = value.usage
if (value.type === "finish" && usage) yield* recordUsage(usage)
})
const run = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
Effect.gen(function* () {
const span = yield* Current
const parentSessionID = span?.attributes.get(ATTR_OPENCODE_SESSION_PARENT_ID)
const observed = effect.pipe(
Effect.annotateSpans({
[ATTR_GEN_AI_AGENT_NAME]: input.agent,
[ATTR_GEN_AI_CONVERSATION_ID]: input.sessionID,
[ATTR_OPENCODE_AGENT_STEP_INDEX]: input.step,
[ATTR_OPENCODE_AGENT_STEP_TRIGGER]: input.trigger,
...(input.retryAttempt === undefined ? {} : { [ATTR_OPENCODE_RETRY_ATTEMPT]: input.retryAttempt }),
...(input.delivery === undefined ? {} : { [ATTR_OPENCODE_SESSION_INPUT_DELIVERY]: input.delivery }),
...(input.compacted ? { [ATTR_GEN_AI_CONVERSATION_COMPACTED]: true } : {}),
...(typeof parentSessionID === "string" ? { [ATTR_OPENCODE_SESSION_PARENT_ID]: parentSessionID } : {}),
}),
)
const traced = observed.pipe(Effect.withTracerEnabled(true))
if (!span) return yield* traced
return yield* traced.pipe(Effect.withParentSpan(span, { captureStackTrace: false }))
})
return { observe: observeEvent, run }
}
export const retryScheduled = (input: {
readonly attempt: number
readonly maxAttempts: number
readonly delayMs: number
readonly retryAfterMs?: number
readonly errorType: string
}) =>
event(EVENT_OPENCODE_RETRY_SCHEDULED, {
[ATTR_OPENCODE_RETRY_ATTEMPT]: input.attempt,
[ATTR_OPENCODE_RETRY_MAX_ATTEMPTS]: input.maxAttempts,
[ATTR_OPENCODE_RETRY_DELAY_MS]: input.delayMs,
[ATTR_OPENCODE_RETRY_DELAY_SOURCE]: input.retryAfterMs === undefined ? "backoff" : "max(backoff,retry_after)",
[ATTR_OPENCODE_RETRY_DECISION]: "scheduled",
[ATTR_ERROR_TYPE]: input.errorType,
})
export const retryStopped = (input: {
readonly decision: RetryDecision
readonly attempt: number
readonly maxAttempts: number
readonly errorType: string
}) =>
event(EVENT_OPENCODE_RETRY_STOPPED, {
[ATTR_OPENCODE_RETRY_DECISION]: input.decision,
[ATTR_OPENCODE_RETRY_ATTEMPT]: input.attempt,
[ATTR_OPENCODE_RETRY_MAX_ATTEMPTS]: input.maxAttempts,
[ATTR_ERROR_TYPE]: input.errorType,
})
function recordUsage(usage: Usage) {
return withCurrent((span) => {
const add = (key: string, value: number | undefined) => {
if (value === undefined) return
const current = span.attributes.get(key)
span.attribute(key, (typeof current === "number" ? current : 0) + value)
}
add(ATTR_GEN_AI_USAGE_INPUT_TOKENS, usage.inputTokens)
add(ATTR_GEN_AI_USAGE_OUTPUT_TOKENS, usage.outputTokens)
add(ATTR_GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS, usage.cacheReadInputTokens)
add(ATTR_GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS, usage.cacheWriteInputTokens)
add(ATTR_GEN_AI_USAGE_REASONING_OUTPUT_TOKENS, usage.reasoningTokens)
})
}
function event(name: string, attributes: Record<string, unknown>) {
return Effect.gen(function* () {
const span = yield* Current
if (!span) return
const time = yield* Clock.currentTimeNanos
yield* observe(Effect.sync(() => span.event(name, time, attributes)))
})
}
function withCurrent(f: (span: Span) => void) {
return Effect.gen(function* () {
const span = yield* Current
if (span) yield* observe(Effect.sync(() => f(span)))
})
}
function classify(f: () => string) {
return Effect.sync(f).pipe(Effect.catchCause(() => Effect.succeed("unknown")))
}

121
packages/core/src/observability/http.ts Normal file
View file

@ -0,0 +1,121 @@
export * as HttpTelemetry from "./http"
import { Cause, Clock, Effect, Exit, Option } from "effect"
import { HttpClient, HttpClientError, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
import {
ATTR_ERROR_TYPE,
ATTR_HTTP_REQUEST_METHOD,
ATTR_HTTP_RESPONSE_STATUS_CODE,
ATTR_SERVER_ADDRESS,
ATTR_SERVER_PORT,
ATTR_URL_FULL,
ATTR_URL_PATH,
ATTR_URL_SCHEME,
ATTR_OPENCODE_ERROR_SOURCE,
ATTR_OPENCODE_ERROR_STAGE,
} from "./semconv"
import { ToolTelemetry } from "./tool"
import { AgentTelemetry } from "./agent"
export const use = <A, E, R>(
http: HttpClient.HttpClient,
request: HttpClientRequest.HttpClientRequest,
consume: (response: HttpClientResponse.HttpClientResponse) => Effect.Effect<A, E, R>,
validate?: (
response: HttpClientResponse.HttpClientResponse,
) => Effect.Effect<HttpClientResponse.HttpClientResponse, HttpClientError.HttpClientError>,
): Effect.Effect<A, E | HttpClientError.HttpClientError, R> =>
Effect.gen(function* () {
const observe = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
Effect.catchCauseIf(
effect,
(cause) => !Cause.hasInterrupts(cause),
() => Effect.void,
)
const state: { responseStatus?: number } = {}
const execute = http.execute(request).pipe(
Effect.flatMap(validate ?? Effect.succeed),
Effect.tap((response) => Effect.sync(() => (state.responseStatus = response.status))),
Effect.flatMap(consume),
)
const parent =
(yield* ToolTelemetry.Current) ??
(yield* AgentTelemetry.Current) ??
Option.getOrUndefined(yield* Effect.option(Effect.currentSpan))
if (!parent) return yield* execute
const url = URL.canParse(request.url) ? new URL(request.url) : undefined
const port = url?.port
? Number(url.port)
: url?.protocol === "https:"
? 443
: url?.protocol === "http:"
? 80
: undefined
const span = yield* Effect.makeSpan(request.method, {
kind: "client",
parent,
attributes: {
[ATTR_HTTP_REQUEST_METHOD]: request.method,
...(url
? {
[ATTR_SERVER_ADDRESS]: url.hostname,
...(port === undefined ? {} : { [ATTR_SERVER_PORT]: port }),
[ATTR_URL_FULL]: safeUrl(url),
[ATTR_URL_PATH]: url.pathname,
[ATTR_URL_SCHEME]: url.protocol.slice(0, -1),
}
: {}),
},
}).pipe(
Effect.withTracerEnabled(true),
Effect.catchCause(() => Effect.succeed(undefined)),
)
if (!span) return yield* execute
return yield* execute.pipe(
Effect.provideService(HttpClient.TracerPropagationEnabled, false),
Effect.onExit((exit) =>
observe(
Effect.gen(function* () {
if (state.responseStatus !== undefined)
span.attribute(ATTR_HTTP_RESPONSE_STATUS_CODE, state.responseStatus)
if (Exit.isSuccess(exit) || Cause.hasInterruptsOnly(exit.cause)) {
span.end(yield* Clock.currentTimeNanos, exit)
return
}
const error = Option.getOrUndefined(Exit.findErrorOption(exit))
const type = HttpClientError.isHttpClientError(error)
? error.reason._tag
: error instanceof Error
? error.name
: "unknown"
const status =
HttpClientError.isHttpClientError(error) && "response" in error.reason
? error.reason.response.status
: undefined
if (status !== undefined) span.attribute(ATTR_HTTP_RESPONSE_STATUS_CODE, status)
span.attribute(ATTR_ERROR_TYPE, type)
span.attribute(ATTR_OPENCODE_ERROR_SOURCE, "transport")
span.attribute(
ATTR_OPENCODE_ERROR_STAGE,
status !== undefined ? "response" : state.responseStatus !== undefined ? "response_stream" : "request",
)
span.end(yield* Clock.currentTimeNanos, Exit.fail(new Error(type)))
}),
),
),
Effect.withParentSpan(span, { captureStackTrace: false }),
Effect.withTracerEnabled(false),
)
})
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.hash = ""
return safe.toString()
}

83
packages/core/src/observability/otlp.ts
View file

@ -1,10 +1,17 @@
import { Layer } from "effect"
import { Effect, Layer } from "effect"
import { OtlpLogger } from "effect/unstable/observability"
import {
ATTR_DEPLOYMENT_ENVIRONMENT_NAME,
ATTR_OPENCODE_CLIENT,
ATTR_OPENCODE_RUN,
ATTR_SERVICE_INSTANCE_ID,
} from "./semconv"
import { Flag } from "../flag/flag"
import { InstallationChannel, InstallationVersion } from "../installation/version"
import { runID } from "./shared"
const endpoint = Flag.OTEL_EXPORTER_OTLP_ENDPOINT
let installedContextManager: { readonly manager: { disable(): unknown }; references: number } | undefined
const headers = Flag.OTEL_EXPORTER_OTLP_HEADERS
? Flag.OTEL_EXPORTER_OTLP_HEADERS.split(",").reduce(
@ -34,15 +41,16 @@ function resourceAttributes() {
}
export function resource(): { serviceName: string; serviceVersion: string; attributes: Record<string, string> } {
const attributes = resourceAttributes()
return {
serviceName: "opencode",
serviceVersion: InstallationVersion,
attributes: {
...resourceAttributes(),
"deployment.environment.name": InstallationChannel,
"opencode.client": Flag.OPENCODE_CLIENT,
"opencode.run": runID,
"service.instance.id": runID,
...attributes,
[ATTR_DEPLOYMENT_ENVIRONMENT_NAME]: attributes[ATTR_DEPLOYMENT_ENVIRONMENT_NAME] ?? InstallationChannel,
[ATTR_OPENCODE_CLIENT]: Flag.OPENCODE_CLIENT,
[ATTR_OPENCODE_RUN]: runID,
[ATTR_SERVICE_INSTANCE_ID]: runID,
},
}
}
@ -52,28 +60,53 @@ export function loggers() {
return [OtlpLogger.make({ url: `${endpoint}/v1/logs`, resource: resource(), headers })]
}
export async function tracingLayer() {
export const tracingLayer = Effect.gen(function* () {
if (!endpoint) return Layer.empty
const NodeSdk = await import("@effect/opentelemetry/NodeSdk")
const OTLP = await import("@opentelemetry/exporter-trace-otlp-http")
const SdkBase = await import("@opentelemetry/sdk-trace-base")
const { AsyncLocalStorageContextManager } = await import("@opentelemetry/context-async-hooks")
const { context } = await import("@opentelemetry/api")
const NodeSdk = yield* Effect.promise(() => import("@effect/opentelemetry/NodeSdk"))
const OTLP = yield* Effect.promise(() => import("@opentelemetry/exporter-trace-otlp-http"))
const SdkBase = yield* Effect.promise(() => import("@opentelemetry/sdk-trace-base"))
const { AsyncLocalStorageContextManager } = yield* Effect.promise(() => import("@opentelemetry/context-async-hooks"))
const { context } = yield* Effect.promise(() => import("@opentelemetry/api"))
// The Effect Node SDK does not register a global context manager, but the AI SDK uses it to parent spans.
const manager = new AsyncLocalStorageContextManager()
manager.enable()
context.setGlobalContextManager(manager)
return NodeSdk.layer(() => ({
resource: resource(),
spanProcessor: new SdkBase.BatchSpanProcessor(
new OTLP.OTLPTraceExporter({
url: `${endpoint}/v1/traces`,
headers,
const contextManager = Layer.effectDiscard(
Effect.acquireRelease(
Effect.sync(() => {
// The Effect Node SDK does not register a global context manager, but the AI SDK uses it to parent spans.
if (installedContextManager) {
installedContextManager.references += 1
return { installed: true, manager: installedContextManager.manager }
}
const manager = new AsyncLocalStorageContextManager().enable()
const installed = context.setGlobalContextManager(manager)
if (!installed) manager.disable()
if (installed) installedContextManager = { manager, references: 1 }
return { installed, manager }
}),
({ installed, manager }) =>
Effect.sync(() => {
if (!installed) return
if (installedContextManager?.manager !== manager) return
installedContextManager.references -= 1
if (installedContextManager.references > 0) return
installedContextManager = undefined
context.disable()
manager.disable()
}),
),
}))
}
)
return Layer.merge(
contextManager,
NodeSdk.layer(() => ({
resource: resource(),
spanProcessor: new SdkBase.BatchSpanProcessor(
new OTLP.OTLPTraceExporter({
url: `${endpoint}/v1/traces`,
headers,
}),
),
})),
)
})
export * as Otlp from "./otlp"

57
packages/core/src/observability/semconv.ts Normal file
View file

@ -0,0 +1,57 @@
export * as CoreSemconv from "./semconv"
export const ATTR_DEPLOYMENT_ENVIRONMENT_NAME = "deployment.environment.name"
export const ATTR_ERROR_TYPE = "error.type"
export const ATTR_HTTP_REQUEST_METHOD = "http.request.method"
export const ATTR_HTTP_RESPONSE_STATUS_CODE = "http.response.status_code"
export const ATTR_SERVER_ADDRESS = "server.address"
export const ATTR_SERVER_PORT = "server.port"
export const ATTR_SERVICE_INSTANCE_ID = "service.instance.id"
export const ATTR_SERVICE_NAMESPACE = "service.namespace"
export const ATTR_URL_FULL = "url.full"
export const ATTR_URL_PATH = "url.path"
export const ATTR_URL_SCHEME = "url.scheme"
export const ATTR_GEN_AI_AGENT_NAME = "gen_ai.agent.name"
export const ATTR_GEN_AI_CONVERSATION_COMPACTED = "gen_ai.conversation.compacted"
export const ATTR_GEN_AI_CONVERSATION_ID = "gen_ai.conversation.id"
export const ATTR_GEN_AI_OPERATION_NAME = "gen_ai.operation.name"
export const ATTR_GEN_AI_TOOL_CALL_ID = "gen_ai.tool.call.id"
export const ATTR_GEN_AI_TOOL_NAME = "gen_ai.tool.name"
export const ATTR_GEN_AI_TOOL_TYPE = "gen_ai.tool.type"
export const ATTR_GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS = "gen_ai.usage.cache_creation.input_tokens"
export const ATTR_GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS = "gen_ai.usage.cache_read.input_tokens"
export const ATTR_GEN_AI_USAGE_INPUT_TOKENS = "gen_ai.usage.input_tokens"
export const ATTR_GEN_AI_USAGE_OUTPUT_TOKENS = "gen_ai.usage.output_tokens"
export const ATTR_GEN_AI_USAGE_REASONING_OUTPUT_TOKENS = "gen_ai.usage.reasoning.output_tokens"
export const GEN_AI_OPERATION_NAME_VALUE_EXECUTE_TOOL = "execute_tool"
export const GEN_AI_OPERATION_NAME_VALUE_INVOKE_AGENT = "invoke_agent"
export const ATTR_OPENCODE_AGENT_STEP_INDEX = "opencode.agent.step.index"
export const ATTR_OPENCODE_AGENT_STEP_TRIGGER = "opencode.agent.step.trigger"
export const ATTR_OPENCODE_CLIENT = "opencode.client"
export const ATTR_OPENCODE_COMPACTION_REASON = "opencode.compaction.reason"
export const ATTR_OPENCODE_ERROR_SOURCE = "opencode.error.source"
export const ATTR_OPENCODE_ERROR_STAGE = "opencode.error.stage"
export const ATTR_OPENCODE_RETRY_ATTEMPT = "opencode.retry.attempt"
export const ATTR_OPENCODE_RETRY_DELAY_MS = "opencode.retry.delay_ms"
export const ATTR_OPENCODE_RETRY_DELAY_SOURCE = "opencode.retry.delay_source"
export const ATTR_OPENCODE_RETRY_DECISION = "opencode.retry.decision"
export const ATTR_OPENCODE_RETRY_MAX_ATTEMPTS = "opencode.retry.max_attempts"
export const ATTR_OPENCODE_RUN = "opencode.run"
export const ATTR_OPENCODE_SESSION_INPUT_COUNT = "opencode.session.input.count"
export const ATTR_OPENCODE_SESSION_INPUT_DELIVERY = "opencode.session.input.delivery"
export const ATTR_OPENCODE_SESSION_PARENT_ID = "opencode.session.parent.id"
export const ATTR_OPENCODE_SUBAGENT_AGENT_NAME = "opencode.subagent.agent.name"
export const ATTR_OPENCODE_SUBAGENT_SESSION_ID = "opencode.subagent.session.id"
export const ATTR_OPENCODE_TOOL_OUTCOME = "opencode.tool.outcome"
export const EVENT_OPENCODE_COMPACTION_COMPLETED = "opencode.session.compaction.completed"
export const EVENT_OPENCODE_COMPACTION_FAILED = "opencode.session.compaction.failed"
export const EVENT_OPENCODE_COMPACTION_STARTED = "opencode.session.compaction.started"
export const EVENT_OPENCODE_PROVIDER_TOOL_CALLED = "opencode.provider_tool.called"
export const EVENT_OPENCODE_PROVIDER_TOOL_COMPLETED = "opencode.provider_tool.completed"
export const EVENT_OPENCODE_RETRY_SCHEDULED = "opencode.provider.retry.scheduled"
export const EVENT_OPENCODE_RETRY_STOPPED = "opencode.provider.retry.stopped"
export const EVENT_OPENCODE_SESSION_INPUT_PROMOTED = "opencode.session.input.promoted"

29
packages/core/src/observability/session.ts Normal file
View file

@ -0,0 +1,29 @@
export * as SessionTelemetry from "./session"
import { Context, Effect, Option } from "effect"
import type { AnySpan } from "effect/Tracer"
import { ParentSpan } from "effect/Tracer"
// undefined inherits the ambient parent, null explicitly detaches background work.
export const TraceParent = Context.Reference<AnySpan | null | undefined>("@opencode/SessionTelemetry/TraceParent", {
defaultValue: () => undefined,
})
export function makeExecution<Key>() {
const parents = new Map<Key, AnySpan>()
const drain = <A, E, R>(key: Key, effect: Effect.Effect<A, E, R>) => {
const parent = parents.get(key)
const observed = effect.pipe(Effect.provideService(TraceParent, parent))
return parent === undefined ? observed : observed.pipe(Effect.withParentSpan(parent, { captureStackTrace: false }))
}
const resume = <A, E, R>(key: Key, effect: Effect.Effect<A, E, R>) =>
Effect.gen(function* () {
const override = yield* TraceParent
const ambient = Option.getOrUndefined(yield* Effect.serviceOption(ParentSpan))
const parent = override === null ? undefined : (override ?? ambient)
if (parent !== undefined && !parents.has(key)) parents.set(key, parent)
return yield* effect
})
const settled = (key: Key) => Effect.sync(() => parents.delete(key)).pipe(Effect.asVoid)
return { drain, resume, settled }
}

125
packages/core/src/observability/tool.ts Normal file
View file

@ -0,0 +1,125 @@
export * as ToolTelemetry from "./tool"
import { Cause, Clock, Context, Effect, Exit } from "effect"
import type { Span } from "effect/Tracer"
import {
ATTR_ERROR_TYPE,
ATTR_GEN_AI_AGENT_NAME,
ATTR_GEN_AI_CONVERSATION_ID,
ATTR_GEN_AI_OPERATION_NAME,
ATTR_GEN_AI_TOOL_CALL_ID,
ATTR_GEN_AI_TOOL_NAME,
ATTR_GEN_AI_TOOL_TYPE,
ATTR_OPENCODE_ERROR_SOURCE,
ATTR_OPENCODE_ERROR_STAGE,
ATTR_OPENCODE_SESSION_PARENT_ID,
ATTR_OPENCODE_SUBAGENT_AGENT_NAME,
ATTR_OPENCODE_SUBAGENT_SESSION_ID,
ATTR_OPENCODE_TOOL_OUTCOME,
GEN_AI_OPERATION_NAME_VALUE_EXECUTE_TOOL,
} from "./semconv"
import { AgentTelemetry } from "./agent"
import { SessionTelemetry } from "./session"
export const Current = Context.Reference<Span | undefined>("@opencode/ToolTelemetry/Current", {
defaultValue: () => undefined,
})
const observe = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
Effect.catchCauseIf(
effect,
(cause) => !Cause.hasInterrupts(cause),
() => Effect.void,
)
export const execute = <A, E, R>(
input: {
readonly sessionID: string
readonly agent: string
readonly call: { readonly id: string; readonly name: string }
},
effect: Effect.Effect<A, E, R>,
errorType: (cause: unknown) => string,
resultErrorType?: (result: A) => string | undefined,
) =>
Effect.gen(function* () {
const toolSpan = yield* Current
const agentSpan = yield* AgentTelemetry.Current
const parent = toolSpan ?? agentSpan
const parentSessionID = agentSpan?.attributes.get(ATTR_OPENCODE_SESSION_PARENT_ID)
const span = yield* Effect.makeSpan(`execute_tool ${input.call.name}`, {
kind: "internal",
parent,
attributes: {
[ATTR_GEN_AI_OPERATION_NAME]: GEN_AI_OPERATION_NAME_VALUE_EXECUTE_TOOL,
[ATTR_GEN_AI_TOOL_NAME]: input.call.name,
[ATTR_GEN_AI_TOOL_TYPE]: "function",
[ATTR_GEN_AI_TOOL_CALL_ID]: input.call.id,
[ATTR_GEN_AI_AGENT_NAME]: input.agent,
[ATTR_GEN_AI_CONVERSATION_ID]: input.sessionID,
...(typeof parentSessionID === "string" ? { [ATTR_OPENCODE_SESSION_PARENT_ID]: parentSessionID } : {}),
},
}).pipe(
Effect.withTracerEnabled(true),
Effect.catchCause(() => Effect.succeed(undefined)),
)
if (!span) return yield* effect
return yield* effect.pipe(
Effect.tap((settlement) =>
observe(
Effect.gen(function* () {
const type = resultErrorType?.(settlement)
span.attribute(ATTR_OPENCODE_TOOL_OUTCOME, type ? "error" : "completed")
if (!type) return
span.attribute(ATTR_OPENCODE_ERROR_SOURCE, "tool")
span.attribute(ATTR_OPENCODE_ERROR_STAGE, "execution")
span.attribute(ATTR_ERROR_TYPE, type)
span.end(yield* Clock.currentTimeNanos, Exit.fail(new Error(type)))
}),
),
),
Effect.onExit((exit) => {
if (span.status._tag === "Ended") return Effect.void
if (Exit.isSuccess(exit))
return observe(
Effect.gen(function* () {
span.end(yield* Clock.currentTimeNanos, exit)
}),
)
const canceled = Cause.hasInterruptsOnly(exit.cause)
return observe(
Effect.gen(function* () {
const type = canceled
? "canceled"
: yield* Effect.sync(() => errorType(Cause.squash(exit.cause))).pipe(
Effect.catchCause(() => Effect.succeed("unknown")),
)
span.attribute(ATTR_OPENCODE_TOOL_OUTCOME, canceled ? "canceled" : "error")
span.attribute(ATTR_OPENCODE_ERROR_SOURCE, canceled ? "cancellation" : "tool")
span.attribute(ATTR_OPENCODE_ERROR_STAGE, "execution")
span.attribute(ATTR_ERROR_TYPE, type)
span.end(yield* Clock.currentTimeNanos, Exit.fail(new Error(type)))
}),
)
}),
Effect.withParentSpan(span, { captureStackTrace: false }),
Effect.provideService(Current, span),
Effect.withTracerEnabled(false),
)
})
export const child = (input: { readonly agent: string; readonly sessionID: string }) =>
Effect.gen(function* () {
const span = yield* Current
if (span)
yield* observe(
Effect.sync(() => {
span.attribute(ATTR_OPENCODE_SUBAGENT_AGENT_NAME, input.agent)
span.attribute(ATTR_OPENCODE_SUBAGENT_SESSION_ID, input.sessionID)
}),
)
return {
resume: <A, E, R>(effect: Effect.Effect<A, E, R>, background: boolean) =>
effect.pipe(Effect.provideService(SessionTelemetry.TraceParent, background ? null : span)),
}
})

23
packages/core/src/session.ts
View file

@ -1,7 +1,7 @@
export * as SessionV2 from "./session"
export * from "./session/schema"
import { DateTime, Effect, Layer, Schema, Context, Stream, Scope } from "effect"
import { Cause, DateTime, Effect, Layer, Schema, Context, Option, Stream, Scope } from "effect"
import { ListAnchor } from "@opencode-ai/schema/session"
import { and, asc, desc, eq, gt, isNull, like, lt, or, type SQL } from "drizzle-orm"
import { ProjectV2 } from "./project"
@ -42,7 +42,7 @@ import { SkillV2 } from "./skill"
import { Job } from "./job"
import { CommandV2 } from "./command"
import { Shell } from "./shell"
import { Shell as ShellSchema } from "@opencode-ai/schema/shell"
import type { Shell as ShellSchema } from "@opencode-ai/schema/shell"
import { KeyedMutex } from "./effect/keyed-mutex"
import { fileURLToPath } from "url"
@ -506,6 +506,17 @@ const layer = Layer.effect(
}
return admitted
}),
).pipe(
Effect.tapCause((cause) =>
Effect.logError("Failed to admit Session prompt", cause).pipe(
Effect.annotateLogs({
operation: "session.prompt",
sessionID: input.sessionID,
...(input.id === undefined ? {} : { messageID: input.id }),
errorType: promptErrorType(cause),
}),
),
),
),
),
command: Effect.fn("V2Session.command")(function* (input) {
@ -720,6 +731,14 @@ const layer = Layer.effect(
}),
)
function promptErrorType(cause: Cause.Cause<unknown>) {
if (Cause.hasInterruptsOnly(cause)) return "canceled"
const error = Option.getOrUndefined(Cause.findErrorOption(cause))
if (error && typeof error === "object" && "_tag" in error && typeof error._tag === "string") return error._tag
const failure = Cause.squash(cause)
return failure instanceof Error ? failure.name : "unknown"
}
function missingShellOutput() {
const output = "Shell command output is no longer available."
return {

42
packages/core/src/session/execution/local.ts
View file

@ -10,6 +10,7 @@ import { SessionStore } from "../store"
import { SessionExecution } from "../execution"
import { toSessionError } from "../to-session-error"
import { UserInterruptedError } from "../error"
import { SessionTelemetry } from "../../observability/session"
export function terminal(exit: Exit.Exit<void, SessionRunner.RunError>, reason?: "user" | "shutdown" | "superseded") {
if (Exit.isSuccess(exit)) return { type: "succeeded" as const }
@ -19,6 +20,11 @@ export function terminal(exit: Exit.Exit<void, SessionRunner.RunError>, reason?:
return { type: "failed" as const, error: toSessionError(failure) }
}
function errorType(cause: Cause.Cause<unknown>) {
const error = Cause.squash(cause)
return error instanceof Error ? error.name : "unknown"
}
/** Current-process routing for implicit-local Locations. Future remote placement belongs here. */
const layer = Layer.effect(
SessionExecution.Service,
@ -26,13 +32,19 @@ const layer = Layer.effect(
const store = yield* SessionStore.Service
const locations = yield* LocationServiceMap.Service
const events = yield* EventV2.Service
const reportLifecycle = <A>(sessionID: SessionSchema.ID, effect: Effect.Effect<A>) =>
const telemetry = SessionTelemetry.makeExecution<SessionSchema.ID>()
const reportLifecycle = <A>(sessionID: SessionSchema.ID, phase: string, effect: Effect.Effect<A>) =>
effect.pipe(
Effect.tapCause((cause) =>
Cause.hasInterruptsOnly(cause)
? Effect.void
: Effect.logError("Failed to publish Session execution lifecycle", cause).pipe(
Effect.annotateLogs({ sessionID }),
Effect.annotateLogs({
operation: "session.execution.lifecycle",
phase,
sessionID,
errorType: errorType(cause),
}),
),
),
Effect.asVoid,
@ -42,25 +54,34 @@ const layer = Layer.effect(
SessionRunner.RunError,
"user" | "shutdown" | "superseded"
>({
started: (sessionID) => reportLifecycle(sessionID, events.publish(SessionEvent.Execution.Started, { sessionID })),
started: (sessionID) =>
reportLifecycle(sessionID, "started", events.publish(SessionEvent.Execution.Started, { sessionID })),
drain: Effect.fnUntraced(function* (sessionID: SessionSchema.ID, force) {
const session = yield* store.get(sessionID)
if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`))
return yield* SessionRunner.Service.use((runner) => runner.drain({ sessionID, force })).pipe(
const drain = SessionRunner.Service.use((runner) => runner.drain({ sessionID, force })).pipe(
Effect.provide(locations.get(session.location)),
Effect.tapCause((cause) =>
Cause.hasInterruptsOnly(cause)
? Effect.void
: Effect.logError("Failed to drain Session", cause).pipe(Effect.annotateLogs({ sessionID })),
: Effect.logError("Failed to drain Session", cause).pipe(
Effect.annotateLogs({
operation: "session.execution.drain",
sessionID,
errorType: toSessionError(Cause.squash(cause)).type,
}),
),
),
)
return yield* telemetry.drain(sessionID, drain)
}),
// One terminal observation per busy period, covering every coalesced drain.
settled: (sessionID, exit, reason) =>
reportLifecycle(
settled: (sessionID, exit, reason) => {
const outcome = terminal(exit, reason)
return reportLifecycle(
sessionID,
outcome.type,
Effect.gen(function* () {
const outcome = terminal(exit, reason)
if (outcome.type === "succeeded") {
yield* events.publish(SessionEvent.Execution.Succeeded, { sessionID })
return
@ -74,13 +95,14 @@ const layer = Layer.effect(
error: outcome.error,
})
}),
),
).pipe(Effect.ensuring(telemetry.settled(sessionID)))
},
})
return SessionExecution.Service.of({
active: coordinator.active,
interrupt: (sessionID) => coordinator.interrupt(sessionID, "user"),
resume: coordinator.run,
resume: (sessionID) => telemetry.resume(sessionID, coordinator.run(sessionID)),
wake: coordinator.wake,
awaitIdle: coordinator.awaitIdle,
})

360
packages/core/src/session/runner/llm.ts
View file

@ -49,6 +49,7 @@ import { llmClient } from "../../effect/app-node-platform"
import { StepFailedError, UserInterruptedError } from "../error"
import { toSessionError } from "../to-session-error"
import { SessionRunnerRetry } from "./retry"
import { AgentTelemetry } from "../../observability/agent"
/**
* Runs one durable coding-agent Session until it settles.
@ -174,25 +175,28 @@ const layer = Layer.effect(
sessionID: SessionSchema.ID,
promotion: SessionInput.Delivery | undefined,
step: number,
trigger: AgentTelemetry.ModelCallTrigger,
retryAttempt: number,
recoverOverflow?: typeof compaction.compactAfterOverflow,
assistantMessageID?: SessionMessage.ID,
) {
const session = yield* getSession(sessionID)
const session = yield* AgentTelemetry.stage("session", getSession(sessionID))
if (session.location.directory !== location.directory || session.location.workspaceID !== location.workspaceID)
return yield* Effect.interrupt
const agent = yield* agents.select(session.agent)
const agent = yield* AgentTelemetry.stage("agent", agents.select(session.agent))
yield* AgentTelemetry.identify({ agent: agent.id, parentSessionID: session.parentID })
// Establish what the model knows before admitting what the user said, so
// a blocked first step leaves pending inputs untouched.
const checkpoint = yield* InstructionCheckpoint.prepare(
db,
events,
loadInstructions(agent, session.id),
session.id,
const checkpoint = yield* AgentTelemetry.stage(
"instructions",
InstructionCheckpoint.prepare(db, events, loadInstructions(agent, session.id), session.id),
)
const toolFibers = yield* FiberSet.make<void, ToolOutputStore.Error | UserInterruptedError>()
const ownedToolFibers: Array<Fiber.Fiber<void, ToolOutputStore.Error | UserInterruptedError>> = []
let needsContinuation = false
let currentStep = step
let currentTrigger = trigger
let inputDelivery: SessionInput.Delivery | undefined
if (promotion) {
let promoted = 0
if (promotion === "steer") promoted = yield* SessionInput.promoteSteers(db, events, session.id)
@ -200,16 +204,27 @@ const layer = Layer.effect(
promoted += Number(yield* SessionInput.promoteNextQueued(db, events, session.id))
promoted += yield* SessionInput.promoteSteers(db, events, session.id)
}
if (promoted > 0) currentStep = 1
if (promoted > 0) {
currentStep = 1
currentTrigger = "input"
inputDelivery = promotion
yield* AgentTelemetry.inputPromoted({ delivery: promotion, count: promoted })
}
}
const resolved = yield* models.resolve(session)
const resolved = yield* AgentTelemetry.stage("model_resolution", models.resolve(session))
const model = resolved.model
const entries = yield* SessionHistory.entriesForRunner(db, session.id, checkpoint.baselineSeq)
const entries = yield* AgentTelemetry.stage(
"history",
SessionHistory.entriesForRunner(db, session.id, checkpoint.baselineSeq),
)
const context = entries.map((entry) => entry.message)
const isLastStep = agent.info?.steps !== undefined && currentStep >= agent.info.steps
const toolMaterialization = isLastStep
? undefined
: yield* tools.materialize({ permissions: agent.info?.permissions, model })
: yield* AgentTelemetry.stage(
"tool_materialization",
tools.materialize({ permissions: agent.info?.permissions, model }),
)
const promptCacheKey = /^ses_[0-9a-f]{64}$/.test(session.id) ? session.id.slice(4) : session.id
const request = LLM.request({
model,
@ -228,12 +243,17 @@ const layer = Layer.effect(
toolChoice: isLastStep ? "none" : undefined,
})
// Automatic compaction completed; rebuild the request from compacted history.
if (
!(yield* SessionInput.pendingCompaction(db, session.id)) &&
(yield* compaction.compactIfNeeded({ sessionID: session.id, messages: context, request }))
)
return { _tag: "RestartAfterCompaction", step: currentStep } as const
const startSnapshot = yield* snapshots.capture()
if (!(yield* SessionInput.pendingCompaction(db, session.id))) {
const compacted = yield* AgentTelemetry.stage(
"compaction",
compaction.compactIfNeeded({ sessionID: session.id, messages: context, request }),
)
if (compacted) {
yield* AgentTelemetry.compactionCompleted("automatic")
return { _tag: "RestartAfterCompaction", step: currentStep } as const
}
}
const startSnapshot = yield* AgentTelemetry.stage("snapshot", snapshots.capture())
const publisher = createLLMEventPublisher(events, {
sessionID: session.id,
agent: agent.id,
@ -251,65 +271,83 @@ const layer = Layer.effect(
const publish = (event: LLMEvent, outputPaths: ReadonlyArray<string> = [], error?: SessionError.Error) =>
serialized(publisher.publish(event, outputPaths, error))
let overflowFailure: ProviderErrorEvent | undefined
const providerStream = llm.stream(request).pipe(
Stream.runForEach((event) =>
Effect.gen(function* () {
if (overflowFailure || publisher.hasProviderError()) return
if (LLMEvent.is.providerError(event)) {
if (isContextOverflowFailure(event) && !publisher.hasRetryEvidence()) {
overflowFailure = event
return
}
}
yield* publish(event)
if (event.type !== "tool-call" || event.providerExecuted) return
if (!toolMaterialization) {
yield* serialized(
publisher.failUnsettledTools({
type: "tool.execution",
message: "Tools are disabled after the maximum agent steps",
}),
)
return
}
needsContinuation = true
const assistantMessageID = yield* publisher.assistantMessageID(event.id)
ownedToolFibers.push(
yield* Effect.uninterruptibleMask((restore) =>
restore(
toolMaterialization.settle({
sessionID: session.id,
agent: agent.id,
assistantMessageID,
call: event,
}),
).pipe(
Effect.flatMap((settlement) =>
publish(
LLMEvent.toolResult({
id: event.id,
name: event.name,
result: settlement.result,
output: settlement.output,
const telemetry = AgentTelemetry.modelCall({
sessionID: session.id,
agent: agent.id,
step: currentStep,
trigger: currentTrigger,
retryAttempt: retryAttempt > 0 ? retryAttempt + 1 : undefined,
delivery: inputDelivery,
compacted: context.some((message) => message.type === "compaction"),
})
const providerStream = AgentTelemetry.stage(
"model",
telemetry.run(
llm.stream(request).pipe(
Stream.runForEach((event) =>
Effect.gen(function* () {
if (overflowFailure || publisher.hasProviderError()) return
if (LLMEvent.is.providerError(event)) {
if (isContextOverflowFailure(event) && !publisher.hasRetryEvidence()) {
overflowFailure = event
return
}
}
yield* publish(event)
yield* telemetry.observe(event)
if (event.type !== "tool-call" || event.providerExecuted) return
if (!toolMaterialization) {
yield* serialized(
publisher.failUnsettledTools({
type: "tool.execution",
message: "Tools are disabled after the maximum agent steps",
}),
)
return
}
needsContinuation = true
const assistantMessageID = yield* publisher.assistantMessageID(event.id)
ownedToolFibers.push(
yield* Effect.uninterruptibleMask((restore) =>
restore(
toolMaterialization.settle({
sessionID: session.id,
agent: agent.id,
assistantMessageID,
call: event,
}),
settlement.outputPaths ?? [],
settlement.error,
).pipe(
Effect.andThen(
settlement.error?.type === "permission.rejected"
? serialized(publisher.failAssistant(settlement.error)).pipe(
Effect.andThen(Effect.fail(new UserInterruptedError())),
)
: Effect.void,
Effect.flatMap((settlement) =>
AgentTelemetry.stage(
"tool_publication",
publish(
LLMEvent.toolResult({
id: event.id,
name: event.name,
result: settlement.result,
output: settlement.output,
}),
settlement.outputPaths ?? [],
settlement.error,
),
).pipe(
Effect.andThen(
settlement.error?.type === "permission.rejected"
? serialized(publisher.failAssistant(settlement.error)).pipe(
Effect.andThen(Effect.fail(new UserInterruptedError())),
)
: Effect.void,
),
),
),
),
),
),
).pipe(FiberSet.run(toolFibers)),
)
}),
).pipe(FiberSet.run(toolFibers)),
)
}),
),
Effect.ensuring(serialized(publisher.flush())),
),
),
Effect.ensuring(serialized(publisher.flush())),
)
// Captures the end snapshot, diffs it against the step's start, and durably ends the
@ -323,16 +361,19 @@ const layer = Layer.effect(
.files({ from: startSnapshot, to: endSnapshot })
.pipe(Effect.catch(() => Effect.succeed(undefined)))
: undefined
yield* serialized(
events.publish(SessionEvent.Step.Ended, {
sessionID: session.id,
assistantMessageID: yield* publisher.startAssistant(),
finish: settlement.finish,
cost: 0,
tokens: settlement.tokens,
snapshot: endSnapshot,
files,
}),
yield* AgentTelemetry.stage(
"step_publication",
serialized(
events.publish(SessionEvent.Step.Ended, {
sessionID: session.id,
assistantMessageID: yield* publisher.startAssistant(),
finish: settlement.finish,
cost: 0,
tokens: settlement.tokens,
snapshot: endSnapshot,
files,
}),
),
)
})
@ -362,11 +403,10 @@ const layer = Layer.effect(
const llmFailure = streamFailure instanceof LLMError ? streamFailure : undefined
if (llmFailure && !publisher.hasProviderError()) {
const error = toSessionError(llmFailure)
if (
SessionRunnerRetry.isRetryable(llmFailure) &&
!publisher.hasRetryEvidence() &&
(agent.info?.steps === undefined || currentStep < agent.info.steps)
) {
const retryable = SessionRunnerRetry.isRetryable(llmFailure)
const hasOutput = publisher.hasRetryEvidence()
const stepLimited = agent.info?.steps !== undefined && currentStep >= agent.info.steps
if (retryable && !hasOutput && !stepLimited) {
return yield* new SessionRunnerRetry.RetryableFailure({
cause: llmFailure,
assistantMessageID: yield* publisher.startAssistant(),
@ -374,6 +414,12 @@ const layer = Layer.effect(
step: currentStep,
})
}
yield* AgentTelemetry.retryStopped({
decision: retryable ? (hasOutput ? "output_started" : "step_limit") : "non_retryable",
attempt: retryAttempt + 1,
maxAttempts: SessionRunnerRetry.maxAttempts,
errorType: error.type,
})
yield* serialized(publisher.failAssistant(error))
}
// Provider error events only arrive from the stream, so the flag is final here.
@ -441,16 +487,17 @@ const layer = Layer.effect(
: false
if (hostedResultMissing && !publisher.stepSettlement())
yield* serialized(
publisher.failAssistant({
type: "tool.result-missing",
message: "Provider did not return a tool result",
}),
publisher.failAssistant({ type: "tool.result-missing", message: "Provider did not return a tool result" }),
)
const stepFailure = publisher.stepFailure()
const stepSettlement = publisher.stepSettlement()
const stepEndedCleanly =
!streamInterrupted && !toolsInterrupted && infraError === undefined && !providerFailed && !stepFailure
!streamInterrupted &&
!toolsInterrupted &&
infraError === undefined &&
!providerFailed &&
!stepFailure
if (stepSettlement && stepEndedCleanly) yield* publishStepEnd(stepSettlement)
if (stepFailure) yield* serialized(publisher.publishStepFailure())
@ -474,6 +521,7 @@ const layer = Layer.effect(
sessionID: SessionSchema.ID,
promotion: SessionInput.Delivery | undefined,
step: number,
trigger: AgentTelemetry.ModelCallTrigger,
) {
// Compaction restarts rebuild the request from compacted history without re-promoting.
// Overflow recovery is one-shot: a post-compaction attempt must not recover another
@ -481,10 +529,24 @@ const layer = Layer.effect(
let recoverOverflow: typeof compaction.compactAfterOverflow | undefined = compaction.compactAfterOverflow
let currentPromotion = promotion
let currentStep = step
let currentTrigger = trigger
let retryAttempt = 0
let assistantMessageID: SessionMessage.ID | undefined
while (true) {
const attempt = yield* Effect.suspend(() =>
attemptStep(sessionID, currentPromotion, currentStep, recoverOverflow, assistantMessageID),
AgentTelemetry.resetStage.pipe(
Effect.andThen(
attemptStep(
sessionID,
currentPromotion,
currentStep,
currentTrigger,
retryAttempt,
recoverOverflow,
assistantMessageID,
),
),
),
).pipe(
Effect.tapError((error) =>
error instanceof SessionRunnerRetry.RetryableFailure
@ -492,22 +554,33 @@ const layer = Layer.effect(
currentStep = error.step + 1
assistantMessageID = error.assistantMessageID
currentPromotion = undefined
currentTrigger = "retry"
retryAttempt += 1
})
: Effect.void,
),
Effect.retryOrElse(SessionRunnerRetry.schedule(events, sessionID), (error) => {
if (!(error instanceof SessionRunnerRetry.RetryableFailure)) return Effect.fail(error)
return events
.publish(SessionEvent.Step.Failed, {
sessionID,
assistantMessageID: error.assistantMessageID,
error: error.error,
})
.pipe(Effect.andThen(Effect.fail(error.cause)))
return AgentTelemetry.retryStopped({
decision: "exhausted",
attempt: retryAttempt,
maxAttempts: SessionRunnerRetry.maxAttempts,
errorType: error.error.type,
}).pipe(
Effect.andThen(
events.publish(SessionEvent.Step.Failed, {
sessionID,
assistantMessageID: error.assistantMessageID,
error: error.error,
}),
),
Effect.andThen(Effect.fail(error.cause)),
)
}),
)
if (attempt._tag === "Completed") return { needsContinuation: attempt.needsContinuation, step: attempt.step }
if (attempt._tag === "RestartAfterOverflowCompaction") recoverOverflow = undefined
currentTrigger = "compaction"
yield* Effect.yieldNow
currentPromotion = undefined
currentStep = attempt.step
@ -519,6 +592,7 @@ const layer = Layer.effect(
) {
const pending = yield* SessionInput.pendingCompaction(db, sessionID)
if (!pending) return false
yield* AgentTelemetry.compactionStarted("manual")
const session = yield* getSession(sessionID)
return yield* Effect.uninterruptibleMask((restore) =>
Effect.gen(function* () {
@ -530,8 +604,12 @@ const layer = Layer.effect(
})
}),
).pipe(Effect.exit)
if (Exit.isSuccess(compacted) && compacted.value) return true
if (Exit.isSuccess(compacted) && compacted.value) {
yield* AgentTelemetry.compactionCompleted("manual")
return true
}
yield* events.publish(SessionEvent.Compaction.Failed, { sessionID })
yield* AgentTelemetry.compactionFailed("manual")
if (Exit.isFailure(compacted)) return yield* Effect.failCause(compacted.cause)
return true
}),
@ -543,44 +621,56 @@ const layer = Layer.effect(
readonly sessionID: SessionSchema.ID
readonly force: boolean
}) {
yield* runPendingCompaction(input.sessionID)
const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, "steer")
const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, input.sessionID, "queue")
if (!input.force && !hasSteer && !hasQueue) return
yield* failInterruptedTools(input.sessionID)
let promotion: SessionInput.Delivery | undefined = hasSteer ? "steer" : hasQueue ? "queue" : undefined
let shouldRun = input.force || hasSteer || hasQueue
while (shouldRun) {
let needsContinuation = true
let step = 1
// Repeat steps while continuation is needed. A step needs continuation only
// when it recorded local tool calls whose results the model has not yet seen;
// a provider error suppresses it. Pending steers also continue the loop so
// interjections are answered before the session goes idle.
while (needsContinuation) {
const result = yield* runStep(input.sessionID, promotion, step)
// Steer/queue promotion inside runStep has already made the pending input a visible
// user message by this point, so the first-user-message check below is reliable.
if (!titleAttempted.has(input.sessionID)) {
titleAttempted.add(input.sessionID)
forkTitle(title.generateForFirstPrompt(yield* getSession(input.sessionID)).pipe(Effect.ignore))
}
needsContinuation = result.needsContinuation
step = result.step + 1
if (needsContinuation) {
promotion = (yield* SessionInput.pendingCompaction(db, input.sessionID)) ? undefined : "steer"
continue
const run = Effect.gen(function* () {
yield* AgentTelemetry.stage("compaction", runPendingCompaction(input.sessionID))
const hasSteer = yield* AgentTelemetry.stage("input", SessionInput.hasPending(db, input.sessionID, "steer"))
const hasQueue = hasSteer
? false
: yield* AgentTelemetry.stage("input", SessionInput.hasPending(db, input.sessionID, "queue"))
if (!input.force && !hasSteer && !hasQueue) return
yield* AgentTelemetry.stage("tool_recovery", failInterruptedTools(input.sessionID))
let promotion: SessionInput.Delivery | undefined = hasSteer ? "steer" : hasQueue ? "queue" : undefined
let trigger: AgentTelemetry.ModelCallTrigger = hasSteer || hasQueue ? "input" : "resume"
let shouldRun = input.force || hasSteer || hasQueue
while (shouldRun) {
let needsContinuation = true
let step = 1
// Repeat steps while continuation is needed. A step needs continuation only
// when it recorded local tool calls whose results the model has not yet seen;
// a provider error suppresses it. Pending steers also continue the loop so
// interjections are answered before the session goes idle.
while (needsContinuation) {
const result = yield* runStep(input.sessionID, promotion, step, trigger)
// Steer/queue promotion inside runStep has already made the pending input a visible
// user message by this point, so the first-user-message check below is reliable.
if (!titleAttempted.has(input.sessionID)) {
titleAttempted.add(input.sessionID)
forkTitle(title.generateForFirstPrompt(yield* getSession(input.sessionID)).pipe(Effect.ignore))
}
needsContinuation = result.needsContinuation
step = result.step + 1
if (needsContinuation) {
promotion = (yield* SessionInput.pendingCompaction(db, input.sessionID)) ? undefined : "steer"
trigger = "tool_result"
continue
}
yield* runPendingCompaction(input.sessionID)
promotion = "steer"
trigger = "tool_result"
needsContinuation = yield* SessionInput.hasPending(db, input.sessionID, "steer")
}
yield* runPendingCompaction(input.sessionID)
promotion = "steer"
needsContinuation = yield* SessionInput.hasPending(db, input.sessionID, "steer")
const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, "steer")
const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, input.sessionID, "queue")
shouldRun = hasSteer || hasQueue
promotion = hasSteer ? "steer" : hasQueue ? "queue" : undefined
trigger = hasSteer || hasQueue ? "input" : "resume"
}
yield* runPendingCompaction(input.sessionID)
const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, "steer")
const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, input.sessionID, "queue")
shouldRun = hasSteer || hasQueue
promotion = hasSteer ? "steer" : hasQueue ? "queue" : undefined
}
})
return yield* AgentTelemetry.invoke(
{ sessionID: input.sessionID, errorType: (cause) => toSessionError(cause).type },
run,
)
})
return Service.of({ drain })

36
packages/core/src/session/runner/retry.ts
View file

@ -4,11 +4,14 @@ import { LLMError } from "@opencode-ai/llm"
import { SessionError } from "@opencode-ai/schema/session-error"
import { Data, Duration, Effect, Schedule } from "effect"
import { EventV2 } from "../../event"
import { AgentTelemetry } from "../../observability/agent"
import { SessionEvent } from "../event"
import { SessionMessage } from "../message"
import { SessionSchema } from "../schema"
import type { SessionRunner } from "./index"
export const maxAttempts = 5
export class RetryableFailure extends Data.TaggedError("SessionRunner.RetryableFailure")<{
readonly cause: LLMError
readonly assistantMessageID: SessionMessage.ID
@ -45,7 +48,7 @@ const retryAfter = (failure: RetryableFailure) => {
export const schedule = (events: EventV2.Interface, sessionID: SessionSchema.ID) =>
Schedule.exponential("2 seconds").pipe(
Schedule.take(4),
Schedule.take(maxAttempts - 1),
Schedule.setInputType<RetryableFailure | SessionRunner.RunError>(),
Schedule.passthrough,
Schedule.while(({ input }) => input instanceof RetryableFailure),
@ -53,15 +56,24 @@ export const schedule = (events: EventV2.Interface, sessionID: SessionSchema.ID)
const minimum = failure instanceof RetryableFailure ? retryAfter(failure) : undefined
return Effect.succeed(minimum === undefined ? delay : Duration.max(delay, Duration.millis(minimum)))
}),
Schedule.tap((metadata) =>
metadata.input instanceof RetryableFailure
? events.publish(SessionEvent.RetryScheduled, {
sessionID,
assistantMessageID: metadata.input.assistantMessageID,
attempt: metadata.attempt + 1,
at: metadata.now + Duration.toMillis(metadata.duration),
error: metadata.input.error,
})
: Effect.void,
),
Schedule.tap((metadata) => {
const failure = metadata.input
if (!(failure instanceof RetryableFailure)) return Effect.void
return Effect.gen(function* () {
yield* events.publish(SessionEvent.RetryScheduled, {
sessionID,
assistantMessageID: failure.assistantMessageID,
attempt: metadata.attempt + 1,
at: metadata.now + Duration.toMillis(metadata.duration),
error: failure.error,
})
yield* AgentTelemetry.retryScheduled({
attempt: metadata.attempt + 1,
maxAttempts,
delayMs: Duration.toMillis(metadata.duration),
retryAfterMs: retryAfter(failure),
errorType: failure.error.type,
})
})
}),
)

31
packages/core/src/tool/execute.ts
View file

@ -3,6 +3,8 @@ export * as ExecuteTool from "./execute"
import { CodeMode, Tool, toolError } from "@opencode-ai/codemode"
import { ToolOutput } from "@opencode-ai/llm"
import { Effect, Ref, Schema } from "effect"
import { ToolTelemetry } from "../observability/tool"
import { toSessionError } from "../session/to-session-error"
import { definition, make, settle, type AnyTool } from "./tool"
const ExecuteFile = Schema.Struct({
@ -115,18 +117,31 @@ export const create = (options: {
(name, registration, input) =>
Effect.gen(function* () {
const index = yield* Ref.getAndUpdate(callIndex, (index) => index + 1)
const current = options.current(name)
if (!current || current.identity !== registration.identity)
return yield* Effect.fail(toolError(`Stale tool call: ${name}`))
const output = yield* settle(
current.tool,
{ type: "tool-call", id: context.toolCallID, name, input },
const state = { stale: false }
const output = yield* ToolTelemetry.execute(
{
sessionID: context.sessionID,
agent: context.agent,
assistantMessageID: context.assistantMessageID,
toolCallID: context.toolCallID,
call: { id: `${context.toolCallID}/${index + 1}`, name },
},
Effect.gen(function* () {
const current = options.current(name)
if (!current || current.identity !== registration.identity) {
state.stale = true
return yield* Effect.fail(toolError(`Stale tool call: ${name}`))
}
return yield* settle(
current.tool,
{ type: "tool-call", id: context.toolCallID, name, input },
{
sessionID: context.sessionID,
agent: context.agent,
assistantMessageID: context.assistantMessageID,
toolCallID: context.toolCallID,
},
)
}),
(cause) => (state.stale ? "tool.stale" : toSessionError(cause).type),
).pipe(Effect.mapError((failure) => toolError(failure.message, failure)))
const outputFileParts = outputFiles(output)
if (outputFileParts.length > 0)

31
packages/core/src/tool/registry.ts
View file

@ -16,6 +16,7 @@ import { ToolHooks } from "./hooks"
import { makeLocationNode } from "../effect/app-node"
import { SessionError } from "@opencode-ai/schema/session-error"
import { toSessionError } from "../session/to-session-error"
import { ToolTelemetry } from "../observability/tool"
export type ExecuteInput = {
readonly sessionID: SessionSchema.ID
@ -216,15 +217,27 @@ const registryLayer = Layer.effect(
...Array.from(direct, ([name, registration]) => definition(name, registration.tool)),
...(execute ? [definition("execute", execute)] : []),
],
settle: (input) => {
if (input.call.name === "execute" && execute) return settleTool(input, execute)
const registration = direct.get(input.call.name)
if (registration) return settleWith(input, registration.identity)
return Effect.succeed({
result: { type: "error", value: `Unknown tool: ${input.call.name}` },
error: { type: "tool.unknown", message: `Unknown tool: ${input.call.name}` },
})
},
settle: (input) =>
ToolTelemetry.execute(
input,
Effect.suspend(() => {
if (input.call.name === "execute" && execute) return settleTool(input, execute)
const registration = direct.get(input.call.name)
if (registration) return settleWith(input, registration.identity)
return Effect.succeed({
result: { type: "error" as const, value: `Unknown tool: ${input.call.name}` },
error: { type: "tool.unknown" as const, message: `Unknown tool: ${input.call.name}` },
})
}),
(cause) => toSessionError(cause).type,
(settlement) => {
if (settlement.error) return settlement.error.type
if (input.call.name !== "execute") return
const structured = settlement.output?.structured
if (typeof structured !== "object" || structured === null || !("error" in structured)) return
return structured.error === true ? "tool.execution" : undefined
},
),
}
}),
})

4
packages/core/src/tool/subagent.ts
View file

@ -5,6 +5,7 @@ import type { PluginContext } from "@opencode-ai/plugin/v2/effect"
import { Effect, Schema, Scope } from "effect"
import { AgentV2 } from "../agent"
import { PluginRuntime } from "../plugin/runtime"
import { ToolTelemetry } from "../observability/tool"
import { SessionSchema } from "../session/schema"
import { Tool } from "./tool"
@ -131,13 +132,14 @@ export const Plugin = {
(error) => new ToolFailure({ message: `Parent session not found: ${context.sessionID}`, error }),
),
)
const telemetry = yield* ToolTelemetry.child({ agent: input.agent, sessionID: child.id })
const background = input.background === true
const run = Effect.gen(function* () {
// The child session owns its agent/model (set at create); prompt only admits input.
yield* runtime.session.prompt({ sessionID: child.id, prompt: { text: input.prompt }, resume: false })
yield* runtime.session.resume(child.id)
yield* telemetry.resume(runtime.session.resume(child.id), background)
return yield* latestAssistantText(child.id)
}).pipe(Effect.onInterrupt(() => runtime.session.interrupt(child.id)))

31
packages/core/src/tool/webfetch.ts
View file

@ -6,6 +6,7 @@ import { Duration, Effect, Schema } from "effect"
import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
import { Parser } from "htmlparser2"
import TurndownService from "turndown"
import { HttpTelemetry } from "../observability/http"
import { PermissionV2 } from "../permission"
import { collectBoundedResponseBody } from "./http-body"
import { Tool } from "./tool"
@ -84,7 +85,21 @@ const assertHttpUrl = (url: URL) => {
}
const execute = (http: HttpClient.HttpClient, url: string, format: Format, userAgent = browserUserAgent) =>
http.execute(request(url, format, userAgent)).pipe(Effect.flatMap(HttpClientResponse.filterStatusOk))
HttpTelemetry.use(
http,
request(url, format, userAgent),
(response) =>
Effect.gen(function* () {
const contentType = response.headers["content-type"] || ""
const mime = mimeFrom(contentType)
if (isImageAttachment(mime))
return yield* Effect.fail(new Error(`Unsupported fetched image content type: ${mime}`))
if (!isTextualMime(mime))
return yield* Effect.fail(new Error(`Unsupported fetched file content type: ${mime}`))
return { body: yield* collectBody(response), contentType }
}),
HttpClientResponse.filterStatusOk,
)
const collectBody = (response: HttpClientResponse.HttpClientResponse) =>
collectBoundedResponseBody(
@ -144,18 +159,8 @@ export const Plugin = {
source: { type: "tool", messageID: context.assistantMessageID, callID: context.toolCallID },
})
const { body, contentType } = yield* Effect.gen(function* () {
const response = yield* execute(http, input.url, input.format).pipe(
Effect.catchIf(isCloudflareChallenge, () => execute(http, input.url, input.format, "opencode")),
)
const contentType = response.headers["content-type"] || ""
const mime = mimeFrom(contentType)
if (isImageAttachment(mime))
return yield* Effect.fail(new Error(`Unsupported fetched image content type: ${mime}`))
if (!isTextualMime(mime))
return yield* Effect.fail(new Error(`Unsupported fetched file content type: ${mime}`))
return { body: yield* collectBody(response), contentType }
}).pipe(
const { body, contentType } = yield* execute(http, input.url, input.format).pipe(
Effect.catchIf(isCloudflareChallenge, () => execute(http, input.url, input.format, "opencode")),
Effect.timeoutOrElse({
duration: Duration.seconds(input.timeout ?? DEFAULT_TIMEOUT_SECONDS),
orElse: () => Effect.fail(new Error("Request timed out")),

16
packages/core/src/tool/websearch.ts
View file

@ -7,6 +7,7 @@ import { HttpClient, HttpClientRequest } from "effect/unstable/http"
import { makeLocationNode } from "../effect/app-node"
import { truthy } from "../flag/flag"
import { InstallationVersion } from "../installation/version"
import { HttpTelemetry } from "../observability/http"
import { PositiveInt } from "../schema"
import { PermissionV2 } from "../permission"
import { Tool } from "./tool"
@ -167,13 +168,16 @@ const callMcp = <F extends Schema.Struct.Fields>(
}),
)
return yield* Effect.gen(function* () {
const response = yield* HttpClient.filterStatusOk(http).execute(request)
const body = yield* collectBoundedResponseBody(
response,
MAX_RESPONSE_BYTES,
() => new Error(`${tool} response exceeded ${MAX_RESPONSE_BYTES} bytes`),
return yield* HttpTelemetry.use(
HttpClient.filterStatusOk(http),
request,
(response) =>
collectBoundedResponseBody(
response,
MAX_RESPONSE_BYTES,
() => new Error(`${tool} response exceeded ${MAX_RESPONSE_BYTES} bytes`),
).pipe(Effect.flatMap((body) => parseResponse(body.toString("utf8")))),
)
return yield* parseResponse(body.toString("utf8"))
}).pipe(
Effect.timeoutOrElse({
duration: Duration.seconds(25),

96
packages/core/test/effect/observability.test.ts
View file

@ -1,11 +1,23 @@
import { afterEach, describe, expect, test } from "bun:test"
import { NodeFileSystem } from "@effect/platform-node"
import { Effect, Layer, Logger } from "effect"
import {
ATTR_DEPLOYMENT_ENVIRONMENT_NAME,
ATTR_OPENCODE_CLIENT,
ATTR_OPENCODE_RUN,
ATTR_SERVICE_INSTANCE_ID,
ATTR_SERVICE_NAMESPACE,
} from "@opencode-ai/core/observability/semconv"
import { Deferred, Effect, Fiber, Layer, Logger, Option, Tracer } from "effect"
import { ParentSpan, type Span } from "effect/Tracer"
import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
import fs from "fs/promises"
import os from "os"
import path from "path"
import { fileLogger } from "../../src/observability/logging"
import { resource } from "../../src/observability/otlp"
import { SessionTelemetry } from "../../src/observability/session"
import { HttpTelemetry } from "../../src/observability/http"
import { it } from "../lib/effect"
const otelResourceAttributes = process.env.OTEL_RESOURCE_ATTRIBUTES
const opencodeClient = process.env.OPENCODE_CLIENT
@ -20,8 +32,7 @@ afterEach(() => {
describe("resource", () => {
test("parses and decodes OTEL resource attributes", () => {
process.env.OTEL_RESOURCE_ATTRIBUTES =
"service.namespace=anomalyco,team=platform%2Cobservability,label=hello%3Dworld,key%2Fname=value%20here"
process.env.OTEL_RESOURCE_ATTRIBUTES = `${ATTR_SERVICE_NAMESPACE}=anomalyco,team=platform%2Cobservability,label=hello%3Dworld,key%2Fname=value%20here`
expect(resource().attributes).toMatchObject({
"service.namespace": "anomalyco",
@ -32,26 +43,87 @@ describe("resource", () => {
})
test("drops OTEL resource attributes when any entry is invalid", () => {
process.env.OTEL_RESOURCE_ATTRIBUTES = "service.namespace=anomalyco,broken"
process.env.OTEL_RESOURCE_ATTRIBUTES = `${ATTR_SERVICE_NAMESPACE}=anomalyco,broken`
expect(resource().attributes["service.namespace"]).toBeUndefined()
expect(resource().attributes["opencode.client"]).toBeDefined()
expect(resource().attributes[ATTR_SERVICE_NAMESPACE]).toBeUndefined()
expect(resource().attributes[ATTR_OPENCODE_CLIENT]).toBeDefined()
})
test("keeps built-in attributes when env values conflict", () => {
process.env.OPENCODE_CLIENT = "cli"
process.env.OTEL_RESOURCE_ATTRIBUTES =
"opencode.client=web,service.instance.id=override,service.namespace=anomalyco"
process.env.OTEL_RESOURCE_ATTRIBUTES = `${ATTR_OPENCODE_CLIENT}=web,${ATTR_SERVICE_INSTANCE_ID}=override,${ATTR_SERVICE_NAMESPACE}=anomalyco`
expect(resource().attributes).toMatchObject({
"opencode.client": "cli",
"service.namespace": "anomalyco",
[ATTR_OPENCODE_CLIENT]: "cli",
[ATTR_SERVICE_NAMESPACE]: "anomalyco",
})
expect(resource().attributes["service.instance.id"]).not.toBe("override")
expect(resource().attributes["opencode.run"]).toMatch(/^[0-9a-f]{8}$/)
expect(resource().attributes[ATTR_SERVICE_INSTANCE_ID]).not.toBe("override")
expect(resource().attributes[ATTR_OPENCODE_RUN]).toMatch(/^[0-9a-f]{8}$/)
})
test("uses deployment environment from OTEL resource attributes", () => {
process.env.OTEL_RESOURCE_ATTRIBUTES = `${ATTR_DEPLOYMENT_ENVIRONMENT_NAME}=development`
expect(resource().attributes[ATTR_DEPLOYMENT_ENVIRONMENT_NAME]).toBe("development")
})
})
it.effect("retains an execution trace parent until the execution settles", () =>
Effect.gen(function* () {
const telemetry = SessionTelemetry.makeExecution<string>()
const started = yield* Deferred.make<void>()
let parent: Span | undefined
yield* Effect.useSpan("parent", (span) =>
Effect.gen(function* () {
parent = span
const joiner = yield* telemetry
.resume("session", Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)))
.pipe(Effect.provideService(SessionTelemetry.TraceParent, span), Effect.forkChild)
yield* Deferred.await(started)
yield* Fiber.interrupt(joiner)
}),
)
const retained = Option.getOrUndefined(yield* telemetry.drain("session", Effect.serviceOption(ParentSpan)))
expect(retained).toBe(parent)
yield* telemetry.settled("session")
const released = Option.getOrUndefined(yield* telemetry.drain("session", Effect.serviceOption(ParentSpan)))
expect(released).toBeUndefined()
}),
)
it.effect("retains an external ambient trace parent", () =>
Effect.gen(function* () {
const telemetry = SessionTelemetry.makeExecution<string>()
const parent = Tracer.externalSpan({ traceId: "1".repeat(32), spanId: "2".repeat(16) })
yield* telemetry.resume("session", Effect.void).pipe(Effect.withParentSpan(parent))
const retained = Option.getOrUndefined(yield* telemetry.drain("session", Effect.serviceOption(ParentSpan)))
expect(retained).toBe(parent)
}),
)
it.effect("applies HTTP response validation without a parent span", () =>
Effect.gen(function* () {
const request = HttpClientRequest.get("https://example.test/missing")
const http = HttpClient.make((request) =>
Effect.succeed(HttpClientResponse.fromWeb(request, new Response("missing", { status: 404 }))),
)
const exit = yield* HttpTelemetry.use(
http,
request,
Effect.succeed,
HttpClientResponse.filterStatusOk,
).pipe(Effect.exit)
expect(exit._tag).toBe("Failure")
}),
)
test("falls back to local logging when OTLP initialization fails", async () => {
const dir = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-observability-test-"))
await using _ = {

18
packages/core/test/mcp.test.ts
View file

@ -15,7 +15,8 @@ import { SessionV2 } from "@opencode-ai/core/session"
import { McpTool } from "@opencode-ai/core/tool/mcp"
import { ToolRegistry } from "@opencode-ai/core/tool/registry"
import { ToolOutputStore } from "@opencode-ai/core/tool-output-store"
import { Deferred, Effect, Fiber, Layer, Stream } from "effect"
import { ATTR_ERROR_TYPE } from "@opencode-ai/core/observability/semconv"
import { Deferred, Effect, Fiber, Layer, Stream, Tracer } from "effect"
import { testEffect } from "./lib/effect"
import { settleTool, toolDefinitions, toolIdentity, waitForTool } from "./lib/tool"
@ -293,6 +294,14 @@ it.effect("waits for permission before calling an MCP tool", () =>
it.effect("does not call MCP when permission is blocked", () =>
Effect.gen(function* () {
const spans: Tracer.NativeSpan[] = []
const tracer = Tracer.make({
span(options) {
const span = new Tracer.NativeSpan(options)
spans.push(span)
return span
},
})
calls = 0
assertion = yield* Deferred.make<PermissionV2.AssertInput>()
decision = Effect.fail(new PermissionV2.BlockedError({ rules: [], permission: "demo_search", resources: ["*"] }))
@ -308,12 +317,17 @@ it.effect("does not call MCP when permission is blocked", () =>
name: "execute",
input: { code: "return await tools.demo.search({})" },
},
})
}).pipe(Effect.provideService(Tracer.Tracer, tracer))
expect(settlement.result).toEqual({ type: "text", value: "Unable to execute demo_search" })
expect(settlement.output?.structured).toEqual({
toolCalls: [{ tool: "demo.search", status: "error" }],
error: true,
})
expect(calls).toBe(0)
const outer = spans.find((span) => span.name === "execute_tool execute")
const nested = spans.find((span) => span.name === "execute_tool demo_search")
expect(outer?.attributes.get(ATTR_ERROR_TYPE)).toBe("tool.execution")
expect(nested?.attributes.get(ATTR_ERROR_TYPE)).toBe("tool.execution")
expect(nested?.parent._tag === "Some" && nested.parent.value === outer).toBeTrue()
}),
)

30
packages/core/test/session-runner-recorded.test.ts
View file

@ -1,5 +1,11 @@
import { HttpRecorder } from "@opencode-ai/http-recorder"
import { HttpRecorderInternal } from "@opencode-ai/http-recorder/internal"
import {
ATTR_GEN_AI_CONVERSATION_ID,
ATTR_GEN_AI_USAGE_INPUT_TOKENS,
ATTR_GEN_AI_USAGE_OUTPUT_TOKENS,
ATTR_HTTP_REQUEST_METHOD,
} from "@opencode-ai/core/observability/semconv"
import * as OpenAIChat from "@opencode-ai/llm/protocols/openai-chat"
import { Auth, LLMClient, RequestExecutor } from "@opencode-ai/llm/route"
import { Database } from "@opencode-ai/core/database/database"
@ -39,7 +45,7 @@ import { ReferenceGuidance } from "@opencode-ai/core/reference/guidance"
import { McpGuidance } from "@opencode-ai/core/mcp/guidance"
import { describe, expect } from "bun:test"
import { eq } from "drizzle-orm"
import { Effect, Layer } from "effect"
import { Effect, Layer, Tracer } from "effect"
import path from "node:path"
import { testEffect } from "./lib/effect"
@ -93,6 +99,14 @@ const runnerLayer = AppNodeBuilder.build(SessionRunnerLLM.node, [
[PermissionV2.node, permission],
[ToolOutputStore.node, ToolOutputStore.nodeWithoutConfig],
])
const spans: Tracer.NativeSpan[] = []
const tracer = Tracer.make({
span(options) {
const span = new Tracer.NativeSpan(options)
spans.push(span)
return span
},
})
const execution = Layer.effect(
SessionExecution.Service,
Effect.gen(function* () {
@ -108,7 +122,10 @@ const execution = Layer.effect(
awaitIdle: coordinator.awaitIdle,
})
}),
).pipe(Layer.provide(runnerLayer))
).pipe(
Layer.provide(runnerLayer),
Layer.updateService(Tracer.Tracer, () => tracer),
)
const it = testEffect(
AppNodeBuilder.build(
LayerNode.group([
@ -149,6 +166,7 @@ const sessionID = SessionV2.ID.make("ses_runner_recorded")
describe("SessionRunnerLLM recorded", () => {
it.effect("executes one recorded V2 prompt through the recorded HTTP transport", () =>
Effect.gen(function* () {
spans.length = 0
const { db } = yield* Database.Service
yield* db
.insert(ProjectTable)
@ -185,6 +203,14 @@ describe("SessionRunnerLLM recorded", () => {
expect(messages[1]?.type === "assistant" ? messages[1].content : []).toMatchObject([
{ type: "text", text: "Hello!" },
])
const agent = spans.find((span) => span.name === "invoke_agent")
const model = spans.find((span) => span.name === "chat gpt-4o-mini")
const http = spans.find((span) => span.attributes.get(ATTR_HTTP_REQUEST_METHOD) === "POST")
expect(agent?.attributes.get(ATTR_GEN_AI_CONVERSATION_ID)).toBe(sessionID)
expect(model?.parent._tag === "Some" ? model.parent.value.spanId : undefined).toBe(agent?.spanId)
expect(http?.parent._tag === "Some" ? http.parent.value.spanId : undefined).toBe(model?.spanId)
expect(model?.attributes.get(ATTR_GEN_AI_USAGE_INPUT_TOKENS)).toBeNumber()
expect(model?.attributes.get(ATTR_GEN_AI_USAGE_OUTPUT_TOKENS)).toBeNumber()
expect(
(yield* db
.select({ type: EventTable.type })

43
packages/core/test/session-runner-tool-registry.test.ts
View file

@ -7,8 +7,22 @@ import { SessionV2 } from "@opencode-ai/core/session"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { ToolOutputStore } from "@opencode-ai/core/tool-output-store"
import { ToolRegistry } from "@opencode-ai/core/tool/registry"
import { ATTR_ERROR_TYPE, ATTR_GEN_AI_TOOL_CALL_ID } from "@opencode-ai/core/observability/semconv"
import { executeTool, settleTool, testModel, toolDefinitions } from "./lib/tool"
import { Cause, Deferred, Effect, Exit, Fiber, Layer, Option, Schema, SchemaGetter, SchemaIssue, Scope } from "effect"
import {
Cause,
Deferred,
Effect,
Exit,
Fiber,
Layer,
Option,
Schema,
SchemaGetter,
SchemaIssue,
Scope,
Tracer,
} from "effect"
import { testEffect } from "./lib/effect"
const bounds: ToolOutputStore.BoundInput[] = []
@ -212,6 +226,33 @@ describe("ToolRegistry", () => {
}),
)
it.effect("traces unknown and stale tool attempts", () =>
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 service = yield* ToolRegistry.Service
yield* service.register({ echo: make() })
const materialized = yield* service.materialize({ model: testModel })
yield* materialized.settle(call("missing", "call-unknown")).pipe(Effect.provideService(Tracer.Tracer, tracer))
yield* service.register({ echo: make() })
yield* materialized.settle(call("echo", "call-stale")).pipe(Effect.provideService(Tracer.Tracer, tracer))
expect(spans.map((span) => span.name)).toEqual(["execute_tool missing", "execute_tool echo"])
expect(spans[0]?.attributes.get(ATTR_ERROR_TYPE)).toBe("tool.unknown")
expect(spans[0]?.attributes.get(ATTR_GEN_AI_TOOL_CALL_ID)).toBe("call-unknown")
expect(spans[1]?.attributes.get(ATTR_ERROR_TYPE)).toBe("tool.stale")
expect(spans[1]?.attributes.get(ATTR_GEN_AI_TOOL_CALL_ID)).toBe("call-stale")
expect(spans.every((span) => span.status._tag === "Ended" && span.status.exit._tag === "Failure")).toBeTrue()
}),
)
it.effect("propagates retention failures through settlement", () =>
Effect.gen(function* () {
const service = yield* ToolRegistry.Service

274
packages/core/test/session-runner.test.ts
View file

@ -8,9 +8,42 @@ import {
TransportReason,
InvalidRequestReason,
RateLimitReason,
Usage,
type LLMClientShape,
type LLMRequest,
} from "@opencode-ai/llm"
import {
ATTR_ERROR_TYPE,
ATTR_GEN_AI_AGENT_NAME,
ATTR_GEN_AI_CONVERSATION_ID,
ATTR_GEN_AI_OPERATION_NAME,
ATTR_GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS,
ATTR_GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS,
ATTR_GEN_AI_USAGE_INPUT_TOKENS,
ATTR_GEN_AI_USAGE_OUTPUT_TOKENS,
ATTR_GEN_AI_USAGE_REASONING_OUTPUT_TOKENS,
ATTR_GEN_AI_TOOL_CALL_ID,
ATTR_GEN_AI_TOOL_NAME,
GEN_AI_OPERATION_NAME_VALUE_EXECUTE_TOOL,
GEN_AI_OPERATION_NAME_VALUE_INVOKE_AGENT,
ATTR_OPENCODE_AGENT_STEP_INDEX,
ATTR_OPENCODE_AGENT_STEP_TRIGGER,
ATTR_OPENCODE_COMPACTION_REASON,
ATTR_OPENCODE_ERROR_STAGE,
ATTR_OPENCODE_ERROR_SOURCE,
ATTR_OPENCODE_RETRY_DECISION,
ATTR_OPENCODE_RETRY_DELAY_SOURCE,
ATTR_OPENCODE_RETRY_MAX_ATTEMPTS,
ATTR_OPENCODE_RETRY_ATTEMPT,
ATTR_OPENCODE_RETRY_DELAY_MS,
ATTR_OPENCODE_SESSION_INPUT_COUNT,
ATTR_OPENCODE_SESSION_INPUT_DELIVERY,
ATTR_OPENCODE_SESSION_PARENT_ID,
EVENT_OPENCODE_COMPACTION_COMPLETED,
EVENT_OPENCODE_RETRY_SCHEDULED,
EVENT_OPENCODE_RETRY_STOPPED,
EVENT_OPENCODE_SESSION_INPUT_PROMOTED,
} from "@opencode-ai/core/observability/semconv"
import * as OpenAIChat from "@opencode-ai/llm/protocols/openai-chat"
import { Database } from "@opencode-ai/core/database/database"
import { makeLocationNode } from "@opencode-ai/core/effect/app-node"
@ -61,7 +94,7 @@ import { McpGuidance } from "@opencode-ai/core/mcp/guidance"
import { ModelV2 } from "@opencode-ai/core/model"
import { Location } from "@opencode-ai/core/location"
import { ProviderV2 } from "@opencode-ai/core/provider"
import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Schema, Stream } from "effect"
import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Schema, Stream, Tracer } from "effect"
import { TestClock } from "effect/testing"
import { asc, eq } from "drizzle-orm"
import { testEffect } from "./lib/effect"
@ -259,6 +292,14 @@ const runnerLayer = AppNodeBuilder.build(SessionRunnerLLM.node, [
[McpGuidance.node, mcpGuidance],
[ToolOutputStore.node, ToolOutputStore.nodeWithoutConfig],
])
const spans: Tracer.NativeSpan[] = []
const tracer = Tracer.make({
span(options) {
const span = new Tracer.NativeSpan(options)
spans.push(span)
return span
},
})
const execution = Layer.effect(
SessionExecution.Service,
Effect.gen(function* () {
@ -274,7 +315,10 @@ const execution = Layer.effect(
awaitIdle: coordinator.awaitIdle,
})
}),
).pipe(Layer.provide(runnerLayer))
).pipe(
Layer.provide(runnerLayer),
Layer.updateService(Tracer.Tracer, () => tracer),
)
const it = testEffect(
AppNodeBuilder.build(
LayerNode.group([
@ -338,6 +382,7 @@ const insertSession = (id: SessionV2.ID) =>
const setup = Effect.gen(function* () {
const { db } = yield* Database.Service
requests.length = 0
response = []
systemBaseline = "Initial context"
systemRemoved = false
@ -356,6 +401,7 @@ const setup = Effect.gen(function* () {
toolExecutionsReady = 5
activeToolExecutions = 0
maxActiveToolExecutions = 0
spans.length = 0
yield* db
.insert(ProjectTable)
.values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
@ -643,6 +689,180 @@ const verifyPartialFlushOnInterruption = (kind: FragmentKind) =>
})
describe("SessionRunnerLLM", () => {
it.effect("parents forked tool spans under the V2 agent turn", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Use echo" }), resume: false })
const usage = new Usage({
inputTokens: 8,
outputTokens: 3,
cacheReadInputTokens: 2,
cacheWriteInputTokens: 1,
reasoningTokens: 1,
})
responses = [
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-telemetry", name: "echo", input: { text: "hello" } }),
LLMEvent.stepFinish({ index: 0, reason: "tool-calls", usage }),
LLMEvent.finish({ reason: "tool-calls", usage }),
],
[],
]
yield* session.resume(sessionID)
const agent = spans.find((span) => span.name === "invoke_agent")
const tool = spans.find((span) => span.name === "execute_tool echo")
expect(agent?.attributes).toMatchObject(
new Map([
[ATTR_GEN_AI_OPERATION_NAME, GEN_AI_OPERATION_NAME_VALUE_INVOKE_AGENT],
[ATTR_GEN_AI_AGENT_NAME, "build"],
[ATTR_GEN_AI_CONVERSATION_ID, sessionID],
]),
)
expect(agent?.events.find(([name]) => name === EVENT_OPENCODE_SESSION_INPUT_PROMOTED)?.[2]).toMatchObject({
[ATTR_OPENCODE_SESSION_INPUT_DELIVERY]: "steer",
[ATTR_OPENCODE_SESSION_INPUT_COUNT]: 1,
})
expect(ancestorNames(agent)).toContain("SessionRunner.drain")
expect(agent?.attributes.get(ATTR_GEN_AI_USAGE_INPUT_TOKENS)).toBe(8)
expect(agent?.attributes.get(ATTR_GEN_AI_USAGE_OUTPUT_TOKENS)).toBe(3)
expect(agent?.attributes.get(ATTR_GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS)).toBe(2)
expect(agent?.attributes.get(ATTR_GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS)).toBe(1)
expect(agent?.attributes.get(ATTR_GEN_AI_USAGE_REASONING_OUTPUT_TOKENS)).toBe(1)
expect(tool?.attributes).toMatchObject(
new Map([
[ATTR_GEN_AI_OPERATION_NAME, GEN_AI_OPERATION_NAME_VALUE_EXECUTE_TOOL],
[ATTR_GEN_AI_TOOL_NAME, "echo"],
[ATTR_GEN_AI_TOOL_CALL_ID, "call-telemetry"],
[ATTR_GEN_AI_CONVERSATION_ID, sessionID],
]),
)
expect(ancestorNames(tool)).toContain("invoke_agent")
expect(ancestorNames(tool)).not.toContain("SessionRunner.attemptStep")
expect(ancestorNames(tool)).not.toContain("chat fake-model")
expect(tool?.status._tag === "Ended" && tool.status.exit._tag).toBe("Success")
requests.length = 0
spans.length = 0
}),
)
it.effect("tracks V2 subagent sessions with their parent session", () =>
Effect.gen(function* () {
yield* setup
const agents = yield* AgentV2.Service
yield* agents.transform((editor) =>
editor.update(AgentV2.ID.make("explore"), (agent) => {
agent.mode = "subagent"
}),
)
const session = yield* SessionV2.Service
const child = yield* session.create({
id: otherSessionID,
parentID: sessionID,
agent: AgentV2.ID.make("explore"),
})
yield* session.prompt({
sessionID: child.id,
prompt: PromptInput.Prompt.make({ text: "Explore" }),
resume: false,
})
responses = [
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-subagent-telemetry", name: "echo", input: { text: "found" } }),
LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }),
LLMEvent.finish({ reason: "tool-calls" }),
],
[],
]
yield* session.resume(child.id)
const agent = spans.find((span) => span.name === "invoke_agent")
const tool = spans.find((span) => span.name === "execute_tool echo")
expect(agent?.attributes).toMatchObject(
new Map([
[ATTR_GEN_AI_OPERATION_NAME, GEN_AI_OPERATION_NAME_VALUE_INVOKE_AGENT],
[ATTR_GEN_AI_AGENT_NAME, "explore"],
[ATTR_GEN_AI_CONVERSATION_ID, child.id],
[ATTR_OPENCODE_SESSION_PARENT_ID, sessionID],
]),
)
expect(tool?.attributes).toMatchObject(
new Map([
[ATTR_GEN_AI_AGENT_NAME, "explore"],
[ATTR_GEN_AI_CONVERSATION_ID, child.id],
[ATTR_OPENCODE_SESSION_PARENT_ID, sessionID],
]),
)
expect(ancestorNames(tool)).toContain("invoke_agent")
}),
)
it.effect("distinguishes input and tool-result model calls", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Use echo twice" }),
resume: false,
})
responses = [
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-context-1", name: "echo", input: { text: "first" } }),
LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }),
LLMEvent.finish({ reason: "tool-calls" }),
],
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-context-2", name: "echo", input: { text: "second" } }),
LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }),
LLMEvent.finish({ reason: "tool-calls" }),
],
[],
]
yield* session.resume(sessionID)
const tools = spans.filter((span) => span.name === "execute_tool echo")
expect(tools).toHaveLength(2)
expect(tools[0]?.attributes.get(ATTR_OPENCODE_AGENT_STEP_INDEX)).toBe(1)
expect(tools[0]?.attributes.get(ATTR_OPENCODE_AGENT_STEP_TRIGGER)).toBe("input")
expect(tools[1]?.attributes.get(ATTR_OPENCODE_AGENT_STEP_INDEX)).toBe(2)
expect(tools[1]?.attributes.get(ATTR_OPENCODE_AGENT_STEP_TRIGGER)).toBe("tool_result")
}),
)
it.effect("marks failed tool spans before returning the error to the model", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Fail tool" }), resume: false })
responses = [
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-telemetry-failure", name: "echo", input: {} }),
LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }),
LLMEvent.finish({ reason: "tool-calls" }),
],
[],
]
yield* session.resume(sessionID)
const tool = spans.find((span) => span.name === "execute_tool echo")
expect(tool?.attributes.get(ATTR_ERROR_TYPE)).toBe("tool.execution")
expect(tool?.status._tag === "Ended" && tool.status.exit._tag).toBe("Failure")
requests.length = 0
spans.length = 0
}),
)
it.effect("advertises and executes a location registered tool", () =>
Effect.gen(function* () {
yield* setup
@ -1445,6 +1665,12 @@ describe("SessionRunnerLLM", () => {
yield* session.resume(sessionID)
expect(requests).toHaveLength(2)
expect(
spans
.filter((span) => span.name === "invoke_agent")
.at(-1)
?.events.find(([name]) => name === EVENT_OPENCODE_COMPACTION_COMPLETED)?.[2],
).toMatchObject({ [ATTR_OPENCODE_COMPACTION_REASON]: "automatic" })
expect(userTexts(requests[0])[0]).toContain("## Objective")
expect(userTexts(requests[1])).toHaveLength(1)
expect(userTexts(requests[1])[0]).toContain("<summary>\n## Objective\n- Preserve the task\n</summary>")
@ -3429,6 +3655,9 @@ describe("SessionRunnerLLM", () => {
{ type: "assistant", finish: "error", error: { type: "aborted", message: "Step interrupted" } },
])
expect(yield* recordedEventTypes(sessionID)).toContain("session.step.failed.1")
const agent = spans.find((span) => span.name === "invoke_agent")
expect(agent?.attributes.get(ATTR_ERROR_TYPE)).toBe("canceled")
expect(agent?.status._tag === "Ended" && agent.status.exit._tag).toBe("Failure")
yield* session.interrupt(sessionID)
}),
)
@ -3766,7 +3995,20 @@ describe("SessionRunnerLLM", () => {
expect(requests).toHaveLength(2)
const eventTypes = yield* recordedEventTypes(sessionID)
expect(eventTypes).toContain("session.retry.scheduled.1")
expect(
spans
.find((span) => span.name === "invoke_agent")
?.events.find(([name]) => name === EVENT_OPENCODE_RETRY_SCHEDULED)?.[2],
).toMatchObject({
[ATTR_OPENCODE_RETRY_ATTEMPT]: 2,
[ATTR_OPENCODE_RETRY_MAX_ATTEMPTS]: 5,
[ATTR_OPENCODE_RETRY_DELAY_MS]: 2_000,
[ATTR_OPENCODE_RETRY_DELAY_SOURCE]: "backoff",
[ATTR_OPENCODE_RETRY_DECISION]: "scheduled",
[ATTR_ERROR_TYPE]: "provider.transport",
})
expect(eventTypes.filter((type) => type === "session.step.started.1")).toHaveLength(2)
expect(spans.find((span) => span.name === "invoke_agent")?.attributes.has(ATTR_OPENCODE_ERROR_STAGE)).toBeFalse()
expect(yield* session.context(sessionID)).toMatchObject([
{ type: "user" },
{ type: "assistant", finish: "stop", content: [{ type: "text", text: "Recovered" }] },
@ -3826,8 +4068,26 @@ describe("SessionRunnerLLM", () => {
{ attempt: 4, at: 14_000 },
{ attempt: 5, at: 30_000 },
])
expect(
spans
.find((span) => span.name === "invoke_agent")
?.events.find(([name]) => name === EVENT_OPENCODE_RETRY_STOPPED)?.[2],
).toMatchObject({
[ATTR_OPENCODE_RETRY_DECISION]: "exhausted",
[ATTR_OPENCODE_RETRY_ATTEMPT]: 5,
[ATTR_OPENCODE_RETRY_MAX_ATTEMPTS]: 5,
})
expect((yield* recordedEventTypes(sessionID)).filter((type) => type === "session.step.started.1")).toHaveLength(5)
expect((yield* session.context(sessionID)).filter((message) => message.type === "assistant")).toHaveLength(1)
const agent = spans.find((span) => span.name === "invoke_agent")
expect(agent?.attributes.get(ATTR_ERROR_TYPE)).toBe("provider.transport")
expect(agent?.attributes.get(ATTR_OPENCODE_ERROR_SOURCE)).toBe("provider")
expect(agent?.attributes.get(ATTR_OPENCODE_ERROR_STAGE)).toBe("model")
if (agent?.status._tag === "Ended" && agent.status.exit._tag === "Failure") {
const failure = Cause.squash(agent.status.exit.cause)
expect(failure).toMatchObject({ message: "provider.transport" })
expect(failure).not.toMatchObject({ message: "Provider unavailable" })
}
}),
)
@ -4310,3 +4570,13 @@ describe("SessionRunnerLLM", () => {
}),
)
})
function ancestorNames(span: Tracer.NativeSpan | undefined) {
const names: string[] = []
let current = span?.parent._tag === "Some" ? span.parent.value : undefined
while (current?._tag === "Span") {
names.push(current.name)
current = current.parent._tag === "Some" ? current.parent.value : undefined
}
return names
}

40
packages/core/test/tool-subagent.test.ts
View file

@ -1,5 +1,9 @@
import { describe, expect } from "bun:test"
import { DateTime, Effect, Layer, Schema } from "effect"
import { DateTime, Effect, Layer, Schema, Tracer } from "effect"
import {
ATTR_OPENCODE_SUBAGENT_AGENT_NAME,
ATTR_OPENCODE_SUBAGENT_SESSION_ID,
} from "@opencode-ai/core/observability/semconv"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
import { LayerNode } from "@opencode-ai/core/effect/layer-node"
import { makeGlobalNode } from "@opencode-ai/core/effect/app-node"
@ -22,6 +26,7 @@ import { PluginRuntime } from "@opencode-ai/core/plugin/runtime"
import { SubagentTool } from "@opencode-ai/core/tool/subagent"
import { ToolRegistry } from "@opencode-ai/core/tool/registry"
import { ToolOutputStore } from "@opencode-ai/core/tool-output-store"
import { SessionTelemetry } from "@opencode-ai/core/observability/session"
import { tmpdir } from "./fixture/tmpdir"
import { testEffect } from "./lib/effect"
import { executeTool, settleTool, testModel, toolIdentity, waitForTool } from "./lib/tool"
@ -30,6 +35,7 @@ const childText = "child final response"
const childModel = ModelV2.Ref.make({ id: ModelV2.ID.make("child"), providerID: ProviderV2.ID.make("test") })
const parentModel = ModelV2.Ref.make({ id: ModelV2.ID.make("parent"), providerID: ProviderV2.ID.make("test") })
const tokens = { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }
let resumedWith: Tracer.AnySpan | undefined
const outputSessionID = (value: unknown) => Schema.decodeUnknownSync(SubagentTool.Output)(value).sessionID
@ -76,7 +82,11 @@ const executionNode = makeGlobalNode({
})
return SessionExecution.Service.of({
active: Effect.succeed(new Set()),
resume: complete,
resume: (sessionID) =>
Effect.gen(function* () {
resumedWith = (yield* SessionTelemetry.TraceParent) ?? undefined
return yield* complete(sessionID)
}),
wake: () => Effect.void,
interrupt: () => Effect.void,
awaitIdle: (sessionID) => complete(sessionID).pipe(Effect.exit, Effect.asVoid),
@ -171,6 +181,14 @@ describe("SubagentTool", () => {
const locations = yield* LocationServiceMap.Service
const registry = yield* ToolRegistry.Service.pipe(Effect.provide(locations.get(parent.location)))
yield* waitForTool(registry, SubagentTool.name)
const spans: Tracer.NativeSpan[] = []
const tracer = Tracer.make({
span(options) {
const span = new Tracer.NativeSpan(options)
spans.push(span)
return span
},
})
const settled = yield* settleTool(registry, {
sessionID: parent.id,
@ -181,7 +199,7 @@ describe("SubagentTool", () => {
name: SubagentTool.name,
input: { agent: "reviewer", description: "review", prompt: "review this" },
},
})
}).pipe(Effect.provideService(Tracer.Tracer, tracer))
expect(settled.output?.structured).toMatchObject({ status: "completed", output: childText })
const child = yield* sessions.get(outputSessionID(settled.output?.structured))
@ -191,6 +209,10 @@ describe("SubagentTool", () => {
agent: "reviewer",
model: childModel,
})
const span = spans.find((span) => span.name === "execute_tool subagent")
expect(span?.attributes.get(ATTR_OPENCODE_SUBAGENT_AGENT_NAME)).toBe("reviewer")
expect(span?.attributes.get(ATTR_OPENCODE_SUBAGENT_SESSION_ID)).toBe(child.id)
expect(resumedWith?.spanId).toBe(span?.spanId)
const fallback = yield* settleTool(registry, {
sessionID: parent.id,
@ -255,6 +277,15 @@ describe("SubagentTool", () => {
const locations = yield* LocationServiceMap.Service
const registry = yield* ToolRegistry.Service.pipe(Effect.provide(locations.get(parent.location)))
yield* waitForTool(registry, SubagentTool.name)
resumedWith = undefined
const spans: Tracer.NativeSpan[] = []
const tracer = Tracer.make({
span(options) {
const span = new Tracer.NativeSpan(options)
spans.push(span)
return span
},
})
const settled = yield* settleTool(registry, {
sessionID: parent.id,
@ -265,7 +296,7 @@ describe("SubagentTool", () => {
name: SubagentTool.name,
input: { agent: "reviewer", description: "background review", prompt: "review", background: true },
},
})
}).pipe(Effect.provideService(Tracer.Tracer, tracer))
const childID = outputSessionID(settled.output?.structured)
expect(settled.output?.structured).toMatchObject({ status: "running" })
@ -274,6 +305,7 @@ describe("SubagentTool", () => {
expect(synthetic).toHaveLength(1)
expect(synthetic[0]?.text).toContain(`<subagent id="${childID}" state="completed"`)
expect(synthetic[0]?.text).toContain(childText)
expect(resumedWith).toBeUndefined()
}),
),
),

55
packages/core/test/tool-webfetch.test.ts
View file

@ -1,5 +1,12 @@
import { describe, expect, test } from "bun:test"
import { Duration, Effect, Fiber, Layer, Schema } from "effect"
import { Duration, Effect, Fiber, Layer, Schema, Tracer } from "effect"
import {
ATTR_ERROR_TYPE,
ATTR_HTTP_REQUEST_METHOD,
ATTR_HTTP_RESPONSE_STATUS_CODE,
ATTR_SERVER_PORT,
ATTR_URL_FULL,
} from "@opencode-ai/core/observability/semconv"
import * as TestClock from "effect/testing/TestClock"
import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
@ -88,9 +95,21 @@ describe("WebFetchTool registration", () => {
reset()
const registry = yield* ToolRegistry.Service
const url = "http://example.com/public"
const spans: Tracer.NativeSpan[] = []
const tracer = Tracer.make({
span(options) {
const span = new Tracer.NativeSpan(options)
spans.push(span)
return span
},
})
expect((yield* toolDefinitions(registry)).map((tool) => tool.name)).toEqual(["webfetch"])
expect(yield* settleTool(registry, call({ url, format: "text", timeout: 4 }))).toEqual({
expect(
yield* settleTool(registry, call({ url, format: "text", timeout: 4 })).pipe(
Effect.provideService(Tracer.Tracer, tracer),
),
).toEqual({
result: { type: "text", value: "hello" },
output: {
structured: { url, contentType: "text/plain", format: "text", output: "hello" },
@ -101,6 +120,38 @@ describe("WebFetchTool registration", () => {
{ sessionID, action: "webfetch", resources: [url], save: ["*"], metadata: { url, format: "text", timeout: 4 } },
])
expect(requests).toMatchObject([{ url, headers: { accept: expect.stringContaining("text/plain;q=1.0") } }])
const tool = spans.find((span) => span.name === "execute_tool webfetch")
const http = spans.find((span) => span.attributes.get(ATTR_HTTP_REQUEST_METHOD) === "GET")
expect(http?.name).toBe("GET")
expect(http?.attributes.get(ATTR_SERVER_PORT)).toBe(80)
expect(http?.attributes.get(ATTR_URL_FULL)).toBe(url)
expect(http?.parent._tag === "Some" ? http.parent.value.spanId : undefined).toBe(tool?.spanId)
expect(requests[0]?.headers.traceparent).toBeUndefined()
}),
)
it.effect("marks rejected HTTP statuses on the HTTP span", () =>
Effect.gen(function* () {
reset()
respond = () => Effect.succeed(new Response("missing", { status: 404 }))
const registry = yield* ToolRegistry.Service
const spans: Tracer.NativeSpan[] = []
const tracer = Tracer.make({
span(options) {
const span = new Tracer.NativeSpan(options)
spans.push(span)
return span
},
})
yield* settleTool(registry, call({ url: "https://example.com/missing", format: "text" })).pipe(
Effect.provideService(Tracer.Tracer, tracer),
)
const http = spans.find((span) => span.attributes.get(ATTR_HTTP_REQUEST_METHOD) === "GET")
expect(http?.attributes.get(ATTR_HTTP_RESPONSE_STATUS_CODE)).toBe(404)
expect(http?.attributes.get(ATTR_ERROR_TYPE)).toBe("StatusCodeError")
expect(http?.status._tag === "Ended" && http.status.exit._tag).toBe("Failure")
}),
)