diff --git a/packages/llm/src/route/transport/http.ts b/packages/llm/src/route/transport/http.ts index 00508957a7..6d7dd84a0e 100644 --- a/packages/llm/src/route/transport/http.ts +++ b/packages/llm/src/route/transport/http.ts @@ -1,11 +1,11 @@ -import { Effect, Stream } from "effect" -import { Headers, HttpClientRequest } from "effect/unstable/http" +import { Cause, Duration, Effect, Stream } from "effect" +import { Headers, HttpClientRequest, type HttpClientResponse } from "effect/unstable/http" import { Auth } from "../auth" import { render as renderEndpoint } from "../endpoint" import { Framing, type Framing as FramingDef } from "../framing" import type { Transport, TransportPrepareInput } from "./index" import * as ProviderShared from "../../protocols/shared" -import { mergeJsonRecords, type LLMRequest } from "../../schema" +import { LLMError, TransportReason, mergeJsonRecords, type LLMRequest } from "../../schema" export type JsonRequestInput = TransportPrepareInput @@ -68,6 +68,29 @@ export interface HttpJsonTransport extends Transport) => HttpJsonTransport } +const timeoutError = (provider: string, timeout: Duration.Duration) => + new LLMError({ + module: "RequestExecutor", + method: "execute", + reason: new TransportReason({ + message: `Provider ${provider} stream timed out after ${Duration.toMillis(timeout)}ms without data`, + kind: "Timeout", + }), + }) + +const readStream = (prepared: HttpPrepared, provider: string) => (response: HttpClientResponse.HttpClientResponse) => + prepared.framing.frame( + response.stream.pipe( + Stream.mapError((error) => + ProviderShared.eventError( + provider, + `Failed to read ${provider} stream`, + ProviderShared.errorText(error), + ), + ), + ), + ) + export const httpJson = (input: HttpJsonInput): HttpJsonTransport => ({ id: "http-json", with: (patch) => httpJson({ ...input, ...patch }), @@ -80,26 +103,28 @@ export const httpJson = (input: HttpJsonInput): HttpJs framing: input.framing, })), ), - frames: (prepared, request, runtime) => - Stream.unwrap( - runtime.http - .execute(prepared.request) - .pipe( - Effect.map((response) => - prepared.framing.frame( - response.stream.pipe( - Stream.mapError((error) => - ProviderShared.eventError( - `${request.model.provider}/${request.model.route.id}`, - `Failed to read ${request.model.provider}/${request.model.route.id} stream`, - ProviderShared.errorText(error), - ), - ), - ), - ), + frames: (prepared, request, runtime) => { + const provider = `${request.model.provider}/${request.model.route.id}` + const timeout = request.http?.timeout + if (timeout === undefined) { + return Stream.unwrap(runtime.http.execute(prepared.request).pipe(Effect.map(readStream(prepared, provider)))) + } + const execute = runtime.http + .execute(prepared.request) + .pipe( + Effect.timeout(timeout), + Effect.mapError((error) => (Cause.isTimeoutError(error) ? timeoutError(provider, timeout) : error)), + Effect.map((response) => + readStream(prepared, provider)(response).pipe( + Stream.timeoutOrElse({ + duration: timeout, + orElse: () => Stream.fail(timeoutError(provider, timeout)), + }), ), ), - ), + ) + return Stream.unwrap(execute) + }, }) export const sseJson = { diff --git a/packages/llm/src/schema/options.ts b/packages/llm/src/schema/options.ts index c02af6d1ed..747d2d5ff0 100644 --- a/packages/llm/src/schema/options.ts +++ b/packages/llm/src/schema/options.ts @@ -54,6 +54,7 @@ export class HttpOptions extends Schema.Class("LLM.HttpOptions")({ body: Schema.optional(JsonSchema), headers: Schema.optional(Schema.Record(Schema.String, Schema.String)), query: Schema.optional(Schema.Record(Schema.String, Schema.String)), + timeout: Schema.optional(Schema.DurationFromMillis), }) {} export namespace HttpOptions { @@ -67,8 +68,9 @@ export const mergeHttpOptions = (...items: ReadonlyArray item?.body)) const headers = mergeStringRecords(...items.map((item) => item?.headers)) const query = mergeStringRecords(...items.map((item) => item?.query)) - if (!body && !headers && !query) return undefined - return new HttpOptions({ body, headers, query }) + const timeout = items.findLast((item) => item?.timeout !== undefined)?.timeout + if (!body && !headers && !query && timeout === undefined) return undefined + return new HttpOptions({ body, headers, query, ...(timeout === undefined ? {} : { timeout }) }) } export class GenerationOptions extends Schema.Class("LLM.GenerationOptions")({ diff --git a/packages/llm/test/transport-timeout.test.ts b/packages/llm/test/transport-timeout.test.ts new file mode 100644 index 0000000000..bb1e19f2e1 --- /dev/null +++ b/packages/llm/test/transport-timeout.test.ts @@ -0,0 +1,107 @@ +import { describe, expect, test } from "bun:test" +import { Cause, Duration, Effect, Exit, Fiber, Option, Stream } from "effect" +import * as TestClock from "effect/testing/TestClock" +import { LLM, LLMError, LLMEvent } from "../src" +import * as OpenAIChat from "../src/protocols/openai-chat" +import { HttpOptions, Model, mergeHttpOptions } from "../src/schema" +import { LLMClient } from "../src/route" +import { testEffect } from "./lib/effect" +import { dynamicResponse, fixedResponse } from "./lib/http" +import { deltaChunk } from "./lib/openai-chunks" +import { sseEvents } from "./lib/sse" + +const model = Model.make({ id: "fake-model", provider: "fake", route: OpenAIChat.route }) + +const request = (timeout?: number) => + LLM.request({ + model, + prompt: "Say hello.", + http: timeout === undefined ? undefined : { timeout: Duration.millis(timeout) }, + }) + +const hangingHeaders = dynamicResponse(() => Effect.never) + +const hangingBody = dynamicResponse((input) => + Effect.sync(() => + input.respond(new ReadableStream({ start() {} }), { headers: { "content-type": "text/event-stream" } }), + ), +) + +const expectTimeoutExit = (exit: Exit.Exit) => { + if (Exit.isSuccess(exit)) { + throw new Error(`expected a Timeout failure, stream completed with ${exit.value.length} events`) + } + const error = Option.getOrThrow(Cause.findErrorOption(exit.cause)) + expect(error).toBeInstanceOf(LLMError) + if (!(error instanceof LLMError)) throw new Error("expected LLMError") + expect(error.reason).toMatchObject({ _tag: "Transport", kind: "Timeout" }) +} + +describe("http transport timeout", () => { + testEffect(hangingHeaders).effect( + "ends the stream with a Timeout error when the provider never sends response headers", + () => + Effect.gen(function* () { + const fiber = yield* LLMClient.stream(request(1000)).pipe(Stream.runCollect, Effect.forkChild) + yield* TestClock.adjust(2000) + expectTimeoutExit(yield* Fiber.join(fiber).pipe(Effect.exit)) + }), + ) + + testEffect(hangingBody).effect( + "ends the stream with a Timeout error when the response body never emits", + () => + Effect.gen(function* () { + const fiber = yield* LLMClient.stream(request(1000)).pipe(Stream.runCollect, Effect.forkChild) + yield* TestClock.adjust(2000) + expectTimeoutExit(yield* Fiber.join(fiber).pipe(Effect.exit)) + }), + ) + + testEffect(fixedResponse(sseEvents(deltaChunk({ role: "assistant", content: "Hello" })))).effect( + "completes normally when the stream finishes within the timeout", + () => + Effect.gen(function* () { + const events = yield* LLMClient.stream(request(1000)).pipe(Stream.runCollect) + expect(events.some(LLMEvent.is.textDelta)).toBe(true) + }), + ) + + testEffect(hangingHeaders).effect( + "applies the route default timeout when the request omits http", + () => + Effect.gen(function* () { + const defaultModel = Model.make({ + id: "fake-model", + provider: "fake", + route: OpenAIChat.route.with({ http: { timeout: Duration.millis(500) } }), + }) + const fiber = yield* LLMClient.stream(LLM.request({ model: defaultModel, prompt: "Say hello." })).pipe( + Stream.runCollect, + Effect.forkChild, + ) + yield* TestClock.adjust(1000) + expectTimeoutExit(yield* Fiber.join(fiber).pipe(Effect.exit)) + }), + ) +}) + +describe("HttpOptions.timeout merging", () => { + test("keeps existing merge behavior when no timeout is set", () => { + expect(mergeHttpOptions(new HttpOptions({ headers: { "x-a": "1" } }), undefined)).toEqual( + new HttpOptions({ headers: { "x-a": "1" } }), + ) + expect(mergeHttpOptions()).toBeUndefined() + expect(new HttpOptions({ headers: { "x-a": "1" } }).timeout).toBeUndefined() + }) + + test("merges timeout with last-wins semantics", () => { + const merged = mergeHttpOptions( + new HttpOptions({ timeout: Duration.millis(1000) }), + new HttpOptions({ headers: { "x-a": "1" }, timeout: Duration.millis(2500) }), + undefined, + ) + expect(merged?.headers).toEqual({ "x-a": "1" }) + expect(Duration.toMillis(merged?.timeout ?? Duration.zero)).toBe(2500) + }) +})