feat(plugin): wrap session HTTP requests
This commit is contained in:
Aiden Cline
parent
003b22edda
commit
b1f86ee72b
15 changed files with 309 additions and 114 deletions
|
|
@ -5,7 +5,7 @@ import { Endpoint, type EndpointPatch } from "./endpoint"
|
|||
import { RequestExecutor } from "./executor"
|
||||
import { Framing } from "./framing"
|
||||
import { HttpTransport } from "./transport"
|
||||
import type { HttpRequestTransform, Transport, TransportRuntime } from "./transport"
|
||||
import type { HttpMiddleware, Transport, TransportRuntime } from "./transport"
|
||||
import { WebSocketExecutor } from "./transport"
|
||||
import type { Protocol } from "./protocol"
|
||||
import { applyCachePolicy } from "../cache-policy"
|
||||
|
|
@ -96,7 +96,10 @@ export interface RoutePatch<Body, Prepared> extends RouteDefaultsInput {
|
|||
|
||||
type RouteMappedModelInput = RouteModelInput | RouteRoutedModelInput
|
||||
|
||||
const makeRouteModel = <Options extends ProviderOptions = ProviderOptions>(route: AnyRoute, mapped: RouteMappedModelInput) => {
|
||||
const makeRouteModel = <Options extends ProviderOptions = ProviderOptions>(
|
||||
route: AnyRoute,
|
||||
mapped: RouteMappedModelInput,
|
||||
) => {
|
||||
const provider = route.provider ?? ("provider" in mapped ? mapped.provider : undefined)
|
||||
if (!provider) throw new Error(`Route.model(${route.id}) requires a provider`)
|
||||
if (!endpointBaseURL(route.endpoint))
|
||||
|
|
@ -150,7 +153,7 @@ export interface Interface {
|
|||
}
|
||||
|
||||
export interface StreamOptions {
|
||||
readonly transform?: HttpRequestTransform
|
||||
readonly http?: HttpMiddleware
|
||||
}
|
||||
|
||||
export interface StreamMethod {
|
||||
|
|
@ -302,7 +305,7 @@ function makeFromTransport<Body, Prepared, Frame, Event, State>(
|
|||
auth: routeInput.auth ?? Auth.none,
|
||||
encodeBody,
|
||||
headers: routeInput.headers,
|
||||
transform: options?.transform,
|
||||
middleware: options?.http,
|
||||
}),
|
||||
streamPrepared: (prepared: Prepared, request: LLMRequest, runtime: TransportRuntime) => {
|
||||
const route = `${request.model.provider}/${request.model.route.id}`
|
||||
|
|
|
|||
|
|
@ -23,4 +23,4 @@ export type { ApiKeyMode, AuthOverride, ProviderAuthOption } from "./auth-option
|
|||
export type { Definition as EndpointFn, EndpointInput } from "./endpoint"
|
||||
export type { Definition as FramingDef } from "./framing"
|
||||
export type { Protocol as ProtocolDef } from "./protocol"
|
||||
export type { HttpRequest, HttpRequestTransform, Transport as TransportDef, TransportRuntime } from "./transport"
|
||||
export type { HttpHandler, HttpMiddleware, Transport as TransportDef, TransportRuntime } from "./transport"
|
||||
|
|
|
|||
|
|
@ -1,11 +1,11 @@
|
|||
import { Effect, Stream } from "effect"
|
||||
import { Headers, HttpClientRequest } from "effect/unstable/http"
|
||||
import { Headers, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
|
||||
import { Auth } from "../auth"
|
||||
import { render as renderEndpoint } from "../endpoint"
|
||||
import { Framing } from "../framing"
|
||||
import type { Transport, TransportPrepareInput } from "./index"
|
||||
import type { HttpMiddleware, Transport, TransportPrepareInput } from "./index"
|
||||
import * as ProviderShared from "../../protocols/shared"
|
||||
import { mergeJsonRecords, type LLMRequest } from "../../schema"
|
||||
import { LLMError, mergeJsonRecords, type LLMRequest } from "../../schema"
|
||||
|
||||
export type JsonRequestInput<Body> = TransportPrepareInput<Body>
|
||||
|
||||
|
|
@ -18,7 +18,9 @@ export interface JsonRequestParts<Body = unknown> {
|
|||
|
||||
export interface HttpPrepared<Frame> {
|
||||
readonly request: HttpClientRequest.HttpClientRequest
|
||||
readonly web: Request
|
||||
readonly framing: Framing.Definition<Frame>
|
||||
readonly middleware?: HttpMiddleware
|
||||
}
|
||||
|
||||
const applyQuery = (url: string, query: Record<string, string> | undefined) => {
|
||||
|
|
@ -74,21 +76,62 @@ export const httpJson = <Body, Frame>(input: HttpJsonInput<Body, Frame>): HttpJs
|
|||
prepare: (prepareInput) =>
|
||||
Effect.gen(function* () {
|
||||
const parts = yield* jsonRequestParts({ ...prepareInput })
|
||||
const request = { url: parts.url, method: "POST", headers: { ...parts.headers }, body: parts.bodyText }
|
||||
yield* (prepareInput.transform?.(request) ?? Effect.void)
|
||||
const request = ProviderShared.jsonPost({
|
||||
url: parts.url,
|
||||
body: parts.bodyText,
|
||||
headers: parts.headers,
|
||||
})
|
||||
return {
|
||||
request: ProviderShared.jsonPost({
|
||||
url: request.url,
|
||||
body: request.body ?? "",
|
||||
headers: Headers.fromInput(request.headers),
|
||||
}),
|
||||
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(
|
||||
runtime.http
|
||||
.execute(prepared.request)
|
||||
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)),
|
||||
)
|
||||
.pipe(
|
||||
Effect.map((response) =>
|
||||
prepared.framing.frame(
|
||||
|
|
|
|||
|
|
@ -10,14 +10,8 @@ export interface TransportRuntime {
|
|||
readonly webSocket?: WebSocketExecutorInterface
|
||||
}
|
||||
|
||||
export interface HttpRequest {
|
||||
url: string
|
||||
readonly method: string
|
||||
headers: Record<string, string>
|
||||
body: string | undefined
|
||||
}
|
||||
|
||||
export type HttpRequestTransform = (request: HttpRequest) => Effect.Effect<void>
|
||||
export type HttpHandler = (request: Request) => Effect.Effect<Response, Error>
|
||||
export type HttpMiddleware = (request: Request, handler: HttpHandler) => Effect.Effect<Response, Error>
|
||||
|
||||
export interface Transport<Body, Prepared, Frame> {
|
||||
readonly id: string
|
||||
|
|
@ -36,7 +30,7 @@ export interface TransportPrepareInput<Body> {
|
|||
readonly auth: Auth.Definition
|
||||
readonly encodeBody: (body: Body) => string
|
||||
readonly headers?: (input: { readonly request: LLMRequest }) => Record<string, string>
|
||||
readonly transform?: HttpRequestTransform
|
||||
readonly middleware?: HttpMiddleware
|
||||
}
|
||||
|
||||
export * as HttpTransport from "./http"
|
||||
|
|
|
|||
|
|
@ -146,12 +146,18 @@ describe("request option precedence", () => {
|
|||
prompt: "Say hello.",
|
||||
}),
|
||||
{
|
||||
transform: (request) =>
|
||||
Effect.sync(() => {
|
||||
expect(request.headers.authorization).toBe("Bearer fresh-key")
|
||||
request.url = "https://proxy.test/v1/chat/completions"
|
||||
request.headers["x-plugin"] = "transformed"
|
||||
request.body = JSON.stringify({ transformed: true })
|
||||
http: (request, handler) =>
|
||||
Effect.gen(function* () {
|
||||
expect(request.headers.get("authorization")).toBe("Bearer fresh-key")
|
||||
const headers = new Headers(request.headers)
|
||||
headers.set("x-plugin", "transformed")
|
||||
return yield* handler(
|
||||
new Request("https://proxy.test/v1/chat/completions", {
|
||||
method: request.method,
|
||||
headers,
|
||||
body: JSON.stringify({ transformed: true }),
|
||||
}),
|
||||
)
|
||||
}),
|
||||
},
|
||||
).pipe(
|
||||
|
|
@ -171,6 +177,42 @@ describe("request option precedence", () => {
|
|||
),
|
||||
)
|
||||
|
||||
it.effect("transforms the HTTP response before protocol decoding", () =>
|
||||
Effect.gen(function* () {
|
||||
const response = yield* LLMClient.generate(
|
||||
LLM.request({
|
||||
model: OpenAIChat.route
|
||||
.with({ endpoint: { baseURL: "https://api.openai.test/v1/" }, auth: Auth.bearer("test") })
|
||||
.model({ id: "gpt-4o-mini" }),
|
||||
prompt: "Say hello.",
|
||||
}),
|
||||
{
|
||||
http: (request, handler) =>
|
||||
Effect.gen(function* () {
|
||||
const response = yield* handler(request)
|
||||
const body = yield* Effect.promise(() => response.text())
|
||||
return new Response(body.replace("network", "hooked"), {
|
||||
status: response.status,
|
||||
headers: response.headers,
|
||||
})
|
||||
}),
|
||||
},
|
||||
).pipe(
|
||||
Effect.provide(
|
||||
dynamicResponse((input) =>
|
||||
Effect.succeed(
|
||||
input.respond(sseEvents(deltaChunk({ content: "network" }, "stop")), {
|
||||
headers: { "content-type": "text/event-stream" },
|
||||
}),
|
||||
),
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
expect(response.text).toBe("hooked")
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("applies raw body overlays after protocol lowering", () =>
|
||||
LLMClient.generate(
|
||||
LLM.request({
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue