feat(opencode): wire LLM-native stream path behind opt-in flag (audit gap #4 phase 1)

Adds the parallel `runNative()` path inside `session/llm.ts` so a narrow slice of sessions can flow through `@opencode-ai/llm` instead of the AI SDK `streamText`. Behavior is gated and shipped off by default; only callers that opt in see any difference. The full migration plan (audit gap #4) is parallel-path-with-flag, prove parity test-by-test, flip default last. This commit is phase 1: get the wire-up in place behind a flag with one protocol so we can see whether the design holds before committing to the full migration. Wire-up summary: - New flag `OPENCODE_EXPERIMENTAL_LLM_NATIVE` (also enabled by the umbrella `OPENCODE_EXPERIMENTAL`). Off by default. - The session-LLM `live` layer now consumes `RequestExecutor.Service`, and the `defaultLayer` provides `RequestExecutor.defaultLayer` so a Node fetch HTTP client backs every native stream. - `runNative(input)` returns `Stream<Event> | undefined`. `undefined` means "fall through to AI SDK." It returns a real stream only when every gate passes: the flag is set, the caller populated `input.nativeMessages` (the bridge needs typed `MessageV2.WithParts`, not the AI SDK `messages` array), the session has zero tools (Phase 2 will lift this), and the bridge routes the model to a protocol in `NATIVE_PROTOCOLS`. - `NATIVE_PROTOCOLS` is a single-entry set today: `anthropic-messages`. Other adapters are imported and registered with the client so the Phase 2 expansion is a one-line edit, not an architecture change. - Stream wiring: client.stream(req) -> Stream.flatMap(event -> fromIterable(map.map(event))) -> Stream.concat(suspended fromIterable(map.flush())) -> Stream.provideService( RequestExecutor.Service, executor). The flush stream is built lazily with `Stream.unwrap(Effect.sync(...))` so it observes the mapper final state after every upstream event has been mapped. - The mapper (`LLMNativeEvents.mapper`) emits AI-SDK-shaped session events from `LLMEvent` so downstream consumers see one shape. What this does NOT do (deferred to later phases): - No tool support on the native path (skipped, falls through). - No parity harness yet; Phase 2 builds it. - No production traffic; flag is off by default and no production caller populates `nativeMessages`. - No reasoning/cache/multi-modal coverage. Anthropic supports reasoning and cache via existing patches, so those start working as soon as a caller routes a real session through. Verification: opencode typecheck clean, bridge tests still green (33/0/0 across llm-native.test.ts + llm-bridge.test.ts); LLM package tests green (123/0/0).
2026-04-27 13:50:24 -04:00
parent 0ba8ca63b6
commit fc3a1bfd34
2 changed files with 103 additions and 1 deletions
@@ -5,6 +5,18 @@ import * as Stream from "effect/Stream"
 import { streamText, wrapLanguageModel, type ModelMessage, type Tool, tool, jsonSchema } from "ai"
 import { mergeDeep } from "remeda"
 import { GitLabWorkflowLanguageModel } from "gitlab-ai-provider"
+import {
+  AnthropicMessages,
+  BedrockConverse,
+  Gemini,
+  LLMClient,
+  OpenAIChat,
+  OpenAICompatibleChat,
+  OpenAIResponses,
+  ProviderPatch,
+  RequestExecutor,
+  type Protocol,
+} from "@opencode-ai/llm"
 import { ProviderTransform } from "@/provider/transform"
 import { Config } from "@/config/config"
 import { InstanceState } from "@/effect/instance-state"
@@ -23,6 +35,8 @@ 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 { LLMNative } from "./llm-native"
+import { LLMNativeEvents } from "./llm-native-events"

 const log = Log.create({ service: "llm" })
 export const OUTPUT_TOKEN_MAX = ProviderTransform.OUTPUT_TOKEN_MAX
@@ -63,7 +77,12 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/LL
 const live: Layer.Layer<
  Service,
  never,
-  Auth.Service | Config.Service | Provider.Service | Plugin.Service | Permission.Service
+  | Auth.Service
+  | Config.Service
+  | Provider.Service
+  | Plugin.Service
+  | Permission.Service
+  | RequestExecutor.Service
 > = Layer.effect(
  Service,
  Effect.gen(function* () {
@@ -72,6 +91,11 @@ const live: Layer.Layer<
    const provider = yield* Provider.Service
    const plugin = yield* Plugin.Service
    const perm = yield* Permission.Service
+    // Required by the LLM-native stream path. The default layer wires it on
+    // top of `FetchHttpClient.layer`. Yielded here (not inside `runNative`)
+    // so the executor instance is shared across every native stream the
+    // service hands out.
+    const executor = yield* RequestExecutor.Service

    const run = Effect.fn("LLM.run")(function* (input: StreamRequest) {
      const l = log
@@ -420,6 +444,73 @@ const live: Layer.Layer<
      })
    })

+    // ----- Phase 1: LLM-native opt-in path -----
+    //
+    // `runNative` returns the session-shaped Stream when (and only when) the
+    // request matches a narrow opt-in profile we've actively wired:
+    //
+    //   - The flag `OPENCODE_EXPERIMENTAL_LLM_NATIVE` is set.
+    //   - The caller populated `input.nativeMessages` with `MessageV2.WithParts`
+    //     (the AI SDK `messages` array isn't enough — the LLM-native bridge
+    //     needs the typed parts).
+    //   - The bridge can route the model to one of the protocols listed in
+    //     `NATIVE_PROTOCOLS` (today: Anthropic only).
+    //   - The session has no tools (Phase 2 will lift this).
+    //
+    // Otherwise it returns `undefined` and the caller falls through to the
+    // existing AI SDK path. The return shape is deliberately narrow — we are
+    // not yet committed to native-by-default for any provider.
+    const NATIVE_PROTOCOLS = new Set<Protocol>(["anthropic-messages"])
+    const NATIVE_ADAPTERS = [
+      AnthropicMessages.adapter,
+      OpenAIChat.adapter,
+      OpenAIResponses.adapter,
+      Gemini.adapter,
+      OpenAICompatibleChat.adapter,
+      BedrockConverse.adapter,
+    ]
+
+    const nativeClient = LLMClient.make({
+      adapters: NATIVE_ADAPTERS,
+      patches: ProviderPatch.defaults,
+    })
+
+    const runNative = Effect.fn("LLM.runNative")(function* (input: StreamRequest) {
+      if (!Flag.OPENCODE_EXPERIMENTAL_LLM_NATIVE) return undefined
+      if (!input.nativeMessages || input.nativeMessages.length === 0) return undefined
+      if (Object.keys(input.tools).length > 0) return undefined
+
+      const item = yield* provider.getProvider(input.model.providerID)
+      const llmRequest = yield* LLMNative.request({
+        id: input.user.id,
+        provider: item,
+        model: input.model,
+        system: input.system,
+        messages: input.nativeMessages,
+      })
+      if (!NATIVE_PROTOCOLS.has(llmRequest.model.protocol)) return undefined
+
+      log.info("native stream", {
+        sessionID: input.sessionID,
+        modelID: input.model.id,
+        providerID: input.model.providerID,
+        protocol: llmRequest.model.protocol,
+      })
+
+      // Stateful LLMEvent → SessionEvent translator. `map.map(event)` is called
+      // per-element, `map.flush()` emits the remaining `*-end` events for any
+      // text/reasoning/tool-input parts left open at stream close. The flush
+      // stream is built lazily (`Stream.unwrap(Effect.sync(...))`) so it
+      // observes the mapper's final state after `mapConcat` has consumed every
+      // upstream event.
+      const map = LLMNativeEvents.mapper()
+      return nativeClient.stream(llmRequest).pipe(
+        Stream.flatMap((event) => Stream.fromIterable(map.map(event))),
+        Stream.concat(Stream.unwrap(Effect.sync(() => Stream.fromIterable(map.flush())))),
+        Stream.provideService(RequestExecutor.Service, executor),
+      )
+    })
+
    const stream: Interface["stream"] = (input) =>
      Stream.scoped(
        Stream.unwrap(
@@ -429,6 +520,9 @@ const live: Layer.Layer<
              (ctrl) => Effect.sync(() => ctrl.abort()),
            )

+            const native = yield* runNative({ ...input, abort: ctrl.signal })
+            if (native) return native
+
            const result = yield* run({ ...input, abort: ctrl.signal })

            return Stream.fromAsyncIterable(result.fullStream, (e) => (e instanceof Error ? e : new Error(String(e))))
@@ -448,6 +542,7 @@ export const defaultLayer = Layer.suspend(() =>
    Layer.provide(Config.defaultLayer),
    Layer.provide(Provider.defaultLayer),
    Layer.provide(Plugin.defaultLayer),
+    Layer.provide(RequestExecutor.defaultLayer),
  ),
 )