foxhunt/services/api/src/grpc/monitoring_handler.rs

//! Monitoring Service Handler - Direct Prometheus scraping for training metrics
//!
//! Serves `GetLiveTrainingMetrics`, `StreamTrainingMetrics`, and `GetEpochHistory`
//! by querying Prometheus directly, eliminating the monitoring-service backend.
//!
//! The merged monitoring.proto defines 16 RPCs (system health + training).
//! This handler implements the 3 training RPCs with real Prometheus scraping;
//! the 13 system health RPCs return UNIMPLEMENTED until real health checks are wired.

use std::collections::{HashMap, VecDeque};
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;

use anyhow::{Context, Result};
use futures::Stream;
use serde::Deserialize;
use tokio::sync::RwLock;
use tonic::{Request, Response, Status};
use tracing::{error, info, instrument};

use crate::monitoring::monitoring_service_server::{MonitoringService, MonitoringServiceServer};
use crate::monitoring::{
    AcknowledgeAlertRequest, AcknowledgeAlertResponse, AlertEvent,
    ClusterPodsResponse, EpochFinancialSnapshot, GetActiveAlertsRequest, GetActiveAlertsResponse,
    GetEpochHistoryRequest, GetEpochHistoryResponse, GetHealthCheckRequest,
    GetHealthCheckResponse, GetLatencyMetricsRequest, GetLatencyMetricsResponse,
    GetLiveTrainingMetricsRequest, GetLiveTrainingMetricsResponse, GetMetricsRequest,
    GetMetricsResponse, GetSystemStatusRequest, GetSystemStatusResponse,
    GetThroughputMetricsRequest, GetThroughputMetricsResponse, GpuSnapshot, HealthCheck,
    HealthStatus, Metric, MetricType, MetricsEvent, ServiceHealth, ServiceState, ServiceStatus,
    StreamTrainingMetricsRequest, SubscribeClusterPodsRequest, SystemHealth, SystemMetrics,
    SystemStatus, SystemStatusChangeType, SystemStatusEvent, TrainingSession,
};

// ============================================================================
// Prometheus client (ported from monitoring_service/src/prometheus_client.rs)
// ============================================================================

/// A single Prometheus instant-query result entry
#[derive(Debug, Deserialize)]
struct PromResult {
    metric: HashMap<String, String>,
    value: (f64, String), // (timestamp, value_string)
}

#[derive(Debug, Deserialize)]
struct PromData {
    result: Vec<PromResult>,
}

#[derive(Debug, Deserialize)]
struct PromResponse {
    status: String,
    data: PromData,
}

/// Parsed metric value with model/fold labels
#[derive(Debug, Clone)]
pub struct MetricSample {
    pub name: String,
    pub model: String,
    pub fold: String,
    pub value: f64,
}

struct PrometheusClient {
    http: reqwest::Client,
    base_url: String,
}

impl PrometheusClient {
    fn new(base_url: &str) -> Self {
        let http = reqwest::Client::builder()
            .timeout(Duration::from_secs(5))
            .build()
            .unwrap_or_else(|_| reqwest::Client::new());
        Self {
            http,
            base_url: base_url.trim_end_matches('/').to_owned(),
        }
    }

    /// Execute an instant query against Prometheus
    async fn query(&self, promql: &str) -> Result<Vec<PromResult>> {
        let url = format!("{}/api/v1/query", self.base_url);
        let resp = self
            .http
            .get(&url)
            .query(&[("query", promql)])
            .send()
            .await
            .context("Prometheus HTTP request failed")?;

        if !resp.status().is_success() {
            anyhow::bail!("Prometheus returned HTTP {}", resp.status());
        }

        let body: PromResponse = resp
            .json()
            .await
            .context("Failed to parse Prometheus response")?;

        if body.status != "success" {
            anyhow::bail!("Prometheus query status: {}", body.status);
        }

        Ok(body.data.result)
    }

