fix(provider): recover stalled model streams

packages/core/src/v1/config/provider.ts

@@ -114,9 +114,9 @@ export const Info = Schema.Struct({
           description:
             "Timeout in milliseconds to wait for response headers. Provider integrations may set defaults. Set to false to disable timeout.",
         }),
-        chunkTimeout: Schema.optional(PositiveInt).annotate({
+        chunkTimeout: Schema.optional(Schema.Union([Schema.Finite, Schema.Literal(false)])).annotate({
           description:
-            "Timeout in milliseconds between streamed SSE chunks for this provider. If no chunk arrives within this window, the request is aborted.",
+            "Timeout in milliseconds between streamed SSE chunks for this provider. Set to false, zero, or a negative number to disable the timeout.",
         }),
       }),
       [Schema.Record(Schema.String, Schema.Any)],

packages/opencode/src/provider/error.ts

@@ -13,13 +13,23 @@ export class HeaderTimeoutError extends Error {
 }
 
 export class ResponseStreamError extends Error {
-  public override readonly name = "ProviderResponseStreamError"
+  public override readonly name: string = "ProviderResponseStreamError"
 
   constructor(message: string, options?: ErrorOptions) {
     super(message, options)
   }
 }
 
+export class ResponseStreamTimeoutError extends ResponseStreamError {
+  public override readonly name = "ProviderResponseStreamTimeoutError"
+
+  constructor(public readonly ms: number) {
+    super(
+      `Provider response stream timed out after ${ms}ms of inactivity. Check provider connectivity or increase provider.options.chunkTimeout before retrying.`,
+    )
+  }
+}
+
 function isOpenAiErrorRetryable(e: APICallError) {
   const status = e.statusCode
   if (!status) return e.isRetryable

packages/opencode/src/provider/provider.ts

@@ -18,7 +18,7 @@ import { iife } from "@/util/iife"
 import { Global } from "@opencode-ai/core/global"
 import path from "path"
 import { pathToFileURL } from "url"
-import { Effect, Layer, Context, Schema, Types } from "effect"
+import { Effect, Layer, Context, Option, Schema, Types } from "effect"
 import { EffectBridge } from "@/effect/bridge"
 import { InstanceState } from "@/effect/instance-state"
 import { EffectPromise } from "@/effect/promise"
@@ -31,57 +31,10 @@ import { ModelV2 } from "@opencode-ai/core/model"
 import { ModelStatus } from "./model-status"
 import { RuntimeFlags } from "@/effect/runtime-flags"
 import { ProviderError } from "./error"
+import { StreamLiveness } from "./stream-liveness"
 
 const OPENAI_HEADER_TIMEOUT_DEFAULT = 300_000
 
-function wrapSSE(res: Response, ms: number, ctl: AbortController) {
-  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 reader = res.body.getReader()
-  const body = new ReadableStream<Uint8Array>({
-    async pull(ctrl) {
-      const part = await new Promise<Awaited<ReturnType<typeof reader.read>>>((resolve, reject) => {
-        const id = setTimeout(() => {
-          const err = new ProviderError.ResponseStreamError("SSE read timed out")
-          ctl.abort(err)
-          void reader.cancel(err)
-          reject(err)
-        }, ms)
-
-        reader.read().then(
-          (part) => {
-            clearTimeout(id)
-            resolve(part)
-          },
-          (err) => {
-            clearTimeout(id)
-            reject(err)
-          },
-        )
-      })
-
-      if (part.done) {
-        ctrl.close()
-        return
-      }
-
-      ctrl.enqueue(part.value)
-    },
-    async cancel(reason) {
-      ctl.abort(reason)
-      await reader.cancel(reason)
-    },
-  })
-
-  return new Response(body, {
-    headers: new Headers(res.headers),
-    status: res.status,
-    statusText: res.statusText,
-  })
-}
-
 function timeoutController(ms: number) {
   const ctl = new AbortController()
   const id = setTimeout(() => ctl.abort(new ProviderError.HeaderTimeoutError(ms)), ms)
@@ -91,6 +44,14 @@ function timeoutController(ms: number) {
   }
 }
 
+const decodeJson = Schema.decodeUnknownOption(Schema.UnknownFromJsonString)
+
+function streamsResponse(body: BodyInit | null | undefined) {
+  if (typeof body !== "string") return false
+  const value = Option.getOrUndefined(decodeJson(body))
+  return isRecord(value) && value.stream === true
+}
+
 function googleVertexAnthropicBaseURL(project: string | undefined, location: string | undefined) {
   if (!project) return
   if (location !== "eu" && location !== "us") return
@@ -1170,6 +1131,7 @@ interface State {
   sdk: Map<string, BundledSDK>
   modelLoaders: Record<string, CustomModelLoader>
   varsLoaders: Record<string, CustomVarsLoader>
+  streamLiveness: StreamLiveness.Detector
 }
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/Provider") {}
@@ -1357,6 +1319,7 @@ const layer = Layer.effect(
           [providerID: string]: CustomVarsLoader
         } = {}
         const sdk = new Map<string, BundledSDK>()
+        const streamLiveness = StreamLiveness.create()
         const discoveryLoaders: {
           [providerID: string]: CustomDiscoverModels
         } = {}
@@ -1664,6 +1627,7 @@ const layer = Layer.effect(
           sdk,
           modelLoaders,
           varsLoaders,
+          streamLiveness,
         }
       }),
     )
