From 60673aaef38f395811c9616fdde80e53b8660bd3 Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Mon, 21 Sep 2026 22:41:06 -0500 Subject: [PATCH] fix(core): run session HTTP hooks on the AI SDK route (#50487) --- packages/core/src/aisdk.ts | 78 +++++++++++++-- .../src/plugin/provider/github-copilot.ts | 1 - packages/core/test/aisdk.test.ts | 95 +++++++++++++++++++ 3 files changed, 163 insertions(+), 11 deletions(-) diff --git a/packages/core/src/aisdk.ts b/packages/core/src/aisdk.ts index 14206bfbc29..4da9ec1326b 100644 --- a/packages/core/src/aisdk.ts +++ b/packages/core/src/aisdk.ts @@ -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[0], init?: BunFetchRequestInit) => Promise + +// 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 }>() + +function throughMiddleware( + store: { http: HttpMiddleware; context: Context.Context }, + send: Fetch, + input: Parameters[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()("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 } 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() + 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({ diff --git a/packages/core/src/plugin/provider/github-copilot.ts b/packages/core/src/plugin/provider/github-copilot.ts index 5529d51d123..311f84fd241 100644 --- a/packages/core/src/plugin/provider/github-copilot.ts +++ b/packages/core/src/plugin/provider/github-copilot.ts @@ -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) => diff --git a/packages/core/test/aisdk.test.ts b/packages/core/test/aisdk.test.ts index 1dcb9fb9c6a..47c775febbb 100644 --- a/packages/core/test/aisdk.test.ts +++ b/packages/core/test/aisdk.test.ts @@ -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[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[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