//! Generate SVG and PNG plots from benchmark JSON results.
//!
//! Reads JSON files produced by `melin-bench --json` and generates:
//! 1. **Latency CDF** — percentile plot comparing benchmark configs
//! 2. **Saturation curve** — throughput vs latency at multiple load levels
//! 3. **Pipeline breakdown** — stage utilization bar chart
//!
//! Usage:
//!   melin-plot latency-cdf -o latency.svg results/*.json
//!   melin-plot saturation -o saturation.svg sweep/*.json
//!   melin-plot pipeline -o pipeline.svg --stats pipeline-stats.log

use std::collections::{BTreeMap, HashMap};
use std::fs;
use std::path::PathBuf;

use plotters::prelude::*;
use serde::Deserialize;

/// Render a chart to both SVG and PNG. The block receives `root` as a
/// `DrawingArea` and should build the full chart on it. The macro expands
/// the body twice — once for each backend — so all chart-building code
/// works through the backend-agnostic `DrawingBackend` trait.
macro_rules! render_both {
    ($path:expr, $size:expr, |$root:ident| $body:block) => {{
        // SVG
        {
            let $root = SVGBackend::new(&$path, $size).into_drawing_area();
            $body
            $root.present().unwrap();
            eprintln!("wrote {}", $path.display());
        }
        // PNG
        {
            let png_path = $path.with_extension("png");
            let $root = BitMapBackend::new(&png_path, $size).into_drawing_area();
            $body
            $root.present().unwrap();
            eprintln!("wrote {}", png_path.display());
        }
    }};
}

// --- JSON schema matching melin-bench output ---

#[derive(Debug, Deserialize)]
#[allow(dead_code)]
struct BenchResult {
    label: String,
    #[allow(dead_code)]
    measured_orders: u64,
    #[allow(dead_code)]
    wall_ms: f64,
    throughput_ops: f64,
    latency: BTreeMap<String, f64>,
}

/// Health data embedded in the JSON output, from the health poller.
#[derive(Debug, Deserialize)]
struct HealthResult {
    #[allow(dead_code)]
    label: String,
    throughput_ops: f64,
    #[serde(default)]
    health: Vec<HealthPoint>,
}

/// A single health sample from the Prometheus metrics poller.
///
/// Every server-side `melin_*` metric is a typed field here. The
/// `_assert_all_metrics_plotted` function below destructures this struct
/// without `..`, so adding a field without listing it there is a compile
/// error — that is how we guarantee every metric has an explicit plotting
/// decision.
#[derive(Debug, Deserialize, Default)]
struct HealthPoint {
    elapsed_secs: f64,
    active_connections: u64,
    events_processed: u64,
    journal_sequence: u64,
    replication_lag: u64,
    input_queue_depth: u64,
    // Field-only by design: this is the y-axis bound for the queue-depth
    // chart, not a series. No `HealthMetric` variant.
    input_queue_capacity: u64,
    pipeline_healthy: bool,
    trading_active: bool,
    #[serde(default, rename = "melin_replicas_connected")]
    replicas_connected: u64,
    #[serde(default, rename = "melin_replica_acked_sequence_slot_0")]
    replica_acked_sequence_slot_0: u64,
    #[serde(default, rename = "melin_replica_acked_sequence_slot_1")]
    replica_acked_sequence_slot_1: u64,
    #[serde(default, rename = "melin_replica_in_memory_sequence_slot_0")]
    replica_in_memory_sequence_slot_0: u64,
    #[serde(default, rename = "melin_replica_in_memory_sequence_slot_1")]
    replica_in_memory_sequence_slot_1: u64,
    #[serde(default, rename = "melin_replica_lag_slot_0")]
    replica_lag_slot_0: u64,
    #[serde(default, rename = "melin_replica_lag_slot_1")]
    replica_lag_slot_1: u64,
    #[serde(default, rename = "melin_replica_bytes_sent_total_slot_0")]
    replica_bytes_sent_total_slot_0: u64,
    #[serde(default, rename = "melin_replica_bytes_sent_total_slot_1")]
    replica_bytes_sent_total_slot_1: u64,
    #[serde(default, rename = "melin_replica_ack_latency_us_slot_0")]
    replica_ack_latency_us_slot_0: u64,
    #[serde(default, rename = "melin_replica_ack_latency_us_slot_1")]
    replica_ack_latency_us_slot_1: u64,
    #[serde(default, rename = "melin_replica_catching_up_slot_0")]
    replica_catching_up_slot_0: u64,
    #[serde(default, rename = "melin_replica_catching_up_slot_1")]
    replica_catching_up_slot_1: u64,
    #[serde(default, rename = "melin_replica_evictions_total")]
    replica_evictions_total: u64,
    #[serde(default, rename = "melin_replication_ring_depth_slot_0")]
    replication_ring_depth_slot_0: u64,
    #[serde(default, rename = "melin_replication_ring_depth_slot_1")]
    replication_ring_depth_slot_1: u64,
    #[serde(default, rename = "melin_fastest_replica_cursor")]
    fastest_replica_cursor: u64,
    #[serde(default, rename = "melin_stage_busy_total_stage_journal")]
    stage_busy_total_journal: u64,
    #[serde(default, rename = "melin_stage_busy_total_stage_matching")]
    stage_busy_total_matching: u64,
    #[serde(default, rename = "melin_stage_busy_total_stage_response")]
    stage_busy_total_response: u64,
    #[serde(default, rename = "melin_stage_idle_total_stage_journal")]
    stage_idle_total_journal: u64,
    #[serde(default, rename = "melin_stage_idle_total_stage_matching")]
    stage_idle_total_matching: u64,
    #[serde(default, rename = "melin_stage_idle_total_stage_response")]
    stage_idle_total_response: u64,
    #[serde(default, rename = "melin_response_gate_total_blocker_journal")]
    response_gate_total_journal: u64,
    #[serde(default, rename = "melin_response_gate_total_blocker_replication")]
    response_gate_total_replication: u64,
    #[serde(default, rename = "melin_durability_policy_degraded")]
    durability_policy_degraded: u64,
    /// Catch-all for genuinely-unknown metrics the server added without
    /// updating this struct. A non-empty map at load time is logged as a
    /// warning so we know to add a typed field + plot.
    #[serde(flatten)]
    unknown: HashMap<String, serde_json::Value>,
}

/// Type-safe handle to a single metric on `HealthPoint`. Plot helpers take
/// `HealthMetric` values instead of stringly-typed names: renaming a struct
/// field cascades through `HealthMetric::read` as a compile error, and
/// adding a variant without a match arm fails exhaustiveness.
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
enum HealthMetric {
    ActiveConnections,
    EventsProcessed,
    JournalSequence,
    ReplicationLag,
    InputQueueDepth,
    PipelineHealthy,
    TradingActive,
    ReplicasConnected,
    ReplicaAckedSequenceSlot0,
    ReplicaAckedSequenceSlot1,
    ReplicaInMemorySequenceSlot0,
    ReplicaInMemorySequenceSlot1,
    ReplicaLagSlot0,
    ReplicaLagSlot1,
    ReplicaBytesSentTotalSlot0,
    ReplicaBytesSentTotalSlot1,
    ReplicaAckLatencyUsSlot0,
    ReplicaAckLatencyUsSlot1,
    ReplicaCatchingUpSlot0,
    ReplicaCatchingUpSlot1,
    ReplicaEvictionsTotal,
    ReplicationRingDepthSlot0,
    ReplicationRingDepthSlot1,
    FastestReplicaCursor,
    StageBusyTotalJournal,
    StageBusyTotalMatching,
    StageBusyTotalResponse,
    StageIdleTotalJournal,
    StageIdleTotalMatching,
    StageIdleTotalResponse,
    ResponseGateTotalJournal,
    ResponseGateTotalReplication,
    DurabilityPolicyDegraded,
}

impl HealthMetric {
    fn read(self, h: &HealthPoint) -> f64 {
        match self {
            Self::ActiveConnections => h.active_connections as f64,
            Self::EventsProcessed => h.events_processed as f64,
            Self::JournalSequence => h.journal_sequence as f64,
            Self::ReplicationLag => h.replication_lag as f64,
            Self::InputQueueDepth => h.input_queue_depth as f64,
            Self::PipelineHealthy => h.pipeline_healthy as u64 as f64,
            Self::TradingActive => h.trading_active as u64 as f64,
            Self::ReplicasConnected => h.replicas_connected as f64,
            Self::ReplicaAckedSequenceSlot0 => h.replica_acked_sequence_slot_0 as f64,
            Self::ReplicaAckedSequenceSlot1 => h.replica_acked_sequence_slot_1 as f64,
            Self::ReplicaInMemorySequenceSlot0 => h.replica_in_memory_sequence_slot_0 as f64,
            Self::ReplicaInMemorySequenceSlot1 => h.replica_in_memory_sequence_slot_1 as f64,
            Self::ReplicaLagSlot0 => h.replica_lag_slot_0 as f64,
            Self::ReplicaLagSlot1 => h.replica_lag_slot_1 as f64,
            Self::ReplicaBytesSentTotalSlot0 => h.replica_bytes_sent_total_slot_0 as f64,
            Self::ReplicaBytesSentTotalSlot1 => h.replica_bytes_sent_total_slot_1 as f64,
            Self::ReplicaAckLatencyUsSlot0 => h.replica_ack_latency_us_slot_0 as f64,
            Self::ReplicaAckLatencyUsSlot1 => h.replica_ack_latency_us_slot_1 as f64,
            Self::ReplicaCatchingUpSlot0 => h.replica_catching_up_slot_0 as f64,
            Self::ReplicaCatchingUpSlot1 => h.replica_catching_up_slot_1 as f64,
            Self::ReplicaEvictionsTotal => h.replica_evictions_total as f64,
            Self::ReplicationRingDepthSlot0 => h.replication_ring_depth_slot_0 as f64,
            Self::ReplicationRingDepthSlot1 => h.replication_ring_depth_slot_1 as f64,
            Self::FastestReplicaCursor => h.fastest_replica_cursor as f64,
            Self::StageBusyTotalJournal => h.stage_busy_total_journal as f64,
            Self::StageBusyTotalMatching => h.stage_busy_total_matching as f64,
            Self::StageBusyTotalResponse => h.stage_busy_total_response as f64,
            Self::StageIdleTotalJournal => h.stage_idle_total_journal as f64,
            Self::StageIdleTotalMatching => h.stage_idle_total_matching as f64,
            Self::StageIdleTotalResponse => h.stage_idle_total_response as f64,
            Self::ResponseGateTotalJournal => h.response_gate_total_journal as f64,
            Self::ResponseGateTotalReplication => h.response_gate_total_replication as f64,
            Self::DurabilityPolicyDegraded => h.durability_policy_degraded as f64,
        }
    }
}

