From f372d5ce292a3ddf24e95f1c2452834500e33a0f Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Tue, 10 Feb 2026 11:21:39 +0530 Subject: [PATCH] Move skill ping/health-check from frontend to Rust backend The frontend ping loop only runs while React is active, but the Rust backend keeps skills running independently. This adds a PingScheduler (modeled after CronScheduler) that pings all running skills every 5 minutes from a background Tokio task and acts on auth/network failures. The redundant frontend ping loop is removed since SkillProvider already listens for the Tauri events the Rust scheduler emits. Co-Authored-By: Claude Opus 4.6 --- src-tauri/src/lib.rs | 6 + src-tauri/src/runtime/mod.rs | 2 + src-tauri/src/runtime/ping_scheduler.rs | 302 ++++++++++++++++++++++++ src-tauri/src/runtime/qjs_engine.rs | 12 + src/lib/skills/manager.ts | 99 -------- src/providers/SkillProvider.tsx | 2 - 6 files changed, 322 insertions(+), 101 deletions(-) create mode 100644 src-tauri/src/runtime/ping_scheduler.rs diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 02c56f1db..508456b5f 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -318,6 +318,12 @@ pub fn run() { cron.start(); }); + // Start the ping scheduler (health-checks running skills) + let ping = engine.ping_scheduler(); + tauri::async_runtime::spawn(async move { + ping.start(); + }); + // Auto-start skills in background (no delay needed for QuickJS - // lightweight contexts don't have V8's memory reservation issue) let engine_clone = engine.clone(); diff --git a/src-tauri/src/runtime/mod.rs b/src-tauri/src/runtime/mod.rs index c4acfb5f1..337d494e8 100644 --- a/src-tauri/src/runtime/mod.rs +++ b/src-tauri/src/runtime/mod.rs @@ -19,6 +19,8 @@ pub mod bridge; #[cfg(not(any(target_os = "android", target_os = "ios")))] pub mod cron_scheduler; #[cfg(not(any(target_os = "android", target_os = "ios")))] +pub mod ping_scheduler; +#[cfg(not(any(target_os = "android", target_os = "ios")))] pub mod skill_registry; #[cfg(not(any(target_os = "android", target_os = "ios")))] pub mod qjs_engine; diff --git a/src-tauri/src/runtime/ping_scheduler.rs b/src-tauri/src/runtime/ping_scheduler.rs new file mode 100644 index 000000000..ad93036d1 --- /dev/null +++ b/src-tauri/src/runtime/ping_scheduler.rs @@ -0,0 +1,302 @@ +//! PingScheduler — background Tokio task that periodically health-checks running skills. +//! +//! Every 5 minutes the scheduler pings all skills whose status is `Running` by +//! sending an RPC `skill/ping` message. The response is interpreted as follows: +//! +//! - `null` / `{ ok: true }` → healthy, no action +//! - `{ ok: false, errorType: "auth" }` → stop the skill and set an error status +//! - `{ ok: false, errorType: "network" }` → update `connection_status` in the +//! skill's published state to `"error"` but keep the skill running +//! +//! Architecture follows the same pattern as `CronScheduler`: a background Tokio +//! task with `tokio::select!` for a tick interval + a stop signal via a watch channel. + +use std::sync::Arc; +use std::time::Duration; + +use parking_lot::RwLock; +use serde::Deserialize; +use tauri::{AppHandle, Emitter}; +use tokio::sync::watch; + +use crate::runtime::skill_registry::SkillRegistry; +use crate::runtime::types::{events, SkillMessage, SkillStatus}; + +/// Interval between ping sweeps. +const PING_INTERVAL: Duration = Duration::from_secs(5 * 60); + +/// Per-skill timeout when waiting for a ping reply. +const PING_TIMEOUT: Duration = Duration::from_secs(30); + +/// Deserialized result from a skill's `onPing()` handler. +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct PingResult { + ok: bool, + #[serde(default)] + error_type: Option, + #[serde(default)] + error_message: Option, +} + +/// Background ping scheduler that health-checks running skills. +pub struct PingScheduler { + /// Reference to the skill registry (set after engine initialisation). + registry: Arc>>>, + /// Tauri app handle for emitting events to the frontend. + app_handle: Arc>>, + /// Watch channel to signal the tick loop to stop. + stop_tx: watch::Sender, +} + +impl PingScheduler { + pub fn new() -> Self { + let (stop_tx, _) = watch::channel(false); + Self { + registry: Arc::new(RwLock::new(None)), + app_handle: Arc::new(RwLock::new(None)), + stop_tx, + } + } + + /// Set the skill registry (called after engine initialisation). + pub fn set_registry(&self, registry: Arc) { + *self.registry.write() = Some(registry); + } + + /// Set the Tauri app handle for emitting events. + pub fn set_app_handle(&self, handle: AppHandle) { + *self.app_handle.write() = Some(handle); + } + + /// Start the background ping loop. Returns the Tokio task handle. + /// + /// Must be called from within a Tokio runtime context. + pub fn start(&self) -> tokio::task::JoinHandle<()> { + let registry = self.registry.clone(); + let app_handle = self.app_handle.clone(); + let mut stop_rx = self.stop_tx.subscribe(); + + tokio::spawn(async move { + log::info!("[ping] Scheduler started ({}s interval)", PING_INTERVAL.as_secs()); + + loop { + tokio::select! { + _ = tokio::time::sleep(PING_INTERVAL) => { + let reg = registry.read().clone(); + let handle = app_handle.read().clone(); + Self::tick(®, &handle).await; + } + _ = stop_rx.changed() => { + log::info!("[ping] Scheduler stopped"); + break; + } + } + } + }) + } + + /// Stop the scheduler's tick loop. + #[allow(dead_code)] + pub fn stop(&self) { + let _ = self.stop_tx.send(true); + } + + /// Ping all running skills concurrently and act on failures. + async fn tick(registry: &Option>, app_handle: &Option) { + let registry = match registry { + Some(r) => r, + None => return, + }; + + // Collect running skill IDs + let running: Vec = registry + .list_skills() + .into_iter() + .filter(|s| s.status == SkillStatus::Running) + .map(|s| s.skill_id) + .collect(); + + if running.is_empty() { + return; + } + + log::debug!("[ping] Pinging {} running skill(s)", running.len()); + + // Ping all skills concurrently + let futures: Vec<_> = running + .into_iter() + .map(|skill_id| { + let registry = Arc::clone(registry); + let app_handle = app_handle.clone(); + async move { + Self::ping_skill(&skill_id, ®istry, &app_handle).await; + } + }) + .collect(); + + futures::future::join_all(futures).await; + } + + /// Ping a single skill and handle the result. + async fn ping_skill( + skill_id: &str, + registry: &Arc, + app_handle: &Option, + ) { + log::debug!("[ping] Pinging skill '{}'", skill_id); + + // Send the RPC message + let (tx, rx) = tokio::sync::oneshot::channel(); + if let Err(e) = registry.send_message( + skill_id, + SkillMessage::Rpc { + method: "skill/ping".to_string(), + params: serde_json::json!({}), + reply: tx, + }, + ) { + log::warn!("[ping] Failed to send ping to '{}': {}", skill_id, e); + return; + } + + // Wait for the reply with a timeout + let reply = match tokio::time::timeout(PING_TIMEOUT, rx).await { + Ok(Ok(result)) => result, + Ok(Err(_)) => { + log::warn!("[ping] Ping channel closed for '{}'", skill_id); + return; + } + Err(_) => { + log::warn!("[ping] Ping timed out for '{}'", skill_id); + return; + } + }; + + // Parse the result + let value = match reply { + Ok(v) => v, + Err(e) => { + log::warn!("[ping] Ping RPC error for '{}': {}", skill_id, e); + return; + } + }; + + // null / { ok: true } → healthy + if value.is_null() { + return; + } + + let ping_result: PingResult = match serde_json::from_value(value) { + Ok(r) => r, + Err(e) => { + log::debug!( + "[ping] Could not parse ping result for '{}': {} — treating as healthy", + skill_id, + e + ); + return; + } + }; + + if ping_result.ok { + return; + } + + // ----- Handle failure ----- + let error_type = ping_result.error_type.as_deref().unwrap_or("unknown"); + let error_message = ping_result + .error_message + .as_deref() + .unwrap_or("Ping failed"); + + log::warn!( + "[ping] Skill '{}' ping failed: type={}, message={}", + skill_id, + error_type, + error_message + ); + + match error_type { + "auth" => { + // Auth failure: stop the skill and emit error event + log::info!("[ping] Stopping skill '{}' due to auth failure", skill_id); + + if let Err(e) = registry.stop_skill(skill_id).await { + log::error!("[ping] Failed to stop skill '{}': {}", skill_id, e); + } + + if let Some(handle) = app_handle { + let payload = serde_json::json!({ + "skill_id": skill_id, + "status": "error", + "error": error_message, + }); + let _ = handle.emit(events::SKILL_STATUS_CHANGED, &payload); + } + } + _ => { + // Network or other error: update published state, keep running + if let Some(snap) = registry.get_skill(skill_id) { + // We need to update the skill's published_state through the + // registry. The SkillState is behind an Arc>, which + // we can reach via the snapshot's backing state. However, the + // registry only exposes snapshots (copies). We use an RPC + // message to let the skill instance update its own state. + // + // A simpler approach: directly update published_state via the + // SkillState Arc that the registry entry holds. Since + // SkillRegistry doesn't expose the Arc directly, we send a + // state/set RPC to the skill, which is the same mechanism + // the frontend uses. + let _ = snap; // used for logging context + + // Send a state update via RPC (skills handle "state/set" + // in their reverse-RPC handler, but here we update the + // published_state directly through the skill message loop). + let (tx, rx) = tokio::sync::oneshot::channel(); + let _ = registry.send_message( + skill_id, + SkillMessage::Rpc { + method: "state/set".to_string(), + params: serde_json::json!({ + "partial": { + "connection_status": "error", + "connection_error": error_message, + } + }), + reply: tx, + }, + ); + // Don't block on the reply — fire-and-forget + let _ = tokio::time::timeout(Duration::from_secs(5), rx).await; + } + + // Also emit a state-changed event so the frontend picks it up + if let Some(handle) = app_handle { + let mut state_map = std::collections::HashMap::new(); + state_map.insert( + "connection_status".to_string(), + serde_json::Value::String("error".to_string()), + ); + state_map.insert( + "connection_error".to_string(), + serde_json::Value::String(error_message.to_string()), + ); + + let payload = serde_json::json!({ + "skillId": skill_id, + "state": state_map, + }); + let _ = handle.emit("skill-state-changed", &payload); + } + } + } + } +} + +impl Default for PingScheduler { + fn default() -> Self { + Self::new() + } +} diff --git a/src-tauri/src/runtime/qjs_engine.rs b/src-tauri/src/runtime/qjs_engine.rs index a8981023f..15ec60382 100644 --- a/src-tauri/src/runtime/qjs_engine.rs +++ b/src-tauri/src/runtime/qjs_engine.rs @@ -11,6 +11,7 @@ use tauri::{AppHandle, Emitter}; use crate::runtime::cron_scheduler::CronScheduler; use crate::runtime::manifest::SkillManifest; +use crate::runtime::ping_scheduler::PingScheduler; use crate::runtime::preferences::PreferencesStore; use crate::runtime::skill_registry::SkillRegistry; use crate::runtime::socket_manager::SocketManager; @@ -24,6 +25,8 @@ pub struct RuntimeEngine { registry: Arc, /// Global cron scheduler for timed skill triggers. cron_scheduler: Arc, + /// Background ping scheduler for skill health checks. + ping_scheduler: Arc, /// Persistent user enable/disable preferences for skills. preferences: Arc, /// Base data directory for skills (platform-aware). @@ -44,6 +47,8 @@ impl RuntimeEngine { let registry = Arc::new(SkillRegistry::new()); let cron_scheduler = Arc::new(CronScheduler::new()); cron_scheduler.set_registry(Arc::clone(®istry)); + let ping_scheduler = Arc::new(PingScheduler::new()); + ping_scheduler.set_registry(Arc::clone(®istry)); let preferences = Arc::new(PreferencesStore::new(&skills_data_dir)); log::info!("[runtime] QuickJS RuntimeEngine created"); @@ -51,6 +56,7 @@ impl RuntimeEngine { Ok(Self { registry, cron_scheduler, + ping_scheduler, preferences, skills_data_dir, skills_source_dir: RwLock::new(None), @@ -70,8 +76,14 @@ impl RuntimeEngine { Arc::clone(&self.cron_scheduler) } + /// Get a clone of the ping scheduler Arc. + pub fn ping_scheduler(&self) -> Arc { + Arc::clone(&self.ping_scheduler) + } + /// Set the Tauri app handle for emitting events to the frontend. pub fn set_app_handle(&self, handle: AppHandle) { + self.ping_scheduler.set_app_handle(handle.clone()); *self.app_handle.write() = Some(handle); } diff --git a/src/lib/skills/manager.ts b/src/lib/skills/manager.ts index 3357dd10e..4f4ff2ad3 100644 --- a/src/lib/skills/manager.ts +++ b/src/lib/skills/manager.ts @@ -16,7 +16,6 @@ import type { SetupResult, SkillToolDefinition, SkillOptionDefinition, - PingResult, } from "./types"; import { store } from "../../store"; import { @@ -33,8 +32,6 @@ import { class SkillManager { private runtimes = new Map(); - private pingIntervalId: ReturnType | null = null; - static PING_INTERVAL_MS = 5 * 60 * 1000; // 5 minutes /** * Get skill-specific load parameters (e.g., session data for Telegram) @@ -134,10 +131,6 @@ class SkillManager { const tools = await runtime.listTools(); store.dispatch(setSkillTools({ skillId, tools })); store.dispatch(setSkillStatus({ skillId, status: "ready" })); - // Fire an initial ping (non-blocking) - this.pingSkill(skillId).catch((err) => { - console.warn(`[SkillManager] Initial ping failed for ${skillId}:`, err); - }); syncToolsToBackend(); } catch (err) { const msg = err instanceof Error ? err.message : String(err); @@ -348,7 +341,6 @@ class SkillManager { * Stop all running skills. */ async stopAll(): Promise { - this.stopPingLoop(); const ids = Array.from(this.runtimes.keys()); await Promise.all(ids.map((id) => this.stopSkill(id))); } @@ -394,97 +386,6 @@ class SkillManager { } } - // ----------------------------------------------------------------------- - // Ping / health-check - // ----------------------------------------------------------------------- - - /** - * Start the periodic ping loop. Call once after skills are started. - */ - startPingLoop(): void { - if (this.pingIntervalId) return; - this.pingIntervalId = setInterval(() => { - this.pingAllSkills().catch((err) => { - console.error("[SkillManager] Ping loop error:", err); - }); - }, SkillManager.PING_INTERVAL_MS); - } - - /** - * Stop the periodic ping loop. - */ - stopPingLoop(): void { - if (this.pingIntervalId) { - clearInterval(this.pingIntervalId); - this.pingIntervalId = null; - } - } - - /** - * Ping all ready skills in parallel and handle results. - */ - private async pingAllSkills(): Promise { - const entries: [string, SkillRuntime][] = []; - for (const [id, runtime] of this.runtimes) { - const status = this.getSkillStatus(id); - if (status === "ready" && runtime.isRunning) { - entries.push([id, runtime]); - } - } - if (entries.length === 0) return; - - await Promise.allSettled( - entries.map(([id]) => this.pingSkill(id)), - ); - } - - /** - * Ping a single skill and interpret the result. - * - null / ok:true → healthy, no-op - * - ok:false, errorType:"auth" → stop skill + set error - * - ok:false, errorType:"network" → update connection_status to error (keep running) - */ - private async pingSkill(skillId: string): Promise { - const runtime = this.runtimes.get(skillId); - if (!runtime || !runtime.isRunning) return; - - let result: PingResult | null; - try { - result = await runtime.ping(); - } catch (err) { - console.warn(`[SkillManager] Ping RPC failed for ${skillId}:`, err); - return; - } - - // null means onPing is not implemented → treat as healthy - if (!result || result.ok) return; - - console.warn( - `[SkillManager] Ping failed for ${skillId}: ${result.errorType} — ${result.errorMessage}`, - ); - - if (result.errorType === "auth") { - store.dispatch( - setSkillError({ skillId, error: result.errorMessage ?? "Auth error" }), - ); - await this.stopSkill(skillId); - } else { - // Network error — update state but keep skill running - const currentState = - store.getState().skills.skillStates[skillId] ?? {}; - store.dispatch( - setSkillState({ - skillId, - state: { - ...currentState, - connection_status: "error", - connection_error: result.errorMessage ?? "Connection error", - }, - }), - ); - } - } - // ----------------------------------------------------------------------- // Reverse RPC handling // ----------------------------------------------------------------------- diff --git a/src/providers/SkillProvider.tsx b/src/providers/SkillProvider.tsx index 52e9d75c9..f9c1f5f80 100644 --- a/src/providers/SkillProvider.tsx +++ b/src/providers/SkillProvider.tsx @@ -160,7 +160,6 @@ export default function SkillProvider({ children }: { children: ReactNode }) { // Discover skills from the QuickJS runtime engine const manifests = await discoverSkills(); await registerAndStart(manifests); - skillManager.startPingLoop(); } catch (err) { console.error('[SkillProvider] Failed to discover skills:', err); } @@ -169,7 +168,6 @@ export default function SkillProvider({ children }: { children: ReactNode }) { init(); return () => { - skillManager.stopPingLoop(); skillManager.stopAll().catch(console.error); initRef.current = false; };