From 1eec3e640a4e74a711d3fdee3427d473fe2c50e8 Mon Sep 17 00:00:00 2001 From: Aiden Cline Date: Sun, 2 Aug 2026 11:44:53 -0500 Subject: [PATCH] fix(ai): expose raw HTTP responses to hooks --- packages/ai/src/route/executor.ts | 43 +++++++++++++-- packages/ai/src/route/transport/http.ts | 50 ++--------------- packages/ai/src/route/transport/index.ts | 6 +-- packages/ai/test/compile.test.ts | 48 ++++++++++++++++- packages/core/src/plugin/promise.ts | 5 +- packages/core/test/plugin/promise.test.ts | 66 ++++++++++++++++++----- 6 files changed, 145 insertions(+), 73 deletions(-) diff --git a/packages/ai/src/route/executor.ts b/packages/ai/src/route/executor.ts index ab97bdaba8..0a19fe95c5 100644 --- a/packages/ai/src/route/executor.ts +++ b/packages/ai/src/route/executor.ts @@ -1,4 +1,4 @@ -import { Cause, Context, Effect, Layer } from "effect" +import { Cause, Context, Effect, Layer, Stream } from "effect" import { FetchHttpClient, Headers, @@ -20,9 +20,13 @@ import { classifyProviderFailure } from "../provider-error" export interface Interface { readonly execute: ( request: HttpClientRequest.HttpClientRequest, + middleware?: HttpMiddleware, ) => Effect.Effect } +export type HttpHandler = (request: Request) => Effect.Effect +export type HttpMiddleware = (request: Request, handler: HttpHandler) => Effect.Effect + export class Service extends Context.Service()("@opencode/LLM/RequestExecutor") {} const BODY_LIMIT = 16_384 @@ -282,12 +286,41 @@ export const layer: Layer.Layer = Layer.e Service, Effect.gen(function* () { const http = yield* HttpClient.HttpClient - const executeOnce = (request: HttpClientRequest.HttpClientRequest) => + const executeOnce = (request: HttpClientRequest.HttpClientRequest, middleware?: HttpMiddleware) => Effect.gen(function* () { const redactedNames = yield* Headers.CurrentRedactedNames - return yield* http - .execute(request) - .pipe(Effect.mapError(toHttpError(redactedNames)), Effect.flatMap(statusError(request, redactedNames))) + if (!middleware) + return yield* http + .execute(request) + .pipe(Effect.mapError(toHttpError(redactedNames)), Effect.flatMap(statusError(request, redactedNames))) + + let sent = request + const origins = new WeakMap() + const response = yield* HttpClientRequest.toWeb(request).pipe( + Effect.flatMap((web) => + middleware(web, (input) => + Effect.gen(function* () { + sent = HttpClientRequest.fromWeb(input) + if (input.body) + sent = HttpClientRequest.bodyUint8Array( + sent, + new Uint8Array(yield* Effect.promise(() => input.arrayBuffer())), + input.headers.get("content-type") ?? undefined, + ) + const response = yield* http + .execute(sent) + .pipe(Effect.mapError((cause) => (cause instanceof Error ? cause : new Error(String(cause))))) + const body = yield* Stream.toReadableStreamEffect(response.stream) + const web = new Response(body, { status: response.status, headers: response.headers }) + origins.set(web, sent) + return web + }), + ), + ), + Effect.mapError(toHttpError(redactedNames)), + ) + const origin = origins.get(response) ?? sent + return yield* statusError(origin, redactedNames)(HttpClientResponse.fromWeb(origin, response)) }) return Service.of({ execute: executeOnce, diff --git a/packages/ai/src/route/transport/http.ts b/packages/ai/src/route/transport/http.ts index a1204c47f1..223afb353c 100644 --- a/packages/ai/src/route/transport/http.ts +++ b/packages/ai/src/route/transport/http.ts @@ -1,11 +1,11 @@ import { Effect, Stream } from "effect" -import { Headers, HttpClientRequest, HttpClientResponse } from "effect/unstable/http" +import { Headers, HttpClientRequest } from "effect/unstable/http" import { Auth } from "../auth" import { render as renderEndpoint } from "../endpoint" import { Framing } from "../framing" import type { HttpMiddleware, Transport, TransportPrepareInput } from "./index" import * as ProviderShared from "../../protocols/shared" -import { LLMError, mergeJsonRecords, type LLMRequest } from "../../schema" +import { mergeJsonRecords, type LLMRequest } from "../../schema" export type JsonRequestInput = TransportPrepareInput @@ -18,7 +18,6 @@ export interface JsonRequestParts { export interface HttpPrepared { readonly request: HttpClientRequest.HttpClientRequest - readonly web: Request readonly framing: Framing.Definition readonly middleware?: HttpMiddleware } @@ -83,55 +82,14 @@ export const httpJson = (input: HttpJsonInput): HttpJs }) return { request, - web: new Request(parts.url, { method: "POST", headers: parts.headers, body: parts.bodyText }), framing: input.framing, middleware: prepareInput.middleware, } }), frames: (prepared, request, runtime) => Stream.unwrap( - Effect.gen(function* () { - const request = prepared.web - const execute = (input: Request) => - Effect.tryPromise({ - try: () => input.text(), - catch: (cause) => (cause instanceof Error ? cause : new Error(String(cause))), - }).pipe( - Effect.flatMap((body) => - runtime.http.execute( - ProviderShared.jsonPost({ - url: input.url, - body, - headers: Headers.fromInput(input.headers), - }), - ), - ), - Effect.flatMap((response) => - Stream.toReadableStreamEffect(response.stream).pipe( - Effect.map( - (body) => - new Response(body, { - status: response.status, - headers: response.headers, - }), - ), - ), - ), - ) - return yield* prepared.middleware ? prepared.middleware(request, execute) : execute(request) - }) - .pipe( - Effect.mapError((error) => - error instanceof LLMError - ? error - : ProviderShared.eventError( - `${request.model.provider}/${request.model.route.id}`, - `Failed to execute ${request.model.provider}/${request.model.route.id} request`, - ProviderShared.errorText(error), - ), - ), - Effect.map((response) => HttpClientResponse.fromWeb(prepared.request, response)), - ) + runtime.http + .execute(prepared.request, prepared.middleware) .pipe( Effect.map((response) => prepared.framing.frame( diff --git a/packages/ai/src/route/transport/index.ts b/packages/ai/src/route/transport/index.ts index 275f64c06c..1e9e826bc9 100644 --- a/packages/ai/src/route/transport/index.ts +++ b/packages/ai/src/route/transport/index.ts @@ -1,7 +1,7 @@ import type { Effect, Stream } from "effect" import { Endpoint } from "../endpoint" import { Auth } from "../auth" -import type { Interface as RequestExecutorInterface } from "../executor" +import type { HttpMiddleware, Interface as RequestExecutorInterface } from "../executor" import type { Interface as WebSocketExecutorInterface } from "./websocket" import type { LLMError, LLMRequest } from "../../schema" @@ -10,9 +10,6 @@ export interface TransportRuntime { readonly webSocket?: WebSocketExecutorInterface } -export type HttpHandler = (request: Request) => Effect.Effect -export type HttpMiddleware = (request: Request, handler: HttpHandler) => Effect.Effect - export interface Transport { readonly id: string readonly prepare: (input: TransportPrepareInput) => Effect.Effect @@ -34,4 +31,5 @@ export interface TransportPrepareInput { } export * as HttpTransport from "./http" +export type { HttpHandler, HttpMiddleware } from "../executor" export { WebSocketExecutor, WebSocketTransport } from "./websocket" diff --git a/packages/ai/test/compile.test.ts b/packages/ai/test/compile.test.ts index 741b1f7dda..57ffbb3ddc 100644 --- a/packages/ai/test/compile.test.ts +++ b/packages/ai/test/compile.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "bun:test" -import { Effect, Schema } from "effect" +import { Effect, Ref, Schema } from "effect" import { HttpClientRequest } from "effect/unstable/http" import { LLM, mergeProviderOptions } from "../src" import { AnthropicMessages, OpenAIChat } from "../src/protocols" @@ -151,9 +151,10 @@ describe("request option precedence", () => { expect(request.headers.get("authorization")).toBe("Bearer fresh-key") const headers = new Headers(request.headers) headers.set("x-plugin", "transformed") + headers.set("content-type", "application/custom+json") return yield* handler( new Request("https://proxy.test/v1/chat/completions", { - method: request.method, + method: "PUT", headers, body: JSON.stringify({ transformed: true }), }), @@ -166,7 +167,9 @@ describe("request option precedence", () => { Effect.gen(function* () { const web = yield* HttpClientRequest.toWeb(input.request).pipe(Effect.orDie) expect(web.url).toBe("https://proxy.test/v1/chat/completions") + expect(web.method).toBe("PUT") expect(web.headers.get("x-plugin")).toBe("transformed") + expect(web.headers.get("content-type")).toBe("application/custom+json") expect(decodeJson(input.text)).toEqual({ transformed: true }) return input.respond(sseEvents(deltaChunk({}, "stop")), { headers: { "content-type": "text/event-stream" }, @@ -213,6 +216,47 @@ describe("request option precedence", () => { }), ) + it.effect("can inspect an error response and retry the native request", () => + Effect.gen(function* () { + const attempts = yield* Ref.make(0) + const response = yield* LLMClient.generate( + LLM.request({ + model: OpenAIChat.route + .with({ endpoint: { baseURL: "https://api.openai.test/v1/" }, auth: Auth.bearer("stale") }) + .model({ id: "gpt-4o-mini" }), + prompt: "Say hello.", + }), + { + http: (request, handler) => + Effect.gen(function* () { + const retry = request.clone() + const response = yield* handler(request) + expect(response.status).toBe(401) + const headers = new Headers(retry.headers) + headers.set("authorization", "Bearer refreshed") + return yield* handler(new Request(retry, { headers })) + }), + }, + ).pipe( + Effect.provide( + dynamicResponse((input) => + Effect.gen(function* () { + yield* Ref.update(attempts, (value) => value + 1) + if (input.request.headers.authorization !== "Bearer refreshed") + return input.respond("unauthorized", { status: 401 }) + return input.respond(sseEvents(deltaChunk({ content: "retried" }, "stop")), { + headers: { "content-type": "text/event-stream" }, + }) + }), + ), + ), + ) + + expect(response.text).toBe("retried") + expect(yield* Ref.get(attempts)).toBe(2) + }), + ) + it.effect("applies raw body overlays after protocol lowering", () => LLMClient.generate( LLM.request({ diff --git a/packages/core/src/plugin/promise.ts b/packages/core/src/plugin/promise.ts index 36d903b9ee..32e5342af2 100644 --- a/packages/core/src/plugin/promise.ts +++ b/packages/core/src/plugin/promise.ts @@ -277,14 +277,15 @@ export function fromPromise(plugin: Plugin) { const request = event.request const output = { ...event, - request: (input: Request) => Effect.runPromiseWith(context)(request(input)), + request: (input: Request) => + Effect.runPromiseWith(context)(request(input), { signal: input.signal }), } return Effect.promise(() => Promise.resolve(Reflect.apply(callback, undefined, [output]))).pipe( Effect.tap(() => Effect.sync(() => { event.request = (input) => Effect.tryPromise({ - try: () => output.request(input), + try: (signal) => output.request(new Request(input, { signal })), catch: (cause) => (cause instanceof Error ? cause : new Error(String(cause))), }) }), diff --git a/packages/core/test/plugin/promise.test.ts b/packages/core/test/plugin/promise.test.ts index a095630779..b0c01fe44f 100644 --- a/packages/core/test/plugin/promise.test.ts +++ b/packages/core/test/plugin/promise.test.ts @@ -1,6 +1,6 @@ import { describe, expect } from "bun:test" import { Message, SystemPart } from "@opencode-ai/ai" -import { DateTime, Effect, Schema } from "effect" +import { DateTime, Deferred, Effect, Fiber, Schema } from "effect" import { Agent } from "@opencode-ai/core/agent" import { Catalog } from "@opencode-ai/core/catalog" import { Model } from "@opencode-ai/core/model" @@ -148,7 +148,9 @@ describe("fromPromise", () => { expect((await ctx.agent.get({ agentID: Agent.ID.make("reviewer") })).data).toMatchObject({ description: "Reviews code", }) - await expect(ctx.agent.get({ agentID: Agent.ID.make("missing") })).rejects.toThrow("Agent not found: missing") + await expect(ctx.agent.get({ agentID: Agent.ID.make("missing") })).rejects.toThrow( + "Agent not found: missing", + ) const models = (await ctx.catalog.model.list()).data expect(models.find((model) => model.providerID === "test" && model.id === "alias")).toMatchObject({ modelID: "gpt-5", @@ -254,6 +256,44 @@ describe("fromPromise", () => { }), ) + it.effect("interrupts the Effect request through a promise session HTTP hook", () => + Effect.gen(function* () { + const plugin = yield* Plugin.Service + const hooks = yield* PluginHooks.Service + const host = yield* PluginHost.make(plugin) + yield* PluginPromise.fromPromise( + define({ + id: "promise-session-http-interrupt", + setup: async (ctx) => { + await ctx.session.hook("http", (event) => { + const request = event.request + event.request = (input) => request(input) + }) + }, + }), + ).effect(host) + const started = yield* Deferred.make() + const interrupted = yield* Deferred.make() + const event: SessionHooks["http"] = { + sessionID: Session.ID.make("ses_promise_session_http_interrupt"), + agent: Agent.ID.make("build"), + model: Model.Ref.make({ providerID: Provider.ID.make("test"), id: Model.ID.make("model") }), + request: () => + Deferred.succeed(started, undefined).pipe( + Effect.andThen(Effect.never), + Effect.onInterrupt(() => Deferred.succeed(interrupted, undefined)), + ), + } + + yield* hooks.trigger("session", "http", event) + const fiber = yield* event.request(new Request("https://provider.test")).pipe(Effect.forkChild) + yield* Deferred.await(started) + yield* Fiber.interrupt(fiber) + + expect(yield* Deferred.isDone(interrupted)).toBeTrue() + }), + ) + it.effect("disposes a hook registration on request", () => Effect.gen(function* () { const agents = yield* Agent.Service @@ -348,19 +388,17 @@ describe("fromPromise", () => { id: "promise-tool", setup: async (ctx) => { await ctx.tool.transform((tools) => { - tools.add( - { - name: "hello", - options: { codemode: false }, - description: "Hello", - input: Schema.Struct({ name: Schema.String }), - output: Schema.String, - execute: async ({ name }, context) => { - await context.progress({ phase: "greeting" }) - return { output: `Hello, ${name}!` } - }, + tools.add({ + name: "hello", + options: { codemode: false }, + description: "Hello", + input: Schema.Struct({ name: Schema.String }), + output: Schema.String, + execute: async ({ name }, context) => { + await context.progress({ phase: "greeting" }) + return { output: `Hello, ${name}!` } }, - ) + }) }) }, })