/// Compile-time check: every field on `HealthPoint` must be named here. If
/// a new typed field is added without updating this destructure, the
/// compiler errors with "pattern does not mention field `...`".
///
/// `HealthMetric::read` already enforces exhaustiveness over the *variants*
/// it knows about — this destructure is the complementary guard that
/// catches a new field being added without adding a matching variant. The
/// `// plotted by` comments document where each field is consumed; fields
/// that have no `HealthMetric` variant by design (`input_queue_capacity`,
/// `elapsed_secs`, `unknown`) are noted in their corresponding row.
#[allow(dead_code, unused_variables)]
fn _assert_all_metrics_plotted(p: &HealthPoint) {
    let HealthPoint {
        elapsed_secs,                      // x-axis for every plot
        active_connections,                // plot_health_extra_metric
        events_processed,                  // plot_health_events
        journal_sequence,                  // plot_health_journal_seq
        replication_lag,                   // plot_health_replication_lag
        input_queue_depth,                 // plot_health_queue_depth
        input_queue_capacity,              // y-axis bound for plot_health_queue_depth
        pipeline_healthy,                  // plot_health_extra_metric
        trading_active,                    // plot_health_extra_metric
        replicas_connected,                // plot_health_extra_metric
        replica_acked_sequence_slot_0,     // plot_health_extra_metric (per-slot)
        replica_acked_sequence_slot_1,     // plot_health_extra_metric (per-slot)
        replica_in_memory_sequence_slot_0, // plot_health_extra_metric (per-slot)
        replica_in_memory_sequence_slot_1, // plot_health_extra_metric (per-slot)
        replica_lag_slot_0,                // plot_health_extra_metric (per-slot)
        replica_lag_slot_1,                // plot_health_extra_metric (per-slot)
        replica_bytes_sent_total_slot_0,   // plot_health_extra_metric (per-slot)
        replica_bytes_sent_total_slot_1,   // plot_health_extra_metric (per-slot)
        replica_ack_latency_us_slot_0,     // plot_health_extra_metric (per-slot)
        replica_ack_latency_us_slot_1,     // plot_health_extra_metric (per-slot)
        replica_catching_up_slot_0,        // plot_health_extra_metric (per-slot)
        replica_catching_up_slot_1,        // plot_health_extra_metric (per-slot)
        replica_evictions_total,           // plot_health_extra_metric
        replication_ring_depth_slot_0,     // plot_health_extra_metric (per-slot)
        replication_ring_depth_slot_1,     // plot_health_extra_metric (per-slot)
        fastest_replica_cursor,            // plot_health_extra_metric
        stage_busy_total_journal,          // plot_health_utilization (derivative)
        stage_busy_total_matching,         // plot_health_utilization (derivative)
        stage_busy_total_response,         // plot_health_utilization (derivative)
        stage_idle_total_journal,          // plot_health_utilization (derivative)
        stage_idle_total_matching,         // plot_health_utilization (derivative)
        stage_idle_total_response,         // plot_health_utilization (derivative)
        response_gate_total_journal,       // plot_health_extra_metric
        response_gate_total_replication,   // plot_health_extra_metric
        durability_policy_degraded,        // plot_health_extra_metric
        unknown,                           // warned at load time, not plotted
    } = p;
}

/// Print a warning if the loaded health result contains metric keys that
/// don't correspond to any typed field. New server-side metrics land here
/// and are a signal to add a typed field + plot.
fn warn_unknown_metrics(result: &HealthResult, filename: &str) {
    let mut seen: std::collections::BTreeSet<&str> = std::collections::BTreeSet::new();
    for h in &result.health {
        for k in h.unknown.keys() {
            seen.insert(k.as_str());
        }
    }
    if !seen.is_empty() {
        let keys: Vec<&str> = seen.into_iter().collect();
        eprintln!(
            "warning: {filename} has unrecognized health metrics (no plot generated): {}",
            keys.join(", ")
        );
    }
}

/// Pipeline stats parsed from tracing output.
#[derive(Debug)]
struct PipelineStage {
    name: String,
    pct_busy: f64,
}

// --- Colors ---
// Professional palette suitable for investor presentations.
const COLOR_FSYNC: RGBColor = RGBColor(41, 98, 255); // blue
const COLOR_NO_PERSIST: RGBColor = RGBColor(0, 200, 83); // green
const COLOR_SINGLE: RGBColor = RGBColor(255, 145, 0); // amber
const COLORS: [RGBColor; 6] = [
    COLOR_FSYNC,
    COLOR_NO_PERSIST,
    COLOR_SINGLE,
    RGBColor(156, 39, 176), // purple
    RGBColor(244, 67, 54),  // red
    RGBColor(0, 188, 212),  // cyan
];

fn main() {
    let args: Vec<String> = std::env::args().collect();

    if args.len() < 2 {
        print_usage();
        std::process::exit(1);
    }

    match args[1].as_str() {
        "latency-cdf" => cmd_latency_cdf(&args[2..]),
        "saturation" => cmd_saturation(&args[2..]),
        "sweep" => cmd_sweep(&args[2..]),
        "stability" => cmd_stability(&args[2..]),
        "health" => cmd_health(&args[2..]),
        "pipeline" => cmd_pipeline(&args[2..]),
        "all" => cmd_all(&args[2..]),
        _ => {
            eprintln!("unknown command: {}", args[1]);
            print_usage();
            std::process::exit(1);
        }
    }
}

fn print_usage() {
    eprintln!("Usage: melin-plot <command> [options] <files...>");
    eprintln!();
    eprintln!("Commands:");
    eprintln!("  latency-cdf  Percentile plot comparing benchmark configs");
    eprintln!("  saturation   Throughput vs latency at multiple load levels");
    eprintln!("  sweep        Parameter vs throughput (e.g. window depth sweep)");
    eprintln!("  stability    Latency stability over time (from time-series JSON)");
    eprintln!("  health       Server health metrics over time (queue depth, events, etc.)");
    eprintln!("  pipeline     Pipeline stage utilization bar chart");
    eprintln!("  all          Generate all plots from a results directory");
    eprintln!();
    eprintln!("Options:");
    eprintln!("  -o <file>    Output SVG path (default: <command>.svg)");
    eprintln!("  --stats <f>  Pipeline stats log file (for pipeline command)");
    eprintln!();
    eprintln!("Examples:");
    eprintln!("  melin-plot latency-cdf results/1-fsync.json results/2-no-persist.json");
    eprintln!("  melin-plot saturation sweep/*.json");
    eprintln!("  melin-plot pipeline --stats /tmp/melin-server.log");
    eprintln!("  melin-plot all results/");
}

// --- Argument parsing (minimal, no clap dependency for this tool) ---

struct PlotArgs {
    output: PathBuf,
    stats_file: Option<PathBuf>,
    files: Vec<PathBuf>,
}

fn parse_args(args: &[String], default_output: &str) -> PlotArgs {
    let mut output = PathBuf::from(default_output);
    let mut stats_file = None;
    let mut files = Vec::new();
    let mut i = 0;

    while i < args.len() {
        match args[i].as_str() {
            "-o" => {
                i += 1;
                if i < args.len() {
                    output = PathBuf::from(&args[i]);
                }
            }
            "--stats" => {
                i += 1;
                if i < args.len() {
                    stats_file = Some(PathBuf::from(&args[i]));
                }
            }
            _ => {
                files.push(PathBuf::from(&args[i]));
            }
        }
        i += 1;
    }

    PlotArgs {
        output,
        stats_file,
        files,
    }
}

/// A loaded benchmark result paired with its source filename.
struct LoadedResult {
    result: BenchResult,
    filename: String,
}

fn load_results(files: &[PathBuf]) -> Vec<LoadedResult> {
    let mut results = Vec::new();
    for path in files {
        let data = fs::read_to_string(path).unwrap_or_else(|e| {
            eprintln!("warning: cannot read {}: {}", path.display(), e);
            String::new()
        });
        if data.is_empty() {
            continue;
        }
        match serde_json::from_str::<BenchResult>(&data) {
            Ok(r) => {
                let filename = path
                    .file_name()
                    .map(|f| f.to_string_lossy().into_owned())
                    .unwrap_or_default();
                results.push(LoadedResult {
                    result: r,
                    filename,
                });
            }
            Err(e) => eprintln!("warning: cannot parse {}: {}", path.display(), e),
        }
    }
    results
}

// --- Percentile extraction ---

/// Standard percentile labels in order, mapped to quantile values.
const PERCENTILES: &[(&str, &str, f64)] = &[
    ("min_us", "min", 0.0),
    ("p50_us", "p50", 50.0),
    ("p90_us", "p90", 90.0),
    ("p99_us", "p99", 99.0),
    ("p99.9_us", "p99.9", 99.9),
    ("p99.9.9_us", "p99.99", 99.99),
    ("p99.9.9.9_us", "p99.999", 99.999),
    ("p99.9.9.9.9_us", "p99.9999", 99.9999),
    ("p99.9.9.9.9.9_us", "p99.99999", 99.99999),
    ("max_us", "max", 100.0),
];

