fix(tui): add WebSocket heartbeat + auto-reconnect for silent drops (#32997)

tui_gateway/ws.py

@@ -384,6 +384,22 @@ async def handle_ws(ws: Any) -> None:
             # response dict, which we write here from the loop.
             req_id = req.get("id") if isinstance(req, dict) else None
             req_method = req.get("method") if isinstance(req, dict) else None
+
+            if req_method == "gateway.ping":
+                ok = await transport.write_async(
+                    {
+                        "jsonrpc": "2.0",
+                        "result": {"ok": True},
+                        "id": req_id,
+                    }
+                )
+                if not ok:
+                    disconnect_reason = "send_failed_after_heartbeat"
+                    send_failures += 1
+                    _log.warning("ws heartbeat reply send failed peer=%s id=%s", peer, req_id)
+                    break
+                continue
+
             try:
                 resp = await asyncio.to_thread(server.dispatch, req, transport)
             except Exception:

ui-tui/src/__tests__/gatewayClient.test.ts

@@ -97,7 +97,7 @@ const { FakeWebSocket } = vi.hoisted(() => {
 
 vi.mock('undici', () => ({ WebSocket: FakeWebSocket }))
 
-import { GatewayClient } from '../gatewayClient.js'
+import { GatewayClient, RECONNECT_BASE_MS, RECONNECT_MAX_MS, WS_HEARTBEAT_DEAD_MS, WS_HEARTBEAT_INTERVAL_MS } from '../gatewayClient.js'
 
 describe('GatewayClient websocket attach mode', () => {
   const originalWebSocket = globalThis.WebSocket
@@ -493,4 +493,87 @@ describe('GatewayClient websocket attach mode', () => {
 
     gw.kill()
   })
+
+  it('keeps a healthy idle websocket open when heartbeat acknowledgements arrive (issue #32997)', async () => {
+    vi.useFakeTimers()
+    process.env.HERMES_TUI_GATEWAY_URL = 'ws://gateway.test/api/ws?token=abc'
+    const gw = new GatewayClient()
+
+    try {
+      gw.start()
+      const socket = FakeWebSocket.instances[0]!
+
+      socket.open()
+      await vi.advanceTimersByTimeAsync(WS_HEARTBEAT_INTERVAL_MS)
+
+      const heartbeat = JSON.parse(socket.sent.at(-1) ?? '{}') as { id: string; method: string }
+
+      expect(heartbeat.method).toBe('gateway.ping')
+      socket.message(JSON.stringify({ id: heartbeat.id, jsonrpc: '2.0', result: { ok: true } }))
+
+      await vi.advanceTimersByTimeAsync(WS_HEARTBEAT_DEAD_MS + WS_HEARTBEAT_INTERVAL_MS)
+      expect(socket.readyState).toBe(FakeWebSocket.OPEN)
+      expect(FakeWebSocket.instances).toHaveLength(1)
+    } finally {
+      gw.kill()
+      vi.useRealTimers()
+    }
+  })
+
+  it('auto-reconnects after a missing heartbeat acknowledgement (issue #32997)', async () => {
+    vi.useFakeTimers()
+    process.env.HERMES_TUI_GATEWAY_URL = 'ws://gateway.test/api/ws?token=abc'
+    const gw = new GatewayClient()
+
+    try {
+      gw.start()
+      const first = FakeWebSocket.instances[0]!
+
+      first.open()
+      await vi.advanceTimersByTimeAsync(WS_HEARTBEAT_INTERVAL_MS)
+      expect(JSON.parse(first.sent.at(-1) ?? '{}')).toMatchObject({ method: 'gateway.ping' })
+      await vi.advanceTimersByTimeAsync(WS_HEARTBEAT_DEAD_MS + WS_HEARTBEAT_INTERVAL_MS)
+      await vi.advanceTimersByTimeAsync(RECONNECT_BASE_MS)
+      expect(FakeWebSocket.instances.length).toBeGreaterThanOrEqual(2)
+    } finally {
+      gw.kill()
+      vi.useRealTimers()
+    }
+  })
+
+  it('does not double-reconnect when the exit subscriber restarts immediately', async () => {
+    vi.useFakeTimers()
+    process.env.HERMES_TUI_GATEWAY_URL = 'ws://gateway.test/api/ws?token=abc'
+    const gw = new GatewayClient()
+
+    try {
+      gw.on('exit', () => gw.start())
+      gw.start()
+      const first = FakeWebSocket.instances[0]!
+
+      first.open()
+      gw.drain()
+      await Promise.resolve()
+      first.close(1011)
+
+      expect(FakeWebSocket.instances).toHaveLength(2)
+      await vi.advanceTimersByTimeAsync(RECONNECT_BASE_MS)
+      expect(FakeWebSocket.instances).toHaveLength(2)
+    } finally {
+      gw.kill()
+      vi.useRealTimers()
+    }
+  })
+
+  it('does not auto-reconnect after an intentional kill() (issue #32997)', async () => {
+    vi.useFakeTimers()
+    process.env.HERMES_TUI_GATEWAY_URL = 'ws://gateway.test/api/ws?token=abc'
+    const gw = new GatewayClient()
+    gw.start()
+    FakeWebSocket.instances[0]!.open()
+    gw.kill() // sets disposed
+    await vi.advanceTimersByTimeAsync(WS_HEARTBEAT_DEAD_MS + RECONNECT_MAX_MS + 1000)
+    expect(FakeWebSocket.instances.length).toBe(1) // no reconnect attempted
+    vi.useRealTimers()
+  })
 })

ui-tui/src/gatewayClient.ts

@@ -21,6 +21,18 @@ const WS_OPEN = 1
 const WS_CLOSING = 2
 const WS_CLOSED = 3
 
+// Keepalive + dead-connection detection. A silent drop (macOS sleep, proxy
+// idle timeout, VPN reconnect) kills the TCP socket without a `close` event,
+// so the client hangs forever (issue #32997). Browser/undici WebSocket does
+// not expose an acknowledged ping/pong API, so this uses a small JSON-RPC
+// heartbeat that the TUI gateway explicitly answers. Healthy idle sockets stay
+// open; only a missing heartbeat ack forces close -> reconnect.
+export const WS_HEARTBEAT_INTERVAL_MS = 15_000
+export const WS_HEARTBEAT_DEAD_MS = 45_000
+// Exponential backoff for reconnect attempts after a transport drop.
+export const RECONNECT_BASE_MS = 1_000
+export const RECONNECT_MAX_MS = 30_000
+
 const getWebSocketCtor = (): typeof WebSocket =>
   typeof WebSocket === 'undefined' ? (UndiciWebSocket as unknown as typeof WebSocket) : WebSocket
 
@@ -149,6 +161,15 @@ export class GatewayClient extends EventEmitter {
   private drainGeneration = 0
   private stdoutRl: ReturnType<typeof createInterface> | null = null
   private stderrRl: ReturnType<typeof createInterface> | null = null
+  private heartbeatTimer: ReturnType<typeof setInterval> | null = null
+  private reconnectTimer: ReturnType<typeof setTimeout> | null = null
+  private reconnectAttempts = 0
+  private lastActivityAt = 0
+  private heartbeatSeq = 0
+  private heartbeatPendingId: string | null = null
+  private heartbeatSentAt = 0
+  // Set on kill() so we never auto-reconnect after an intentional shutdown.
+  private disposed = false
 
   constructor() {
     super()
@@ -210,6 +231,96 @@ export class GatewayClient extends EventEmitter {
     }
   }
 
+  private startHeartbeat(ws: WebSocket) {
+    this.stopHeartbeat()
+    this.lastActivityAt = Date.now()
+    this.heartbeatPendingId = null
+    this.heartbeatSentAt = 0
+    this.heartbeatTimer = setInterval(() => {
+      if (this.ws !== ws || ws.readyState !== WS_OPEN) {
+        return
+      }
+
+      const now = Date.now()
+
+      if (this.heartbeatPendingId && now - this.heartbeatSentAt > WS_HEARTBEAT_DEAD_MS) {
+        this.lifecycle('[lifecycle] websocket silent drop detected (heartbeat ack timeout); forcing reconnect')
+        this.stopHeartbeat()
+
+        try {
+          ws.close()
+        } catch {
+          // ignore
+        }
+
+        return
+      }
+
+      if (this.heartbeatPendingId) {
+        return
+      }
+
+      const id = `h${++this.heartbeatSeq}`
+
+      this.heartbeatPendingId = id
+      this.heartbeatSentAt = now
+
+      try {
+        ws.send(JSON.stringify({ id, jsonrpc: '2.0', method: 'gateway.ping', params: { last_activity_ms: this.lastActivityAt } }))
+      } catch {
+        this.lifecycle('[lifecycle] websocket heartbeat send failed; forcing reconnect')
+        this.stopHeartbeat()
+
+        try {
+          ws.close()
+        } catch {
+          // ignore
+        }
+      }
+    }, WS_HEARTBEAT_INTERVAL_MS)
+    this.heartbeatTimer.unref?.()
+  }
+
+  private stopHeartbeat() {
+    if (this.heartbeatTimer !== null) {
+      clearInterval(this.heartbeatTimer)
+      this.heartbeatTimer = null
+    }
+
+    this.heartbeatPendingId = null
+    this.heartbeatSentAt = 0
+  }
+
+  private scheduleReconnect() {
+    if (this.disposed || this.reconnectTimer !== null) {
+      return
+    }
+
+    const delay = Math.min(RECONNECT_BASE_MS * 2 ** this.reconnectAttempts, RECONNECT_MAX_MS)
+    this.reconnectAttempts += 1
+    this.lifecycle(`[lifecycle] scheduling gateway reconnect in ${delay}ms (attempt ${this.reconnectAttempts})`)
+    this.publish({ type: 'gateway.reconnecting', payload: { attempt: this.reconnectAttempts, delay_ms: delay } })
+    this.reconnectTimer = setTimeout(() => {
+      this.reconnectTimer = null
+
+      if (this.disposed) {
+        return
+      }
+
+      this.start()
+    }, delay)
+    this.reconnectTimer.unref?.()
+  }
+
+  private clearReconnect() {
+    if (this.reconnectTimer !== null) {
+      clearTimeout(this.reconnectTimer)
+      this.reconnectTimer = null
+    }
+
+    this.reconnectAttempts = 0
+  }
+
   private resetStartupState() {
     // Reject any in-flight RPCs left over from the previous transport
     // before we swap. Otherwise the old transport's stale exit/close
@@ -258,6 +369,14 @@ export class GatewayClient extends EventEmitter {
     this.lifecycle(`[lifecycle] transport exit code=${code ?? 'null'} reason=${reason ?? 'none'}`)
     this.rejectPending(new Error(reason || `gateway exited${code === null ? '' : ` (${code})`}`))
 
+    // Self-heal: a dropped transport (real close OR silent drop caught by the
+    // heartbeat) should reconnect instead of stranding the UI on a dead socket
+    // (issue #32997). Intentional shutdown sets `disposed` and skips this.
+    // Schedule before the synchronous 'exit' emission: useMainApp's existing
+    // recovery subscriber may call start() immediately, and start() cancels this
+    // timer so there is only one recovery owner.
+    this.scheduleReconnect()
+
     if (this.subscribed) {
       this.emit('exit', code)
     } else {
@@ -320,6 +439,7 @@ export class GatewayClient extends EventEmitter {
   }
 
   private handleWebSocketFrame(raw: unknown) {
+    this.lastActivityAt = Date.now()
     const text = asWireText(raw)
 
     if (!text) {
@@ -453,6 +573,9 @@ export class GatewayClient extends EventEmitter {
               resolve()
             }
 
+            this.lastActivityAt = Date.now()
+            this.clearReconnect()
+            this.startHeartbeat(ws)
             this.connectSidecarMirror()
           },
           { once: true }
@@ -504,6 +627,7 @@ export class GatewayClient extends EventEmitter {
         }
 
         this.pushLog(`[lifecycle] websocket close code=${ev.code}`)
+        this.stopHeartbeat()
         this.ws = null
         this.wsConnectPromise = null
         this.handleTransportExit(ev.code, `gateway websocket closed${ev.code ? ` (${ev.code})` : ''}`)
@@ -521,13 +645,18 @@ export class GatewayClient extends EventEmitter {
   }
 
   start() {
+    this.disposed = false
+    this.clearReconnect()
+
     const root = process.env.HERMES_PYTHON_SRC_ROOT ?? resolve(import.meta.dirname, '../../')
     const attachUrl = resolveGatewayAttachUrl()
     const sidecarUrl = resolveSidecarUrl()
 
     this.attachUrl = attachUrl
     this.sidecarUrl = sidecarUrl
     this.resetStartupState()
+    this.clearReconnect()
+    this.stopHeartbeat()
 
     if (this.proc && !this.proc.killed && this.proc.exitCode === null) {
       this.lifecycle(`[lifecycle] replacing live gateway child ${describeChild(this.proc)}`)
@@ -549,6 +678,14 @@ export class GatewayClient extends EventEmitter {
 
   private dispatch(msg: Record<string, unknown>) {
     const id = msg.id as string | undefined
+
+    if (id && id === this.heartbeatPendingId) {
+      this.heartbeatPendingId = null
+      this.heartbeatSentAt = 0
+
+      return
+    }
+
     const p = id ? this.pending.get(id) : undefined
 
     if (p) {
@@ -776,6 +913,9 @@ export class GatewayClient extends EventEmitter {
   }
 
   kill(reason = 'requested') {
+    this.disposed = true
+    this.clearReconnect()
+    this.stopHeartbeat()
     const proc = this.proc
     const killed = proc?.kill()
 

ui-tui/src/gatewayTypes.ts

@@ -642,6 +642,7 @@ export type GatewayEvent =
   | { payload?: { no_speech_limit?: boolean; text?: string }; session_id?: string; type: 'voice.transcript' }
   | { payload?: { reason?: string }; session_id?: string; type: 'dashboard.new_session_requested' }
   | { payload: { line: string }; session_id?: string; type: 'gateway.stderr' }
+  | { payload?: { attempt?: number; delay_ms?: number }; session_id?: string; type: 'gateway.reconnecting' }
   | {
       payload?: { level?: 'info' | 'warn' | 'error'; message?: string }
       session_id?: string