feat(gpu-monitor): introduce GPU monitoring plugin with validation, build scripts, and documentation; enhance plugin development guidelines

src-tauri/Cargo.lock

@@ -4940,6 +4940,7 @@ dependencies = [
  "md-5 0.10.6",
  "notify",
  "nucleo-matcher",
+ "nyaterm-gpu",
  "nyaterm-mcp-protocol",
  "nyaterm-otp",
  "nyaterm-plugin-runtime",
@@ -4996,6 +4997,13 @@ dependencies = [
  "zmodem2",
 ]
 
+[[package]]
+name = "nyaterm-gpu"
+version = "0.1.0"
+dependencies = [
+ "serde",
+]
+
 [[package]]
 name = "nyaterm-mcp-protocol"
 version = "0.1.0"
@@ -5015,6 +5023,7 @@ version = "0.1.0"
 dependencies = [
  "async-trait",
  "hex",
+ "nyaterm-gpu",
  "regex",
  "semver",
  "serde",

src-tauri/Cargo.toml

@@ -85,6 +85,7 @@ tauri-build = { version = "2", features = [] }
 tokio = { version = "1", features = ["test-util"] }
 
 [dependencies]
+nyaterm-gpu = { path = "crates/nyaterm-gpu" }
 nyaterm-otp = { path = "crates/otp" }
 nyaterm-mcp-protocol = { path = "crates/nyaterm-mcp-protocol" }
 nyaterm-plugin-runtime = { path = "crates/nyaterm-plugin-runtime" }

src-tauri/crates/nyaterm-gpu/Cargo.lock

@@ -0,0 +1,75 @@
+# This file is automatically @generated by Cargo.
+# It is not intended for manual editing.
+version = 4
+
+[[package]]
+name = "nyaterm-gpu"
+version = "0.1.0"
+dependencies = [
+ "serde",
+]
+
+[[package]]
+name = "proc-macro2"
+version = "1.0.107"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9"
+dependencies = [
+ "unicode-ident",
+]
+
+[[package]]
+name = "quote"
+version = "1.0.47"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001"
+dependencies = [
+ "proc-macro2",
+]
+
+[[package]]
+name = "serde"
+version = "1.0.229"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba"
+dependencies = [
+ "serde_core",
+ "serde_derive",
+]
+
+[[package]]
+name = "serde_core"
+version = "1.0.229"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48"
+dependencies = [
+ "serde_derive",
+]
+
+[[package]]
+name = "serde_derive"
+version = "1.0.229"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "syn"
+version = "3.0.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8593e8e72159ed2257d083c7a454a85cbf854f37a0966d8d483aff8c8a3ebcee"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "unicode-ident",
+]
+
+[[package]]
+name = "unicode-ident"
+version = "1.0.26"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954"

src-tauri/crates/nyaterm-gpu/Cargo.toml

@@ -0,0 +1,8 @@
+[package]
+name = "nyaterm-gpu"
+version = "0.1.0"
+edition = "2024"
+rust-version = "1.94"
+
+[dependencies]
+serde = { version = "1", features = ["derive"] }

src-tauri/crates/nyaterm-gpu/src/lib.rs

@@ -0,0 +1,332 @@
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+
+#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
+#[serde(deny_unknown_fields)]
+pub struct RemoteGpuOverview {
+    pub available: bool,
+    pub driver_version: String,
+    pub cuda_version: String,
+    pub gpus: Vec<RemoteGpu>,
+    pub processes: Vec<RemoteGpuProcess>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(deny_unknown_fields)]
+pub struct RemoteGpu {
+    pub index: u32,
+    pub uuid: String,
+    pub name: String,
+    pub temperature_c: Option<f64>,
+    pub utilization_gpu_percent: Option<f64>,
+    pub utilization_memory_percent: Option<f64>,
+    pub memory_total_mb: u64,
+    pub memory_used_mb: u64,
+    pub memory_free_mb: u64,
+    pub power_draw_w: Option<f64>,
+    pub power_limit_w: Option<f64>,
+    pub fan_speed_percent: Option<f64>,
+    pub pstate: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(deny_unknown_fields)]
+pub struct RemoteGpuProcess {
+    pub gpu_uuid: String,
+    pub gpu_index: Option<u32>,
+    pub pid: u32,
+    pub process_name: String,
+    pub used_memory_mb: u64,
+}
+
+enum GpuParseSection {
+    None,
+    Gpus,
+    Processes,
+}
+
+pub fn parse_gpu_overview_output(output: &str) -> RemoteGpuOverview {
+    let mut overview = RemoteGpuOverview::default();
+    let mut section = GpuParseSection::None;
+    let mut process_rows = Vec::new();
+
+    for line in output.lines() {
+        match line {
+            "GPU_CSV_BEGIN" => {
+                section = GpuParseSection::Gpus;
+                continue;
+            }
+            "GPU_CSV_END" => {
+                section = GpuParseSection::None;
+                continue;
+            }
+            "GPU_PROCESS_CSV_BEGIN" => {
+                section = GpuParseSection::Processes;
+                continue;
+            }
+            "GPU_PROCESS_CSV_END" => {
+                section = GpuParseSection::None;
+                continue;
+            }
+            _ => {}
+        }
+
+        let cols: Vec<&str> = line.split('\t').collect();
+        if cols.first() == Some(&"GPU_AVAILABLE") && cols.len() >= 2 {
+            overview.available = cols[1] == "1";
+            continue;
+        }
+        if cols.first() == Some(&"GPU_CUDA_VERSION") && cols.len() >= 2 {
+            overview.cuda_version = cols[1].trim().to_string();
+            continue;
+        }
+
+        match section {
+            GpuParseSection::Gpus => {
+                if let Some(gpu) = parse_gpu_csv_line(line) {
+                    if overview.driver_version.is_empty() {
+                        overview.driver_version = gpu.1;
+                    }
+                    overview.gpus.push(gpu.0);
+                }
+            }
+            GpuParseSection::Processes => {
+                if !line.trim().is_empty() {
+                    process_rows.push(line.to_string());
+                }
+            }
+            GpuParseSection::None => {}
+        }
+    }
+
+    let gpu_indexes: HashMap<String, u32> = overview
+        .gpus
+        .iter()
+        .map(|gpu| (gpu.uuid.clone(), gpu.index))
+        .collect();
+    overview.processes = process_rows
+        .iter()
+        .filter_map(|line| parse_gpu_process_csv_line(line, &gpu_indexes))
+        .collect();
+
+    overview
+}
+
+fn parse_gpu_csv_line(line: &str) -> Option<(RemoteGpu, String)> {
+    let cols = parse_csv_line(line);
+    if cols.len() < 14 {
+        return None;
+    }
+
+    let gpu = RemoteGpu {
+        index: parse_u32(&cols[0]).unwrap_or(0),
+        uuid: cols[1].trim().to_string(),
+        name: cols[2].trim().to_string(),
+        temperature_c: parse_optional_f64(&cols[4]),
+        utilization_gpu_percent: parse_optional_f64(&cols[5]),
+        utilization_memory_percent: parse_optional_f64(&cols[6]),
+        memory_total_mb: parse_u64(&cols[7]).unwrap_or(0),
+        memory_used_mb: parse_u64(&cols[8]).unwrap_or(0),
+        memory_free_mb: parse_u64(&cols[9]).unwrap_or(0),
+        power_draw_w: parse_optional_f64(&cols[10]),
+        power_limit_w: parse_optional_f64(&cols[11]),
+        fan_speed_percent: parse_optional_f64(&cols[12]),
+        pstate: clean_text(&cols[13]),
+    };
+
+    Some((gpu, clean_text(&cols[3])))
+}
+
+fn parse_gpu_process_csv_line(
+    line: &str,
+    gpu_indexes: &HashMap<String, u32>,
+) -> Option<RemoteGpuProcess> {
+    let cols = parse_csv_line(line);
+    if cols.len() < 4 {
+        return None;
+    }
+
+    let gpu_uuid = clean_text(&cols[0]);
+    if gpu_uuid.is_empty() || is_missing(&gpu_uuid) {
+        return None;
+    }
+
+    Some(RemoteGpuProcess {
+        gpu_index: gpu_indexes.get(&gpu_uuid).copied(),
+        gpu_uuid,
+        pid: parse_u32(&cols[1]).unwrap_or(0),
+        used_memory_mb: parse_u64(&cols[2]).unwrap_or(0),
+        process_name: cols[3..].join(",").trim().to_string(),
+    })
+}
+
+fn parse_csv_line(line: &str) -> Vec<String> {
+    let chars: Vec<char> = line.chars().collect();
+    let mut values = Vec::new();
+    let mut current = String::new();
+    let mut in_quotes = false;
+    let mut index = 0;
+
+    while index < chars.len() {
+        let ch = chars[index];
+        match ch {
+            '"' if in_quotes && chars.get(index + 1) == Some(&'"') => {
+                current.push('"');
+                index += 1;
+            }
+            '"' => {
+                in_quotes = !in_quotes;
+            }
+            ',' if !in_quotes => {
+                values.push(current.trim().to_string());
+                current.clear();
+            }
+            _ => current.push(ch),
+        }
+        index += 1;
+    }
+
+    values.push(current.trim().to_string());
+    values
+}
+
+fn clean_text(value: &str) -> String {
+    value.trim().to_string()
+}
+
+fn is_missing(value: &str) -> bool {
+    let normalized = value.trim().to_ascii_lowercase();
+    matches!(
+        normalized.as_str(),
+        "" | "n/a" | "na" | "[n/a]" | "[not supported]" | "not supported"
+    )
+}
+
+fn parse_optional_f64(value: &str) -> Option<f64> {
+    if is_missing(value) {
+        return None;
+    }
+    numeric_prefix(value).and_then(|value| value.parse().ok())
+}
+
+fn parse_u64(value: &str) -> Option<u64> {
+    if is_missing(value) {
+        return None;
+    }
+    numeric_prefix(value).and_then(|value| value.parse::<f64>().ok().map(|number| number as u64))
+}
+
+fn parse_u32(value: &str) -> Option<u32> {
+    parse_u64(value).and_then(|value| u32::try_from(value).ok())
+}
+
+fn numeric_prefix(value: &str) -> Option<&str> {
+    let trimmed = value.trim().trim_end_matches('%').trim();
+    let end = trimmed
+        .char_indices()
+        .take_while(|(_, ch)| ch.is_ascii_digit() || *ch == '.' || *ch == '-')
+        .map(|(index, ch)| index + ch.len_utf8())
+        .last()
+        .unwrap_or(0);
+
+    (end > 0).then_some(&trimmed[..end])
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parses_unavailable_gpu_output() {
+        let overview = parse_gpu_overview_output("GPU_AVAILABLE\t0\n");
+
+        assert!(!overview.available);
+        assert!(overview.gpus.is_empty());
+        assert!(overview.processes.is_empty());
+    }
+
+    #[test]
+    fn parses_single_gpu_with_metrics() {
+        let raw = concat!(
+            "GPU_AVAILABLE\t1\n",
+            "GPU_CUDA_VERSION\t12.4\n",
+            "GPU_CSV_BEGIN\n",
+            "0, GPU-abc, NVIDIA RTX 4090, 550.54.14, 43, 77, 31, 24564, 12345, 12219, 180.5, 450.0, 46, P2\n",
+            "GPU_CSV_END\n",
+            "GPU_PROCESS_CSV_BEGIN\n",
+            "GPU_PROCESS_CSV_END\n",
+        );
+
+        let overview = parse_gpu_overview_output(raw);
+
+        assert!(overview.available);
+        assert_eq!(overview.driver_version, "550.54.14");
+        assert_eq!(overview.cuda_version, "12.4");
+        assert_eq!(overview.gpus[0].name, "NVIDIA RTX 4090");
+        assert_eq!(overview.gpus[0].utilization_gpu_percent, Some(77.0));
+        assert_eq!(overview.gpus[0].memory_used_mb, 12345);
+        assert_eq!(overview.gpus[0].power_draw_w, Some(180.5));
+    }
+
+    #[test]
+    fn parses_multiple_gpus_and_missing_optional_metrics() {
+        let raw = concat!(
+            "GPU_AVAILABLE\t1\n",
+            "GPU_CUDA_VERSION\t12.2\n",
+            "GPU_CSV_BEGIN\n",
+            "0, GPU-a, NVIDIA A100, 535.1, N/A, 0, 0, 40536, 0, 40536, [Not Supported], 400, N/A, P0\n",
+            "1, GPU-b, NVIDIA T4, 535.1, 55, 99, 80, 15360, 15000, 360, 67.2, 70.0, [Not Supported], P0\n",
+            "GPU_CSV_END\n",
+            "GPU_PROCESS_CSV_BEGIN\n",
+            "GPU_PROCESS_CSV_END\n",
+        );
+
+        let overview = parse_gpu_overview_output(raw);
+
+        assert_eq!(overview.gpus.len(), 2);
+        assert_eq!(overview.gpus[0].temperature_c, None);
+        assert_eq!(overview.gpus[0].power_draw_w, None);
+        assert_eq!(overview.gpus[0].fan_speed_percent, None);
+        assert_eq!(overview.gpus[1].fan_speed_percent, None);
+    }
+
+    #[test]
+    fn maps_gpu_processes_to_indexes() {
+        let raw = concat!(
+            "GPU_AVAILABLE\t1\n",
+            "GPU_CUDA_VERSION\t12.2\n",
+            "GPU_CSV_BEGIN\n",
+            "0, GPU-a, NVIDIA A100, 535.1, 40, 30, 20, 40536, 1000, 39536, 80, 400, 20, P0\n",
+            "1, GPU-b, NVIDIA T4, 535.1, 44, 10, 9, 15360, 512, 14848, 35, 70, 30, P8\n",
+            "GPU_CSV_END\n",
+            "GPU_PROCESS_CSV_BEGIN\n",
+            "GPU-b, 4242, 2048, python\n",
+            "GPU_PROCESS_CSV_END\n",
+        );
+
+        let overview = parse_gpu_overview_output(raw);
+
+        assert_eq!(overview.processes.len(), 1);
+        assert_eq!(overview.processes[0].gpu_index, Some(1));
+        assert_eq!(overview.processes[0].pid, 4242);
+        assert_eq!(overview.processes[0].used_memory_mb, 2048);
+    }
+
+    #[test]
+    fn keeps_commas_inside_quoted_process_names() {
+        let raw = concat!(
+            "GPU_AVAILABLE\t1\n",
+            "GPU_CUDA_VERSION\t12.2\n",
+            "GPU_CSV_BEGIN\n",
+            "0, GPU-a, NVIDIA A100, 535.1, 40, 30, 20, 40536, 1000, 39536, 80, 400, 20, P0\n",
+            "GPU_CSV_END\n",
+            "GPU_PROCESS_CSV_BEGIN\n",
+            "GPU-a, 100, 512, \"python, worker\"\n",
+            "GPU_PROCESS_CSV_END\n",
+        );
+
+        let overview = parse_gpu_overview_output(raw);
+
+        assert_eq!(overview.processes[0].process_name, "python, worker");
+    }
+}

src-tauri/crates/nyaterm-plugin-runtime/Cargo.lock

@@ -391,12 +391,20 @@ dependencies = [
  "windows-sys",
 ]
 
+[[package]]
+name = "nyaterm-gpu"
+version = "0.1.0"
+dependencies = [
+ "serde",
+]
+
 [[package]]
 name = "nyaterm-plugin-runtime"
 version = "0.1.0"
 dependencies = [
  "async-trait",
  "hex",
+ "nyaterm-gpu",
  "nyaterm-plugin-sdk",
  "regex",
  "semver",

src-tauri/crates/nyaterm-plugin-runtime/Cargo.toml

@@ -5,6 +5,7 @@ edition = "2024"
 rust-version = "1.94"
 
 [dependencies]
+nyaterm-gpu = { path = "../nyaterm-gpu" }
 async-trait = "0.1"
 hex = "0.4"
 regex = "1"
@@ -21,7 +22,7 @@ uuid = { version = "1", features = ["v4"] }
 zip = { version = "8", default-features = false, features = ["deflate"] }
 
 [dev-dependencies]
-tokio = { version = "1", features = ["rt-multi-thread"] }
+tokio = { version = "1", features = ["rt-multi-thread", "test-util"] }
 
 [features]
 sdk-fixture = ["dep:nyaterm-plugin-sdk"]
@@ -34,3 +35,8 @@ optional = true
 name = "sdk-fixture"
 path = "src/bin/sdk-fixture.rs"
 required-features = ["sdk-fixture"]
+
+[[bin]]
+name = "gpu-fixture"
+path = "../../../plugins/examples/gpu-monitor/backend/src/main.rs"
+required-features = ["sdk-fixture"]

src-tauri/crates/nyaterm-plugin-runtime/src/lib.rs

@@ -1,7 +1,9 @@
 //! Tauri-independent installation, manifest validation and sidecar transport.
 pub mod diagnostics;
 pub mod manifest;
+pub mod monitoring;
 pub mod package;
+pub mod probe;
 pub mod registry;
 pub mod sidecar;
 

src-tauri/crates/nyaterm-plugin-runtime/src/manifest.rs

@@ -29,6 +29,29 @@ pub struct Contributions {
     pub panels: Vec<Panel>,
     #[serde(default)]
     pub commands: Vec<Command>,
+    #[serde(default)]
+    pub probes: Vec<Probe>,
+    #[serde(default)]
+    pub monitors: Vec<Monitor>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct Probe {
+    pub id: String,
+    pub title: String,
+    pub entry: String,
+    pub timeout_ms: u64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct Monitor {
+    pub id: String,
+    pub title: String,
+    pub schema: String,
+    pub method: String,
+    pub panel: String,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -126,6 +149,7 @@ pub fn validate_permission(permission: &str) -> Result<()> {
         "filesystem.read",
         "storage",
         "native",
+        "remote.probe",
     ]
     .contains(&permission)
     {
@@ -188,6 +212,8 @@ impl Manifest {
         if self.permissions.len() > 24
             || self.contributions.panels.len() > 16
             || self.contributions.commands.len() > 64
+            || self.contributions.probes.len() > 16
+            || self.contributions.monitors.len() > 16
         {
             return Err(invalid("Too many plugin permissions or contributions"));
         }
@@ -208,6 +234,22 @@ impl Manifest {
             return Err(invalid("Too many network origins"));
         }
         let mut ids = HashSet::new();
+        for probe in &self.contributions.probes {
+            validate_path(&probe.entry)?;
+            if !valid_id(&probe.id)
+                || !ids.insert(&probe.id)
+                || probe.title.trim().is_empty()
+                || probe.title.len() > 120
+                || !probe.entry.starts_with("assets/probes/")
+                || !probe.entry.ends_with(".sh")
+                || !(1000..=30_000).contains(&probe.timeout_ms)
+                || !permissions.contains(&"remote.probe".to_string())
+            {
+                return Err(invalid(
+                    "Invalid probe declaration or missing remote.probe permission",
+                ));
+            }
+        }
         for panel in &self.contributions.panels {
             if !valid_id(&panel.id)
                 || panel.title.trim().is_empty()
@@ -242,6 +284,28 @@ impl Manifest {
                 return Err(invalid("Too many command menu locations"));
             }
         }
+        let mut schemas = HashSet::new();
+        for monitor in &self.contributions.monitors {
+            if !valid_id(&monitor.id)
+                || !ids.insert(&monitor.id)
+                || monitor.title.trim().is_empty()
+                || monitor.title.len() > 120
+                || monitor.schema != "gpu.v1"
+                || !schemas.insert(&monitor.schema)
+                || !monitor.method.starts_with("monitor/")
+                || monitor.method.len() > 128
+                || monitor.method.len() <= 8
+                || self.backend.is_none()
+                || self.contributions.probes.is_empty()
+                || !self
+                    .contributions
+                    .panels
+                    .iter()
+                    .any(|p| p.id == monitor.panel)
+            {
+                return Err(invalid("Invalid monitor declaration"));
+            }
+        }
         if let Some(backend) = &self.backend {
             if !self.permissions.iter().any(|p| p == "native")
                 || backend.executables.is_empty()

src-tauri/crates/nyaterm-plugin-runtime/src/monitoring.rs

@@ -0,0 +1,266 @@
+//! Demand-driven, shared monitor actors. No Tauri or SSH dependency.
+use crate::{Result, invalid};
+use async_trait::async_trait;
+use nyaterm_gpu::RemoteGpuOverview;
+use serde::Serialize;
+use serde_json::Value;
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+use tokio::sync::{Notify, watch};
+use tokio_util::sync::CancellationToken;
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct MonitorKey {
+    pub plugin_id: String,
+    pub version: String,
+    pub window: String,
+    pub session: String,
+    pub monitor_id: String,
+}
+
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MonitorSnapshot {
+    pub revision: u64,
+    pub session_id: String,
+    pub overview: Option<RemoteGpuOverview>,
+    pub error: bool,
+    pub refreshing: bool,
+    pub paused: bool,
+}
+
+#[async_trait]
+pub trait Collector: Send + Sync {
+    async fn collect(&self, scope: &str) -> Result<Value>;
+    fn status(&self, _message: &str) {}
+}
+
+struct Demand {
+    token: String,
+    cancellation: CancellationToken,
+    interval: Duration,
+}
+
+pub struct MonitorTask {
+    demands: Mutex<HashMap<String, Demand>>,
+    paused: Mutex<bool>,
+    manual: Mutex<bool>,
+    in_flight: Mutex<Option<CancellationToken>>,
+    wake: Notify,
+    stop: CancellationToken,
+    snapshot: watch::Sender<MonitorSnapshot>,
+}
+
+impl MonitorTask {
+    pub fn new(session: String, collector: Arc<dyn Collector>) -> Arc<Self> {
+        let (snapshot, _) = watch::channel(MonitorSnapshot {
+            revision: 0,
+            session_id: session,
+            overview: None,
+            error: false,
+            refreshing: false,
+            paused: false,
+        });
+        let task = Arc::new(Self {
+            demands: Mutex::new(HashMap::new()),
+            paused: Mutex::new(false),
+            manual: Mutex::new(false),
+            wake: Notify::new(),
+            in_flight: Mutex::new(None),
+            stop: CancellationToken::new(),
+            snapshot,
+        });
+        let running = task.clone();
+        tokio::spawn(async move {
+            running.run(collector).await;
+        });
+        task
+    }
+
+    pub fn add(&self, id: String, token: String, cancellation: CancellationToken, seconds: u64) {
+        self.demands.lock().unwrap().insert(
+            id,
+            Demand {
+                token,
+                cancellation,
+                interval: Duration::from_secs(seconds.clamp(3, 3600)),
+            },
+        );
+        self.wake.notify_one();
+    }
+
+    pub fn remove(&self, id: &str) {
+        let mut demands = self.demands.lock().unwrap();
+        demands.remove(id);
+        if demands.is_empty() {
+            self.stop.cancel();
+        }
+        self.wake.notify_one();
+    }
+
+    pub fn is_stopped(&self) -> bool {
+        self.stop.is_cancelled()
+    }
+    pub fn cancel(&self) {
+        self.stop.cancel();
+    }
+    pub fn snapshot(&self) -> MonitorSnapshot {
+        self.snapshot.borrow().clone()
+    }
+    pub fn watch(&self) -> watch::Receiver<MonitorSnapshot> {
+        self.snapshot.subscribe()
+    }
+    pub fn refresh(&self) {
+        *self.paused.lock().unwrap() = false;
+        *self.manual.lock().unwrap() = true;
+        self.wake.notify_one();
+    }
+    pub fn pause(&self) {
+        *self.paused.lock().unwrap() = true;
+        *self.manual.lock().unwrap() = false;
+        if let Some(cancellation) = self.in_flight.lock().unwrap().as_ref() {
+            cancellation.cancel();
+        }
+        self.snapshot.send_modify(|s| {
+            s.paused = true;
+            s.refreshing = false;
+            s.revision += 1;
+        });
+        self.wake.notify_one();
+    }
+
+    async fn run(self: Arc<Self>, collector: Arc<dyn Collector>) {
+        let mut failures = 0;
+        let mut started = false;
+        let mut unavailable = false;
+        let mut last_completion: Option<tokio::time::Instant> = None;
+        loop {
+            let demand = {
+                let mut demands = self.demands.lock().unwrap();
+                demands.retain(|_, d| !d.cancellation.is_cancelled());
+                demands
+                    .values()
+                    .min_by_key(|d| d.interval)
+                    .map(|d| (d.token.clone(), d.cancellation.clone(), d.interval))
+            };
+            let Some((token, cancellation, interval)) = demand else {
+                if self.stop.is_cancelled() {
+                    return;
+                }
+                tokio::select! { () = self.stop.cancelled() => return, () = self.wake.notified() => continue }
+            };
+            let paused = *self.paused.lock().unwrap();
+            let manual = *self.manual.lock().unwrap();
+            let next =
+                last_completion.map_or_else(tokio::time::Instant::now, |last| last + interval);
+            if (paused || unavailable) && !manual {
+                tokio::select! {
+                    () = self.stop.cancelled() => return,
+                    () = cancellation.cancelled() => continue,
+                    () = self.wake.notified() => continue,
+                }
+            }
+            if !manual && tokio::time::Instant::now() < next {
+                tokio::select! {
+                    () = self.stop.cancelled() => return,
+                    () = cancellation.cancelled() => continue,
+                    () = self.wake.notified() => continue,
+                    () = tokio::time::sleep_until(next) => {},
+                }
+            }
+            // Pause can arrive while waiting; a wake must not issue another tick.
+            if *self.paused.lock().unwrap() {
+                continue;
+            }
+            *self.manual.lock().unwrap() = false;
+            let resuming = self.snapshot.borrow().paused;
+            let collection = CancellationToken::new();
+            *self.in_flight.lock().unwrap() = Some(collection.clone());
+            if *self.paused.lock().unwrap() {
+                collection.cancel();
+            }
+            self.snapshot.send_modify(|s| {
+                s.refreshing = true;
+                s.paused = false;
+                s.revision += 1;
+            });
+            let result = tokio::select! {
+                biased;
+                () = self.stop.cancelled() => return,
+                () = collection.cancelled() => {
+                    *self.in_flight.lock().unwrap() = None;
+                    self.snapshot.send_modify(|s| { s.refreshing = false; s.paused = *self.paused.lock().unwrap(); s.revision += 1; });
+                    collector.status("Monitor collection paused");
+                    continue;
+                },
+                () = cancellation.cancelled() => { *self.in_flight.lock().unwrap() = None; last_completion = None; continue; },
+                result = tokio::time::timeout(Duration::from_secs(45), collector.collect(&token)) =>
+                    result.unwrap_or_else(|_| Err(invalid("Monitor collection timed out"))),
+            }.and_then(validate_gpu);
+            *self.in_flight.lock().unwrap() = None;
+            if self.stop.is_cancelled() || cancellation.is_cancelled() || collection.is_cancelled()
+            {
+                continue;
+            }
+            if !started {
+                collector.status("Monitor collection started");
+                started = true;
+            }
+            match result {
+                Ok(overview) => {
+                    if failures > 0 || resuming {
+                        collector.status("Monitor collection recovered");
+                    }
+                    failures = 0;
+                    unavailable = !overview.available;
+                    self.snapshot.send_modify(|s| {
+                        s.overview = Some(overview);
+                        s.error = false;
+                        s.refreshing = false;
+                        s.revision += 1;
+                    });
+                }
+                Err(_) => {
+                    if failures == 0 {
+                        collector.status("Monitor collection failed");
+                    }
+                    failures += 1;
+                    self.snapshot.send_modify(|s| {
+                        if failures >= 3 {
+                            s.overview = None;
+                        }
+                        s.error = true;
+                        s.refreshing = false;
+                        s.revision += 1;
+                    });
+                }
+            }
+            // Schedule after completion: no overlapping requests or catch-up bursts.
+            last_completion = Some(tokio::time::Instant::now());
+        }
+    }
+}
+
+pub fn validate_gpu(value: Value) -> Result<RemoteGpuOverview> {
+    if serde_json::to_vec(&value)?.len() > 1024 * 1024 {
+        return Err(invalid("Monitor result exceeds 1 MiB"));
+    }
+    let overview: RemoteGpuOverview = serde_json::from_value(value)?;
+    if overview.gpus.len() > 128
+        || overview.processes.len() > 4096
+        || overview.driver_version.len() > 256
+        || overview.cuda_version.len() > 256
+        || overview
+            .gpus
+            .iter()
+            .any(|g| g.name.len() > 1024 || g.uuid.len() > 256 || g.pstate.len() > 128)
+        || overview
+            .processes
+            .iter()
+            .any(|p| p.process_name.len() > 4096 || p.gpu_uuid.len() > 256)
+    {
+        return Err(invalid("Monitor result exceeds GPU schema limits"));
+    }
+    Ok(overview)
+}

src-tauri/crates/nyaterm-plugin-runtime/src/package.rs

@@ -18,6 +18,7 @@ pub struct PackagePreview {
     pub manifest: Manifest,
     pub digest: String,
     pub expanded_bytes: u64,
+    pub probe_scripts: BTreeMap<String, String>,
 }
 
 pub struct PreparedPackage {
@@ -120,6 +121,18 @@ pub fn prepare(path: &Path, staging_parent: &Path, app_version: &str) -> Result<
     for panel in &manifest.contributions.panels {
         checked_file(directory.path(), &panel.entry)?;
     }
+    let mut probe_scripts = BTreeMap::new();
+    for probe in &manifest.contributions.probes {
+        let path = checked_file(directory.path(), &probe.entry)?;
+        if std::fs::metadata(&path)?.len() > 64 * 1024 {
+            return Err(invalid("Probe script exceeds 64 KiB"));
+        }
+        let script = std::fs::read_to_string(path)?;
+        if script.is_empty() || script.contains('\0') {
+            return Err(invalid("Probe script must be nonempty UTF-8 without NUL"));
+        }
+        probe_scripts.insert(probe.id.clone(), script);
+    }
     if let Some(backend) = &manifest.backend {
         for executable in backend.executables.values() {
             let path = checked_file(directory.path(), executable)?;
@@ -137,6 +150,7 @@ pub fn prepare(path: &Path, staging_parent: &Path, app_version: &str) -> Result<
             manifest,
             digest: package_digest,
             expanded_bytes,
+            probe_scripts,
         },
         directory,
     })

src-tauri/crates/nyaterm-plugin-runtime/src/probe.rs

@@ -0,0 +1,93 @@
+//! Fixed, reviewed package scripts. Caller supplies identity, never shell text.
+use crate::{Result, invalid, package::checked_file};
+use serde::{Deserialize, Serialize};
+use std::path::Path;
+use std::time::Duration;
+use tokio_util::sync::CancellationToken;
+
+#[async_trait::async_trait]
+pub trait ProbeExecutor: Send + Sync {
+    async fn execute(
+        &self,
+        script: &str,
+        timeout: Duration,
+        cancellation: CancellationToken,
+    ) -> Result<ProbeOutput>;
+}
+
+pub async fn execute_declared(
+    executor: &dyn ProbeExecutor,
+    root: &Path,
+    manifest: &crate::manifest::Manifest,
+    request: ProbeRequest,
+    cancellation: CancellationToken,
+) -> Result<ProbeOutput> {
+    if cancellation.is_cancelled() {
+        return Err(invalid("Probe cancelled"));
+    }
+    let probe = manifest
+        .contributions
+        .probes
+        .iter()
+        .find(|p| p.id == request.probe_id)
+        .ok_or_else(|| invalid("Unknown declared probe"))?;
+    let script = read_script(root, probe)?;
+    let timeout = Duration::from_millis(probe.timeout_ms);
+    let result = tokio::select! {
+        biased;
+        () = cancellation.cancelled() => return Err(invalid("Probe cancelled")),
+        result = tokio::time::timeout(timeout, executor.execute(&script, timeout, cancellation.clone())) =>
+            result.map_err(|_| invalid("Probe timed out"))??,
+    };
+    if result.stdout.len() + result.stderr.len() > 1024 * 1024 {
+        return Err(invalid("Probe output exceeds 1 MiB"));
+    }
+    Ok(result)
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct ProbeRequest {
+    pub probe_id: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ProbeOutput {
+    pub stdout: String,
+    pub stderr: String,
+    pub exit_status: Option<u32>,
+}
+
+pub fn read_script(root: &Path, probe: &crate::manifest::Probe) -> Result<String> {
+    let path = checked_file(root, &probe.entry)?;
+    if std::fs::metadata(&path)?.len() > 64 * 1024 {
+        return Err(invalid("Probe script exceeds 64 KiB"));
+    }
+    use std::io::Read;
+    let mut bytes = Vec::new();
+    std::fs::File::open(path)?
+        .take(64 * 1024 + 1)
+        .read_to_end(&mut bytes)?;
+    if bytes.len() > 64 * 1024 || bytes.is_empty() || bytes.contains(&0) {
+        return Err(invalid("Invalid probe script"));
+    }
+    String::from_utf8(bytes).map_err(|_| invalid("Probe script must be UTF-8"))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn rejects_shell_and_cross_session_overrides() {
+        for input in [
+            r#"{"probeId":"gpu","command":"rm -rf /"}"#,
+            r#"{"probeId":"gpu","sessionId":"other"}"#,
+            r#"{"probeId":"gpu","stdin":"password"}"#,
+            r#"{"probeId":"gpu","args":["--evil"]}"#,
+        ] {
+            assert!(serde_json::from_str::<ProbeRequest>(input).is_err());
+        }
+        assert!(serde_json::from_str::<ProbeRequest>(r#"{"probeId":"gpu"}"#).is_ok());
+    }
+}

src-tauri/crates/nyaterm-plugin-runtime/src/registry.rs

@@ -202,10 +202,11 @@ impl Registry {
         // New versions are always reviewed before reactivation, including permission changes.
         plugin.enabled = false;
         plugin.granted_permissions.retain(|p| {
-            plugin.versions[&plugin.active_version]
-                .manifest
-                .permissions
-                .contains(p)
+            p != "remote.probe"
+                && plugin.versions[&plugin.active_version]
+                    .manifest
+                    .permissions
+                    .contains(p)
         });
         let installed = plugin.clone();
         if let Err(error) = self.save(next) {
@@ -241,6 +242,39 @@ impl Registry {
         }
         plugin.enabled = enabled;
         plugin.granted_permissions = permissions;
+        if enabled
+            && plugin
+                .granted_permissions
+                .iter()
+                .any(|p| p == "remote.probe")
+        {
+            let schemas = plugin
+                .active()?
+                .manifest
+                .contributions
+                .monitors
+                .iter()
+                .map(|m| m.schema.clone())
+                .collect::<Vec<_>>();
+            for other in next.values().filter(|p| p.id != id && p.enabled) {
+                if other
+                    .granted_permissions
+                    .iter()
+                    .any(|p| p == "remote.probe")
+                    && other
+                        .active()?
+                        .manifest
+                        .contributions
+                        .monitors
+                        .iter()
+                        .any(|m| schemas.contains(&m.schema))
+                {
+                    return Err(invalid(
+                        "Another enabled plugin already supplies this monitor schema; disable it first",
+                    ));
+                }
+            }
+        }
         self.save(next)
     }
 
@@ -258,7 +292,7 @@ impl Registry {
         plugin.enabled = false;
         plugin
             .granted_permissions
-            .retain(|p| record.manifest.permissions.contains(p));
+            .retain(|p| p != "remote.probe" && record.manifest.permissions.contains(p));
         self.save(next)
     }
 

src-tauri/crates/nyaterm-plugin-runtime/tests/gpu_plugin.rs

@@ -0,0 +1,178 @@
+#![cfg(feature = "sdk-fixture")]
+use nyaterm_plugin_runtime::{
+    Result,
+    manifest::Manifest,
+    monitoring::validate_gpu,
+    sidecar::{HostHandler, Sidecar},
+};
+use serde_json::{Value, json};
+use std::{
+    sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    },
+    time::Duration,
+};
+
+struct Host(AtomicUsize);
+#[async_trait::async_trait]
+impl HostHandler for Host {
+    async fn call(&self, method: &str, params: Value) -> Result<Value> {
+        assert_eq!(method, "host/remote/probe");
+        assert_eq!(
+            params,
+            json!({"scopeToken":"gpu-scope","input":{"probeId":"gpu-overview"}})
+        );
+        self.0.fetch_add(1, Ordering::SeqCst);
+        Ok(json!({"stdout":concat!(
+            "GPU_AVAILABLE\t1\nGPU_CUDA_VERSION\t12.4\nGPU_CSV_BEGIN\n",
+            "0, GPU-a, NVIDIA RTX 4090, 550.54, 43, 77, 31, 24564, 12345, 12219, 180.5, 450, 46, P2\n",
+            "GPU_CSV_END\nGPU_PROCESS_CSV_BEGIN\nGPU-a, 4242, 2048, python\nGPU_PROCESS_CSV_END\n"
+        ),"stderr":"","exitStatus":0}))
+    }
+}
+
+#[tokio::test]
+async fn real_gpu_backend_reuses_process_and_parses_reverse_probe_response() {
+    let root = tempfile::tempdir().unwrap();
+    std::fs::create_dir(root.path().join("bin")).unwrap();
+    let executable = if cfg!(windows) {
+        "bin/gpu.exe"
+    } else {
+        "bin/gpu"
+    };
+    std::fs::copy(
+        env!("CARGO_BIN_EXE_gpu-fixture"),
+        root.path().join(executable),
+    )
+    .unwrap();
+    let mut value: Value = serde_json::from_slice(include_bytes!(
+        "../../../../plugins/examples/gpu-monitor/manifest.json"
+    ))
+    .unwrap();
+    value["backend"]["executables"] =
+        json!({format!("{}-{}", std::env::consts::OS, std::env::consts::ARCH):executable});
+    let manifest = Manifest::parse(&serde_json::to_vec(&value).unwrap(), "1.2.12").unwrap();
+    let host = Arc::new(Host(AtomicUsize::new(0)));
+    let sidecar = Sidecar::spawn(
+        root.path(),
+        &manifest,
+        "1.2.12",
+        &["native".into(), "remote.probe".into()],
+        host.clone(),
+    )
+    .await
+    .unwrap();
+    for _ in 0..2 {
+        let value = sidecar
+            .request(
+                "monitor/collect",
+                json!({"scopeToken":"gpu-scope","input":null}),
+                Duration::from_secs(5),
+            )
+            .await
+            .unwrap();
+        let overview = validate_gpu(value).unwrap();
+        assert_eq!(overview.gpus[0].name, "NVIDIA RTX 4090");
+        assert_eq!(overview.gpus[0].utilization_gpu_percent, Some(77.0));
+        assert_eq!(overview.processes[0].gpu_index, Some(0));
+    }
+    assert_eq!(host.0.load(Ordering::SeqCst), 2);
+    sidecar.stop();
+}
+
+/// Opt-in verification of the final archive, not the debug fixture binary.
+#[tokio::test]
+#[ignore = "Build/package GPU plugin and set NYATERM_GPU_PACKAGE"]
+async fn installed_release_package_collects_using_its_fixed_probe() {
+    use nyaterm_plugin_runtime::{
+        package::prepare,
+        probe::{ProbeExecutor, ProbeOutput, ProbeRequest, execute_declared},
+        registry::Registry,
+    };
+    use tokio_util::sync::CancellationToken;
+    struct FixedExecutor;
+    #[async_trait::async_trait]
+    impl ProbeExecutor for FixedExecutor {
+        async fn execute(
+            &self,
+            script: &str,
+            _: Duration,
+            _: CancellationToken,
+        ) -> Result<ProbeOutput> {
+            assert_eq!(
+                script,
+                include_str!("../../../../plugins/examples/gpu-monitor/assets/probes/gpu.sh")
+            );
+            assert!(script.contains("nvidia-smi"));
+            Ok(ProbeOutput { stdout: "GPU_AVAILABLE\t1\nGPU_CUDA_VERSION\t12.4\nGPU_CSV_BEGIN\n0, GPU-a, NVIDIA RTX 4090, 550.54, 43, 77, 31, 24564, 12345, 12219, 180.5, 450, 46, P2\nGPU_CSV_END\nGPU_PROCESS_CSV_BEGIN\nGPU-a, 4242, 2048, python\nGPU_PROCESS_CSV_END\n".into(), stderr: String::new(), exit_status: Some(0) })
+        }
+    }
+    struct InstalledHost {
+        root: std::path::PathBuf,
+        manifest: Manifest,
+        calls: AtomicUsize,
+    }
+    #[async_trait::async_trait]
+    impl HostHandler for InstalledHost {
+        async fn call(&self, method: &str, params: Value) -> Result<Value> {
+            assert_eq!(method, "host/remote/probe");
+            assert_eq!(params["scopeToken"], "gpu-scope");
+            let request: ProbeRequest = serde_json::from_value(params["input"].clone())?;
+            let output = execute_declared(
+                &FixedExecutor,
+                &self.root,
+                &self.manifest,
+                request,
+                CancellationToken::new(),
+            )
+            .await?;
+            self.calls.fetch_add(1, Ordering::SeqCst);
+            Ok(serde_json::to_value(output)?)
+        }
+    }
+    let package = std::env::var_os("NYATERM_GPU_PACKAGE").expect("NYATERM_GPU_PACKAGE is required");
+    let temp = tempfile::tempdir().unwrap();
+    let prepared = prepare(std::path::Path::new(&package), temp.path(), "1.2.12").unwrap();
+    let digest = prepared.preview.digest.clone();
+    let mut registry = Registry::open(temp.path().join("plugins"), "1.2.12".into()).unwrap();
+    registry.install(prepared, &digest).unwrap();
+    let permissions = vec![
+        "native".into(),
+        "remote.probe".into(),
+    ];
+    registry
+        .configure("nyaterm.gpu", true, permissions.clone())
+        .unwrap();
+    let root = registry.active_path("nyaterm.gpu").unwrap();
+    let manifest = registry
+        .get("nyaterm.gpu")
+        .unwrap()
+        .active()
+        .unwrap()
+        .manifest
+        .clone();
+    let host = Arc::new(InstalledHost {
+        root: root.clone(),
+        manifest: manifest.clone(),
+        calls: AtomicUsize::new(0),
+    });
+    let sidecar = Sidecar::spawn(&root, &manifest, "1.2.12", &permissions, host.clone())
+        .await
+        .unwrap();
+    for _ in 0..2 {
+        let result = sidecar
+            .request(
+                "monitor/collect",
+                json!({"scopeToken":"gpu-scope","input":null}),
+                Duration::from_secs(5),
+            )
+            .await
+            .unwrap();
+        let overview = validate_gpu(result).unwrap();
+        assert_eq!(overview.gpus[0].name, "NVIDIA RTX 4090");
+        assert_eq!(overview.processes[0].pid, 4242);
+    }
+    assert_eq!(host.calls.load(Ordering::SeqCst), 2);
+    sidecar.stop();
+}

src-tauri/crates/nyaterm-plugin-runtime/tests/monitor_packages.rs

@@ -0,0 +1,111 @@
+use nyaterm_plugin_runtime::{manifest::Manifest, package::prepare, registry::Registry};
+use serde_json::{Value, json};
+use sha2::{Digest, Sha256};
+use std::{collections::BTreeMap, io::Write, path::Path};
+use zip::write::SimpleFileOptions;
+
+fn manifest(id: &str, version: &str) -> Value {
+    json!({
+        "manifestVersion":1,"id":id,"name":"GPU","version":version,"publisher":"Test","description":"GPU","engine":">=1",
+        "permissions":["native","remote.probe"],
+        "backend":{"transport":"stdio-jsonl","executables":{"windows-x86_64":"bin/backend.exe"}},
+        "contributions":{
+            "panels":[{"id":"overview","title":"GPU","entry":"ui/index.html"}],
+            "probes":[{"id":"probe","title":"GPU probe","entry":"assets/probes/gpu.sh","timeoutMs":15000}],
+            "monitors":[{"id":"gpu","title":"GPU","schema":"gpu.v1","method":"monitor/collect","panel":"overview"}]
+        }
+    })
+}
+fn package(path: &Path, manifest: &Value, script: &[u8]) {
+    let files = BTreeMap::from([
+        ("manifest.json", serde_json::to_vec(manifest).unwrap()),
+        ("ui/index.html", b"<html></html>".to_vec()),
+        ("assets/probes/gpu.sh", script.to_vec()),
+        ("bin/backend.exe", b"fixture".to_vec()),
+    ]);
+    let checksums: BTreeMap<_, _> = files
+        .iter()
+        .map(|(k, v)| (*k, hex::encode(Sha256::digest(v))))
+        .collect();
+    let mut zip = zip::ZipWriter::new(std::fs::File::create(path).unwrap());
+    for (name, data) in files {
+        zip.start_file(name, SimpleFileOptions::default()).unwrap();
+        zip.write_all(&data).unwrap();
+    }
+    zip.start_file("checksums.json", SimpleFileOptions::default())
+        .unwrap();
+    zip.write_all(&serde_json::to_vec(&checksums).unwrap())
+        .unwrap();
+    zip.finish().unwrap();
+}
+#[test]
+fn validates_monitor_references_permissions_and_paths() {
+    let valid = manifest("test.gpu", "1.0.0");
+    assert!(Manifest::parse(&serde_json::to_vec(&valid).unwrap(), "1.2.12").is_ok());
+    for (pointer, replacement) in [
+        (
+            "/contributions/probes/0/entry",
+            json!("assets/probes/../escape.sh"),
+        ),
+        ("/contributions/probes/0/entry", json!("ui/probe.sh")),
+        ("/contributions/probes/0/timeoutMs", json!(30001)),
+        ("/contributions/monitors/0/panel", json!("missing")),
+        ("/contributions/monitors/0/method", json!("ui/collect")),
+        ("/contributions/monitors/0/schema", json!("arbitrary")),
+        ("/permissions", json!(["native"])),
+    ] {
+        let mut invalid = valid.clone();
+        *invalid.pointer_mut(pointer).unwrap() = replacement;
+        assert!(
+            Manifest::parse(&serde_json::to_vec(&invalid).unwrap(), "1.2.12").is_err(),
+            "{pointer}"
+        );
+    }
+}
+
+#[test]
+fn installation_exposes_reviewed_source_and_rejects_bad_scripts() {
+    let root = tempfile::tempdir().unwrap();
+    let path = root.path().join("gpu.nyap");
+    let manifest = manifest("test.gpu", "1.0.0");
+    package(&path, &manifest, b"printf metrics\n");
+    let prepared = prepare(&path, root.path(), "1.2.12").unwrap();
+    assert_eq!(prepared.preview.probe_scripts["probe"], "printf metrics\n");
+    for script in [vec![b'x'; 64 * 1024 + 1], vec![0], vec![255], vec![]] {
+        package(&path, &manifest, &script);
+        assert!(prepare(&path, root.path(), "1.2.12").is_err());
+    }
+}
+
+#[test]
+fn monitor_conflicts_and_updates_cannot_reuse_execution_grants() {
+    let root = tempfile::tempdir().unwrap();
+    let mut registry = Registry::open(root.path().join("installed"), "1.2.12".into()).unwrap();
+    let path = root.path().join("gpu.nyap");
+    for id in ["test.gpu", "other.gpu"] {
+        package(&path, &manifest(id, "1.0.0"), b"printf metrics\n");
+        let prepared = prepare(&path, root.path(), "1.2.12").unwrap();
+        let digest = prepared.preview.digest.clone();
+        registry.install(prepared, &digest).unwrap();
+    }
+    let grants = || vec!["native".into(), "remote.probe".into()];
+    registry.configure("test.gpu", true, grants()).unwrap();
+    assert!(registry.configure("other.gpu", true, grants()).is_err());
+    assert!(registry.get("test.gpu").unwrap().enabled);
+    assert!(!registry.get("other.gpu").unwrap().enabled);
+    package(&path, &manifest("test.gpu", "1.0.1"), b"printf updated\n");
+    let prepared = prepare(&path, root.path(), "1.2.12").unwrap();
+    let digest = prepared.preview.digest.clone();
+    let updated = registry.install(prepared, &digest).unwrap();
+    assert!(!updated.enabled);
+    assert!(!updated.granted_permissions.contains(&"remote.probe".into()));
+    registry.configure("test.gpu", true, grants()).unwrap();
+    registry.activate_version("test.gpu", "1.0.0").unwrap();
+    assert!(
+        !registry
+            .get("test.gpu")
+            .unwrap()
+            .granted_permissions
+            .contains(&"remote.probe".into())
+    );
+}

src-tauri/crates/nyaterm-plugin-runtime/tests/monitoring.rs

@@ -0,0 +1,198 @@
+use async_trait::async_trait;
+use nyaterm_plugin_runtime::{
+    Error, Result,
+    monitoring::{Collector, MonitorTask, validate_gpu},
+};
+use serde_json::{Value, json};
+use std::{
+    collections::VecDeque,
+    sync::{
+        Arc, Mutex,
+        atomic::{AtomicUsize, Ordering},
+    },
+    time::Duration,
+};
+use tokio_util::sync::CancellationToken;
+
+struct Fake {
+    calls: AtomicUsize,
+    active: AtomicUsize,
+    peak: AtomicUsize,
+    delay: Duration,
+    results: Mutex<VecDeque<Result<Value>>>,
+}
+impl Fake {
+    fn new(results: Vec<Result<Value>>) -> Arc<Self> {
+        Arc::new(Self {
+            calls: AtomicUsize::new(0),
+            active: AtomicUsize::new(0),
+            peak: AtomicUsize::new(0),
+            delay: Duration::ZERO,
+            results: Mutex::new(results.into()),
+        })
+    }
+}
+struct Active<'a>(&'a AtomicUsize);
+impl Drop for Active<'_> {
+    fn drop(&mut self) {
+        self.0.fetch_sub(1, Ordering::SeqCst);
+    }
+}
+#[async_trait]
+impl Collector for Fake {
+    async fn collect(&self, scope: &str) -> Result<Value> {
+        assert!(scope.starts_with("scope-"));
+        self.calls.fetch_add(1, Ordering::SeqCst);
+        let active = self.active.fetch_add(1, Ordering::SeqCst) + 1;
+        self.peak.fetch_max(active, Ordering::SeqCst);
+        let _guard = Active(&self.active);
+        tokio::time::sleep(self.delay).await;
+        self.results
+            .lock()
+            .unwrap()
+            .pop_front()
+            .unwrap_or_else(|| Ok(overview(true)))
+    }
+}
+fn overview(available: bool) -> Value {
+    json!({"available":available,"driver_version":"550","cuda_version":"12","gpus":[],"processes":[]})
+}
+async fn settle() {
+    for _ in 0..24 {
+        tokio::task::yield_now().await;
+    }
+}
+async fn advance(seconds: u64) {
+    tokio::time::advance(Duration::from_secs(seconds)).await;
+    settle().await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn shares_demands_stops_after_last_and_never_catches_up() {
+    let fake = Arc::new(Fake {
+        delay: Duration::from_secs(5),
+        ..Arc::try_unwrap(Fake::new(vec![])).ok().unwrap()
+    });
+    let task = MonitorTask::new("ssh-1".into(), fake.clone());
+    task.add("a".into(), "scope-a".into(), CancellationToken::new(), 3);
+    task.add("b".into(), "scope-b".into(), CancellationToken::new(), 3);
+    settle().await;
+    assert_eq!(fake.calls.load(Ordering::SeqCst), 1);
+    advance(4).await;
+    assert_eq!(fake.calls.load(Ordering::SeqCst), 1);
+    advance(1).await;
+    assert!(task.snapshot().overview.is_some());
+    task.remove("a");
+    advance(3).await;
+    assert_eq!(fake.calls.load(Ordering::SeqCst), 2);
+    assert_eq!(fake.peak.load(Ordering::SeqCst), 1);
+    task.remove("b");
+    settle().await;
+    assert_eq!(fake.active.load(Ordering::SeqCst), 0);
+    advance(100).await;
+    assert_eq!(fake.calls.load(Ordering::SeqCst), 2);
+}
+
+#[tokio::test(start_paused = true)]
+async fn unavailable_and_pause_require_explicit_refresh() {
+    let fake = Fake::new(vec![Ok(overview(false)), Ok(overview(true))]);
+    let task = MonitorTask::new("ssh-1".into(), fake.clone());
+    task.add("a".into(), "scope-a".into(), CancellationToken::new(), 3);
+    settle().await;
+    advance(30).await;
+    assert_eq!(fake.calls.load(Ordering::SeqCst), 1);
+    task.refresh();
+    settle().await;
+    assert!(task.snapshot().overview.unwrap().available);
+    task.pause();
+    task.refresh();
+    task.pause();
+    advance(30).await;
+    assert!(task.snapshot().paused);
+    assert_eq!(fake.calls.load(Ordering::SeqCst), 2);
+    task.refresh();
+    settle().await;
+    assert!(!task.snapshot().paused);
+    assert_eq!(fake.calls.load(Ordering::SeqCst), 3);
+    task.cancel();
+}
+
+#[tokio::test(start_paused = true)]
+async fn failures_preserve_then_clear_and_can_recover() {
+    let fail = || Err(Error::Runtime("failure".into()));
+    let fake = Fake::new(vec![
+        Ok(overview(true)),
+        fail(),
+        fail(),
+        fail(),
+        Ok(overview(true)),
+    ]);
+    let task = MonitorTask::new("ssh-1".into(), fake);
+    task.add("a".into(), "scope-a".into(), CancellationToken::new(), 3);
+    settle().await;
+    let first = task.snapshot().revision;
+    advance(3).await;
+    assert!(task.snapshot().error && task.snapshot().overview.is_some());
+    advance(3).await;
+    assert!(task.snapshot().overview.is_some());
+    advance(3).await;
+    assert!(task.snapshot().overview.is_none());
+    task.refresh();
+    settle().await;
+    assert!(!task.snapshot().error && task.snapshot().overview.is_some());
+    assert!(task.snapshot().revision > first);
+    task.cancel();
+}
+
+#[tokio::test(start_paused = true)]
+async fn revocation_drops_inflight_and_rejects_late_publication() {
+    let fake = Arc::new(Fake {
+        delay: Duration::from_secs(5),
+        ..Arc::try_unwrap(Fake::new(vec![])).ok().unwrap()
+    });
+    let cancellation = CancellationToken::new();
+    let task = MonitorTask::new("ssh-1".into(), fake.clone());
+    task.add("a".into(), "scope-a".into(), cancellation.clone(), 3);
+    settle().await;
+    cancellation.cancel();
+    task.remove("a");
+    settle().await;
+    assert_eq!(fake.active.load(Ordering::SeqCst), 0);
+    advance(30).await;
+    assert!(task.snapshot().overview.is_none());
+}
+
+#[test]
+fn rejects_untrusted_schema_and_excessive_results() {
+    let mut value = overview(true);
+    value["connectionId"] = json!("other");
+    assert!(validate_gpu(value).is_err());
+    let mut value = overview(true);
+    value["processes"] = json!([{"gpu_uuid":"GPU-a","gpu_index":0,"pid":1,"process_name":"x".repeat(4097),"used_memory_mb":0}]);
+    assert!(validate_gpu(value).is_err());
+    assert!(validate_gpu(json!({"available":true})).is_err());
+}
+
+#[tokio::test(start_paused = true)]
+async fn stopping_backend_cancels_collection_and_new_demands_remain_paused() {
+    let fake = Arc::new(Fake {
+        delay: Duration::from_secs(5),
+        ..Arc::try_unwrap(Fake::new(vec![])).ok().unwrap()
+    });
+    let task = MonitorTask::new("ssh-1".into(), fake.clone());
+    task.add("a".into(), "scope-a".into(), CancellationToken::new(), 3);
+    settle().await;
+    task.pause();
+    settle().await;
+    assert_eq!(fake.active.load(Ordering::SeqCst), 0);
+    task.add("b".into(), "scope-b".into(), CancellationToken::new(), 3);
+    advance(60).await;
+    assert_eq!(fake.calls.load(Ordering::SeqCst), 1);
+    assert!(task.snapshot().paused && task.snapshot().overview.is_none());
+    task.refresh();
+    settle().await;
+    assert_eq!(fake.calls.load(Ordering::SeqCst), 2);
+    advance(5).await;
+    assert!(task.snapshot().overview.is_some());
+    task.cancel();
+}

src-tauri/crates/nyaterm-plugin-runtime/tests/probes.rs

@@ -0,0 +1,173 @@
+use nyaterm_plugin_runtime::{
+    Result,
+    manifest::Manifest,
+    probe::{ProbeExecutor, ProbeOutput, ProbeRequest, execute_declared},
+};
+use serde_json::json;
+use std::{
+    sync::{
+        Mutex,
+        atomic::{AtomicUsize, Ordering},
+    },
+    time::Duration,
+};
+use tokio_util::sync::CancellationToken;
+
+struct Executor {
+    scripts: Mutex<Vec<String>>,
+    mode: u8,
+    active: AtomicUsize,
+}
+struct Active<'a>(&'a AtomicUsize);
+impl Drop for Active<'_> {
+    fn drop(&mut self) {
+        self.0.fetch_sub(1, Ordering::SeqCst);
+    }
+}
+#[async_trait::async_trait]
+impl ProbeExecutor for Executor {
+    async fn execute(
+        &self,
+        script: &str,
+        _: Duration,
+        _: CancellationToken,
+    ) -> Result<ProbeOutput> {
+        self.scripts.lock().unwrap().push(script.into());
+        self.active.fetch_add(1, Ordering::SeqCst);
+        let _guard = Active(&self.active);
+        if self.mode == 2 {
+            std::future::pending::<()>().await;
+        }
+        Ok(ProbeOutput {
+            stdout: if self.mode == 1 {
+                "x".repeat(1024 * 1024)
+            } else {
+                "metrics".into()
+            },
+            stderr: "err".into(),
+            exit_status: Some(0),
+        })
+    }
+}
+fn fixture() -> (tempfile::TempDir, Manifest) {
+    let root = tempfile::tempdir().unwrap();
+    std::fs::create_dir_all(root.path().join("assets/probes")).unwrap();
+    std::fs::write(root.path().join("assets/probes/gpu.sh"), "printf metrics\n").unwrap();
+    let manifest = Manifest::parse(&serde_json::to_vec(&json!({
+        "manifestVersion":1,"id":"test.gpu","name":"Test","version":"1.0.0","publisher":"Test","description":"Test","engine":">=1",
+        "permissions":["remote.probe"],"contributions":{"probes":[{"id":"gpu","title":"GPU","entry":"assets/probes/gpu.sh","timeoutMs":1000}]}
+    })).unwrap(), "1.2.12").unwrap();
+    (root, manifest)
+}
+
+#[tokio::test]
+async fn runs_only_the_declared_script_and_bounds_combined_output() {
+    let (root, manifest) = fixture();
+    let executor = Executor {
+        scripts: Mutex::new(vec![]),
+        mode: 0,
+        active: AtomicUsize::new(0),
+    };
+    let request = || ProbeRequest {
+        probe_id: "gpu".into(),
+    };
+    let result = execute_declared(
+        &executor,
+        root.path(),
+        &manifest,
+        request(),
+        CancellationToken::new(),
+    )
+    .await
+    .unwrap();
+    assert_eq!(result.exit_status, Some(0));
+    assert_eq!(*executor.scripts.lock().unwrap(), vec!["printf metrics\n"]);
+    assert!(
+        execute_declared(
+            &executor,
+            root.path(),
+            &manifest,
+            ProbeRequest {
+                probe_id: "other".into()
+            },
+            CancellationToken::new()
+        )
+        .await
+        .is_err()
+    );
+    assert_eq!(executor.scripts.lock().unwrap().len(), 1);
+    let huge = Executor {
+        mode: 1,
+        ..executor
+    };
+    assert!(
+        execute_declared(
+            &huge,
+            root.path(),
+            &manifest,
+            request(),
+            CancellationToken::new()
+        )
+        .await
+        .is_err()
+    );
+}
+
+#[tokio::test(start_paused = true)]
+async fn timeout_and_revocation_drop_only_the_execution_future() {
+    let (root, manifest) = fixture();
+    let executor = Executor {
+        scripts: Mutex::new(vec![]),
+        mode: 2,
+        active: AtomicUsize::new(0),
+    };
+    assert!(
+        execute_declared(
+            &executor,
+            root.path(),
+            &manifest,
+            ProbeRequest {
+                probe_id: "gpu".into()
+            },
+            CancellationToken::new()
+        )
+        .await
+        .unwrap_err()
+        .to_string()
+        .contains("timed out")
+    );
+    assert_eq!(executor.active.load(Ordering::SeqCst), 0);
+    let cancellation = CancellationToken::new();
+    let operation = execute_declared(
+        &executor,
+        root.path(),
+        &manifest,
+        ProbeRequest {
+            probe_id: "gpu".into(),
+        },
+        cancellation.clone(),
+    );
+    tokio::pin!(operation);
+    tokio::select! {
+        result = &mut operation => panic!("Cancelled before the test revoked execution: {result:?}"),
+        () = async { tokio::task::yield_now().await; assert_eq!(executor.active.load(Ordering::SeqCst), 1); cancellation.cancel(); } => {},
+    }
+    assert!(operation.await.is_err());
+    assert_eq!(executor.active.load(Ordering::SeqCst), 0);
+    let token = CancellationToken::new();
+    token.cancel();
+    assert!(
+        execute_declared(
+            &executor,
+            root.path(),
+            &manifest,
+            ProbeRequest {
+                probe_id: "gpu".into()
+            },
+            token
+        )
+        .await
+        .is_err()
+    );
+    assert_eq!(executor.active.load(Ordering::SeqCst), 0);
+}

src-tauri/src/cmd/plugins.rs

@@ -160,3 +160,53 @@ pub async fn stop_plugin_backend(
 ) -> AppResult<()> {
     manager.stop_backend(&app, &plugin_id).await
 }
+
+#[tauri::command]
+pub async fn subscribe_plugin_monitor(
+    app: tauri::AppHandle,
+    window: tauri::WebviewWindow,
+    manager: tauri::State<'_, Arc<PluginManager>>,
+    token: String,
+    monitor_id: String,
+    interval_seconds: u64,
+) -> AppResult<crate::core::plugins::MonitorSubscription> {
+    manager
+        .inner()
+        .subscribe_monitor(&app, window.label(), &token, &monitor_id, interval_seconds)
+        .await
+}
+
+#[tauri::command]
+pub fn unsubscribe_plugin_monitor(
+    window: tauri::WebviewWindow,
+    manager: tauri::State<'_, Arc<PluginManager>>,
+    token: String,
+    subscription_id: String,
+) -> AppResult<()> {
+    manager.unsubscribe_monitor(window.label(), &token, &subscription_id)
+}
+
+#[tauri::command]
+pub async fn refresh_plugin_monitor(
+    app: tauri::AppHandle,
+    window: tauri::WebviewWindow,
+    manager: tauri::State<'_, Arc<PluginManager>>,
+    token: String,
+    subscription_id: String,
+) -> AppResult<()> {
+    manager
+        .refresh_monitor(&app, window.label(), &token, &subscription_id)
+        .await
+}
+
+#[tauri::command]
+pub async fn get_plugin_probe_scripts(
+    app: tauri::AppHandle,
+    manager: tauri::State<'_, Arc<PluginManager>>,
+    plugin_id: String,
+    expected_version: String,
+) -> AppResult<std::collections::HashMap<String, String>> {
+    manager
+        .probe_scripts(&app, &plugin_id, &expected_version)
+        .await
+}

src-tauri/src/core/monitoring/gpu.rs

@@ -1,40 +1,4 @@
-use serde::Serialize;
-use std::collections::HashMap;
-
-#[derive(Debug, Clone, Serialize, Default, PartialEq)]
-pub struct RemoteGpuOverview {
-    pub available: bool,
-    pub driver_version: String,
-    pub cuda_version: String,
-    pub gpus: Vec<RemoteGpu>,
-    pub processes: Vec<RemoteGpuProcess>,
-}
-
-#[derive(Debug, Clone, Serialize, PartialEq)]
-pub struct RemoteGpu {
-    pub index: u32,
-    pub uuid: String,
-    pub name: String,
-    pub temperature_c: Option<f64>,
-    pub utilization_gpu_percent: Option<f64>,
-    pub utilization_memory_percent: Option<f64>,
-    pub memory_total_mb: u64,
-    pub memory_used_mb: u64,
-    pub memory_free_mb: u64,
-    pub power_draw_w: Option<f64>,
-    pub power_limit_w: Option<f64>,
-    pub fan_speed_percent: Option<f64>,
-    pub pstate: String,
-}
-
-#[derive(Debug, Clone, Serialize, PartialEq)]
-pub struct RemoteGpuProcess {
-    pub gpu_uuid: String,
-    pub gpu_index: Option<u32>,
-    pub pid: u32,
-    pub process_name: String,
-    pub used_memory_mb: u64,
-}
+pub use nyaterm_gpu::{RemoteGpuOverview, parse_gpu_overview_output};
 
 pub const GPU_OVERVIEW_SCRIPT: &str = r#"sh -s <<'NYATERM_GPU_SCRIPT'
 LC_ALL=C
@@ -73,295 +37,3 @@ fi
 printf "GPU_PROCESS_CSV_END\n"
 NYATERM_GPU_SCRIPT
 "#;
-
-enum GpuParseSection {
-    None,
-    Gpus,
-    Processes,
-}
-
-pub fn parse_gpu_overview_output(output: &str) -> RemoteGpuOverview {
-    let mut overview = RemoteGpuOverview::default();
-    let mut section = GpuParseSection::None;
-    let mut process_rows = Vec::new();
-
-    for line in output.lines() {
-        match line {
-            "GPU_CSV_BEGIN" => {
-                section = GpuParseSection::Gpus;
-                continue;
-            }
-            "GPU_CSV_END" => {
-                section = GpuParseSection::None;
-                continue;
-            }
-            "GPU_PROCESS_CSV_BEGIN" => {
-                section = GpuParseSection::Processes;
-                continue;
-            }
-            "GPU_PROCESS_CSV_END" => {
-                section = GpuParseSection::None;
-                continue;
-            }
-            _ => {}
-        }
-
-        let cols: Vec<&str> = line.split('\t').collect();
-        if cols.first() == Some(&"GPU_AVAILABLE") && cols.len() >= 2 {
-            overview.available = cols[1] == "1";
-            continue;
-        }
-        if cols.first() == Some(&"GPU_CUDA_VERSION") && cols.len() >= 2 {
-            overview.cuda_version = cols[1].trim().to_string();
-            continue;
-        }
-
-        match section {
-            GpuParseSection::Gpus => {
-                if let Some(gpu) = parse_gpu_csv_line(line) {
-                    if overview.driver_version.is_empty() {
-                        overview.driver_version = gpu.1;
-                    }
-                    overview.gpus.push(gpu.0);
-                }
-            }
-            GpuParseSection::Processes => {
-                if !line.trim().is_empty() {
-                    process_rows.push(line.to_string());
-                }
-            }
-            GpuParseSection::None => {}
-        }
-    }
-
-    let gpu_indexes: HashMap<String, u32> = overview
-        .gpus
-        .iter()
-        .map(|gpu| (gpu.uuid.clone(), gpu.index))
-        .collect();
-    overview.processes = process_rows
-        .iter()
-        .filter_map(|line| parse_gpu_process_csv_line(line, &gpu_indexes))
-        .collect();
-
-    overview
-}
-
-fn parse_gpu_csv_line(line: &str) -> Option<(RemoteGpu, String)> {
-    let cols = parse_csv_line(line);
-    if cols.len() < 14 {
-        return None;
-    }
-
-    let gpu = RemoteGpu {
-        index: parse_u32(&cols[0]).unwrap_or(0),
-        uuid: cols[1].trim().to_string(),
-        name: cols[2].trim().to_string(),
-        temperature_c: parse_optional_f64(&cols[4]),
-        utilization_gpu_percent: parse_optional_f64(&cols[5]),
-        utilization_memory_percent: parse_optional_f64(&cols[6]),
-        memory_total_mb: parse_u64(&cols[7]).unwrap_or(0),
-        memory_used_mb: parse_u64(&cols[8]).unwrap_or(0),
-        memory_free_mb: parse_u64(&cols[9]).unwrap_or(0),
-        power_draw_w: parse_optional_f64(&cols[10]),
-        power_limit_w: parse_optional_f64(&cols[11]),
-        fan_speed_percent: parse_optional_f64(&cols[12]),
-        pstate: clean_text(&cols[13]),
-    };
-
-    Some((gpu, clean_text(&cols[3])))
-}
-
-fn parse_gpu_process_csv_line(
-    line: &str,
-    gpu_indexes: &HashMap<String, u32>,
-) -> Option<RemoteGpuProcess> {
-    let cols = parse_csv_line(line);
-    if cols.len() < 4 {
-        return None;
-    }
-
-    let gpu_uuid = clean_text(&cols[0]);
-    if gpu_uuid.is_empty() || is_missing(&gpu_uuid) {
-        return None;
-    }
-
-    Some(RemoteGpuProcess {
-        gpu_index: gpu_indexes.get(&gpu_uuid).copied(),
-        gpu_uuid,
-        pid: parse_u32(&cols[1]).unwrap_or(0),
-        used_memory_mb: parse_u64(&cols[2]).unwrap_or(0),
-        process_name: cols[3..].join(",").trim().to_string(),
-    })
-}
-
-fn parse_csv_line(line: &str) -> Vec<String> {
-    let chars: Vec<char> = line.chars().collect();
-    let mut values = Vec::new();
-    let mut current = String::new();
-    let mut in_quotes = false;
-    let mut index = 0;
-
-    while index < chars.len() {
-        let ch = chars[index];
-        match ch {
-            '"' if in_quotes && chars.get(index + 1) == Some(&'"') => {
-                current.push('"');
-                index += 1;
-            }
-            '"' => {
-                in_quotes = !in_quotes;
-            }
-            ',' if !in_quotes => {
-                values.push(current.trim().to_string());
-                current.clear();
-            }
-            _ => current.push(ch),
-        }
-        index += 1;
-    }
-
-    values.push(current.trim().to_string());
-    values
-}
-
-fn clean_text(value: &str) -> String {
-    value.trim().to_string()
-}
-
-fn is_missing(value: &str) -> bool {
-    let normalized = value.trim().to_ascii_lowercase();
-    matches!(
-        normalized.as_str(),
-        "" | "n/a" | "na" | "[n/a]" | "[not supported]" | "not supported"
-    )
-}
-
-fn parse_optional_f64(value: &str) -> Option<f64> {
-    if is_missing(value) {
-        return None;
-    }
-    numeric_prefix(value).and_then(|value| value.parse().ok())
-}
-
-fn parse_u64(value: &str) -> Option<u64> {
-    if is_missing(value) {
-        return None;
-    }
-    numeric_prefix(value).and_then(|value| value.parse::<f64>().ok().map(|number| number as u64))
-}
-
-fn parse_u32(value: &str) -> Option<u32> {
-    parse_u64(value).and_then(|value| u32::try_from(value).ok())
-}
-
-fn numeric_prefix(value: &str) -> Option<&str> {
-    let trimmed = value.trim().trim_end_matches('%').trim();
-    let end = trimmed
-        .char_indices()
-        .take_while(|(_, ch)| ch.is_ascii_digit() || *ch == '.' || *ch == '-')
-        .map(|(index, ch)| index + ch.len_utf8())
-        .last()
-        .unwrap_or(0);
-
-    (end > 0).then_some(&trimmed[..end])
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn parses_unavailable_gpu_output() {
-        let overview = parse_gpu_overview_output("GPU_AVAILABLE\t0\n");
-
-        assert!(!overview.available);
-        assert!(overview.gpus.is_empty());
-        assert!(overview.processes.is_empty());
-    }
-
-    #[test]
-    fn parses_single_gpu_with_metrics() {
-        let raw = concat!(
-            "GPU_AVAILABLE\t1\n",
-            "GPU_CUDA_VERSION\t12.4\n",
-            "GPU_CSV_BEGIN\n",
-            "0, GPU-abc, NVIDIA RTX 4090, 550.54.14, 43, 77, 31, 24564, 12345, 12219, 180.5, 450.0, 46, P2\n",
-            "GPU_CSV_END\n",
-            "GPU_PROCESS_CSV_BEGIN\n",
-            "GPU_PROCESS_CSV_END\n",
-        );
-
-        let overview = parse_gpu_overview_output(raw);
-
-        assert!(overview.available);
-        assert_eq!(overview.driver_version, "550.54.14");
-        assert_eq!(overview.cuda_version, "12.4");
-        assert_eq!(overview.gpus[0].name, "NVIDIA RTX 4090");
-        assert_eq!(overview.gpus[0].utilization_gpu_percent, Some(77.0));
-        assert_eq!(overview.gpus[0].memory_used_mb, 12345);
-        assert_eq!(overview.gpus[0].power_draw_w, Some(180.5));
-    }
-
-    #[test]
-    fn parses_multiple_gpus_and_missing_optional_metrics() {
-        let raw = concat!(
-            "GPU_AVAILABLE\t1\n",
-            "GPU_CUDA_VERSION\t12.2\n",
-            "GPU_CSV_BEGIN\n",
-            "0, GPU-a, NVIDIA A100, 535.1, N/A, 0, 0, 40536, 0, 40536, [Not Supported], 400, N/A, P0\n",
-            "1, GPU-b, NVIDIA T4, 535.1, 55, 99, 80, 15360, 15000, 360, 67.2, 70.0, [Not Supported], P0\n",
-            "GPU_CSV_END\n",
-            "GPU_PROCESS_CSV_BEGIN\n",
-            "GPU_PROCESS_CSV_END\n",
-        );
-
-        let overview = parse_gpu_overview_output(raw);
-
-        assert_eq!(overview.gpus.len(), 2);
-        assert_eq!(overview.gpus[0].temperature_c, None);
-        assert_eq!(overview.gpus[0].power_draw_w, None);
-        assert_eq!(overview.gpus[0].fan_speed_percent, None);
-        assert_eq!(overview.gpus[1].fan_speed_percent, None);
-    }
-
-    #[test]
-    fn maps_gpu_processes_to_indexes() {
-        let raw = concat!(
-            "GPU_AVAILABLE\t1\n",
-            "GPU_CUDA_VERSION\t12.2\n",
-            "GPU_CSV_BEGIN\n",
-            "0, GPU-a, NVIDIA A100, 535.1, 40, 30, 20, 40536, 1000, 39536, 80, 400, 20, P0\n",
-            "1, GPU-b, NVIDIA T4, 535.1, 44, 10, 9, 15360, 512, 14848, 35, 70, 30, P8\n",
-            "GPU_CSV_END\n",
-            "GPU_PROCESS_CSV_BEGIN\n",
-            "GPU-b, 4242, 2048, python\n",
-            "GPU_PROCESS_CSV_END\n",
-        );
-
-        let overview = parse_gpu_overview_output(raw);
-
-        assert_eq!(overview.processes.len(), 1);
-        assert_eq!(overview.processes[0].gpu_index, Some(1));
-        assert_eq!(overview.processes[0].pid, 4242);
-        assert_eq!(overview.processes[0].used_memory_mb, 2048);
-    }
-
-    #[test]
-    fn keeps_commas_inside_quoted_process_names() {
-        let raw = concat!(
-            "GPU_AVAILABLE\t1\n",
-            "GPU_CUDA_VERSION\t12.2\n",
-            "GPU_CSV_BEGIN\n",
-            "0, GPU-a, NVIDIA A100, 535.1, 40, 30, 20, 40536, 1000, 39536, 80, 400, 20, P0\n",
-            "GPU_CSV_END\n",
-            "GPU_PROCESS_CSV_BEGIN\n",
-            "GPU-a, 100, 512, \"python, worker\"\n",
-            "GPU_PROCESS_CSV_END\n",
-        );
-
-        let overview = parse_gpu_overview_output(raw);
-
-        assert_eq!(overview.processes[0].process_name, "python, worker");
-    }
-}

src-tauri/src/core/plugins/gateway.rs

@@ -67,6 +67,7 @@ impl PluginManager {
             "host/filesystem/read" => "filesystem.read",
             "host/storage/get" | "host/storage/set" => "storage",
             "host/network/request" => "",
+            "host/remote/probe" => "remote.probe",
             _ => return Err(AppError::Config("Unknown plugin host capability".into())),
         };
         if !permission.is_empty() {
@@ -75,6 +76,33 @@ impl PluginManager {
         let operation = async {
             let manager = app.state::<Arc<SessionManager>>().inner().clone();
             match method {
+                "host/remote/probe" => {
+                    let request: nyaterm_plugin_runtime::probe::ProbeRequest =
+                        serde_json::from_value(params.clone())?;
+                    let root = self
+                        .registry
+                        .read()
+                        .await
+                        .as_ref()
+                        .ok_or_else(|| self.registry_error())?
+                        .active_path(&scope.plugin_id)
+                        .map_err(plugin_error)?;
+                    self.scope(app, token, caller_window).await?;
+                    let executor = super::probe::SshProbeExecutor {
+                        sessions: manager,
+                        session_id: session_id(&scope)?.into(),
+                    };
+                    let result = nyaterm_plugin_runtime::probe::execute_declared(
+                        &executor,
+                        &root,
+                        &plugin.active().map_err(plugin_error)?.manifest,
+                        request,
+                        scope.cancellation.child_token(),
+                    )
+                    .await
+                    .map_err(plugin_error)?;
+                    serde_json::to_value(result).map_err(Into::into)
+                }
                 "host/session" => match &scope.session_id {
                     Some(id) => {
                         let info = require_session(app, &scope.window_label, id).await?;

src-tauri/src/core/plugins/mod.rs

@@ -1,5 +1,8 @@
 mod gateway;
 mod lifecycle;
+mod monitoring;
+pub use monitoring::MonitorSubscription;
+mod probe;
 mod protocol;
 
 use crate::cmd::app::AppLockState;
@@ -97,6 +100,7 @@ pub struct PluginManager {
     backends: Mutex<HashMap<String, BackendSlot>>,
     storage_lock: Mutex<()>,
     diagnostics: Diagnostics,
+    monitors: std::sync::Mutex<monitoring::Monitors>,
 }
 
 impl PluginManager {
@@ -116,6 +120,7 @@ impl PluginManager {
             backends: Mutex::new(HashMap::new()),
             storage_lock: Mutex::new(()),
             diagnostics: Diagnostics::default(),
+            monitors: std::sync::Mutex::new(monitoring::Monitors::default()),
         })
     }
 
@@ -192,7 +197,6 @@ impl PluginManager {
                 "Plugin version changed; review its permissions again".into(),
             ));
         }
-        self.revoke_plugin(id).await;
         self.registry
             .write()
             .await
@@ -200,6 +204,7 @@ impl PluginManager {
             .ok_or_else(|| self.registry_error())?
             .configure(id, enabled, permissions)
             .map_err(plugin_error)?;
+        self.revoke_plugin(id).await;
         let _ = app.emit("plugins-changed", ());
         Ok(())
     }
@@ -310,10 +315,12 @@ impl PluginManager {
         if let Some(scope) = scopes.remove(token) {
             scope.cancellation.cancel();
         }
+        self.monitors.lock().unwrap().remove_scope(token);
         Ok(())
     }
 
     pub async fn revoke_window(&self, label: &str) {
+        self.monitors.lock().unwrap().revoke(|k| k.window == label);
         self.scopes.lock().await.retain(|_, scope| {
             if scope.window_label == label {
                 scope.cancellation.cancel();
@@ -329,6 +336,10 @@ impl PluginManager {
     }
 
     pub async fn revoke_session(&self, session_id: &str) {
+        self.monitors
+            .lock()
+            .unwrap()
+            .revoke(|k| k.session == session_id);
         self.scopes.lock().await.retain(|_, scope| {
             if scope.session_id.as_deref() == Some(session_id) {
                 scope.cancellation.cancel();
@@ -340,6 +351,7 @@ impl PluginManager {
     }
 
     pub async fn revoke_all(&self) {
+        self.monitors.lock().unwrap().revoke(|_| true);
         for (_, scope) in self.scopes.lock().await.drain() {
             scope.cancellation.cancel();
         }
@@ -353,6 +365,7 @@ impl PluginManager {
     }
 
     async fn revoke_plugin(&self, id: &str) {
+        self.monitors.lock().unwrap().reset_plugin(id);
         self.diagnostics.stop(id);
         self.scopes.lock().await.retain(|_, scope| {
             if scope.plugin_id == id {
@@ -444,12 +457,34 @@ impl PluginManager {
         token: &str,
         method: &str,
         params: Value,
+    ) -> AppResult<Value> {
+        self.backend_request(app, window_label, token, method, params, false)
+            .await
+    }
+
+    async fn backend_request(
+        self: &Arc<Self>,
+        app: &tauri::AppHandle,
+        window_label: &str,
+        token: &str,
+        method: &str,
+        params: Value,
+        monitor_request: bool,
     ) -> AppResult<Value> {
         let scope = self.scope(app, token, Some(window_label)).await?;
         let _lease = self.lifecycle.acquire(&scope.plugin_id, false)?;
         let plugin = self.plugin(&scope.plugin_id).await?;
         plugin.require("native").map_err(plugin_error)?;
-        if !(method.starts_with("ui/")
+        if !((monitor_request
+            && plugin
+                .active()
+                .map_err(plugin_error)?
+                .manifest
+                .contributions
+                .monitors
+                .iter()
+                .any(|m| m.method == method))
+            || method.starts_with("ui/")
             || plugin
                 .active()
                 .map_err(plugin_error)?
@@ -622,6 +657,16 @@ impl PluginManager {
         ensure_unlocked(app)?;
         let _lease = self.lifecycle.acquire(id, true)?;
         self.plugin(id).await?;
+        self.monitors.lock().unwrap().pause_plugin(id);
+        if let Some(ticket) = self.diagnostics.ticket(id) {
+            self.diagnostics.update(
+                &ticket,
+                None,
+                "info",
+                "monitor",
+                "Monitor collection paused",
+            );
+        }
         self.diagnostics.stop(id);
         if let Some(backend) = self.backends.lock().await.remove(id)
             && let Some(sidecar) = backend.lock().await.take()

src-tauri/src/core/plugins/monitoring.rs

@@ -0,0 +1,320 @@
+use super::{PluginManager, plugin_error, require_session};
+use crate::error::{AppError, AppResult};
+use nyaterm_plugin_runtime::monitoring::{Collector, MonitorKey, MonitorSnapshot, MonitorTask};
+use serde::Serialize;
+use serde_json::Value;
+use std::collections::{HashMap, HashSet};
+use std::sync::{Arc, Weak};
+use std::time::Instant;
+use tauri::{Emitter, Manager};
+use tokio_util::sync::CancellationToken;
+
+#[derive(Default)]
+pub(super) struct Monitors {
+    tasks: HashMap<MonitorKey, Arc<MonitorTask>>,
+    subscriptions: HashMap<String, Subscription>,
+    paused_plugins: HashSet<String>,
+}
+
+struct Subscription {
+    token: String,
+    key: MonitorKey,
+    task: Arc<MonitorTask>,
+    cancellation: CancellationToken,
+}
+
+impl Monitors {
+    pub fn remove_scope(&mut self, token: &str) {
+        let ids = self
+            .subscriptions
+            .iter()
+            .filter(|(_, s)| s.token == token)
+            .map(|(id, _)| id.clone())
+            .collect::<Vec<_>>();
+        for id in ids {
+            self.remove(&id);
+        }
+    }
+
+    pub fn remove(&mut self, id: &str) {
+        if let Some(s) = self.subscriptions.remove(id) {
+            s.cancellation.cancel();
+            s.task.remove(id);
+        }
+        self.tasks.retain(|_, t| !t.is_stopped());
+    }
+
+    pub fn revoke(&mut self, matches: impl Fn(&MonitorKey) -> bool) {
+        let ids = self
+            .subscriptions
+            .iter()
+            .filter(|(_, s)| matches(&s.key))
+            .map(|(id, _)| id.clone())
+            .collect::<Vec<_>>();
+        for id in ids {
+            self.remove(&id);
+        }
+    }
+
+    pub fn reset_plugin(&mut self, id: &str) {
+        self.revoke(|k| k.plugin_id == id);
+        self.paused_plugins.remove(id);
+    }
+
+    pub fn pause_plugin(&mut self, id: &str) {
+        self.paused_plugins.insert(id.into());
+        for (key, task) in &self.tasks {
+            if key.plugin_id == id {
+                task.pause();
+            }
+        }
+    }
+}
+
+#[derive(Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MonitorSubscription {
+    pub subscription_id: String,
+    pub snapshot: MonitorSnapshot,
+}
+
+struct PluginCollector {
+    manager: Weak<PluginManager>,
+    app: tauri::AppHandle,
+    method: String,
+    plugin_id: String,
+}
+
+#[async_trait::async_trait]
+impl Collector for PluginCollector {
+    fn status(&self, message: &str) {
+        if let Some(host) = self.manager.upgrade()
+            && let Some(ticket) = host.diagnostics.ticket(&self.plugin_id)
+        {
+            host.diagnostics
+                .update(&ticket, None, "info", "monitor", message);
+        }
+    }
+    async fn collect(&self, scope: &str) -> nyaterm_plugin_runtime::Result<Value> {
+        let manager = self
+            .manager
+            .upgrade()
+            .ok_or_else(|| nyaterm_plugin_runtime::Error::Runtime("Plugin host stopped".into()))?;
+        let owned = manager
+            .scope(&self.app, scope, None)
+            .await
+            .map_err(|e| nyaterm_plugin_runtime::Error::Runtime(e.to_string()))?;
+        manager
+            .backend_request(
+                &self.app,
+                &owned.window_label,
+                scope,
+                &self.method,
+                Value::Null,
+                true,
+            )
+            .await
+            .map_err(|e| nyaterm_plugin_runtime::Error::Runtime(e.to_string()))
+    }
+}
+
+impl PluginManager {
+    pub async fn subscribe_monitor(
+        self: &Arc<Self>,
+        app: &tauri::AppHandle,
+        window: &str,
+        token: &str,
+        monitor_id: &str,
+        interval_seconds: u64,
+    ) -> AppResult<MonitorSubscription> {
+        let scope = self.scope(app, token, Some(window)).await?;
+        let _lease = self.lifecycle.acquire(&scope.plugin_id, false)?;
+        let plugin = self.plugin(&scope.plugin_id).await?;
+        plugin.require("native").map_err(plugin_error)?;
+        plugin.require("remote.probe").map_err(plugin_error)?;
+        let session = scope.session_id.as_ref().ok_or_else(|| {
+            AppError::Config("Select an active SSH terminal for monitoring".into())
+        })?;
+        let info = require_session(app, window, session).await?;
+        if info.session_type != crate::core::SessionType::SSH
+            || !info.remote_stats_enabled
+            || info.ssh_profile == Some(crate::config::SshProfile::NetworkDevice)
+        {
+            return Err(AppError::Config(
+                "Remote probes are unavailable for this session".into(),
+            ));
+        }
+        let monitor = plugin
+            .active()
+            .map_err(plugin_error)?
+            .manifest
+            .contributions
+            .monitors
+            .iter()
+            .find(|m| m.id == monitor_id)
+            .ok_or_else(|| AppError::Config("Unknown monitor".into()))?;
+        let key = MonitorKey {
+            plugin_id: scope.plugin_id.clone(),
+            version: scope.version.clone(),
+            window: window.into(),
+            session: session.clone(),
+            monitor_id: monitor_id.into(),
+        };
+        let id = uuid::Uuid::new_v4().to_string();
+        let cancellation = scope.cancellation.child_token();
+        let task = {
+            let mut monitors = self.monitors.lock().unwrap();
+            monitors.tasks.retain(|_, t| !t.is_stopped());
+            if monitors.subscriptions.len() >= 256 || monitors.tasks.len() >= 128 {
+                return Err(AppError::Config("Too many monitor subscriptions".into()));
+            }
+            let paused = monitors.paused_plugins.contains(&scope.plugin_id);
+            let task = monitors
+                .tasks
+                .entry(key.clone())
+                .or_insert_with(|| {
+                    let task = MonitorTask::new(
+                        session.clone(),
+                        Arc::new(PluginCollector {
+                            manager: Arc::downgrade(self),
+                            app: app.clone(),
+                            method: monitor.method.clone(),
+                            plugin_id: scope.plugin_id.clone(),
+                        }),
+                    );
+                    if paused {
+                        task.pause();
+                    }
+                    task
+                })
+                .clone();
+            monitors.subscriptions.insert(
+                id.clone(),
+                Subscription {
+                    token: token.into(),
+                    key,
+                    task: task.clone(),
+                    cancellation: cancellation.clone(),
+                },
+            );
+            task.add(
+                id.clone(),
+                token.into(),
+                cancellation.clone(),
+                interval_seconds,
+            );
+            task
+        };
+        if let Err(error) = self.scope(app, token, Some(window)).await {
+            self.monitors.lock().unwrap().remove(&id);
+            return Err(error);
+        }
+        let mut updates = task.watch();
+        let manager = Arc::downgrade(self);
+        let app = app.clone();
+        let label = window.to_owned();
+        let sub_id = id.clone();
+        let scope_token = token.to_owned();
+        let expiry = scope.expires.saturating_duration_since(Instant::now());
+        tokio::spawn(async move {
+            let deadline = tokio::time::sleep(expiry);
+            tokio::pin!(deadline);
+            loop {
+                tokio::select! {
+                    () = cancellation.cancelled() => break,
+                    () = &mut deadline => break,
+                    changed = updates.changed() => {
+                        if changed.is_err() { break; }
+                        let Some(host) = manager.upgrade() else { break; };
+                        if host.scope(&app, &scope_token, Some(&label)).await.is_err() { break; }
+                        let snapshot = updates.borrow_and_update().clone();
+                        if let Some(window) = app.get_webview_window(&label) {
+                            let _ = window.emit("plugin-monitor-updated", MonitorSubscription {
+                                subscription_id: sub_id.clone(), snapshot,
+                            });
+                        }
+                    }
+                }
+            }
+            if let Some(host) = manager.upgrade() {
+                host.monitors.lock().unwrap().remove(&sub_id);
+            }
+        });
+        Ok(MonitorSubscription {
+            subscription_id: id,
+            snapshot: task.snapshot(),
+        })
+    }
+
+    pub fn unsubscribe_monitor(&self, window: &str, token: &str, id: &str) -> AppResult<()> {
+        let mut monitors = self.monitors.lock().unwrap();
+        if monitors
+            .subscriptions
+            .get(id)
+            .is_some_and(|s| s.token != token || s.key.window != window)
+        {
+            return Err(AppError::Config(
+                "Monitor subscription belongs to another scope".into(),
+            ));
+        }
+        monitors.remove(id);
+        Ok(())
+    }
+
+    pub async fn refresh_monitor(
+        &self,
+        app: &tauri::AppHandle,
+        window: &str,
+        token: &str,
+        id: &str,
+    ) -> AppResult<()> {
+        let scope = self.scope(app, token, Some(window)).await?;
+        let mut monitors = self.monitors.lock().unwrap();
+        let subscription = monitors
+            .subscriptions
+            .get(id)
+            .filter(|s| s.token == token && s.key.window == window)
+            .ok_or_else(|| AppError::Config("Monitor subscription is unavailable".into()))?;
+        let task = subscription.task.clone();
+        monitors.paused_plugins.remove(&scope.plugin_id);
+        task.refresh();
+        Ok(())
+    }
+
+    pub async fn probe_scripts(
+        &self,
+        app: &tauri::AppHandle,
+        id: &str,
+        version: &str,
+    ) -> AppResult<HashMap<String, String>> {
+        super::ensure_unlocked(app)?;
+        let _lease = self.lifecycle.acquire(id, false)?;
+        let plugin = self.plugin(id).await?;
+        if plugin.active_version != version {
+            return Err(AppError::Config(
+                "Plugin version changed; review again".into(),
+            ));
+        }
+        let root = self
+            .registry
+            .read()
+            .await
+            .as_ref()
+            .ok_or_else(|| self.registry_error())?
+            .active_path(id)
+            .map_err(plugin_error)?;
+        plugin
+            .active()
+            .map_err(plugin_error)?
+            .manifest
+            .contributions
+            .probes
+            .iter()
+            .map(|p| {
+                nyaterm_plugin_runtime::probe::read_script(&root, p)
+                    .map(|s| (p.id.clone(), s))
+                    .map_err(plugin_error)
+            })
+            .collect()
+    }
+}

src-tauri/src/core/plugins/probe.rs

@@ -0,0 +1,33 @@
+use crate::core::SessionManager;
+use nyaterm_plugin_runtime::{
+    Result,
+    probe::{ProbeExecutor, ProbeOutput},
+};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio_util::sync::CancellationToken;
+
+pub(super) struct SshProbeExecutor {
+    pub sessions: Arc<SessionManager>,
+    pub session_id: String,
+}
+
+#[async_trait::async_trait]
+impl ProbeExecutor for SshProbeExecutor {
+    async fn execute(
+        &self,
+        script: &str,
+        timeout: Duration,
+        cancellation: CancellationToken,
+    ) -> Result<ProbeOutput> {
+        crate::core::remote_exec::exec_ssh_session_probe(
+            &self.sessions,
+            &self.session_id,
+            script.as_bytes(),
+            timeout,
+            cancellation,
+        )
+        .await
+        .map_err(|e| nyaterm_plugin_runtime::Error::Runtime(e.to_string()))
+    }
+}

src-tauri/src/core/remote_exec.rs

@@ -5,6 +5,100 @@ use russh::ChannelMsg;
 use serde::Serialize;
 use std::sync::Arc;
 use std::time::Duration;
+use tokio_util::sync::CancellationToken;
+
+/// A private exec channel, never `SessionCommand::Write` or the user's PTY.
+/// The guard closes the channel even when its enclosing RPC future is dropped.
+struct ProbeChannel(Option<russh::Channel<russh::client::Msg>>);
+
+impl Drop for ProbeChannel {
+    fn drop(&mut self) {
+        if let Some(channel) = self.0.take() {
+            tokio::spawn(async move {
+                let _ = tokio::time::timeout(Duration::from_secs(2), channel.close()).await;
+            });
+        }
+    }
+}
+
+pub async fn exec_ssh_session_probe(
+    manager: &Arc<SessionManager>,
+    session_id: &str,
+    script: &[u8],
+    timeout: Duration,
+    cancellation: CancellationToken,
+) -> AppResult<nyaterm_plugin_runtime::probe::ProbeOutput> {
+    ensure_remote_exec_enabled(manager, session_id).await?;
+    let info = manager.session_info(session_id).await?;
+    if info.ssh_profile == Some(crate::config::SshProfile::NetworkDevice) {
+        return Err(AppError::Config(
+            "Remote probes are disabled for network devices".into(),
+        ));
+    }
+    let ssh_handle = get_ssh_handle(manager, session_id).await?;
+    let operation = async {
+        let handle_mtx = ssh_handle.target_handle();
+        let channel = handle_mtx
+            .lock()
+            .await
+            .channel_open_session()
+            .await
+            .map_err(|e| AppError::Channel(e.to_string()))?;
+        let mut guard = ProbeChannel(Some(channel));
+        let channel = guard.0.as_mut().unwrap();
+        channel
+            .exec(true, b"sh -s".as_slice())
+            .await
+            .map_err(|e| AppError::Channel(e.to_string()))?;
+        channel
+            .data(script)
+            .await
+            .map_err(|e| AppError::Channel(e.to_string()))?;
+        channel
+            .eof()
+            .await
+            .map_err(|e| AppError::Channel(e.to_string()))?;
+        let mut stdout = Vec::new();
+        let mut stderr = Vec::new();
+        let mut exit_status = None;
+        while let Some(message) = channel.wait().await {
+            match message {
+                ChannelMsg::Data { data } => {
+                    if stdout.len() + stderr.len() + data.len() > 1024 * 1024 {
+                        return Err(AppError::Config("Probe output exceeds 1 MiB".into()));
+                    }
+                    stdout.extend_from_slice(&data);
+                }
+                ChannelMsg::ExtendedData { data, .. } => {
+                    if stdout.len() + stderr.len() + data.len() > 1024 * 1024 {
+                        return Err(AppError::Config("Probe output exceeds 1 MiB".into()));
+                    }
+                    stderr.extend_from_slice(&data);
+                }
+                ChannelMsg::ExitStatus {
+                    exit_status: status,
+                } => exit_status = Some(status),
+                ChannelMsg::Close => break,
+                _ => {}
+            }
+            if stdout.len() + stderr.len() > 1024 * 1024 {
+                return Err(AppError::Config("Probe output exceeds 1 MiB".into()));
+            }
+        }
+        Ok(nyaterm_plugin_runtime::probe::ProbeOutput {
+            stdout: String::from_utf8(stdout)
+                .map_err(|_| AppError::Config("Probe stdout must be UTF-8".into()))?,
+            stderr: String::from_utf8(stderr)
+                .map_err(|_| AppError::Config("Probe stderr must be UTF-8".into()))?,
+            exit_status,
+        })
+    };
+    tokio::select! {
+        () = cancellation.cancelled() => Err(AppError::Cancelled("Probe cancelled".into())),
+        result = tokio::time::timeout(timeout, operation) =>
+            result.map_err(|_| AppError::Cancelled("Probe timed out".into()))?,
+    }
+}
 
 #[derive(Debug, Clone, Serialize)]
 pub struct RemoteCommandOutput {

src-tauri/src/lib.rs

@@ -156,6 +156,10 @@ pub fn run() {
             },
         )
         .invoke_handler(tauri::generate_handler![
+            cmd::plugins::subscribe_plugin_monitor,
+            cmd::plugins::unsubscribe_plugin_monitor,
+            cmd::plugins::refresh_plugin_monitor,
+            cmd::plugins::get_plugin_probe_scripts,
             cmd::plugins::get_plugin_diagnostics,
             cmd::plugins::clear_plugin_logs,
             cmd::plugins::stop_plugin_backend,