fix(core): default provider header and chunk timeouts to five minutes

This commit is contained in:
Aiden Cline
2026-09-23 20:11:06 -05:00
parent 0c4bbc3cd1
commit 54f7ab6ada
10 changed files with 341 additions and 36 deletions
+41 -6
View File
@@ -1,11 +1,19 @@
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,
} from "../../schema/index.js"
import { RequestExecutor } from "../executor.js"
export type JsonRequestInput<Body> = TransportPrepareInput<Body>
@@ -87,17 +95,44 @@ 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: "request" | "read", 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: () => 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,
}
}),
})
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 }),
+15 -2
View File
@@ -48,10 +48,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>
/** Applied to `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. */
headerTimeout: Schema.optional(HttpTimeout),
/** Time allowed between streamed response chunks once headers have arrived. */
chunkTimeout: Schema.optional(HttpTimeout),
}) {}
export namespace HttpOptions {
@@ -70,8 +81,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")({
+142
View File
@@ -0,0 +1,142 @@
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, sseRaw } 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 SSE = { headers: { "content-type": "text/event-stream" } }
const silentServer = dynamicResponse(() => Effect.never)
const slowHeadersServer = dynamicResponse((input) =>
Effect.sleep("10 minutes").pipe(
Effect.as(input.respond(sseEvents(deltaChunk({ role: "assistant", content: "Hi" }), finishChunk("stop")), SSE)),
),
)
// 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(sseRaw(`data: ${JSON.stringify(deltaChunk({ content: "Hi" }))}`)))
},
async pull(controller) {
Deferred.doneUnsafe(stalled, Effect.void)
await released
controller.enqueue(encoder.encode(sseEvents(finishChunk("stop"))))
controller.close()
},
}),
SSE,
),
),
)
return { layer, stalled, resume: () => resume() }
})
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",
})
}),
)
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()
}),
)
})
+42 -21
View File
@@ -16,6 +16,7 @@ import type {
SharedV3ProviderOptions,
} from "@ai-sdk/provider"
import {
DEFAULT_HTTP_TIMEOUT_MS,
FinishReason,
LLMEvent,
AIError,
@@ -128,23 +129,30 @@ function prepareOptions(model: RuntimeInfo, pkg: string) {
}
const customFetch = options.fetch
const chunkTimeout = options.chunkTimeout
const timeouts = Provider.timeouts(options)
const headerTimeout = timeouts.headerTimeout ?? DEFAULT_HTTP_TIMEOUT_MS
const chunkTimeout = timeouts.chunkTimeout ?? DEFAULT_HTTP_TIMEOUT_MS
delete options.headerTimeout
delete options.chunkTimeout
delete options.compaction
delete options.transport
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()
// Covers only 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))
@@ -155,11 +163,13 @@ function prepareOptions(model: RuntimeInfo, pkg: string) {
const send: Fetch = typeof customFetch === "function" ? customFetch : fetch
const middleware = httpMiddleware.getStore()
const res = middleware
? await throughMiddleware(middleware, send, input, { ...opts, timeout: false })
: await send(input, { ...opts, timeout: false })
if (!chunkAbortCtl || typeof chunkTimeout !== "number") return res
return wrapSSE(res, chunkTimeout, chunkAbortCtl)
const res = await (
middleware
? throughMiddleware(middleware, send, input, { ...opts, timeout: false })
: send(input, { ...opts, timeout: false })
).finally(() => clearTimeout(headerTimer))
if (chunkTimeout === false) return res
return wrapSSE(res, chunkTimeout, ctl)
}
return options
@@ -215,6 +225,8 @@ function throughMiddleware(
)
}
const HEADER_TIMEOUT_MESSAGE = "Response headers timed out"
export class InitError extends Schema.TaggedError<InitError>()("AISDK.InitError", {
providerID: Provider.ID,
cause: Schema.Defect(),
@@ -447,9 +459,17 @@ function requestSettings(settings: Readonly<Record<string, unknown>> | undefined
const result = Object.fromEntries(
Object.entries(settings).filter(
([key]) =>
!["apiKey", "authToken", "baseURL", "chunkTimeout", "compaction", "fetch", "timeout", "transport"].includes(
key,
),
![
"apiKey",
"authToken",
"baseURL",
"chunkTimeout",
"compaction",
"fetch",
"headerTimeout",
"timeout",
"transport",
].includes(key),
),
)
return Object.keys(result).length === 0 ? undefined : result
@@ -951,7 +971,7 @@ const errorValue = (error: unknown) => (ProviderShared.isRecord(error) ? error.v
// 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([
@@ -969,6 +989,7 @@ const NETWORK_ERROR_MESSAGES = new Set([
"terminated",
"other side closed",
"sse read timed out",
HEADER_TIMEOUT_MESSAGE.toLowerCase(),
])
const NativeErrorShape = Schema.Struct({
+6 -1
View File
@@ -1,7 +1,7 @@
export * as ModelResolver from "./model-resolver.js"
import { makeLocationNode } from "@opencode/util/effect/app-node"
import { LanguageModel, ProviderConfigurationError } from "@opencode/ai"
import { HttpOptions, LanguageModel, mergeHttpOptions, ProviderConfigurationError } from "@opencode/ai"
import { Auth } from "@opencode/ai/route"
import { Context, Effect, Layer, Schema, Struct } from "effect"
import { AISDK } from "./aisdk.js"
@@ -237,6 +237,11 @@ const resolveCatalogModel = Effect.fn("ModelResolver.resolveCatalogModel")(funct
compatibility: resolved.compatibility
? Object.assign({}, runtime.compatibility, resolved.compatibility)
: runtime.compatibility,
// Timeouts are transport policy, so they land on the route's HTTP defaults instead of package settings.
defaults: {
...runtime.defaults,
http: mergeHttpOptions(runtime.defaults?.http, new HttpOptions(Provider.timeouts(configured))),
},
})
},
catch: (cause) =>
+14 -4
View File
@@ -1,10 +1,10 @@
export * as Provider from "./provider.js"
import { Context, Effect, Layer, Schema, Stream, Struct } from "effect"
import { Context, Effect, Layer, Option, Schema, Stream, Struct } from "effect"
import { Provider } from "@opencode/schema/provider"
import { Model } from "@opencode/schema/model"
import { makeLocationNode } from "@opencode/util/effect/app-node"
import type { ProviderPackageDefinition } from "@opencode/ai"
import { HttpTimeout, type ProviderPackageDefinition } from "@opencode/ai"
import { isRecord } from "@opencode/ai/utils/record"
import { Npm } from "@opencode/util/npm"
import type { DeepMutable } from "./schema.js"
@@ -133,8 +133,8 @@ export const loadPackage = Effect.fn("Provider.loadPackage")(function* (input: s
})
/** opencode settings consumed in Core; native packages never receive them. */
const CORE_KEYS = ["chunkTimeout", "compaction", "fetch", "timeout", "transport"] as const
const PROVIDER_ONLY_KEYS = ["chunkTimeout", "timeout", "transport"] as const
const CORE_KEYS = ["chunkTimeout", "compaction", "fetch", "headerTimeout", "timeout", "transport"] as const
const PROVIDER_ONLY_KEYS = ["chunkTimeout", "headerTimeout", "timeout", "transport"] as const
export function nativeSettings(settings: Settings): Settings {
return Struct.omit(settings, CORE_KEYS)
@@ -144,6 +144,16 @@ export function modelSettings(settings: Settings | undefined) {
return settings && Struct.omit(settings, PROVIDER_ONLY_KEYS)
}
const decodeTimeout = Schema.decodeUnknownOption(HttpTimeout)
/** Request timeouts from provider settings; invalid values are dropped so the transport default applies. */
export function timeouts(settings: Readonly<Record<string, unknown>>) {
return {
headerTimeout: Option.getOrUndefined(decodeTimeout(settings.headerTimeout)),
chunkTimeout: Option.getOrUndefined(decodeTimeout(settings.chunkTimeout)),
}
}
export function mergeOverlay(
base: Readonly<Record<string, unknown>> | undefined,
overlay: Readonly<Record<string, unknown>> | undefined,
+1 -2
View File
@@ -271,9 +271,8 @@ export const layer = Layer.effect(
const request = LLMRequest.update(base, {
model: route === base.model.route ? base.model : LanguageModel.update(base.model, { route }),
http: new HttpOptions({
body: base.http?.body,
...base.http,
headers: Object.keys(modelHook.headers).length === 0 ? undefined : modelHook.headers,
query: base.http?.query,
}),
})
// History selects native windows against the catalog route before hooks run. A newly installed
+39
View File
@@ -779,6 +779,45 @@ it.effect("sends AI SDK requests directly when no HTTP hook middleware is attach
}),
)
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)).toBeTrue()
}),
)
it.effect("emits malformed AI SDK tool input without executing it", () =>
Effect.gen(function* () {
const aisdk = yield* AISDK.Service
+19
View File
@@ -287,6 +287,25 @@ 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/openai"), {
settings: {
apiKey: "secret",
baseURL: "https://openai.example/v1",
headerTimeout: false,
chunkTimeout: 60_000,
},
}),
)
const prepared = yield* compileRequest(LLM.request({ model: resolved, prompt: "Hello" }))
expect(resolved.defaults?.http).toMatchObject({ headerTimeout: false, chunkTimeout: 60_000 })
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(
@@ -131,6 +131,28 @@ Both fields can also be set on a model or variant when only part of a provider's
}
```
## Timeouts
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,
},
},
},
}
```
A request that exceeds either limit fails with a transport error and follows the normal retry policy.
## Packages
The `package` field selects the runtime that communicates with a provider. Use the compatible runtime for APIs that