From f5535dcaf141ab14f8fbc40efb56d30145479b61 Mon Sep 17 00:00:00 2001 From: jwcrystal <121911854+jwcrystal@users.noreply.github.com> Date: Fri, 17 Apr 2026 23:48:39 +0800 Subject: [PATCH] fix: recover from sleep/wake disconnection with connection state tracking and immediate health check (#940) When the computer sleeps and wakes, the SSE/WS event stream drops silently. Messages appeared sent (optimistic insert) but never reached the OpenCode server, and the user had no indication the system was disconnected. Three fixes: 1. Connection state tracking: add onDisconnect callback to the event pipeline. Stream failures set isConnected=false in useConfigStore; successful reconnect sets isConnected=true. 2. Send guard: optimisticSend, respondToPermission, and respondToQuestion now check isConnected before making API calls, throwing a clear error that surfaces as a toast to the user. The /compact command also checks connection with error feedback. 3. Faster server recovery: add triggerHealthCheck() to the server lifecycle and wire it into the WS event stream runtime. When the upstream OpenCode connection fails, the server immediately checks health and restarts if needed, instead of waiting up to 15s for the periodic health check. --- packages/ui/src/components/chat/ChatInput.tsx | 23 ++++++++++++------- packages/ui/src/sync/event-pipeline.ts | 22 ++++++++++++++---- packages/ui/src/sync/session-actions.ts | 11 +++++++++ packages/ui/src/sync/sync-context.tsx | 4 ++++ packages/web/server/index.js | 2 ++ .../web/server/lib/event-stream/runtime.js | 6 +++++ packages/web/server/lib/opencode/lifecycle.js | 20 ++++++++++++++++ .../lib/opencode/startup-pipeline-runtime.js | 2 ++ 8 files changed, 78 insertions(+), 12 deletions(-) diff --git a/packages/ui/src/components/chat/ChatInput.tsx b/packages/ui/src/components/chat/ChatInput.tsx index 2f04f958..6c93cb21 100644 --- a/packages/ui/src/components/chat/ChatInput.tsx +++ b/packages/ui/src/components/chat/ChatInput.tsx @@ -1463,14 +1463,21 @@ const ChatInputComponent: React.FC = ({ onOpenSettings, scrollTo return; } else if (commandName === 'compact' && currentSessionId) { - const { opencodeClient } = await import('@/lib/opencode/client'); - const sdk = opencodeClient.getSdkClient(); - const configState = useConfigStore.getState(); - await sdk.session.summarize({ - sessionID: currentSessionId, - modelID: configState.currentModelId || '', - providerID: configState.currentProviderId || '', - }); + try { + if (!useConfigStore.getState().isConnected) { + throw new Error("Connection lost. Please wait for reconnection."); + } + const { opencodeClient } = await import('@/lib/opencode/client'); + const sdk = opencodeClient.getSdkClient(); + const configState = useConfigStore.getState(); + await sdk.session.summarize({ + sessionID: currentSessionId, + modelID: configState.currentModelId || '', + providerID: configState.currentProviderId || '', + }); + } catch (error) { + toast.error(error instanceof Error ? error.message : 'Failed to compact session'); + } return; } } diff --git a/packages/ui/src/sync/event-pipeline.ts b/packages/ui/src/sync/event-pipeline.ts index 6ebb2e41..7b7a25e1 100644 --- a/packages/ui/src/sync/event-pipeline.ts +++ b/packages/ui/src/sync/event-pipeline.ts @@ -33,6 +33,8 @@ export type EventPipelineInput = { routeDirectory?: (directory: string, payload: Event) => string /** Called after stream reconnects (visibility restore or heartbeat timeout). */ onReconnect?: () => void + /** Called when the stream disconnects (heartbeat timeout, network error, or transport failure). */ + onDisconnect?: () => void transport?: "auto" | "ws" | "sse" } @@ -142,9 +144,10 @@ type DirectoryQueue = { } export function createEventPipeline(input: EventPipelineInput) { - const { sdk, onEvent, onReconnect, routeDirectory, transport = "auto" } = input + const { sdk, onEvent, onReconnect, onDisconnect, routeDirectory, transport = "auto" } = input const abort = new AbortController() let hasConnected = false + let disconnected = false let lastEventId: string | undefined let wsFallbackUntil = 0 @@ -242,6 +245,7 @@ export function createEventPipeline(input: EventPipelineInput) { let heartbeat: ReturnType | undefined const markConnected = () => { + disconnected = false if (hasConnected) { onReconnect?.() return @@ -506,9 +510,19 @@ export function createEventPipeline(input: EventPipelineInput) { const code = typeof error === "object" && error !== null ? (error as { code?: unknown }).code : undefined if (currentTransport === "ws" && code === "WS_FALLBACK") { retryDelayMs = 0 - } else if (!isAbortError(error) && !streamErrorLogged) { - streamErrorLogged = true - console.error("[event-pipeline] stream failed", error) + } else if (!isAbortError(error)) { + if (!streamErrorLogged) { + streamErrorLogged = true + console.error("[event-pipeline] stream failed", error) + } + // Notify consumer that the stream has disconnected, so it can + // update connection state (e.g. set isConnected = false). + // Guard: only fire once per disconnection cycle to avoid repeated + // setState calls on every failed retry attempt. + if (!disconnected) { + disconnected = true + onDisconnect?.() + } } } finally { abort.signal.removeEventListener("abort", onAbort) diff --git a/packages/ui/src/sync/session-actions.ts b/packages/ui/src/sync/session-actions.ts index e4a05890..d3d6a357 100644 --- a/packages/ui/src/sync/session-actions.ts +++ b/packages/ui/src/sync/session-actions.ts @@ -10,6 +10,7 @@ import { useInputStore } from "./input-store" import type { ChildStoreManager } from "./child-store" import { opencodeClient } from "@/lib/opencode/client" import { useGlobalSessionsStore } from "@/stores/useGlobalSessionsStore" +import { useConfigStore } from "@/stores/useConfigStore" import { registerSessionDirectory } from "./sync-refs" // Reference set by SyncProvider — allows actions to access SDK and stores @@ -325,6 +326,10 @@ export async function optimisticSend(input: { throw new Error("Optimistic refs not set — is useSync() mounted?") } + if (!useConfigStore.getState().isConnected) { + throw new Error("Connection lost. Please wait for reconnection.") + } + const store = dirStore() const messageID = ascendingId("msg") const textPartId = ascendingId("prt") @@ -408,6 +413,9 @@ export async function respondToPermission( requestId: string, response: "once" | "always" | "reject", ): Promise { + if (!useConfigStore.getState().isConnected) { + throw new Error("Connection lost. Please wait for reconnection.") + } const result = await getRequestReplyClient("permission", sessionId, requestId).permission.reply({ requestID: requestId, reply: response, @@ -439,6 +447,9 @@ export async function respondToQuestion( requestId: string, answers: string[] | string[][], ): Promise { + if (!useConfigStore.getState().isConnected) { + throw new Error("Connection lost. Please wait for reconnection.") + } const result = await getRequestReplyClient("question", sessionId, requestId).question.reply({ requestID: requestId, answers: answers as Array>, diff --git a/packages/ui/src/sync/sync-context.tsx b/packages/ui/src/sync/sync-context.tsx index 5b41cd99..7a586fa7 100644 --- a/packages/ui/src/sync/sync-context.tsx +++ b/packages/ui/src/sync/sync-context.tsx @@ -1318,6 +1318,7 @@ export function SyncProvider(props: { handleEvent(directory, payload, childStores, routingIndex) }, onReconnect: () => { + useConfigStore.setState({ isConnected: true }) for (const [dir, store] of childStores.children) { if (reconnectResyncing.has(dir)) continue if (getReconnectCandidateSessionIds(store.getState()).length === 0) continue @@ -1332,6 +1333,9 @@ export function SyncProvider(props: { }) } }, + onDisconnect: () => { + useConfigStore.setState({ isConnected: false }) + }, }) return cleanup }, [props.sdk, childStores, routingIndex, messageStreamTransport]) diff --git a/packages/web/server/index.js b/packages/web/server/index.js index 4a5aea38..ec8dcfdf 100644 --- a/packages/web/server/index.js +++ b/packages/web/server/index.js @@ -870,6 +870,7 @@ const waitForOpenCodeReady = (...args) => openCodeLifecycleRuntime.waitForOpenCo const waitForAgentPresence = (...args) => openCodeLifecycleRuntime.waitForAgentPresence(...args); const refreshOpenCodeAfterConfigChange = (...args) => openCodeLifecycleRuntime.refreshOpenCodeAfterConfigChange(...args); const startHealthMonitoring = () => openCodeLifecycleRuntime.startHealthMonitoring(HEALTH_CHECK_INTERVAL); +const triggerHealthCheck = () => openCodeLifecycleRuntime.triggerHealthCheck(); const scheduledTasksRuntime = createScheduledTasksRuntime({ projectConfigRuntime, listProjects: async () => { @@ -1155,6 +1156,7 @@ async function main(options = {}) { setupProxy, scheduleOpenCodeApiDetection, bootstrapOpenCodeAtStartup, + triggerHealthCheck, staticRoutesRuntime, process, crypto, diff --git a/packages/web/server/lib/event-stream/runtime.js b/packages/web/server/lib/event-stream/runtime.js index a25895dc..3eefdea9 100644 --- a/packages/web/server/lib/event-stream/runtime.js +++ b/packages/web/server/lib/event-stream/runtime.js @@ -54,6 +54,7 @@ export function createMessageStreamWsRuntime({ getOpenCodeAuthHeaders, processForwardedEventPayload, wsClients, + triggerHealthCheck, fetchImpl = fetch, }) { const wsServer = new WebSocketServer({ @@ -136,6 +137,10 @@ export function createMessageStreamWsRuntime({ if (!controller.signal.aborted) { sendMessageStreamWsFrame(socket, { type: 'error', message: 'Failed to connect to OpenCode event stream' }); socket.close(1011, 'Failed to connect to OpenCode event stream'); + // Trigger immediate health check so the server detects and + // restarts a dead OpenCode process without waiting for the next + // periodic interval (up to 15 s). + triggerHealthCheck?.(); } return; } @@ -146,6 +151,7 @@ export function createMessageStreamWsRuntime({ message: `OpenCode event stream unavailable (${upstream.status})`, }); socket.close(1011, 'OpenCode event stream unavailable'); + triggerHealthCheck?.(); return; } diff --git a/packages/web/server/lib/opencode/lifecycle.js b/packages/web/server/lib/opencode/lifecycle.js index 9d85d028..a4be1884 100644 --- a/packages/web/server/lib/opencode/lifecycle.js +++ b/packages/web/server/lib/opencode/lifecycle.js @@ -714,6 +714,25 @@ export const createOpenCodeLifecycleRuntime = (deps) => { } }; + /** + * Perform an immediate (one-shot) health check and restart OpenCode if it's + * not healthy. Callers on the SSE / WS proxy path use this to trigger + * recovery without waiting for the next periodic interval (up to 15 s). + */ + const triggerHealthCheck = async () => { + if (!state.openCodeProcess || state.isShuttingDown || state.isRestartingOpenCode) return; + + try { + const healthy = await isOpenCodeProcessHealthy(); + if (!healthy) { + console.log('[lifecycle] immediate health check: OpenCode not healthy, restarting...'); + await restartOpenCode(); + } + } catch (error) { + console.error(`[lifecycle] immediate health check error: ${error.message}`); + } + }; + const startHealthMonitoring = (healthCheckIntervalMs) => { if (state.healthCheckInterval) { clearInterval(state.healthCheckInterval); @@ -743,6 +762,7 @@ export const createOpenCodeLifecycleRuntime = (deps) => { refreshOpenCodeAfterConfigChange, bootstrapOpenCodeAtStartup, startHealthMonitoring, + triggerHealthCheck, waitForPortRelease, }; }; diff --git a/packages/web/server/lib/opencode/startup-pipeline-runtime.js b/packages/web/server/lib/opencode/startup-pipeline-runtime.js index 697f0d3c..5860a3e3 100644 --- a/packages/web/server/lib/opencode/startup-pipeline-runtime.js +++ b/packages/web/server/lib/opencode/startup-pipeline-runtime.js @@ -22,6 +22,7 @@ export const createStartupPipelineRuntime = (dependencies) => { getOpenCodeAuthHeaders, processForwardedEventPayload, messageStreamWsClients, + triggerHealthCheck, terminalHeartbeatIntervalMs, terminalRebindWindowMs, terminalMaxRebindsPerWindow, @@ -76,6 +77,7 @@ export const createStartupPipelineRuntime = (dependencies) => { getOpenCodeAuthHeaders, processForwardedEventPayload, wsClients: messageStreamWsClients, + triggerHealthCheck, }); setupProxy(app);