From 2b5c7ff07c66e3821ccf2d302a5b02638d96a2a9 Mon Sep 17 00:00:00 2001 From: Indigo Karasu Date: Tue, 7 Jul 2026 23:18:37 -0700 Subject: [PATCH 1/2] fix(tui): add WebSocket heartbeat + auto-reconnect for silent drops Fixes #32997 The Ink TUI WebSocket client had no heartbeat and no auto-reconnect, so a silent connection drop (macOS sleep, proxy idle timeout, VPN reconnect) left the UI stranded at a dead socket with only Ctrl+C as an escape. Add a periodic ping keepalive plus dead-connection detection that forces a reconnect with exponential backoff; never reconnect after an intentional kill(). Supersedes #30114 (which only restarted on the exit event and could not heal silent drops) by moving reconnect into the transport client and adding the heartbeat. --- ui-tui/src/__tests__/gatewayClient.test.ts | 32 +++++++- ui-tui/src/gatewayClient.ts | 95 ++++++++++++++++++++++ ui-tui/src/gatewayTypes.ts | 1 + 3 files changed, 127 insertions(+), 1 deletion(-) diff --git a/ui-tui/src/__tests__/gatewayClient.test.ts b/ui-tui/src/__tests__/gatewayClient.test.ts index 2a2384b38be8d..ba79505c13411 100644 --- a/ui-tui/src/__tests__/gatewayClient.test.ts +++ b/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,34 @@ describe('GatewayClient websocket attach mode', () => { gw.kill() }) + + it('auto-reconnects after a silent dead connection (issue #32997)', async () => { + vi.useFakeTimers() + process.env.HERMES_TUI_GATEWAY_URL = 'ws://gateway.test/api/ws?token=abc' + const gw = new GatewayClient() + gw.start() + const first = FakeWebSocket.instances[0]! + first.open() + // Advance past one heartbeat tick (no inbound frames -> lastActivityAt stale). + await vi.advanceTimersByTimeAsync(WS_HEARTBEAT_INTERVAL_MS) + // Advance past the dead window -> heartbeat forces close -> handleTransportExit -> reconnect scheduled. + await vi.advanceTimersByTimeAsync(WS_HEARTBEAT_DEAD_MS) + // Advance past the base backoff -> start() opens a new socket. + await vi.advanceTimersByTimeAsync(RECONNECT_BASE_MS) + expect(FakeWebSocket.instances.length).toBeGreaterThanOrEqual(2) + 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() + }) }) diff --git a/ui-tui/src/gatewayClient.ts b/ui-tui/src/gatewayClient.ts index 122b5089de4a5..3c572e2fe0a8b 100644 --- a/ui-tui/src/gatewayClient.ts +++ b/ui-tui/src/gatewayClient.ts @@ -21,6 +21,17 @@ 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). We send a periodic ping to stop +// intermediaries reaping an idle socket AND watch inbound activity; if nothing +// arrives within the dead window we force the socket closed -> 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 +160,12 @@ export class GatewayClient extends EventEmitter { private drainGeneration = 0 private stdoutRl: ReturnType | null = null private stderrRl: ReturnType | null = null + private heartbeatTimer: ReturnType | null = null + private reconnectTimer: ReturnType | null = null + private reconnectAttempts = 0 + private lastActivityAt = 0 + // Set on kill() so we never auto-reconnect after an intentional shutdown. + private disposed = false constructor() { super() @@ -210,6 +227,69 @@ export class GatewayClient extends EventEmitter { } } + private startHeartbeat(ws: WebSocket) { + this.stopHeartbeat() + this.lastActivityAt = Date.now() + this.heartbeatTimer = setInterval(() => { + // Keepalive: nudge the socket so proxies/OS don't reap an idle connection. + // Browser WebSocket has no ping(); undici (Node) does — guard either way. + try { + ;(ws as unknown as { ping?: (data?: string) => void }).ping?.() + } catch { + // best effort + } + + if (Date.now() - this.lastActivityAt > WS_HEARTBEAT_DEAD_MS) { + this.lifecycle('[lifecycle] websocket silent drop detected (no activity); 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 + } + } + + 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 @@ -263,6 +343,11 @@ export class GatewayClient extends EventEmitter { } else { this.pendingExit = 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. + this.scheduleReconnect() } private connectSidecarMirror() { @@ -320,6 +405,7 @@ export class GatewayClient extends EventEmitter { } private handleWebSocketFrame(raw: unknown) { + this.lastActivityAt = Date.now() const text = asWireText(raw) if (!text) { @@ -453,6 +539,9 @@ export class GatewayClient extends EventEmitter { resolve() } + this.lastActivityAt = Date.now() + this.clearReconnect() + this.startHeartbeat(ws) this.connectSidecarMirror() }, { once: true } @@ -506,6 +595,7 @@ export class GatewayClient extends EventEmitter { this.pushLog(`[lifecycle] websocket close code=${ev.code}`) this.ws = null this.wsConnectPromise = null + this.stopHeartbeat() this.handleTransportExit(ev.code, `gateway websocket closed${ev.code ? ` (${ev.code})` : ''}`) }) ws.addEventListener('error', () => { @@ -528,6 +618,8 @@ export class GatewayClient extends EventEmitter { 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)}`) @@ -776,6 +868,9 @@ export class GatewayClient extends EventEmitter { } kill(reason = 'requested') { + this.disposed = true + this.clearReconnect() + this.stopHeartbeat() const proc = this.proc const killed = proc?.kill() diff --git a/ui-tui/src/gatewayTypes.ts b/ui-tui/src/gatewayTypes.ts index ee6b8d78c451e..27069b62d7375 100644 --- a/ui-tui/src/gatewayTypes.ts +++ b/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 From 383a000d6af33dee5b562e558b63014e7fa866d8 Mon Sep 17 00:00:00 2001 From: Indigo Karasu Date: Fri, 10 Jul 2026 08:56:43 -0700 Subject: [PATCH 2/2] fix(tui): acknowledge websocket heartbeats --- tui_gateway/ws.py | 16 +++++ ui-tui/src/__tests__/gatewayClient.test.ts | 79 ++++++++++++++++++---- ui-tui/src/gatewayClient.ts | 79 +++++++++++++++++----- 3 files changed, 144 insertions(+), 30 deletions(-) diff --git a/tui_gateway/ws.py b/tui_gateway/ws.py index 2ab4798df1205..0ed6088f20dfc 100644 --- a/tui_gateway/ws.py +++ b/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: diff --git a/ui-tui/src/__tests__/gatewayClient.test.ts b/ui-tui/src/__tests__/gatewayClient.test.ts index ba79505c13411..8b08b551089be 100644 --- a/ui-tui/src/__tests__/gatewayClient.test.ts +++ b/ui-tui/src/__tests__/gatewayClient.test.ts @@ -494,22 +494,75 @@ describe('GatewayClient websocket attach mode', () => { gw.kill() }) - it('auto-reconnects after a silent dead connection (issue #32997)', async () => { + 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() - gw.start() - const first = FakeWebSocket.instances[0]! - first.open() - // Advance past one heartbeat tick (no inbound frames -> lastActivityAt stale). - await vi.advanceTimersByTimeAsync(WS_HEARTBEAT_INTERVAL_MS) - // Advance past the dead window -> heartbeat forces close -> handleTransportExit -> reconnect scheduled. - await vi.advanceTimersByTimeAsync(WS_HEARTBEAT_DEAD_MS) - // Advance past the base backoff -> start() opens a new socket. - await vi.advanceTimersByTimeAsync(RECONNECT_BASE_MS) - expect(FakeWebSocket.instances.length).toBeGreaterThanOrEqual(2) - gw.kill() - vi.useRealTimers() + + 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 () => { diff --git a/ui-tui/src/gatewayClient.ts b/ui-tui/src/gatewayClient.ts index 3c572e2fe0a8b..06b244afb7f32 100644 --- a/ui-tui/src/gatewayClient.ts +++ b/ui-tui/src/gatewayClient.ts @@ -23,9 +23,10 @@ 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). We send a periodic ping to stop -// intermediaries reaping an idle socket AND watch inbound activity; if nothing -// arrives within the dead window we force the socket closed -> reconnect. +// 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. @@ -164,6 +165,9 @@ export class GatewayClient extends EventEmitter { private reconnectTimer: ReturnType | 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 @@ -230,17 +234,41 @@ export class GatewayClient extends EventEmitter { private startHeartbeat(ws: WebSocket) { this.stopHeartbeat() this.lastActivityAt = Date.now() + this.heartbeatPendingId = null + this.heartbeatSentAt = 0 this.heartbeatTimer = setInterval(() => { - // Keepalive: nudge the socket so proxies/OS don't reap an idle connection. - // Browser WebSocket has no ping(); undici (Node) does — guard either way. - try { - ;(ws as unknown as { ping?: (data?: string) => void }).ping?.() - } catch { - // best effort + if (this.ws !== ws || ws.readyState !== WS_OPEN) { + return } - if (Date.now() - this.lastActivityAt > WS_HEARTBEAT_DEAD_MS) { - this.lifecycle('[lifecycle] websocket silent drop detected (no activity); forcing reconnect') + 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 { @@ -258,6 +286,9 @@ export class GatewayClient extends EventEmitter { clearInterval(this.heartbeatTimer) this.heartbeatTimer = null } + + this.heartbeatPendingId = null + this.heartbeatSentAt = 0 } private scheduleReconnect() { @@ -338,16 +369,19 @@ 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 { this.pendingExit = 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. - this.scheduleReconnect() } private connectSidecarMirror() { @@ -593,9 +627,9 @@ export class GatewayClient extends EventEmitter { } this.pushLog(`[lifecycle] websocket close code=${ev.code}`) + this.stopHeartbeat() this.ws = null this.wsConnectPromise = null - this.stopHeartbeat() this.handleTransportExit(ev.code, `gateway websocket closed${ev.code ? ` (${ev.code})` : ''}`) }) ws.addEventListener('error', () => { @@ -611,6 +645,9 @@ 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() @@ -641,6 +678,14 @@ export class GatewayClient extends EventEmitter { private dispatch(msg: Record) { 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) {