@@ -1735,21 +1699,27 @@ const layer = Layer.effect(
         if (existing) return existing
 
         const customFetch = options["fetch"]
-        const chunkTimeout = options["chunkTimeout"]
+        const configuredChunkTimeout = options["chunkTimeout"]
+        const fixedChunkTimeout =
+          typeof configuredChunkTimeout === "number" && Number.isFinite(configuredChunkTimeout)
+            ? configuredChunkTimeout
+            : undefined
+        const streamTimeoutDisabled =
+          configuredChunkTimeout === false || (fixedChunkTimeout !== undefined && fixedChunkTimeout <= 0)
         const headerTimeout = options["headerTimeout"]
         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 streamAbortController = streamTimeoutDisabled ? undefined : new AbortController()
           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 (streamAbortController) signals.push(streamAbortController.signal)
           if (headerTimeoutCtl) signals.push(headerTimeoutCtl.signal)
           if (options["timeout"] !== undefined && options["timeout"] !== null && options["timeout"] !== false)
             signals.push(AbortSignal.timeout(options["timeout"]))
@@ -1763,8 +1733,14 @@ const layer = Layer.effect(
             timeout: false,
           }).finally(() => headerTimeoutCtl?.clear())
 
-          if (!chunkAbortCtl) return res
-          return wrapSSE(res, chunkTimeout, chunkAbortCtl)
+          if (!streamAbortController) return res
+          return s.streamLiveness.wrap({
+            response: res,
+            bucket: `${model.providerID}:${model.api.npm}`,
+            controller: streamAbortController,
+            timeout: fixedChunkTimeout,
+            stream: streamsResponse(opts.body),
+          })
         }
 
         const bundledLoader = BUNDLED_PROVIDERS[model.api.npm]

packages/opencode/src/provider/stream-liveness.ts

@@ -0,0 +1,111 @@
+import { ProviderError } from "./error"
+
+export type Policy = {
+  initial: number
+  minimum: number
+  maximum: number
+  multiplier: number
+  historySize: number
+}
+
+export const defaultPolicy = {
+  initial: 900_000,
+  minimum: 900_000,
+  maximum: 1_800_000,
+  multiplier: 2,
+  historySize: 32,
+} satisfies Policy
+
+export type Detector = ReturnType<typeof create>
+
+export function create(policy: Policy = defaultPolicy, now = () => performance.now()) {
+  const histories = new Map<string, number[]>()
+
+  function deadline(bucket: string) {
+    const values = histories.get(bucket)
+    if (!values?.length) return policy.initial
+    return Math.min(policy.maximum, Math.max(policy.minimum, Math.max(...values) * policy.multiplier))
+  }
+
+  function observe(bucket: string, elapsed: number) {
+    if (!Number.isFinite(elapsed) || elapsed < 0) return
+    const values = histories.get(bucket) ?? []
+    values.push(elapsed)
+    if (values.length > policy.historySize) values.shift()
+    histories.set(bucket, values)
+  }
+
+  function wrap(input: {
+    response: Response
+    bucket: string
+    controller: AbortController
+    timeout?: number | false
+    stream?: boolean
+  }) {
+    const fixed =
+      typeof input.timeout === "number" && Number.isFinite(input.timeout) ? input.timeout : undefined
+    if (input.timeout === false || (fixed !== undefined && fixed <= 0)) return input.response
+    if (!input.response.body) return input.response
+    if (!input.stream && !input.response.headers.get("content-type")?.includes("text/event-stream")) return input.response
+
+    const ms = fixed ?? deadline(input.bucket)
+    const reader = input.response.body.getReader()
+    let maximum = 0
+    const body = new ReadableStream<Uint8Array>({
+      async pull(controller) {
+        const started = now()
+        const part = await new Promise<Awaited<ReturnType<typeof reader.read>>>((resolve, reject) => {
+          const id = setTimeout(() => {
+            const error = new ProviderError.ResponseStreamTimeoutError(ms)
+            input.controller.abort(error)
+            void reader.cancel(error).catch(() => {})
+            reject(error)
+          }, ms)
+
+          const read = () =>
+            reader.read().then(
+              (part) => {
+                if (!part.done && part.value.byteLength === 0) {
+                  void read()
+                  return
+                }
+                clearTimeout(id)
+                resolve(part)
+              },
+              (error) => {
+                clearTimeout(id)
+                reject(error)
+              },
+            )
+          void read()
+        })
+
+        maximum = Math.max(maximum, now() - started)
+        if (part.done) {
+          observe(input.bucket, maximum)
+          controller.close()
+          return
+        }
+        controller.enqueue(part.value)
+      },
+      async cancel(reason) {
+        input.controller.abort(reason)
+        await reader.cancel(reason)
+      },
+    })
+
+    return new Response(body, {
+      headers: new Headers(input.response.headers),
+      status: input.response.status,
+      statusText: input.response.statusText,
+    })
+  }
+
+  return {
+    deadline,
+    observe,
+    wrap,
+  }
+}
+
+export * as StreamLiveness from "./stream-liveness"

packages/opencode/src/session/message-v2.ts

@@ -665,6 +665,18 @@ export function fromError(
         },
         { cause: e },
       ).toObject()
+    case e instanceof ProviderError.ResponseStreamTimeoutError:
+      return new APIError(
+        {
+          message: e.message,
+          isRetryable: true,
+          metadata: {
+            code: e.name,
+            timeoutMs: String(e.ms),
+          },
+        },
+        { cause: e },
+      ).toObject()
     case e instanceof ProviderError.ResponseStreamError:
       return new APIError(
         {