fix(core): ignore SSE comment heartbeats

nix/hashes.json

@@ -1,8 +1,8 @@
 {
   "nodeModules": {
-    "x86_64-linux": "sha256-SBRJwBuBi++5+x8ECufVqMKZMiw9OzF1gr6r/94KpPo=",
-    "aarch64-linux": "sha256-A/FpmriQw8HCwLffkYQ2FEycUE9dWW5PbSzCaiAvaLU=",
-    "aarch64-darwin": "sha256-9YXDArPsKlBPZoFuvjXCFoGd0pmOEDAG6qN6VhdSi64=",
-    "x86_64-darwin": "sha256-m9I3g+UZnwmKflXaZReu9y+4S853wDkq9/zZ2YTEO3I="
+    "x86_64-linux": "sha256-JEqi00PCle+o5OfBlJJaZtXd+4sYB3o+rvYiESlN4dY=",
+    "aarch64-linux": "sha256-zk3Uk1SQyeRrQ7BuFwlOnQAptUHIkq+oPdfd+sTEq5U=",
+    "aarch64-darwin": "sha256-3BOd3EcqimoG3rTI6lTHe91YVlCoEi8/68eT1lbOi0c=",
+    "x86_64-darwin": "sha256-X7wGmjiMloF5Zhuc20kAxLC+tl613YNXRgA+dQjP2WM="
   }
 }

packages/core/src/aisdk.ts