fn extract_percentiles(result: &BenchResult) -> Vec<(f64, f64)> {
    let mut points = Vec::new();
    for &(key, _, pct) in PERCENTILES {
        if let Some(&val) = result.latency.get(key) {
            points.push((pct, val));
        }
    }
    points
}

/// Format throughput as a compact string (e.g. "4.0M/s").
fn format_throughput(ops: f64) -> String {
    if ops >= 1_000_000.0 {
        format!("{:.1}M/s", ops / 1_000_000.0)
    } else if ops >= 1_000.0 {
        format!("{:.0}K/s", ops / 1_000.0)
    } else {
        format!("{:.0}/s", ops)
    }
}

/// Extract a human-readable mode name from a filename like "1-fsync.json".
///
/// Strips the leading number prefix and extension, then title-cases the
/// remainder: "1-fsync.json" -> "Fsync", "4-replication.json" -> "Replication",
/// "2-no-persist.json" -> "No-persist".
fn mode_from_filename(filename: &str) -> String {
    let stem = filename.strip_suffix(".json").unwrap_or(filename);
    // Strip leading digit-dash prefix (e.g. "1-", "42-").
    let name = stem
        .find('-')
        .and_then(|pos| {
            if stem[..pos].chars().all(|c| c.is_ascii_digit()) {
                Some(&stem[pos + 1..])
            } else {
                None
            }
        })
        .unwrap_or(stem);
    // Title-case the first character.
    let mut chars = name.chars();
    match chars.next() {
        Some(c) => format!("{}{}", c.to_uppercase(), chars.as_str()),
        None => name.to_string(),
    }
}

/// Build a label for the latency CDF plot: "Mode (throughput)".
///
/// Uses the source filename to infer the mode name, then appends throughput.
/// Example: "1-fsync.json" with 4.0M ops/s -> "Fsync (4.0M/s)".
fn cdf_label(loaded: &LoadedResult) -> String {
    let mode = mode_from_filename(&loaded.filename);
    let tp = format_throughput(loaded.result.throughput_ops);
    format!("{mode} ({tp})")
}

/// Build a label for saturation plots using the filename stem directly.
///
/// Example: "w256.json" -> "w256", "i100.json" -> "i100".
fn saturation_label(loaded: &LoadedResult) -> String {
    loaded
        .filename
        .strip_suffix(".json")
        .unwrap_or(&loaded.filename)
        .to_string()
}

// =====================================================================
// Command: latency-cdf
// =====================================================================

fn cmd_latency_cdf(args: &[String]) {
    let opts = parse_args(args, "latency-cdf.svg");
    let results = load_results(&opts.files);
    if results.is_empty() {
        eprintln!("error: no result files loaded");
        std::process::exit(1);
    }
    plot_latency_cdf(&results, &opts.output);
}

fn plot_latency_cdf(results: &[LoadedResult], output: &PathBuf) {
    // Filter out low-throughput results (e.g. single-order latency at ~13K ops/s)
    // that are incomparable with peak-load benchmarks on the same CDF axes.
    let results: Vec<&LoadedResult> = results
        .iter()
        .filter(|r| r.result.throughput_ops >= 100_000.0)
        .collect();
    if results.is_empty() {
        eprintln!("warning: no results >= 100K ops/s for CDF plot, skipping");
        return;
    }

    render_both!(output, (900, 500), |root| {
        root.fill(&WHITE).unwrap();

        // Find max latency across all results for Y-axis range.
        let max_lat = results
            .iter()
            .flat_map(|r| r.result.latency.values())
            .cloned()
            .fold(0.0f64, f64::max)
            * 1.1;

        // Percentile labels for X-axis.
        let pct_labels: Vec<&str> = PERCENTILES.iter().map(|&(_, label, _)| label).collect();

        let mut chart = ChartBuilder::on(&root)
            .caption("Latency by Percentile", ("sans-serif", 22).into_font())
            .margin(15)
            .x_label_area_size(50)
            .y_label_area_size(70)
            .build_cartesian_2d(0usize..pct_labels.len().saturating_sub(1), 0.0..max_lat)
            .unwrap();

        chart
            .configure_mesh()
            .x_desc("Percentile")
            .y_desc("Latency (µs)")
            .x_label_formatter(&|idx| pct_labels.get(*idx).copied().unwrap_or("").to_string())
            .y_label_formatter(&|v| {
                if *v >= 1000.0 {
                    format!("{:.1}ms", v / 1000.0)
                } else {
                    format!("{:.0}µs", v)
                }
            })
            .draw()
            .unwrap();

        for (i, loaded) in results.iter().enumerate() {
            let points = extract_percentiles(&loaded.result);
            let color = COLORS[i % COLORS.len()];
            let label = cdf_label(loaded);

            // Map percentile values to x-indices.
            let indexed: Vec<(usize, f64)> = points
                .iter()
                .filter_map(|(pct, val)| {
                    PERCENTILES
                        .iter()
                        .position(|&(_, _, p)| (p - pct).abs() < 0.000001)
                        .map(|idx| (idx, *val))
                })
                .collect();

            chart
                .draw_series(LineSeries::new(indexed.clone(), color.stroke_width(2)))
                .unwrap()
                .label(label)
                .legend(move |(x, y)| {
                    PathElement::new(vec![(x, y), (x + 20, y)], color.stroke_width(2))
                });

            // Draw dots at each data point.
            chart
                .draw_series(
                    indexed
                        .iter()
                        .map(|&(x, y)| Circle::new((x, y), 3, color.filled())),
                )
                .unwrap();
        }

        chart
            .configure_series_labels()
            .position(SeriesLabelPosition::UpperLeft)
            .background_style(WHITE.mix(0.8))
            .border_style(BLACK.mix(0.3))
            .draw()
            .unwrap();
    });
}

// =====================================================================
// Command: saturation
// =====================================================================

fn cmd_saturation(args: &[String]) {
    let opts = parse_args(args, "saturation.svg");
    let loaded = load_results(&opts.files);
    if loaded.len() < 2 {
        eprintln!("error: need at least 2 result files for saturation curve");
        std::process::exit(1);
    }
    plot_saturation(&loaded, &opts.output);
}

fn plot_saturation(results: &[LoadedResult], output: &PathBuf) {
    // Sort by throughput for the curve.
    let mut sorted: Vec<&LoadedResult> = results.iter().collect();
    sorted.sort_by(|a, b| {
        a.result
            .throughput_ops
            .partial_cmp(&b.result.throughput_ops)
            .unwrap()
    });

    let max_throughput = sorted
        .last()
        .map(|r| r.result.throughput_ops)
        .unwrap_or(1.0)
        * 1.1;
    let max_latency = sorted
        .iter()
        .filter_map(|r| r.result.latency.get("p99_us"))
        .cloned()
        .fold(0.0f64, f64::max)
        * 1.3;

    render_both!(output, (900, 500), |root| {
        root.fill(&WHITE).unwrap();

        let mut chart = ChartBuilder::on(&root)
            .caption(
                "Throughput vs Latency (Saturation Curve)",
                ("sans-serif", 22).into_font(),
            )
            .margin(15)
            .x_label_area_size(50)
            .y_label_area_size(70)
            .build_cartesian_2d(0.0..max_throughput, 0.0..max_latency)
            .unwrap();

        chart
            .configure_mesh()
            .x_desc("Throughput (orders/sec)")
            .y_desc("Latency (µs)")
            .x_label_formatter(&|v| {
                if *v >= 1_000_000.0 {
                    format!("{:.1}M", v / 1_000_000.0)
                } else if *v >= 1_000.0 {
                    format!("{:.0}K", v / 1_000.0)
                } else {
                    format!("{:.0}", v)
                }
            })
            .y_label_formatter(&|v| {
                if *v >= 1000.0 {
                    format!("{:.1}ms", v / 1000.0)
                } else {
                    format!("{:.0}µs", v)
                }
            })
            .draw()
            .unwrap();

        // Plot p50 line.
        let p50_points: Vec<(f64, f64)> = sorted
            .iter()
            .filter_map(|r| {
                r.result
                    .latency
                    .get("p50_us")
                    .map(|lat| (r.result.throughput_ops, *lat))
            })
            .collect();

        if !p50_points.is_empty() {
            chart
                .draw_series(LineSeries::new(
                    p50_points.clone(),
                    COLOR_FSYNC.stroke_width(2),
                ))
                .unwrap()
                .label("p50")
                .legend(move |(x, y)| {
                    PathElement::new(vec![(x, y), (x + 20, y)], COLOR_FSYNC.stroke_width(2))
                });
            chart
                .draw_series(
                    p50_points
                        .iter()
                        .map(|&(x, y)| Circle::new((x, y), 4, COLOR_FSYNC.filled())),
                )
                .unwrap();
        }

        // Plot p99 line.
        let p99_points: Vec<(f64, f64)> = sorted
            .iter()
            .filter_map(|r| {
                r.result
                    .latency
                    .get("p99_us")
                    .map(|lat| (r.result.throughput_ops, *lat))
            })
            .collect();

        if !p99_points.is_empty() {
            chart
                .draw_series(LineSeries::new(
                    p99_points.clone(),
                    COLOR_NO_PERSIST.stroke_width(2),
                ))
                .unwrap()
                .label("p99")
                .legend(move |(x, y)| {
                    PathElement::new(vec![(x, y), (x + 20, y)], COLOR_NO_PERSIST.stroke_width(2))
                });
            chart
                .draw_series(
                    p99_points
                        .iter()
                        .map(|&(x, y)| Circle::new((x, y), 4, COLOR_NO_PERSIST.filled())),
                )
                .unwrap();
        }

        // Plot p99.9 line.
        let p999_points: Vec<(f64, f64)> = sorted
            .iter()
            .filter_map(|r| {
                r.result
                    .latency
                    .get("p99.9_us")
                    .map(|lat| (r.result.throughput_ops, *lat))
            })
            .collect();

        if !p999_points.is_empty() {
            chart
                .draw_series(LineSeries::new(
                    p999_points.clone(),
                    COLOR_SINGLE.stroke_width(2),
                ))
                .unwrap()
                .label("p99.9")
                .legend(move |(x, y)| {
                    PathElement::new(vec![(x, y), (x + 20, y)], COLOR_SINGLE.stroke_width(2))
                });
            chart
                .draw_series(
                    p999_points
                        .iter()
                        .map(|&(x, y)| Circle::new((x, y), 4, COLOR_SINGLE.filled())),
                )
                .unwrap();
        }

        // Label each data point with its filename stem (e.g. "w32", "w256")
        // above the p99.9 dot for readability.
        let labels: Vec<(f64, f64, String)> = sorted
            .iter()
            .filter_map(|r| {
                r.result
                    .latency
                    .get("p99.9_us")
                    .or_else(|| r.result.latency.get("p99_us"))
                    .map(|lat| (r.result.throughput_ops, *lat, saturation_label(r)))
            })
            .collect();
        if !labels.is_empty() {
            chart
                .draw_series(labels.iter().map(|(x, y, label)| {
                    Text::new(
                        label.clone(),
                        (*x, *y + max_latency * 0.04),
                        ("sans-serif", 11).into_font(),
                    )
                }))
                .unwrap();
        }

        chart
            .configure_series_labels()
            .position(SeriesLabelPosition::UpperLeft)
            .background_style(WHITE.mix(0.8))
            .border_style(BLACK.mix(0.3))
            .draw()
            .unwrap();
    });
}

