BuzzbeeSCD/openchamber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(connection): improve resilience with watchdog and reconnection logic

Browse Source
This commit is contained in:
Bohdan Triapitsyn
2025-12-19 01:21:09 +02:00
parent 3e027facae
commit 71574acf19
4 changed files with 193 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/desktop/src-tauri/src/main.rs
+93
View File
@@ -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<u16>,
api_prefix: String,
cli_available: bool,
}
let mut last_snapshot: Option<HealthSnapshot> = 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());
packages/desktop/src-tauri/src/opencode_manager.rs
+19
View File
@@ -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<bool> {
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
packages/ui/src/hooks/useEventStream.ts
+43 -3
View File
@@ -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<string, unknown>) => {
if (streamDebugEnabled()) {
console.debug(`[MessageTracker] ${messageId}: ${event}`, extraData);
@@ -195,6 +212,17 @@ export const useEventStream = () => {
const lastEventTimestampRef = React.useRef<number>(Date.now());
const isDesktopRuntimeRef = React.useRef<boolean>(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 });
packages/ui/src/lib/opencode/client.ts
+38 -12
View File
@@ -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<void> => {
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<unknown>) => void;
headers?: Record<string, string>;
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<unknown>) => {
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);