diff --git a/packages/desktop/src-tauri/src/main.rs b/packages/desktop/src-tauri/src/main.rs index 0a24b794..cb62c112 100644 --- a/packages/desktop/src-tauri/src/main.rs +++ b/packages/desktop/src-tauri/src/main.rs @@ -491,6 +491,99 @@ fn main() { let _ = app_handle.emit("openchamber:runtime-ready", ()); }); + // Sidecar watchdog: restart on unexpected exit and notify UI + { + let app_handle = app.app_handle().clone(); + let runtime = runtime.clone(); + tauri::async_runtime::spawn(async move { + let mut backoff_ms: u64 = 1000; + loop { + if runtime.opencode_manager().is_shutting_down() { + break; + } + + let mut sleep_ms = backoff_ms; + + match runtime.opencode_manager().is_child_running().await { + Ok(true) => { + sleep_ms = 1000; + backoff_ms = 1000; + } + Ok(false) => { + let _ = app_handle.emit("server.instance.disposed", ()); + if runtime.opencode_manager().is_cli_available() { + if let Err(err) = runtime.opencode_manager().ensure_running().await { + warn!("[desktop:watchdog] Failed to restart OpenCode: {err}"); + } else { + backoff_ms = 1000; + } + } + } + Err(err) => { + warn!("[desktop:watchdog] Failed to check child status: {err}"); + } + } + + tokio::time::sleep(Duration::from_millis(sleep_ms)).await; + backoff_ms = (backoff_ms * 2).min(8000); + } + }); + } + + // Health and wake monitor: emit health and port updates to webview + { + let app_handle = app.app_handle().clone(); + let runtime = runtime.clone(); + tauri::async_runtime::spawn(async move { + #[derive(Clone, Serialize)] + struct HealthSnapshot { + ok: bool, + port: Option, + api_prefix: String, + cli_available: bool, + } + + let mut last_snapshot: Option = None; + let mut last_tick = Instant::now(); + + loop { + if runtime.opencode_manager().is_shutting_down() { + break; + } + + let now = Instant::now(); + let gap_ms = now.saturating_duration_since(last_tick).as_millis() as u64; + last_tick = now; + + let snapshot = HealthSnapshot { + ok: runtime.opencode_manager().is_ready(), + port: runtime.opencode_manager().current_port(), + api_prefix: runtime.opencode_manager().api_prefix(), + cli_available: opencode_manager::check_cli_exists(), + }; + + let changed = match &last_snapshot { + Some(prev) => prev.ok != snapshot.ok + || prev.port != snapshot.port + || prev.api_prefix != snapshot.api_prefix + || prev.cli_available != snapshot.cli_available, + None => true, + }; + + if changed { + let _ = app_handle.emit("openchamber:health-changed", &snapshot); + last_snapshot = Some(snapshot.clone()); + } + + if gap_ms > 15000 { + let _ = app_handle.emit("openchamber:wake", ()); + } + + tokio::time::sleep(Duration::from_secs(5)).await; + } + }); + } + spawn_assistant_notifications(app.app_handle().clone(), runtime.clone()); spawn_session_activity_tracker(app.app_handle().clone(), runtime.clone()); diff --git a/packages/desktop/src-tauri/src/opencode_manager.rs b/packages/desktop/src-tauri/src/opencode_manager.rs index 7d701500..c35aa7eb 100644 --- a/packages/desktop/src-tauri/src/opencode_manager.rs +++ b/packages/desktop/src-tauri/src/opencode_manager.rs @@ -233,6 +233,25 @@ impl OpenCodeManager { self.is_ready.load(Ordering::SeqCst) } + pub fn is_shutting_down(&self) -> bool { + self.shutting_down.load(Ordering::SeqCst) + } + + pub async fn is_child_running(&self) -> Result { + let mut guard = self.child.lock().await; + if let Some(child) = guard.as_mut() { + match child.try_wait()? { + None => return Ok(true), + Some(_status) => { + *guard = None; + self.is_ready.store(false, Ordering::SeqCst); + return Ok(false); + } + } + } + Ok(false) + } + pub fn rewrite_path(&self, incoming_path: &str) -> String { // Strip /api prefix to get OpenCode path let result = incoming_path diff --git a/packages/ui/src/hooks/useEventStream.ts b/packages/ui/src/hooks/useEventStream.ts index a50f0f50..f4c8a209 100644 --- a/packages/ui/src/hooks/useEventStream.ts +++ b/packages/ui/src/hooks/useEventStream.ts @@ -160,6 +160,23 @@ export const useEventStream = () => { [setEventStreamStatus] ); + const bootstrapState = React.useCallback( + async (reason: string) => { + if (streamDebugEnabled()) { + console.info('[useEventStream] Bootstrapping state:', reason); + } + try { + await Promise.all([ + loadSessions(), + currentSessionId ? loadMessages(currentSessionId) : Promise.resolve(), + ]); + } catch (error) { + console.warn('[useEventStream] Bootstrap failed:', reason, error); + } + }, + [currentSessionId, loadMessages, loadSessions] + ); + const trackMessage = React.useCallback((messageId: string, event?: string, extraData?: Record) => { if (streamDebugEnabled()) { console.debug(`[MessageTracker] ${messageId}: ${event}`, extraData); @@ -195,6 +212,17 @@ export const useEventStream = () => { const lastEventTimestampRef = React.useRef(Date.now()); const isDesktopRuntimeRef = React.useRef(false); + const maybeBootstrapIfStale = React.useCallback( + (reason: string) => { + const now = Date.now(); + if (now - lastEventTimestampRef.current > 25000) { + void bootstrapState(reason); + lastEventTimestampRef.current = now; + } + }, + [bootstrapState] + ); + React.useEffect(() => { if (typeof window !== 'undefined') { const apis = (window as typeof window & { __OPENCHAMBER_RUNTIME_APIS__?: { runtime?: { isDesktop?: boolean } } }).__OPENCHAMBER_RUNTIME_APIS__; @@ -325,6 +353,11 @@ export const useEventStream = () => { case 'server.connected': checkConnection(); break; + case 'global.disposed': + case 'server.instance.disposed': { + void bootstrapState('server_disposed_event'); + break; + } case 'openchamber:session-activity': { if (!isDesktopRuntimeRef.current) break; const sessionId = typeof props.sessionId === 'string' ? props.sessionId : null; @@ -821,7 +854,8 @@ export const useEventStream = () => { applySessionMetadata, trackMessage, reportMessage, - updateSessionActivityPhase + updateSessionActivityPhase, + bootstrapState ]); const shouldHoldConnection = React.useCallback(() => { @@ -921,7 +955,9 @@ export const useEventStream = () => { publishStatus('connected', null); checkConnection(); - if (shouldRefresh && currentSessionId) { + if (shouldRefresh) { + void bootstrapState('sse_reconnected'); + } else if (currentSessionId) { setTimeout(() => { loadMessages(currentSessionId) .then(() => requestSessionMetadataRefresh(currentSessionId)) @@ -980,7 +1016,8 @@ export const useEventStream = () => { effectiveDirectory, updateSessionActivityPhase, waitForDesktopBridge, - debugConnectionState + debugConnectionState, + bootstrapState ]); const scheduleReconnect = React.useCallback((hint?: string) => { @@ -1064,6 +1101,7 @@ export const useEventStream = () => { if (visibilityStateRef.current === 'visible') { clearPauseTimeout(); + maybeBootstrapIfStale('visibility_restore'); if (pendingResumeRef.current || !unsubscribeRef.current) { console.info('[useEventStream] Visibility restored, triggering soft refresh...'); if (!isDesktopRuntimeRef.current) { @@ -1088,6 +1126,7 @@ export const useEventStream = () => { if (visibilityStateRef.current === 'visible') { clearPauseTimeout(); + maybeBootstrapIfStale('window_focus'); if (pendingResumeRef.current || !unsubscribeRef.current) { console.info('[useEventStream] Window focused after pause, triggering soft refresh...'); @@ -1111,6 +1150,7 @@ export const useEventStream = () => { const handleOnline = () => { onlineStatusRef.current = true; + maybeBootstrapIfStale('network_restored'); if (pendingResumeRef.current || !unsubscribeRef.current) { publishStatus('connecting', 'Network restored'); startStream({ resetAttempts: true }); diff --git a/packages/ui/src/lib/opencode/client.ts b/packages/ui/src/lib/opencode/client.ts index d07a0bd7..732403fd 100644 --- a/packages/ui/src/lib/opencode/client.ts +++ b/packages/ui/src/lib/opencode/client.ts @@ -633,6 +633,8 @@ class OpencodeService { const abortController = new AbortController(); this.sseAbortController = abortController; + let lastEventId: string | undefined; + console.log('[OpencodeClient] Starting SSE subscription...'); // Start async generator in background with reconnect on failure @@ -645,13 +647,21 @@ class OpencodeService { const connect = async (attempt: number): Promise => { try { - const result = await this.client.event.subscribe({ + const subscribeOptions: { + query?: { directory?: string }; + signal: AbortSignal; + sseDefaultRetryDelay: number; + sseMaxRetryDelay: number; + onSseError?: (error: unknown) => void; + onSseEvent: (event: StreamEvent) => void; + headers?: Record; + lastEventId?: string; + } = { query: resolvedDirectory ? { directory: resolvedDirectory } : undefined, signal: abortController.signal, - sseMaxRetryAttempts: 2, - sseDefaultRetryDelay: 500, - sseMaxRetryDelay: 8000, - onSseError: (error) => { + sseDefaultRetryDelay: 3000, + sseMaxRetryDelay: 30000, + onSseError: (error: unknown) => { if (error instanceof Error && error.name === 'AbortError') { return; } @@ -661,14 +671,23 @@ class OpencodeService { } }, onSseEvent: (event: StreamEvent) => { - if (!abortController.signal.aborted) { - const payload = event.data; - if (payload && typeof payload === 'object') { - onMessage(payload as Event); - } + if (abortController.signal.aborted) return; + if (event.id && typeof event.id === 'string') { + lastEventId = event.id; + } + const payload = event.data; + if (payload && typeof payload === 'object') { + onMessage(payload as Event); } }, - }); + }; + + if (lastEventId) { + subscribeOptions.lastEventId = lastEventId; + subscribeOptions.headers = { ...(subscribeOptions.headers || {}), 'Last-Event-ID': lastEventId }; + } + + const result = await this.client.event.subscribe(subscribeOptions); if (onOpen && !abortController.signal.aborted) { console.log('[OpencodeClient] SSE connection opened'); @@ -682,6 +701,13 @@ class OpencodeService { break; } } + + if (!abortController.signal.aborted) { + // Stream ended unexpectedly; attempt reconnect + const delay = Math.min(3000 * Math.pow(2, attempt), 30000); + await new Promise((resolve) => setTimeout(resolve, delay)); + await connect(attempt + 1); + } } catch (error: unknown) { if ((error as Error)?.name === 'AbortError' || abortController.signal.aborted) { console.log('[OpencodeClient] SSE stream aborted normally'); @@ -691,7 +717,7 @@ class OpencodeService { if (onError) { onError(error); } - const delay = Math.min(500 * Math.pow(2, attempt), 8000); + const delay = Math.min(3000 * Math.pow(2, attempt), 30000); await new Promise((resolve) => setTimeout(resolve, delay)); if (!abortController.signal.aborted) { await connect(attempt + 1);