// =====================================================================
// Command: sweep — parameter vs throughput
// =====================================================================

fn cmd_sweep(args: &[String]) {
    let opts = parse_args(args, "sweep.svg");
    let results = load_results(&opts.files);
    if results.len() < 2 {
        eprintln!("error: need at least 2 result files for sweep plot");
        std::process::exit(1);
    }
    plot_sweep(&results, &opts.output);
}

/// Extract the numeric value from a filename stem.
/// "w256.json" → 256, "i100.json" → 100, "w32.json" → 32.
fn extract_sweep_value(filename: &str) -> Option<f64> {
    let stem = filename.strip_suffix(".json").unwrap_or(filename);
    // Strip leading non-digit prefix (e.g. "w" from "w256").
    let digits: String = stem.chars().skip_while(|c| !c.is_ascii_digit()).collect();
    digits.parse::<f64>().ok()
}

/// Extract the parameter name prefix from a filename stem.
/// "w256.json" → "Window depth", "i100.json" → "Instruments", "c64.json" → "Clients".
fn sweep_x_label(filename: &str) -> &'static str {
    let stem = filename.strip_suffix(".json").unwrap_or(filename);
    if stem.starts_with('w') {
        "Window depth"
    } else if stem.starts_with('i') {
        "Instruments"
    } else if stem.starts_with('c') {
        "Clients"
    } else if stem.starts_with('a') {
        "Accounts"
    } else {
        "Parameter"
    }
}

/// One parameter-sweep data point: the swept parameter value paired with the
/// throughput and p99 latency the engine achieved at that value.
struct SweepPoint {
    x: f64,
    throughput: f64,
    p99_latency: f64,
}

fn plot_sweep(results: &[LoadedResult], output: &PathBuf) {
    let mut points: Vec<SweepPoint> = results
        .iter()
        .filter_map(|r| {
            let x = extract_sweep_value(&r.filename)?;
            let p99_latency = r.result.latency.get("p99_us").copied().unwrap_or(0.0);
            Some(SweepPoint {
                x,
                throughput: r.result.throughput_ops,
                p99_latency,
            })
        })
        .collect();
    points.sort_by(|a, b| a.x.partial_cmp(&b.x).unwrap());

    if points.is_empty() {
        eprintln!("error: no valid sweep data (could not extract numeric values from filenames)");
        std::process::exit(1);
    }

    let x_label = sweep_x_label(&results[0].filename);
    let max_x = points.last().map(|p| p.x).unwrap_or(1.0) * 1.15;
    let max_throughput = points.iter().map(|p| p.throughput).fold(0.0f64, f64::max) * 1.15;
    let max_latency = points.iter().map(|p| p.p99_latency).fold(0.0f64, f64::max) * 1.3;

    render_both!(output, (900, 500), |root| {
        root.fill(&WHITE).unwrap();

        // Split into left (throughput) and right (latency) areas.
        let (left, right) = root.split_horizontally(800);

        let mut chart = ChartBuilder::on(&left)
            .caption(
                format!("Throughput & Latency vs {x_label}"),
                ("sans-serif", 22).into_font(),
            )
            .margin(15)
            .x_label_area_size(50)
            .y_label_area_size(70)
            .right_y_label_area_size(70)
            .build_cartesian_2d(0.0..max_x, 0.0..max_throughput)
            .unwrap()
            .set_secondary_coord(0.0..max_x, 0.0..max_latency);

        chart
            .configure_mesh()
            .x_desc(x_label)
            .y_desc("Throughput (orders/sec)")
            .x_label_formatter(&|v| {
                if *v >= 1000.0 {
                    format!("{:.0}K", v / 1000.0)
                } else {
                    format!("{:.0}", v)
                }
            })
            .y_label_formatter(&|v| {
                if *v >= 1_000_000.0 {
                    format!("{:.1}M", v / 1_000_000.0)
                } else if *v >= 1_000.0 {
                    format!("{:.0}K", v / 1_000.0)
                } else {
                    format!("{:.0}", v)
                }
            })
            .draw()
            .unwrap();

        chart
            .configure_secondary_axes()
            .y_desc("p99 Latency (µs)")
            .y_label_formatter(&|v| {
                if *v >= 1000.0 {
                    format!("{:.1}ms", v / 1000.0)
                } else {
                    format!("{:.0}µs", v)
                }
            })
            .draw()
            .unwrap();

        // Throughput line (primary Y-axis).
        let tp_points: Vec<(f64, f64)> = points.iter().map(|p| (p.x, p.throughput)).collect();
        chart
            .draw_series(LineSeries::new(
                tp_points.clone(),
                COLOR_FSYNC.stroke_width(2),
            ))
            .unwrap()
            .label("Throughput")
            .legend(move |(x, y)| {
                PathElement::new(vec![(x, y), (x + 20, y)], COLOR_FSYNC.stroke_width(2))
            });
        chart
            .draw_series(
                tp_points
                    .iter()
                    .map(|&(x, y)| Circle::new((x, y), 4, COLOR_FSYNC.filled())),
            )
            .unwrap();

        // p99 latency line (secondary Y-axis).
        let lat_points: Vec<(f64, f64)> = points.iter().map(|p| (p.x, p.p99_latency)).collect();
        chart
            .draw_secondary_series(LineSeries::new(
                lat_points.clone(),
                COLOR_SINGLE.stroke_width(2),
            ))
            .unwrap()
            .label("p99 Latency")
            .legend(move |(x, y)| {
                PathElement::new(vec![(x, y), (x + 20, y)], COLOR_SINGLE.stroke_width(2))
            });
        chart
            .draw_secondary_series(
                lat_points
                    .iter()
                    .map(|&(x, y)| Circle::new((x, y), 4, COLOR_SINGLE.filled())),
            )
            .unwrap();

        chart
            .configure_series_labels()
            .position(SeriesLabelPosition::UpperLeft)
            .background_style(WHITE.mix(0.8))
            .border_style(BLACK.mix(0.3))
            .draw()
            .unwrap();

        // We don't draw on the right area — it's just padding for the secondary axis.
        let _ = right;
    });
}

// =====================================================================
// Command: stability — latency stability over time
// =====================================================================

/// Time-series data embedded in the JSON output.
#[derive(Debug, Deserialize)]
struct TimeSeriesResult {
    #[allow(dead_code)]
    label: String,
    throughput_ops: f64,
    #[serde(default)]
    time_series: Vec<TimeSeriesPoint>,
}

/// Deserialized from JSON time-series output; fields exist for schema
/// compatibility even when not all are plotted.
#[derive(Debug, Deserialize)]
#[allow(dead_code)]
struct TimeSeriesPoint {
    elapsed_secs: f64,
    p99_us: f64,
    p999_us: f64,
    p9999_us: f64,
}

fn cmd_stability(args: &[String]) {
    let opts = parse_args(args, "stability.svg");
    if opts.files.is_empty() {
        eprintln!("error: need at least 1 result file with time_series data");
        std::process::exit(1);
    }

    let mut all_results = Vec::new();
    for path in &opts.files {
        let data = fs::read_to_string(path).unwrap_or_else(|e| {
            eprintln!("warning: cannot read {}: {}", path.display(), e);
            String::new()
        });
        if data.is_empty() {
            continue;
        }
        match serde_json::from_str::<TimeSeriesResult>(&data) {
            Ok(r) if !r.time_series.is_empty() => {
                let filename = path
                    .file_name()
                    .map(|f| f.to_string_lossy().into_owned())
                    .unwrap_or_default();
                all_results.push((r, filename));
            }
            Ok(_) => eprintln!(
                "warning: {} has no time_series data, skipping",
                path.display()
            ),
            Err(e) => eprintln!("warning: cannot parse {}: {}", path.display(), e),
        }
    }

    if all_results.is_empty() {
        eprintln!("error: no files with time_series data found");
        std::process::exit(1);
    }

    plot_stability(&all_results, &opts.output);
}

