# Monitoring Service Implementation Plan

> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

**Goal:** Build a monitoring_service that bridges live Prometheus training metrics to gRPC, enabling `fxt train monitor` and the web dashboard to see real-time training progress for any running job (CI or service-managed).

**Architecture:** A new standalone Rust service queries the in-cluster Prometheus HTTP API for `foxhunt_training_*`, `foxhunt_hyperopt_*`, `dcgm_*`, and `kube_job_*` metrics, groups them by `{model, fold}`, and serves them over gRPC (unary + server-streaming). The fxt CLI and web-gateway connect as gRPC clients only — no direct Prometheus dependency.

**Tech Stack:** Rust, tonic (gRPC), reqwest (Prometheus HTTP client), axum (metrics endpoint), serde_json, clap

---

## Task 1: Add hyperopt Prometheus metrics to common

**Files:**
- Modify: `crates/common/src/metrics/training_metrics.rs`

**Step 1: Add 6 hyperopt metric registrations to `init()`**

Append inside `init()` after the `register_gauge("foxhunt_training_active_workers", ...)` line:

```rust
    // Hyperopt gauges (model only)
    _ = register_gauge_vec(
        "foxhunt_hyperopt_trial_current",
        "Current trial number (1-indexed)",
        m,
    );
    _ = register_gauge_vec(
        "foxhunt_hyperopt_trial_total",
        "Total trials configured",
        m,
    );
    _ = register_gauge_vec(
        "foxhunt_hyperopt_best_objective",
        "Best objective value so far (Sharpe)",
        m,
    );
    _ = register_gauge_vec(
        "foxhunt_hyperopt_mode",
        "1=hyperopt active, 0=training only",
        m,
    );

    // Hyperopt counters (model only)
    _ = register_counter_vec(
        "foxhunt_hyperopt_trials_failed_total",
        "Failed trials count",
        m,
    );

    // Hyperopt histogram (model only)
    _ = register_histogram_vec(
        "foxhunt_hyperopt_trial_duration_seconds",
        "Duration of each trial",
        m,
        vec![5.0, 15.0, 30.0, 60.0, 120.0, 300.0, 600.0],
    );
```

**Step 2: Add typed helper functions**

Append after `set_active_workers()`:

```rust
pub fn set_hyperopt_trial(model: &str, current: f64, total: f64) {
    set_gauge_vec("foxhunt_hyperopt_trial_current", &[model], current);
    set_gauge_vec("foxhunt_hyperopt_trial_total", &[model], total);
}

pub fn set_hyperopt_best_objective(model: &str, value: f64) {
    set_gauge_vec("foxhunt_hyperopt_best_objective", &[model], value);
}

pub fn record_hyperopt_trial_duration(model: &str, secs: f64) {
    observe_histogram_vec("foxhunt_hyperopt_trial_duration_seconds", &[model], secs);
}

pub fn record_hyperopt_trial_failure(model: &str) {
    increment_counter_vec("foxhunt_hyperopt_trials_failed_total", &[model], 1.0);
}

pub fn set_hyperopt_mode(model: &str, active: bool) {
    set_gauge_vec(
        "foxhunt_hyperopt_mode",
        &[model],
        if active { 1.0 } else { 0.0 },
    );
}
```

**Step 3: Add unit tests**

