systemprompt_api/services/health/

monitor.rs

//! Background process monitor for managed services.
//!
//! [`ProcessMonitor`] spawns a polling loop that checks each running service's
//! PID against the live process table, marks vanished processes as crashed, and
//! reports aggregate state via [`HealthSummary`] and [`ModuleHealth`]. Each
//! completed cycle advances a counter that callers observe through
//! [`ProcessMonitor::completed_cycles`].
//!
//! Copyright (c) systemprompt.io — Business Source License 1.1.
//! See <https://systemprompt.io> for licensing details.

use anyhow::Result;
use std::sync::Arc;
use std::time::Duration;
use systemprompt_database::ServiceRepository;
use systemprompt_loader::subprocess;
use systemprompt_traits::OwnedTask;
use tokio::sync::watch;
use tracing::{info, warn};

#[derive(Debug)]
pub struct ProcessMonitor {
    repository: ServiceRepository,
    monitor_handle: Option<OwnedTask<()>>,
    check_interval: Duration,
    cycles: Arc<watch::Sender<u64>>,
}

impl ProcessMonitor {
    pub fn new(repository: ServiceRepository) -> Self {
        Self::with_interval(repository, Duration::from_secs(30))
    }

    pub fn with_interval(repository: ServiceRepository, interval: Duration) -> Self {
        Self {
            repository,
            monitor_handle: None,
            check_interval: interval,
            cycles: Arc::new(watch::Sender::new(0)),
        }
    }

    #[must_use]
    pub fn completed_cycles(&self) -> watch::Receiver<u64> {
        self.cycles.subscribe()
    }

    pub fn start(&mut self) {
        if self.monitor_handle.is_some() {
            warn!("Process monitor already started");
            return;
        }

        info!("Starting centralized process monitoring");

        let repository = self.repository.clone();
        let interval = self.check_interval;
        let cycles = Arc::clone(&self.cycles);

        let handle = OwnedTask::spawn("process_monitor", async move {
            Self::monitor_loop(repository, interval, &cycles).await;
        });

        self.monitor_handle = Some(handle);
        info!("Centralized process monitoring started");
    }

    pub fn stop(&mut self) {
        if let Some(handle) = self.monitor_handle.take() {
            info!("Stopping process monitoring");
            handle.abort();
            info!("Process monitoring stopped");
        }
    }

    pub const fn is_running(&self) -> bool {
        self.monitor_handle.is_some()
    }

    async fn monitor_loop(
        repository: ServiceRepository,
        check_interval: Duration,
        cycles: &watch::Sender<u64>,
    ) {
        info!(
            interval_secs = check_interval.as_secs(),
            "Process monitor loop started"
        );

        let mut interval = tokio::time::interval(check_interval);

        loop {
            interval.tick().await;

            if let Err(e) = Self::perform_monitoring_cycle(&repository).await {
                warn!(error = %e, "Monitoring cycle failed");
            }
            cycles.send_modify(|completed| *completed = completed.saturating_add(1));
        }
    }

    async fn perform_monitoring_cycle(repository: &ServiceRepository) -> Result<()> {
        let services = repository.list_running_services_with_pid().await?;

        if services.is_empty() {
            return Ok(());
        }

        let mut healthy_count = 0;
        let mut crashed_count = 0;

        for service in services {
            if let Some(pid) = service.pid {
                if Self::pid_is_running(pid).await {
                    healthy_count += 1;
                } else {
                    repository.mark_service_crashed(&service.name).await?;

                    crashed_count += 1;
                    warn!(
                        module = %service.module_name,
                        service = %service.name,
                        pid = pid,
                        "Detected crashed service"
                    );
                }
            }
        }

        if crashed_count == 0 {
            info!(healthy = healthy_count, "All services healthy");
        } else {
            warn!(
                healthy = healthy_count,
                crashed = crashed_count,
                "Service health check completed with failures"
            );
        }

        Ok(())
    }

    async fn pid_is_running(pid: i32) -> bool {
        match u32::try_from(pid) {
            Ok(pid) => subprocess::is_running(pid).await,
            Err(_) => false,
        }
    }

    pub async fn health_check_all(&self) -> Result<HealthSummary> {
        info!("Running health check on all services");

        let services = self.repository.list_running_services_with_pid().await?;

        let mut summary = HealthSummary::default();

        for service in services {
            if let Some(pid) = service.pid {
                let healthy = Self::pid_is_running(pid).await;

                info!(
                    module = %service.module_name,
                    service = %service.name,
                    pid = pid,
                    healthy = healthy,
                    "Service health status"
                );

                *summary
                    .modules
                    .entry(service.module_name.as_str().to_owned())
                    .or_insert_with(ModuleHealth::default) += if healthy {
                    ModuleHealth {
                        healthy: 1,
                        crashed: 0,
                    }
                } else {
                    ModuleHealth {
                        healthy: 0,
                        crashed: 1,
                    }
                };
            }
        }

        let total_healthy = summary.modules.values().map(|m| m.healthy).sum::<u32>();
        let total_crashed = summary.modules.values().map(|m| m.crashed).sum::<u32>();

        if total_crashed == 0 {
            info!(healthy = total_healthy, "All services are healthy");
        } else {
            warn!(
                healthy = total_healthy,
                total = total_healthy + total_crashed,
                "Some services are unhealthy"
            );
        }

        Ok(summary)
    }
}

#[derive(Debug, Default)]
pub struct HealthSummary {
    pub modules: std::collections::HashMap<String, ModuleHealth>,
}

#[derive(Debug, Default, Copy, Clone)]
pub struct ModuleHealth {
    pub healthy: u32,
    pub crashed: u32,
}

impl std::ops::AddAssign for ModuleHealth {
    fn add_assign(&mut self, other: Self) {
        self.healthy += other.healthy;
        self.crashed += other.crashed;
    }
}

impl HealthSummary {
    pub fn total_healthy(&self) -> u32 {
        self.modules.values().map(|m| m.healthy).sum()
    }

    pub fn total_crashed(&self) -> u32 {
        self.modules.values().map(|m| m.crashed).sum()
    }

    pub fn is_all_healthy(&self) -> bool {
        self.total_crashed() == 0
    }
}