fn plot_stability(results: &[(TimeSeriesResult, String)], output: &PathBuf) {
    let max_time = results
        .iter()
        .flat_map(|(r, _)| r.time_series.iter().map(|p| p.elapsed_secs))
        .fold(0.0f64, f64::max)
        * 1.05;

    let max_lat = results
        .iter()
        .flat_map(|(r, _)| r.time_series.iter().map(|p| p.p9999_us))
        .fold(0.0f64, f64::max)
        * 1.2;

    render_both!(output, (900, 500), |root| {
        root.fill(&WHITE).unwrap();

        let mut chart = ChartBuilder::on(&root)
            .caption(
                "Latency Stability Over Time",
                ("sans-serif", 22).into_font(),
            )
            .margin(15)
            .x_label_area_size(50)
            .y_label_area_size(70)
            .build_cartesian_2d(0.0..max_time, 0.0..max_lat)
            .unwrap();

        chart
            .configure_mesh()
            .x_desc("Time (seconds)")
            .y_desc("Latency (µs)")
            .x_label_formatter(&|v| format!("{:.0}s", v))
            .y_label_formatter(&|v| {
                if *v >= 1000.0 {
                    format!("{:.1}ms", v / 1000.0)
                } else {
                    format!("{:.0}µs", v)
                }
            })
            .draw()
            .unwrap();

        for (i, (result, filename)) in results.iter().enumerate() {
            let color = COLORS[i % COLORS.len()];
            let mode = mode_from_filename(filename);
            let tp = format_throughput(result.throughput_ops);

            // p99.99 line — fewer points than p99.9, easier to read.
            let p9999_points: Vec<(f64, f64)> = result
                .time_series
                .iter()
                .map(|p| (p.elapsed_secs, p.p9999_us))
                .collect();
            chart
                .draw_series(LineSeries::new(p9999_points, color.stroke_width(2)))
                .unwrap()
                .label(format!("{mode} p99.99 ({tp})"))
                .legend(move |(x, y)| {
                    PathElement::new(vec![(x, y), (x + 20, y)], color.stroke_width(2))
                });
        }

        chart
            .configure_series_labels()
            .position(SeriesLabelPosition::UpperRight)
            .background_style(WHITE.mix(0.8))
            .border_style(BLACK.mix(0.3))
            .draw()
            .unwrap();
    });
}

// =====================================================================
// Command: health — server health metrics over time
// =====================================================================

fn cmd_health(args: &[String]) {
    let opts = parse_args(args, "health");
    if opts.files.is_empty() {
        eprintln!("error: need at least 1 result file with health data");
        std::process::exit(1);
    }

    let mut all_results = Vec::new();
    for path in &opts.files {
        let data = fs::read_to_string(path).unwrap_or_else(|e| {
            eprintln!("warning: cannot read {}: {}", path.display(), e);
            String::new()
        });
        if data.is_empty() {
            continue;
        }
        match serde_json::from_str::<HealthResult>(&data) {
            Ok(r) if !r.health.is_empty() => {
                let filename = path
                    .file_name()
                    .map(|f| f.to_string_lossy().into_owned())
                    .unwrap_or_default();
                warn_unknown_metrics(&r, &filename);
                all_results.push((r, filename));
            }
            Ok(_) => eprintln!("warning: {} has no health data, skipping", path.display()),
            Err(e) => eprintln!("warning: cannot parse {}: {}", path.display(), e),
        }
    }

    if all_results.is_empty() {
        eprintln!("error: no files with health data found");
        std::process::exit(1);
    }

    // Generate one SVG per metric. The output path is used as a stem:
    // "-o health" → "health-queue-depth.svg", "health-events.svg", etc.
    let stem = opts.output.to_string_lossy().to_string();
    plot_health_queue_depth(
        &all_results,
        &PathBuf::from(format!("{stem}-queue-depth.svg")),
    );
    plot_health_events(&all_results, &PathBuf::from(format!("{stem}-events.svg")));
    plot_health_journal_seq(
        &all_results,
        &PathBuf::from(format!("{stem}-journal-seq.svg")),
    );

    // Replication lag plot — only if any file has non-zero lag.
    let has_repl_lag = all_results
        .iter()
        .any(|(r, _)| r.health.iter().any(|h| h.replication_lag > 0));
    if has_repl_lag {
        plot_health_replication_lag(
            &all_results,
            &PathBuf::from(format!("{stem}-replication-lag.svg")),
        );
    }

    emit_health_metric_plots(&all_results, &stem);
}

/// One entry in the per-metric plot catalog: filename suffix, the metric
/// series to render, per-series legend labels, chart title and y-axis.
struct PlotSpec {
    suffix: &'static str,
    metrics: &'static [HealthMetric],
    labels: &'static [&'static str],
    title: &'static str,
    y_label: &'static str,
}

/// Render every typed `HealthMetric` over time, plus the derived
/// utilization chart. Called from both `cmd_health` and `cmd_all` so the
/// two paths can't drift.
fn emit_health_metric_plots(results: &[(HealthResult, String)], stem: &str) {
    use HealthMetric::*;
    let specs: &[PlotSpec] = &[
        PlotSpec {
            suffix: "replica-lag",
            metrics: &[ReplicaLagSlot0, ReplicaLagSlot1],
            labels: &["slot 0", "slot 1"],
            title: "Per-Replica Lag Over Time",
            y_label: "Lag (events)",
        },
        PlotSpec {
            suffix: "replica-ack-latency",
            metrics: &[ReplicaAckLatencyUsSlot0, ReplicaAckLatencyUsSlot1],
            labels: &["slot 0", "slot 1"],
            title: "Replica Ack Latency Over Time",
            y_label: "Latency (\u{00b5}s)",
        },
        PlotSpec {
            suffix: "replica-bytes-sent",
            metrics: &[ReplicaBytesSentTotalSlot0, ReplicaBytesSentTotalSlot1],
            labels: &["slot 0", "slot 1"],
            title: "Cumulative Bytes Sent Per Replica",
            y_label: "Bytes",
        },
        PlotSpec {
            suffix: "replication-ring-depth",
            metrics: &[ReplicationRingDepthSlot0, ReplicationRingDepthSlot1],
            labels: &["slot 0", "slot 1"],
            title: "Replication Ring Depth Over Time",
            y_label: "Depth (batches)",
        },
        PlotSpec {
            suffix: "fastest-replica-cursor",
            metrics: &[FastestReplicaCursor],
            labels: &["fastest"],
            title: "Fastest-Replica Cursor Over Time",
            y_label: "Sequence",
        },
        PlotSpec {
            suffix: "replica-evictions",
            metrics: &[ReplicaEvictionsTotal],
            labels: &["evictions"],
            title: "Replica Evictions (cumulative)",
            y_label: "Evictions",
        },
        PlotSpec {
            suffix: "replicas-connected",
            metrics: &[ReplicasConnected],
            labels: &["connected"],
            title: "Replicas Connected Over Time",
            y_label: "Count",
        },
        PlotSpec {
            suffix: "active-connections",
            metrics: &[ActiveConnections],
            labels: &["connections"],
            title: "Authenticated Client Connections Over Time",
            y_label: "Count",
        },
        PlotSpec {
            suffix: "pipeline-healthy",
            metrics: &[PipelineHealthy],
            labels: &["healthy"],
            title: "Pipeline Healthy (1 = healthy, 0 = degraded)",
            y_label: "Status",
        },
        PlotSpec {
            suffix: "trading-active",
            metrics: &[TradingActive],
            labels: &["trading"],
            title: "Trading Active (1 = accepting orders, 0 = halted)",
            y_label: "Status",
        },
        PlotSpec {
            suffix: "replica-acked-sequence",
            metrics: &[ReplicaAckedSequenceSlot0, ReplicaAckedSequenceSlot1],
            labels: &["slot 0", "slot 1"],
            title: "Per-Replica Acked Sequence (journal-persisted)",
            y_label: "Sequence",
        },
        PlotSpec {
            suffix: "replica-in-memory-sequence",
            metrics: &[ReplicaInMemorySequenceSlot0, ReplicaInMemorySequenceSlot1],
            labels: &["slot 0", "slot 1"],
            title: "Per-Replica In-Memory Sequence (pre-journal)",
            y_label: "Sequence",
        },
        PlotSpec {
            suffix: "replica-catching-up",
            metrics: &[ReplicaCatchingUpSlot0, ReplicaCatchingUpSlot1],
            labels: &["slot 0", "slot 1"],
            title: "Per-Replica Catching-Up Status (1 = replaying from journal)",
            y_label: "Status",
        },
        PlotSpec {
            suffix: "durability-degraded",
            metrics: &[DurabilityPolicyDegraded],
            labels: &["degraded"],
            title: "Durability Policy Degraded (1 = clamped below target node count)",
            y_label: "Status",
        },
        PlotSpec {
            suffix: "response-gate",
            metrics: &[ResponseGateTotalJournal, ResponseGateTotalReplication],
            labels: &["journal", "replication"],
            title: "Response Gate Bottleneck (cumulative events)",
            y_label: "Events",
        },
    ];

    for spec in specs {
        plot_health_extra_metric(
            results,
            &PathBuf::from(format!("{stem}-{}.svg", spec.suffix)),
            spec.metrics,
            spec.labels,
            spec.title,
            spec.y_label,
        );
    }

    // Derived pipeline-stage utilization from the busy/idle counters.
    plot_health_utilization(results, &PathBuf::from(format!("{stem}-utilization.svg")));
}

