diff --git a/packages/core/src/aisdk.ts b/packages/core/src/aisdk.ts index b604dac664b6..de1e33f250a0 100644 --- a/packages/core/src/aisdk.ts +++ b/packages/core/src/aisdk.ts @@ -23,10 +23,12 @@ export interface LanguageEvent { language?: LanguageModelV3 } -function wrapSSE(res: Response, ms: number, ctl: AbortController) { +function wrapSSE(res: Response, ms: number, ctl: AbortController, extraContentTypes: string[] = []) { if (typeof ms !== "number" || ms <= 0) return res if (!res.body) return res - if (!res.headers.get("content-type")?.includes("text/event-stream")) return res + const contentType = res.headers.get("content-type") ?? "" + const chunkedContentTypes = ["text/event-stream", "eventstream", ...extraContentTypes] + if (!chunkedContentTypes.some((type) => contentType.includes(type))) return res const reader = res.body.getReader() const body = new ReadableStream({ @@ -82,6 +84,8 @@ function prepareOptions(model: ModelV2.Info, pkg: string) { const customFetch = options.fetch const chunkTimeout = options.chunkTimeout delete options.chunkTimeout + const chunkTimeoutContentTypes = options.chunkTimeoutContentTypes + delete options.chunkTimeoutContentTypes options.fetch = async (input: Parameters[0], init?: RequestInit) => { const opts = { ...(init ?? {}) } const signals = [ @@ -115,7 +119,7 @@ function prepareOptions(model: ModelV2.Info, pkg: string) { timeout: false, }) if (!chunkAbortCtl || typeof chunkTimeout !== "number") return res - return wrapSSE(res, chunkTimeout, chunkAbortCtl) + return wrapSSE(res, chunkTimeout, chunkAbortCtl, Array.isArray(chunkTimeoutContentTypes) ? chunkTimeoutContentTypes : []) } return options diff --git a/packages/core/src/v1/config/provider.ts b/packages/core/src/v1/config/provider.ts index d54a3f08f926..3fa164fe22a3 100644 --- a/packages/core/src/v1/config/provider.ts +++ b/packages/core/src/v1/config/provider.ts @@ -112,6 +112,10 @@ export const Info = Schema.Struct({ description: "Timeout in milliseconds between streamed SSE chunks for this provider. If no chunk arrives within this window, the request is aborted.", }), + chunkTimeoutContentTypes: Schema.optional(Schema.mutable(Schema.Array(Schema.String))).annotate({ + description: + "Additional response content-type substrings (besides text/event-stream and eventstream) that should be monitored by chunkTimeout. Useful for providers using non-standard streaming content types.", + }), }), [Schema.Record(Schema.String, Schema.Any)], ), diff --git a/packages/opencode/src/provider/provider.ts b/packages/opencode/src/provider/provider.ts index c979fc44659c..e632e0628960 100644 --- a/packages/opencode/src/provider/provider.ts +++ b/packages/opencode/src/provider/provider.ts @@ -34,10 +34,12 @@ import { ProviderError } from "./error" const OPENAI_HEADER_TIMEOUT_DEFAULT = 10_000 -function wrapSSE(res: Response, ms: number, ctl: AbortController) { +function wrapSSE(res: Response, ms: number, ctl: AbortController, extraContentTypes: string[] = []) { if (typeof ms !== "number" || ms <= 0) return res if (!res.body) return res - if (!res.headers.get("content-type")?.includes("text/event-stream")) return res + const contentType = res.headers.get("content-type") ?? "" + const chunkedContentTypes = ["text/event-stream", "eventstream", ...extraContentTypes] + if (!chunkedContentTypes.some((type) => contentType.includes(type))) return res const reader = res.body.getReader() const body = new ReadableStream({ @@ -1709,6 +1711,8 @@ const layer = Layer.effect( const headerTimeout = options["headerTimeout"] delete options["chunkTimeout"] delete options["headerTimeout"] + const chunkTimeoutContentTypes = options["chunkTimeoutContentTypes"] + delete options["chunkTimeoutContentTypes"] options["fetch"] = async (input: any, init?: BunFetchRequestInit) => { const fetchFn = customFetch ?? fetch @@ -1734,7 +1738,7 @@ const layer = Layer.effect( }).finally(() => headerTimeoutCtl?.clear()) if (!chunkAbortCtl) return res - return wrapSSE(res, chunkTimeout, chunkAbortCtl) + return wrapSSE(res, chunkTimeout, chunkAbortCtl, Array.isArray(chunkTimeoutContentTypes) ? chunkTimeoutContentTypes : []) } const bundledLoader = BUNDLED_PROVIDERS[model.api.npm]