diff --git a/packages/opencode/src/provider/provider.ts b/packages/opencode/src/provider/provider.ts index 867ef2ca2f..97f83bb51d 100644 --- a/packages/opencode/src/provider/provider.ts +++ b/packages/opencode/src/provider/provider.ts @@ -91,6 +91,41 @@ function timeoutController(ms: number) { } } +// Applies the `headerTimeout`, `chunkTimeout` (SSE idle) and `timeout` provider options at the +// fetch layer, on top of `options.fetch` when a custom fetch is configured. +function timeoutFetch(options: Record) { + const customFetch = options["fetch"] + const chunkTimeout = options["chunkTimeout"] ?? 300_000 + const headerTimeout = options["headerTimeout"] ?? 300_000 + const timeout = options["timeout"] + + return async (input: any, init?: BunFetchRequestInit) => { + const fetchFn = customFetch ?? fetch + const opts = init ?? {} + const chunkAbortCtl = typeof chunkTimeout === "number" && chunkTimeout > 0 ? new AbortController() : undefined + const headerTimeoutMs = headerTimeout === false ? undefined : headerTimeout + const headerTimeoutCtl = typeof headerTimeoutMs === "number" ? timeoutController(headerTimeoutMs) : undefined + const signals: AbortSignal[] = [] + + if (opts.signal) signals.push(opts.signal) + if (chunkAbortCtl) signals.push(chunkAbortCtl.signal) + if (headerTimeoutCtl) signals.push(headerTimeoutCtl.signal) + if (timeout !== undefined && timeout !== null && timeout !== false) signals.push(AbortSignal.timeout(timeout)) + + const combined = signals.length === 0 ? null : signals.length === 1 ? signals[0] : AbortSignal.any(signals) + if (combined) opts.signal = combined + + const res = await fetchFn(input, { + ...opts, + // @ts-ignore see here: https://github.com/oven-sh/bun/issues/16682 + timeout: false, + }).finally(() => headerTimeoutCtl?.clear()) + + if (!chunkAbortCtl) return res + return wrapSSE(res, chunkTimeout, chunkAbortCtl) + } +} + function googleVertexAnthropicBaseURL(project: string | undefined, location: string | undefined) { if (!project) return if (location !== "eu" && location !== "us") return @@ -810,7 +845,7 @@ function custom(dep: CustomDep): Record { ) } - const { createAiGateway } = yield* Effect.promise(() => import("ai-gateway-provider")) + const { createAiGateway, parseAiGatewayOptions } = yield* Effect.promise(() => import("ai-gateway-provider")) const { createUnified } = yield* Effect.promise(() => import("ai-gateway-provider/providers/unified")) const { createOpenAI } = yield* Effect.promise(() => import("ai-gateway-provider/providers/openai")) const { createAnthropic } = yield* Effect.promise(() => import("ai-gateway-provider/providers/anthropic")) @@ -835,15 +870,30 @@ function custom(dep: CustomDep): Record { }, } - const aigateway = createAiGateway({ - accountId, - gateway, - apiKey: apiToken, - ...(Object.values(opts).some((v) => v !== undefined) ? { options: opts } : {}), - }) return { autoload: true, - async getModel(_sdk: any, modelID: string, _options?: Record) { + async getModel(_sdk: any, modelID: string, options?: Record) { + // This loader builds its own clients instead of using the SDK from resolveSDK, so the + // timeout options have to be applied to the requests it makes explicitly. + const gatewayFetch = timeoutFetch(options ?? {}) + // ai-gateway-provider's REST path always calls the global fetch. Its binding path hands + // the request to `run`, so send the same request the REST path would (options as + // request-level cf-aig-* headers) through the timeout-aware fetch instead. + const aigateway = createAiGateway({ + binding: { + run(body, init) { + const headers = parseAiGatewayOptions(opts) + headers.set("Content-Type", "application/json") + headers.set("cf-aig-authorization", `Bearer ${apiToken}`) + return gatewayFetch(`https://gateway.ai.cloudflare.com/v1/${accountId}/${gateway}`, { + body: JSON.stringify(body), + headers, + method: "POST", + signal: init?.signal, + }) + }, + }, + }) // Model IDs use Unified API format: provider/model (e.g., "anthropic/claude-sonnet-4-5"). // OpenAI and Anthropic ride their native passthrough routes so agents get the Responses // and Messages APIs; new OpenAI models reject tools+reasoning_effort on chat completions. @@ -877,6 +927,7 @@ function custom(dep: CustomDep): Record { baseURL: `https://api.cloudflare.com/client/v4/accounts/${accountId}/ai/v1`, apiKey: apiToken, headers: { "cf-aig-gateway-id": gateway }, + fetch: gatewayFetch as typeof fetch, })(modelID) }, options: {}, @@ -1795,39 +1846,10 @@ const layer = Layer.effect( const existing = s.sdk.get(key) if (existing) return existing - const customFetch = options["fetch"] - const chunkTimeout = options["chunkTimeout"] ?? 300_000 - const headerTimeout = options["headerTimeout"] ?? 300_000 + options["fetch"] = timeoutFetch(options) delete options["chunkTimeout"] delete options["headerTimeout"] - options["fetch"] = async (input: any, init?: BunFetchRequestInit) => { - const fetchFn = customFetch ?? fetch - const opts = init ?? {} - const chunkAbortCtl = typeof chunkTimeout === "number" && chunkTimeout > 0 ? new AbortController() : undefined - const headerTimeoutMs = headerTimeout === false ? undefined : headerTimeout - const headerTimeoutCtl = typeof headerTimeoutMs === "number" ? timeoutController(headerTimeoutMs) : undefined - const signals: AbortSignal[] = [] - - if (opts.signal) signals.push(opts.signal) - if (chunkAbortCtl) signals.push(chunkAbortCtl.signal) - if (headerTimeoutCtl) signals.push(headerTimeoutCtl.signal) - if (options["timeout"] !== undefined && options["timeout"] !== null && options["timeout"] !== false) - signals.push(AbortSignal.timeout(options["timeout"])) - - const combined = signals.length === 0 ? null : signals.length === 1 ? signals[0] : AbortSignal.any(signals) - if (combined) opts.signal = combined - - const res = await fetchFn(input, { - ...opts, - // @ts-ignore see here: https://github.com/oven-sh/bun/issues/16682 - timeout: false, - }).finally(() => headerTimeoutCtl?.clear()) - - if (!chunkAbortCtl) return res - return wrapSSE(res, chunkTimeout, chunkAbortCtl) - } - const bundledLoader = BUNDLED_PROVIDERS[model.api.npm] if (bundledLoader) { const factory = await bundledLoader() diff --git a/packages/opencode/test/provider/header-timeout.test.ts b/packages/opencode/test/provider/header-timeout.test.ts index 38b884dc7f..3442105aa7 100644 --- a/packages/opencode/test/provider/header-timeout.test.ts +++ b/packages/opencode/test/provider/header-timeout.test.ts @@ -244,6 +244,112 @@ it.live("OpenAI API auth gets default headerTimeout", () => }), ) +const gatewayModels = { + "native passthrough": "anthropic/claude-sonnet-4-6", + "REST catalog": "google/gemini-2.5-flash", +} + +for (const [route, modelID] of Object.entries(gatewayModels)) { + it.live(`cloudflare-ai-gateway ${route} applies chunkTimeout when the SSE body stalls`, () => + Effect.gen(function* () { + yield* provideTmpdirInstance( + () => + Effect.gen(function* () { + const urls = yield* setupGateway(() => + Promise.resolve(new Response(new ReadableStream(), { headers: { "content-type": "text/event-stream" } })), + ) + const provider = yield* Provider.Service + const model = yield* provider.getModel( + ProviderV2.ID.make("cloudflare-ai-gateway"), + ModelV2.ID.make(modelID), + ) + const result = streamText({ + model: yield* provider.getLanguage(model), + onError() {}, + messages: [{ role: "user", content: "hello" }], + }) + + const error = yield* Effect.promise(() => firstStreamError(result.fullStream)) + expect(urls).toHaveLength(1) + expect(error).toBeInstanceOf(ProviderError.ResponseStreamError) + }), + { config: gatewayConfig({ chunkTimeout: 50 }) }, + ) + }), + ) +} + +it.live("cloudflare-ai-gateway applies headerTimeout when response headers do not arrive", () => + Effect.gen(function* () { + yield* provideTmpdirInstance( + () => + Effect.gen(function* () { + const urls = yield* setupGateway( + (init) => + new Promise((_, reject) => init?.signal?.addEventListener("abort", () => reject(init.signal?.reason))), + ) + const provider = yield* Provider.Service + const model = yield* provider.getModel( + ProviderV2.ID.make("cloudflare-ai-gateway"), + ModelV2.ID.make(gatewayModels["native passthrough"]), + ) + const result = streamText({ + model: yield* provider.getLanguage(model), + onError() {}, + messages: [{ role: "user", content: "hello" }], + }) + + const error = yield* Effect.promise(() => firstStreamError(result.fullStream)) + expect(urls).toEqual(["https://gateway.ai.cloudflare.com/v1/test-account/test-gateway"]) + expect(String(error)).toContain("response headers timed out") + }), + { config: gatewayConfig({ headerTimeout: 50 }) }, + ) + }), +) + +// Routes the gateway provider's requests to `respond` through the configured custom fetch, which +// the timeout wrapper calls instead of the global fetch. Returns the requested URLs. +function setupGateway(respond: (init?: RequestInit) => Promise) { + return Effect.gen(function* () { + yield* Env.use.set("CLOUDFLARE_ACCOUNT_ID", "test-account") + yield* Env.use.set("CLOUDFLARE_GATEWAY_ID", "test-gateway") + yield* Env.use.set("CLOUDFLARE_API_TOKEN", "test-token") + const provider = yield* Provider.Service + const configured = yield* provider.getProvider(ProviderV2.ID.make("cloudflare-ai-gateway")) + const urls: string[] = [] + configured.options.fetch = (input: string, init?: RequestInit) => { + urls.push(input) + return respond(init) + } + return urls + }) +} + +function gatewayConfig(options: Record) { + return { + provider: { + "cloudflare-ai-gateway": { + options, + models: { + // The gateway loader builds its own client; a bundled npm keeps resolveSDK off the network. + [gatewayModels["REST catalog"]]: { name: "Gemini 2.5 Flash", provider: { npm: "@ai-sdk/openai-compatible" } }, + }, + }, + }, + } +} + +async function firstStreamError(stream: AsyncIterable<{ type: string; error?: unknown }>) { + try { + for await (const part of stream) { + if (part.type === "error") return part.error + } + } catch (error) { + return error + } +} + function providerConfig(url: string, options: Record = {}) { const config = testProviderConfig(url) return {