/// Plot input queue depth over time — the primary saturation indicator.
/// Shows depth as a percentage of capacity when capacity is available.
fn plot_health_queue_depth(results: &[(HealthResult, String)], output: &PathBuf) {
    let max_time = health_max_time(results) * 1.05;
    // Use capacity from the first sample as y-axis max when available.
    let capacity = results
        .iter()
        .flat_map(|(r, _)| r.health.iter().map(|h| h.input_queue_capacity))
        .max()
        .unwrap_or(0);
    let max_depth = if capacity > 0 {
        capacity as f64
    } else {
        results
            .iter()
            .flat_map(|(r, _)| {
                r.health
                    .iter()
                    .map(|h| HealthMetric::InputQueueDepth.read(h))
            })
            .fold(1.0f64, f64::max)
            * 1.2
    };

    render_both!(output, (900, 500), |root| {
        root.fill(&WHITE).unwrap();

        let mut chart = ChartBuilder::on(&root)
            .caption(
                "Input Queue Depth Over Time",
                ("sans-serif", 22).into_font(),
            )
            .margin(15)
            .x_label_area_size(50)
            .y_label_area_size(80)
            .build_cartesian_2d(0.0..max_time, 0.0..max_depth)
            .unwrap();

        chart
            .configure_mesh()
            .x_desc("Time (seconds)")
            .y_desc("Queue Depth")
            .x_label_formatter(&|v| format!("{:.0}s", v))
            .y_label_formatter(&|v| {
                if capacity > 0 {
                    format!("{:.1}%", v / capacity as f64 * 100.0)
                } else {
                    format_large_number(*v)
                }
            })
            .draw()
            .unwrap();

        for (i, (result, filename)) in results.iter().enumerate() {
            let color = COLORS[i % COLORS.len()];
            let mode = mode_from_filename(filename);
            let tp = format_throughput(result.throughput_ops);

            let points: Vec<(f64, f64)> = result
                .health
                .iter()
                .map(|h| (h.elapsed_secs, HealthMetric::InputQueueDepth.read(h)))
                .collect();

            chart
                .draw_series(LineSeries::new(points, color.stroke_width(2)))
                .unwrap()
                .label(format!("{mode} ({tp})"))
                .legend(move |(x, y)| {
                    PathElement::new(vec![(x, y), (x + 20, y)], color.stroke_width(2))
                });
        }

        chart
            .configure_series_labels()
            .position(SeriesLabelPosition::UpperLeft)
            .background_style(WHITE.mix(0.8))
            .border_style(BLACK.mix(0.3))
            .draw()
            .unwrap();
    });
}

/// Plot events processed over time — shows throughput ramp-up and steady state.
fn plot_health_events(results: &[(HealthResult, String)], output: &PathBuf) {
    let max_time = health_max_time(results) * 1.05;
    let max_events = results
        .iter()
        .flat_map(|(r, _)| {
            r.health
                .iter()
                .map(|h| HealthMetric::EventsProcessed.read(h))
        })
        .fold(1.0f64, f64::max)
        * 1.1;

    render_both!(output, (900, 500), |root| {
        root.fill(&WHITE).unwrap();

        let mut chart = ChartBuilder::on(&root)
            .caption("Events Processed Over Time", ("sans-serif", 22).into_font())
            .margin(15)
            .x_label_area_size(50)
            .y_label_area_size(80)
            .build_cartesian_2d(0.0..max_time, 0.0..max_events)
            .unwrap();

        chart
            .configure_mesh()
            .x_desc("Time (seconds)")
            .y_desc("Events Processed")
            .x_label_formatter(&|v| format!("{:.0}s", v))
            .y_label_formatter(&|v| format_large_number(*v))
            .draw()
            .unwrap();

        for (i, (result, filename)) in results.iter().enumerate() {
            let color = COLORS[i % COLORS.len()];
            let mode = mode_from_filename(filename);
            let tp = format_throughput(result.throughput_ops);

            let points: Vec<(f64, f64)> = result
                .health
                .iter()
                .map(|h| (h.elapsed_secs, HealthMetric::EventsProcessed.read(h)))
                .collect();

            chart
                .draw_series(LineSeries::new(points, color.stroke_width(2)))
                .unwrap()
                .label(format!("{mode} ({tp})"))
                .legend(move |(x, y)| {
                    PathElement::new(vec![(x, y), (x + 20, y)], color.stroke_width(2))
                });
        }

        chart
            .configure_series_labels()
            .position(SeriesLabelPosition::UpperLeft)
            .background_style(WHITE.mix(0.8))
            .border_style(BLACK.mix(0.3))
            .draw()
            .unwrap();
    });
}

/// Plot journal sequence over time — shows durability progress and fsync cadence.
fn plot_health_journal_seq(results: &[(HealthResult, String)], output: &PathBuf) {
    let max_time = health_max_time(results) * 1.05;
    let max_seq = results
        .iter()
        .flat_map(|(r, _)| {
            r.health
                .iter()
                .map(|h| HealthMetric::JournalSequence.read(h))
        })
        .fold(1.0f64, f64::max)
        * 1.1;

    render_both!(output, (900, 500), |root| {
        root.fill(&WHITE).unwrap();

        let mut chart = ChartBuilder::on(&root)
            .caption("Journal Sequence Over Time", ("sans-serif", 22).into_font())
            .margin(15)
            .x_label_area_size(50)
            .y_label_area_size(80)
            .build_cartesian_2d(0.0..max_time, 0.0..max_seq)
            .unwrap();

        chart
            .configure_mesh()
            .x_desc("Time (seconds)")
            .y_desc("Journal Sequence")
            .x_label_formatter(&|v| format!("{:.0}s", v))
            .y_label_formatter(&|v| format_large_number(*v))
            .draw()
            .unwrap();

        for (i, (result, filename)) in results.iter().enumerate() {
            let color = COLORS[i % COLORS.len()];
            let mode = mode_from_filename(filename);
            let tp = format_throughput(result.throughput_ops);

            let points: Vec<(f64, f64)> = result
                .health
                .iter()
                .map(|h| (h.elapsed_secs, HealthMetric::JournalSequence.read(h)))
                .collect();

            chart
                .draw_series(LineSeries::new(points, color.stroke_width(2)))
                .unwrap()
                .label(format!("{mode} ({tp})"))
                .legend(move |(x, y)| {
                    PathElement::new(vec![(x, y), (x + 20, y)], color.stroke_width(2))
                });
        }

        chart
            .configure_series_labels()
            .position(SeriesLabelPosition::UpperLeft)
            .background_style(WHITE.mix(0.8))
            .border_style(BLACK.mix(0.3))
            .draw()
            .unwrap();
    });
}

/// Plot replication lag over time — only generated when non-zero lag exists.
fn plot_health_replication_lag(results: &[(HealthResult, String)], output: &PathBuf) {
    let max_time = health_max_time(results) * 1.05;
    let max_lag = results
        .iter()
        .flat_map(|(r, _)| {
            r.health
                .iter()
                .map(|h| HealthMetric::ReplicationLag.read(h))
        })
        .fold(1.0f64, f64::max)
        * 1.2;

    render_both!(output, (900, 500), |root| {
        root.fill(&WHITE).unwrap();

        let mut chart = ChartBuilder::on(&root)
            .caption("Replication Lag Over Time", ("sans-serif", 22).into_font())
            .margin(15)
            .x_label_area_size(50)
            .y_label_area_size(80)
            .build_cartesian_2d(0.0..max_time, 0.0..max_lag)
            .unwrap();

        chart
            .configure_mesh()
            .x_desc("Time (seconds)")
            .y_desc("Replication Lag (events)")
            .x_label_formatter(&|v| format!("{:.0}s", v))
            .y_label_formatter(&|v| format_large_number(*v))
            .draw()
            .unwrap();

        for (i, (result, filename)) in results.iter().enumerate() {
            let color = COLORS[i % COLORS.len()];
            let mode = mode_from_filename(filename);
            let tp = format_throughput(result.throughput_ops);

            let points: Vec<(f64, f64)> = result
                .health
                .iter()
                .map(|h| (h.elapsed_secs, HealthMetric::ReplicationLag.read(h)))
                .collect();

            chart
                .draw_series(LineSeries::new(points, color.stroke_width(2)))
                .unwrap()
                .label(format!("{mode} ({tp})"))
                .legend(move |(x, y)| {
                    PathElement::new(vec![(x, y), (x + 20, y)], color.stroke_width(2))
                });
        }

        chart
            .configure_series_labels()
            .position(SeriesLabelPosition::UpperLeft)
            .background_style(WHITE.mix(0.8))
            .border_style(BLACK.mix(0.3))
            .draw()
            .unwrap();
    });
}

