//! gRPC streaming data fetcher for the TUI dashboard.
//!
//! Spawns one tokio task per stream. Each task sends [`StateUpdate`] variants
//! through an unbounded channel that the event loop drains every tick.

// Channel send() is intentionally fire-and-forget (UI best-effort).
// Integer division is used for time formatting (intentional truncation).
#![allow(
    clippy::let_underscore_must_use,
    clippy::integer_division,
    clippy::modulo_arithmetic
)]

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tonic::Request;
use tracing::{debug, error, warn};

use crate::grpc::FoxhuntClient;
use crate::proto::{broker_gateway, data_acquisition, ml, ml_training, monitoring, risk, trading};

use super::state::{
    AccountData, AlertData, CircuitBreakerData, ConnectionStatus, DataCacheData, DataFeedData,
    ExecutionData, GpuData, ModelStatusData, OrderEventData, PodData, PodGroupData, PositionData,
    RiskAlertData, RiskData, ServiceData, SystemResources, TrainingHistoryEntry,
    TrainingSessionData,
};

/// Empty filter string means "fetch all" in gRPC requests.
const FILTER_ALL: &str = "";

// ---------------------------------------------------------------------------
// StateUpdate – one variant per data source
// ---------------------------------------------------------------------------

/// Update sent from a background gRPC stream to the event loop.
pub enum StateUpdate {
    /// Training sessions + GPU telemetry (from StreamTrainingMetrics).
    TrainingMetrics {
        sessions: Vec<TrainingSessionData>,
        gpu: GpuData,
        resources: SystemResources,
        active_jobs: u32,
    },
    /// Service health statuses (from GetSystemStatus polling).
    Services(Vec<ServiceData>),
    /// Account financials (from StreamAccountState).
    Account(AccountData),
    /// Open trading positions (from StreamPortfolioSummary).
    Positions(Vec<PositionData>),
    /// Trade executions (from StreamExecutions).
    Execution(ExecutionData),
    /// Order lifecycle events (from StreamOrders).
    Order(OrderEventData),
    /// Risk metrics (from StreamRiskMetrics).
    Risk(RiskData),
    /// Circuit breaker statuses (from StreamCircuitBreakerStatus).
    CircuitBreakers(Vec<CircuitBreakerData>),
    /// Risk alerts (from StreamRiskAlerts).
    RiskAlert(RiskAlertData),
    /// ML model statuses (from StreamModelStatus).
    ModelStatuses(Vec<ModelStatusData>),
    /// System alerts (from StreamAlerts).
    Alert(AlertData),
    /// Data download jobs (from StreamDownloadStatus).
    DataFeeds(Vec<DataFeedData>),
    /// Aggregate data cache stats (computed from download job summaries).
    DataCache(DataCacheData),
    /// Portfolio summary — equity, margin, pnl (from StreamPortfolioSummary).
    Portfolio {
        equity: f64,
        cash: f64,
        margin_used: f64,
        daily_pnl: f64,
        total_pnl: f64,
    },
    /// Cluster pod statuses (from SubscribeClusterPods).
    Pods(Vec<PodGroupData>),
    /// Completed/failed training runs (from ListTrainingJobs polling).
    TrainingHistory(Vec<TrainingHistoryEntry>),
    /// gRPC connection status change.
    Connection(ConnectionStatus),
}

// ---------------------------------------------------------------------------
// DataFetcher
// ---------------------------------------------------------------------------

/// Manages background gRPC stream tasks.
///
/// Call [`DataFetcher::start`] to spawn all stream listeners. Drop the
/// [`CancellationToken`] (or call `.cancel()`) for graceful shutdown.
/// Maximum reconnection attempts before reporting a connection error.
const MAX_RECONNECT_ATTEMPTS: u32 = 10;

pub struct DataFetcher {
    tx: mpsc::UnboundedSender<StateUpdate>,
    cancel: CancellationToken,
    /// Shared flag: set to `true` once any stream receives its first
    /// successful message, triggering a single `ConnectionStatus::Connected`.
    connected_sent: Arc<AtomicBool>,
}

impl DataFetcher {
    /// Create a new fetcher. Returns the fetcher and the receiving half of the
    /// update channel.
    pub fn new() -> (Self, mpsc::UnboundedReceiver<StateUpdate>) {
        let (tx, rx) = mpsc::unbounded_channel();
        let cancel = CancellationToken::new();
        let connected_sent = Arc::new(AtomicBool::new(false));
        (Self { tx, cancel, connected_sent }, rx)
    }

    /// Cancellation token — clone and pass to the event loop so it can
    /// signal shutdown.
    pub fn cancel_token(&self) -> CancellationToken {
        self.cancel.clone()
    }

    /// Spawn all background stream tasks.
    pub fn start(&self, client: &FoxhuntClient) {
        self.spawn_training_metrics(client);
        self.spawn_training_history(client);
        self.spawn_system_status(client);
        self.spawn_account_state(client);
        self.spawn_executions(client);
        self.spawn_orders(client);
        self.spawn_portfolio_summary(client);
        self.spawn_download_status(client);
        self.spawn_risk_metrics(client);
        self.spawn_circuit_breakers(client);
        self.spawn_risk_alerts(client);
        self.spawn_model_status(client);
        self.spawn_alerts(client);
        self.spawn_pods_stream(client);
    }

    // ── Stream spawners ─────────────────────────────────────────────