@@ -34,6 +34,7 @@ import {
 import { Auth, Endpoint, RequestExecutor, type AnyRoute } from "@opencode-ai/ai/route"
 import { ProviderShared } from "@opencode-ai/ai/protocols/shared"
 import { Cause, Context, Effect, Layer, Option, Schema, Scope, Stream } from "effect"
+import { makeParser } from "effect/unstable/encoding/Sse"
 import type { ID, Info } from "./model.js"
 import { Provider } from "./provider.js"
 import { State } from "./state.js"
@@ -65,15 +66,23 @@ 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 deadline: number | undefined
+  const parser = makeParser((event) => {
+    if (event._tag === "Event") deadline = Date.now() + ms
+  })
   const body = new ReadableStream<Uint8Array>({
     async pull(ctrl) {
+      const expires = deadline ?? Date.now() + ms
+      deadline = expires
       const part = await new Promise<Awaited<ReturnType<typeof reader.read>>>((resolve, reject) => {
+        const remaining = Math.max(0, expires - Date.now())
         const id = setTimeout(() => {
           const err = new Error("SSE read timed out")
           ctl.abort(err)
           void reader.cancel(err)
           reject(err)
-        }, ms)
+        }, remaining)
 
         reader.read().then(
           (part) => {
@@ -92,6 +101,7 @@ function wrapSSE(res: Response, ms: number, ctl: AbortController) {
         return
       }
 
+      parser.feed(decoder.decode(part.value, { stream: true }))
       ctrl.enqueue(part.value)
     },
     async cancel(reason) {

packages/core/test/aisdk.test.ts

@@ -1,6 +1,7 @@
 import { APICallError } from "@ai-sdk/provider"
 import type { LanguageModelV3, LanguageModelV3StreamPart } from "@ai-sdk/provider"
 import { createMistral } from "@ai-sdk/mistral"
+import { createOpenAICompatible } from "@ai-sdk/openai-compatible"
 import { AISDK } from "@opencode-ai/core/aisdk"
 import { SessionRunnerRetry } from "@opencode-ai/core/session/runner/retry"
 import { toSessionError } from "@opencode-ai/core/session/to-session-error"
@@ -412,6 +413,63 @@ it.effect("moves a tool image through the real Mistral provider as a user messag
   }),
 )
 
+it.effect("does not treat SSE comment heartbeats as model progress", () =>
+  Effect.gen(function* () {
+    const aisdk = yield* AISDK.Service
+    const encoder = new TextEncoder()
+    let heartbeat: ReturnType<typeof setInterval> | undefined
+    const customFetch = Object.assign(
+      async () =>
+        new Response(
+          new ReadableStream({
+            start(controller) {
+              controller.enqueue(
+                encoder.encode(
+                  'data: {"id":"response-1","object":"chat.completion.chunk","created":0,"model":"api-model","choices":[{"index":0,"delta":{"content":"partial"},"finish_reason":null}]}\n\n',
+                ),
+              )
+              heartbeat = setInterval(() => controller.enqueue(encoder.encode(": keepalive\n\n")), 5)
+            },
+            cancel() {
+              if (heartbeat) clearInterval(heartbeat)
+            },
+          }),
+          { headers: { "content-type": "text/event-stream" } },
+        ),
+      { preconnect: fetch.preconnect },
+    )
+    yield* aisdk.hook.sdk((event) => {
+      event.sdk = createOpenAICompatible({
+        ...event.options,
+        name: String(event.options.name),
+        baseURL: String(event.options.baseURL),
+      })
+    })
+    const resolved = yield* aisdk.model(
+      model("@ai-sdk/openai-compatible", {
+        apiKey: "test",
+        baseURL: "https://example.test/v1",
+        chunkTimeout: 25,
+        fetch: customFetch,
+      }),
+    )
+    const result = yield* LLMClient.generate(LLM.request({ model: resolved, prompt: "Hello" })).pipe(
+      Effect.provide(client),
+      Effect.result,
+      Effect.ensuring(
+        Effect.sync(() => {
+          if (heartbeat) clearInterval(heartbeat)
+        }),
+      ),
+    )
+
+    expect(result).toMatchObject({
+      _tag: "Failure",
+      failure: { reason: { message: expect.stringContaining("SSE read timed out") } },
+    })
+  }),
+)
+
 it.effect("emits malformed AI SDK tool input without executing it", () =>
   Effect.gen(function* () {
     const aisdk = yield* AISDK.Service

packages/tui/package.json

@@ -59,11 +59,6 @@
       "node": "./src/attention-sounds.node.ts",
       "default": "./src/attention-sounds.bun.ts"
     },
-    "#terminal-win32": {
-      "bun": "./src/terminal-win32.bun.ts",
-      "node": "./src/terminal-win32.node.ts",
-      "default": "./src/terminal-win32.bun.ts"
-    },
     "#string-width": {
       "bun": "./src/util/string-width.bun.ts",
       "node": "./src/util/string-width.node.ts",

packages/tui/src/app.tsx

@@ -96,7 +96,6 @@ import { CommandPaletteDialog } from "./component/command-palette"
 import { COMMAND_PALETTE_COMMAND, Keymap, type KeymapCommand } from "./context/keymap"
 
 import { DialogVariant } from "./component/dialog-variant"
-import { win32DisableProcessedInput, win32FlushInputBuffer } from "./terminal-win32"
 import { destroyRenderer } from "./util/renderer"
 import { cliErrorMessage, errorFormat } from "./util/error"
 import { AttentionProvider } from "./context/attention"
@@ -266,7 +265,6 @@ export const run = Effect.fn("Tui.run")(function* (input: TuiInput) {
             Effect.catch((error) => Effect.sync(() => log("error", "Failed to dispose TUI clipboard", { error }))),
           ),
       )
-      win32DisableProcessedInput()
       const finalizers = new Set<() => Promise<void>>()
       yield* Effect.addFinalizer(() =>
         Effect.promise(async () => {
@@ -450,7 +448,6 @@ export const run = Effect.fn("Tui.run")(function* (input: TuiInput) {
     }),
   )
   yield* Effect.sync(() => {
-    win32FlushInputBuffer()
     if (result.reason !== undefined)
       process.stderr.write((cliErrorMessage(result.reason) ?? errorFormat(result.reason)) + "\n")
     if (result.epilogue) process.stdout.write(result.epilogue + "\n")

packages/tui/src/terminal-win32.bun.ts

@@ -1,130 +0,0 @@
-import { dlopen, ptr } from "bun:ffi"
-import type { ReadStream } from "node:tty"
-
-const STD_INPUT_HANDLE = -10
-const ENABLE_PROCESSED_INPUT = 0x0001
-
-const kernel = () =>
-  dlopen("kernel32.dll", {
-    GetStdHandle: { args: ["i32"], returns: "ptr" },
-    GetConsoleMode: { args: ["ptr", "ptr"], returns: "i32" },
-    SetConsoleMode: { args: ["ptr", "u32"], returns: "i32" },
-    FlushConsoleInputBuffer: { args: ["ptr"], returns: "i32" },
-  })
-
-let k32: ReturnType<typeof kernel> | undefined
-
-function load() {
-  if (process.platform !== "win32") return false
-  try {
-    k32 ??= kernel()
-    return true
-  } catch {
-    return false
-  }
-}
-
-/**
- * Clear ENABLE_PROCESSED_INPUT on the console stdin handle.
- */
-export function win32DisableProcessedInput() {
-  if (process.platform !== "win32") return
-  if (!process.stdin.isTTY) return
-  if (!load()) return
-
-  const handle = k32!.symbols.GetStdHandle(STD_INPUT_HANDLE)
-  const buf = new Uint32Array(1)
-  if (k32!.symbols.GetConsoleMode(handle, ptr(buf)) === 0) return
-
-  const mode = buf[0]!
-  if ((mode & ENABLE_PROCESSED_INPUT) === 0) return
-  k32!.symbols.SetConsoleMode(handle, mode & ~ENABLE_PROCESSED_INPUT)
-}
-
-/**
- * Discard any queued console input (mouse events, key presses, etc.).
- */
-export function win32FlushInputBuffer() {
-  if (process.platform !== "win32") return
-  if (!process.stdin.isTTY) return
-  if (!load()) return
-
-  const handle = k32!.symbols.GetStdHandle(STD_INPUT_HANDLE)
-  k32!.symbols.FlushConsoleInputBuffer(handle)
-}
-
-let unhook: (() => void) | undefined
-
-/**
- * Keep ENABLE_PROCESSED_INPUT disabled.
- *
- * On Windows, Ctrl+C becomes a CTRL_C_EVENT (instead of stdin input) when
- * ENABLE_PROCESSED_INPUT is set. Various runtimes can re-apply console modes
- * (sometimes on a later tick), and the flag is console-global, not per-process.
- *
- * We combine:
- * - A `setRawMode(...)` hook to re-clear after known raw-mode toggles.
- * - A low-frequency poll as a backstop for native/external mode changes.
- */
-export function win32InstallCtrlCGuard() {
-  if (process.platform !== "win32") return
-  if (!process.stdin.isTTY) return
-  if (!load()) return
-  if (unhook) return unhook
-
-  const stdin = process.stdin as ReadStream
-  const original = stdin.setRawMode
-
-  const handle = k32!.symbols.GetStdHandle(STD_INPUT_HANDLE)
-  const buf = new Uint32Array(1)
-
-  if (k32!.symbols.GetConsoleMode(handle, ptr(buf)) === 0) return
-  const initial = buf[0]!
-
-  const enforce = () => {
-    if (k32!.symbols.GetConsoleMode(handle, ptr(buf)) === 0) return
-    const mode = buf[0]!
-    if ((mode & ENABLE_PROCESSED_INPUT) === 0) return
-    k32!.symbols.SetConsoleMode(handle, mode & ~ENABLE_PROCESSED_INPUT)
-  }
-
-  // Some runtimes can re-apply console modes on the next tick; enforce twice.
-  const later = () => {
-    enforce()
-    setImmediate(enforce)
-  }
-
-  let wrapped: ReadStream["setRawMode"] | undefined
-
-  if (typeof original === "function") {
-    wrapped = (mode: boolean) => {
-      const result = original.call(stdin, mode)
-      later()
-      return result
-    }
-
-    stdin.setRawMode = wrapped
-  }
-
-  // Ensure it's cleared immediately too (covers any earlier mode changes).
-  later()
-
-  const interval = setInterval(enforce, 100)
-  interval.unref()
-
-  let done = false
-  unhook = () => {
-    if (done) return
-    done = true
-
-    clearInterval(interval)
-    if (wrapped && stdin.setRawMode === wrapped) {
-      stdin.setRawMode = original
-    }
-
-    k32!.symbols.SetConsoleMode(handle, initial)
-    unhook = undefined
-  }
-
-  return unhook
-}

packages/tui/src/terminal-win32.node.ts

@@ -1,77 +0,0 @@
-import { dlopen } from "node:ffi"
-import type { ReadStream } from "node:tty"
-
-const STD_INPUT_HANDLE = -10
-const ENABLE_PROCESSED_INPUT = 0x0001
-
-const kernel = () =>
-  dlopen("kernel32.dll", {
-    GetStdHandle: { arguments: ["i32"], return: "pointer" },
-    GetConsoleMode: { arguments: ["pointer", "pointer"], return: "i32" },
-    SetConsoleMode: { arguments: ["pointer", "u32"], return: "i32" },
-    FlushConsoleInputBuffer: { arguments: ["pointer"], return: "i32" },
-  }).functions
-
-let k32: ReturnType<typeof kernel> | undefined
-
-function load() {
-  if (process.platform !== "win32") return false
-  try {
-    k32 ??= kernel()
-    return true
-  } catch {
-    return false
-  }
-}
-
-export function win32DisableProcessedInput() {
-  if (process.platform !== "win32" || !process.stdin.isTTY || !load()) return
-  const handle = k32!.GetStdHandle(STD_INPUT_HANDLE)
-  const buffer = new Uint32Array(1)
-  if (k32!.GetConsoleMode(handle, buffer) === 0) return
-  const mode = buffer[0]!
-  if ((mode & ENABLE_PROCESSED_INPUT) === 0) return
-  k32!.SetConsoleMode(handle, mode & ~ENABLE_PROCESSED_INPUT)
-}
-
-export function win32FlushInputBuffer() {
-  if (process.platform !== "win32" || !process.stdin.isTTY || !load()) return
-  k32!.FlushConsoleInputBuffer(k32!.GetStdHandle(STD_INPUT_HANDLE))
-}
-
-let unhook: (() => void) | undefined
-
-export function win32InstallCtrlCGuard() {
-  if (process.platform !== "win32" || !process.stdin.isTTY || !load() || unhook) return unhook
-  const stdin = process.stdin as ReadStream
-  const original = stdin.setRawMode
-  const handle = k32!.GetStdHandle(STD_INPUT_HANDLE)
-  const buffer = new Uint32Array(1)
-  if (k32!.GetConsoleMode(handle, buffer) === 0) return
-  const initial = buffer[0]!
-  const enforce = () => {
-    if (k32!.GetConsoleMode(handle, buffer) === 0) return
-    const mode = buffer[0]!
-    if ((mode & ENABLE_PROCESSED_INPUT) !== 0) k32!.SetConsoleMode(handle, mode & ~ENABLE_PROCESSED_INPUT)
-  }
-  const later = () => {
-    enforce()
-    setImmediate(enforce)
-  }
-  const wrapped: ReadStream["setRawMode"] = (mode) => {
-    const result = original.call(stdin, mode)
-    later()
-    return result
-  }
-  stdin.setRawMode = wrapped
-  later()
-  const interval = setInterval(enforce, 100)
-  interval.unref()
-  unhook = () => {
-    clearInterval(interval)
-    if (stdin.setRawMode === wrapped) stdin.setRawMode = original
-    k32!.SetConsoleMode(handle, initial)
-    unhook = undefined
-  }
-  return unhook
-}

packages/tui/src/terminal-win32.ts

@@ -1 +0,0 @@
-export { win32DisableProcessedInput, win32FlushInputBuffer, win32InstallCtrlCGuard } from "#terminal-win32"