mirror of
https://gitee.com/zhangsken/opencode.git
synced 2026-10-16 10:30:01 +08:00
fix(core): run session HTTP hooks on the AI SDK route (#50487)
This commit is contained in:
+68
-10
@@ -30,10 +30,12 @@ import {
|
||||
type ToolDefinition,
|
||||
type UsageInput,
|
||||
} from "@opencode/ai"
|
||||
import { Auth, Endpoint, RequestExecutor, type AnyRoute } from "@opencode/ai/route"
|
||||
import { Auth, Endpoint, RequestExecutor, type AnyRoute, type HttpMiddleware } from "@opencode/ai/route"
|
||||
import { ProviderShared } from "@opencode/ai/protocols/shared"
|
||||
import { Cause, Context, Effect, Layer, Option, Schema, Scope, Stream } from "effect"
|
||||
import { makeParser } from "effect/unstable/encoding/Sse"
|
||||
import { HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
|
||||
import { AsyncLocalStorage } from "node:async_hooks"
|
||||
import type { ID, RuntimeInfo } from "./model.js"
|
||||
import { Provider } from "./provider.js"
|
||||
import { State } from "./state.js"
|
||||
@@ -151,10 +153,11 @@ function prepareOptions(model: RuntimeInfo, pkg: string) {
|
||||
}
|
||||
}
|
||||
|
||||
const res = await (typeof customFetch === "function" ? customFetch : fetch)(input, {
|
||||
...opts,
|
||||
timeout: false,
|
||||
})
|
||||
const send: Fetch = typeof customFetch === "function" ? customFetch : fetch
|
||||
const middleware = httpMiddleware.getStore()
|
||||
const res = middleware
|
||||
? await throughMiddleware(middleware, send, input, { ...opts, timeout: false })
|
||||
: await send(input, { ...opts, timeout: false })
|
||||
if (!chunkAbortCtl || typeof chunkTimeout !== "number") return res
|
||||
return wrapSSE(res, chunkTimeout, chunkAbortCtl)
|
||||
}
|
||||
@@ -162,6 +165,56 @@ function prepareOptions(model: RuntimeInfo, pkg: string) {
|
||||
return options
|
||||
}
|
||||
|
||||
type Fetch = (input: Parameters<typeof fetch>[0], init?: BunFetchRequestInit) => Promise<Response>
|
||||
|
||||
// HTTP hook middleware is scoped to one model request, but the SDK's fetch is baked into the
|
||||
// cached language model, so the active middleware rides along in async context instead.
|
||||
const httpMiddleware = new AsyncLocalStorage<{ http: HttpMiddleware; context: Context.Context<never> }>()
|
||||
|
||||
function throughMiddleware(
|
||||
store: { http: HttpMiddleware; context: Context.Context<never> },
|
||||
send: Fetch,
|
||||
input: Parameters<typeof fetch>[0],
|
||||
init: BunFetchRequestInit,
|
||||
) {
|
||||
const toError = (cause: unknown) => (cause instanceof Error ? cause : new Error(String(cause)))
|
||||
const request = input instanceof Request ? new Request(input, init) : new Request(String(input), init)
|
||||
return Effect.runPromiseWith(store.context)(
|
||||
Effect.gen(function* () {
|
||||
// Hooks see a byte body like on the native route, so they may convert it to a web Request
|
||||
// as many times as they like without contending for one stream.
|
||||
const body = request.body ? new Uint8Array(yield* Effect.promise(() => request.arrayBuffer())) : undefined
|
||||
const response = yield* store.http(
|
||||
body
|
||||
? HttpClientRequest.bodyUint8Array(
|
||||
HttpClientRequest.fromWeb(request),
|
||||
body,
|
||||
request.headers.get("content-type") ?? undefined,
|
||||
)
|
||||
: HttpClientRequest.fromWeb(request),
|
||||
(sent) =>
|
||||
Effect.gen(function* () {
|
||||
const web = yield* HttpClientRequest.toWeb(sent)
|
||||
const response = yield* Effect.tryPromise(async () =>
|
||||
send(web.url, {
|
||||
...init,
|
||||
method: web.method,
|
||||
headers: web.headers,
|
||||
body: web.body ? await web.arrayBuffer() : undefined,
|
||||
}),
|
||||
)
|
||||
return HttpClientResponse.fromWeb(sent, response)
|
||||
}).pipe(Effect.mapError(toError)),
|
||||
)
|
||||
const stream = [204, 205, 304].includes(response.status)
|
||||
? null
|
||||
: yield* Stream.toReadableStreamEffect(response.stream)
|
||||
return new Response(stream, { status: response.status, headers: response.headers })
|
||||
}),
|
||||
{ signal: init.signal ?? undefined },
|
||||
)
|
||||
}
|
||||
|
||||
export class InitError extends Schema.TaggedError<InitError>()("AISDK.InitError", {
|
||||
providerID: Provider.ID,
|
||||
cause: Schema.Defect(),
|
||||
@@ -344,7 +397,8 @@ function modelFromLanguage(info: RuntimeInfo, language: LanguageModelV3) {
|
||||
model: (input) =>
|
||||
LanguageModel.make({ ...input, provider: "provider" in input ? input.provider : providerID, route }),
|
||||
prepareTransport: (body) => Effect.succeed(body),
|
||||
streamPrepared: (prepared) => streamLanguage(language, prepared as LanguageModelV3CallOptions),
|
||||
streamPrepared: (prepared, _request, _runtime, options) =>
|
||||
streamLanguage(language, prepared as LanguageModelV3CallOptions, options?.http),
|
||||
}
|
||||
return LanguageModel.make({
|
||||
id: info.modelID ?? info.id,
|
||||
@@ -642,14 +696,18 @@ function metadataProviderOptions(input: ProviderMetadata | undefined): SharedV3P
|
||||
return Object.fromEntries(Object.entries(input).map(([key, value]) => [key, jsonObject(value)]))
|
||||
}
|
||||
|
||||
function streamLanguage(language: LanguageModelV3, options: LanguageModelV3CallOptions) {
|
||||
function streamLanguage(language: LanguageModelV3, options: LanguageModelV3CallOptions, http?: HttpMiddleware) {
|
||||
const state = { step: 0, toolNames: {} as Record<string, string> }
|
||||
return Stream.concat(
|
||||
Stream.make(LLMEvent.stepStart({ index: state.step })),
|
||||
Stream.unwrap(
|
||||
Effect.tryPromise({
|
||||
try: () => language.doStream(options),
|
||||
catch: (error) => llmError(error, "request"),
|
||||
Effect.gen(function* () {
|
||||
const context = yield* Effect.context<never>()
|
||||
return yield* Effect.tryPromise({
|
||||
try: () =>
|
||||
http ? httpMiddleware.run({ http, context }, () => language.doStream(options)) : language.doStream(options),
|
||||
catch: (error) => llmError(error, "request"),
|
||||
})
|
||||
}).pipe(
|
||||
Effect.map((result) =>
|
||||
Stream.fromReadableStream({
|
||||
|
||||
@@ -258,7 +258,6 @@ export const GithubCopilotPlugin = define({
|
||||
evt.sdk = mod.createOpenaiCompatible(evt.options)
|
||||
}),
|
||||
)
|
||||
// Runs for every route, unlike http.request, which the AI SDK route bypasses.
|
||||
yield* ctx.session.hook(
|
||||
"model.request",
|
||||
(evt) =>
|
||||
|
||||
@@ -23,6 +23,7 @@ import { LLMClient, RequestExecutor } from "@opencode/ai/route"
|
||||
import { compileRequest } from "@opencode/ai/route/client"
|
||||
import { expect } from "bun:test"
|
||||
import { Effect, Layer } from "effect"
|
||||
import { HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
|
||||
import { testEffect } from "./lib/effect"
|
||||
|
||||
const it = testEffect(AISDK.locationLayer)
|
||||
@@ -596,6 +597,100 @@ it.effect("does not treat SSE comment heartbeats as model progress", () =>
|
||||
}),
|
||||
)
|
||||
|
||||
const chatChunk = (text: string) =>
|
||||
`data: ${JSON.stringify({
|
||||
id: "response-1",
|
||||
object: "chat.completion.chunk",
|
||||
created: 0,
|
||||
model: "api-model",
|
||||
choices: [{ index: 0, delta: { content: text }, finish_reason: "stop" }],
|
||||
})}\n\ndata: [DONE]\n\n`
|
||||
|
||||
const compatibleModel = Effect.fn(function* (customFetch: typeof fetch) {
|
||||
const aisdk = yield* AISDK.Service
|
||||
yield* aisdk.hook.sdk((event) => {
|
||||
event.sdk = createOpenAICompatible({
|
||||
...event.options,
|
||||
name: String(event.options.name),
|
||||
baseURL: String(event.options.baseURL),
|
||||
})
|
||||
})
|
||||
return yield* aisdk.model(
|
||||
model("@ai-sdk/openai-compatible", { apiKey: "test", baseURL: "https://example.test/v1", fetch: customFetch }),
|
||||
)
|
||||
})
|
||||
|
||||
it.effect("routes AI SDK requests and responses through HTTP hook middleware", () =>
|
||||
Effect.gen(function* () {
|
||||
const sent: Array<{ url: string; headers: Headers; body: string }> = []
|
||||
const resolved = yield* compatibleModel(
|
||||
Object.assign(
|
||||
async (input: Parameters<typeof fetch>[0], init?: RequestInit) => {
|
||||
sent.push({
|
||||
url: String(input),
|
||||
headers: new Headers(init?.headers),
|
||||
body: new TextDecoder().decode(init?.body as ArrayBuffer),
|
||||
})
|
||||
return new Response(chatChunk("upstream"), { headers: { "content-type": "text/event-stream" } })
|
||||
},
|
||||
{ preconnect: fetch.preconnect },
|
||||
),
|
||||
)
|
||||
const seen: string[] = []
|
||||
const response = yield* LLMClient.generate(LLM.request({ model: resolved, prompt: "Hello" }), {
|
||||
http: (request, handler) =>
|
||||
Effect.gen(function* () {
|
||||
// Read the body twice the way session hooks do, to prove it is not a single-use stream.
|
||||
const first = yield* HttpClientRequest.toWeb(request)
|
||||
const second = yield* HttpClientRequest.toWeb(request)
|
||||
seen.push(`${request.method} ${request.url}`)
|
||||
seen.push(yield* Effect.promise(() => first.text()))
|
||||
seen.push(yield* Effect.promise(() => second.text()))
|
||||
const upstream = yield* handler(HttpClientRequest.setHeader(request, "x-hook", "applied"))
|
||||
seen.push(`status ${upstream.status}`)
|
||||
return HttpClientResponse.fromWeb(
|
||||
upstream.request,
|
||||
new Response(chatChunk("rewritten"), { headers: { "content-type": "text/event-stream" } }),
|
||||
)
|
||||
}),
|
||||
}).pipe(Effect.provide(client))
|
||||
|
||||
expect(sent).toHaveLength(1)
|
||||
expect(sent[0]?.url).toBe("https://example.test/v1/chat/completions")
|
||||
expect(sent[0]?.headers.get("x-hook")).toBe("applied")
|
||||
expect(sent[0]?.headers.get("authorization")).toBe("Bearer test")
|
||||
expect(JSON.parse(sent[0]?.body ?? "")).toMatchObject({ model: "api-model" })
|
||||
expect(seen).toEqual([
|
||||
"POST https://example.test/v1/chat/completions",
|
||||
sent[0]?.body,
|
||||
sent[0]?.body,
|
||||
"status 200",
|
||||
])
|
||||
expect(response.events.filter(LLMEvent.is.textDelta).map((event) => event.text)).toEqual(["rewritten"])
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("sends AI SDK requests directly when no HTTP hook middleware is attached", () =>
|
||||
Effect.gen(function* () {
|
||||
const bodies: unknown[] = []
|
||||
const resolved = yield* compatibleModel(
|
||||
Object.assign(
|
||||
async (_input: Parameters<typeof fetch>[0], init?: RequestInit) => {
|
||||
bodies.push(init?.body)
|
||||
return new Response(chatChunk("upstream"), { headers: { "content-type": "text/event-stream" } })
|
||||
},
|
||||
{ preconnect: fetch.preconnect },
|
||||
),
|
||||
)
|
||||
const response = yield* LLMClient.generate(LLM.request({ model: resolved, prompt: "Hello" })).pipe(
|
||||
Effect.provide(client),
|
||||
)
|
||||
expect(bodies).toHaveLength(1)
|
||||
expect(typeof bodies[0]).toBe("string")
|
||||
expect(response.events.filter(LLMEvent.is.textDelta).map((event) => event.text)).toEqual(["upstream"])
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("emits malformed AI SDK tool input without executing it", () =>
|
||||
Effect.gen(function* () {
|
||||
const aisdk = yield* AISDK.Service
|
||||
|
||||
Reference in New Issue
Block a user