feat(fxt): add spawn_pods_stream with service grouping

Adds a new gRPC stream spawner that subscribes to SubscribeClusterPods
and groups pods by service name for the pods cockpit view. Follows the
same backoff/reconnect pattern used by all other stream spawners.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

bin/fxt/src/tui/data_fetcher.rs

@@ -25,7 +25,7 @@ use crate::proto::{broker_gateway, data_acquisition, ml, monitoring, risk, tradi
 
 use super::state::{
     AccountData, AlertData, CircuitBreakerData, ConnectionStatus, DataCacheData, DataFeedData,
-    ExecutionData, GpuData, ModelStatusData, OrderEventData, PodGroupData, PositionData,
+    ExecutionData, GpuData, ModelStatusData, OrderEventData, PodData, PodGroupData, PositionData,
     RiskAlertData, RiskData, ServiceData, SystemResources, TrainingSessionData,
 };
 
@@ -128,6 +128,7 @@ impl DataFetcher {
         self.spawn_risk_alerts(client);
         self.spawn_model_status(client);
         self.spawn_alerts(client);
+        self.spawn_pods_stream(client);
     }
 
     // ── Stream spawners ─────────────────────────────────────────────
@@ -823,6 +824,69 @@ impl DataFetcher {
             }
         });
     }
+
+    /// 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) => {
+                                            debug!("ClusterPods stream ended");
+                                            break;
+                                        }
+                                        Err(e) => {
+                                            warn!("ClusterPods stream error: {e}");
+                                            break;
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        debug!("ClusterPods connect failed: {e}");
+                        let _ = tx.send(StateUpdate::Connection(ConnectionStatus::Reconnecting {
+                            attempt: backoff.attempts,
+                        }));
+                    }
+                }
+
+                if backoff.attempts >= MAX_RECONNECT_ATTEMPTS {
+                    warn!("ClusterPods exceeded {MAX_RECONNECT_ATTEMPTS} reconnect attempts, giving up");
+                    return;
+                }
+                tokio::select! {
+                    _ = cancel.cancelled() => return,
+                    _ = tokio::time::sleep(backoff.next_delay()) => {}
+                }
+            }
+        });
+    }
 }
 
 // ---------------------------------------------------------------------------
@@ -1130,6 +1194,45 @@ fn convert_download_job(j: &data_acquisition::DownloadJobSummary) -> DataFeedDat
     }
 }
 
+/// 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, pods)| {
+            let ready_count = pods.iter().filter(|p| p.ready).count();
+            let total_count = pods.len();
+            PodGroupData {
+                service,
+                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