fix(app): recover from a silently dead event stream

packages/app/src/context/server-sdk.test.ts

@@ -1,5 +1,11 @@
 import { describe, expect, test } from "bun:test"
-import { adaptServerEvent, coalesceServerEvents, enqueueServerEvent, resumeStreamAfterPageShow } from "./server-sdk"
+import {
+  adaptServerEvent,
+  coalesceServerEvents,
+  createActivityTrackingFetch,
+  enqueueServerEvent,
+  resumeStreamAfterPageShow,
+} from "./server-sdk"
 import type { OpenCodeEvent } from "@opencode-ai/client/promise"
 import type { Event } from "@opencode-ai/sdk/v2/client"
 
@@ -15,6 +21,44 @@ describe("resumeStreamAfterPageShow", () => {
   })
 })
 
+describe("createActivityTrackingFetch", () => {
+  const stream = (chunks: string[], contentType = "text/event-stream") =>
+    new Response(
+      new ReadableStream<Uint8Array>({
+        start(controller) {
+          for (const chunk of chunks) controller.enqueue(new TextEncoder().encode(chunk))
+          controller.close()
+        },
+      }),
+      { headers: { "content-type": contentType } },
+    )
+
+  test("reports heartbeat frames the SSE parser discards", async () => {
+    let activity = 0
+    const fetch = createActivityTrackingFetch(
+      async () => stream([": heartbeat\n\n", ": heartbeat\n\n"]),
+      () => activity++,
+    )
+
+    const response = await fetch("http://localhost/api/event")
+    expect(await response.text()).toBe(": heartbeat\n\n: heartbeat\n\n")
+    // One on response, one per delivered chunk.
+    expect(activity).toBe(3)
+  })
+
+  test("passes non-stream responses through untouched", async () => {
+    let activity = 0
+    const original = stream(["{}"], "application/json")
+    const fetch = createActivityTrackingFetch(
+      async () => original,
+      () => activity++,
+    )
+
+    expect(await fetch("http://localhost/api/health")).toBe(original)
+    expect(activity).toBe(0)
+  })
+})
+
 describe("adaptServerEvent", () => {
   test("preserves V2 events while adapting permission requests for existing consumers", () => {
     const current = {

packages/app/src/context/server-sdk.tsx

@@ -164,6 +164,47 @@ export function resumeStreamAfterPageShow(event: PageTransitionEvent, start: ()
   start()
 }
 
+// Structural shape of the only thing the SDKs ever call. Narrower than
+// `typeof fetch`, whose runtime-specific statics a wrapper cannot carry over.
+type StreamFetch = (input: URL | RequestInfo, init?: RequestInit) => Promise<Response>
+
+/**
+ * Wraps a fetch so every byte delivered on an event stream is reported back.
+ *
+ * Liveness has to be observed at the byte level rather than at the event level:
+ * the v2 server heartbeats with an SSE comment frame, which the SSE parser
+ * discards without yielding an event, so a quiet-but-healthy stream is
+ * indistinguishable from a dead one to the consumer of the async iterator.
+ */
+export function createActivityTrackingFetch(base: StreamFetch, onActivity: () => void): StreamFetch {
+  return async (input, init) => {
+    const response = await base(input, init)
+    if (!response.body) return response
+    if (!response.headers.get("content-type")?.includes("text/event-stream")) return response
+    onActivity()
+    const reader = response.body.getReader()
+    const body = new ReadableStream<Uint8Array>({
+      async pull(controller) {
+        const next = await reader.read()
+        if (next.done) {
+          controller.close()
+          return
+        }
+        onActivity()
+        controller.enqueue(next.value)
+      },
+      cancel(reason) {
+        return reader.cancel(reason)
+      },
+    })
+    return new Response(body, {
+      headers: response.headers,
+      status: response.status,
+      statusText: response.statusText,
+    })
+  }
+}
+
 type ServerEventEmitter = ReturnType<typeof createGlobalEmitter<{ [key: string]: ServerEvent }>>
 type ServerSDKBase = {
   server: ServerConnection.Any
@@ -199,10 +240,17 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
     }
   })()
 
-  const eventApi = createApiForServer({ server: server.http, fetch: eventFetch })
+  // Updated on every byte the event stream delivers, including heartbeat frames
+  // the SSE parser drops. Read by the per-attempt stall watchdog in start().
+  let activity = 0
+  const streamFetch = createActivityTrackingFetch(eventFetch ?? globalThis.fetch.bind(globalThis), () => {
+    activity = Date.now()
+  }) as typeof fetch
+
+  const eventApi = createApiForServer({ server: server.http, fetch: streamFetch })
   const eventSdk = createSdkForServer({
     signal: abort.signal,
-    fetch: eventFetch,
+    fetch: streamFetch,
     server: server.http,
   })
   const protocol = detectServerProtocol(server.http, platform.fetch ?? globalThis.fetch)
@@ -218,6 +266,12 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
   const FLUSH_FRAME_MS = 16
   const STREAM_YIELD_MS = 8
   const RECONNECT_DELAY_MS = 250
+  // Servers heartbeat every 10s (v1) / 15s (v2), so three missed beats means the
+  // socket is gone even though no error ever surfaced — common when a NAT or
+  // proxy drops the connection, or the host suspends, leaving the read hanging
+  // forever instead of failing.
+  const STREAM_STALL_MS = 45_000
+  const STREAM_STALL_CHECK_MS = 5_000
 
   let queue: Queued[] = []
   let buffer: Queued[] = []
@@ -267,10 +321,18 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
       // oxlint-disable-next-line no-unmodified-loop-condition -- `started` is set to false by stop() which also aborts; both flags are checked to allow graceful exit
       while (!abort.signal.aborted && started && generation === active) {
         attempt = new AbortController()
+        const controller = attempt
         const onAbort = () => {
-          attempt?.abort()
+          controller.abort()
         }
         abort.signal.addEventListener("abort", onAbort)
+        activity = Date.now()
+        const watchdog = setInterval(() => {
+          const idle = Date.now() - activity
+          if (idle < STREAM_STALL_MS) return
+          console.warn("[global-sdk] event stream stalled, reconnecting", { url: server.http.url, idle })
+          controller.abort()
+        }, STREAM_STALL_CHECK_MS)
         try {
           const kind = await protocol
           const events =
@@ -300,6 +362,7 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
             })
           }
         } finally {
+          clearInterval(watchdog)
           abort.signal.removeEventListener("abort", onAbort)
           attempt = undefined
         }

packages/app/src/context/server-sync.test.ts

@@ -10,7 +10,13 @@ import type {
 import { QueryClient } from "@tanstack/solid-query"
 import { canDisposeDirectory, pickDirectoriesToEvict } from "./global-sync/eviction"
 import { estimateRootSessionTotal, loadRootSessions } from "./global-sync/session-load"
-import { loadActiveSessionsQuery, loadMcpQuery, loadMcpResourcesQuery, seedActiveSessionStatuses } from "./server-sync"
+import {
+  loadActiveSessionsQuery,
+  loadMcpQuery,
+  loadMcpResourcesQuery,
+  reconcileActiveSessionStatuses,
+  seedActiveSessionStatuses,
+} from "./server-sync"
 import { ServerScope } from "@/utils/server-scope"
 import { createServerSession } from "./server-session"
 import type { ServerApi } from "@/utils/server"
@@ -100,6 +106,59 @@ describe("active session query", () => {
       next: 10,
     })
   })
