LeXwDeX · LeXwDeX · Aug 7, 2026 · Aug 7, 2026
diff --git 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<Body> = TransportPrepareInput<Body>
 
@@ -68,6 +68,29 @@ export interface HttpJsonTransport<Body, Frame> extends Transport<Body, HttpPrep
   readonly with: (patch: HttpJsonPatch<Body, Frame>) => HttpJsonTransport<Body, Frame>
 }
 
+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 = <Frame>(prepared: HttpPrepared<Frame>, 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 = <Body, Frame>(input: HttpJsonInput<Body, Frame>): HttpJsonTransport<Body, Frame> => ({
   id: "http-json",
   with: (patch) => httpJson({ ...input, ...patch }),
@@ -80,26 +103,28 @@ export const httpJson = <Body, Frame>(input: HttpJsonInput<Body, Frame>): 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
@@ -54,6 +54,7 @@ export class HttpOptions extends Schema.Class<HttpOptions>("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<HttpOptions | undefined
   const body = mergeJsonRecords(...items.map((item) => 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<GenerationOptions>("LLM.GenerationOptions")({

diff --git a/packages/llm/test/transport-timeout.test.ts 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<readonly LLMEvent[], LLMError>) => {
+  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)
+  })
+})