feat(plugin): wire session request hook
This commit is contained in:
parent
518d64d684
commit
7edaa05869
17 changed files with 447 additions and 73 deletions
|
|
@ -8,6 +8,7 @@ import { HttpTransport } from "./transport"
|
|||
import type { Transport, TransportRuntime } from "./transport"
|
||||
import { WebSocketExecutor } from "./transport"
|
||||
import type { Protocol } from "./protocol"
|
||||
import type { CallOptions } from "./request-transform"
|
||||
import { applyCachePolicy } from "../cache-policy"
|
||||
import * as ProviderShared from "../protocols/shared"
|
||||
import type { LLMError, PreparedRequestOf, ProtocolID, ProviderOptions } from "../schema"
|
||||
|
|
@ -47,7 +48,11 @@ export interface Route<Body, Prepared = unknown> {
|
|||
readonly body: RouteBody<Body>
|
||||
readonly with: (patch: RoutePatch<Body, Prepared>) => Route<Body, Prepared>
|
||||
readonly model: (input: RouteMappedModelInput) => Model
|
||||
readonly prepareTransport: (body: Body, request: LLMRequest) => Effect.Effect<Prepared, LLMError>
|
||||
readonly prepareTransport: (
|
||||
body: Body,
|
||||
request: LLMRequest,
|
||||
options?: CallOptions,
|
||||
) => Effect.Effect<Prepared, LLMError>
|
||||
readonly streamPrepared: (
|
||||
prepared: Prepared,
|
||||
request: LLMRequest,
|
||||
|
|
@ -158,11 +163,11 @@ export interface Interface {
|
|||
}
|
||||
|
||||
export interface StreamMethod {
|
||||
(request: LLMRequest): Stream.Stream<LLMEvent, LLMError>
|
||||
(request: LLMRequest, options?: CallOptions): Stream.Stream<LLMEvent, LLMError>
|
||||
}
|
||||
|
||||
export interface GenerateMethod {
|
||||
(request: LLMRequest): Effect.Effect<LLMResponse, LLMError>
|
||||
(request: LLMRequest, options?: CallOptions): Effect.Effect<LLMResponse, LLMError>
|
||||
}
|
||||
|
||||
export class Service extends Context.Service<Service, Interface>()("@opencode/LLMClient") {}
|
||||
|
|
@ -297,7 +302,7 @@ function makeFromTransport<Body, Prepared, Frame, Event, State>(
|
|||
})
|
||||
},
|
||||
model: (input) => makeRouteModel(route, input),
|
||||
prepareTransport: (body, request) =>
|
||||
prepareTransport: (body, request, options) =>
|
||||
routeInput.transport.prepare({
|
||||
body,
|
||||
request,
|
||||
|
|
@ -305,6 +310,7 @@ function makeFromTransport<Body, Prepared, Frame, Event, State>(
|
|||
auth: routeInput.auth ?? Auth.none,
|
||||
encodeBody,
|
||||
headers: routeInput.headers,
|
||||
transformRequest: options?.transformRequest,
|
||||
}),
|
||||
streamPrepared: (prepared: Prepared, request: LLMRequest, runtime: TransportRuntime) => {
|
||||
const route = `${request.model.provider}/${request.model.route.id}`
|
||||
|
|
@ -373,14 +379,14 @@ export function make<Body, Prepared, Frame, Event, State>(
|
|||
// `compile` is the important boundary: it turns a common `LLMRequest` into a
|
||||
// validated provider body plus transport-private prepared data, but does not
|
||||
// execute transport.
|
||||
const compile = Effect.fn("LLM.compile")(function* (request: LLMRequest) {
|
||||
const compile = Effect.fn("LLM.compile")(function* (request: LLMRequest, options?: CallOptions) {
|
||||
const resolved = applyCachePolicy(resolveRequestOptions(request))
|
||||
const route = resolved.model.route
|
||||
|
||||
const body = yield* route.body
|
||||
.from(resolved)
|
||||
.pipe(Effect.flatMap(ProviderShared.validateWith(Schema.decodeUnknownEffect(route.body.schema))))
|
||||
const prepared = yield* route.prepareTransport(body, resolved)
|
||||
const prepared = yield* route.prepareTransport(body, resolved, options)
|
||||
|
||||
return {
|
||||
request: resolved,
|
||||
|
|
@ -403,17 +409,17 @@ const prepareWith = Effect.fn("LLMClient.prepare")(function* (request: LLMReques
|
|||
})
|
||||
})
|
||||
|
||||
const streamRequestWith = (runtime: TransportRuntime) => (request: LLMRequest) =>
|
||||
const streamRequestWith = (runtime: TransportRuntime) => (request: LLMRequest, options?: CallOptions) =>
|
||||
Stream.unwrap(
|
||||
Effect.gen(function* () {
|
||||
const compiled = yield* compile(request)
|
||||
const compiled = yield* compile(request, options)
|
||||
return compiled.route.streamPrepared(compiled.prepared, compiled.request, runtime)
|
||||
}),
|
||||
)
|
||||
|
||||
const generateWith = (stream: Interface["stream"]) =>
|
||||
Effect.fn("LLM.generate")(function* (request: LLMRequest) {
|
||||
const state = yield* stream(request).pipe(Stream.runFold(LLMResponse.empty, LLMResponse.reduce))
|
||||
Effect.fn("LLM.generate")(function* (request: LLMRequest, options?: CallOptions) {
|
||||
const state = yield* stream(request, options).pipe(Stream.runFold(LLMResponse.empty, LLMResponse.reduce))
|
||||
const response = LLMResponse.complete(state)
|
||||
if (response) return response
|
||||
return yield* ProviderShared.eventError(
|
||||
|
|
@ -425,17 +431,17 @@ const generateWith = (stream: Interface["stream"]) =>
|
|||
export const prepare = <Body = unknown>(request: LLMRequest) =>
|
||||
prepareWith(request) as Effect.Effect<PreparedRequestOf<Body>, LLMError>
|
||||
|
||||
export function stream(request: LLMRequest): Stream.Stream<LLMEvent, LLMError> {
|
||||
export function stream(request: LLMRequest, options?: CallOptions): Stream.Stream<LLMEvent, LLMError> {
|
||||
return Stream.unwrap(
|
||||
Effect.gen(function* () {
|
||||
return (yield* Service).stream(request)
|
||||
return (yield* Service).stream(request, options)
|
||||
}),
|
||||
) as Stream.Stream<LLMEvent, LLMError>
|
||||
}
|
||||
|
||||
export function generate(request: LLMRequest): Effect.Effect<LLMResponse, LLMError> {
|
||||
export function generate(request: LLMRequest, options?: CallOptions): Effect.Effect<LLMResponse, LLMError> {
|
||||
return Effect.gen(function* () {
|
||||
return yield* (yield* Service).generate(request)
|
||||
return yield* (yield* Service).generate(request, options)
|
||||
}) as Effect.Effect<LLMResponse, LLMError>
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ export type {
|
|||
Interface as LLMClientShape,
|
||||
Service as LLMClientService,
|
||||
} from "./client"
|
||||
export type { CallOptions, RequestData, RequestTransform, RequestValue } from "./request-transform"
|
||||
export * from "./executor"
|
||||
export { Auth } from "./auth"
|
||||
export { AuthOptions } from "./auth-options"
|
||||
|
|
|
|||
14
packages/ai/src/route/request-transform.ts
Normal file
14
packages/ai/src/route/request-transform.ts
Normal file
|
|
@ -0,0 +1,14 @@
|
|||
import type { Effect } from "effect"
|
||||
|
||||
export type RequestValue = null | boolean | number | string | RequestValue[] | { [key: string]: RequestValue }
|
||||
|
||||
export interface RequestData {
|
||||
readonly headers: Record<string, string>
|
||||
readonly body: Record<string, RequestValue>
|
||||
}
|
||||
|
||||
export type RequestTransform = (request: RequestData) => Effect.Effect<RequestData>
|
||||
|
||||
export interface CallOptions {
|
||||
readonly transformRequest?: RequestTransform
|
||||
}
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
import { Effect, Stream } from "effect"
|
||||
import { Effect, Schema, Stream } from "effect"
|
||||
import { Headers, HttpClientRequest } from "effect/unstable/http"
|
||||
import { Auth } from "../auth"
|
||||
import { render as renderEndpoint } from "../endpoint"
|
||||
|
|
@ -6,6 +6,7 @@ import { Framing } from "../framing"
|
|||
import type { Transport, TransportPrepareInput } from "./index"
|
||||
import * as ProviderShared from "../../protocols/shared"
|
||||
import { mergeJsonRecords, type LLMRequest } from "../../schema"
|
||||
import type { RequestValue } from "../request-transform"
|
||||
|
||||
export type JsonRequestInput<Body> = TransportPrepareInput<Body>
|
||||
|
||||
|
|
@ -86,24 +87,61 @@ const bodyWithOverlay = <Body>(body: Body, request: LLMRequest, encodeBody: (bod
|
|||
return yield* ProviderShared.invalidRequest("http.body can only overlay JSON object request bodies")
|
||||
})
|
||||
|
||||
export const jsonRequestParts = <Body>(input: JsonRequestInput<Body>) =>
|
||||
const isRequestValue = (value: unknown): value is RequestValue => {
|
||||
if (value === null || ["boolean", "number", "string"].includes(typeof value)) return true
|
||||
if (Array.isArray(value)) return value.every(isRequestValue)
|
||||
return ProviderShared.isRecord(value) && Object.values(value).every(isRequestValue)
|
||||
}
|
||||
|
||||
export const isRequestBody = (value: unknown): value is Record<string, RequestValue> =>
|
||||
ProviderShared.isRecord(value) && Object.values(value).every(isRequestValue)
|
||||
|
||||
const decodeJson = Schema.decodeUnknownEffect(Schema.UnknownFromJsonString)
|
||||
|
||||
export const decodeRequestBody = (text: string) =>
|
||||
decodeJson(text).pipe(
|
||||
Effect.mapError(() => ProviderShared.invalidRequest("Request hooks require a JSON object body")),
|
||||
Effect.flatMap((body) =>
|
||||
isRequestBody(body)
|
||||
? Effect.succeed(body)
|
||||
: Effect.fail(ProviderShared.invalidRequest("Request hooks require a JSON object body")),
|
||||
),
|
||||
)
|
||||
|
||||
export const jsonRequestBaseParts = <Body>(input: JsonRequestInput<Body>) =>
|
||||
Effect.gen(function* () {
|
||||
const url = applyQuery(
|
||||
renderEndpoint(input.endpoint, { request: input.request, body: input.body }).toString(),
|
||||
input.request.http?.query,
|
||||
)
|
||||
const body = yield* bodyWithOverlay(input.body, input.request, input.encodeBody)
|
||||
return {
|
||||
url,
|
||||
jsonBody: body.jsonBody,
|
||||
bodyText: body.bodyText,
|
||||
headers: {
|
||||
...input.headers?.({ request: input.request }),
|
||||
...input.request.http?.headers,
|
||||
},
|
||||
}
|
||||
})
|
||||
|
||||
export const jsonRequestParts = <Body>(input: JsonRequestInput<Body>) =>
|
||||
Effect.gen(function* () {
|
||||
const base = yield* jsonRequestBaseParts(input)
|
||||
const transformRequest = input.transformRequest
|
||||
const transformed = transformRequest
|
||||
? yield* transformRequest({ headers: base.headers, body: yield* decodeRequestBody(base.bodyText) })
|
||||
: { headers: base.headers, body: base.jsonBody }
|
||||
const bodyText = transformRequest ? ProviderShared.encodeJson(transformed.body) : base.bodyText
|
||||
const headers = yield* Auth.toEffect(input.auth)({
|
||||
request: input.request,
|
||||
method: "POST",
|
||||
url,
|
||||
body: body.bodyText,
|
||||
headers: Headers.fromInput({
|
||||
...input.headers?.({ request: input.request }),
|
||||
...input.request.http?.headers,
|
||||
}),
|
||||
url: base.url,
|
||||
body: bodyText,
|
||||
headers: Headers.fromInput(transformed.headers),
|
||||
})
|
||||
return { url, jsonBody: body.jsonBody, bodyText: body.bodyText, headers }
|
||||
return { url: base.url, jsonBody: transformed.body, bodyText, headers }
|
||||
})
|
||||
|
||||
export interface HttpJsonInput<_Body, Frame> {
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import { Auth } from "../auth"
|
|||
import type { Interface as RequestExecutorInterface } from "../executor"
|
||||
import type { Interface as WebSocketExecutorInterface } from "./websocket"
|
||||
import type { LLMError, LLMRequest } from "../../schema"
|
||||
import type { RequestTransform } from "../request-transform"
|
||||
|
||||
export interface TransportRuntime {
|
||||
readonly http: RequestExecutorInterface
|
||||
|
|
@ -27,6 +28,7 @@ export interface TransportPrepareInput<Body> {
|
|||
readonly auth: Auth.Definition
|
||||
readonly encodeBody: (body: Body) => string
|
||||
readonly headers?: (input: { readonly request: LLMRequest }) => Record<string, string>
|
||||
readonly transformRequest?: RequestTransform
|
||||
}
|
||||
|
||||
export * as HttpTransport from "./http"
|
||||
|
|
|
|||
|
|
@ -1,6 +1,8 @@
|
|||
import { Cause, Context, Effect, Layer, Queue, Stream } from "effect"
|
||||
import { Headers } from "effect/unstable/http"
|
||||
import { encodeJson } from "../../protocols/shared"
|
||||
import { LLMError, TransportReason } from "../../schema"
|
||||
import { Auth } from "../auth"
|
||||
import * as HttpTransport from "./http"
|
||||
import type { Transport } from "./index"
|
||||
|
||||
|
|
@ -228,13 +230,24 @@ export const json = <Body, Message>(input: JsonInput<Body, Message>): JsonTransp
|
|||
with: (patch) => json({ ...input, ...patch }),
|
||||
prepare: (prepareInput) =>
|
||||
Effect.gen(function* () {
|
||||
const parts = yield* HttpTransport.jsonRequestParts({
|
||||
...prepareInput,
|
||||
const parts = yield* HttpTransport.jsonRequestBaseParts(prepareInput)
|
||||
const message = input.encodeMessage(yield* input.toMessage(parts.jsonBody))
|
||||
const transformRequest = prepareInput.transformRequest
|
||||
const transformed = transformRequest
|
||||
? yield* transformRequest({ headers: parts.headers, body: yield* HttpTransport.decodeRequestBody(message) })
|
||||
: { headers: parts.headers, body: yield* HttpTransport.decodeRequestBody(message) }
|
||||
const bodyText = transformRequest ? encodeJson(transformed.body) : message
|
||||
const headers = yield* Auth.toEffect(prepareInput.auth)({
|
||||
request: prepareInput.request,
|
||||
method: "POST",
|
||||
url: parts.url,
|
||||
body: bodyText,
|
||||
headers: Headers.fromInput(transformed.headers),
|
||||
})
|
||||
return {
|
||||
url: yield* webSocketUrl(parts.url),
|
||||
headers: parts.headers,
|
||||
message: input.encodeMessage(yield* input.toMessage(parts.jsonBody)),
|
||||
headers,
|
||||
message: bodyText,
|
||||
}
|
||||
}),
|
||||
frames: (prepared, _request, runtime) => {
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
import { describe, expect, test } from "bun:test"
|
||||
import { Effect, Schema } from "effect"
|
||||
import { Effect, Schema, Stream } from "effect"
|
||||
import { HttpClientRequest } from "effect/unstable/http"
|
||||
import { LLM, mergeProviderOptions } from "../src"
|
||||
import { AnthropicMessages, OpenAIChat } from "../src/protocols"
|
||||
|
|
@ -136,6 +136,41 @@ describe("request option precedence", () => {
|
|||
),
|
||||
)
|
||||
|
||||
it.effect("transforms provider-native headers and JSON", () =>
|
||||
Effect.gen(function* () {
|
||||
const model = OpenAIChat.route
|
||||
.with({ endpoint: { baseURL: "https://api.openai.test/v1/" }, auth: Auth.bearer("test") })
|
||||
.model({ id: "gpt-4o-mini" })
|
||||
yield* LLMClient.stream(LLM.request({ model, prompt: "Say hello." }), {
|
||||
transformRequest: (request) =>
|
||||
Effect.sync(() => {
|
||||
const body = { ...request.body, store: true }
|
||||
delete body.stream_options
|
||||
return {
|
||||
headers: { ...request.headers, "x-plugin": "enabled" },
|
||||
body,
|
||||
}
|
||||
}),
|
||||
}).pipe(
|
||||
Stream.runDrain,
|
||||
Effect.provide(
|
||||
dynamicResponse((input) =>
|
||||
Effect.gen(function* () {
|
||||
const web = yield* HttpClientRequest.toWeb(input.request).pipe(Effect.orDie)
|
||||
expect(web.headers.get("authorization")).toBe("Bearer test")
|
||||
expect(web.headers.get("x-plugin")).toBe("enabled")
|
||||
expect(decodeJson(input.text)).toMatchObject({ store: true })
|
||||
expect(decodeJson(input.text)).not.toHaveProperty("stream_options")
|
||||
return input.respond(sseEvents(deltaChunk({}, "stop")), {
|
||||
headers: { "content-type": "text/event-stream" },
|
||||
})
|
||||
}),
|
||||
),
|
||||
),
|
||||
)
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("rejects raw body overlays for protocol-owned roots", () =>
|
||||
Effect.gen(function* () {
|
||||
const model = OpenAIChat.route
|
||||
|
|
|
|||
|
|
@ -251,16 +251,34 @@ describe("OpenAI Responses route", () => {
|
|||
}),
|
||||
),
|
||||
)
|
||||
const response = yield* LLMClient.generate(
|
||||
const text: string[] = []
|
||||
yield* LLMClient.stream(
|
||||
LLM.request({
|
||||
model: OpenAI.configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).responsesWebSocket(
|
||||
"gpt-4.1-mini",
|
||||
),
|
||||
prompt: "Say hello.",
|
||||
}),
|
||||
).pipe(Effect.provide(LLMClient.layer.pipe(Layer.provide(deps))))
|
||||
{
|
||||
transformRequest: (request) =>
|
||||
Effect.sync(() => {
|
||||
expect(request.body.type).toBe("response.create")
|
||||
expect(request.body.stream).toBeUndefined()
|
||||
const body = { ...request.body, plugin: true }
|
||||
delete body.store
|
||||
return { headers: { ...request.headers, "x-plugin": "enabled" }, body }
|
||||
}),
|
||||
},
|
||||
).pipe(
|
||||
Stream.runForEach((event) =>
|
||||
Effect.sync(() => {
|
||||
if (LLMEvent.is.textDelta(event)) text.push(event.text)
|
||||
}),
|
||||
),
|
||||
Effect.provide(LLMClient.layer.pipe(Layer.provide(deps))),
|
||||
)
|
||||
|
||||
expect(response.text).toBe("Hi")
|
||||
expect(text.join("")).toBe("Hi")
|
||||
expect(opened).toEqual([{ url: "wss://api.openai.test/v1/responses", authorization: "Bearer test" }])
|
||||
expect(closed).toBe(true)
|
||||
expect(sent).toHaveLength(1)
|
||||
|
|
@ -268,7 +286,7 @@ describe("OpenAI Responses route", () => {
|
|||
type: "response.create",
|
||||
model: "gpt-4.1-mini",
|
||||
input: [{ role: "user", content: [{ type: "input_text", text: "Say hello." }] }],
|
||||
store: false,
|
||||
plugin: true,
|
||||
})
|
||||
}),
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue