fix(provider): ignore SSE comment heartbeats for chunk timeout

packages/opencode/src/provider/provider.ts

@@ -40,15 +40,28 @@ function wrapSSE(res: Response, ms: number, ctl: AbortController) {
   if (!res.headers.get("content-type")?.includes("text/event-stream")) return res
 
   const reader = res.body.getReader()
+  const decoder = new TextDecoder()
+  let buffer = ""
+  let deadline: number | undefined
+
+  function observedDataEvent(value: Uint8Array) {
+    buffer += decoder.decode(value, { stream: true })
+    const events = buffer.split(/\r?\n\r?\n/)
+    buffer = events.pop() ?? ""
+    return events.some((event) => /^data\s*:/m.test(event))
+  }
+
   const body = new ReadableStream<Uint8Array>({
     async pull(ctrl) {
+      if (deadline === undefined) deadline = Date.now() + ms
       const part = await new Promise<Awaited<ReturnType<typeof reader.read>>>((resolve, reject) => {
+        const remaining = Math.max(0, deadline - Date.now())
         const id = setTimeout(() => {
           const err = new ProviderError.ResponseStreamError("SSE read timed out")
           ctl.abort(err)
           void reader.cancel(err)
           reject(err)
-        }, ms)
+        }, remaining)
 
         reader.read().then(
           (part) => {
@@ -67,6 +80,7 @@ function wrapSSE(res: Response, ms: number, ctl: AbortController) {
         return
       }
 
+      if (observedDataEvent(part.value)) deadline = Date.now() + ms
       ctrl.enqueue(part.value)
     },
     async cancel(reason) {

packages/opencode/test/provider/header-timeout.test.ts

@@ -80,6 +80,36 @@ it.live("chunkTimeout raises a response stream error when SSE body stalls", () =
   }),
 )
 
+it.live("chunkTimeout ignores SSE comment heartbeats", () =>
+  Effect.gen(function* () {
+    const server = yield* Effect.acquireRelease(
+      Effect.promise(() => keepaliveBodyServer(20)),
+      (server) => Effect.sync(() => server.server.close()),
+    )
+
+    yield* provideTmpdirInstance(
+      () =>
+        Effect.gen(function* () {
+          const provider = yield* Provider.Service
+          const model = yield* provider.getModel(ProviderV2.ID.make("test"), ModelV2.ID.make("test-model"))
+          const result = streamText({
+            model: yield* provider.getLanguage(model),
+            onError() {},
+            messages: [{ role: "user", content: "hello" }],
+          })
+
+          const error = yield* Effect.promise(async () => {
+            for await (const part of result.fullStream) {
+              if (part.type === "error") return part.error
+            }
+          })
+          expect(error).toBeInstanceOf(ProviderError.ResponseStreamError)
+        }),
+      { config: providerConfig(server.url, { chunkTimeout: 50 }) },
+    )
+  }),
+)
+
 it.live("headerTimeout aborts when response headers do not arrive", () =>
   Effect.gen(function* () {
     const server = yield* Effect.acquireRelease(
@@ -211,6 +241,19 @@ async function delayedBodyServer(delay: number): Promise<{ server: Server; url:
   return { server, url: `http://127.0.0.1:${address.port}` }
 }
 
+async function keepaliveBodyServer(interval: number): Promise<{ server: Server; url: string }> {
+  const server = createServer((_, res) => {
+    res.writeHead(200, { "content-type": "text/event-stream" })
+    res.write('data: {"choices":[{"delta":{"content":"partial"}}]}\n\n')
+    const keepalive = setInterval(() => res.write(": keepalive\n\n"), interval)
+    res.on("close", () => clearInterval(keepalive))
+  })
+  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve))
+  const address = server.address()
+  if (!address || typeof address === "string") throw new Error("server did not bind to a TCP port")
+  return { server, url: `http://127.0.0.1:${address.port}` }
+}
+
 function withAuthContent<A, E, R>(self: Effect.Effect<A, E, R>, value: Record<string, unknown> = defaultAuthContent()) {
   return Effect.acquireUseRelease(
     Effect.sync(() => {