+
+  test("clears statuses for runs that finished while the stream was down", () => {
+    const session = createServerSession({} as OpencodeClient)
+    session.set("session_status", "ses_done", { type: "busy" })
+    session.set("session_status", "ses_still_running", { type: "busy" })
+    session.set("session_status", "ses_idle", { type: "idle" })
+
+    reconcileActiveSessionStatuses(Object.assign(session, { sync: async () => undefined }), {
+      ses_still_running: { type: "running" },
+    })
+
+    expect(session.data.session_status.ses_done).toEqual({ type: "idle" })
+    expect(session.data.session_status.ses_still_running).toEqual({ type: "busy" })
+  })
+
+  test("reloads every session holding messages, and only those", () => {
+    const session = createServerSession({} as OpencodeClient)
+    const synced: string[] = []
+    session.set("session_status", "ses_running", { type: "busy" })
+    session.set("message", "ses_running", [])
+    // ses_unopened is active but holds no messages, so it loads on demand.
+
+    const resync = reconcileActiveSessionStatuses(
+      Object.assign(session, {
+        sync: async (sessionID: string) => void synced.push(sessionID),
+      }),
+      { ses_running: { type: "running" }, ses_unopened: { type: "running" } },
+    )
+
+    expect(session.data.session_status.ses_running).toEqual({ type: "busy" })
+    expect(resync).toEqual(["ses_running"])
+    expect(synced).toEqual(["ses_running"])
+  })
+
+  test("reloads a finished session whose status a racing bootstrap already reset", () => {
+    const session = createServerSession({} as OpencodeClient)
+    const synced: string[] = []
+    // A concurrent directory bootstrap rewrites session_status from the server
+    // first, so nothing here still reads busy - but the timeline is frozen
+    // mid-message and must be reloaded anyway.
+    session.set("session_status", "ses_done", { type: "idle" })
+    session.set("message", "ses_done", [])
+
+    const resync = reconcileActiveSessionStatuses(
+      Object.assign(session, {
+        sync: async (sessionID: string) => void synced.push(sessionID),
+      }),
+      {},
+    )
+
+    expect(resync).toEqual(["ses_done"])
+    expect(synced).toEqual(["ses_done"])
+  })
 })
 
 describe("pickDirectoriesToEvict", () => {

packages/app/src/context/server-sync.tsx

@@ -176,6 +176,40 @@ export function seedActiveSessionStatuses(
   }
 }
 
+/**
+ * Repairs state that drifted while the event stream was down.
+ *
+ * Status is otherwise driven purely by `session.execution.*` events, so a run
+ * that finished during the outage stays busy forever — a spinner with nothing
+ * behind it. A run that kept going lost its deltas, so its timeline is frozen
+ * part-way through a message. Both need the server to be re-read.
+ *
+ * Unlike {@link seedActiveSessionStatuses} this overwrites what events wrote,
+ * so it is only safe once the server has been asked what is actually running.
+ */
+export function reconcileActiveSessionStatuses(
+  session: Pick<ServerSession, "data" | "set" | "sync">,
+  active: SessionActiveOutput | Record<string, SessionStatus>,
+) {
+  for (const [sessionID, status] of Object.entries(session.data.session_status)) {
+    if (!status || status.type === "idle") continue
+    if (active[sessionID] !== undefined) continue
+    session.set("session_status", sessionID, { type: "idle" })
+  }
+  // Reload by what is held locally, never by what the status now reads. A
+  // directory bootstrap racing this one also rewrites session_status from the
+  // server, so by the time this runs a finished session may already read idle
+  // and would look like it needs nothing — while its timeline is still frozen
+  // mid-message. Any session holding messages may have missed events; the rest
+  // load on demand when they are opened.
+  const resync = Object.keys(session.data.message)
+  for (const sessionID of resync)
+    void session.sync(sessionID, { force: true }).catch((err) => {
+      console.warn("[server-sync] failed to resync session after reconnect", { sessionID, err })
+    })
+  return resync
+}
+
 function makeQueryOptionsApi(
   scope: ServerScope,
   serverSDK: () => OpencodeClient,
@@ -243,18 +277,21 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) {
       active: async () => {
         if ((await serverSDK.protocol) === "v1") {
           const statuses = (await serverSDK.client.session.status()).data ?? {}
-          seedActiveSessionStatuses(session, statuses)
-          for (const sessionID of Object.keys(statuses)) {
-            void session.resolve(sessionID).catch(() => undefined)
-          }
-          return Object.fromEntries(
+          const running = Object.fromEntries(
             Object.entries(statuses).flatMap(([sessionID, status]) =>
               status.type === "idle" ? [] : [[sessionID, { type: "running" as const }]],
             ),
           )
+          seedActiveSessionStatuses(session, statuses)
+          reconcileActiveSessionStatuses(session, running)
+          for (const sessionID of Object.keys(statuses)) {
+            void session.resolve(sessionID).catch(() => undefined)
+          }
+          return running
         }
         const active = await serverSDK.api.session.active()
         seedActiveSessionStatuses(session, active)
+        reconcileActiveSessionStatuses(session, active)
         for (const sessionID of Object.keys(active)) {
           void session.resolve(sessionID).catch(() => undefined)
         }
@@ -293,6 +330,7 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) {
 
   let bootedAt = 0
   let bootingRoot = false
+  let connected = false
   let eventFrame: number | undefined
   let eventTimer: ReturnType<typeof setTimeout> | undefined
 
@@ -539,8 +577,15 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) {
     homeSessions.refresh(event.type)
 
     if (directory === "global") {
-      if (eventType === "server.connected" && activeSessionsQuery.data === undefined && !activeSessionsQuery.isFetching)
-        void activeSessionsQuery.refetch()
+      if (eventType === "server.connected") {
+        // Every connect after the first one means the stream had dropped, so the
+        // statuses and timelines held locally may have drifted while it was down
+        // and have to be reconciled against the server.
+        const reconnected = connected
+        connected = true
+        if (!activeSessionsQuery.isFetching && (reconnected || activeSessionsQuery.data === undefined))
+          void activeSessionsQuery.refetch()
+      }
       applyGlobalEvent({
         event,
         project: globalStore.project,