refactor(core): simplify provider timeout wiring

fix(core): default provider header and chunk timeouts to five minutes
2026-09-03 07:26:20 +00:00 · 2026-09-02 18:23:14 -05:00 · 2026-09-02 17:55:42 -05:00
10 changed files with 331 additions and 31 deletions
@@ -1,11 +1,20 @@
-import { Effect } from "effect"
+import { Duration, Effect, Stream } from "effect"
 import { Headers, HttpClientRequest } from "effect/unstable/http"
 import { Auth } from "../auth.js"
 import { render as renderEndpoint } from "../endpoint.js"
 import { Framing } from "../framing.js"
 import type { HttpMiddleware, Transport, TransportPrepareInput } from "./index.js"
 import * as ProviderShared from "../../protocols/shared.js"
-import { mergeJsonRecords, type LLMRequest } from "../../schema/index.js"
+import {
+  AIError,
+  DEFAULT_HTTP_TIMEOUT_MS,
+  mergeJsonRecords,
+  TransportError,
+  type HttpContext,
+  type HttpTimeout,
+  type LLMRequest,
+  type TransportOperation,
+} from "../../schema/index.js"
 import { RequestExecutor } from "../executor.js"

 export type JsonRequestInput<Body> = TransportPrepareInput<Body>
@@ -87,17 +96,45 @@ export const httpJson = <Body, Frame>(input: HttpJsonInput<Body, Frame>): HttpJs
        middleware: prepareInput.middleware,
      }
    }),
-  execute: (prepared, _request, runtime) =>
+  execute: (prepared, request, runtime) =>
    Effect.gen(function* () {
-      const response = yield* runtime.http.execute(prepared.request, prepared.middleware)
+      const timeout = (operation: TransportOperation, message: string, http?: HttpContext) =>
+        new AIError({
+          reason: new TransportError({
+            message,
+            transport: "http",
+            operation,
+            code: "Timeout",
+            url: prepared.request.url,
+            http,
+          }),
+        })
+      const response = yield* runtime.http.execute(prepared.request, prepared.middleware).pipe(
+        Effect.timeoutOrElse({
+          duration: timeoutDuration(request.http?.headerTimeout),
+          orElse: () => Effect.fail(timeout("request", "Timed out waiting for response headers")),
+        }),
+      )
+      const http = RequestExecutor.responseHttp(response)
      return {
-        frames: prepared.framing.frame(RequestExecutor.responseStream(response)),
-        http: RequestExecutor.responseHttp(response),
+        frames: prepared.framing.frame(
+          RequestExecutor.responseStream(response).pipe(
+            Stream.timeoutOrElse({
+              duration: timeoutDuration(request.http?.chunkTimeout),
+              orElse: () => Stream.fail(timeout("read", "Timed out waiting for response data", http)),
+            }),
+          ),
+        ),
+        http,
        body: prepared.framing.body,
      }
    }),
 })

