refactor(session): consume native LLM events

Author: kitlangton

bun.lock

@@ -105,6 +105,7 @@
     "@octokit/graphql": "9.0.2",
     "@octokit/rest": "catalog:",
     "@openauthjs/openauth": "catalog:",
+    "@opencode-ai/llm": "workspace:*",
     "@opencode-ai/plugin": "workspace:*",
     "@opencode-ai/script": "workspace:*",
     "@opencode-ai/sdk": "workspace:*",

packages/opencode/package.json

@@ -0,0 +1,223 @@
+import { ContentBlockID, FinishReason, LLMEvent, ProviderMetadata, ToolCallID, ToolResultValue, Usage } from "@opencode-ai/llm"
+import { Effect, Schema } from "effect"
+import { type streamText } from "ai"
+import { errorMessage } from "@/util/error"
+
+type Result = Awaited<ReturnType<typeof streamText>>
+type AISDKEvent = Result["fullStream"] extends AsyncIterable<infer T> ? T : never
+
+export function adapterState() {
+  return {
+    step: 0,
+    text: 0,
+    reasoning: 0,
+    currentTextID: undefined as ContentBlockID | undefined,
+    currentReasoningID: undefined as ContentBlockID | undefined,
+    toolNames: {} as Record<string, string>,
+  }
+}
+
+const contentBlockID = (value: string) => ContentBlockID.make(value)
+const toolCallID = (value: string) => ToolCallID.make(value)
+
+function finishReason(value: string | undefined): FinishReason {
+  return Schema.is(FinishReason)(value) ? value : "unknown"
+}
+
+function providerMetadata(value: unknown): ProviderMetadata | undefined {
+  return Schema.is(ProviderMetadata)(value) ? value : undefined
+}
+
+function usage(value: unknown): Usage | undefined {
+  if (!value || typeof value !== "object") return undefined
+  const item = value as {
+    inputTokens?: number
+    outputTokens?: number
+    totalTokens?: number
+    reasoningTokens?: number
+    cachedInputTokens?: number
+    inputTokenDetails?: { cacheReadTokens?: number; cacheWriteTokens?: number }
+    outputTokenDetails?: { reasoningTokens?: number }
+  }
+  const result = Object.fromEntries(
+    Object.entries({
+      inputTokens: item.inputTokens,
+      outputTokens: item.outputTokens,
+      totalTokens: item.totalTokens,
+      reasoningTokens: item.outputTokenDetails?.reasoningTokens ?? item.reasoningTokens,
+      cacheReadInputTokens: item.inputTokenDetails?.cacheReadTokens ?? item.cachedInputTokens,
+      cacheWriteInputTokens: item.inputTokenDetails?.cacheWriteTokens,
+    }).filter((entry) => entry[1] !== undefined),
+  )
+  return new Usage(result)
+}
+
+export function toLLMEvents(
+  state: ReturnType<typeof adapterState>,
+  event: AISDKEvent,
+): Effect.Effect<ReadonlyArray<LLMEvent>, unknown> {
+  switch (event.type) {
+    case "start":
+      return Effect.succeed([])
+
+    case "start-step":
+      return Effect.succeed([LLMEvent.stepStart({ index: state.step })])
+
+    case "finish-step":
+      return Effect.sync(() => [
+        LLMEvent.stepFinish({
+          index: state.step++,
+          reason: finishReason(event.finishReason),
+          usage: usage(event.usage),
+          providerMetadata: providerMetadata(event.providerMetadata),
+        }),
+      ])
+
+    case "finish":
+      return Effect.sync(() => {
+        state.toolNames = {}
+        return [
+          LLMEvent.requestFinish({
+            reason: finishReason(event.finishReason),
+            usage: usage(event.totalUsage),
+          }),
+        ]
+      })
+
+    case "text-start":
+      return Effect.sync(() => {
+        state.currentTextID = contentBlockID(event.id ?? `text-${state.text++}`)
+        return [
+          LLMEvent.textStart({
+            id: state.currentTextID,
+            providerMetadata: providerMetadata(event.providerMetadata),
+          }),
+        ]
+      })
+
+    case "text-delta":
+      return Effect.succeed([
+        LLMEvent.textDelta({
+          id: event.id ? contentBlockID(event.id) : (state.currentTextID ?? contentBlockID(`text-${state.text++}`)),
+          text: event.text,
+        }),
+      ])
+
+    case "text-end":
+      return Effect.succeed([
+        LLMEvent.textEnd({
+          id: event.id ? contentBlockID(event.id) : (state.currentTextID ?? contentBlockID(`text-${state.text++}`)),
+          providerMetadata: providerMetadata(event.providerMetadata),
+        }),
+      ])
+
+    case "reasoning-start":
+      return Effect.sync(() => {
+        state.currentReasoningID = contentBlockID(event.id)
+        return [
+          LLMEvent.reasoningStart({
+            id: state.currentReasoningID,
+            providerMetadata: providerMetadata(event.providerMetadata),
+          }),
+        ]
+      })
+
+    case "reasoning-delta":
+      return Effect.succeed([
+        LLMEvent.reasoningDelta({
+          id: event.id ? contentBlockID(event.id) : (state.currentReasoningID ?? contentBlockID(`reasoning-${state.reasoning++}`)),
+          text: event.text,
+        }),
+      ])
+
+    case "reasoning-end":
+      return Effect.sync(() => {
+        const id = contentBlockID(event.id)
+        state.currentReasoningID = undefined
+        return [
+          LLMEvent.reasoningEnd({
+            id,
+            providerMetadata: providerMetadata(event.providerMetadata),
+          }),
+        ]
+      })
+
+    case "tool-input-start":
+      return Effect.sync(() => {
+        state.toolNames[event.id] = event.toolName
+        return [
+          LLMEvent.toolInputStart({
+            id: toolCallID(event.id),
+            name: event.toolName,
+            providerMetadata: providerMetadata(event.providerMetadata),
+          }),
+        ]
+      })
+
+    case "tool-input-delta":
+      return Effect.succeed([
+        LLMEvent.toolInputDelta({
+          id: toolCallID(event.id),
+          name: state.toolNames[event.id] ?? "unknown",
+          text: event.delta ?? "",
+        }),
+      ])
+
+    case "tool-input-end":
+      return Effect.succeed([
+        LLMEvent.toolInputEnd({
+          id: toolCallID(event.id),
+          name: state.toolNames[event.id] ?? "unknown",
+        }),
+      ])
+
+    case "tool-call":
+      return Effect.sync(() => {
+        state.toolNames[event.toolCallId] = event.toolName
+        return [
+          LLMEvent.toolCall({
+            id: toolCallID(event.toolCallId),
+            name: event.toolName,
+            input: event.input,
+            providerExecuted: "providerExecuted" in event ? event.providerExecuted : undefined,
+            providerMetadata: providerMetadata(event.providerMetadata),
+          }),
+        ]
+      })
+
+    case "tool-result":
+      return Effect.sync(() => {
+        const name = state.toolNames[event.toolCallId] ?? "unknown"
+        delete state.toolNames[event.toolCallId]
+        return [
+          LLMEvent.toolResult({
+            id: toolCallID(event.toolCallId),
+            name,
+            result: ToolResultValue.make(event.output),
+            providerExecuted: "providerExecuted" in event ? event.providerExecuted : undefined,
+          }),
+        ]
+      })
+
+    case "tool-error":
+      return Effect.sync(() => {
+        const name = state.toolNames[event.toolCallId] ?? "unknown"
+        delete state.toolNames[event.toolCallId]
+        return [
+          LLMEvent.toolError({
+            id: toolCallID(event.toolCallId),
+            name,
+            message: errorMessage(event.error),
+          }),
+        ]
+      })
+
+    case "error":
+      return Effect.fail(event.error)
+
+    default:
+      return Effect.succeed([])
+  }
+}
+
+export * as LLMAISDK from "./llm-ai-sdk"