/// Plot one or more `HealthMetric` series over time. Each series renders as
/// a line per result file. The plot is emitted whenever at least one input
/// file has health samples — even when every sample reads zero, since a
/// flat-zero line is meaningful for binary status metrics ("never degraded",
/// "never evicted") and counter metrics that didn't fire.
fn plot_health_extra_metric(
    results: &[(HealthResult, String)],
    output: &PathBuf,
    metrics: &[HealthMetric],
    series_labels: &[&str], // e.g. ["slot 0", "slot 1"]
    title: &str,
    y_label: &str,
) {
    // Skip only when literally no samples exist — the chart would have
    // nothing to draw on the x-axis.
    if !results.iter().any(|(r, _)| !r.health.is_empty()) {
        return;
    }

    let max_time = health_max_time(results) * 1.05;
    let max_val = results
        .iter()
        .flat_map(|(r, _)| {
            r.health
                .iter()
                .flat_map(|h| metrics.iter().map(|m| m.read(h)))
        })
        .fold(1.0f64, f64::max)
        * 1.2;

    render_both!(output, (900, 500), |root| {
        root.fill(&WHITE).unwrap();

        let mut chart = ChartBuilder::on(&root)
            .caption(title, ("sans-serif", 22).into_font())
            .margin(15)
            .x_label_area_size(50)
            .y_label_area_size(80)
            .build_cartesian_2d(0.0..max_time, 0.0..max_val)
            .unwrap();

        chart
            .configure_mesh()
            .x_desc("Time (seconds)")
            .y_desc(y_label)
            .x_label_formatter(&|v| format!("{:.0}s", v))
            .y_label_formatter(&|v| format_large_number(*v))
            .draw()
            .unwrap();

        for (i, (result, filename)) in results.iter().enumerate() {
            let mode = mode_from_filename(filename);
            let tp = format_throughput(result.throughput_ops);

            for (slot, metric) in metrics.iter().enumerate() {
                let color_idx = i * metrics.len() + slot;
                let color = COLORS[color_idx % COLORS.len()];
                let label = series_labels.get(slot).unwrap_or(&"");

                let points: Vec<(f64, f64)> = result
                    .health
                    .iter()
                    .map(|h| (h.elapsed_secs, metric.read(h)))
                    .collect();

                chart
                    .draw_series(LineSeries::new(points, color.stroke_width(2)))
                    .unwrap()
                    .label(format!("{mode} {label} ({tp})"))
                    .legend(move |(x, y)| {
                        PathElement::new(vec![(x, y), (x + 20, y)], color.stroke_width(2))
                    });
            }
        }

        chart
            .configure_series_labels()
            .position(SeriesLabelPosition::UpperLeft)
            .background_style(WHITE.mix(0.8))
            .border_style(BLACK.mix(0.3))
            .draw()
            .unwrap();
    });
}

/// Plot per-stage utilization over time. Computes `delta(busy) / delta(busy+idle)`
/// between successive health samples to derive instantaneous utilization percentage.
fn plot_health_utilization(results: &[(HealthResult, String)], output: &PathBuf) {
    const STAGES: [&str; 3] = ["journal", "matching", "response"];
    const BUSY: [HealthMetric; 3] = [
        HealthMetric::StageBusyTotalJournal,
        HealthMetric::StageBusyTotalMatching,
        HealthMetric::StageBusyTotalResponse,
    ];
    const IDLE: [HealthMetric; 3] = [
        HealthMetric::StageIdleTotalJournal,
        HealthMetric::StageIdleTotalMatching,
        HealthMetric::StageIdleTotalResponse,
    ];

    // Skip only when no sample saw any busy-counter tick: with zero
    // activity there is no utilization curve to draw.
    let has_data = results.iter().any(|(r, _)| {
        r.health
            .iter()
            .any(|h| BUSY.iter().any(|m| m.read(h) > 0.0))
    });
    if !has_data {
        return;
    }

    let max_time = health_max_time(results) * 1.05;

    render_both!(output, (900, 500), |root| {
        root.fill(&WHITE).unwrap();

        let mut chart = ChartBuilder::on(&root)
            .caption("Pipeline Stage Utilization", ("sans-serif", 22).into_font())
            .margin(15)
            .x_label_area_size(50)
            .y_label_area_size(70)
            .build_cartesian_2d(0.0..max_time, 0.0..105.0)
            .unwrap();

        chart
            .configure_mesh()
            .x_desc("Time (seconds)")
            .y_desc("Utilization %")
            .x_label_formatter(&|v| format!("{:.0}s", v))
            .y_label_formatter(&|v| format!("{:.0}%", v))
            .draw()
            .unwrap();

        for (i, (result, filename)) in results.iter().enumerate() {
            let mode = mode_from_filename(filename);
            let tp = format_throughput(result.throughput_ops);

            for (stage_idx, stage_name) in STAGES.iter().enumerate() {
                let color_idx = i * STAGES.len() + stage_idx;
                let color = COLORS[color_idx % COLORS.len()];

                // Compute instantaneous utilization from delta of counters.
                let points: Vec<(f64, f64)> = result
                    .health
                    .windows(2)
                    .filter_map(|pair| {
                        let (prev, cur) = (&pair[0], &pair[1]);
                        let d_busy = BUSY[stage_idx].read(cur) - BUSY[stage_idx].read(prev);
                        let d_idle = IDLE[stage_idx].read(cur) - IDLE[stage_idx].read(prev);
                        let total = d_busy + d_idle;
                        if total > 0.0 {
                            Some((cur.elapsed_secs, d_busy / total * 100.0))
                        } else {
                            None
                        }
                    })
                    .collect();

                if !points.is_empty() {
                    chart
                        .draw_series(LineSeries::new(points, color.stroke_width(2)))
                        .unwrap()
                        .label(format!("{mode} {stage_name} ({tp})"))
                        .legend(move |(x, y)| {
                            PathElement::new(vec![(x, y), (x + 20, y)], color.stroke_width(2))
                        });
                }
            }
        }

        chart
            .configure_series_labels()
            .position(SeriesLabelPosition::UpperRight)
            .background_style(WHITE.mix(0.8))
            .border_style(BLACK.mix(0.3))
            .draw()
            .unwrap();
    });
}

/// Maximum elapsed time across all health samples.
fn health_max_time(results: &[(HealthResult, String)]) -> f64 {
    results
        .iter()
        .flat_map(|(r, _)| r.health.iter().map(|h| h.elapsed_secs))
        .fold(0.0f64, f64::max)
}

/// Format a large number compactly for Y-axis labels: 1.5M, 100K, 42.
fn format_large_number(v: f64) -> String {
    if v >= 1_000_000.0 {
        format!("{:.1}M", v / 1_000_000.0)
    } else if v >= 1_000.0 {
        format!("{:.0}K", v / 1_000.0)
    } else {
        format!("{:.0}", v)
    }
}

// =====================================================================
// Command: pipeline
// =====================================================================

fn cmd_pipeline(args: &[String]) {
    let opts = parse_args(args, "pipeline.svg");

    // Try loading stats from --stats file or from JSON files.
    let stages = if let Some(ref stats_path) = opts.stats_file {
        parse_pipeline_stats(&fs::read_to_string(stats_path).unwrap_or_default())
    } else {
        eprintln!("error: --stats <log-file> required for pipeline command");
        std::process::exit(1);
    };

    if stages.is_empty() {
        eprintln!("error: no pipeline stats found in log file");
        std::process::exit(1);
    }

    plot_pipeline(&stages, &opts.output);
}

fn parse_pipeline_stats(log: &str) -> Vec<PipelineStage> {
    let mut stages = Vec::new();
    for line in log.lines() {
        // Match: stage="matching" pct_busy=18.48%
        if let Some(stage_start) = line.find("stage=\"") {
            let rest = &line[stage_start + 7..];
            if let Some(end) = rest.find('"') {
                let name = rest[..end].to_string();
                if let Some(pct_start) = line.find("pct_busy=") {
                    let pct_rest = &line[pct_start + 9..];
                    if let Some(pct_end) = pct_rest.find('%')
                        && let Ok(pct) = pct_rest[..pct_end].parse::<f64>()
                    {
                        stages.push(PipelineStage {
                            name,
                            pct_busy: pct,
                        });
                    }
                }
            }
        }
    }
    stages
}

fn plot_pipeline(stages: &[PipelineStage], output: &PathBuf) {
    let max_pct = stages
        .iter()
        .map(|s| s.pct_busy)
        .fold(0.0f64, f64::max)
        .max(10.0)
        * 1.3;

    render_both!(output, (700, 400), |root| {
        root.fill(&WHITE).unwrap();

        let mut chart = ChartBuilder::on(&root)
            .caption("Pipeline Stage Utilization", ("sans-serif", 22).into_font())
            .margin(15)
            .x_label_area_size(50)
            .y_label_area_size(70)
            .build_cartesian_2d(0usize..stages.len(), 0.0..max_pct)
            .unwrap();

        chart
            .configure_mesh()
            .x_desc("Stage")
            .y_desc("Busy %")
            .x_label_formatter(&|idx| stages.get(*idx).map(|s| s.name.clone()).unwrap_or_default())
            .y_label_formatter(&|v| format!("{:.0}%", v))
            .draw()
            .unwrap();

        chart
            .draw_series(stages.iter().enumerate().map(|(i, stage)| {
                let color = COLORS[i % COLORS.len()];
                let x0 = i;
                let x1 = i + 1;
                Rectangle::new([(x0, 0.0), (x1, stage.pct_busy)], color.filled())
            }))
            .unwrap();

        // Value labels on top of bars.
        chart
            .draw_series(stages.iter().enumerate().map(|(i, stage)| {
                Text::new(
                    format!("{:.1}%", stage.pct_busy),
                    (i, stage.pct_busy + max_pct * 0.03),
                    ("sans-serif", 14).into_font(),
                )
            }))
            .unwrap();
    });
}

// =====================================================================
// Command: all — generate all plots from a results directory
// =====================================================================

