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

packages/opencode/src/provider/provider.ts

@@ -40,15 +40,29 @@ 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) {
+ const current = deadline ?? Date.now() + ms
+ deadline = current
  const part = await new Promise<Awaited<ReturnType<typeof reader.read>>>((resolve, reject) => {
+ const remaining = Math.max(0, current - 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 +81,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,41 @@ 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 () => {
+ try {
+ for await (const part of result.fullStream) {
+ if (part.type === "error") return part.error
+ }
+ } catch (error) {
+ return error
+ }
+ return undefined
+ })
+ 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 +246,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(() => {