packages/opencode/src/session/llm-ai-sdk.ts

@@ -3,6 +3,7 @@ import * as Log from "@opencode-ai/core/util/log"
 import { Context, Effect, Layer, Record } from "effect"
 import * as Stream from "effect/Stream"
 import { streamText, wrapLanguageModel, type ModelMessage, type Tool, tool, jsonSchema } from "ai"
+import type { LLMEvent } from "@opencode-ai/llm"
 import { mergeDeep } from "remeda"
 import { GitLabWorkflowLanguageModel } from "gitlab-ai-provider"
 import { ProviderTransform } from "@/provider/transform"
@@ -24,10 +25,10 @@ import { InstallationVersion } from "@opencode-ai/core/installation/version"
 import { EffectBridge } from "@/effect/bridge"
 import * as Option from "effect/Option"
 import * as OtelTracer from "@effect/opentelemetry/Tracer"
+import { LLMAISDK } from "./llm-ai-sdk"
 
 const log = Log.create({ service: "llm" })
 export const OUTPUT_TOKEN_MAX = ProviderTransform.OUTPUT_TOKEN_MAX
-type Result = Awaited<ReturnType<typeof streamText>>
 
 // Avoid re-instantiating remeda's deep merge types in this hot LLM path; the runtime behavior is still mergeDeep.
 const mergeOptions = (target: Record<string, any>, source: Record<string, any> | undefined): Record<string, any> =>
@@ -52,10 +53,8 @@ export type StreamRequest = StreamInput & {
   abort: AbortSignal
 }
 
-export type Event = Result["fullStream"] extends AsyncIterable<infer T> ? T : never
-
 export interface Interface {
-  readonly stream: (input: StreamInput) => Stream.Stream<Event, unknown>
+  readonly stream: (input: StreamInput) => Stream.Stream<LLMEvent, unknown>
 }
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/LLM") {}
@@ -427,7 +426,11 @@ const live: Layer.Layer<
 
             const result = yield* run({ ...input, abort: ctrl.signal })
 
-            return Stream.fromAsyncIterable(result.fullStream, (e) => (e instanceof Error ? e : new Error(String(e))))
+            const state = LLMAISDK.adapterState()
+            return Stream.fromAsyncIterable(result.fullStream, (e) => (e instanceof Error ? e : new Error(String(e)))).pipe(
+              Stream.mapEffect((event) => LLMAISDK.toLLMEvents(state, event)),
+              Stream.flatMap((events) => Stream.fromIterable(events)),
+            )
           }),
         ),
       )