    fn spawn_training_metrics(&self, client: &FoxhuntClient) {
        let mut mon = client.monitoring();
        let tx = self.tx.clone();
        let cancel = self.cancel.clone();
        let connected_sent = self.connected_sent.clone();

        tokio::spawn(async move {
            let mut backoff = Backoff::new();
            loop {
                let req = Request::new(monitoring::StreamTrainingMetricsRequest {
                    model_filter: FILTER_ALL.into(),
                    interval_seconds: 3,
                });
                match mon.stream_training_metrics(req).await {
                    Ok(resp) => {
                        backoff.reset();
                        let mut stream = resp.into_inner();
                        loop {
                            tokio::select! {
                                _ = cancel.cancelled() => return,
                                msg = stream.message() => {
                                    match msg {
                                        Ok(Some(m)) => {
                                            if !connected_sent.swap(true, Ordering::Relaxed) {
                                                let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Connected));
                                            }
                                            let sessions = dedup_sessions(&m.sessions);
                                            let gpu = m.gpu.as_ref().map(convert_gpu).unwrap_or_default();
                                            let resources = SystemResources {
                                                cpu_usage_pct: f64::from(m.cpu_percent),
                                                ram_used_gb: f64::from(m.memory_used_mb) / 1024.0,
                                                ram_total_gb: f64::from(m.memory_total_mb) / 1024.0,
                                                gpu_cluster_utilization: gpu.utilization,
                                            };
                                            let active_jobs = m.active_k8s_jobs;
                                            let _ = tx.send(StateUpdate::TrainingMetrics { sessions, gpu, resources, active_jobs });
                                        }
                                        Ok(None) => {
                                            debug!("TrainingMetrics stream ended");
                                            break;
                                        }
                                        Err(e) => {
                                            warn!("TrainingMetrics stream error: {e}");
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                    }
                    Err(e) => {
                        warn!("TrainingMetrics connect failed: {e}");
                        let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Reconnecting {
                            attempt: backoff.attempts,
                        }));
                    }
                }

                if backoff.attempts >= MAX_RECONNECT_ATTEMPTS {
                    error!("TrainingMetrics exceeded {MAX_RECONNECT_ATTEMPTS} reconnect attempts, giving up");
                    let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Error(
                        format!("Failed to connect after {MAX_RECONNECT_ATTEMPTS} attempts"),
                    )));
                    return;
                }

                tokio::select! {
                    _ = cancel.cancelled() => return,
                    _ = tokio::time::sleep(backoff.next_delay()) => {}
                }
            }
        });
    }

    /// Training history — polls ListTrainingJobs every 30s for the History tab.
    fn spawn_training_history(&self, client: &FoxhuntClient) {
        let mut ml_train = client.ml_training();
        let tx = self.tx.clone();
        let cancel = self.cancel.clone();

        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(30));
            loop {
                tokio::select! {
                    _ = cancel.cancelled() => return,
                    _ = interval.tick() => {}
                }

                let req = Request::new(ml_training::ListTrainingJobsRequest {
                    page: 0,
                    page_size: 50,
                    status_filter: 0, // all statuses
                    model_type_filter: String::new(),
                    start_time: 0,
                    end_time: 0,
                });
                match ml_train.list_training_jobs(req).await {
                    Ok(resp) => {
                        let inner = resp.into_inner();
                        let entries: Vec<TrainingHistoryEntry> = inner
                            .jobs
                            .iter()
                            .map(|j| {
                                let status = match ml_training::TrainingStatus::try_from(j.status) {
                                    Ok(ml_training::TrainingStatus::Pending) => "pending",
                                    Ok(ml_training::TrainingStatus::Running) => "running",
                                    Ok(ml_training::TrainingStatus::Completed) => "completed",
                                    Ok(ml_training::TrainingStatus::Failed) => "failed",
                                    Ok(ml_training::TrainingStatus::Stopped) => "stopped",
                                    Ok(ml_training::TrainingStatus::Paused) => "paused",
                                    _ => "unknown",
                                };
                                TrainingHistoryEntry {
                                    job_id: j.job_id.clone(),
                                    model: j.model_type.clone(),
                                    status: status.to_owned(),
                                    final_loss: j.final_loss,
                                    best_val: j.best_validation_score,
                                    created_at: j.created_at,
                                    started_at: j.started_at,
                                    completed_at: j.completed_at,
                                    description: j.description.clone(),
                                }
                            })
                            .collect();
                        let _ = tx.send(StateUpdate::TrainingHistory(entries));
                    }
                    Err(e) => {
                        debug!("ListTrainingJobs poll failed: {e}");
                    }
                }
            }
        });
    }

    /// Service statuses — poll GetSystemStatus every 5s (the stream event
    /// only carries SystemMetrics, not individual ServiceStatus entries).
    /// After 3 consecutive failures, reports a connection error to indicate
    /// stale data in the services cockpit.
    fn spawn_system_status(&self, client: &FoxhuntClient) {
        let mut mon = client.monitoring();
        let tx = self.tx.clone();
        let cancel = self.cancel.clone();
        let connected_sent = self.connected_sent.clone();

        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(5));
            let mut consecutive_failures: u32 = 0;
            loop {
                tokio::select! {
                    _ = cancel.cancelled() => return,
                    _ = interval.tick() => {}
                }

                let req = Request::new(monitoring::GetSystemStatusRequest {
                    service_names: Vec::new(),
                });
                match mon.get_system_status(req).await {
                    Ok(resp) => {
                        consecutive_failures = 0;
                        if !connected_sent.swap(true, Ordering::Relaxed) {
                            let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Connected));
                        }
                        let inner = resp.into_inner();
                        let services: Vec<ServiceData> = inner
                            .service_statuses
                            .iter()
                            .map(convert_service_status)
                            .collect();
                        let _ = tx.send(StateUpdate::Services(services));
                    }
                    Err(e) => {
                        consecutive_failures = consecutive_failures.saturating_add(1);
                        debug!("SystemStatus poll failed ({consecutive_failures}/3): {e}");
                        if consecutive_failures >= 3 {
                            warn!("SystemStatus: {consecutive_failures} consecutive failures, reporting stale");
                            let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Error(
                                format!("SystemStatus poll failed {consecutive_failures} times: {e}"),
                            )));
                        }
                    }
                }
            }
        });
    }

    fn spawn_account_state(&self, client: &FoxhuntClient) {
        let mut broker = client.broker_gateway();
        let tx = self.tx.clone();
        let cancel = self.cancel.clone();
        let connected_sent = self.connected_sent.clone();

        tokio::spawn(async move {
            let mut backoff = Backoff::new();
            loop {
                let req = Request::new(broker_gateway::StreamAccountStateRequest {
                    interval_seconds: 3,
                });
                match broker.stream_account_state(req).await {
                    Ok(resp) => {
                        backoff.reset();
                        let mut stream = resp.into_inner();
                        loop {
                            tokio::select! {
                                _ = cancel.cancelled() => return,
                                msg = stream.message() => {
                                    match msg {
                                        Ok(Some(acct)) => {
                                            if !connected_sent.swap(true, Ordering::Relaxed) {
                                                let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Connected));
                                            }
                                            let _ = tx.send(StateUpdate::Account(convert_account(&acct)));
                                        }
                                        Ok(None) => break,
                                        Err(e) => {
                                            warn!("AccountState stream error: {e}");
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                    }
                    Err(e) => {
                        warn!("AccountState connect failed: {e}");
                    }
                }

                if backoff.attempts >= MAX_RECONNECT_ATTEMPTS {
                    error!("AccountState exceeded {MAX_RECONNECT_ATTEMPTS} reconnect attempts, giving up");
                    return;
                }
                tokio::select! {
                    _ = cancel.cancelled() => return,
                    _ = tokio::time::sleep(backoff.next_delay()) => {}
                }
            }
        });
    }

    fn spawn_executions(&self, client: &FoxhuntClient) {
        let mut broker = client.broker_gateway();
        let tx = self.tx.clone();
        let cancel = self.cancel.clone();
        let connected_sent = self.connected_sent.clone();

        tokio::spawn(async move {
            let mut backoff = Backoff::new();
            loop {
                let req = Request::new(broker_gateway::StreamExecutionsRequest {
                    account_id: None,
                    symbol: None,
                });
                match broker.stream_executions(req).await {
                    Ok(resp) => {
                        backoff.reset();
                        let mut stream = resp.into_inner();
                        loop {
                            tokio::select! {
                                _ = cancel.cancelled() => return,
                                msg = stream.message() => {
                                    match msg {
                                        Ok(Some(exec)) => {
                                            if !connected_sent.swap(true, Ordering::Relaxed) {
                                                let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Connected));
                                            }
                                            let _ = tx.send(StateUpdate::Execution(convert_execution(&exec)));
                                        }
                                        Ok(None) => break,
                                        Err(e) => {
                                            warn!("Executions stream error: {e}");
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                    }
                    Err(e) => {
                        warn!("Executions connect failed: {e}");
                    }
                }

                if backoff.attempts >= MAX_RECONNECT_ATTEMPTS {
                    error!("Executions exceeded {MAX_RECONNECT_ATTEMPTS} reconnect attempts, giving up");
                    return;
                }
                tokio::select! {
                    _ = cancel.cancelled() => return,
                    _ = tokio::time::sleep(backoff.next_delay()) => {}
                }
            }
        });
    }

    fn spawn_orders(&self, client: &FoxhuntClient) {
        let mut trd = client.trading();
        let tx = self.tx.clone();
        let cancel = self.cancel.clone();
        let connected_sent = self.connected_sent.clone();

        tokio::spawn(async move {
            let mut backoff = Backoff::new();
            loop {
                let req = Request::new(trading::StreamOrdersRequest {
                    account_id: None,
                    symbol: None,
                });
                match trd.stream_orders(req).await {
                    Ok(resp) => {
                        backoff.reset();
                        let mut stream = resp.into_inner();
                        loop {
                            tokio::select! {
                                _ = cancel.cancelled() => return,
                                msg = stream.message() => {
                                    match msg {
                                        Ok(Some(evt)) => {
                                            if !connected_sent.swap(true, Ordering::Relaxed) {
                                                let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Connected));
                                            }
                                            let _ = tx.send(StateUpdate::Order(convert_order_event(&evt)));
                                        }
                                        Ok(None) => break,
                                        Err(e) => {
                                            warn!("Orders stream error: {e}");
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                    }
                    Err(e) => {
                        warn!("Orders connect failed: {e}");
                    }
                }

                if backoff.attempts >= MAX_RECONNECT_ATTEMPTS {
                    error!("Orders exceeded {MAX_RECONNECT_ATTEMPTS} reconnect attempts, giving up");
                    return;
                }
                tokio::select! {
                    _ = cancel.cancelled() => return,
                    _ = tokio::time::sleep(backoff.next_delay()) => {}
                }
            }
        });
    }

    fn spawn_portfolio_summary(&self, client: &FoxhuntClient) {
        let mut trd = client.trading();
        let tx = self.tx.clone();
        let cancel = self.cancel.clone();
        let connected_sent = self.connected_sent.clone();

        tokio::spawn(async move {
            let mut backoff = Backoff::new();
            loop {
                let req = Request::new(trading::StreamPortfolioSummaryRequest {
                    account_id: FILTER_ALL.into(),
                    interval_seconds: 3,
                });
                match trd.stream_portfolio_summary(req).await {
                    Ok(resp) => {
                        backoff.reset();
                        let mut stream = resp.into_inner();
                        loop {
                            tokio::select! {
                                _ = cancel.cancelled() => return,
                                msg = stream.message() => {
                                    match msg {
                                        Ok(Some(ps)) => {
                                            if !connected_sent.swap(true, Ordering::Relaxed) {
                                                let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Connected));
                                            }
                                            let positions = ps.positions.iter().map(convert_position).collect();
                                            let _ = tx.send(StateUpdate::Positions(positions));
                                            let _ = tx.send(StateUpdate::Portfolio {
                                                equity: ps.total_value,
                                                cash: ps.buying_power,
                                                margin_used: ps.margin_used,
                                                daily_pnl: ps.day_pnl,
                                                total_pnl: ps.unrealized_pnl + ps.realized_pnl,
                                            });
                                        }
                                        Ok(None) => break,
                                        Err(e) => {
                                            warn!("PortfolioSummary stream error: {e}");
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                    }
                    Err(e) => {
                        warn!("PortfolioSummary connect failed: {e}");
                    }
                }

                if backoff.attempts >= MAX_RECONNECT_ATTEMPTS {
                    error!("PortfolioSummary exceeded {MAX_RECONNECT_ATTEMPTS} reconnect attempts, giving up");
                    return;
                }
                tokio::select! {
                    _ = cancel.cancelled() => return,
                    _ = tokio::time::sleep(backoff.next_delay()) => {}
                }
            }
        });
    }

    fn spawn_download_status(&self, client: &FoxhuntClient) {
        let mut da = client.data_acquisition();
        let tx = self.tx.clone();
        let cancel = self.cancel.clone();
        let connected_sent = self.connected_sent.clone();

        tokio::spawn(async move {
            let mut backoff = Backoff::new();
            loop {
                let req = Request::new(data_acquisition::StreamDownloadStatusRequest {
                    interval_seconds: 5,
                });
                match da.stream_download_status(req).await {
                    Ok(resp) => {
                        backoff.reset();
                        let mut stream = resp.into_inner();
                        loop {
                            tokio::select! {
                                _ = cancel.cancelled() => return,
                                msg = stream.message() => {
                                    match msg {
                                        Ok(Some(list)) => {
                                            if !connected_sent.swap(true, Ordering::Relaxed) {
                                                let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Connected));
                                            }
                                            let feeds = list.jobs.iter().map(convert_download_job).collect();
                                            let _ = tx.send(StateUpdate::DataFeeds(feeds));
                                            let cache = aggregate_cache_stats(&list.jobs);
                                            let _ = tx.send(StateUpdate::DataCache(cache));
                                        }
                                        Ok(None) => break,
                                        Err(e) => {
                                            warn!("DownloadStatus stream error: {e}");
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                    }
                    Err(e) => {
                        warn!("DownloadStatus connect failed: {e}");
                    }
                }

                if backoff.attempts >= MAX_RECONNECT_ATTEMPTS {
                    error!("DownloadStatus exceeded {MAX_RECONNECT_ATTEMPTS} reconnect attempts, giving up");
                    return;
                }
                tokio::select! {
                    _ = cancel.cancelled() => return,
                    _ = tokio::time::sleep(backoff.next_delay()) => {}
                }
            }
        });
    }

    /// Risk metrics — streams instead of polling.
    fn spawn_risk_metrics(&self, client: &FoxhuntClient) {
        let mut risk_client = client.risk();
        let tx = self.tx.clone();
        let cancel = self.cancel.clone();
        let connected_sent = self.connected_sent.clone();

        tokio::spawn(async move {
            let mut backoff = Backoff::new();
            loop {
                let req = Request::new(risk::StreamRiskMetricsRequest {
                    portfolio_id: None,
                    interval_seconds: 3,
                });
                match risk_client.stream_risk_metrics(req).await {
                    Ok(resp) => {
                        backoff.reset();
                        let mut stream = resp.into_inner();
                        loop {
                            tokio::select! {
                                _ = cancel.cancelled() => return,
                                msg = stream.message() => {
                                    match msg {
                                        Ok(Some(resp_inner)) => {
                                            if !connected_sent.swap(true, Ordering::Relaxed) {
                                                let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Connected));
                                            }
                                            if let Some(metrics) = resp_inner.metrics {
                                                let _ = tx.send(StateUpdate::Risk(convert_risk_metrics(&metrics)));
                                            }
                                        }
                                        Ok(None) => break,
                                        Err(e) => {
                                            warn!("RiskMetrics stream error: {e}");
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                    }
                    Err(e) => {
                        warn!("RiskMetrics connect failed: {e}");
                    }
                }

                if backoff.attempts >= MAX_RECONNECT_ATTEMPTS {
                    error!("RiskMetrics exceeded {MAX_RECONNECT_ATTEMPTS} reconnect attempts, giving up");
                    return;
                }
                tokio::select! {
                    _ = cancel.cancelled() => return,
                    _ = tokio::time::sleep(backoff.next_delay()) => {}
                }
            }
        });
    }

    /// Circuit breaker statuses — streams instead of polling.
    fn spawn_circuit_breakers(&self, client: &FoxhuntClient) {
        let mut risk_client = client.risk();
        let tx = self.tx.clone();
        let cancel = self.cancel.clone();
        let connected_sent = self.connected_sent.clone();

        tokio::spawn(async move {
            let mut backoff = Backoff::new();
            loop {
                let req = Request::new(risk::StreamCircuitBreakerStatusRequest {
                    symbol: None,
                    interval_seconds: 2,
                });
                match risk_client.stream_circuit_breaker_status(req).await {
                    Ok(resp) => {
                        backoff.reset();
                        let mut stream = resp.into_inner();
                        loop {
                            tokio::select! {
                                _ = cancel.cancelled() => return,
                                msg = stream.message() => {
                                    match msg {
                                        Ok(Some(resp_inner)) => {
                                            if !connected_sent.swap(true, Ordering::Relaxed) {
                                                let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Connected));
                                            }
                                            let cbs = resp_inner
                                                .circuit_breakers
                                                .iter()
                                                .map(convert_circuit_breaker)
                                                .collect();
                                            let _ = tx.send(StateUpdate::CircuitBreakers(cbs));
                                        }
                                        Ok(None) => break,
                                        Err(e) => {
                                            warn!("CircuitBreaker stream error: {e}");
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                    }
                    Err(e) => {
                        warn!("CircuitBreaker connect failed: {e}");
                    }
                }

                if backoff.attempts >= MAX_RECONNECT_ATTEMPTS {
                    error!("CircuitBreaker exceeded {MAX_RECONNECT_ATTEMPTS} reconnect attempts, giving up");
                    return;
                }
                tokio::select! {
                    _ = cancel.cancelled() => return,
                    _ = tokio::time::sleep(backoff.next_delay()) => {}
                }
            }
        });
    }

    /// Risk alerts — real-time risk violation stream.
    fn spawn_risk_alerts(&self, client: &FoxhuntClient) {
        let mut risk_client = client.risk();
        let tx = self.tx.clone();
        let cancel = self.cancel.clone();
        let connected_sent = self.connected_sent.clone();

        tokio::spawn(async move {
            let mut backoff = Backoff::new();
            loop {
                let req = Request::new(risk::StreamRiskAlertsRequest {
                    min_severity: 0, // all severities
                    alert_types: Vec::new(),
                });
                match risk_client.stream_risk_alerts(req).await {
                    Ok(resp) => {
                        backoff.reset();
                        let mut stream = resp.into_inner();
                        loop {
                            tokio::select! {
                                _ = cancel.cancelled() => return,
                                msg = stream.message() => {
                                    match msg {
                                        Ok(Some(evt)) => {
                                            if !connected_sent.swap(true, Ordering::Relaxed) {
                                                let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Connected));
                                            }
                                            let _ = tx.send(StateUpdate::RiskAlert(convert_risk_alert(&evt)));
                                        }
                                        Ok(None) => break,
                                        Err(e) => {
                                            warn!("RiskAlerts stream error: {e}");
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                    }
                    Err(e) => {
                        debug!("RiskAlerts connect failed: {e}");
                    }
                }

                if backoff.attempts >= MAX_RECONNECT_ATTEMPTS {
                    error!("RiskAlerts exceeded {MAX_RECONNECT_ATTEMPTS} reconnect attempts, giving up");
                    return;
                }
                tokio::select! {
                    _ = cancel.cancelled() => return,
                    _ = tokio::time::sleep(backoff.next_delay()) => {}
                }
            }
        });
    }

    /// ML model statuses — streams model health.
    fn spawn_model_status(&self, client: &FoxhuntClient) {
        let mut ml_client = client.ml();
        let tx = self.tx.clone();
        let cancel = self.cancel.clone();
        let connected_sent = self.connected_sent.clone();

        tokio::spawn(async move {
            let mut backoff = Backoff::new();
            loop {
                let req = Request::new(ml::StreamModelStatusRequest {
                    model_name: FILTER_ALL.into(), // all models
                    interval_seconds: 5,
                });
                match ml_client.stream_model_status(req).await {
                    Ok(resp) => {
                        backoff.reset();
                        let mut stream = resp.into_inner();
                        loop {
                            tokio::select! {
                                _ = cancel.cancelled() => return,
                                msg = stream.message() => {
                                    match msg {
                                        Ok(Some(status_resp)) => {
                                            if !connected_sent.swap(true, Ordering::Relaxed) {
                                                let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Connected));
                                            }
                                            let statuses = status_resp
                                                .model_statuses
                                                .iter()
                                                .map(convert_model_status)
                                                .collect();
                                            let _ = tx.send(StateUpdate::ModelStatuses(statuses));
                                        }
                                        Ok(None) => break,
                                        Err(e) => {
                                            warn!("ModelStatus stream error: {e}");
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                    }
                    Err(e) => {
                        debug!("ModelStatus connect failed: {e}");
                    }
                }

                if backoff.attempts >= MAX_RECONNECT_ATTEMPTS {
                    error!("ModelStatus exceeded {MAX_RECONNECT_ATTEMPTS} reconnect attempts, giving up");
                    return;
                }
                tokio::select! {
                    _ = cancel.cancelled() => return,
                    _ = tokio::time::sleep(backoff.next_delay()) => {}
                }
            }
        });
    }

    /// System alerts from monitoring service.
    fn spawn_alerts(&self, client: &FoxhuntClient) {
        let mut mon = client.monitoring();
        let tx = self.tx.clone();
        let cancel = self.cancel.clone();
        let connected_sent = self.connected_sent.clone();

        tokio::spawn(async move {
            let mut backoff = Backoff::new();
            loop {
                let req = Request::new(monitoring::StreamAlertsRequest {
                    min_severity: None,
                    service_names: Vec::new(),
                    alert_types: Vec::new(),
                });
                match mon.stream_alerts(req).await {
                    Ok(resp) => {
                        backoff.reset();
                        let mut stream = resp.into_inner();
                        loop {
                            tokio::select! {
                                _ = cancel.cancelled() => return,
                                msg = stream.message() => {
                                    match msg {
                                        Ok(Some(evt)) => {
                                            if !connected_sent.swap(true, Ordering::Relaxed) {
                                                let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Connected));
                                            }
                                            if let Some(alert) = evt.alert {
                                                let _ = tx.send(StateUpdate::Alert(convert_alert(&alert)));
                                            }
                                        }
                                        Ok(None) => break,
                                        Err(e) => {
                                            warn!("Alerts stream error: {e}");
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                    }
                    Err(e) => {
                        debug!("Alerts connect failed: {e}");
                    }
                }

                if backoff.attempts >= MAX_RECONNECT_ATTEMPTS {
                    error!("Alerts exceeded {MAX_RECONNECT_ATTEMPTS} reconnect attempts, giving up");
                    return;
                }
                tokio::select! {
                    _ = cancel.cancelled() => return,
                    _ = tokio::time::sleep(backoff.next_delay()) => {}
                }
            }
        });
    }

    /// Kubernetes pod statuses — streams from SubscribeClusterPods, grouped by
    /// service for the pods cockpit.
    fn spawn_pods_stream(&self, client: &FoxhuntClient) {
        let mut mon = client.monitoring();
        let tx = self.tx.clone();
        let cancel = self.cancel.clone();
        let connected_sent = self.connected_sent.clone();

        tokio::spawn(async move {
            let mut backoff = Backoff::new();
            loop {
                let req = Request::new(monitoring::SubscribeClusterPodsRequest {
                    interval_seconds: 5,
                });
                match mon.subscribe_cluster_pods(req).await {
                    Ok(resp) => {
                        backoff.reset();
                        let mut stream = resp.into_inner();
                        loop {
                            tokio::select! {
                                _ = cancel.cancelled() => return,
                                msg = stream.message() => {
                                    match msg {
                                        Ok(Some(m)) => {
                                            if !connected_sent.swap(true, Ordering::Relaxed) {
                                                let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Connected));
                                            }
                                            let groups = group_pods(m.pods);
                                            let _ = tx.send(StateUpdate::Pods(groups));
                                        }
                                        Ok(None) => break,
                                        Err(e) => {
                                            warn!("ClusterPods stream error: {e}");
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                    }
                    Err(e) => {
                        warn!("ClusterPods connect failed: {e}");
                        let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Reconnecting {
                            attempt: backoff.attempts,
                        }));
                    }
                }

                if backoff.attempts >= MAX_RECONNECT_ATTEMPTS {
                    error!("ClusterPods exceeded {MAX_RECONNECT_ATTEMPTS} reconnect attempts, giving up");
                    return;
                }
                tokio::select! {
                    _ = cancel.cancelled() => return,
                    _ = tokio::time::sleep(backoff.next_delay()) => {}
                }
            }
        });
    }
}

// ---------------------------------------------------------------------------
// Backoff helper
// ---------------------------------------------------------------------------

struct Backoff {
    attempts: u32,
}

impl Backoff {
    const MAX_DELAY: Duration = Duration::from_secs(30);

    fn new() -> Self {
        Self { attempts: 0 }
    }

    fn reset(&mut self) {
        self.attempts = 0;
    }

    fn next_delay(&mut self) -> Duration {
        let secs = 2_u64.saturating_pow(self.attempts).min(30);
        self.attempts = self.attempts.saturating_add(1);
        Duration::from_secs(secs).min(Self::MAX_DELAY)
    }
}

// ---------------------------------------------------------------------------
// Nanosecond timestamp → HH:MM:SS helper
// ---------------------------------------------------------------------------

fn nanos_to_hms(nanos: i64) -> String {
    if nanos <= 0 {
        return "--:--:--".to_owned();
    }
    let secs = nanos / 1_000_000_000;
    format!(
        "{:02}:{:02}:{:02}",
        (secs / 3600) % 24,
        (secs / 60) % 60,
        secs % 60
    )
}

// ---------------------------------------------------------------------------
// Proto → state conversions
// ---------------------------------------------------------------------------

/// Deduplicate training sessions by model name, keeping the entry with the
/// highest epoch. The API gateway can report multiple metric rows for the same
/// model (e.g. a stale epoch-0 hyperopt entry alongside the live session).
/// When two entries share a model name but have different folds, the one with
/// the higher epoch wins — fold is metadata, not a separate session.
fn dedup_sessions(raw: &[monitoring::TrainingSession]) -> Vec<TrainingSessionData> {
    use std::collections::HashMap;
    let mut best: HashMap<&str, &monitoring::TrainingSession> = HashMap::new();
    for s in raw {
        let key = s.model.as_str();
        let entry = best.entry(key).or_insert(s);
        if s.current_epoch > entry.current_epoch {
            *entry = s;
        }
    }
    let mut sessions: Vec<TrainingSessionData> =
        best.into_values().map(convert_training_session).collect();
    sessions.sort_by(|a, b| a.model.cmp(&b.model));
    sessions
}

fn convert_training_session(s: &monitoring::TrainingSession) -> TrainingSessionData {
    TrainingSessionData {
        model: s.model.clone(),
        fold: s.fold.clone(),
        is_hyperopt: s.is_hyperopt,
        epoch: s.current_epoch as u32,
        loss: f64::from(s.epoch_loss),
        val_loss: f64::from(s.validation_loss),
        batches_per_sec: f64::from(s.batches_per_second),
        iteration_secs: f64::from(s.iteration_seconds),
        epoch_duration_secs: f64::from(s.epoch_duration_seconds),
        sharpe: f64::from(s.epoch_sharpe),
        sortino: f64::from(s.epoch_sortino),
        win_rate: f64::from(s.epoch_win_rate),
        max_dd: f64::from(s.epoch_max_drawdown),
        profit_factor: f64::from(s.epoch_profit_factor),
        total_return: f64::from(s.epoch_total_return),
        total_trades: s.epoch_total_trades,
        action_buy_pct: f64::from(s.action_buy_pct),
        action_sell_pct: f64::from(s.action_sell_pct),
        action_hold_pct: f64::from(s.action_hold_pct),
        q_value_mean: f64::from(s.q_value_mean),
        q_value_max: f64::from(s.q_value_max),
        policy_entropy: f64::from(s.policy_entropy),
        kl_divergence: f64::from(s.kl_divergence),
        advantage_mean: f64::from(s.advantage_mean),
        replay_buffer_size: s.replay_buffer_size,
        gradient_norm: f64::from(s.gradient_norm),
        learning_rate: f64::from(s.learning_rate),
        nan_detected: s.nan_detected,
        gradient_explosions: s.gradient_explosions,
        feature_errors: s.feature_errors,
        checkpoint_saves: s.checkpoint_saves,
        checkpoint_failures: s.checkpoint_failures,
        hyperopt_trial_current: s.hyperopt_trial_current,
        hyperopt_trial_total: s.hyperopt_trial_total,
        hyperopt_best_objective: f64::from(s.hyperopt_best_objective),
        hyperopt_trials_failed: s.hyperopt_trials_failed,
        hyperopt_trial_epoch: s.hyperopt_trial_epoch,
        hyperopt_elapsed_secs: f64::from(s.hyperopt_elapsed_seconds),
        epoch_history: Vec::new(),
    }
}

fn convert_gpu(g: &monitoring::GpuSnapshot) -> GpuData {
    GpuData {
        name: String::from("GPU 0"),
        utilization: f64::from(g.utilization_percent),
        memory_used_gb: f64::from(g.memory_used_mb) / 1024.0,
        memory_total_gb: f64::from(g.memory_total_mb) / 1024.0,
        temperature_c: g.temperature_celsius as u32,
        power_watts: g.power_watts as u32,
        power_limit_watts: 0, // not in proto
    }
}

fn convert_service_status(s: &monitoring::ServiceStatus) -> ServiceData {
    let healthy = s.health == monitoring::ServiceHealth::Healthy as i32;
    let uptime = if s.uptime_seconds > 86400 {
        format!("{}d {}h", s.uptime_seconds / 86400, (s.uptime_seconds % 86400) / 3600)
    } else if s.uptime_seconds > 3600 {
        format!("{}h {}m", s.uptime_seconds / 3600, (s.uptime_seconds % 3600) / 60)
    } else {
        format!("{}m", s.uptime_seconds / 60)
    };

    let latency_ms = s
        .metadata
        .get("latency_ms")
        .and_then(|v| v.parse::<f64>().ok())
        .unwrap_or(0.0);

    ServiceData {
        name: s.service_name.clone(),
        healthy,
        uptime,
        version: s.version.clone().unwrap_or_default(),
        latency_ms,
        error: s.error_message.clone(),
    }
}

fn convert_account(a: &broker_gateway::GetAccountStateResponse) -> AccountData {
    AccountData {
        equity: a.equity,
        cash: a.cash_balance,
        margin_used: a.margin_used,
        daily_pnl: a.realized_pnl,
        total_pnl: a.unrealized_pnl + a.realized_pnl,
    }
}

fn convert_position(p: &trading::Position) -> PositionData {
    let qty = p.quantity as i64;
    let market_price = if p.quantity.abs() > f64::EPSILON {
        p.market_value / p.quantity
    } else {
        0.0
    };
    PositionData {
        symbol: p.symbol.clone(),
        qty,
        entry_price: p.average_price,
        market_price,
        unrealized_pnl: p.unrealized_pnl,
    }
}

fn convert_execution(e: &broker_gateway::ExecutionEvent) -> ExecutionData {
    let side = match e.side {
        x if x == broker_gateway::OrderSide::Buy as i32 => "BUY",
        x if x == broker_gateway::OrderSide::Sell as i32 => "SELL",
        _ => "?",
    };

    let exec_type = match e.exec_type {
        x if x == broker_gateway::ExecutionType::Trade as i32 => "FILL",
        x if x == broker_gateway::ExecutionType::New as i32 => "NEW",
        x if x == broker_gateway::ExecutionType::Canceled as i32 => "CANCEL",
        x if x == broker_gateway::ExecutionType::Rejected as i32 => "REJECT",
        _ => "?",
    };

    ExecutionData {
        time: nanos_to_hms(e.transact_time),
        symbol: e.symbol.clone(),
        side: side.to_owned(),
        qty: e.last_qty as i64,
        price: e.last_price,
        status: exec_type.to_owned(),
    }
}

fn convert_order_event(evt: &trading::OrderEvent) -> OrderEventData {
    let order = evt.order.as_ref();
    let side = order
        .map(|o| match o.side {
            x if x == trading::OrderSide::Buy as i32 => "BUY",
            x if x == trading::OrderSide::Sell as i32 => "SELL",
            _ => "?",
        })
        .unwrap_or("?");

    let status = order
        .map(|o| match o.status {
            x if x == trading::OrderStatus::Pending as i32 => "PENDING",
            x if x == trading::OrderStatus::Submitted as i32 => "SUBMITTED",
            x if x == trading::OrderStatus::PartiallyFilled as i32 => "PARTIAL",
            x if x == trading::OrderStatus::Filled as i32 => "FILLED",
            x if x == trading::OrderStatus::Cancelled as i32 => "CANCELLED",
            x if x == trading::OrderStatus::Rejected as i32 => "REJECTED",
            _ => "?",
        })
        .unwrap_or("?");

    let symbol = order.map_or_else(String::new, |o| o.symbol.clone());
    let qty = order.map_or(0.0, |o| o.quantity);
    let filled_qty = order.map_or(0.0, |o| o.filled_quantity);

    OrderEventData {
        time: nanos_to_hms(evt.timestamp),
        order_id: evt.order_id.clone(),
        symbol,
        side: side.to_owned(),
        status: status.to_owned(),
        qty,
        filled_qty,
    }
}

fn convert_risk_metrics(m: &risk::RiskMetrics) -> RiskData {
    RiskData {
        global_kill_switch: None,
        portfolio_kill_switch: None,
        strategy_kill_switch: None,
        instrument_kill_switch: None,
        current_drawdown_pct: m.current_drawdown * 100.0,
        max_drawdown_pct: m.max_drawdown * 100.0,
        high_water_mark: 0.0, // not in proto
        circuit_breakers: Vec::new(),
        position_limits: Vec::new(),
    }
}

fn convert_circuit_breaker(cb: &risk::CircuitBreakerStatus) -> CircuitBreakerData {
    CircuitBreakerData {
        name: cb.name.clone(),
        threshold: None, // not in proto
        current: None,
        tripped: cb.is_triggered,
    }
}

fn convert_risk_alert(evt: &risk::RiskAlertEvent) -> RiskAlertData {
    let severity = match evt.severity {
        x if x == risk::RiskAlertSeverity::Info as i32 => "INFO",
        x if x == risk::RiskAlertSeverity::Warning as i32 => "WARN",
        x if x == risk::RiskAlertSeverity::Critical as i32 => "CRIT",
        x if x == risk::RiskAlertSeverity::Emergency as i32 => "EMRG",
        _ => "?",
    };

    let alert_type = match evt.alert_type {
        x if x == risk::RiskAlertType::VarBreach as i32 => "VaR",
        x if x == risk::RiskAlertType::PositionLimit as i32 => "PosLimit",
        x if x == risk::RiskAlertType::Drawdown as i32 => "Drawdown",
        x if x == risk::RiskAlertType::Concentration as i32 => "Conc",
        x if x == risk::RiskAlertType::Liquidity as i32 => "Liq",
        x if x == risk::RiskAlertType::Correlation as i32 => "Corr",
        _ => "Risk",
    };

    RiskAlertData {
        time: nanos_to_hms(evt.timestamp),
        severity: severity.to_owned(),
        alert_type: alert_type.to_owned(),
        message: evt.message.clone(),
    }
}

fn convert_model_status(s: &ml::ModelStatus) -> ModelStatusData {
    let state = match s.state {
        x if x == ml::ModelState::Loading as i32 => "LOADING",
        x if x == ml::ModelState::Ready as i32 => "READY",
        x if x == ml::ModelState::Predicting as i32 => "ACTIVE",
        x if x == ml::ModelState::Training as i32 => "TRAINING",
        x if x == ml::ModelState::Error as i32 => "ERROR",
        x if x == ml::ModelState::Offline as i32 => "OFFLINE",
        _ => "?",
    };

    let health = match s.health {
        x if x == ml::ModelHealth::Healthy as i32 => "OK",
        x if x == ml::ModelHealth::Degraded as i32 => "DEGRADED",
        x if x == ml::ModelHealth::Unhealthy as i32 => "UNHEALTHY",
        x if x == ml::ModelHealth::Critical as i32 => "CRITICAL",
        _ => "?",
    };

    let last_pred = if s.last_prediction > 0 {
        nanos_to_hms(s.last_prediction)
    } else {
        "never".to_owned()
    };

    ModelStatusData {
        model_name: s.model_name.clone(),
        state: state.to_owned(),
        health: health.to_owned(),
        last_prediction: last_pred,
    }
}

fn convert_alert(a: &monitoring::Alert) -> AlertData {
    let severity = match a.severity {
        x if x == monitoring::AlertSeverity::Info as i32 => "INFO",
        x if x == monitoring::AlertSeverity::Warning as i32 => "WARN",
        x if x == monitoring::AlertSeverity::Critical as i32 => "CRIT",
        x if x == monitoring::AlertSeverity::Emergency as i32 => "EMRG",
        _ => "?",
    };

    AlertData {
        time: nanos_to_hms(a.triggered_at),
        severity: severity.to_owned(),
        service: a.service_name.clone(),
        title: a.title.clone(),
    }
}

fn convert_download_job(j: &data_acquisition::DownloadJobSummary) -> DataFeedData {
    let status = match j.status {
        x if x == data_acquisition::DownloadStatus::Completed as i32 => "OK",
        x if x == data_acquisition::DownloadStatus::Downloading as i32 => "ACTIVE",
        x if x == data_acquisition::DownloadStatus::Pending as i32 => "PENDING",
        x if x == data_acquisition::DownloadStatus::Failed as i32 => "ERROR",
        _ => "UNKNOWN",
    };

    DataFeedData {
        symbol: j.symbols.join(","),
        records_per_sec: 0, // Not available in proto DownloadJobSummary
        latency_us: 0,     // Not available in proto DownloadJobSummary
        status: status.to_owned(),
    }
}

/// Group a flat list of [`PodInfo`] into [`PodGroupData`] buckets keyed by
/// service name. Pods with an empty service label are bucketed under `(other)`.
fn group_pods(pods: Vec<monitoring::PodInfo>) -> Vec<PodGroupData> {
    use std::collections::BTreeMap;

    let mut groups: BTreeMap<String, Vec<PodData>> = BTreeMap::new();
    for p in pods {
        let svc = if p.service.is_empty() {
            "(other)".into()
        } else {
            p.service.clone()
        };
        groups.entry(svc).or_default().push(PodData {
            name: p.name,
            service: p.service,
            status: p.status,
            restarts: p.restarts,
            age: p.age,
            node: p.node,
            ready: p.ready,
            version: p.version,
        });
    }

    groups
        .into_iter()
        .map(|(service, group_pods)| {
            let ready_count = group_pods.iter().filter(|p| p.ready).count();
            let total_count = group_pods.len();
            PodGroupData {
                service,
                pods: group_pods,
                ready_count,
                total_count,
            }
        })
        .collect()
}

/// Build aggregate [`DataCacheData`] from download job summaries.
///
/// Uses completed job count as a proxy for total records (no per-record
/// count in the proto) and derives the date range from job `start_date` /
/// `end_date` strings.
fn aggregate_cache_stats(jobs: &[data_acquisition::DownloadJobSummary]) -> DataCacheData {
    let completed_jobs: Vec<_> = jobs
        .iter()
        .filter(|j| j.status == data_acquisition::DownloadStatus::Completed as i32)
        .collect();

    // completed_jobs.len() is the number of download *jobs*, not market data
    // records. The proto doesn't expose a per-job record count, so show 0 and
    // let the cockpit render "--" for the field.
    let total_records = 0_u64;

    // Derive date range from the string fields across all completed jobs.
    let mut oldest: Option<&str> = None;
    let mut newest: Option<&str> = None;
    for j in &completed_jobs {
        if !j.start_date.is_empty() {
            oldest = Some(match oldest {
                Some(cur) if cur <= j.start_date.as_str() => cur,
                _ => j.start_date.as_str(),
            });
        }
        if !j.end_date.is_empty() {
            newest = Some(match newest {
                Some(cur) if cur >= j.end_date.as_str() => cur,
                _ => j.end_date.as_str(),
            });
        }
    }

    DataCacheData {
        total_records,
        cache_hit_rate: 0.0, // not available from download status
        disk_usage_gb: 0.0,  // not available from download status
        oldest_date: oldest.unwrap_or("--").to_owned(),
        newest_date: newest.unwrap_or("--").to_owned(),
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::shadow_unrelated, clippy::shadow_reuse, clippy::assertions_on_result_states, clippy::str_to_string, clippy::field_reassign_with_default, clippy::let_underscore_must_use, clippy::indexing_slicing, clippy::string_add, clippy::string_add_assign)]
mod tests {
    use super::*;

    fn make_pod(name: &str, service: &str, ready: bool) -> monitoring::PodInfo {
        monitoring::PodInfo {
            name: name.into(),
            service: service.into(),
            namespace: "foxhunt".into(),
            status: if ready { "Running".into() } else { "Pending".into() },
            restarts: 0,
            age: "1h".into(),
            node: "node-1".into(),
            ready,
            version: "v1".into(),
        }
    }

    #[test]
    fn group_pods_groups_by_service() {
        let pods = vec![
            make_pod("api-1", "api", true),
            make_pod("api-2", "api", true),
            make_pod("trade-1", "trading-service", true),
            make_pod("trade-2", "trading-service", false),
        ];
        let groups = group_pods(pods);
        assert_eq!(groups.len(), 2);
        assert_eq!(groups[0].service, "api");
        assert_eq!(groups[0].total_count, 2);
        assert_eq!(groups[0].ready_count, 2);
        assert_eq!(groups[1].service, "trading-service");
        assert_eq!(groups[1].total_count, 2);
        assert_eq!(groups[1].ready_count, 1);
    }

    #[test]
    fn group_pods_empty_service_becomes_other() {
        let pods = vec![
            make_pod("orphan-1", "", true),
            make_pod("api-1", "api", true),
        ];
        let groups = group_pods(pods);
        assert_eq!(groups.len(), 2);
        // BTreeMap sorts alphabetically: "(other)" < "api"
        assert_eq!(groups[0].service, "(other)");
        assert_eq!(groups[0].total_count, 1);
        assert_eq!(groups[1].service, "api");
    }

    #[test]
    fn group_pods_empty_input() {
        let groups = group_pods(vec![]);
        assert!(groups.is_empty());
    }

    #[test]
    fn group_pods_ready_count_accuracy() {
        let pods = vec![
            make_pod("a", "svc", true),
            make_pod("b", "svc", false),
            make_pod("c", "svc", true),
            make_pod("d", "svc", false),
            make_pod("e", "svc", true),
        ];
        let groups = group_pods(pods);
        assert_eq!(groups.len(), 1);
        assert_eq!(groups[0].ready_count, 3);
        assert_eq!(groups[0].total_count, 5);
    }

    #[test]
    fn group_pods_sorted_by_service_name() {
        let pods = vec![
            make_pod("z", "zebra", true),
            make_pod("a", "alpha", true),
            make_pod("m", "middle", true),
        ];
        let groups = group_pods(pods);
        let names: Vec<_> = groups.iter().map(|g| g.service.as_str()).collect();
        assert_eq!(names, vec!["alpha", "middle", "zebra"]);
    }

    fn make_training_session(model: &str, epoch: f32) -> monitoring::TrainingSession {
        monitoring::TrainingSession {
            model: model.into(),
            fold: "fold-0".into(),
            current_epoch: epoch,
            epoch_loss: 0.5,
            validation_loss: 0.6,
            batches_per_second: 100.0,
            iteration_seconds: 0.01,
            epoch_duration_seconds: 30.0,
            epoch_sharpe: 1.2,
            epoch_sortino: 1.5,
            epoch_win_rate: 0.55,
            epoch_max_drawdown: 0.1,
            epoch_profit_factor: 1.8,
            epoch_total_return: 0.05,
            epoch_total_trades: 100,
            action_buy_pct: 0.33,
            action_sell_pct: 0.33,
            action_hold_pct: 0.34,
            q_value_mean: 0.5,
            q_value_max: 1.0,
            policy_entropy: 0.8,
            kl_divergence: 0.01,
            advantage_mean: 0.1,
            replay_buffer_size: 10000,
            gradient_norm: 0.5,
            learning_rate: 0.001,
            checkpoint_saves: 1,
            ..Default::default()
        }
    }

    #[test]
    fn convert_training_session_maps_fields() {
        let proto = make_training_session("DQN", 5.0);
        let data = convert_training_session(&proto);
        assert_eq!(data.model, "DQN");
        assert_eq!(data.epoch, 5);
        assert!((data.loss - 0.5).abs() < 1e-6);
        assert!((data.sharpe - 1.2).abs() < 1e-4);
        assert_eq!(data.total_trades, 100);
        assert!(data.epoch_history.is_empty());
    }

    #[test]
    fn convert_training_session_f32_to_f64() {
        let proto = make_training_session("PPO", 1.0);
        let data = convert_training_session(&proto);
        // f32→f64 should preserve value within f32 precision
        assert!((data.val_loss - 0.6).abs() < 1e-4);
        assert!((data.batches_per_sec - 100.0).abs() < 1e-4);
        assert!((data.learning_rate - 0.001).abs() < 1e-6);
    }

    #[test]
    fn convert_gpu_snapshot() {
        let snap = monitoring::GpuSnapshot {
            utilization_percent: 85.0,
            memory_used_mb: 4096.0,
            memory_total_mb: 8192.0,
            temperature_celsius: 72.0,
            power_watts: 150.0,
        };
        let data = convert_gpu(&snap);
        assert!((data.utilization - 85.0).abs() < 1e-6);
        assert!((data.memory_used_gb - 4.0).abs() < 1e-6);
        assert!((data.memory_total_gb - 8.0).abs() < 1e-6);
        assert_eq!(data.temperature_c, 72);
        assert_eq!(data.power_watts, 150);
    }

    fn make_download_job(status: i32, start: &str, end: &str) -> data_acquisition::DownloadJobSummary {
        data_acquisition::DownloadJobSummary {
            job_id: "j1".into(),
            status,
            dataset: "ohlcv-1m".into(),
            symbols: vec!["ES.FUT".into()],
            start_date: start.into(),
            end_date: end.into(),
            progress_percentage: 100.0,
            actual_cost_usd: 0.0,
            created_at: 0,
            completed_at: 0,
        }
    }

    #[test]
    fn aggregate_cache_stats_date_range() {
        let completed = data_acquisition::DownloadStatus::Completed as i32;
        let jobs = vec![
            make_download_job(completed, "2025-01-01", "2025-03-31"),
            make_download_job(completed, "2024-07-01", "2024-09-30"),
            make_download_job(completed, "2025-04-01", "2025-06-30"),
        ];
        let stats = aggregate_cache_stats(&jobs);
        assert_eq!(stats.oldest_date, "2024-07-01");
        assert_eq!(stats.newest_date, "2025-06-30");
    }

    #[test]
    fn aggregate_cache_stats_empty_jobs() {
        let stats = aggregate_cache_stats(&[]);
        assert_eq!(stats.oldest_date, "--");
        assert_eq!(stats.newest_date, "--");
        assert_eq!(stats.total_records, 0);
    }

    #[test]
    fn aggregate_cache_stats_skips_non_completed() {
        let pending = data_acquisition::DownloadStatus::Pending as i32;
        let completed = data_acquisition::DownloadStatus::Completed as i32;
        let jobs = vec![
            make_download_job(pending, "2020-01-01", "2020-12-31"),
            make_download_job(completed, "2025-01-01", "2025-03-31"),
        ];
        let stats = aggregate_cache_stats(&jobs);
        // Pending job dates should be excluded
        assert_eq!(stats.oldest_date, "2025-01-01");
        assert_eq!(stats.newest_date, "2025-03-31");
    }
}
