From fc3a1bfd34cba4d795288c4d951bd6e23900d1d2 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Mon, 27 Apr 2026 13:50:24 -0400 Subject: [PATCH] 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 | 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). --- packages/core/src/flag/flag.ts | 7 ++ packages/opencode/src/session/llm.ts | 97 +++++++++++++++++++++++++++- 2 files changed, 103 insertions(+), 1 deletion(-) diff --git a/packages/core/src/flag/flag.ts b/packages/core/src/flag/flag.ts index a3b8133b64..c190ec8f00 100644 --- a/packages/core/src/flag/flag.ts +++ b/packages/core/src/flag/flag.ts @@ -67,6 +67,13 @@ export const Flag = { OPENCODE_ENABLE_EXA: truthy("OPENCODE_ENABLE_EXA") || OPENCODE_EXPERIMENTAL || truthy("OPENCODE_EXPERIMENTAL_EXA"), OPENCODE_EXPERIMENTAL_BASH_DEFAULT_TIMEOUT_MS: number("OPENCODE_EXPERIMENTAL_BASH_DEFAULT_TIMEOUT_MS"), OPENCODE_EXPERIMENTAL_OUTPUT_TOKEN_MAX: number("OPENCODE_EXPERIMENTAL_OUTPUT_TOKEN_MAX"), + // Opt-in to the LLM-native stream path in `session/llm.ts`. Today this + // routes a narrow slice of sessions (text-only, Anthropic, with explicit + // `nativeMessages` populated by the caller) through the + // `@opencode-ai/llm` core stack instead of `streamText` from the AI SDK. + // Everything else falls through to the existing path. The flag will go + // away once parity is proven across all six protocols. + OPENCODE_EXPERIMENTAL_LLM_NATIVE: OPENCODE_EXPERIMENTAL || truthy("OPENCODE_EXPERIMENTAL_LLM_NATIVE"), OPENCODE_EXPERIMENTAL_OXFMT: OPENCODE_EXPERIMENTAL || truthy("OPENCODE_EXPERIMENTAL_OXFMT"), OPENCODE_EXPERIMENTAL_LSP_TY: truthy("OPENCODE_EXPERIMENTAL_LSP_TY"), OPENCODE_EXPERIMENTAL_LSP_TOOL: OPENCODE_EXPERIMENTAL || truthy("OPENCODE_EXPERIMENTAL_LSP_TOOL"), diff --git a/packages/opencode/src/session/llm.ts b/packages/opencode/src/session/llm.ts index 406fd3b608..8bb876e460 100644 --- a/packages/opencode/src/session/llm.ts +++ b/packages/opencode/src/session/llm.ts @@ -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()("@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(["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), ), )