fix(opencode): apply provider timeouts to Cloudflare AI Gateway models (#51549)
This commit is contained in:
parent
b471c2b449
commit
35fc7a776c
2 changed files with 166 additions and 38 deletions
|
|
@ -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<string, any>) {
|
||||
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<string, CustomLoader> {
|
|||
)
|
||||
}
|
||||
|
||||
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<string, CustomLoader> {
|
|||
},
|
||||
}
|
||||
|
||||
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<string, any>) {
|
||||
async getModel(_sdk: any, modelID: string, options?: Record<string, any>) {
|
||||
// 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<string, CustomLoader> {
|
|||
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()
|
||||
|
|
|
|||
|
|
@ -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<Response>) {
|
||||
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<string, unknown>) {
|
||||
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<string, unknown> = {}) {
|
||||
const config = testProviderConfig(url)
|
||||
return {
|
||||
|
|
|
|||
Loading…
Reference in a new issue