mirror of
https://github.com/anomalyco/opencode.git
synced 2026-11-14 12:10:45 +03:00
fix(core): default provider header and chunk timeouts to five minutes (#49229)
Co-authored-by: rekram1-node <rekram1-node@users.noreply.github.com>
This commit is contained in:
committed by
GitHub
Unverified
parent
ba0eeddff4
commit
4c0d0ff478
@@ -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<Body> = TransportPrepareInput<Body>
|
||||
@@ -87,17 +95,53 @@ 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 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: <Body>() => httpJson<Body, string>({ framing: Framing.sse }),
|
||||
|
||||
@@ -48,10 +48,23 @@ 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 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<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
|
||||
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<GenerationOptions>("LLM.GenerationOptions")({
|
||||
|
||||
@@ -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<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("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()
|
||||
}),
|
||||
)
|
||||
})
|
||||
@@ -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 }
|
||||
|
||||
+40
-21
@@ -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<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)
|
||||
// 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<InitError>()("AISDK.InitError", {
|
||||
providerID: Provider.ID,
|
||||
cause: Schema.Defect(),
|
||||
@@ -447,9 +457,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 +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({
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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<Record<string, unknown>>) {
|
||||
return {
|
||||
timeout: timeout(settings.timeout),
|
||||
headerTimeout: timeout(settings.headerTimeout) ?? DEFAULT_TIMEOUT_MS,
|
||||
chunkTimeout: timeout(settings.chunkTimeout) ?? DEFAULT_TIMEOUT_MS,
|
||||
}
|
||||
}
|
||||
|
||||
export function mergeOverlay(
|
||||
base: Readonly<Record<string, unknown>> | undefined,
|
||||
overlay: Readonly<Record<string, unknown>> | undefined,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<void>
|
||||
readonly closeAll: Effect.Effect<void>
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -634,6 +634,36 @@ describe("SessionModelTransport", () => {
|
||||
)
|
||||
})
|
||||
|
||||
test("disables the idle timeout when chunkTimeout is false", async () => {
|
||||
const started = Deferred.makeUnsafe<void>()
|
||||
const messages = queue<string | Uint8Array, AIError>()
|
||||
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<void>()
|
||||
const messages = queue<string | Uint8Array, AIError>()
|
||||
|
||||
@@ -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))
|
||||
}),
|
||||
)
|
||||
})
|
||||
@@ -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),
|
||||
}),
|
||||
|
||||
@@ -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),
|
||||
}),
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user