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 <noreply@anthropic.com>

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();

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;

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<String>,
+    #[serde(default)]
+    error_message: Option<String>,
+}
+
+/// Background ping scheduler that health-checks running skills.
+pub struct PingScheduler {
+    /// Reference to the skill registry (set after engine initialisation).
+    registry: Arc<RwLock<Option<Arc<SkillRegistry>>>>,
+    /// Tauri app handle for emitting events to the frontend.
+    app_handle: Arc<RwLock<Option<AppHandle>>>,
+    /// Watch channel to signal the tick loop to stop.
+    stop_tx: watch::Sender<bool>,
+}
+
+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<SkillRegistry>) {
+        *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(&reg, &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<Arc<SkillRegistry>>, app_handle: &Option<AppHandle>) {
+        let registry = match registry {
+            Some(r) => r,
+            None => return,
+        };
+
+        // Collect running skill IDs
+        let running: Vec<String> = 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, &registry, &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<SkillRegistry>,
+        app_handle: &Option<AppHandle>,
+    ) {
+        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<RwLock<>>, 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()
+    }
+}

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<SkillRegistry>,
     /// Global cron scheduler for timed skill triggers.
     cron_scheduler: Arc<CronScheduler>,
+    /// Background ping scheduler for skill health checks.
+    ping_scheduler: Arc<PingScheduler>,
     /// Persistent user enable/disable preferences for skills.
     preferences: Arc<PreferencesStore>,
     /// 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(&registry));
+        let ping_scheduler = Arc::new(PingScheduler::new());
+        ping_scheduler.set_registry(Arc::clone(&registry));
         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<PingScheduler> {
+        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);
     }
 

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<string, SkillRuntime>();
-  private pingIntervalId: ReturnType<typeof setInterval> | 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<void> {
-    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<void> {
-    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<void> {
-    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
   // -----------------------------------------------------------------------

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;
     };