fix(provider): apply chunkTimeout to non-SSE streaming protocols

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<Uint8Array>({
@@ -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<typeof fetch>[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

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)],
     ),

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<Uint8Array>({
@@ -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]