fn cmd_all(args: &[String]) {
    let opts = parse_args(args, ".");

    let dir = if opts.files.is_empty() {
        PathBuf::from(".")
    } else {
        opts.files[0].clone()
    };

    // Find all JSON files in the directory.
    let mut json_files: Vec<PathBuf> = Vec::new();
    if let Ok(entries) = fs::read_dir(&dir) {
        for entry in entries.flatten() {
            let path = entry.path();
            if path.extension().is_some_and(|e| e == "json") {
                json_files.push(path);
            }
        }
    }
    json_files.sort();

    if json_files.is_empty() {
        eprintln!("error: no JSON files found in {}", dir.display());
        std::process::exit(1);
    }

    let results = load_results(&json_files);
    if results.is_empty() {
        eprintln!("error: no valid results loaded");
        std::process::exit(1);
    }

    // 1. Latency CDF (plot_latency_cdf filters out < 100K ops/s internally).
    let cdf_path = dir.join("latency-cdf.svg");
    plot_latency_cdf(&results, &cdf_path);

    // 2. Saturation curve (only if we have multiple load levels).
    if results.len() >= 2 {
        let sat_path = dir.join("saturation.svg");
        plot_saturation(&results, &sat_path);
    } else {
        eprintln!("skipping saturation plot (need >= 2 result files)");
    }

    // 3. Health metrics (one plot per metric for any file with health data).
    let mut health_results = Vec::new();
    for path in &json_files {
        let data = fs::read_to_string(path).unwrap_or_default();
        if let Ok(r) = serde_json::from_str::<HealthResult>(&data)
            && !r.health.is_empty()
        {
            let filename = path
                .file_name()
                .map(|f| f.to_string_lossy().into_owned())
                .unwrap_or_default();
            warn_unknown_metrics(&r, &filename);
            health_results.push((r, filename));
        }
    }
    if !health_results.is_empty() {
        let stem = dir.join("health").to_string_lossy().to_string();
        plot_health_queue_depth(
            &health_results,
            &PathBuf::from(format!("{stem}-queue-depth.svg")),
        );
        plot_health_events(
            &health_results,
            &PathBuf::from(format!("{stem}-events.svg")),
        );
        plot_health_journal_seq(
            &health_results,
            &PathBuf::from(format!("{stem}-journal-seq.svg")),
        );
        let has_repl_lag = health_results
            .iter()
            .any(|(r, _)| r.health.iter().any(|h| h.replication_lag > 0));
        if has_repl_lag {
            plot_health_replication_lag(
                &health_results,
                &PathBuf::from(format!("{stem}-replication-lag.svg")),
            );
        }
        emit_health_metric_plots(&health_results, &stem);
    } else {
        eprintln!("skipping health plots (no files with health data)");
    }

    // 4. Pipeline breakdown (look for server log).
    let log_candidates = ["server.log", "melin-server.log"];
    for log_name in &log_candidates {
        let log_path = dir.join(log_name);
        if log_path.exists() {
            let stages = parse_pipeline_stats(&fs::read_to_string(&log_path).unwrap_or_default());
            if !stages.is_empty() {
                let pipe_path = dir.join("pipeline.svg");
                plot_pipeline(&stages, &pipe_path);
            }
            break;
        }
    }

    eprintln!("all plots generated in {}", dir.display());
}

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

    fn sample_health_json() -> &'static str {
        // Mirrors the JSON layout written by bench/src/main.rs — typed fields
        // for fixed metrics plus sanitized Prometheus labels for the rest.
        r#"{
            "label": "fsync",
            "throughput_ops": 100000.0,
            "health": [{
                "elapsed_secs": 1.5,
                "active_connections": 8,
                "events_processed": 50000,
                "journal_sequence": 50000,
                "replication_lag": 0,
                "input_queue_depth": 128,
                "input_queue_capacity": 1048576,
                "pipeline_healthy": true,
                "trading_active": true,
                "melin_replicas_connected": 2,
                "melin_replica_lag_slot_0": 12,
                "melin_replica_lag_slot_1": 7,
                "melin_replica_evictions_total": 3,
                "melin_fastest_replica_cursor": 49993,
                "melin_stage_busy_total_stage_journal": 999,
                "melin_stage_idle_total_stage_journal": 1,
                "melin_future_metric": 42
            }]
        }"#
    }

    #[test]
    fn deserializes_typed_fields_and_captures_unknowns() {
        let r: HealthResult = serde_json::from_str(sample_health_json()).expect("parse");
        let h = &r.health[0];
        assert_eq!(h.active_connections, 8);
        assert_eq!(h.replicas_connected, 2);
        assert_eq!(h.replica_lag_slot_0, 12);
        assert_eq!(h.replica_lag_slot_1, 7);
        assert_eq!(h.replica_evictions_total, 3);
        assert_eq!(h.stage_busy_total_journal, 999);
        // Metrics without a typed field land in `unknown` for warn-time
        // visibility — they are not silently dropped.
        assert!(
            h.unknown.contains_key("melin_future_metric"),
            "unknown map should capture metrics absent from HealthPoint"
        );
        assert_eq!(h.unknown.len(), 1);
    }

    /// Exercise every `HealthMetric` variant. A recursive `.read()` arm
    /// would stack-overflow this test before producing a value, and a
    /// new variant added without a `match` arm would fail to compile —
    /// together these guarantee every metric is reachable and terminates.
    #[test]
    fn health_metric_read_covers_every_variant() {
        // JSON encoded so every typed field gets a distinct sentinel value;
        // we then assert read() returns that exact value per variant.
        let json = r#"{"label":"x","throughput_ops":0.0,"health":[{
            "elapsed_secs": 0.0,
            "active_connections": 1,
            "events_processed": 2,
            "journal_sequence": 3,
            "replication_lag": 4,
            "input_queue_depth": 5,
            "input_queue_capacity": 6,
            "pipeline_healthy": true,
            "trading_active": false,
            "melin_replicas_connected": 7,
            "melin_replica_acked_sequence_slot_0": 8,
            "melin_replica_acked_sequence_slot_1": 9,
            "melin_replica_in_memory_sequence_slot_0": 10,
            "melin_replica_in_memory_sequence_slot_1": 11,
            "melin_replica_lag_slot_0": 12,
            "melin_replica_lag_slot_1": 13,
            "melin_replica_bytes_sent_total_slot_0": 14,
            "melin_replica_bytes_sent_total_slot_1": 15,
            "melin_replica_ack_latency_us_slot_0": 16,
            "melin_replica_ack_latency_us_slot_1": 17,
            "melin_replica_catching_up_slot_0": 1,
            "melin_replica_catching_up_slot_1": 0,
            "melin_replica_evictions_total": 18,
            "melin_replication_ring_depth_slot_0": 19,
            "melin_replication_ring_depth_slot_1": 20,
            "melin_fastest_replica_cursor": 21,
            "melin_stage_busy_total_stage_journal": 22,
            "melin_stage_busy_total_stage_matching": 23,
            "melin_stage_busy_total_stage_response": 24,
            "melin_stage_idle_total_stage_journal": 25,
            "melin_stage_idle_total_stage_matching": 26,
            "melin_stage_idle_total_stage_response": 27,
            "melin_response_gate_total_blocker_journal": 28,
            "melin_response_gate_total_blocker_replication": 29,
            "melin_durability_policy_degraded": 1
        }]}"#;
        let r: HealthResult = serde_json::from_str(json).expect("parse");
        let h = &r.health[0];

        // Pair every variant with its expected value. Adding a variant
        // without a row here is a non-exhaustive match in the body, which
        // is fine — but the variant won't be tested. Keep the list aligned
        // with HealthMetric.
        let expected: &[(HealthMetric, f64)] = &[
            (HealthMetric::ActiveConnections, 1.0),
            (HealthMetric::EventsProcessed, 2.0),
            (HealthMetric::JournalSequence, 3.0),
            (HealthMetric::ReplicationLag, 4.0),
            (HealthMetric::InputQueueDepth, 5.0),
            (HealthMetric::PipelineHealthy, 1.0),
            (HealthMetric::TradingActive, 0.0),
            (HealthMetric::ReplicasConnected, 7.0),
            (HealthMetric::ReplicaAckedSequenceSlot0, 8.0),
            (HealthMetric::ReplicaAckedSequenceSlot1, 9.0),
            (HealthMetric::ReplicaInMemorySequenceSlot0, 10.0),
            (HealthMetric::ReplicaInMemorySequenceSlot1, 11.0),
            (HealthMetric::ReplicaLagSlot0, 12.0),
            (HealthMetric::ReplicaLagSlot1, 13.0),
            (HealthMetric::ReplicaBytesSentTotalSlot0, 14.0),
            (HealthMetric::ReplicaBytesSentTotalSlot1, 15.0),
            (HealthMetric::ReplicaAckLatencyUsSlot0, 16.0),
            (HealthMetric::ReplicaAckLatencyUsSlot1, 17.0),
            (HealthMetric::ReplicaCatchingUpSlot0, 1.0),
            (HealthMetric::ReplicaCatchingUpSlot1, 0.0),
            (HealthMetric::ReplicaEvictionsTotal, 18.0),
            (HealthMetric::ReplicationRingDepthSlot0, 19.0),
            (HealthMetric::ReplicationRingDepthSlot1, 20.0),
            (HealthMetric::FastestReplicaCursor, 21.0),
            (HealthMetric::StageBusyTotalJournal, 22.0),
            (HealthMetric::StageBusyTotalMatching, 23.0),
            (HealthMetric::StageBusyTotalResponse, 24.0),
            (HealthMetric::StageIdleTotalJournal, 25.0),
            (HealthMetric::StageIdleTotalMatching, 26.0),
            (HealthMetric::StageIdleTotalResponse, 27.0),
            (HealthMetric::ResponseGateTotalJournal, 28.0),
            (HealthMetric::ResponseGateTotalReplication, 29.0),
            (HealthMetric::DurabilityPolicyDegraded, 1.0),
        ];

        for (metric, want) in expected {
            let got = metric.read(h);
            assert_eq!(got, *want, "{metric:?} read {got}, expected {want}");
        }
    }

    #[test]
    fn warn_unknown_metrics_is_silent_when_all_typed() {
        // Build a HealthPoint via deserialize with no unknown keys.
        let json = r#"{"label":"x","throughput_ops":0.0,"health":[{
            "elapsed_secs":0.0,"active_connections":0,"events_processed":0,
            "journal_sequence":0,"replication_lag":0,"input_queue_depth":0,
            "input_queue_capacity":0,"pipeline_healthy":false,"trading_active":false
        }]}"#;
        let r: HealthResult = serde_json::from_str(json).expect("parse");
        assert!(r.health[0].unknown.is_empty());
        // Function should run without panicking when there's nothing to warn about.
        warn_unknown_metrics(&r, "test.json");
    }
}