Append to the existing test module (or add a new `#[cfg(test)] mod tests` block if none exists — there isn't one currently in this file):

```rust
#[cfg(test)]
mod tests {
    use super::*;
    use crate::metrics::{get_gauge_value, register_gauge};

    #[test]
    fn test_init_is_idempotent() {
        init();
        init(); // calling twice must not panic
    }

    #[test]
    fn test_set_hyperopt_trial() {
        init();
        set_hyperopt_trial("dqn", 5.0, 20.0);
        // No panic = success (gauge_vec values not directly readable without label matching)
    }

    #[test]
    fn test_set_hyperopt_mode() {
        init();
        set_hyperopt_mode("dqn", true);
        set_hyperopt_mode("dqn", false);
    }
}
```

**Step 4: Run tests**

Run: `SQLX_OFFLINE=true cargo test -p common --lib -- metrics::training_metrics`
Expected: PASS

**Step 5: Build check**

Run: `SQLX_OFFLINE=true cargo check -p common`
Expected: 0 errors, 0 warnings

**Step 6: Commit**

```bash
git add crates/common/src/metrics/training_metrics.rs
git commit -m "feat(metrics): add 6 hyperopt Prometheus metrics to training_metrics"
```

---

## Task 2: Wire hyperopt metrics into training binaries

**Files:**
- Modify: `crates/ml/examples/hyperopt_baseline_rl.rs`
- Modify: `crates/ml/examples/hyperopt_baseline_supervised.rs`

**Step 1: Wire hyperopt metrics into `hyperopt_baseline_rl.rs`**

After `training_metrics::set_active_workers(1.0);` (line ~276), add:

```rust
training_metrics::set_hyperopt_mode(&args.model, true);
```

In `run_dqn_hyperopt()`, after `let start = Instant::now();` (line ~169), add before optimizer call:

```rust
training_metrics::set_hyperopt_trial("dqn", 0.0, args.trials as f64);
```

After the optimizer completes and `result` is available (line ~182), add:

```rust
training_metrics::set_hyperopt_trial("dqn", result.all_trials.len() as f64, args.trials as f64);
training_metrics::set_hyperopt_best_objective("dqn", result.best_objective);
for trial in &result.all_trials {
    training_metrics::record_hyperopt_trial_duration("dqn", trial.elapsed_secs);
    if !trial.success {
        training_metrics::record_hyperopt_trial_failure("dqn");
    }
}
```

Apply the same pattern to `run_ppo_hyperopt()` with `"ppo"` label.

At the end of `main()`, before returning, add:

```rust
training_metrics::set_hyperopt_mode(&args.model, false);
```

**Step 2: Wire hyperopt metrics into `hyperopt_baseline_supervised.rs`**

Same pattern: `set_hyperopt_mode` at start/end, `set_hyperopt_trial` before/after optimize, `set_hyperopt_best_objective` and `record_hyperopt_trial_duration` after each trial. Use the model name from args (e.g. `"tft"`, `"mamba2"`, etc.).

**Step 3: Build check**

Run: `SQLX_OFFLINE=true cargo check -p ml --example hyperopt_baseline_rl --example hyperopt_baseline_supervised`
Expected: 0 errors

**Step 4: Commit**

```bash
git add crates/ml/examples/hyperopt_baseline_rl.rs crates/ml/examples/hyperopt_baseline_supervised.rs
git commit -m "feat(ml): wire hyperopt Prometheus metrics into training binaries"
```

---

## Task 3: Create monitoring.proto

**Files:**
- Create: `services/monitoring_service/proto/monitoring.proto`

**Step 1: Write the proto file**

```protobuf
syntax = "proto3";
package monitoring;

service MonitoringService {
  // Single snapshot of all active training sessions
  rpc GetLiveTrainingMetrics(GetLiveTrainingMetricsRequest)
      returns (GetLiveTrainingMetricsResponse);

  // Server-streaming: pushes updates every N seconds
  rpc StreamTrainingMetrics(StreamTrainingMetricsRequest)
      returns (stream GetLiveTrainingMetricsResponse);
}

message GetLiveTrainingMetricsRequest {
  string model_filter = 1;  // optional: "dqn", "ppo", etc.
}

message StreamTrainingMetricsRequest {
  string model_filter = 1;
  uint32 interval_seconds = 2;  // 0 = server default (3s)
}

message GetLiveTrainingMetricsResponse {
  repeated TrainingSession sessions = 1;
  GpuSnapshot gpu = 2;
  uint32 active_k8s_jobs = 3;
  int64 timestamp = 4;
}

message TrainingSession {
  string model = 1;
  string fold = 2;
  bool is_hyperopt = 3;

  // Epoch/progress
  float current_epoch = 4;
  float epoch_loss = 5;
  float validation_loss = 6;

  // Throughput
  float batches_per_second = 7;
  float batches_processed = 8;
  float iteration_seconds = 9;

  // Eval metrics
  float eval_accuracy = 10;
  float eval_precision = 11;
  float eval_recall = 12;
  float eval_f1 = 13;

  // Checkpoint
  float checkpoint_size_bytes = 14;
  uint32 checkpoint_saves = 15;
  uint32 checkpoint_failures = 16;

  // Health counters
  uint32 nan_detected = 17;
  uint32 gradient_explosions = 18;
  uint32 feature_errors = 19;

  // Hyperopt fields (populated when is_hyperopt=true)
  uint32 hyperopt_trial_current = 20;
  uint32 hyperopt_trial_total = 21;
  float hyperopt_best_objective = 22;
  uint32 hyperopt_trials_failed = 23;
}

message GpuSnapshot {
  float utilization_percent = 1;
  float memory_used_mb = 2;
  float memory_total_mb = 3;
  float temperature_celsius = 4;
  float power_watts = 5;
}
```

**Step 2: Commit**

```bash
mkdir -p services/monitoring_service/proto
git add services/monitoring_service/proto/monitoring.proto
git commit -m "feat(monitoring): add monitoring.proto with gRPC service definition"
```

---

## Task 4: Create monitoring_service crate scaffolding

**Files:**
- Create: `services/monitoring_service/Cargo.toml`
- Create: `services/monitoring_service/build.rs`
- Create: `services/monitoring_service/src/main.rs`
- Create: `services/monitoring_service/src/prometheus_client.rs`
- Create: `services/monitoring_service/src/service.rs`
- Modify: `Cargo.toml` (workspace members)

**Step 1: Write Cargo.toml**

```toml
[package]
name = "monitoring_service"
version.workspace = true
edition.workspace = true
rust-version.workspace = true
authors.workspace = true
license.workspace = true
description = "Monitoring Service - Prometheus-to-gRPC bridge for live training metrics"

[dependencies]
tokio.workspace = true
tonic.workspace = true
tonic-prost.workspace = true
prost.workspace = true
prost-types.workspace = true
serde.workspace = true
serde_json.workspace = true
anyhow.workspace = true
clap.workspace = true
tracing.workspace = true
tracing-subscriber.workspace = true
reqwest = { version = "0.12", features = ["rustls-tls", "json"], default-features = false }
axum.workspace = true
prometheus.workspace = true
tokio-stream.workspace = true
common = { workspace = true }

[build-dependencies]
tonic-prost-build.workspace = true
prost-build.workspace = true

[[bin]]
name = "monitoring_service"
path = "src/main.rs"
```

**Step 2: Write build.rs**

```rust
fn main() -> Result<(), Box<dyn std::error::Error>> {
    tonic_prost_build::configure()
        .build_server(true)
        .build_client(false)
        .compile_protos(&["proto/monitoring.proto"], &["proto"])?;

    println!("cargo:rerun-if-changed=proto/monitoring.proto");

    Ok(())
}
```

**Step 3: Add to workspace members**

In root `Cargo.toml`, add `"services/monitoring_service"` to the members list after `"services/api_gateway"`.

**Step 4: Write `src/prometheus_client.rs`**

HTTP client that queries the Prometheus query API and parses the JSON response into typed structs:

```rust
use anyhow::{Context, Result};
use serde::Deserialize;
use std::collections::HashMap;
use tracing::warn;

/// 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,
}

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

impl PrometheusClient {
    pub fn new(base_url: &str) -> Self {
        let http = reqwest::Client::builder()
            .timeout(std::time::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
    pub async fn fetch_training_metrics(&self) -> Result<Vec<MetricSample>> {
        let results = self
            .query(r#"{__name__=~"foxhunt_training_.*|foxhunt_hyperopt_.*"}"#)
            .await?;
        Ok(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())
    }

    /// Fetch GPU metrics from DCGM exporter
    pub 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(results
            .into_iter()
            .filter_map(|r| {
                let value: f64 = r.value.1.parse().ok()?;
                Some(MetricSample {
                    name: r.metric.get("__name__")?.clone(),
                    model: String::new(),
                    fold: String::new(),
                    value,
                })
            })
            .collect())
    }

    /// Fetch active K8s training job count
    pub async fn fetch_active_jobs(&self) -> Result<u32> {
        let results = self
            .query(r#"kube_job_status_active{namespace="foxhunt",job_name=~"training-.*"}"#)
            .await?;
        let count: f64 = results
            .iter()
            .filter_map(|r| r.value.1.parse::<f64>().ok())
            .sum();
        Ok(count as u32)
    }
}
```

**Step 5: Write `src/service.rs`**

gRPC service implementation that uses `PrometheusClient` to build responses:

```rust
use std::collections::HashMap;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;

use tokio::sync::watch;
use tokio_stream::wrappers::WatchStream;
use tokio_stream::Stream;
use tonic::{Request, Response, Status};
use tracing::{error, info};

use crate::monitoring::{
    monitoring_service_server::MonitoringService, GetLiveTrainingMetricsRequest,
    GetLiveTrainingMetricsResponse, GpuSnapshot, StreamTrainingMetricsRequest, TrainingSession,
};
use crate::prometheus_client::{MetricSample, PrometheusClient};

pub struct MonitoringServiceImpl {
    prom: Arc<PrometheusClient>,
    default_interval: u32,
}

impl MonitoringServiceImpl {
    pub fn new(prom: PrometheusClient, default_interval: u32) -> Self {
        Self {
            prom: Arc::new(prom),
            default_interval,
        }
    }

    async fn build_response(
        prom: &PrometheusClient,
        model_filter: &str,
    ) -> Result<GetLiveTrainingMetricsResponse, Status> {
        let (training, gpu, jobs) = tokio::try_join!(
            prom.fetch_training_metrics(),
            prom.fetch_gpu_metrics(),
            prom.fetch_active_jobs(),
        )
        .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);

        Ok(GetLiveTrainingMetricsResponse {
            sessions,
            gpu: Some(gpu_snapshot),
            active_k8s_jobs: jobs,
            timestamp: chrono::Utc::now().timestamp(),
        })
    }
}

#[tonic::async_trait]
impl MonitoringService for MonitoringServiceImpl {
    async fn get_live_training_metrics(
        &self,
        request: Request<GetLiveTrainingMetricsRequest>,
    ) -> Result<Response<GetLiveTrainingMetricsResponse>, Status> {
        let filter = &request.into_inner().model_filter;
        let resp = Self::build_response(&self.prom, filter).await?;
        Ok(Response::new(resp))
    }

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

    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.max(1).min(60)
        };
        let filter = req.model_filter;
        let prom = self.prom.clone();

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

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

/// 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 {
        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()
        });

        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;
                }
            }
            _ => {}
        }
    }

    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();
    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" => snap.memory_total_mb += s.value as f32, // will add used below
            "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;
    snap
}
```

**Step 6: Write `src/main.rs`**

```rust
#![deny(clippy::unwrap_used, clippy::expect_used, clippy::panic, clippy::indexing_slicing)]

mod prometheus_client;
mod service;

pub mod monitoring {
    #![allow(
        clippy::all,
        clippy::pedantic,
        clippy::restriction,
        clippy::nursery
    )]
    tonic::include_proto!("monitoring");
}

use anyhow::Result;
use clap::Parser;
use tracing::info;

use monitoring::monitoring_service_server::MonitoringServiceServer;
use prometheus_client::PrometheusClient;
use service::MonitoringServiceImpl;

#[derive(Parser, Debug)]
#[command(name = "monitoring-service")]
#[command(about = "Prometheus-to-gRPC bridge for live training metrics")]
struct Args {
    /// Prometheus base URL
    #[arg(long, env = "PROMETHEUS_URL", default_value = "http://gitlab-prometheus-server.foxhunt.svc:80")]
    prometheus_url: String,

    /// gRPC listen port
    #[arg(long, env = "GRPC_PORT", default_value = "50057")]
    grpc_port: u16,

    /// Prometheus metrics port (self-metrics)
    #[arg(long, env = "METRICS_PORT", default_value = "9099")]
    metrics_port: u16,

    /// Default streaming interval in seconds
    #[arg(long, env = "STREAM_INTERVAL_SECS", default_value = "3")]
    stream_interval: u32,
}

#[tokio::main]
async fn main() -> Result<()> {
    let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok();
    if let Err(e) =
        common::observability::init_observability("monitoring_service", otlp_endpoint.as_deref())
    {
        eprintln!("Observability init failed (non-fatal): {e}");
    }

    let args = Args::parse();

    info!("Starting monitoring_service");
    info!("  Prometheus: {}", args.prometheus_url);
    info!("  gRPC port: {}", args.grpc_port);
    info!("  Metrics port: {}", args.metrics_port);
    info!("  Stream interval: {}s", args.stream_interval);

    // Start self-metrics HTTP endpoint
    common::metrics::server::start_metrics_server(args.metrics_port);

    let prom = PrometheusClient::new(&args.prometheus_url);
    let svc = MonitoringServiceImpl::new(prom, args.stream_interval);

    let addr = format!("0.0.0.0:{}", args.grpc_port)
        .parse()
        .map_err(|e| anyhow::anyhow!("Invalid listen address: {e}"))?;

    info!("gRPC server listening on {}", addr);

    tonic::transport::Server::builder()
        .add_service(MonitoringServiceServer::new(svc))
        .serve(addr)
        .await?;

    Ok(())
}
```

**Step 7: Build check**

Run: `SQLX_OFFLINE=true cargo check -p monitoring_service`
Expected: 0 errors

**Step 8: Commit**

```bash
git add services/monitoring_service/ Cargo.toml
git commit -m "feat(monitoring): add monitoring_service crate with Prometheus-to-gRPC bridge"
```

---

## Task 5: Add `fxt train monitor` CLI subcommand

**Files:**
- Create: `bin/fxt/src/commands/train/monitor.rs`
- Modify: `bin/fxt/src/commands/train/mod.rs`
- Modify: `bin/fxt/proto/` (add monitoring.proto or copy it)
- Modify: `bin/fxt/build.rs` (compile monitoring.proto)
- Modify: `bin/fxt/src/proto.rs` or wherever proto modules are declared

**Step 1: Copy proto for fxt**

The fxt crate compiles protos from `bin/fxt/proto/`. Copy or symlink `monitoring.proto`:

```bash
cp services/monitoring_service/proto/monitoring.proto bin/fxt/proto/monitoring.proto
```

**Step 2: Update fxt build.rs**

Add `"proto/monitoring.proto"` to the `compile_protos` list and add a `cargo:rerun-if-changed` line.

**Step 3: Update fxt proto module**

Add a `monitoring` module to the proto declaration block in `bin/fxt/src/` (find where `tonic::include_proto!` is used):

```rust
pub mod monitoring {
    #![allow(clippy::all, clippy::pedantic, clippy::restriction, clippy::nursery)]
    tonic::include_proto!("monitoring");
}
```

**Step 4: Write `monitor.rs`**

```rust
use anyhow::{Context, Result};
use clap::Parser;
use colored::Colorize;
use std::io::{self, Write};

use crate::proto::monitoring::{
    monitoring_service_client::MonitoringServiceClient, GetLiveTrainingMetricsRequest,
    GetLiveTrainingMetricsResponse, StreamTrainingMetricsRequest, TrainingSession,
};

#[derive(Parser, Debug)]
pub struct MonitorCommand {
    /// Print a single snapshot and exit (no live TUI)
    #[arg(long)]
    pub once: bool,

    /// Filter to a specific model (e.g. "dqn", "ppo", "tft")
    #[arg(long)]
    pub model: Option<String>,

    /// Streaming interval in seconds (default: 3)
    #[arg(long, default_value = "3")]
    pub interval: u32,
}

impl MonitorCommand {
    pub async fn run(&self, api_gateway_url: &str, _jwt_token: &str) -> Result<()> {
        // monitoring_service runs on a different port — derive URL from env or default
        let monitoring_url = std::env::var("MONITORING_SERVICE_URL")
            .unwrap_or_else(|_| api_gateway_url.replace(":50050", ":50057"));

        let mut client = MonitoringServiceClient::connect(monitoring_url)
            .await
            .context("Failed to connect to monitoring service")?;

        let model_filter = self.model.clone().unwrap_or_default();

        if self.once {
            let response = client
                .get_live_training_metrics(tonic::Request::new(GetLiveTrainingMetricsRequest {
                    model_filter,
                }))
                .await
                .context("GetLiveTrainingMetrics failed")?;
            render_snapshot(&response.into_inner());
        } else {
            let response = client
                .stream_training_metrics(tonic::Request::new(StreamTrainingMetricsRequest {
                    model_filter,
                    interval_seconds: self.interval,
                }))
                .await
                .context("StreamTrainingMetrics failed")?;

            let mut stream = response.into_inner();
            while let Some(msg) = stream.message().await? {
                render_tui(&msg);
            }
        }

        Ok(())
    }
}

fn render_snapshot(resp: &GetLiveTrainingMetricsResponse) {
    if resp.sessions.is_empty() {
        println!("No active training sessions found.");
        return;
    }

    if let Some(gpu) = &resp.gpu {
        println!(
            "GPU: {:.0}% util | {:.1}/{:.1} GB VRAM | {:.0}C | {:.0}W",
            gpu.utilization_percent,
            gpu.memory_used_mb / 1024.0,
            gpu.memory_total_mb / 1024.0,
            gpu.temperature_celsius,
            gpu.power_watts,
        );
    }
    println!("Active K8s training jobs: {}", resp.active_k8s_jobs);
    println!();

    print_session_table(&resp.sessions);
    print_hyperopt_summary(&resp.sessions);
    print_health_summary(&resp.sessions);
}

fn render_tui(resp: &GetLiveTrainingMetricsResponse) {
    // Clear screen
    print!("\x1B[2J\x1B[H");
    let _ = io::stdout().flush();

    println!("{}", "=".repeat(72).bright_black());
    println!(
        "{}",
        "                  Foxhunt Training Monitor"
            .bright_white()
            .bold()
    );
    println!("{}", "=".repeat(72).bright_black());
    println!();

    render_snapshot(resp);

    println!();
    println!(
        "{}  |  {}",
        format!("Updated {}s ago", 0).bright_black(),
        "Ctrl+C to exit".bright_black()
    );
}

fn print_session_table(sessions: &[TrainingSession]) {
    println!(
        "{:<10} {:<6} {:<7} {:<10} {:<10} {:<9} {:<10}",
        "Model".bright_cyan(),
        "Fold".bright_cyan(),
        "Epoch".bright_cyan(),
        "Loss".bright_cyan(),
        "Val Loss".bright_cyan(),
        "Batch/s".bright_cyan(),
        "Eval Acc".bright_cyan(),
    );
    println!("{}", "-".repeat(72).bright_black());

    for s in sessions {
        let acc = if s.eval_accuracy > 0.0 {
            format!("{:.1}%", s.eval_accuracy * 100.0)
        } else {
            "-".to_owned()
        };
        println!(
            "{:<10} {:<6} {:<7.0} {:<10.4} {:<10.4} {:<9.1} {:<10}",
            s.model, s.fold, s.current_epoch, s.epoch_loss, s.validation_loss,
            s.batches_per_second, acc,
        );
    }
}

fn print_hyperopt_summary(sessions: &[TrainingSession]) {
    let hyperopt: Vec<_> = sessions.iter().filter(|s| s.is_hyperopt).collect();
    if hyperopt.is_empty() {
        return;
    }
    println!();
    for s in &hyperopt {
        println!(
            "Hyperopt ({}): trial {}/{} | best Sharpe {:.2} | {} failures",
            s.model.bright_magenta(),
            s.hyperopt_trial_current,
            s.hyperopt_trial_total,
            s.hyperopt_best_objective,
            s.hyperopt_trials_failed,
        );
    }
}

fn print_health_summary(sessions: &[TrainingSession]) {
    let total_nan: u32 = sessions.iter().map(|s| s.nan_detected).sum();
    let total_grad: u32 = sessions.iter().map(|s| s.gradient_explosions).sum();
    let total_ckpt: u32 = sessions.iter().map(|s| s.checkpoint_saves).sum();
    println!(
        "Health: {} NaN | {} grad explosions | {} checkpoints",
        total_nan, total_grad, total_ckpt
    );
}
```

**Step 5: Update `train/mod.rs`**

Add `pub mod monitor;` and `pub use monitor::MonitorCommand;` to the module declarations. Add `Monitor` variant to the `TrainCommand` enum:

```rust
/// Live training metrics (Prometheus via gRPC)
Monitor {
    #[clap(flatten)]
    monitor_args: monitor::MonitorCommand,
},
```

Add the match arm in `execute_train_command`:

```rust
TrainCommand::Monitor { monitor_args } => monitor_args.run(api_gateway_url, jwt_token).await,
```

**Step 6: Build check**

Run: `SQLX_OFFLINE=true cargo check -p fxt`
Expected: 0 errors

**Step 7: Commit**

```bash
git add bin/fxt/src/commands/train/monitor.rs bin/fxt/src/commands/train/mod.rs bin/fxt/proto/monitoring.proto bin/fxt/build.rs
git commit -m "feat(fxt): add 'fxt train monitor' subcommand with live TUI"
```

---

## Task 6: Web-gateway integration

**Files:**
- Modify: `crates/web-gateway/build.rs` (add monitoring.proto)
- Modify: `crates/web-gateway/src/lib.rs` (add monitoring proto module)
- Modify: `crates/web-gateway/src/config.rs` (add monitoring_service_url)
- Modify: `crates/web-gateway/src/state.rs` (add monitoring_channel)
- Modify: `crates/web-gateway/src/grpc/clients.rs` (add monitoring_client)
- Create: `crates/web-gateway/src/routes/monitoring.rs` (REST endpoint)
- Modify: `crates/web-gateway/src/routes/mod.rs` (mount monitoring routes)
- Modify: `crates/web-gateway/src/grpc/streams.rs` (add training_progress poller)

**Step 1: Add proto compilation**

In `crates/web-gateway/build.rs`, add `"../../services/monitoring_service/proto/monitoring.proto"` to the `compile_protos` call's first argument list, and `"../../services/monitoring_service/proto"` to the include paths. Add the `cargo:rerun-if-changed` line.

**Step 2: Add proto module**

In `crates/web-gateway/src/lib.rs`, inside `pub mod proto`, add:

```rust
pub mod monitoring {
    #![allow(
        clippy::all,
        clippy::pedantic,
        clippy::restriction,
        clippy::nursery
    )]
    tonic::include_proto!("monitoring");
}
```

**Step 3: Add config field**

In `crates/web-gateway/src/config.rs`, add to `GatewayConfig`:

```rust
/// gRPC monitoring service endpoint
pub monitoring_service_url: String,
```

Default: `"https://localhost:50057".to_owned()`

In `from_env()`:

```rust
monitoring_service_url: std::env::var("MONITORING_SERVICE_URL")
    .unwrap_or_else(|_| "https://localhost:50057".to_owned()),
```

**Step 4: Add channel to AppState**

In `crates/web-gateway/src/state.rs`, add `pub monitoring_channel: Option<Channel>` to `AppState`. In `AppState::new()`:

```rust
let monitoring_channel = Channel::from_shared(config.monitoring_service_url.clone())
    .ok()
    .map(|c| c.timeout(GRPC_TIMEOUT).connect_lazy());
```

Add `monitoring_channel` to the `Self { ... }` struct.

**Step 5: Add client factory**

In `crates/web-gateway/src/grpc/clients.rs`, add:

```rust
use crate::proto::monitoring::monitoring_service_client::MonitoringServiceClient;

pub fn monitoring_client(channel: &Channel) -> MonitoringServiceClient<Channel> {
    MonitoringServiceClient::new(channel.clone())
}
```

**Step 6: Add REST endpoint**

Create `crates/web-gateway/src/routes/monitoring.rs`:

```rust
use axum::{extract::State, routing::get, Json, Router};

use crate::error::AppError;
use crate::grpc::clients::monitoring_client;
use crate::proto::monitoring::GetLiveTrainingMetricsRequest;
use crate::state::AppState;

pub fn router() -> Router<AppState> {
    Router::new().route("/live-metrics", get(live_metrics))
}

async fn live_metrics(
    State(state): State<AppState>,
) -> Result<Json<serde_json::Value>, AppError> {
    let channel = state
        .monitoring_channel
        .as_ref()
        .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Monitoring service not configured")))?;
    let mut client = monitoring_client(channel);
    let response = client
        .get_live_training_metrics(tonic::Request::new(GetLiveTrainingMetricsRequest {
            model_filter: String::new(),
        }))
        .await?;
    Ok(Json(
        serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?,
    ))
}
```

**Step 7: Mount routes**

In `crates/web-gateway/src/routes/mod.rs`, add `pub mod monitoring;` and mount:

```rust
.nest("/monitoring", monitoring::router())
```

**Step 8: Add training_progress WS poller**

In `crates/web-gateway/src/grpc/streams.rs`, add a function that polls monitoring_service every 3s and broadcasts to `training_progress` topic:

```rust
/// Poll monitoring_service and broadcast training_progress to WebSocket
async fn poll_training_metrics(channel: Channel, tx: broadcast::Sender<String>) {
    use crate::grpc::clients::monitoring_client;
    use crate::proto::monitoring::GetLiveTrainingMetricsRequest;

    let mut interval = tokio::time::interval(Duration::from_secs(3));
    let mut backoff = Duration::from_secs(1);
    let max_backoff = Duration::from_secs(60);

    loop {
        interval.tick().await;
        let mut client = monitoring_client(&channel);
        match client
            .get_live_training_metrics(tonic::Request::new(GetLiveTrainingMetricsRequest {
                model_filter: String::new(),
            }))
            .await
        {
            Ok(resp) => {
                backoff = Duration::from_secs(1);
                let data = serde_json::to_value(resp.into_inner()).unwrap_or_default();
                let msg = ServerMessage::TrainingProgress { data };
                if let Ok(json) = serde_json::to_string(&msg) {
                    drop(tx.send(json));
                }
            }
            Err(e) => {
                if is_unimplemented(&e) {
                    info!("Monitoring service not available, disabling training_progress poller");
                    return;
                }
                warn!("Training metrics poll failed: {}, retrying in {:?}", e, backoff);
                tokio::time::sleep(backoff).await;
                backoff = (backoff * 2).min(max_backoff);
            }
        }
    }
}
```

Update `start_grpc_stream_bridges` signature to also accept `monitoring_channel: Option<Channel>`, and spawn the poller:

```rust
if let Some(channel) = monitoring_channel {
    let tx_training = ws_broadcast.clone();
    tokio::spawn(async move {
        poll_training_metrics(channel, tx_training).await;
    });
}
```

Update the call site in `main.rs` to pass `state.monitoring_channel.clone()`.

**Step 9: Fix tests**

Update `test_state()` in `crates/web-gateway/src/routes/training.rs` (and any other test helpers) to include `monitoring_channel: None`.

Update config tests to include the new `monitoring_service_url` field.

**Step 10: Build and test**

Run: `SQLX_OFFLINE=true cargo check -p web-gateway`
Run: `SQLX_OFFLINE=true cargo test -p web-gateway --lib`
Expected: 0 errors, all tests pass

**Step 11: Commit**

```bash
git add crates/web-gateway/
git commit -m "feat(web-gateway): integrate monitoring service for live training metrics"
```

---

## Task 7: K8s deployment manifest

**Files:**
- Create: `infra/k8s/services/monitoring-service.yaml`

**Step 1: Write manifest**

Follow the pattern from `ml-training-service.yaml`. No ServiceAccount/RBAC needed (monitoring_service only queries Prometheus HTTP, no K8s API).

```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: monitoring-service
  namespace: foxhunt
  labels:
    app.kubernetes.io/name: monitoring-service
    app.kubernetes.io/part-of: foxhunt
spec:
  replicas: 1
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 0
      maxUnavailable: 1
  selector:
    matchLabels:
      app.kubernetes.io/name: monitoring-service
  template:
    metadata:
      annotations:
        gitlab.com/prometheus_scrape: "true"
        gitlab.com/prometheus_port: "9099"
        gitlab.com/prometheus_path: "/metrics"
      labels:
        app.kubernetes.io/name: monitoring-service
        app.kubernetes.io/part-of: foxhunt
    spec:
      securityContext:
        runAsNonRoot: true
        runAsUser: 1000
        runAsGroup: 1000
        fsGroup: 1000
        seccompProfile:
          type: RuntimeDefault
      imagePullSecrets:
        - name: scw-registry
      nodeSelector:
        k8s.scaleway.com/pool-name: foxhunt
      containers:
        - name: monitoring-service
          image: rg.fr-par.scw.cloud/foxhunt-ci/foxhunt-runtime:latest
          securityContext:
            allowPrivilegeEscalation: false
            readOnlyRootFilesystem: true
            capabilities:
              drop: ["ALL"]
          command: ["/binaries/monitoring_service"]
          ports:
            - containerPort: 50057
              name: grpc
            - containerPort: 9099
              name: metrics
          env:
            - name: PROMETHEUS_URL
              value: "http://gitlab-prometheus-server.foxhunt.svc:80"
            - name: GRPC_PORT
              value: "50057"
            - name: METRICS_PORT
              value: "9099"
            - name: STREAM_INTERVAL_SECS
              value: "3"
            - name: RUST_LOG
              value: info
            - name: OTEL_EXPORTER_OTLP_ENDPOINT
              value: "http://tempo.foxhunt.svc.cluster.local:4317"
          volumeMounts:
            - name: binaries
              mountPath: /binaries
              readOnly: true
            - name: tmp
              mountPath: /tmp
          readinessProbe:
            tcpSocket:
              port: 50057
            initialDelaySeconds: 5
            periodSeconds: 10
          livenessProbe:
            tcpSocket:
              port: 50057
            initialDelaySeconds: 10
            periodSeconds: 15
            failureThreshold: 5
          resources:
            requests:
              cpu: 50m
              memory: 64Mi
            limits:
              cpu: 200m
              memory: 128Mi
      volumes:
        - name: binaries
          persistentVolumeClaim:
            claimName: foxhunt-binaries
            readOnly: true
        - name: tmp
          emptyDir:
            sizeLimit: 10Mi
---
apiVersion: v1
kind: Service
metadata:
  name: monitoring-service
  namespace: foxhunt
  labels:
    app.kubernetes.io/name: monitoring-service
    app.kubernetes.io/part-of: foxhunt
spec:
  selector:
    app.kubernetes.io/name: monitoring-service
  ports:
    - port: 50057
      targetPort: 50057
      name: grpc
    - port: 9099
      targetPort: 9099
      name: metrics
```

**Step 2: Commit**

```bash
git add infra/k8s/services/monitoring-service.yaml
git commit -m "infra: add K8s deployment manifest for monitoring-service"
```

---

## Task 8: CI integration

**Files:**
- Modify: `.gitlab-ci.yml` (add monitoring_service to compile-services job)

**Step 1: Add to compile-services job**

Find the `compile-services` job and add `monitoring_service` to the list of binaries that get compiled and uploaded to S3.

**Step 2: Commit**

```bash
git add .gitlab-ci.yml
git commit -m "ci: add monitoring_service to compile-services job"
```

---

## Task 9: Final build verification

**Step 1: Full workspace check**

Run: `SQLX_OFFLINE=true cargo check --workspace`
Expected: 0 errors, 0 warnings

**Step 2: Run affected tests**

Run: `SQLX_OFFLINE=true cargo test -p common --lib`
Run: `SQLX_OFFLINE=true cargo test -p web-gateway --lib`
Run: `SQLX_OFFLINE=true cargo test -p fxt --lib`
Expected: All pass

**Step 3: Clippy**

Run: `SQLX_OFFLINE=true cargo clippy --workspace -- -D warnings`
Expected: 0 errors, 0 warnings