+// `false` disables a timer; an unset value uses the shared default.
+const timeoutDuration = (value: HttpTimeout | undefined) =>
+  value === false ? Duration.infinity : Duration.millis(value ?? DEFAULT_HTTP_TIMEOUT_MS)
+
 export const sseJson = {
  id: "http-json/sse",
  with: <Body>() => httpJson<Body, string>({ framing: Framing.sse }),
@@ -43,10 +43,21 @@ export const mergeProviderOptions = (
  ...items: ReadonlyArray<ProviderOptions | undefined>
 ): ProviderOptions | undefined => mergeJsonRecords(...items)

+/** Milliseconds for an HTTP timeout, or `false` to disable it. */
+export const HttpTimeout = Schema.Union([Schema.Number.check(Schema.isGreaterThan(0)), Schema.Literal(false)])
+export type HttpTimeout = Schema.Schema.Type<typeof HttpTimeout>
+
+/** Default for `headerTimeout` and `chunkTimeout` when a request leaves them unset. */
+export const DEFAULT_HTTP_TIMEOUT_MS = 300_000
+
 export class HttpOptions extends Schema.Class<HttpOptions>("AI.HttpOptions")({
  body: Schema.optional(JsonSchema),
  headers: Schema.optional(Schema.Record(Schema.String, Schema.String)),
  query: Schema.optional(Schema.Record(Schema.String, Schema.String)),
+  /** Time allowed for response headers to arrive. Defaults to five minutes. */
+  headerTimeout: Schema.optional(HttpTimeout),
+  /** Time allowed between streamed response chunks once headers have arrived. Defaults to five minutes. */
+  chunkTimeout: Schema.optional(HttpTimeout),
 }) {}

 export namespace HttpOptions {
@@ -60,8 +71,10 @@ 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 headerTimeout = items.findLast((item) => item?.headerTimeout !== undefined)?.headerTimeout
+  const chunkTimeout = items.findLast((item) => item?.chunkTimeout !== undefined)?.chunkTimeout
+  if (!body && !headers && !query && headerTimeout === undefined && chunkTimeout === undefined) return undefined
+  return new HttpOptions({ body, headers, query, headerTimeout, chunkTimeout })
 }

 export class GenerationOptions extends Schema.Class<GenerationOptions>("LLM.GenerationOptions")({
@@ -0,0 +1,144 @@
+import { describe, expect } from "bun:test"
+import { Deferred, Effect, Fiber } from "effect"
+import * as TestClock from "effect/testing/TestClock"
+import { HttpOptions, LLM, mergeHttpOptions } from "../src/index.js"
+import { LLMClient } from "../src/route.js"
+import { configure } from "../src/providers/openai.js"
+import { dynamicResponse } from "./lib/http.js"
+import { deltaChunk, finishChunk } from "./lib/openai-chunks.js"
+import { sseEvents } from "./lib/sse.js"
+import { it } from "./lib/effect.js"
+
+const model = configure({ baseURL: "https://api.openai.test/v1/", apiKey: "test" }).chat("gpt-4.1-mini")
+const body = sseEvents(deltaChunk({ role: "assistant", content: "Hi" }), finishChunk("stop"))
+const SSE = { headers: { "content-type": "text/event-stream" } }
+
+// Never produces headers; the header timer is the only way out.
+const silentServer = dynamicResponse(() => Effect.never)
+
+// Sends one chunk, then stalls until `resume` releases the rest of the body.
+const stalledServer = Effect.gen(function* () {
+  const stalled = yield* Deferred.make<void>()
+  let resume = () => {}
+  const released = new Promise<void>((resolve) => {
+    resume = resolve
+  })
+  const encoder = new TextEncoder()
+  const layer = dynamicResponse((input) =>
+    Effect.sync(() =>
+      input.respond(
+        new ReadableStream<Uint8Array>({
+          start(controller) {
+            controller.enqueue(encoder.encode(`data: ${JSON.stringify(deltaChunk({ content: "Hi" }))}\n\n`))
+          },
+          async pull(controller) {
+            Deferred.doneUnsafe(stalled, Effect.void)
+            await released
+            controller.enqueue(encoder.encode(`data: ${JSON.stringify(finishChunk("stop"))}\n\ndata: [DONE]\n\n`))
+            controller.close()
+          },
+        }),
+        SSE,
+      ),
+    ),
+  )
+  return { layer, stalled, resume: () => resume() }
+})
+
+const slowHeadersServer = dynamicResponse((input) =>
+  Effect.sleep("10 minutes").pipe(Effect.as(input.respond(body, SSE))),
+)
+
+describe("HTTP transport timeouts", () => {
+  it.effect("fails when response headers take longer than five minutes", () =>
+    Effect.gen(function* () {
+      const fiber = yield* LLMClient.generate(LLM.request({ model, prompt: "Hello" })).pipe(
+        Effect.provide(silentServer),
+        Effect.flip,
+        Effect.forkChild({ startImmediately: true }),
+      )
+      yield* TestClock.adjust("5 minutes")
+      const error = yield* Fiber.join(fiber)
+
+      expect(error.reason).toMatchObject({
+        _tag: "Transport",
+        transport: "http",
+        operation: "request",
+        code: "Timeout",
+      })
+      expect(error.reason.message).toContain("response headers")
+    }),
+  )
+
+  it.effect("fails when the response body stalls for five minutes", () =>
+    Effect.gen(function* () {
+      const server = yield* stalledServer
+      const fiber = yield* LLMClient.generate(LLM.request({ model, prompt: "Hello" })).pipe(
+        Effect.provide(server.layer),
+        Effect.flip,
+        Effect.forkChild({ startImmediately: true }),
+      )
+      yield* Deferred.await(server.stalled)
+      yield* Effect.yieldNow
+      yield* TestClock.adjust("5 minutes")
+      const error = yield* Fiber.join(fiber)
+
+      expect(error.reason).toMatchObject({ _tag: "Transport", transport: "http", operation: "read", code: "Timeout" })
+      expect(error.reason.http).toMatchObject({ status: 200 })
+    }),
+  )
+
+  it.effect("applies a configured header timeout", () =>
+    Effect.gen(function* () {
+      const fiber = yield* LLMClient.generate(
+        LLM.request({ model, prompt: "Hello", http: { headerTimeout: 1_000 } }),
+      ).pipe(Effect.provide(silentServer), Effect.flip, Effect.forkChild({ startImmediately: true }))
+      yield* TestClock.adjust("1 second")
+      const error = yield* Fiber.join(fiber)
+
+      expect(error.reason).toMatchObject({ _tag: "Transport", operation: "request", code: "Timeout" })
+    }),
+  )
+
+  it.effect("disables the header timeout with false", () =>
+    Effect.gen(function* () {
+      const fiber = yield* LLMClient.generate(
+        LLM.request({ model, prompt: "Hello", http: { headerTimeout: false } }),
+      ).pipe(Effect.provide(slowHeadersServer), Effect.forkChild({ startImmediately: true }))
+      yield* TestClock.adjust("10 minutes")
+      const response = yield* Fiber.join(fiber)
+
+      expect(response.text).toBe("Hi")
+    }),
+  )
+
+  it.effect("disables the chunk timeout with false", () =>
+    Effect.gen(function* () {
+      const server = yield* stalledServer
+      const fiber = yield* LLMClient.generate(
+        LLM.request({ model, prompt: "Hello", http: { chunkTimeout: false } }),
+      ).pipe(Effect.provide(server.layer), Effect.forkChild({ startImmediately: true }))
+      yield* Deferred.await(server.stalled)
+      yield* Effect.yieldNow
+      yield* TestClock.adjust("10 minutes")
+      server.resume()
+      const response = yield* Fiber.join(fiber)
+
+      expect(response.text).toBe("Hi")
+    }),
+  )
+
+  it.effect("merges timeouts with later values winning", () =>
+    Effect.sync(() => {
+      const merged = mergeHttpOptions(
+        new HttpOptions({ headerTimeout: 1_000, chunkTimeout: 2_000 }),
+        new HttpOptions({ headers: { a: "b" } }),
+        new HttpOptions({ chunkTimeout: false }),
+      )
+
+      expect(merged).toMatchObject({ headers: { a: "b" }, headerTimeout: 1_000, chunkTimeout: false })
+      expect(mergeHttpOptions(new HttpOptions({}), undefined)).toBeUndefined()
+      expect(mergeHttpOptions(new HttpOptions({ headerTimeout: false }))?.headerTimeout).toBe(false)
+    }),
+  )
+})
@@ -16,6 +16,7 @@ import type {
  SharedV3ProviderOptions,
 } from "@ai-sdk/provider"
 import {
+  DEFAULT_HTTP_TIMEOUT_MS,
  FinishReason,
  LLMEvent,
  AIError,
@@ -126,21 +127,28 @@ function prepareOptions(model: Info, pkg: string) {
  }

  const customFetch = options.fetch
-  const chunkTimeout = options.chunkTimeout
+  const timeouts = Provider.timeouts(options)
+  const chunkTimeout = timeouts.chunkTimeout ?? DEFAULT_HTTP_TIMEOUT_MS
+  const headerTimeout = timeouts.headerTimeout ?? DEFAULT_HTTP_TIMEOUT_MS
  delete options.chunkTimeout
+  delete options.headerTimeout
  options.fetch = async (input: Parameters<typeof fetch>[0], init?: RequestInit) => {
    const opts = { ...(init ?? {}) }
-    const signals = [
-      opts.signal,
-      typeof chunkTimeout === "number" && chunkTimeout > 0 ? new AbortController() : undefined,
-      options.timeout !== undefined && options.timeout !== null && options.timeout !== false
-        ? AbortSignal.timeout(options.timeout)
-        : undefined,
-    ].filter((item): item is AbortSignal | AbortController => item !== undefined && item !== null)
-    const chunkAbortCtl = signals.find((item): item is AbortController => item instanceof AbortController)
-    const abortSignals = signals.map((item) => (item instanceof AbortController ? item.signal : item))
-    if (abortSignals.length === 1) opts.signal = abortSignals[0]
-    if (abortSignals.length > 1) opts.signal = AbortSignal.any(abortSignals)
+    const ctl = new AbortController()
+    // Only covers the wait for response headers; wrapSSE takes over once the body streams.
+    const headerTimer =
+      headerTimeout === false
+        ? undefined
+        : setTimeout(() => ctl.abort(new Error(HEADER_TIMEOUT_MESSAGE)), headerTimeout)
+    opts.signal = AbortSignal.any(
+      [
+        opts.signal,
+        ctl.signal,
+        options.timeout !== undefined && options.timeout !== null && options.timeout !== false
+          ? AbortSignal.timeout(options.timeout)
+          : undefined,
+      ].filter((item): item is AbortSignal => item !== undefined && item !== null),
+    )

    if (typeof opts.body === "string" && model.body !== undefined) {
      const decoded = Option.getOrUndefined(decodeJson(opts.body))
@@ -152,14 +160,16 @@ function prepareOptions(model: Info, pkg: string) {
    const res = await (typeof customFetch === "function" ? customFetch : fetch)(input, {
      ...opts,
      timeout: false,
-    })
-    if (!chunkAbortCtl || typeof chunkTimeout !== "number") return res
-    return wrapSSE(res, chunkTimeout, chunkAbortCtl)
+    }).finally(() => clearTimeout(headerTimer))
+    if (chunkTimeout === false) return res
+    return wrapSSE(res, chunkTimeout, ctl)
  }

  return options
 }

+const HEADER_TIMEOUT_MESSAGE = "Response headers timed out"
+
 export class InitError extends Schema.TaggedError<InitError>()("AISDK.InitError", {
  providerID: Provider.ID,
  cause: Schema.Defect(),
@@ -388,7 +398,7 @@ function requestSettings(settings: Readonly<Record<string, unknown>> | undefined
  if (settings === undefined) return undefined
  const result = Object.fromEntries(
    Object.entries(settings).filter(
-      ([key]) => !["apiKey", "authToken", "baseURL", "chunkTimeout", "fetch", "timeout"].includes(key),
+      ([key]) => !["apiKey", "authToken", "baseURL", "chunkTimeout", "fetch", "headerTimeout", "timeout"].includes(key),
    ),
  )
  return Object.keys(result).length === 0 ? undefined : result
@@ -838,7 +848,7 @@ function llmError(error: unknown, operation: "request" | "read") {

 // Runtime-generated network failure shapes. The codes mirror the AI SDK's own
 // Bun network error list in handleFetchError; the messages are undici's fetch
-// TypeError and stream termination strings plus our SSE chunk timeout error.
+// TypeError and stream termination strings plus our header and chunk timeout errors.
 // Unrecognized shapes still retry via the UnknownProvider default; this match
 // only adds transport semantics (continuation eligibility, display).
 const NETWORK_ERROR_CODES = new Set([
@@ -856,6 +866,7 @@ const NETWORK_ERROR_MESSAGES = new Set([
  "terminated",
  "other side closed",
  "sse read timed out",
+  HEADER_TIMEOUT_MESSAGE.toLowerCase(),
 ])

 const NativeErrorShape = Schema.Struct({
@@ -1,7 +1,7 @@
 export * as ModelResolver from "./model-resolver.js"

 import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
-import { LanguageModel } from "@opencode-ai/ai"
+import { HttpOptions, LanguageModel, mergeHttpOptions } from "@opencode-ai/ai"
 import { Auth } from "@opencode-ai/ai/route"
 import { Context, Effect, Layer, Schema, Struct } from "effect"
 import { AISDK } from "./aisdk.js"
@@ -128,7 +128,10 @@ const resolveCatalogModel = Effect.fn("ModelResolver.resolveCatalogModel")(funct
  const resolved = prepareRuntimeModel(model, credential)
  const packageName = Provider.packageName(resolved.package)
  const configuration = credential?.type === "key" ? credential.configuration : undefined
-  const configured = { ...resolved.settings, ...credential?.metadata, ...configuration }
+  const merged = { ...resolved.settings, ...credential?.metadata, ...configuration }
+  // Timeouts are transport policy: they become route HTTP defaults rather than provider package settings.
+  const timeouts = Provider.timeouts(merged)
+  const configured = Struct.omit(merged, ["headerTimeout", "chunkTimeout"])
  const mapping = Provider.isAISDK(resolved.package)
    ? AISDKNative.map({
        packageName,
@@ -173,6 +176,7 @@ const resolveCatalogModel = Effect.fn("ModelResolver.resolveCatalogModel")(funct
        compatibility: resolved.compatibility
          ? Object.assign({}, runtime.compatibility, resolved.compatibility)
          : runtime.compatibility,
+        defaults: { ...runtime.defaults, http: mergeHttpOptions(runtime.defaults?.http, new HttpOptions(timeouts)) },
      })
    },
    catch: () => unsupported(resolved),
@@ -1,8 +1,8 @@
 export * as Provider from "./provider.js"

-import { Effect, Schema } from "effect"
+import { Effect, Option, Schema } from "effect"
 import { Provider } from "@opencode-ai/schema/provider"
-import type { ProviderPackageDefinition } from "@opencode-ai/ai"
+import { HttpTimeout, type ProviderPackageDefinition } from "@opencode-ai/ai"
 import { isRecord } from "@opencode-ai/ai/utils/record"
 import { Npm } from "@opencode-ai/util/npm"
 import type { DeepMutable } from "./schema.js"
@@ -136,6 +136,16 @@ export function mergeHeaders(
  )
 }

+const decodeTimeout = Schema.decodeUnknownOption(HttpTimeout)
+
+/** Read request timeouts from provider settings; invalid values are ignored so the default applies. */
+export function timeouts(settings: Readonly<Record<string, unknown>>) {
+  return {
+    headerTimeout: Option.getOrUndefined(decodeTimeout(settings.headerTimeout)),
+    chunkTimeout: Option.getOrUndefined(decodeTimeout(settings.chunkTimeout)),
+  }
+}
+
 export const Request = Provider.Request
 export type Request = Provider.Request

@@ -226,9 +226,8 @@ const applyModelHooks = (hooks: PluginHooks.Interface, scope: HookScope, request
    return LLMRequest.update(request, {
      model: route === request.model.route ? request.model : LanguageModel.update(request.model, { route }),
      http: new HttpOptions({
-        body: request.http?.body,
+        ...request.http,
        headers: Object.keys(event.headers).length === 0 ? undefined : event.headers,
-        query: request.http?.query,
      }),
    })
  })
@@ -701,6 +701,45 @@ it.effect("does not treat SSE comment heartbeats as model progress", () =>
  }),
 )

+it.effect("fails with a retryable transport error when response headers time out", () =>
+  Effect.gen(function* () {
+    const aisdk = yield* AISDK.Service
+    const customFetch = Object.assign(
+      (_input: Parameters<typeof fetch>[0], init?: RequestInit) =>
+        new Promise<Response>((_, reject) => {
+          init?.signal?.addEventListener("abort", () => reject(init.signal?.reason), { once: true })
+        }),
+      { preconnect: fetch.preconnect },
+    )
+    yield* aisdk.hook.sdk((event) => {
+      event.sdk = createOpenAICompatible({
+        ...event.options,
+        name: String(event.options.name),
+        baseURL: String(event.options.baseURL),
+      })
+    })
+    const resolved = yield* aisdk.model(
+      model("@ai-sdk/openai-compatible", {
+        apiKey: "test",
+        baseURL: "https://example.test/v1",
+        headerTimeout: 25,
+        fetch: customFetch,
+      }),
+    )
+    const error = yield* LLMClient.generate(LLM.request({ model: resolved, prompt: "Hello" })).pipe(
+      Effect.provide(client),
+      Effect.flip,
+    )
+
+    expect(error.reason).toMatchObject({
+      _tag: "Transport",
+      operation: "request",
+      message: "Response headers timed out",
+    })
+    expect(SessionRunnerRetry.isRetryable(error)).toBe(true)
+  }),
+)
+
 it.effect("emits malformed AI SDK tool input without executing it", () =>
  Effect.gen(function* () {
    const aisdk = yield* AISDK.Service
@@ -247,6 +247,22 @@ describe("ModelResolver", () => {
    }),
  )

+  it.effect("lifts timeout settings onto native route HTTP defaults", () =>
+    Effect.gen(function* () {
+      const resolved = yield* ModelResolver.fromCatalogModel(
+        model(Provider.aisdk("@ai-sdk/anthropic"), {
+          settings: { baseURL: "https://anthropic.example/v1", headerTimeout: false, chunkTimeout: 60_000 },
+        }),
+        Credential.Key.make({ type: "key", key: "secret" }),
+      )
+      const prepared = yield* compileRequest(LLM.request({ model: resolved, prompt: "Hello" }))
+
+      expect(resolved.defaults?.http).toMatchObject({ headerTimeout: false, chunkTimeout: 60_000 })
+      expect(resolved.defaults?.providerOptions).toBeUndefined()
+      expect(JSON.stringify(prepared.body)).not.toContain("Timeout")
+    }),
+  )
+
  it.effect("keeps catalog apiKey credentials out of provider JSON", () =>
    Effect.gen(function* () {
      const resolved = yield* ModelResolver.fromCatalogModel(
@@ -138,6 +138,33 @@ body:
 }
 ```

+## Timeouts
+
+Provider requests wait up to five minutes for response headers and up to five minutes between streamed response chunks.
+Set `settings.headerTimeout` or `settings.chunkTimeout` to a number of milliseconds to change a limit, or to `false` to
+disable it:
+
+```jsonc title="opencode.jsonc"
+{
+  "$schema": "https://opencode.ai/config.json",
+  "providers": {
+    "anthropic": {
+      "settings": {
+        "headerTimeout": 600000,
+        "chunkTimeout": false,
+      },
+    },
+  },
+}
+```
+
+| Setting         | Default  | Purpose                                                                      |
+| --------------- | -------- | ---------------------------------------------------------------------------- |
+| `headerTimeout` | `300000` | Time to wait for response headers. Stops once headers arrive.                |
+| `chunkTimeout`  | `300000` | Time allowed between streamed response chunks. Resets whenever data arrives. |
+
+A request that times out fails with a transport error and follows the normal retry policy.
+
 ## Package

 The `package` field selects the runtime used to communicate with a provider. For an OpenAI-compatible API, use the
Author	SHA1	Message	Date
Aiden Cline	b85cf3d67f	refactor(core): simplify provider timeout wiring	2026-09-02 18:23:14 -05:00
Aiden Cline	2b72a95502	fix(core): default provider header and chunk timeouts to five minutes	2026-09-02 17:55:42 -05:00