//! monitor: real-time TUI of the managed processes (issue #50). Lists every //! running agent with CPU %, RSS, uptime; alerts on configurable thresholds //! (settings.monitor_thresholds) with a desktop notification + journal event. //! `--json` prints one JSON document per tick for scripts. use super::*; use crate::config::MonitorThresholds; use crate::events::{Event, EventKind}; use anyhow::{Context, Result}; use serde::Serialize; use std::process::Command; use std::time::Duration; /// One measured process row. #[derive(Debug, Clone, Default, Serialize)] pub struct AgentMetric { pub agent: String, pub pid: u32, pub cpu_pct: f64, pub mem_mb: f64, pub uptime_s: u64, /// "running" when the process answers, "dead" otherwise. pub state: String, } /// Raw reading of a process (platform-dependent). #[derive(Debug, Clone, Copy, Default)] pub struct RawReading { /// Total CPU time consumed so far (seconds). pub cpu_total_s: f64, /// Resident set size in bytes. pub rss_bytes: u64, } /// Read the raw counters of a process. Windows: PowerShell Get-Process /// (CPU = total seconds, WorkingSet64 = RSS). Unix: ps. fn raw_reading(pid: u32) -> Option { #[cfg(windows)] { let out = Command::new("powershell") .args([ "-NoProfile", "-Command", &format!( "Get-Process -Id {pid} | Select-Object -ExpandProperty CPU; Get-Process -Id {pid} | Select-Object -ExpandProperty WorkingSet64" ), ]) .output() .ok()?; if !out.status.success() { return None; } let text = String::from_utf8_lossy(&out.stdout); let mut lines = text.lines().filter(|l| !l.trim().is_empty()); let cpu: f64 = lines.next()?.trim().parse().ok()?; let rss: u64 = lines.next()?.trim().parse().ok()?; Some(RawReading { cpu_total_s: cpu, rss_bytes: rss, }) } #[cfg(unix)] { let out = Command::new("ps") .args(["-o", "time=,rss=", "-p", &pid.to_string()]) .output() .ok()?; if !out.status.success() { return None; } let text = String::from_utf8_lossy(&out.stdout); let mut parts = text.split_whitespace(); let time = parts.next()?; let rss_kb: u64 = parts.next()?.parse().ok()?; Some(RawReading { cpu_total_s: parse_ps_time(time), rss_bytes: rss_kb * 1024, }) } } /// Parse `ps` cumulative time "MM:SS" or "HH:MM:SS" into seconds. fn parse_ps_time(t: &str) -> f64 { let parts: Vec = t .split(':') .filter_map(|p| p.parse().ok()) .collect(); match parts.as_slice() { [m, s] => m * 60.0 + s, [h, m, s] => h * 3600.0 + m * 60.0 + s, _ => 0.0, } } /// Turn raw readings (taken `interval_s` apart) into a metric row. /// Pure: unit tested. pub fn to_metric( agent: &str, pid: u32, before: Option, after: RawReading, started_at: Option<&str>, interval_s: f64, ) -> AgentMetric { let cpu_pct = match before { Some(b) if interval_s > 0.0 => { let delta = (after.cpu_total_s - b.cpu_total_s).max(0.0); (delta / interval_s * 100.0).min(100.0 * 64.0) // multi-core cap, generous } _ => 0.0, }; let uptime_s = started_at .and_then(|t| chrono::DateTime::parse_from_rfc3339(t).ok()) .map(|t| (chrono::Utc::now() - t.with_timezone(&chrono::Utc)).num_seconds().max(0) as u64) .unwrap_or(0); AgentMetric { agent: agent.to_string(), pid, cpu_pct, mem_mb: after.rss_bytes as f64 / (1024.0 * 1024.0), uptime_s, state: "running".to_string(), } } /// Decide which alerts fire for a metric against the thresholds. Pure. pub fn alerts_for(m: &AgentMetric, t: &MonitorThresholds) -> Vec { let mut out = Vec::new(); if let Some(cpu) = t.cpu_pct { if cpu > 0.0 && m.cpu_pct > cpu { out.push(format!( "{} cpu {:.1}% > {cpu}%", m.agent, m.cpu_pct )); } } if let Some(mem) = t.mem_mb { if mem > 0 && m.mem_mb > mem as f64 { out.push(format!( "{} rss {:.0} MB > {mem} MB", m.agent, m.mem_mb )); } } out } /// Collect the metrics of every managed agent with a live PID. pub fn collect(app: &App, interval_s: f64) -> Vec { let mut out = Vec::new(); let Ok(sf) = app.state.load() else { return out; }; for (name, entry) in &sf.installed { let Some(pid) = entry.pid else { continue }; let Some(after) = raw_reading(pid) else { out.push(AgentMetric { agent: name.clone(), pid, state: "dead".to_string(), ..Default::default() }); continue; }; let before = if interval_s > 0.0 { std::thread::sleep(Duration::from_secs_f64(interval_s)); raw_reading(pid) } else { None }; out.push(to_metric( name, pid, before, after, entry.started_at.as_deref(), interval_s, )); } out } /// Emit one JSON document (one row per process) — script-friendly. fn print_json_rows(rows: &[AgentMetric]) { crate::output::print_json(&rows); } /// Run the monitor: TUI by default, --json for scripts. pub fn run(app: &App, interval: Option, json: bool) -> Result { let secs = interval.unwrap_or(2).max(1) as f64; let thresholds = app .config .settings .monitor_thresholds .clone() .unwrap_or_default(); if json { // One JSON document per tick, then sleep; Ctrl-C stops. loop { let rows = collect(app, 0.0); print_json_rows(&rows); for m in &rows { fire_alerts(app, m, &thresholds); } std::thread::sleep(Duration::from_secs_f64(secs)); } } tui_loop(app, secs, &thresholds) } /// Fire a desktop notification + journal event when a threshold is crossed. /// Alerts are rate-limited by the caller (one per tick). fn fire_alerts(app: &App, m: &AgentMetric, t: &MonitorThresholds) { for alert in alerts_for(m, t) { let _ = crate::process::desktop_notify("agent-manager monitor", &alert); app.emit( &Event::now(EventKind::Alert) .with_agent(m.agent.clone()) .with_pid(m.pid) .with_reason(format!("monitor: {alert}")), ); } } /// ratatui loop: list of managed processes, refresh every `secs`, q/Esc quits. fn tui_loop(app: &App, secs: f64, thresholds: &MonitorThresholds) -> Result { use crossterm::event::{self, Event, KeyCode, KeyEventKind}; use ratatui::layout::{Constraint, Layout}; use ratatui::style::{Color, Modifier, Style}; use ratatui::text::{Line, Span}; use ratatui::widgets::{Block, Borders, List, ListItem}; use ratatui::DefaultTerminal; let mut terminal = ratatui::init(); let result = (|| -> Result { loop { let rows = collect(app, secs); for m in &rows { fire_alerts(app, m, thresholds); } terminal.draw(|f| { let chunks = Layout::vertical([ Constraint::Length(1), Constraint::Min(0), Constraint::Length(1), ]) .split(f.area()); let title = Line::from(vec![ Span::styled( " am monitor ", Style::default() .fg(Color::Cyan) .add_modifier(Modifier::BOLD), ), Span::styled( format!("{} managed process(es)", rows.len()), Style::default().fg(Color::Gray), ), ]); f.render_widget( Block::default().borders(Borders::TOP).title(title), chunks[0], ); let items: Vec = rows .iter() .map(|m| { let state = if m.state == "dead" { Style::default().fg(Color::Red) } else { Style::default().fg(Color::Green) }; ListItem::new(Line::from(vec![ Span::styled( format!( "{:<20} pid {:>7} cpu {:>6.1}% rss {:>8.0} MB up {}", m.agent, m.pid, m.cpu_pct, m.mem_mb, human_uptime(m.uptime_s), ), state, ), ])) }) .collect(); f.render_widget( List::new(items).block(Block::default().borders(Borders::ALL)), chunks[1], ); let hint = Line::from(Span::styled( " q / Esc: quit ", Style::default().fg(Color::DarkGray), )); f.render_widget(hint, chunks[2]); })?; if event::poll(Duration::from_millis(100))? { if let Event::Key(k) = event::read()? { if k.kind == KeyEventKind::Press && matches!(k.code, KeyCode::Char('q') | KeyCode::Esc) { return Ok(0); } } } } })(); ratatui::restore(); result } fn human_uptime(secs: u64) -> String { if secs < 60 { format!("{secs}s") } else if secs < 3600 { format!("{}m{:02}s", secs / 60, secs % 60) } else { format!("{}h{:02}m", secs / 3600, (secs % 3600) / 60) } } // --------------------------------------------------------------------------- // Tests // --------------------------------------------------------------------------- #[cfg(test)] mod tests { use super::*; #[test] fn to_metric_computes_cpu_and_mem() { let before = RawReading { cpu_total_s: 10.0, rss_bytes: 2 * 1024 * 1024, }; let after = RawReading { cpu_total_s: 12.0, rss_bytes: 4 * 1024 * 1024, }; let m = to_metric("a", 42, Some(before), after, None, 2.0); assert_eq!(m.pid, 42); assert!((m.cpu_pct - 100.0).abs() < 1e-6, "{}", m.cpu_pct); assert!((m.mem_mb - 4.0).abs() < 1e-6); assert_eq!(m.state, "running"); } #[test] fn to_metric_first_sample_zero_cpu() { let after = RawReading { cpu_total_s: 5.0, rss_bytes: 1024 * 1024, }; let m = to_metric("a", 1, None, after, None, 2.0); assert_eq!(m.cpu_pct, 0.0); } #[test] fn to_metric_parses_uptime() { let now = chrono::Utc::now(); let started = now - chrono::Duration::minutes(5); let after = RawReading { cpu_total_s: 1.0, rss_bytes: 1024 * 1024, }; let m = to_metric("a", 1, None, after, Some(&started.to_rfc3339()), 0.0); assert!((260..=320).contains(&m.uptime_s), "{}", m.uptime_s); } #[test] fn alerts_fire_on_thresholds() { let m = AgentMetric { agent: "a".into(), pid: 1, cpu_pct: 95.0, mem_mb: 2000.0, uptime_s: 0, state: "running".into(), }; let t = MonitorThresholds { cpu_pct: Some(80.0), mem_mb: Some(1024), }; let alerts = alerts_for(&m, &t); assert_eq!(alerts.len(), 2); // Below thresholds → nothing. let quiet = AgentMetric { cpu_pct: 10.0, mem_mb: 100.0, ..m.clone() }; assert!(alerts_for(&quiet, &t).is_empty()); // 0 disables the threshold. let off = MonitorThresholds { cpu_pct: Some(0.0), mem_mb: Some(0), }; assert!(alerts_for(&m, &off).is_empty()); } #[test] fn parse_ps_time_formats() { assert!((parse_ps_time("01:30") - 90.0).abs() < 1e-9); assert!((parse_ps_time("1:02:03") - 3723.0).abs() < 1e-9); } }