fix(client): detect stalled event streams and resync on foreground (#47571)

packages/client/src/promise/client.ts

@@ -9,7 +9,7 @@ export type { ClientOptions, RequestOptions } from "./generated/client.js"
 
 export function make(options: ClientOptions) {
   const raw = OpenCode.make(options)
-  const events = SharedEvents.make((signal) => raw.event.subscribe({ signal }))
+  const events = SharedEvents.make((signal, onActivity) => raw.event.subscribe({ signal, onActivity }))
   return {
     ...raw,
     rpc: Object.assign(makeRpc(raw, events), raw.rpc),

packages/client/src/promise/generated/client.ts

@@ -278,6 +278,8 @@ export interface ClientOptions {
 export interface RequestOptions {
   readonly signal?: AbortSignal
   readonly headers?: RequestInit["headers"]
+  /** Reports every chunk a streaming response receives, including keepalive comments that yield no event. */
+  readonly onActivity?: () => void
 }
 
 interface RequestDescriptor {
@@ -369,6 +371,7 @@ export function make(options: ClientOptions) {
           } catch (cause) {
             throw new ClientError("Transport", { cause })
           }
+          if (!next.done) requestOptions?.onActivity?.()
           buffer += decoder.decode(next.value, { stream: !next.done })
           if (buffer.length > maxSseEventBytes) throw new ClientError("SseEventTooLarge")
           const trailingCarriageReturn = !next.done && buffer.endsWith("\r")

packages/client/src/shared-events.ts

@@ -1,10 +1,19 @@
 export * as SharedEvents from "./shared-events.js"
 
-export function make<A extends { readonly type: string }>(connect: (signal: AbortSignal) => AsyncIterable<A>) {
+export type SubscribeOptions = {
+  readonly signal?: AbortSignal
+  /** Reports transport activity on the shared stream, including keepalive frames that carry no event. */
+  readonly onActivity?: () => void
+}
+
+export function make<A extends { readonly type: string }>(
+  connect: (signal: AbortSignal, onActivity: () => void) => AsyncIterable<A>,
+) {
   type Completion = { readonly error: unknown } | Record<string, never>
   type Subscriber = {
     push: (value: A) => void
     finish: (completion: Completion) => void
+    activity?: () => void
   }
   type Connection = {
     controller: AbortController
@@ -26,7 +35,9 @@ export function make<A extends { readonly type: string }>(connect: (signal: Abor
     let completion: Completion = {}
     try {
       if (connection.controller.signal.aborted) return
-      iterator = connect(connection.controller.signal)[Symbol.asyncIterator]()
+      iterator = connect(connection.controller.signal, () => {
+        connection.subscribers.forEach((subscriber) => subscriber.activity?.())
+      })[Symbol.asyncIterator]()
       while (!connection.controller.signal.aborted) {
         const item = await iterator.next()
         if (item.done || connection.controller.signal.aborted) break
@@ -47,7 +58,7 @@ export function make<A extends { readonly type: string }>(connect: (signal: Abor
   }
 
   return {
-    subscribe(options?: { readonly signal?: AbortSignal }): AsyncIterable<A> {
+    subscribe(options?: SubscribeOptions): AsyncIterable<A> {
       return {
         [Symbol.asyncIterator]() {
           const pending: ReturnType<typeof Promise.withResolvers<IteratorResult<A>>>[] = []
@@ -72,6 +83,7 @@ export function make<A extends { readonly type: string }>(connect: (signal: Abor
           }
 
           const subscriber: Subscriber = {
+            activity: options?.onActivity,
             finish(result) {
               finish(result, false)
             },

packages/client/src/solid/connection.ts

@@ -1,4 +1,4 @@
-import { batch, onCleanup, onMount } from "solid-js"
+import { batch, onCleanup } from "solid-js"
 import { createStore } from "solid-js/store"
 import type { OpenCodeClient, OpenCodeEvent } from "../promise"
 
@@ -18,6 +18,11 @@ export type ClientConnectionOptions = {
   readonly onEvent: (event: OpenCodeEvent) => void
   readonly flushInterval?: number
   readonly pageLifecycle?: boolean
+  /**
+   * Abort and reconnect a stream that receives no bytes for this long. The server writes a keepalive
+   * comment every 15 seconds, so a quiet but healthy stream never trips this.
+   */
+  readonly idleTimeout?: number
   readonly log?: {
     readonly debug?: (message: string, data?: Readonly<Record<string, unknown>>) => void
     readonly info?: (message: string, data?: Readonly<Record<string, unknown>>) => void
@@ -27,10 +32,15 @@ export type ClientConnectionOptions = {
 const connectTimeout = 2_000
 const reconnectDelay = 1_000
 const connectionHistoryLimit = 50
+export const defaultIdleTimeout = 45_000
+// Longer than one server keepalive interval: a stream that is silent this long when the page
+// returns to the foreground is probably half-open after the device slept.
+export const foregroundIdleThreshold = 20_000
 
 export function createClientConnection(initialApi: OpenCodeClient, options: ClientConnectionOptions) {
   const abort = new AbortController()
   const history: ClientConnectionEvent[] = []
+  const idleTimeout = options.idleTimeout ?? defaultIdleTimeout
   const [connection, setConnection] = createStore<{
     status: ClientConnectionStatus
     attempt: number
@@ -40,9 +50,12 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie
   let pending: OpenCodeEvent[] = []
   let flushTimer: ReturnType<typeof setTimeout> | undefined
   let stream: AbortController | undefined
+  let current: AbortController | undefined
   let run: Promise<void> | undefined
   let started = false
   let generation = 0
+  let lastActivity = 0
+  let forced = false
 
   function record(status: ClientConnectionEvent["data"]["status"], attempt: number, error?: string) {
     history.push({ type: "client.connection", created: Date.now(), data: { status, attempt, error } })
@@ -63,14 +76,25 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie
   async function connect(signal: AbortSignal, attempt: number) {
     let connectedAt: number | undefined
     const request = new AbortController()
+    current = request
     const cancel = () => request.abort(signal.reason)
     const timeout = setTimeout(() => request.abort(new Error("Timed out connecting to server")), connectTimeout)
     signal.addEventListener("abort", cancel, { once: true })
 
+    // Any received bytes, including keepalive comments, push the stall deadline out. A timer whose
+    // deadline passed while the page was suspended fires as soon as the page resumes.
+    let watchdog: ReturnType<typeof setTimeout> | undefined
+    const touch = () => {
+      lastActivity = Date.now()
+      if (connectedAt === undefined) return
+      clearTimeout(watchdog)
+      watchdog = setTimeout(() => request.abort(new Error("Event stream stalled")), idleTimeout)
+    }
+
     try {
       record(attempt === 0 ? "connecting" : "reconnecting", attempt)
       options.log?.info?.("event stream connecting", { attempt })
-      const iterator = api.event.subscribe({ signal: request.signal })[Symbol.asyncIterator]()
+      const iterator = api.event.subscribe({ signal: request.signal, onActivity: touch })[Symbol.asyncIterator]()
       const first = await iterator.next()
       if (signal.aborted) return { error: undefined, connectedAt }
       if (first.done)
@@ -85,6 +109,7 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie
       clearTimeout(timeout)
       record("connected", attempt)
       connectedAt = Date.now()
+      touch()
       options.log?.info?.("event stream connected")
       publish(first.value)
       setConnection({ status: "connected", attempt: 0, error: undefined })
@@ -92,7 +117,13 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie
       while (!signal.aborted) {
         const event = await iterator.next()
         if (signal.aborted) return { error: undefined, connectedAt }
-        if (event.done) return { error: new Error("Event stream disconnected"), connectedAt }
+        if (event.done)
+          return {
+            error:
+              request.signal.reason instanceof Error ? request.signal.reason : new Error("Event stream disconnected"),
+            connectedAt,
+          }
+        touch()
         if ("durable" in event.value && event.value.durable)
           options.log?.debug?.("event", {
             type: event.value.type,
@@ -106,7 +137,9 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie
       return { error, connectedAt }
     } finally {
       request.abort()
+      if (current === request) current = undefined
       clearTimeout(timeout)
+      clearTimeout(watchdog)
       signal.removeEventListener("abort", cancel)
     }
   }
@@ -140,6 +173,11 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie
           if (attempt === 1) continue
         }
       }
+      // A deliberate resync already knows the old socket is gone; reconnect without backing off.
+      if (forced) {
+        forced = false
+        continue
+      }
       await wait(reconnectDelay, controller.signal)
     }
   }
@@ -147,6 +185,7 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie
   function start() {
     if (started) return run
     started = true
+    forced = false
     const active = ++generation
     const previous = run
     const current = (async () => {
@@ -161,26 +200,45 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie
   }
 
   function stop() {
+    if (!started) return
     started = false
     generation += 1
     stream?.abort()
+    // Nothing is listening once stopped, so consumers must treat their data as stale until start() reconnects.
+    setConnection({ status: "connecting", attempt: 0, error: undefined })
   }
 
-  onMount(() => {
-    if (options.pageLifecycle) {
-      const pagehide = () => stop()
-      const pageshow = (event: PageTransitionEvent) => {
-        if (event.persisted) void start()
-      }
-      window.addEventListener("pagehide", pagehide)
-      window.addEventListener("pageshow", pageshow)
-      onCleanup(() => {
-        window.removeEventListener("pagehide", pagehide)
-        window.removeEventListener("pageshow", pageshow)
-      })
+  // Drop the live request so the reconnect loop replaces it now instead of waiting for the idle watchdog.
+  function resync(reason: string) {
+    if (!started || connection.status !== "connected") return
+    options.log?.info?.("event stream resync", { reason, idle: Date.now() - lastActivity })
+    forced = true
+    current?.abort(new Error(reason))
+  }
+
+  if (options.pageLifecycle) {
+    const pagehide = () => stop()
+    const pageshow = () => void start()
+    // Locking a phone or switching apps hides the document without a pagehide; the socket usually
+    // dies while the page is suspended, and the browser may never report that on the hung read.
+    const visibility = () => {
+      if (document.visibilityState !== "visible") return
+      if (Date.now() - lastActivity < foregroundIdleThreshold) return
+      resync("Page returned to the foreground after the event stream went quiet")
     }
-    void start()
-  })
+    const online = () => resync("Network connection restored")
+    window.addEventListener("pagehide", pagehide)
+    window.addEventListener("pageshow", pageshow)
+    window.addEventListener("online", online)
+    document.addEventListener("visibilitychange", visibility)
+    onCleanup(() => {
+      window.removeEventListener("pagehide", pagehide)
+      window.removeEventListener("pageshow", pageshow)
+      window.removeEventListener("online", online)
+      document.removeEventListener("visibilitychange", visibility)
+    })
+  }
+  void start()
 
   onCleanup(() => {
     stop()
@@ -195,6 +253,7 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie
     error: () => connection.error,
     internal: {
       history: () => history.slice(),
+      resync,
     },
   }
 }

packages/client/test/promise.test.ts

@@ -675,6 +675,31 @@ test("event.subscribe ignores server heartbeat comments", async () => {
   expect(received).toEqual([event])
 })
 
+test("event.subscribe reports heartbeat comments as stream activity", async () => {
+  const event = { id: "evt_sentinel", created: 1, type: "server.connected", data: {} }
+  const encoder = new TextEncoder()
+  const client = OpenCode.make({
+    baseUrl: "http://localhost:3000",
+    fetch: async () =>
+      new Response(
+        new ReadableStream({
+          start(controller) {
+            controller.enqueue(encoder.encode(": heartbeat\n\n"))
+            controller.enqueue(encoder.encode(`data: ${JSON.stringify(event)}\n\n`))
+            controller.enqueue(encoder.encode(": heartbeat\n\n"))
+            controller.close()
+          },
+        }),
+        { headers: { "content-type": "text/event-stream" } },
+      ),
+  })
+  let activity = 0
+  const received = []
+  for await (const item of client.event.subscribe({ onActivity: () => activity++ })) received.push(item)
+  expect(received).toEqual([event])
+  expect(activity).toBe(3)
+})
+
 // Moved from packages/app/e2e/regression/session-timeline-transport.spec.ts
 test("event transport passes through ordinary health requests", async () => {
   const requests: string[] = []

packages/client/test/shared-events.test.ts

@@ -359,3 +359,24 @@ test("synchronous source creation failures reject subscribers without automatic
   await expect(shared.subscribe()[Symbol.asyncIterator]().next()).rejects.toBe(failure)
   expect(attempts).toHaveLength(2)
 })
+
+test("source activity fans out to every subscriber that asked for it", async () => {
+  const events = source()
+  let activity: (() => void) | undefined
+  const shared = SharedEvents.make<Event>((signal, onActivity) => {
+    activity = onActivity
+    return events.connect(signal)
+  })
+  const counts = { first: 0, second: 0 }
+  const first = shared.subscribe({ onActivity: () => counts.first++ })[Symbol.asyncIterator]()
+  const second = shared.subscribe()[Symbol.asyncIterator]()
+  const reads = Promise.all([first.next(), second.next()])
+  activity!()
+  activity!()
+  events.connections[0].push({ type: "server.connected" })
+  await reads
+  expect(counts).toEqual({ first: 2, second: 0 })
+  await first.return!()
+  await second.return!()
+  await events.connections[0].closed
+})

packages/client/test/solid-connection.test.ts

@@ -0,0 +1,153 @@
+import { expect, test } from "bun:test"
+import { createRoot } from "solid-js"
+import { createClientConnection } from "../src/solid"
+import { OpenCode, type OpenCodeEvent } from "../src/promise"
+
+const connected = { id: "evt_connected", created: 1, type: "server.connected", data: {} }
+
+// One fake server whose event streams stay open until the test writes to them or the client aborts.
+function server() {
+  const encoder = new TextEncoder()
+  const streams: {
+    write: (text: string) => void
+    close: () => void
+    aborted: boolean
+  }[] = []
+  const api = OpenCode.make({
+    baseUrl: "http://opencode.local",
+    fetch: async (input, init) => {
+      const request = input instanceof Request ? input : new Request(input, init)
+      let controller!: ReadableStreamDefaultController<Uint8Array>
+      const entry = {
+        write: (text: string) => controller.enqueue(encoder.encode(text)),
+        close: () => controller.close(),
+        aborted: false,
+      }
+      const body = new ReadableStream<Uint8Array>({
+        start(value) {
+          controller = value
+        },
+        cancel() {
+          entry.aborted = true
+        },
+      })
+      request.signal.addEventListener("abort", () => {
+        entry.aborted = true
+        controller.error(request.signal.reason)
+      })
+      streams.push(entry)
+      return new Response(body, { headers: { "content-type": "text/event-stream" } })
+    },
+  })
+  return { api, streams }
+}
+
+function setup(input: ReturnType<typeof server>, idleTimeout: number) {
+  const events: OpenCodeEvent[] = []
+  return createRoot((dispose) => ({
+    events,
+    dispose,
+    connection: createClientConnection(input.api, {
+      idleTimeout,
+      flushInterval: 0,
+      onEvent: (event) => events.push(event),
+    }),
+  }))
+}
+
+async function until(check: () => boolean, timeout = 2_000) {
+  const deadline = Date.now() + timeout
+  while (!check()) {
+    if (Date.now() > deadline) throw new Error("Timed out waiting for condition")
+    await new Promise((resolve) => setTimeout(resolve, 5))
+  }
+}
+
+test("a stream that goes silent past the idle timeout is replaced", async () => {
+  const fake = server()
+  const ctx = setup(fake, 60)
+  try {
+    await until(() => fake.streams.length === 1)
+    fake.streams[0].write(`data: ${JSON.stringify(connected)}\n\n`)
+    await until(() => ctx.connection.status() === "connected")
+
+    await until(() => fake.streams.length === 2)
+    expect(fake.streams[0].aborted).toBe(true)
+    expect(ctx.connection.internal.history().map((item) => item.data)).toContainEqual({
+      status: "disconnected",
+      attempt: 1,
+      error: "Event stream stalled",
+    })
+
+    fake.streams[1].write(`data: ${JSON.stringify({ ...connected, id: "evt_connected_2" })}\n\n`)
+    await until(() => ctx.connection.status() === "connected" && ctx.events.length === 2)
+    expect(ctx.connection.error()).toBeUndefined()
+  } finally {
+    ctx.dispose()
+  }
+})
+
+test("keepalive comments hold a quiet stream open", async () => {
+  const fake = server()
+  const ctx = setup(fake, 60)
+  try {
+    await until(() => fake.streams.length === 1)
+    fake.streams[0].write(`data: ${JSON.stringify(connected)}\n\n`)
+    await until(() => ctx.connection.status() === "connected")
+
+    const heartbeat = setInterval(() => fake.streams[0].write(": heartbeat\n\n"), 20)
+    await new Promise((resolve) => setTimeout(resolve, 250))
+    clearInterval(heartbeat)
+
+    expect(fake.streams).toHaveLength(1)
+    expect(fake.streams[0].aborted).toBe(false)
+    expect(ctx.connection.status()).toBe("connected")
+    expect(ctx.events).toHaveLength(1)
+  } finally {
+    ctx.dispose()
+  }
+})
+
+test("a forced resync replaces the stream immediately and only while connected", async () => {
+  const fake = server()
+  const ctx = setup(fake, 10_000)
+  try {
+    ctx.connection.internal.resync("too early")
+    await until(() => fake.streams.length === 1)
+    expect(fake.streams[0].aborted).toBe(false)
+    fake.streams[0].write(`data: ${JSON.stringify(connected)}\n\n`)
+    await until(() => ctx.connection.status() === "connected")
+
+    const started = Date.now()
+    ctx.connection.internal.resync("Network connection restored")
+    await until(() => fake.streams.length === 2)
+    expect(Date.now() - started).toBeLessThan(500)
+    expect(fake.streams[0].aborted).toBe(true)
+    expect(ctx.connection.internal.history().map((item) => item.data)).toContainEqual({
+      status: "disconnected",
+      attempt: 1,
+      error: "Network connection restored",
+    })
+    fake.streams[1].write(`data: ${JSON.stringify(connected)}\n\n`)
+    await until(() => ctx.connection.status() === "connected")
+  } finally {
+    ctx.dispose()
+  }
+})
+
+test("a stream the server closes reconnects and reports the disconnect", async () => {
+  const fake = server()
+  const ctx = setup(fake, 10_000)
+  try {
+    await until(() => fake.streams.length === 1)
+    fake.streams[0].write(`data: ${JSON.stringify(connected)}\n\n`)
+    await until(() => ctx.connection.status() === "connected")
+
+    fake.streams[0].close()
+    await until(() => ctx.connection.status() === "reconnecting")
+    expect(ctx.connection.error()).toBe("Event stream disconnected")
+    await until(() => fake.streams.length === 2)
+  } finally {
+    ctx.dispose()
+  }
+})