    /// Fetch all training + hyperopt metrics
    async fn fetch_training_metrics(&self) -> Result<Vec<MetricSample>> {
        let results = self
            .query(r#"{__name__=~"foxhunt_training_.*|foxhunt_hyperopt_.*"}"#)
            .await?;
        Ok(parse_samples(results))
    }

    /// Fetch GPU metrics from DCGM exporter
    async fn fetch_gpu_metrics(&self) -> Result<Vec<MetricSample>> {
        let results = self
            .query(r#"{__name__=~"dcgm_gpu_utilization|dcgm_fb_used|dcgm_fb_free|dcgm_gpu_temp|dcgm_power_usage"}"#)
            .await?;
        Ok(parse_samples(results))
    }

    /// Fetch active training worker count (covers both K8s Jobs and CI runner pods)
    async fn fetch_active_jobs(&self) -> Result<u32> {
        let results = self.query("foxhunt_training_active_workers").await?;
        let count: f64 = results
            .iter()
            .filter_map(|r| r.value.1.parse::<f64>().ok())
            .sum();
        #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
        let count = count as u32;
        Ok(count)
    }

    /// Query node-level CPU and memory usage from Prometheus.
    ///
    /// Returns `(cpu_usage_pct, mem_usage_pct, disk_usage_pct)`.
    /// Falls back to zeros on any query failure (best-effort).
    async fn query_system_metrics(&self) -> (f64, f64, f64) {
        // CPU: 1 − idle fraction across all cores, averaged over 1m
        let cpu = self
            .query(
                r#"100 * (1 - avg(rate(node_cpu_seconds_total{mode="idle"}[1m])))"#,
            )
            .await
            .ok()
            .and_then(|r| r.first().and_then(|v| v.value.1.parse::<f64>().ok()))
            .unwrap_or(0.0);

        // Memory: (total − available) / total × 100
        let mem = self
            .query(
                r#"100 * (1 - node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes)"#,
            )
            .await
            .ok()
            .and_then(|r| r.first().and_then(|v| v.value.1.parse::<f64>().ok()))
            .unwrap_or(0.0);

        // Disk: (total − free) / total × 100 on root mount
        let disk = self
            .query(
                r#"100 * (1 - node_filesystem_avail_bytes{mountpoint="/"} / node_filesystem_size_bytes{mountpoint="/"})"#,
            )
            .await
            .ok()
            .and_then(|r| r.first().and_then(|v| v.value.1.parse::<f64>().ok()))
            .unwrap_or(0.0);

        (cpu, mem, disk)
    }

    /// Query cluster-level CPU and memory from node-exporter metrics.
    ///
    /// Returns `(cpu_usage_pct, mem_used_mb, mem_total_mb)`.
    /// Best-effort: returns `(0.0, 0.0, 0.0)` on any query failure.
    #[allow(clippy::cast_possible_truncation)]
    async fn query_cluster_resources(&self) -> (f32, f32, f32) {
        let cpu_pct = self
            .query(
                r#"100 * (1 - avg(rate(node_cpu_seconds_total{mode="idle"}[1m])))"#,
            )
            .await
            .ok()
            .and_then(|r| r.first().and_then(|v| v.value.1.parse::<f64>().ok()))
            .unwrap_or(0.0) as f32;

        let mem_used_mb = self
            .query(
                r#"sum(node_memory_MemTotal_bytes - node_memory_MemAvailable_bytes) / 1024 / 1024"#,
            )
            .await
            .ok()
            .and_then(|r| r.first().and_then(|v| v.value.1.parse::<f64>().ok()))
            .unwrap_or(0.0) as f32;

        let mem_total_mb = self
            .query(r#"sum(node_memory_MemTotal_bytes) / 1024 / 1024"#)
            .await
            .ok()
            .and_then(|r| r.first().and_then(|v| v.value.1.parse::<f64>().ok()))
            .unwrap_or(0.0) as f32;

        (cpu_pct, mem_used_mb, mem_total_mb)
    }
}

fn parse_samples(results: Vec<PromResult>) -> Vec<MetricSample> {
    results
        .into_iter()
        .filter_map(|r| {
            let value: f64 = r.value.1.parse().ok()?;
            Some(MetricSample {
                name: r.metric.get("__name__")?.clone(),
                model: r.metric.get("model").cloned().unwrap_or_default(),
                fold: r.metric.get("fold").cloned().unwrap_or_default(),
                value,
            })
        })
        .collect()
}

// ============================================================================
// Monitoring Service Handler
// ============================================================================

const MAX_EPOCH_HISTORY: usize = 50;

/// Monitoring Service Handler
///
/// Directly scrapes Prometheus for training metrics, GPU stats, and active jobs.
/// Replaces the former proxy pattern that forwarded to monitoring-service backend.
pub struct MonitoringServiceHandler {
    prom: Arc<PrometheusClient>,
    default_interval: u32,
    epoch_histories: Arc<RwLock<HashMap<String, VecDeque<EpochFinancialSnapshot>>>>,
    last_epochs: Arc<RwLock<HashMap<String, f32>>>,
    /// Backend service URLs for health checking: (name, url)
    backends: Arc<Vec<(String, String)>>,
    start_time: std::time::Instant,
}

impl MonitoringServiceHandler {
    pub fn new(prometheus_url: &str, default_interval_secs: u32) -> Self {
        let prom = PrometheusClient::new(prometheus_url);
        info!(
            "MonitoringServiceHandler: Prometheus={}, interval={}s",
            prometheus_url, default_interval_secs
        );
        Self {
            prom: Arc::new(prom),
            default_interval: default_interval_secs,
            epoch_histories: Arc::new(RwLock::new(HashMap::new())),
            last_epochs: Arc::new(RwLock::new(HashMap::new())),
            backends: Arc::new(Vec::new()),
            start_time: std::time::Instant::now(),
        }
    }

    /// Set the list of backend service URLs for health checking
    pub fn with_backends(mut self, backends: Vec<(String, String)>) -> Self {
        self.backends = Arc::new(backends);
        self
    }

    /// Check health of a single backend via gRPC health check protocol.
    ///
    /// `uptime_secs` is the gateway's own uptime — used as an approximation
    /// for the backend service uptime (individual services don't expose this).
    async fn check_backend_health(
        name: &str,
        url: &str,
        uptime_secs: i64,
    ) -> ServiceStatus {
        let channel = match tonic::transport::Channel::from_shared(url.to_string()) {
            Ok(c) => c.connect_lazy(),
            Err(_) => {
                return ServiceStatus {
                    service_name: name.to_string(),
                    health: ServiceHealth::Unhealthy.into(),
                    state: ServiceState::Error.into(),
                    error_message: Some(format!("Invalid URL: {url}")),
                    version: Some(common::build_info::version().to_owned()),
                    ..Default::default()
                };
            }
        };

        let mut hc = tonic_health::pb::health_client::HealthClient::new(channel);
        let req = tonic_health::pb::HealthCheckRequest {
            service: String::new(),
        };

        let start = std::time::Instant::now();
        let (health, state, msg) = match tokio::time::timeout(
            Duration::from_millis(2000),
            hc.check(req),
        )
        .await
        {
            Ok(Ok(_)) => (
                ServiceHealth::Healthy,
                ServiceState::Running,
                None,
            ),
            Ok(Err(s)) if s.code() == tonic::Code::Unimplemented => (
                ServiceHealth::Healthy,
                ServiceState::Running,
                Some("Serving (no health service)".to_string()),
            ),
            Ok(Err(e)) => (
                ServiceHealth::Unhealthy,
                ServiceState::Error,
                Some(format!("gRPC error: {e}")),
            ),
            Err(_) => (
                ServiceHealth::Unhealthy,
                ServiceState::Error,
                Some("Unreachable (2s timeout)".to_string()),
            ),
        };
        let latency_ms = start.elapsed().as_secs_f64() * 1000.0;

        let now = chrono::Utc::now().timestamp();
        let mut metadata = std::collections::HashMap::new();
        metadata.insert("latency_ms".to_string(), format!("{latency_ms:.1}"));

        ServiceStatus {
            service_name: name.to_string(),
            health: health.into(),
            state: state.into(),
            error_message: msg,
            version: Some(common::build_info::version().to_owned()),
            last_health_check: now,
            uptime_seconds: uptime_secs,
            metadata,
            ..Default::default()
        }
    }

    pub fn into_server(self) -> MonitoringServiceServer<MonitoringServiceHandler> {
        MonitoringServiceServer::new(self)
    }

    /// Build a full training metrics snapshot from Prometheus and update epoch history.
    async fn build_response(
        prom: &PrometheusClient,
        model_filter: &str,
        epoch_histories: &RwLock<HashMap<String, VecDeque<EpochFinancialSnapshot>>>,
        last_epochs: &RwLock<HashMap<String, f32>>,
    ) -> Result<GetLiveTrainingMetricsResponse, Status> {
        // Fetch training/GPU/jobs (fail-fast) and cluster resources (best-effort) concurrently.
        let (training_result, (cpu_pct, mem_used_mb, mem_total_mb)) = tokio::join!(
            async {
                tokio::try_join!(
                    prom.fetch_training_metrics(),
                    prom.fetch_gpu_metrics(),
                    prom.fetch_active_jobs(),
                )
            },
            prom.query_cluster_resources(),
        );

        let (training, gpu, jobs) = training_result
            .map_err(|e| Status::internal(format!("Prometheus query failed: {e}")))?;

        let sessions = group_into_sessions(&training, model_filter);
        let gpu_snapshot = build_gpu_snapshot(&gpu);

        // Record epoch history snapshots for sessions with new epoch data
        {
            let mut last = last_epochs.write().await;
            let mut histories = epoch_histories.write().await;
            for session in &sessions {
                let key = format!("{}/{}", session.model, session.fold);
                let prev_epoch = last.get(&key).copied().unwrap_or_default();
                if session.current_epoch > prev_epoch && session.epoch_sharpe != 0.0 {
                    last.insert(key.clone(), session.current_epoch);
                    let snapshot = EpochFinancialSnapshot {
                        epoch: session.current_epoch as u32,
                        sharpe: session.epoch_sharpe,
                        sortino: session.epoch_sortino,
                        win_rate: session.epoch_win_rate,
                        max_drawdown: session.epoch_max_drawdown,
                        profit_factor: session.epoch_profit_factor,
                        total_return: session.epoch_total_return,
                        avg_return: session.epoch_avg_return,
                        total_trades: session.epoch_total_trades,
                        loss: session.epoch_loss,
                        val_loss: session.validation_loss,
                        learning_rate: session.learning_rate,
                        action_buy_pct: session.action_buy_pct,
                        action_sell_pct: session.action_sell_pct,
                        action_hold_pct: session.action_hold_pct,
                    };
                    let history = histories
                        .entry(key)
                        .or_insert_with(|| VecDeque::with_capacity(MAX_EPOCH_HISTORY));
                    if history.len() >= MAX_EPOCH_HISTORY {
                        history.pop_front();
                    }
                    history.push_back(snapshot);
                }
            }
        }

        Ok(GetLiveTrainingMetricsResponse {
            sessions,
            gpu: Some(gpu_snapshot),
            active_k8s_jobs: jobs,
            timestamp: chrono::Utc::now().timestamp(),
            cpu_percent: cpu_pct,
            memory_used_mb: mem_used_mb,
            memory_total_mb: mem_total_mb,
        })
    }
}

#[tonic::async_trait]
impl MonitoringService for MonitoringServiceHandler {
    // ========================================================================
    // Training metrics RPCs (implemented -- direct Prometheus scraping)
    // ========================================================================

    type StreamTrainingMetricsStream =
        Pin<Box<dyn Stream<Item = Result<GetLiveTrainingMetricsResponse, Status>> + Send>>;

    #[instrument(skip(self, request), fields(request_id = %uuid::Uuid::new_v4()), err)]
    async fn get_live_training_metrics(
        &self,
        request: Request<GetLiveTrainingMetricsRequest>,
    ) -> Result<Response<GetLiveTrainingMetricsResponse>, Status> {
        info!("GetLiveTrainingMetrics: scraping Prometheus");
        let filter = &request.into_inner().model_filter;
        let resp =
            Self::build_response(&self.prom, filter, &self.epoch_histories, &self.last_epochs)
                .await?;
        Ok(Response::new(resp))
    }

    #[instrument(skip(self, request), fields(request_id = %uuid::Uuid::new_v4()), err)]
    async fn stream_training_metrics(
        &self,
        request: Request<StreamTrainingMetricsRequest>,
    ) -> Result<Response<Self::StreamTrainingMetricsStream>, Status> {
        let req = request.into_inner();
        let interval_secs = if req.interval_seconds == 0 {
            self.default_interval
        } else {
            req.interval_seconds.clamp(1, 60)
        };
        let filter = req.model_filter;
        let prom = self.prom.clone();
        let epoch_histories = self.epoch_histories.clone();
        let last_epochs = self.last_epochs.clone();

        info!(
            "StreamTrainingMetrics: interval={}s, filter={:?}",
            interval_secs,
            if filter.is_empty() {
                "<all>"
            } else {
                &filter
            }
        );

        let stream = async_stream::stream! {
            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
            loop {
                interval.tick().await;
                match Self::build_response(&prom, &filter, &epoch_histories, &last_epochs).await {
                    Ok(resp) => yield Ok(resp),
                    Err(e) => {
                        error!("Stream tick failed: {}", e);
                        yield Err(e);
                        break;
                    }
                }
            }
        };

        Ok(Response::new(Box::pin(stream)))
    }

    #[instrument(skip(self, request), fields(request_id = %uuid::Uuid::new_v4()), err)]
    async fn get_epoch_history(
        &self,
        request: Request<GetEpochHistoryRequest>,
    ) -> Result<Response<GetEpochHistoryResponse>, Status> {
        let req = request.into_inner();
        let key = format!("{}/{}", req.model, req.fold);
        let histories = self.epoch_histories.read().await;
        let epochs = match histories.get(&key) {
            Some(deque) => {
                let max = if req.max_epochs == 0 {
                    MAX_EPOCH_HISTORY
                } else {
                    req.max_epochs as usize
                };
                deque.iter().rev().take(max).rev().cloned().collect()
            }
            None => vec![],
        };
        Ok(Response::new(GetEpochHistoryResponse {
            model: req.model,
            fold: req.fold,
            epochs,
        }))
    }

    // ========================================================================
    // System health RPCs (stubs -- will be implemented when real health checks are wired)
    // ========================================================================

    type StreamSystemStatusStream =
        Pin<Box<dyn Stream<Item = Result<SystemStatusEvent, Status>> + Send>>;

    #[instrument(skip(self, request), err)]
    async fn get_system_status(
        &self,
        request: Request<GetSystemStatusRequest>,
    ) -> Result<Response<GetSystemStatusResponse>, Status> {
        let req = request.into_inner();
        let backends = self.backends.clone();
        #[allow(clippy::cast_possible_truncation)]
        let uptime_secs = self.start_time.elapsed().as_secs() as i64;

        // Fan-out health checks + system metrics concurrently
        let prom = self.prom.clone();
        let checks: Vec<_> = backends
            .iter()
            .filter(|(name, _)| {
                req.service_names.is_empty() || req.service_names.contains(name)
            })
            .map(|(name, url)| {
                let name = name.clone();
                let url = url.clone();
                async move { Self::check_backend_health(&name, &url, uptime_secs).await }
            })
            .collect();

        let (statuses, (cpu, mem, disk)) = tokio::join!(
            futures::future::join_all(checks),
            prom.query_system_metrics(),
        );

        let healthy_count = statuses
            .iter()
            .filter(|s| s.health == i32::from(ServiceHealth::Healthy))
            .count();
        let total = statuses.len();

        #[allow(clippy::cast_possible_truncation)]
        let overall_health = if healthy_count == total {
            SystemHealth::Healthy
        } else if healthy_count == 0 {
            SystemHealth::Critical
        } else if healthy_count * 2 >= total {
            SystemHealth::Degraded
        } else {
            SystemHealth::Unhealthy
        };

        let critical_issues: Vec<String> = statuses
            .iter()
            .filter(|s| s.health != i32::from(ServiceHealth::Healthy))
            .map(|s| {
                format!(
                    "{}: {}",
                    s.service_name,
                    s.error_message.as_deref().unwrap_or("unhealthy")
                )
            })
            .collect();

        #[allow(clippy::cast_possible_truncation)]
        let response = GetSystemStatusResponse {
            overall_status: Some(SystemStatus {
                overall_health: overall_health.into(),
                healthy_services: healthy_count as i32,
                total_services: total as i32,
                critical_issues,
                system_uptime_seconds: uptime_secs,
                system_metrics: Some(SystemMetrics {
                    cpu_usage_percent: cpu,
                    memory_usage_percent: mem,
                    disk_usage_percent: disk,
                    ..Default::default()
                }),
            }),
            service_statuses: statuses,
            timestamp: chrono::Utc::now().timestamp(),
        };

        Ok(Response::new(response))
    }

    #[instrument(skip(self, request), fields(request_id = %uuid::Uuid::new_v4()), err)]
    async fn stream_system_status(
        &self,
        request: Request<crate::monitoring::StreamSystemStatusRequest>,
    ) -> Result<Response<Self::StreamSystemStatusStream>, Status> {
        let req = request.into_inner();
        let interval_secs = req.update_frequency_seconds.unwrap_or(0);
        let interval_secs = if interval_secs == 0 {
            5u32
        } else {
            (interval_secs as u32).clamp(1, 60)
        };
        let backends = self.backends.clone();
        let start_time = self.start_time;
        let service_filter = req.service_names;

        info!(
            "StreamSystemStatus: interval={}s, filter={:?}",
            interval_secs,
            if service_filter.is_empty() {
                "<all>"
            } else {
                "filtered"
            }
        );

        let prom = self.prom.clone();

        let stream = async_stream::stream! {
            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
            loop {
                interval.tick().await;

                #[allow(clippy::cast_possible_truncation)]
                let uptime_secs = start_time.elapsed().as_secs() as i64;
                let checks: Vec<_> = backends
                    .iter()
                    .filter(|(name, _)| {
                        service_filter.is_empty() || service_filter.contains(name)
                    })
                    .map(|(name, url)| {
                        let name = name.clone();
                        let url = url.clone();
                        async move { Self::check_backend_health(&name, &url, uptime_secs).await }
                    })
                    .collect();

                let (statuses, (cpu, mem, disk)) = tokio::join!(
                    futures::future::join_all(checks),
                    prom.query_system_metrics(),
                );

                let healthy_count = statuses
                    .iter()
                    .filter(|s| s.health == i32::from(ServiceHealth::Healthy))
                    .count();
                let total = statuses.len();

                #[allow(clippy::cast_possible_truncation)]
                let overall_health = if healthy_count == total {
                    SystemHealth::Healthy
                } else if healthy_count == 0 {
                    SystemHealth::Critical
                } else if healthy_count * 2 >= total {
                    SystemHealth::Degraded
                } else {
                    SystemHealth::Unhealthy
                };

                let critical_issues: Vec<String> = statuses
                    .iter()
                    .filter(|s| s.health != i32::from(ServiceHealth::Healthy))
                    .map(|s| {
                        format!(
                            "{}: {}",
                            s.service_name,
                            s.error_message.as_deref().unwrap_or("unhealthy")
                        )
                    })
                    .collect();

                #[allow(clippy::cast_possible_truncation)]
                let system_status = SystemStatus {
                    overall_health: overall_health.into(),
                    healthy_services: healthy_count as i32,
                    total_services: total as i32,
                    critical_issues,
                    system_uptime_seconds: start_time.elapsed().as_secs() as i64,
                    system_metrics: Some(SystemMetrics {
                        cpu_usage_percent: cpu,
                        memory_usage_percent: mem,
                        disk_usage_percent: disk,
                        ..Default::default()
                    }),
                };

                let event = SystemStatusEvent {
                    system_status: Some(system_status),
                    change_type: SystemStatusChangeType::Unspecified.into(),
                    timestamp: chrono::Utc::now().timestamp(),
                };

                yield Ok(event);
            }
        };

        Ok(Response::new(Box::pin(stream)))
    }

    #[instrument(skip(self, request), err)]
    async fn get_health_check(
        &self,
        request: Request<GetHealthCheckRequest>,
    ) -> Result<Response<GetHealthCheckResponse>, Status> {
        let req = request.into_inner();
        let backends = self.backends.clone();
        #[allow(clippy::cast_possible_truncation)]
        let uptime_secs = self.start_time.elapsed().as_secs() as i64;

        // Fan-out health checks
        let checks: Vec<_> = backends
            .iter()
            .filter(|(name, _)| {
                req.service_name.is_none()
                    || req.service_name.as_deref() == Some(name.as_str())
            })
            .map(|(name, url)| {
                let name = name.clone();
                let url = url.clone();
                async move {
                    let start = std::time::Instant::now();
                    let status = Self::check_backend_health(&name, &url, uptime_secs).await;
                    let elapsed = start.elapsed();
                    let healthy = status.health == i32::from(ServiceHealth::Healthy);
                    HealthCheck {
                        check_name: name,
                        status: if healthy {
                            HealthStatus::Healthy.into()
                        } else {
                            HealthStatus::Unhealthy.into()
                        },
                        message: status.error_message,
                        response_time_ms: Some(elapsed.as_secs_f64() * 1000.0),
                        last_checked: chrono::Utc::now().timestamp(),
                        details: HashMap::new(),
                    }
                }
            })
            .collect();

        let health_checks = futures::future::join_all(checks).await;

        let all_healthy = health_checks
            .iter()
            .all(|c| c.status == i32::from(HealthStatus::Healthy));

        let response = GetHealthCheckResponse {
            health_status: if all_healthy {
                HealthStatus::Healthy.into()
            } else {
                HealthStatus::Degraded.into()
            },
            health_checks,
            timestamp: chrono::Utc::now().timestamp(),
        };

        Ok(Response::new(response))
    }

    type StreamMetricsStream =
        Pin<Box<dyn Stream<Item = Result<MetricsEvent, Status>> + Send>>;

    #[instrument(skip(self, request), err)]
    async fn get_metrics(
        &self,
        request: Request<GetMetricsRequest>,
    ) -> Result<Response<GetMetricsResponse>, Status> {
        let req = request.into_inner();

        // If specific metric names requested, query each; otherwise return empty
        let mut metrics = Vec::new();
        for name in &req.metric_names {
            match self.prom.query(name).await {
                Ok(results) => {
                    for r in results {
                        if let Ok(value) = r.value.1.parse::<f64>() {
                            metrics.push(Metric {
                                name: r
                                    .metric
                                    .get("__name__")
                                    .cloned()
                                    .unwrap_or_else(|| name.clone()),
                                metric_type: MetricType::Gauge.into(),
                                value,
                                unit: String::new(),
                                labels: r.metric,
                                timestamp: chrono::Utc::now().timestamp(),
                                statistics: None,
                            });
                        }
                    }
                }
                Err(e) => {
                    error!("Prometheus query for '{}' failed: {}", name, e);
                }
            }
        }

        Ok(Response::new(GetMetricsResponse {
            metrics,
            timestamp: chrono::Utc::now().timestamp(),
        }))
    }

    #[instrument(skip(self, request), fields(request_id = %uuid::Uuid::new_v4()), err)]
    async fn stream_metrics(
        &self,
        request: Request<crate::monitoring::StreamMetricsRequest>,
    ) -> Result<Response<Self::StreamMetricsStream>, Status> {
        let req = request.into_inner();
        let interval_secs = req.update_frequency_seconds.unwrap_or(0);
        let interval_secs = if interval_secs == 0 {
            5u32
        } else {
            (interval_secs as u32).clamp(1, 60)
        };
        let metric_names: Vec<String> = if req.metric_names.is_empty() {
            vec![
                "cpu_usage".to_owned(),
                "memory_usage".to_owned(),
                "gpu_utilization".to_owned(),
            ]
        } else {
            req.metric_names
        };
        let prom = self.prom.clone();

        info!(
            "StreamMetrics: interval={}s, metrics={:?}",
            interval_secs, metric_names
        );

        let stream = async_stream::stream! {
            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
            loop {
                interval.tick().await;

                let mut metrics = Vec::new();
                for name in &metric_names {
                    match prom.query(name).await {
                        Ok(results) => {
                            for r in results {
                                if let Ok(value) = r.value.1.parse::<f64>() {
                                    metrics.push(Metric {
                                        name: r
                                            .metric
                                            .get("__name__")
                                            .cloned()
                                            .unwrap_or_else(|| name.clone()),
                                        metric_type: MetricType::Gauge.into(),
                                        value,
                                        unit: String::new(),
                                        labels: r.metric,
                                        timestamp: chrono::Utc::now().timestamp(),
                                        statistics: None,
                                    });
                                }
                            }
                        }
                        Err(e) => {
                            error!("StreamMetrics: Prometheus query for '{}' failed: {}", name, e);
                        }
                    }
                }

                let event = MetricsEvent {
                    metrics,
                    timestamp: chrono::Utc::now().timestamp(),
                };

                yield Ok(event);
            }
        };

        Ok(Response::new(Box::pin(stream)))
    }

    async fn get_latency_metrics(
        &self,
        _request: Request<GetLatencyMetricsRequest>,
    ) -> Result<Response<GetLatencyMetricsResponse>, Status> {
        Err(Status::unimplemented(
            "GetLatencyMetrics not yet implemented",
        ))
    }

    async fn get_throughput_metrics(
        &self,
        _request: Request<GetThroughputMetricsRequest>,
    ) -> Result<Response<GetThroughputMetricsResponse>, Status> {
        Err(Status::unimplemented(
            "GetThroughputMetrics not yet implemented",
        ))
    }

    type StreamAlertsStream =
        Pin<Box<dyn Stream<Item = Result<AlertEvent, Status>> + Send>>;

    #[instrument(skip(self, request), fields(request_id = %uuid::Uuid::new_v4()), err)]
    async fn stream_alerts(
        &self,
        request: Request<crate::monitoring::StreamAlertsRequest>,
    ) -> Result<Response<Self::StreamAlertsStream>, Status> {
        let _req = request.into_inner();
        Err(Status::unimplemented(
            "StreamAlerts will be wired to Prometheus Alertmanager",
        ))
    }

    async fn acknowledge_alert(
        &self,
        _request: Request<AcknowledgeAlertRequest>,
    ) -> Result<Response<AcknowledgeAlertResponse>, Status> {
        Err(Status::unimplemented(
            "AcknowledgeAlert not yet implemented",
        ))
    }

    async fn get_active_alerts(
        &self,
        _request: Request<GetActiveAlertsRequest>,
    ) -> Result<Response<GetActiveAlertsResponse>, Status> {
        Err(Status::unimplemented(
            "GetActiveAlerts not yet implemented",
        ))
    }

    // ========================================================================
    // Cluster pods RPC (real -- queries Kubernetes API via pods_handler)
    // ========================================================================

    type SubscribeClusterPodsStream =
        Pin<Box<dyn Stream<Item = Result<ClusterPodsResponse, Status>> + Send>>;

    async fn subscribe_cluster_pods(
        &self,
        request: Request<SubscribeClusterPodsRequest>,
    ) -> Result<Response<Self::SubscribeClusterPodsStream>, Status> {
        let req = request.into_inner();
        let interval_secs = if req.interval_seconds == 0 {
            5
        } else {
            req.interval_seconds.clamp(1, 60)
        };

        info!(
            "SubscribeClusterPods: interval={}s, namespace=foxhunt",
            interval_secs
        );

        let stream = async_stream::stream! {
            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
            loop {
                interval.tick().await;
                match super::pods_handler::list_pods("foxhunt").await {
                    Ok(pods) => {
                        yield Ok(ClusterPodsResponse {
                            pods,
                            timestamp: chrono::Utc::now().timestamp(),
                        });
                    }
                    Err(e) => {
                        tracing::warn!("Pod listing failed: {e}");
                        yield Ok(ClusterPodsResponse {
                            pods: vec![],
                            timestamp: chrono::Utc::now().timestamp(),
                        });
                    }
                }
            }
        };

        Ok(Response::new(Box::pin(stream)))
    }
}

// ============================================================================
// Helper functions (ported from monitoring_service/src/service.rs)
// ============================================================================

/// Group flat metric samples into TrainingSession structs keyed by (model, fold)
fn group_into_sessions(samples: &[MetricSample], model_filter: &str) -> Vec<TrainingSession> {
    let mut map: HashMap<(String, String), TrainingSession> = HashMap::new();

    for s in samples {
        // Skip metrics without a model label (e.g. foxhunt_training_active_workers)
        if s.model.is_empty() {
            continue;
        }
        if !model_filter.is_empty() && s.model != model_filter {
            continue;
        }

        let key = (s.model.clone(), s.fold.clone());
        let session = map.entry(key).or_insert_with(|| TrainingSession {
            model: s.model.clone(),
            fold: s.fold.clone(),
            ..Default::default()
        });

        #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
        match s.name.as_str() {
            "foxhunt_training_current_epoch" => session.current_epoch = s.value as f32,
            "foxhunt_training_epoch_loss" => session.epoch_loss = s.value as f32,
            "foxhunt_training_validation_loss" => session.validation_loss = s.value as f32,
            "foxhunt_training_batches_per_second" => session.batches_per_second = s.value as f32,
            "foxhunt_training_batches_processed" => session.batches_processed = s.value as f32,
            "foxhunt_training_iteration_seconds" => session.iteration_seconds = s.value as f32,
            "foxhunt_training_eval_accuracy" => session.eval_accuracy = s.value as f32,
            "foxhunt_training_eval_precision" => session.eval_precision = s.value as f32,
            "foxhunt_training_eval_recall" => session.eval_recall = s.value as f32,
            "foxhunt_training_eval_f1" => session.eval_f1 = s.value as f32,
            "foxhunt_training_checkpoint_size_bytes" => {
                session.checkpoint_size_bytes = s.value as f32;
            }
            "foxhunt_training_checkpoint_saves_total" => {
                session.checkpoint_saves = s.value as u32;
            }
            "foxhunt_training_checkpoint_failures_total" => {
                session.checkpoint_failures = s.value as u32;
            }
            "foxhunt_training_nan_detected_total" => session.nan_detected = s.value as u32,
            "foxhunt_training_gradient_explosion_total" => {
                session.gradient_explosions = s.value as u32;
            }
            "foxhunt_training_feature_errors_total" => session.feature_errors = s.value as u32,
            "foxhunt_hyperopt_trial_current" => {
                session.hyperopt_trial_current = s.value as u32;
                session.is_hyperopt = true;
            }
            "foxhunt_hyperopt_trial_total" => session.hyperopt_trial_total = s.value as u32,
            "foxhunt_hyperopt_best_objective" => {
                session.hyperopt_best_objective = s.value as f32;
            }
            "foxhunt_hyperopt_trials_failed_total" => {
                session.hyperopt_trials_failed = s.value as u32;
            }
            "foxhunt_hyperopt_mode" => {
                if s.value > 0.5 {
                    session.is_hyperopt = true;
                }
            }
            // RL diagnostics
            "foxhunt_training_q_value_mean" => session.q_value_mean = s.value as f32,
            "foxhunt_training_q_value_max" => session.q_value_max = s.value as f32,
            "foxhunt_training_policy_entropy" => session.policy_entropy = s.value as f32,
            "foxhunt_training_kl_divergence" => session.kl_divergence = s.value as f32,
            "foxhunt_training_advantage_mean" => session.advantage_mean = s.value as f32,
            "foxhunt_training_replay_buffer_size" => {
                session.replay_buffer_size = s.value as u32;
            }
            // Gradient & training health
            "foxhunt_training_gradient_norm" => session.gradient_norm = s.value as f32,
            "foxhunt_training_learning_rate" => session.learning_rate = s.value as f32,
            "foxhunt_training_epoch_duration_seconds" => {
                session.epoch_duration_seconds = s.value as f32;
            }
            // Hyperopt intra-trial
            "foxhunt_hyperopt_trial_epoch" => session.hyperopt_trial_epoch = s.value as u32,
            "foxhunt_hyperopt_trial_best_loss" => {
                session.hyperopt_trial_best_loss = s.value as f32;
            }
            "foxhunt_hyperopt_elapsed_seconds" => {
                session.hyperopt_elapsed_seconds = s.value as f32;
            }
            // Epoch-level financial metrics
            "foxhunt_training_epoch_sharpe" => session.epoch_sharpe = s.value as f32,
            "foxhunt_training_epoch_sortino" => session.epoch_sortino = s.value as f32,
            "foxhunt_training_epoch_win_rate" => session.epoch_win_rate = s.value as f32,
            "foxhunt_training_epoch_max_drawdown" => {
                session.epoch_max_drawdown = s.value as f32;
            }
            "foxhunt_training_epoch_profit_factor" => {
                session.epoch_profit_factor = s.value as f32;
            }
            "foxhunt_training_epoch_total_return" => {
                session.epoch_total_return = s.value as f32;
            }
            "foxhunt_training_epoch_avg_return" => {
                session.epoch_avg_return = s.value as f32;
            }
            "foxhunt_training_epoch_total_trades" => {
                session.epoch_total_trades = s.value as u32;
            }
            // Action distribution
            "foxhunt_training_epoch_action_buy_pct" => session.action_buy_pct = s.value as f32,
            "foxhunt_training_epoch_action_sell_pct" => session.action_sell_pct = s.value as f32,
            "foxhunt_training_epoch_action_hold_pct" => session.action_hold_pct = s.value as f32,
            _ => {}
        }
    }

    let mut sessions: Vec<_> = map.into_values().collect();
    sessions.sort_by(|a, b| (&a.model, &a.fold).cmp(&(&b.model, &b.fold)));
    sessions
}

fn build_gpu_snapshot(samples: &[MetricSample]) -> GpuSnapshot {
    let mut snap = GpuSnapshot::default();
    let mut fb_free: f32 = 0.0;
    #[allow(clippy::cast_possible_truncation)]
    for s in samples {
        match s.name.as_str() {
            "dcgm_gpu_utilization" => snap.utilization_percent = s.value as f32,
            "dcgm_fb_used" => snap.memory_used_mb = s.value as f32,
            "dcgm_fb_free" => fb_free = s.value as f32,
            "dcgm_gpu_temp" => snap.temperature_celsius = s.value as f32,
            "dcgm_power_usage" => snap.power_watts = s.value as f32,
            _ => {}
        }
    }
    // dcgm_fb_free + dcgm_fb_used = total
    snap.memory_total_mb = snap.memory_used_mb + fb_free;
    snap
}

// ============================================================================
// Tests (ported from monitoring_service/src/service.rs)
// ============================================================================

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_handler_creation() {
        let handler = MonitoringServiceHandler::new("http://localhost:9090", 3);
        assert_eq!(handler.default_interval, 3);
    }

    #[test]
    fn test_group_empty_samples() {
        let sessions = group_into_sessions(&[], "");
        assert!(sessions.is_empty());
    }

    #[test]
    fn test_group_with_filter() {
        let samples = vec![
            MetricSample {
                name: "foxhunt_training_current_epoch".to_owned(),
                model: "dqn".to_owned(),
                fold: "0".to_owned(),
                value: 5.0,
            },
            MetricSample {
                name: "foxhunt_training_current_epoch".to_owned(),
                model: "ppo".to_owned(),
                fold: "0".to_owned(),
                value: 3.0,
            },
        ];
        let sessions = group_into_sessions(&samples, "dqn");
        assert_eq!(sessions.len(), 1);
        assert_eq!(sessions.first().map(|s| s.model.as_str()), Some("dqn"));
        assert_eq!(sessions.first().map(|s| s.current_epoch), Some(5.0));
    }

    #[test]
    fn test_group_sets_hyperopt_flag() {
        let samples = vec![MetricSample {
            name: "foxhunt_hyperopt_mode".to_owned(),
            model: "dqn".to_owned(),
            fold: "".to_owned(),
            value: 1.0,
        }];
        let sessions = group_into_sessions(&samples, "");
        assert_eq!(sessions.len(), 1);
        assert!(sessions.first().map(|s| s.is_hyperopt).unwrap_or(false));
    }

    #[test]
    fn test_group_new_diagnostic_metrics() {
        let samples = vec![
            MetricSample {
                name: "foxhunt_training_current_epoch".to_owned(),
                model: "dqn".to_owned(),
                fold: "0".to_owned(),
                value: 25.0,
            },
            MetricSample {
                name: "foxhunt_training_q_value_mean".to_owned(),
                model: "dqn".to_owned(),
                fold: "0".to_owned(),
                value: 12.3,
            },
            MetricSample {
                name: "foxhunt_training_q_value_max".to_owned(),
                model: "dqn".to_owned(),
                fold: "0".to_owned(),
                value: 45.6,
            },
            MetricSample {
                name: "foxhunt_training_gradient_norm".to_owned(),
                model: "dqn".to_owned(),
                fold: "0".to_owned(),
                value: 1.23,
            },
            MetricSample {
                name: "foxhunt_training_learning_rate".to_owned(),
                model: "dqn".to_owned(),
                fold: "0".to_owned(),
                value: 0.001,
            },
            MetricSample {
                name: "foxhunt_training_epoch_duration_seconds".to_owned(),
                model: "dqn".to_owned(),
                fold: "0".to_owned(),
                value: 12.5,
            },
            MetricSample {
                name: "foxhunt_training_replay_buffer_size".to_owned(),
                model: "dqn".to_owned(),
                fold: "0".to_owned(),
                value: 50000.0,
            },
        ];
        let sessions = group_into_sessions(&samples, "");
        assert_eq!(sessions.len(), 1);
        let s = &sessions[0];
        assert_eq!(s.current_epoch, 25.0);
        assert!((s.q_value_mean - 12.3).abs() < 0.01);
        assert!((s.q_value_max - 45.6).abs() < 0.01);
        assert!((s.gradient_norm - 1.23).abs() < 0.01);
        assert!((s.learning_rate - 0.001).abs() < 0.0001);
        assert!((s.epoch_duration_seconds - 12.5).abs() < 0.01);
        assert_eq!(s.replay_buffer_size, 50000);
    }

    #[test]
    fn test_group_ppo_diagnostics() {
        let samples = vec![
            MetricSample {
                name: "foxhunt_training_policy_entropy".to_owned(),
                model: "ppo".to_owned(),
                fold: "0".to_owned(),
                value: 1.23,
            },
            MetricSample {
                name: "foxhunt_training_kl_divergence".to_owned(),
                model: "ppo".to_owned(),
                fold: "0".to_owned(),
                value: 0.008,
            },
            MetricSample {
                name: "foxhunt_training_advantage_mean".to_owned(),
                model: "ppo".to_owned(),
                fold: "0".to_owned(),
                value: 0.001,
            },
        ];
        let sessions = group_into_sessions(&samples, "");
        assert_eq!(sessions.len(), 1);
        let s = &sessions[0];
        assert!((s.policy_entropy - 1.23).abs() < 0.01);
        assert!((s.kl_divergence - 0.008).abs() < 0.001);
        assert!((s.advantage_mean - 0.001).abs() < 0.001);
    }

    #[test]
    fn test_group_hyperopt_intra_trial() {
        let samples = vec![
            MetricSample {
                name: "foxhunt_hyperopt_mode".to_owned(),
                model: "dqn".to_owned(),
                fold: "".to_owned(),
                value: 1.0,
            },
            MetricSample {
                name: "foxhunt_hyperopt_trial_epoch".to_owned(),
                model: "dqn".to_owned(),
                fold: "".to_owned(),
                value: 12.0,
            },
            MetricSample {
                name: "foxhunt_hyperopt_trial_best_loss".to_owned(),
                model: "dqn".to_owned(),
                fold: "".to_owned(),
                value: 0.042,
            },
            MetricSample {
                name: "foxhunt_hyperopt_elapsed_seconds".to_owned(),
                model: "dqn".to_owned(),
                fold: "".to_owned(),
                value: 123.4,
            },
        ];
        let sessions = group_into_sessions(&samples, "");
        assert_eq!(sessions.len(), 1);
        let s = &sessions[0];
        assert!(s.is_hyperopt);
        assert_eq!(s.hyperopt_trial_epoch, 12);
        assert!((s.hyperopt_trial_best_loss - 0.042).abs() < 0.001);
        assert!((s.hyperopt_elapsed_seconds - 123.4).abs() < 0.1);
    }

    #[test]
    fn test_group_financial_metrics() {
        let samples = vec![
            MetricSample {
                name: "foxhunt_training_epoch_sharpe".to_owned(),
                model: "dqn".to_owned(),
                fold: "0".to_owned(),
                value: 2.31,
            },
            MetricSample {
                name: "foxhunt_training_epoch_win_rate".to_owned(),
                model: "dqn".to_owned(),
                fold: "0".to_owned(),
                value: 0.552,
            },
            MetricSample {
                name: "foxhunt_training_epoch_max_drawdown".to_owned(),
                model: "dqn".to_owned(),
                fold: "0".to_owned(),
                value: 0.081,
            },
            MetricSample {
                name: "foxhunt_training_epoch_action_buy_pct".to_owned(),
                model: "dqn".to_owned(),
                fold: "0".to_owned(),
                value: 0.35,
            },
        ];
        let sessions = group_into_sessions(&samples, "");
        assert_eq!(sessions.len(), 1);
        let s = &sessions[0];
        assert!((s.epoch_sharpe - 2.31).abs() < 0.01);
        assert!((s.epoch_win_rate - 0.552).abs() < 0.001);
        assert!((s.epoch_max_drawdown - 0.081).abs() < 0.001);
        assert!((s.action_buy_pct - 0.35).abs() < 0.01);
    }

    #[test]
    fn test_build_gpu_snapshot() {
        let samples = vec![
            MetricSample {
                name: "dcgm_gpu_utilization".to_owned(),
                model: String::new(),
                fold: String::new(),
                value: 87.0,
            },
            MetricSample {
                name: "dcgm_fb_used".to_owned(),
                model: String::new(),
                fold: String::new(),
                value: 38200.0,
            },
            MetricSample {
                name: "dcgm_fb_free".to_owned(),
                model: String::new(),
                fold: String::new(),
                value: 9800.0,
            },
            MetricSample {
                name: "dcgm_gpu_temp".to_owned(),
                model: String::new(),
                fold: String::new(),
                value: 62.0,
            },
            MetricSample {
                name: "dcgm_power_usage".to_owned(),
                model: String::new(),
                fold: String::new(),
                value: 245.0,
            },
        ];
        let snap = build_gpu_snapshot(&samples);
        assert_eq!(snap.utilization_percent, 87.0);
        assert_eq!(snap.memory_used_mb, 38200.0);
        assert_eq!(snap.memory_total_mb, 48000.0);
        assert_eq!(snap.temperature_celsius, 62.0);
        assert_eq!(snap.power_watts, 245.0);
    }
}