diff --git a/packages/ai/src/route/transport/http.ts b/packages/ai/src/route/transport/http.ts index c8f94d1a7a1..5e330f61d5c 100644 --- a/packages/ai/src/route/transport/http.ts +++ b/packages/ai/src/route/transport/http.ts @@ -1,11 +1,19 @@ -import { Effect } from "effect" +import { Clock, 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 = TransportPrepareInput @@ -87,17 +95,53 @@ export const httpJson = (input: HttpJsonInput): 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 started = yield* Clock.currentTimeMillis + // Unlike the header and chunk limits, the whole-request budget has no default. + const total = request.http?.timeout ? Duration.millis(request.http.timeout) : Duration.infinity + const response = yield* runtime.http.execute(prepared.request, prepared.middleware).pipe( + Effect.timeoutOrElse({ + duration: Duration.min(timeoutDuration(request.http?.headerTimeout), total), + orElse: () => timeout("request", "Timed out waiting for response headers"), + }), + ) + const http = RequestExecutor.responseHttp(response) + const remaining = Duration.subtract(total, Duration.millis((yield* Clock.currentTimeMillis) - started)) 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)), + }), + Stream.interruptWhen( + Effect.sleep(remaining).pipe( + Effect.andThen(Effect.fail(timeout("read", "Timed out waiting for the response to complete", 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: () => httpJson({ framing: Framing.sse }), diff --git a/packages/ai/src/schema/options.ts b/packages/ai/src/schema/options.ts index 88429a2d9a8..615a4339369 100644 --- a/packages/ai/src/schema/options.ts +++ b/packages/ai/src/schema/options.ts @@ -48,10 +48,23 @@ export const mergeProviderOptions = ( ...items: ReadonlyArray ): 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 + +/** 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("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 the whole request, from send until the response completes. Unbounded when unset. */ + timeout: Schema.optional(HttpTimeout), + /** 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 +83,12 @@ 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 + const headerTimeout = items.findLast((item) => item?.headerTimeout !== undefined)?.headerTimeout + const chunkTimeout = items.findLast((item) => item?.chunkTimeout !== undefined)?.chunkTimeout + if (!body && !headers && !query && timeout === undefined && headerTimeout === undefined && chunkTimeout === undefined) + return undefined + return new HttpOptions({ body, headers, query, timeout, headerTimeout, chunkTimeout }) } export class GenerationOptions extends Schema.Class("LLM.GenerationOptions")({ diff --git a/packages/ai/test/http-timeout.test.ts b/packages/ai/test/http-timeout.test.ts new file mode 100644 index 00000000000..6b45a807007 --- /dev/null +++ b/packages/ai/test/http-timeout.test.ts @@ -0,0 +1,169 @@ +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() + let resume = () => {} + const released = new Promise((resolve) => { + resume = resolve + }) + const encoder = new TextEncoder() + const layer = dynamicResponse((input) => + Effect.sync(() => + input.respond( + new ReadableStream({ + 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("applies a whole-request timeout while the body is still streaming", () => + Effect.gen(function* () { + const server = yield* stalledServer + const fiber = yield* LLMClient.generate( + LLM.request({ model, prompt: "Hello", http: { timeout: 60_000, chunkTimeout: false } }), + ).pipe(Effect.provide(server.layer), Effect.flip, Effect.forkChild({ startImmediately: true })) + yield* Deferred.await(server.stalled) + yield* Effect.yieldNow + yield* TestClock.adjust("1 minute") + const error = yield* Fiber.join(fiber) + + expect(error.reason).toMatchObject({ _tag: "Transport", operation: "read", code: "Timeout" }) + }), + ) + + it.effect("bounds the header wait by a shorter whole-request timeout", () => + Effect.gen(function* () { + const fiber = yield* LLMClient.generate( + LLM.request({ model, prompt: "Hello", http: { timeout: 1_000, headerTimeout: false } }), + ).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("merges timeouts with later values winning", () => + Effect.sync(() => { + const merged = mergeHttpOptions( + new HttpOptions({ timeout: 5_000, headerTimeout: 1_000, chunkTimeout: 2_000 }), + new HttpOptions({ headers: { a: "b" } }), + new HttpOptions({ chunkTimeout: false }), + ) + + expect(merged).toMatchObject({ headers: { a: "b" }, timeout: 5_000, headerTimeout: 1_000, chunkTimeout: false }) + expect(mergeHttpOptions(new HttpOptions({}), undefined)).toBeUndefined() + }), + ) +}) diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 1a450e13981..7341f79490d 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -475,14 +475,16 @@ export type ConfigModelSettings = { compaction?: ProviderCompaction } & { [x: st export type ProviderSettings = { timeout?: number | false - chunkTimeout?: number + headerTimeout?: number | false + chunkTimeout?: number | false compaction?: ProviderCompaction transport?: ProviderTransport } & { [x: string]: any } export type ConfigProviderSettings = { timeout?: number | false - chunkTimeout?: number + headerTimeout?: number | false + chunkTimeout?: number | false compaction?: ProviderCompaction transport?: ProviderTransport } & { [x: string]: JsonValue | null } diff --git a/packages/core/src/aisdk.ts b/packages/core/src/aisdk.ts index cb27a55584c..4b9d5c8e928 100644 --- a/packages/core/src/aisdk.ts +++ b/packages/core/src/aisdk.ts @@ -128,23 +128,29 @@ function prepareOptions(model: RuntimeInfo, pkg: string) { } const customFetch = options.fetch - const chunkTimeout = options.chunkTimeout + const timeouts = Provider.timeouts(options) + delete options.headerTimeout delete options.chunkTimeout delete options.compaction delete options.transport options.fetch = async (input: Parameters[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) + // A plain AbortSignal.timeout would keep running into the body stream, so the header limit uses + // a timer that is cleared once headers arrive. The whole-request limit is meant to span the body. + const headers = new AbortController() + const headerTimer = + timeouts.headerTimeout === false + ? undefined + : setTimeout(() => headers.abort(new Error(HEADER_TIMEOUT_MESSAGE)), timeouts.headerTimeout) + const chunks = new AbortController() + opts.signal = AbortSignal.any( + [ + opts.signal, + headers.signal, + chunks.signal, + timeouts.timeout ? AbortSignal.timeout(timeouts.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 +161,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 (timeouts.chunkTimeout === false) return res + return wrapSSE(res, timeouts.chunkTimeout, chunks) } return options @@ -215,6 +223,8 @@ function throughMiddleware( ) } +const HEADER_TIMEOUT_MESSAGE = "Response headers timed out" + export class InitError extends Schema.TaggedError()("AISDK.InitError", { providerID: Provider.ID, cause: Schema.Defect(), @@ -447,9 +457,17 @@ function requestSettings(settings: Readonly> | 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 +969,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 +987,7 @@ const NETWORK_ERROR_MESSAGES = new Set([ "terminated", "other side closed", "sse read timed out", + HEADER_TIMEOUT_MESSAGE.toLowerCase(), ]) const NativeErrorShape = Schema.Struct({ diff --git a/packages/core/src/model-resolver.ts b/packages/core/src/model-resolver.ts index 56428b0cb5d..63a5cc0f2e1 100644 --- a/packages/core/src/model-resolver.ts +++ b/packages/core/src/model-resolver.ts @@ -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" @@ -122,8 +122,8 @@ export interface Resolved { readonly compaction?: Provider.Compaction /** Provider transport policy; omitted means HTTP. */ readonly transport?: Provider.Transport - /** Milliseconds without streamed data before a WebSocket exchange fails. */ - readonly chunkTimeout?: number + /** Milliseconds without streamed data before a WebSocket exchange fails; `false` disables the limit. */ + readonly chunkTimeout?: number | false } export interface Interface { @@ -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) => @@ -390,7 +395,7 @@ export const layer = Layer.effect( limit: selected.limit, compaction: runtimeInfo.settings?.compaction, transport: provider?.settings?.transport, - chunkTimeout: provider?.settings?.chunkTimeout, + chunkTimeout: Provider.timeout(provider?.settings?.chunkTimeout), } }) return Service.of({ diff --git a/packages/core/src/provider.ts b/packages/core/src/provider.ts index 9204837424d..dc2261735f5 100644 --- a/packages/core/src/provider.ts +++ b/packages/core/src/provider.ts @@ -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" @@ -134,8 +134,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) @@ -145,6 +145,25 @@ export function modelSettings(settings: Settings | undefined) { return settings && Struct.omit(settings, PROVIDER_ONLY_KEYS) } +const decodeTimeout = Schema.decodeUnknownOption(HttpTimeout) + +/** Applied to `headerTimeout` and `chunkTimeout` when provider settings leave them unset or invalid. */ +export const DEFAULT_TIMEOUT_MS = 300_000 + +/** One timeout setting without a default; invalid values are dropped. */ +export function timeout(value: unknown) { + return Option.getOrUndefined(decodeTimeout(value)) +} + +/** Request timeouts from provider settings, resolved for every route. `timeout` has no default. */ +export function timeouts(settings: Readonly>) { + return { + timeout: timeout(settings.timeout), + headerTimeout: timeout(settings.headerTimeout) ?? DEFAULT_TIMEOUT_MS, + chunkTimeout: timeout(settings.chunkTimeout) ?? DEFAULT_TIMEOUT_MS, + } +} + export function mergeOverlay( base: Readonly> | undefined, overlay: Readonly> | undefined, diff --git a/packages/core/src/session/model-request.ts b/packages/core/src/session/model-request.ts index 26cecde25d0..cf6d4ac4dfa 100644 --- a/packages/core/src/session/model-request.ts +++ b/packages/core/src/session/model-request.ts @@ -310,9 +310,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 diff --git a/packages/core/src/session/model-transport.ts b/packages/core/src/session/model-transport.ts index 638ef66af4d..b42b07dc9c2 100644 --- a/packages/core/src/session/model-transport.ts +++ b/packages/core/src/session/model-transport.ts @@ -12,7 +12,7 @@ import { } from "@opencode/ai/route" import { AIError, AIErrorReason, TransportError, type TransportOperation } from "@opencode/ai" import { Hash } from "@opencode/util/hash" -import { Cause, Clock, Context, Effect, Fiber, Layer, Metric, Queue, Scope, Semaphore, Stream } from "effect" +import { Cause, Clock, Context, Duration, Effect, Fiber, Layer, Metric, Queue, Scope, Semaphore, Stream } from "effect" import { Headers } from "effect/unstable/http" import { Socket } from "effect/unstable/socket" import { makeGlobalNode } from "@opencode/util/effect/app-node" @@ -76,7 +76,7 @@ export interface Interface { readonly bind: ( sessionID: SessionSchema.ID, interceptor?: Interceptor, - idleTimeout?: number, + idleTimeout?: number | false, ) => WebSocketChannelExecutor readonly close: (sessionID: SessionSchema.ID) => Effect.Effect readonly closeAll: Effect.Effect @@ -296,7 +296,7 @@ export const makeLayer = (connector: WebSocketConnector) => owner: State, input: WebSocketChannelExchange, interceptor?: Interceptor, - idleTimeout?: number, + idleTimeout?: number | false, ) { if (owner.closed) return yield* transportError("Session WebSocket owner is closed", { @@ -415,7 +415,7 @@ export const makeLayer = (connector: WebSocketConnector) => const token = {} const frames = Stream.fromQueue(active.queue).pipe( Stream.timeoutOrElse({ - duration: idleTimeout ?? IDLE_TIMEOUT, + duration: idleTimeout === false ? Duration.infinity : (idleTimeout ?? IDLE_TIMEOUT), orElse: () => Stream.fail( transportError("Timed out waiting for WebSocket data", { @@ -509,7 +509,7 @@ export const makeLayer = (connector: WebSocketConnector) => const bind = ( sessionID: SessionSchema.ID, interceptor?: Interceptor, - idleTimeout?: number, + idleTimeout?: number | false, ): WebSocketChannelExecutor => ({ execute: (exchange) => { const owner = state(sessionID) diff --git a/packages/core/src/session/runner/retry.ts b/packages/core/src/session/runner/retry.ts index 4885306176e..303201638ad 100644 --- a/packages/core/src/session/runner/retry.ts +++ b/packages/core/src/session/runner/retry.ts @@ -55,10 +55,17 @@ const schedule = Schedule.max([ }), ) +// A timed-out attempt already waited minutes before failing, so the general allowance would let a +// dead provider hold a step for most of an hour. Cap those attempts well below it. +const MAX_TIMEOUT_RETRIES = 3 + +const isTimeout = (error: AIError) => error.reason._tag === "Transport" && error.reason.code === "Timeout" + export const policy = (sessionID: SessionSchema.ID) => Effect.gen(function* () { const step = yield* Schedule.toStep(schedule) let attempt = 1 + let timeouts = 0 return (input: Input) => Effect.gen(function* () { const now = yield* Clock.currentTimeMillis @@ -66,6 +73,7 @@ export const policy = (sessionID: SessionSchema.ID) => if (!next) return { retry: false as const } const [, duration] = next attempt++ + if (isTimeout(input.cause)) timeouts++ const delay = Math.ceil(Duration.toMillis(duration)) const event: PluginHooks.Domains["session"]["retry"] = { sessionID, @@ -73,7 +81,7 @@ export const policy = (sessionID: SessionSchema.ID) => model: input.model, error: input.error, attempt, - decision: input.retry ? { retry: true, delay } : { retry: false }, + decision: input.retry && timeouts <= MAX_TIMEOUT_RETRIES ? { retry: true, delay } : { retry: false }, } yield* input.hook(event) if (!event.decision.retry) return event.decision diff --git a/packages/core/test/aisdk.test.ts b/packages/core/test/aisdk.test.ts index f4a2a2cceaf..b88a8eaaf25 100644 --- a/packages/core/test/aisdk.test.ts +++ b/packages/core/test/aisdk.test.ts @@ -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[0], init?: RequestInit) => + new Promise((_, 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 diff --git a/packages/core/test/session-model-transport.test.ts b/packages/core/test/session-model-transport.test.ts index e38c3fda60e..929079e7b3e 100644 --- a/packages/core/test/session-model-transport.test.ts +++ b/packages/core/test/session-model-transport.test.ts @@ -634,6 +634,36 @@ describe("SessionModelTransport", () => { ) }) + test("disables the idle timeout when chunkTimeout is false", async () => { + const started = Deferred.makeUnsafe() + const messages = queue() + const connector: WebSocketConnector = { + open: () => + Effect.succeed({ + sendText: () => Deferred.succeed(started, undefined), + messages: Stream.fromQueue(messages), + close: Queue.shutdown(messages), + }), + } + + await runWithTestClock( + connector, + Effect.gen(function* () { + const transport = yield* SessionModelTransport.Service + const running = yield* collect(transport.bind(session, undefined, false), exchange("patient")).pipe( + Effect.forkChild({ startImmediately: true }), + ) + yield* Deferred.await(started) + yield* Effect.yieldNow + + yield* TestClock.adjust("2 hours") + yield* Queue.offer(messages, "completed:patient") + + expect(yield* Fiber.join(running)).toEqual(["completed:patient"]) + }), + ) + }) + test("closes a newly opened connection when request creation is interrupted", async () => { const opened = Deferred.makeUnsafe() const messages = queue() diff --git a/packages/core/test/session-runner-retry.test.ts b/packages/core/test/session-runner-retry.test.ts new file mode 100644 index 00000000000..fa549e8246a --- /dev/null +++ b/packages/core/test/session-runner-retry.test.ts @@ -0,0 +1,47 @@ +import { describe, expect } from "bun:test" +import { AIError, ProviderInternalError, TransportError } from "@opencode/ai" +import { Agent } from "@opencode/schema/agent" +import { Model } from "@opencode/schema/model" +import { Provider } from "@opencode/core/provider" +import { SessionSchema } from "@opencode/core/session/schema" +import { SessionRunnerRetry } from "@opencode/core/session/runner/retry" +import { toSessionError } from "@opencode/core/session/to-session-error" +import { Effect } from "effect" +import { it } from "./lib/effect" + +const timeout = new AIError({ + reason: new TransportError({ message: "Timed out", transport: "http", operation: "request", code: "Timeout" }), +}) +const internal = new AIError({ reason: new ProviderInternalError({ message: "internal" }) }) + +const input = (cause: AIError) => ({ + cause, + error: toSessionError(cause), + agent: Agent.ID.make("build"), + model: Model.Ref.make({ id: Model.ID.make("model"), providerID: Provider.ID.make("provider") }), + hook: () => Effect.void, + retry: SessionRunnerRetry.isRetryable(cause), +}) + +// Decisions for `count` consecutive failures of the same cause within one step. +const decisions = (cause: AIError, count: number) => + Effect.gen(function* () { + const policy = yield* SessionRunnerRetry.policy(SessionSchema.ID.make("ses_retry")) + const results: boolean[] = [] + for (let i = 0; i < count; i++) results.push((yield* policy(input(cause))).retry) + return results + }) + +describe("SessionRunnerRetry.policy", () => { + it.effect("stops retrying transport timeouts after three attempts", () => + Effect.gen(function* () { + expect(yield* decisions(timeout, 4)).toEqual([true, true, true, false]) + }), + ) + + it.effect("keeps the general allowance for other transient failures", () => + Effect.gen(function* () { + expect(yield* decisions(internal, 10)).toEqual(Array(10).fill(true)) + }), + ) +}) diff --git a/packages/schema/src/config/provider.ts b/packages/schema/src/config/provider.ts index 1ae813582f5..f110da3513c 100644 --- a/packages/schema/src/config/provider.ts +++ b/packages/schema/src/config/provider.ts @@ -9,7 +9,8 @@ import { optional } from "../schema.js" export const Settings = Schema.StructWithRest( Schema.Struct({ timeout: Schema.Union([Schema.Finite, Schema.Literal(false)]).pipe(optional), - chunkTimeout: Schema.Finite.pipe(optional), + headerTimeout: Schema.Union([Schema.Finite, Schema.Literal(false)]).pipe(optional), + chunkTimeout: Schema.Union([Schema.Finite, Schema.Literal(false)]).pipe(optional), compaction: Provider.Compaction.pipe(optional), transport: Provider.Transport.pipe(optional), }), diff --git a/packages/schema/src/provider.ts b/packages/schema/src/provider.ts index a8493801d9b..9e5ced38765 100644 --- a/packages/schema/src/provider.ts +++ b/packages/schema/src/provider.ts @@ -47,7 +47,8 @@ export type Transport = typeof Transport.Type export const Settings = Schema.StructWithRest( Schema.Struct({ timeout: Schema.Union([Schema.Finite, Schema.Literal(false)]).pipe(optional), - chunkTimeout: Schema.Finite.pipe(optional), + headerTimeout: Schema.Union([Schema.Finite, Schema.Literal(false)]).pipe(optional), + chunkTimeout: Schema.Union([Schema.Finite, Schema.Literal(false)]).pipe(optional), compaction: Compaction.pipe(optional), transport: Transport.pipe(optional), }), diff --git a/packages/schema/test/config.test.ts b/packages/schema/test/config.test.ts index 37c1ecfac0c..7f0f8a61fed 100644 --- a/packages/schema/test/config.test.ts +++ b/packages/schema/test/config.test.ts @@ -27,6 +27,23 @@ describe("Config.Entry", () => { expect(() => Schema.decodeUnknownSync(Config.Info)({ providers: { custom: { canonical: 1 } } })).toThrow() }) + test("round-trips disabled and numeric provider HTTP timeouts", () => { + const input = { + providers: { custom: { settings: { timeout: 900_000, headerTimeout: 600_000, chunkTimeout: false } } }, + } as const + const decoded = Schema.decodeUnknownSync(Config.Info)(input) + expect(Schema.encodeSync(Config.Info)(decoded)).toEqual(input) + expect(Schema.decodeUnknownSync(Provider.Settings)(decoded.providers?.custom?.settings)).toMatchObject({ + timeout: 900_000, + headerTimeout: 600_000, + chunkTimeout: false, + }) + expect(Schema.decodeUnknownSync(Provider.Settings)({ headerTimeout: false, chunkTimeout: 600_000 })).toMatchObject({ + headerTimeout: false, + chunkTimeout: 600_000, + }) + }) + test("accepts disabled, fixed, and random web search selection", () => { const decode = Schema.decodeUnknownSync(Config.Info) diff --git a/services/www/src/docs/content/providers.mdx b/services/www/src/docs/content/providers.mdx index c6ea1b20ea1..ed6797b3438 100644 --- a/services/www/src/docs/content/providers.mdx +++ b/services/www/src/docs/content/providers.mdx @@ -131,6 +131,31 @@ Both fields can also be set on a model or variant when only part of a provider's } ``` +## Timeouts + +HTTP requests wait up to five minutes for response headers and up to five minutes between streamed response chunks. +Set the provider's `settings.headerTimeout` or `settings.chunkTimeout` to a number of milliseconds to change a limit, +or to `false` to disable it. `settings.timeout` additionally bounds the whole request from send until the response +completes; it has no default. These configuration settings apply provider-wide, not per model or variant. On a +WebSocket transport, `chunkTimeout` bounds the gap between frames instead and defaults to thirty minutes. + +```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. OpenCode retries a timed-out request up to three +times before giving up. + ## Packages The `package` field selects the runtime that communicates with a provider. Use the compatible runtime for APIs that