Mirror_Repos/anomalyco_opencode
2
0
Fork 0
Code Issues Pull Requests Actions 65 Packages Projects Releases Wiki Activity

refactor(llm): tighten runtime service boundaries

Browse Source
This commit is contained in:
Kit Langton
2026-05-06 11:47:23 -04:00
parent 7b4f436fc2
commit 9fc1d154c4
42 changed files with 851 additions and 496 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/opencode/src/session/llm-native-tools.ts
+8 -7
View File
@@ -1,13 +1,12 @@
import {
LLM,
LLMClient,
type LLMError,
type LLMEvent,
type LLMRequest,
type FinishReason,
type ContentPart,
type LLMClientShape,
} from "@opencode-ai/llm"
import type { RequestExecutor } from "@opencode-ai/llm/adapter"
import { Cause, Deferred, Effect, FiberSet, Queue, Stream, type Scope } from "effect"
import type { Tool, ToolExecutionOptions } from "ai"
@@ -129,6 +128,7 @@ const dispatchTool = (
// `done` resolves with the accumulated state so the multi-round driver can
// decide whether to recurse.
const runOneRound = (
client: LLMClientShape,
request: LLMRequest,
tools: Record<string, Tool>,
abort: AbortSignal,
@@ -138,7 +138,7 @@ const runOneRound = (
readonly done: Deferred.Deferred<RoundState>
},
never,
Scope.Scope | RequestExecutor.Service
Scope.Scope
> =>
Effect.gen(function* () {
const queue = yield* Queue.unbounded<LLMEvent, LLMError | Cause.Done>()
@@ -148,7 +148,7 @@ const runOneRound = (
yield* Effect.forkScoped(
Effect.gen(function* () {
yield* LLMClient.stream(request).pipe(
yield* client.stream(request).pipe(
Stream.runForEach((event) =>
Effect.gen(function* () {
accumulate(state, event)
@@ -218,16 +218,17 @@ const continuationRequest = (request: LLMRequest, state: RoundState): LLMRequest
* interrupted (e.g. via the abort signal).
*/
export const runWithTools = (input: {
readonly client: LLMClientShape
readonly request: LLMRequest
readonly tools: Record<string, Tool>
readonly abort: AbortSignal
readonly maxSteps?: number
}): Stream.Stream<LLMEvent, LLMError, RequestExecutor.Service> => {
}): Stream.Stream<LLMEvent, LLMError> => {
const maxSteps = input.maxSteps ?? DEFAULT_MAX_STEPS
const round = (request: LLMRequest, step: number): Stream.Stream<LLMEvent, LLMError, RequestExecutor.Service> =>
const round = (request: LLMRequest, step: number): Stream.Stream<LLMEvent, LLMError> =>
Stream.unwrap(
Effect.gen(function* () {
const { events, done } = yield* runOneRound(request, input.tools, input.abort)
const { events, done } = yield* runOneRound(input.client, request, input.tools, input.abort)
const continuation = Stream.unwrap(
Effect.gen(function* () {
const state = yield* Deferred.await(done)
packages/opencode/src/session/llm.ts
+11 -13
View File
@@ -8,6 +8,7 @@ import { mergeDeep } from "remeda"
import { GitLabWorkflowLanguageModel } from "gitlab-ai-provider"
import {
LLMClient,
type LLMClientService,
type ProtocolID,
} from "@opencode-ai/llm"
import { RequestExecutor } from "@opencode-ai/llm/adapter"
@@ -103,7 +104,7 @@ const live: Layer.Layer<
| Provider.Service
| Plugin.Service
| Permission.Service
| RequestExecutor.Service
| LLMClientService
> = Layer.effect(
Service,
Effect.gen(function* () {
@@ -112,11 +113,7 @@ 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 llmClient = yield* LLMClient.Service
const prepare = Effect.fn("LLM.prepareStream")(function* (input: StreamRequest) {
const [language, cfg, item, info] = yield* Effect.all(
@@ -581,14 +578,14 @@ const live: Layer.Layer<
const upstream = filteredNativeTools && filteredNativeTools.length > 0
? LLMNativeTools.runWithTools({
request: llmRequest,
client: llmClient,
tools: filteredAITools,
abort: input.abort,
})
: LLMClient.stream(llmRequest)
: llmClient.stream(llmRequest)
return upstream.pipe(
Stream.flatMap((event) => Stream.fromIterable(map.map(event))),
Stream.concat(Stream.unwrap(Effect.sync(() => Stream.fromIterable(map.flush())))),
Stream.provideService(RequestExecutor.Service, executor),
)
})
@@ -620,15 +617,16 @@ const live: Layer.Layer<
export const layer = live.pipe(Layer.provide(Permission.defaultLayer))
export const defaultLayer = Layer.suspend(() =>
layer.pipe(
export const defaultLayer = Layer.suspend(() => {
const llmClientLayer = LLMClient.layer.pipe(Layer.provide(RequestExecutor.defaultLayer))
return layer.pipe(
Layer.provide(Auth.defaultLayer),
Layer.provide(Config.defaultLayer),
Layer.provide(Provider.defaultLayer),
Layer.provide(Plugin.defaultLayer),
Layer.provide(RequestExecutor.defaultLayer),
),
)
Layer.provide(llmClientLayer),
)
})
function resolveTools(input: Pick<StreamInput, "tools" | "agent" | "permission" | "user">) {
const disabled = Permission.disabled(
packages/opencode/test/session/llm-native-stream.test.ts
+23 -15
View File
@@ -11,7 +11,7 @@ import { LLMNative } from "../../src/session/llm-native"
import { LLMNativeEvents } from "../../src/session/llm-native-events"
import { LLMNativeTools } from "../../src/session/llm-native-tools"
import { ProviderTest } from "../fake/provider"
import { testEffect } from "../lib/effect"
import { it } from "../lib/effect"
import type { MessageV2 } from "../../src/session/message-v2"
import type { Provider } from "../../src/provider/provider"
import type { Tool } from "../../src/tool/tool"
@@ -19,8 +19,8 @@ import type { Tool } from "../../src/tool/tool"
// Inline HTTP layer that returns a single fixed body. Mirrors the
// `fixedResponse` helper in `packages/llm/test/lib/http.ts` — duplicated here
// rather than imported across packages so this test stays self-contained.
const fixedResponse = (body: BodyInit, init: ResponseInit = { headers: { "content-type": "text/event-stream" } }) =>
RequestExecutor.layer.pipe(
const fixedResponse = (body: BodyInit, init: ResponseInit = { headers: { "content-type": "text/event-stream" } }) => {
const requestExecutorLayer = RequestExecutor.layer.pipe(
Layer.provide(
Layer.succeed(
HttpClient.HttpClient,
@@ -30,12 +30,14 @@ const fixedResponse = (body: BodyInit, init: ResponseInit = { headers: { "conten
),
),
)
return Layer.merge(requestExecutorLayer, LLMClient.layer.pipe(Layer.provide(requestExecutorLayer)))
}
// Scripted multi-response HTTP layer. Each request consumes the next body in
// order; the final body repeats if more requests arrive. Mirrors the
// `scriptedResponses` helper in `packages/llm/test/lib/http.ts`.
const scriptedResponses = (bodies: ReadonlyArray<BodyInit>, init: ResponseInit = { headers: { "content-type": "text/event-stream" } }) =>
RequestExecutor.layer.pipe(
const scriptedResponses = (bodies: ReadonlyArray<BodyInit>, init: ResponseInit = { headers: { "content-type": "text/event-stream" } }) => {
const requestExecutorLayer = RequestExecutor.layer.pipe(
Layer.provide(
Layer.unwrap(
Effect.gen(function* () {
@@ -54,6 +56,8 @@ const scriptedResponses = (bodies: ReadonlyArray<BodyInit>, init: ResponseInit =
),
),
)
return Layer.merge(requestExecutorLayer, LLMClient.layer.pipe(Layer.provide(requestExecutorLayer)))
}
// Encode an Anthropic SSE body. Each event becomes a `data:` line; the codec
// also expects `event:` lines but the package's SSE framing only reads the
@@ -91,8 +95,6 @@ const userMessage = (mdl: Provider.Model, id: MessageID, parts: MessageV2.Part[]
parts,
})
const it = testEffect(Layer.empty)
describe("LLMNative stream wire-up (audit gap #4 phase 1)", () => {
it.effect("converts an Anthropic SSE response into session events via the LLMNative path", () =>
Effect.gen(function* () {
@@ -120,7 +122,9 @@ describe("LLMNative stream wire-up (audit gap #4 phase 1)", () => {
{ type: "message_stop" },
])
const events = yield* LLMClient.stream(llmRequest).pipe(
const events = yield* Stream.unwrap(Effect.gen(function* () {
return (yield* LLMClient.Service).stream(llmRequest)
})).pipe(
Stream.flatMap((event) => Stream.fromIterable(map.map(event))),
Stream.concat(Stream.unwrap(Effect.sync(() => Stream.fromIterable(map.flush())))),
Stream.runCollect,
@@ -226,12 +230,14 @@ describe("LLMNative stream wire-up (audit gap #4 phase 1)", () => {
])
const map = LLMNativeEvents.mapper()
const events = yield* LLMNativeTools.runWithTools({
request: llmRequest,
tools: { lookup: aiTool },
abort: new AbortController().signal,
}).pipe(
const events = yield* Stream.unwrap(Effect.gen(function* () {
return LLMNativeTools.runWithTools({
client: yield* LLMClient.Service,
request: llmRequest,
tools: { lookup: aiTool },
abort: new AbortController().signal,
})
})).pipe(
Stream.flatMap((event) => Stream.fromIterable(map.map(event))),
Stream.concat(Stream.unwrap(Effect.sync(() => Stream.fromIterable(map.flush())))),
Stream.runCollect,
@@ -300,7 +306,9 @@ describe("LLMNative stream wire-up (audit gap #4 phase 1)", () => {
tools: [lookupTool],
})
const prepared = yield* LLMClient.prepare(llmRequest)
const prepared = yield* Effect.gen(function* () {
return yield* (yield* LLMClient.Service).prepare(llmRequest)
}).pipe(Effect.provide(LLMClient.layer.pipe(Layer.provide(RequestExecutor.defaultLayer))))
expect(prepared.payload).toMatchObject({
tools: [
{
packages/opencode/test/session/llm-native.test.ts
+18 -12
View File
@@ -1,7 +1,8 @@
import { describe, expect } from "bun:test"
import { LLMClient } from "@opencode-ai/llm"
import { LLMClient, type LLMRequest } from "@opencode-ai/llm"
import { RequestExecutor } from "@opencode-ai/llm/adapter"
import "@opencode-ai/llm/protocols"
import { Cause, Effect, Exit, Layer, Schema } from "effect"
import { Cause, Effect, Layer, Exit, Schema } from "effect"
import { ModelID, ProviderID } from "../../src/provider/schema"
import { LLMNative } from "../../src/session/llm-native"
import { MessageID, PartID, SessionID } from "../../src/session/schema"
@@ -113,7 +114,12 @@ const lookupTool = {
execute: () => Effect.succeed({ title: "", metadata: {}, output: "" }),
} satisfies Tool.Def<typeof lookupParameters>
const it = testEffect(Layer.empty)
const prepare = (request: LLMRequest) =>
Effect.gen(function* () {
return yield* (yield* LLMClient.Service).prepare(request)
})
const it = testEffect(LLMClient.layer.pipe(Layer.provide(RequestExecutor.defaultLayer)))
const isRecord = (value: unknown): value is Record<string, unknown> =>
typeof value === "object" && value !== null && !Array.isArray(value)
@@ -598,7 +604,7 @@ describe("LLMNative.request", () => {
tools: [lookupTool],
toolChoice: "lookup",
})
const prepared = yield* LLMClient.prepare(request)
const prepared = yield* prepare(request)
expect(prepared.payload).toMatchObject({
model: "gpt-5",
@@ -657,7 +663,7 @@ describe("LLMNative.request", () => {
tools: [lookupTool],
toolChoice: "lookup",
})
const prepared = yield* LLMClient.prepare(request)
const prepared = yield* prepare(request)
expect(request.model).toMatchObject({
provider: "anthropic",
@@ -726,7 +732,7 @@ describe("LLMNative.request", () => {
tools: [lookupTool],
toolChoice: "lookup",
})
const prepared = yield* LLMClient.prepare(request)
const prepared = yield* prepare(request)
expect(request.model).toMatchObject({
provider: "togetherai",
@@ -857,7 +863,7 @@ describe("LLMNative.request", () => {
tools: [lookupTool],
toolChoice: "lookup",
})
const prepared = yield* LLMClient.prepare(request)
const prepared = yield* prepare(request)
expect(request.model).toMatchObject({
provider: "google",
@@ -929,7 +935,7 @@ describe("LLMNative.request", () => {
system: ["First", "Second", "Third"],
messages: [userMessage(mdl, userID, [textPart(userID, "hello")])],
})
const prepared = yield* LLMClient.prepare(request)
const prepared = yield* prepare(request)
expect(prepared.payload).toMatchObject({
system: [
@@ -951,7 +957,7 @@ describe("LLMNative.request", () => {
model: mdl,
messages: messageIds.map((id, index) => userMessage(mdl, id, [textPart(id, `m${index}`)])),
})
const prepared = yield* LLMClient.prepare(request)
const prepared = yield* prepare(request)
expect(prepared.payload).toMatchObject({
messages: [
@@ -975,7 +981,7 @@ describe("LLMNative.request", () => {
system: ["You are concise."],
messages: [userMessage(mdl, userID, [textPart(userID, "hello")])],
})
const prepared = yield* LLMClient.prepare(request)
const prepared = yield* prepare(request)
expect(prepared.payload).toMatchObject({
system: [{ text: "You are concise." }, { cachePoint: { type: "default" } }],
@@ -1000,7 +1006,7 @@ describe("LLMNative.request", () => {
system: ["A", "B", "C"],
messages: ids.map((id, index) => userMessage(mdl, id, [textPart(id, `m${index}`)])),
})
const prepared = yield* LLMClient.prepare(request)
const prepared = yield* prepare(request)
// The serialized OpenAI Responses payload has no cache concept; the
// assertion is that nothing in the payload carries a cache marker.
@@ -1076,7 +1082,7 @@ describe("LLMNative.request", () => {
]),
],
})
const prepared = yield* LLMClient.prepare(request)
const prepared = yield* prepare(request)
expect(prepared.payload).toMatchObject({
messages: [