all-smi 0.26.2

// Copyright 2025 Lablup Inc. and Jeongkyu Shin
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::collections::{HashMap, HashSet};

use crate::parsing::common::sanitize_label_value;
use chrono::Local;
use regex::Regex;

use crate::device::types::{GPU_METRIC_UNAVAILABLE, MAX_GPU_FAN_RPM};
use crate::device::{
    AppleSiliconCpuInfo, CpuInfo, CpuPlatformType, GpmMetrics, GpuInfo, MemoryInfo, MigGpuInfo,
    MigInstanceInfo, NvLinkRemoteDevice, NvLinkRemoteType, VgpuHostInfo, VgpuInfo,
};
use crate::storage::info::StorageInfo;

/// Structured return type for [`MetricsParser::parse_metrics`], replacing the
/// previous 6-tuple. Each field holds the parsed device records for a single
/// metric family.
#[derive(Debug, Default)]
pub struct ParsedMetrics {
    pub gpu_info: Vec<GpuInfo>,
    pub cpu_info: Vec<CpuInfo>,
    pub memory_info: Vec<MemoryInfo>,
    pub storage_info: Vec<StorageInfo>,
    pub vgpu_info: Vec<VgpuHostInfo>,
    pub mig_info: Vec<MigGpuInfo>,
    /// Per-process rows parsed from `all_smi_process_*` metric families on
    /// the remote side. Populated by [`MetricsParser::process_process_metrics`]
    /// and consumed by the cluster-wide Users tab aggregator (issue #189).
    /// Empty when the scraped host was not started with `--processes`.
    pub process_info: Vec<ParsedProcessRow>,
}

/// One row emitted by the remote metrics parser for each `(host, pid,
/// gpu_index)` triple. The UI aggregator (see
/// `src/ui/aggregation/user.rs`) groups these by `user` to build the
/// cluster-wide Users tab (issue #189).
///
/// Rows are keyed by `(host, pid, gpu_index)` — not by `pid` alone —
/// because:
/// 1. The same PID on two different hosts refers to two completely
///    different processes.
/// 2. A single GPU process on a multi-GPU host may appear once per GPU it
///    touches, and each appearance carries its own `gpu_index` /
///    `gpu_memory_bytes` readings.
///
/// When `user` is unknown (Windows API mode, scraping a host that did not
/// attribute the process) we render `?` in the UI and let the aggregator
/// group it under the synthetic "unattributed" user.
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct ParsedProcessRow {
    pub host: String,
    pub pid: u32,
    pub user: String,
    pub command: String,
    pub name: String,
    pub gpu_index: u32,
    pub gpu_uuid: String,
    pub gpu_memory_bytes: u64,
    /// CPU percent as an integer tenths-of-percent (e.g. `125` = 12.5 %).
    /// Keeping it as an integer avoids dragging an `f64` into the
    /// `PartialEq`/`Eq` bound on this struct — the downstream aggregator
    /// only uses this for display, never for ranking.
    pub cpu_pct_tenths: u32,
    /// Wall-clock seconds since the process started, mirrored from
    /// `all_smi_process_start_time_seconds`. 0 means "unknown" — the
    /// aggregator treats unknown as "youngest" so mixed fleets don't
    /// let unattributed processes win the "LONGEST" column.
    pub start_time_seconds: u64,
}

impl ParsedProcessRow {
    /// Convert a locally-collected [`crate::device::ProcessInfo`] into
    /// the remote-side row representation.  Used by `view --replay` so
    /// a recorded session (whose process rows are full `ProcessInfo`
    /// objects) flows through the cluster-wide Users tab on playback.
    pub fn from_local_process(process: &crate::device::ProcessInfo, host: &str) -> Self {
        // `start_time` is a HH:MM:SS-ish elapsed string in local mode.
        // Reuse the same parser the API exporter uses.
        let start_seconds =
            crate::api::metrics::process::ProcessMetricExporter::parse_start_time_seconds_public(
                &process.start_time,
            );
        let cpu_pct_tenths = (process.cpu_percent.max(0.0) * 10.0).round() as u32;
        Self {
            host: host.to_string(),
            pid: process.pid,
            user: process.user.clone(),
            command: process.command.clone(),
            name: process.process_name.clone(),
            gpu_index: process.device_id as u32,
            gpu_uuid: process.device_uuid.clone(),
            gpu_memory_bytes: process.used_memory,
            cpu_pct_tenths,
            start_time_seconds: start_seconds,
        }
    }
}

pub struct MetricsParser;

impl MetricsParser {
    pub fn new() -> Self {
        Self
    }

    pub fn parse_metrics(&self, text: &str, host: &str, re: &Regex) -> ParsedMetrics {
        // Limit the maximum size of HashMaps to prevent memory exhaustion
        const MAX_DEVICES_PER_TYPE: usize = 256;
        const MAX_TEXT_SIZE: usize = 10_485_760; // 10MB max input

        // Validate input size
        if text.len() > MAX_TEXT_SIZE {
            eprintln!(
                "Warning: Metrics text too large ({}), truncating to 10MB",
                text.len()
            );
            let truncated = &text[..MAX_TEXT_SIZE];
            return self.parse_metrics(truncated, host, re);
        }

        let mut gpu_info_map: HashMap<String, GpuInfo> = HashMap::with_capacity(16);
        let mut cpu_info_map: HashMap<String, CpuInfo> = HashMap::with_capacity(8);
        let mut memory_info_map: HashMap<String, MemoryInfo> = HashMap::with_capacity(8);
        let mut storage_info_map: HashMap<String, StorageInfo> = HashMap::with_capacity(32);
        // Keyed by (gpu_uuid, vgpu_id) for instance rows, and by (gpu_uuid, "__host__")
        // for host-scoped metrics.
        let mut vgpu_state = VgpuParseState::new();
        // MIG accumulator — keyed by gpu_uuid for parent host rows, and by
        // (gpu_uuid, mig_instance) for per-instance rows.
        let mut mig_state = MigParseState::new();
        // Per-process row accumulator (issue #189). Keyed by
        // `(pid, gpu_index)` so the three metric families (memory,
        // start-time, cpu-percent) collapse into a single row even though
        // they arrive on separate lines.
        let mut process_info_map: HashMap<(u32, u32), ParsedProcessRow> =
            HashMap::with_capacity(32);
        let mut host_instance_name: Option<String> = None;

        for line in text.lines() {
            if let Some((metric_name, labels_str, value)) = parse_prometheus!(line, re) {
                let labels = self.parse_labels(&labels_str);

                // Extract instance name from the first metric that has it
                if host_instance_name.is_none()
                    && let Some(instance) = labels.get("instance")
                {
                    host_instance_name = Some(instance.clone());
                }

                // Process different metric types with size limits.
                // Route vGPU and MIG lines first so they aren't swallowed by
                // the broader `gpu_` prefix below. `nvlink_*` metrics from
                // issue #132 are also hardware-detail lines that must join
                // the per-GPU accumulator — they carry the same GPU base
                // label set as the rest of the GPU rows. NPU families share
                // the same accumulator but may identify themselves with
                // either the GPU (`gpu_uuid`/`gpu_index`) or NPU
                // (`npu_uuid`/`npu_index`) label aliases; see
                // `process_gpu_metrics` for the fallback chain.
                if metric_name.starts_with("vgpu_") {
                    vgpu_state.process(&metric_name, &labels, value, host);
                } else if metric_name == "gpu_mig_mode" || metric_name.starts_with("mig_instance_")
                {
                    mig_state.process(&metric_name, &labels, value, host);
                } else if metric_name.starts_with("gpu_")
                    || metric_name.starts_with("npu_")
                    || metric_name.starts_with("nvlink_")
                    || metric_name == "ane_utilization"
                {
                    if gpu_info_map.len() < MAX_DEVICES_PER_TYPE {
                        self.process_gpu_metrics(
                            &mut gpu_info_map,
                            &metric_name,
                            &labels,
                            value,
                            host,
                        );
                    }
                } else if metric_name.starts_with("cpu_") {
                    if cpu_info_map.len() < MAX_DEVICES_PER_TYPE {
                        self.process_cpu_metrics(
                            &mut cpu_info_map,
                            &metric_name,
                            &labels,
                            value,
                            host,
                        );
                    }
                } else if metric_name.starts_with("memory_") || metric_name.starts_with("swap_") {
                    // Swap metrics share the same per-host accumulator as
                    // memory — they carry identical `instance`/`hostname`/`index`
                    // labels and are conceptually a property of the host's
                    // memory subsystem (issue #220). Routing them together
                    // keeps the `MemoryInfo` row populated consistently.
                    if memory_info_map.len() < MAX_DEVICES_PER_TYPE {
                        self.process_memory_metrics(
                            &mut memory_info_map,
                            &metric_name,
                            &labels,
                            value,
                            host,
                        );
                    }
                } else if (metric_name.starts_with("storage_") || metric_name.starts_with("disk_"))
                    && storage_info_map.len() < MAX_DEVICES_PER_TYPE
                {
                    self.process_storage_metrics(
                        &mut storage_info_map,
                        &metric_name,
                        &labels,
                        value,
                        host,
                    );
                } else if metric_name.starts_with("process_") {
                    // Cap process rows to keep a pathological scrape from
                    // turning into an OOM — 50 k rows is two orders of
                    // magnitude beyond any realistic node (the issue
                    // target is 100 nodes × 50 procs = 5 k).
                    const MAX_PROCESS_ROWS: usize = 50_000;
                    if process_info_map.len() < MAX_PROCESS_ROWS {
                        self.process_process_metrics(
                            &mut process_info_map,
                            &metric_name,
                            &labels,
                            value,
                            host,
                        );
                    }
                }
            }
        }

        // Store instance name in detail field if available, but keep host as the key
        if let Some(instance_name) = host_instance_name {
            self.update_instance_names(
                &mut gpu_info_map,
                &mut cpu_info_map,
                &mut memory_info_map,
                &mut storage_info_map,
                &instance_name,
            );
        }

        ParsedMetrics {
            gpu_info: gpu_info_map.into_values().collect(),
            cpu_info: cpu_info_map.into_values().collect(),
            memory_info: memory_info_map.into_values().collect(),
            storage_info: storage_info_map.into_values().collect(),
            vgpu_info: vgpu_state.finish(),
            mig_info: mig_state.finish(),
            process_info: process_info_map.into_values().collect(),
        }
    }

    /// Absorb a single `all_smi_process_*` metric line into the per-PID
    /// accumulator. Three families are recognised:
    /// - `process_memory_used_bytes` (gauge, bytes)
    /// - `process_start_time_seconds` (gauge, seconds since start)
    /// - `process_cpu_percent` (gauge, %)
    ///
    /// Each family shares the same label set — `pid`, `name`, `user`,
    /// `device_id`, `gpu_index`, `device_uuid`, `command` — so we only
    /// populate the label-derived fields the first time we see a given
    /// `(pid, gpu_index)` pair.
    fn process_process_metrics(
        &self,
        process_info_map: &mut HashMap<(u32, u32), ParsedProcessRow>,
        metric_name: &str,
        labels: &HashMap<String, String>,
        value: f64,
        host: &str,
    ) {
        let Some(pid) = labels.get("pid").and_then(|s| s.parse::<u32>().ok()) else {
            return;
        };
        // `gpu_index` is authoritative for the Users tab, but fall back
        // to `device_id` to stay compatible with any dashboard still
        // emitting only the legacy label.
        let gpu_index = labels
            .get("gpu_index")
            .and_then(|s| s.parse::<u32>().ok())
            .or_else(|| labels.get("device_id").and_then(|s| s.parse::<u32>().ok()))
            .unwrap_or(0);

        let row = process_info_map
            .entry((pid, gpu_index))
            .or_insert_with(|| ParsedProcessRow {
                host: host.to_string(),
                pid,
                gpu_index,
                ..Default::default()
            });

        // Tighter per-field caps than the generic 1024-byte label
        // cap applied by `parse_labels`. These limits bound the total
        // memory the accumulator can hold when a malicious remote host
        // pushes the 50 000-row process cap with maximum-length labels:
        // with 50 000 rows × 3 families × ~300 bytes of label content the
        // accumulator ceiling is ~45 MB per host (vs ~150 MB at the
        // generic 1024 cap, which would compound to several GB across
        // 100 hosts). They also match the exporter-side caps in
        // `src/api/metrics/process.rs` so our own output round-trips
        // unchanged. Treat an incoming value longer than the cap as
        // "take the prefix" (truncated at a UTF-8 boundary) — logging
        // would be too noisy for a per-row hot path.
        const MAX_COMMAND: usize = 256;
        const MAX_NAME: usize = 128;
        const MAX_USER: usize = 128;
        const MAX_UUID: usize = 128;

        fn utf8_truncate(s: &str, max_len: usize) -> String {
            if s.len() <= max_len {
                return s.to_string();
            }
            let mut boundary = max_len;
            while boundary > 0 && !s.is_char_boundary(boundary) {
                boundary -= 1;
            }
            s[..boundary].to_string()
        }

        // Fill label-derived fields on first sighting. Subsequent lines
        // for the same (pid, gpu_index) overwrite only when the new
        // string is non-empty — this keeps a future exporter that
        // truncates labels on a per-metric basis from wiping a real
        // value with an empty string.
        let overwrite_capped = |dst: &mut String, src: Option<&String>, cap: usize| {
            if let Some(v) = src
                && !v.is_empty()
            {
                *dst = utf8_truncate(v, cap);
            }
        };
        overwrite_capped(&mut row.user, labels.get("user"), MAX_USER);
        overwrite_capped(&mut row.command, labels.get("command"), MAX_COMMAND);
        overwrite_capped(&mut row.name, labels.get("name"), MAX_NAME);
        overwrite_capped(&mut row.gpu_uuid, labels.get("device_uuid"), MAX_UUID);

        match metric_name {
            "process_memory_used_bytes" => {
                // Prometheus gauges are doubles on the wire; clamp
                // negatives (which shouldn't happen but let's not panic
                // on garbage) and saturate at u64::MAX.
                let clamped = value.max(0.0).min(u64::MAX as f64) as u64;
                row.gpu_memory_bytes = clamped;
            }
            "process_start_time_seconds" => {
                let clamped = value.max(0.0).min(u64::MAX as f64) as u64;
                row.start_time_seconds = clamped;
            }
            "process_cpu_percent" => {
                // Store as tenths of a percent (integer) so the struct
                // can derive `Eq` without pulling in a float total-order
                // implementation.
                let tenths = (value.max(0.0) * 10.0).round() as u32;
                row.cpu_pct_tenths = tenths;
            }
            _ => {}
        }
    }

    fn parse_labels(&self, labels_str: &str) -> HashMap<String, String> {
        const MAX_LABELS: usize = 100; // Prevent unbounded growth
        const MAX_LABEL_LENGTH: usize = 1024; // Prevent large string allocations
        const MAX_INPUT_LENGTH: usize = 32768; // Maximum label string length to process

        // Limit input size to prevent DoS
        if labels_str.len() > MAX_INPUT_LENGTH {
            eprintln!("Warning: Label string too long, truncating to {MAX_INPUT_LENGTH} bytes");
            return HashMap::new();
        }

        let mut labels: HashMap<String, String> = HashMap::with_capacity(16);
        let mut label_count = 0;

        // Quote-aware splitting: only split on commas that appear outside of
        // a double-quoted value, so that a VM-owner-controlled label value
        // containing `",` cannot break out and inject fake labels.
        for label in split_labels_respecting_quotes(labels_str) {
            if label_count >= MAX_LABELS {
                break;
            }

            // Find the '=' separator without allocating a vector
            if let Some(eq_pos) = label.find('=') {
                let key = &label[..eq_pos];
                let value = &label[eq_pos + 1..];

                // Sanitize the key (trim whitespace + any stray quotes).
                let key_clean = sanitize_label_value(key);
                // For the value we strip the surrounding quotes ourselves
                // and un-escape the Prometheus exposition escape sequences
                // produced by `MetricBuilder::metric`. We never fall back to
                // the naive sanitizer for quoted values — doing so would
                // leave `\"` / `\\` / `\n` / `\r` escapes in place.
                let value_clean = unescape_label_value(value);

                // Check lengths and insert
                if key_clean.len() <= MAX_LABEL_LENGTH && value_clean.len() <= MAX_LABEL_LENGTH {
                    labels.insert(key_clean, value_clean);
                    label_count += 1;
                }
            }
        }
        labels
    }

    fn process_gpu_metrics(
        &self,
        gpu_info_map: &mut HashMap<String, GpuInfo>,
        metric_name: &str,
        labels: &HashMap<String, String>,
        value: f64,
        host: &str,
    ) {
        // NPU lines carry the same accumulator as GPU lines (both funnel into
        // `gpu_info_map`), so a row may label its device name as either `gpu`
        // or `npu` depending on the vendor exporter. Prefer `gpu` when both
        // are present.
        let gpu_name = labels
            .get("gpu")
            .or_else(|| labels.get("npu"))
            .cloned()
            .unwrap_or_default();
        // Accept GPU (`gpu_uuid`/`uuid`) and NPU (`npu_uuid`) label names for
        // backward compatibility. The legacy `uuid`/`index` labels were used
        // by pre-v0.21.0 exporters for both GPUs and NPUs; the new explicit
        // `gpu_*`/`npu_*` labels were introduced in v0.21.0 to disambiguate
        // the base label set across device families.
        let gpu_uuid = labels
            .get("gpu_uuid")
            .or_else(|| labels.get("npu_uuid"))
            .or_else(|| labels.get("uuid"))
            .cloned()
            .unwrap_or_default();
        let gpu_index = labels
            .get("gpu_index")
            .or_else(|| labels.get("npu_index"))
            .or_else(|| labels.get("index"))
            .cloned()
            .unwrap_or_default();

        if gpu_name.is_empty() || gpu_uuid.is_empty() {
            return;
        }

        let gpu_info = gpu_info_map.entry(gpu_uuid.clone()).or_insert_with(|| {
            let mut detail = HashMap::new();
            detail.insert("index".to_string(), gpu_index.clone());
            GpuInfo {
                uuid: gpu_uuid.clone(),
                time: Local::now().format("%Y-%m-%d %H:%M:%S").to_string(),
                name: gpu_name,
                device_type: "GPU".to_string(), // Default to GPU, can be overridden by gpu_info metric
                host_id: host.to_string(),      // Host identifier (e.g., "10.82.128.41:9090")
                hostname: crate::get_label_or_default!(labels, "instance", host), // DNS hostname from instance label
                instance: crate::get_label_or_default!(labels, "instance", host),
                // Start every live field at "no reading" rather than at
                // zero. A scrape only overwrites the fields whose series it
                // actually contains, so a zero default silently re-fabricated
                // the exact value the exporter went out of its way to omit:
                // a remote macOS node with no IOReport would render "0.0%" in
                // the local TUI instead of N/A (issue #325). Exporters that
                // do emit the series overwrite these below, so nothing
                // changes for a node that reports normally.
                utilization: GPU_METRIC_UNAVAILABLE,
                ane_utilization: GPU_METRIC_UNAVAILABLE,
                dla_utilization: None,
                tensorcore_utilization: None,
                temperature: 0,
                used_memory: 0,
                total_memory: 0,
                frequency: 0,
                power_consumption: GPU_METRIC_UNAVAILABLE,
                gpu_core_count: None,
                // Populated by the new threshold / pstate metric handlers
                // below when the scraped exposition contains them. Remote
                // all-smi nodes exporting older builds will leave these None,
                // matching the local graceful-degradation contract.
                temperature_threshold_slowdown: None,
                temperature_threshold_shutdown: None,
                temperature_threshold_max_operating: None,
                temperature_threshold_acoustic: None,
                performance_state: None,
                fan_speed_rpm: None,
                // Hardware details (issue #132) land via dedicated
                // `all_smi_gpu_numa_node_id`, `all_smi_gpu_gsp_firmware_*`,
                // `all_smi_nvlink_remote_device_type`, and GPM metric
                // handlers further down. Until those lines are seen in
                // the scrape these fields stay at their "unavailable"
                // defaults, matching the local graceful-degradation
                // contract.
                numa_node_id: None,
                gsp_firmware_mode: None,
                gsp_firmware_version: None,
                nvlink_remote_devices: Vec::new(),
                gpm_metrics: None,
                detail,
            }
        });

        crate::update_metric_field!(metric_name, value, gpu_info, {
            "gpu_utilization" => utilization as f64,
            "gpu_memory_used_bytes" => used_memory as u64,
            "gpu_memory_total_bytes" => total_memory as u64,
            "gpu_temperature_celsius" => temperature as u32,
            "gpu_power_consumption_watts" => power_consumption as f64,
            "gpu_frequency_mhz" => frequency as u32,
            "ane_utilization" => ane_utilization as f64
        });

        match metric_name {
            "gpu_power_limit_max_watts" => {
                gpu_info
                    .detail
                    .insert("power_limit_max".to_string(), value.to_string());
            }
            "gpu_info" => {
                // Extract device type
                if let Some(device_type) = labels.get("type") {
                    gpu_info.device_type = device_type.clone();
                }

                // Extract all GPU metadata labels in batch
                crate::extract_labels_batch!(
                    labels,
                    gpu_info.detail,
                    [
                        "cuda_version",
                        "driver_version",
                        "architecture",
                        "compute_capability",
                        "firmware",
                        "serial_number",
                        "pci_address",
                        "pci_device",
                        // Apple Silicon: why the live series are missing.
                        // Carried on the identity series so a remote viewer
                        // sees the reason and not just the absence (#325).
                        "native_metrics"
                    ]
                );

                // A node running a build that predates
                // `GpuInfo::fan_speed_rpm` never emits
                // `all_smi_gpu_fan_speed_rpm`. Its reading rides here
                // instead: the exporter turns every `detail` key into a
                // label on this series, so `Fan Speed` arrives as
                // `fan_speed`. Recovering it is what lets a mixed old/new
                // fleet show fan speed in the viewer.
                //
                // Gap-filling only. The dedicated metric stays
                // authoritative whichever order the two lines arrive in:
                // this branch yields to a value already in the field, and
                // the `gpu_fan_speed_rpm` handler assigns unconditionally,
                // so it overwrites anything recovered here. The exporter
                // happens to emit `all_smi_gpu_info` before the fan metric,
                // so on a node publishing both it is that overwrite which
                // decides. `parse_fan_speed_detail` itself enforces the same
                // integer-and-range bound the `gpu_fan_speed_rpm` handler
                // applies below, so a legacy label cannot smuggle in a value
                // the metric path would have rejected.
                if gpu_info.fan_speed_rpm.is_none()
                    && let Some(fan) = labels.get(FAN_SPEED_LEGACY_LABEL)
                    && let Some(rpm) = crate::api::metrics::gpu::parse_fan_speed_detail(fan)
                {
                    gpu_info.fan_speed_rpm = saturating_u32(rpm);
                }
            }
            // Thermal thresholds and P-state. Any round-number reading < 0 is
            // rejected via saturating_cast; the exporter only emits positive
            // u32 values, but we defend against malformed upstreams.
            "gpu_temperature_threshold_slowdown_celsius" => {
                gpu_info.temperature_threshold_slowdown = saturating_u32(value);
            }
            "gpu_temperature_threshold_shutdown_celsius" => {
                gpu_info.temperature_threshold_shutdown = saturating_u32(value);
            }
            "gpu_temperature_threshold_max_operating_celsius" => {
                gpu_info.temperature_threshold_max_operating = saturating_u32(value);
            }
            "gpu_temperature_threshold_acoustic_celsius" => {
                gpu_info.temperature_threshold_acoustic = saturating_u32(value);
            }
            "gpu_performance_state"
                // NVML defines exactly P0–P15 (0..=15). Accept only values
                // in that range so a malicious or buggy upstream cannot emit
                // an out-of-range value (e.g. 9999) that the TUI would
                // render as "P9999" and corrupt the secondary row layout.
                // Fractional inputs (e.g. 1.5) are also rejected — the
                // exporter only emits integer performance state indices.
                if (0.0..=15.0).contains(&value) && value.fract() == 0.0 => {
                    gpu_info.performance_state = saturating_u32(value);
                }
            "gpu_fan_speed_rpm"
                // The exporter only emits a non-negative integer RPM, and
                // omits the series entirely for a card with no tachometer.
                // Reject fractional and out-of-range readings so a hostile
                // upstream cannot inject a value that widens the TUI row or
                // renders as a nonsense fan speed. Shares `MAX_GPU_FAN_RPM`
                // with the reader-side clamps and the exporter's detail
                // fallback so all three agree on one bound.
                if (0.0..=f64::from(MAX_GPU_FAN_RPM)).contains(&value) && value.fract() == 0.0 => {
                    gpu_info.fan_speed_rpm = saturating_u32(value);
                }
            "gpu_numa_node_id" => {
                // NUMA node ids are non-negative on every real system.
                // Cap at a paranoid ceiling so a hostile upstream cannot
                // inject huge values: no real machine exposes more than
                // ~256 NUMA nodes today. Negative inputs, NaN, and
                // fractional values (e.g. -0.5 → saturates to 0 without
                // this guard) are all dropped.
                if value >= 0.0
                    && value.fract() == 0.0
                    && let Some(node) = saturating_i32(value)
                    && (0..=MAX_NUMA_NODE_ID).contains(&node)
                {
                    gpu_info.numa_node_id = Some(node);
                }
            }
            "gpu_gsp_firmware_mode"
                // Exporter emits exactly 0/1/2. Accept only integer values
                // in that range so a malicious upstream cannot seed the UI
                // with a bogus code. Fractional inputs (e.g. 1.5) saturate
                // to 1 without this guard, producing a silently wrong code.
                if (0.0..=2.0).contains(&value) && value.fract() == 0.0 => {
                    gpu_info.gsp_firmware_mode =
                        saturating_u32(value).and_then(|v| u8::try_from(v).ok());
                }
            "gpu_gsp_firmware_version_info" => {
                // The numeric value is always 1 (info-style metric). The
                // payload is the `version` label.
                // Reject control characters (including ANSI escape sequences
                // like ESC[2J) to prevent TUI escape injection when the
                // version string is rendered in the hardware row. A remote
                // Prometheus endpoint could embed `\x1b[...` sequences that
                // would be executed by terminal emulators on display.
                if let Some(version) = labels.get("version") {
                    let trimmed = version.trim();
                    if !trimmed.is_empty()
                        && trimmed.len() <= MAX_GSP_VERSION_LEN
                        && trimmed.chars().all(|c| !c.is_control())
                    {
                        gpu_info.gsp_firmware_version = Some(trimmed.to_string());
                    }
                }
            }
            "nvlink_remote_device_type" => {
                // Info-style metric with `link_index`, `remote_type` and
                // (issue #190) `bandwidth_mb_s` labels. Enforce a defensive
                // per-GPU cap so a malicious upstream cannot explode the
                // `nvlink_remote_devices` vec by emitting thousands of
                // distinct link indices.
                let Some(link_index) = labels.get("link_index").and_then(|s| s.parse::<u32>().ok())
                else {
                    return;
                };
                if link_index >= MAX_NVLINK_PER_GPU {
                    return;
                }
                let remote_type = labels
                    .get("remote_type")
                    .map(|s| NvLinkRemoteType::from_label(s))
                    .unwrap_or_default();
                // Optional bandwidth hint (issue #190). Absent from older
                // exporters — `None` preserves backward compatibility with
                // scrapes predating the topology tab. Reject obviously
                // nonsensical upstream values so the TUI never classifies
                // based on a malicious input.
                let bandwidth_mb_s = labels
                    .get("bandwidth_mb_s")
                    .and_then(|s| s.parse::<u32>().ok())
                    .filter(|&v| v > 0 && v <= MAX_NVLINK_BANDWIDTH_MB_S);
                // Coalesce duplicate link_index emissions — most recent
                // sample wins — so a scrape that contains the same link
                // multiple times doesn't multiply the vector length.
                if let Some(existing) = gpu_info
                    .nvlink_remote_devices
                    .iter_mut()
                    .find(|l| l.link_index == link_index)
                {
                    existing.remote_type = remote_type;
                    existing.bandwidth_mb_s = bandwidth_mb_s;
                } else if gpu_info.nvlink_remote_devices.len() < MAX_NVLINK_PER_GPU as usize {
                    gpu_info.nvlink_remote_devices.push(NvLinkRemoteDevice {
                        link_index,
                        remote_type,
                        bandwidth_mb_s,
                    });
                }
            }
            "gpu_sm_occupancy"
                // GPM fractional utilization — expected in [0.0, 1.0].
                // Values outside that band come from a buggy upstream and
                // are dropped rather than clamped so dashboards can
                // distinguish "unavailable" from "definitely zero".
                if value.is_finite() && (0.0..=1.0).contains(&value) => {
                    ensure_gpm_metrics(gpu_info).sm_occupancy = Some(value as f32);
                }
            "gpu_memory_bandwidth_utilization"
                if value.is_finite() && (0.0..=1.0).contains(&value) => {
                    ensure_gpm_metrics(gpu_info).memory_bandwidth_utilization = Some(value as f32);
                }
            "npu_firmware_info" => {
                // Handle NPU-specific firmware info metric
                crate::extract_label_to_detail!(labels, "firmware", gpu_info.detail);
            }
            _ => {}
        }
    }

    fn process_cpu_metrics(
        &self,
        cpu_info_map: &mut HashMap<String, CpuInfo>,
        metric_name: &str,
        labels: &HashMap<String, String>,
        value: f64,
        host: &str,
    ) {
        let cpu_model = crate::get_label_or_default!(labels, "cpu_model");
        // Keep the full host address including port
        let cpu_index = crate::get_label_or_default!(labels, "index", "0");

        let cpu_key = format!("{host}:{cpu_index}");

        let cpu_info = cpu_info_map.entry(cpu_key).or_insert_with(|| {
            let platform_type = if cpu_model.contains("Apple") {
                CpuPlatformType::AppleSilicon
            } else if cpu_model.contains("Intel") {
                CpuPlatformType::Intel
            } else if cpu_model.contains("AMD") {
                CpuPlatformType::Amd
            } else {
                CpuPlatformType::Other("Unknown".to_string())
            };

            CpuInfo {
                index: 0,                  // Network-parsed CpuInfo: no local AllSmi indexing, default to 0
                host_id: host.to_string(), // Host identifier (e.g., "10.82.128.41:9090")
                hostname: crate::get_label_or_default!(labels, "instance", host), // DNS hostname from instance label
                instance: crate::get_label_or_default!(labels, "instance", host),
                cpu_model: cpu_model.clone(),
                architecture: "".to_string(),
                platform_type,
                socket_count: 1,
                total_cores: 0,
                total_threads: 0,
                base_frequency_mhz: 0,
                max_frequency_mhz: 0,
                cache_size_mb: 0,
                utilization: 0.0,
                temperature: None,
                power_consumption: None,
                per_socket_info: Vec::new(),
                apple_silicon_info: None,
                per_core_utilization: Vec::new(),
                time: Local::now().format("%Y-%m-%d %H:%M:%S").to_string(),
            }
        });

        crate::update_metric_field!(metric_name, value, cpu_info, {
            "cpu_utilization" => utilization as f64,
            "cpu_socket_count" => socket_count as u32,
            "cpu_core_count" => total_cores as u32,
            "cpu_thread_count" => total_threads as u32
        });

        match metric_name {
            "cpu_model" => {
                // Handle all_smi_cpu_model info metric
                if let Some(model) = labels.get("model") {
                    cpu_info.cpu_model = model.clone();

                    // Update platform type based on new model info
                    cpu_info.platform_type = if model.contains("Apple") {
                        CpuPlatformType::AppleSilicon
                    } else if model.contains("Intel") {
                        CpuPlatformType::Intel
                    } else if model.contains("AMD")
                        || model.contains("EPYC")
                        || model.contains("Ryzen")
                    {
                        CpuPlatformType::Amd
                    } else {
                        CpuPlatformType::Other("Unknown".to_string())
                    };
                }
            }
            "cpu_frequency_mhz" => {
                cpu_info.base_frequency_mhz = value as u32;
                cpu_info.max_frequency_mhz = value as u32;
            }
            "cpu_temperature_celsius" => cpu_info.temperature = Some(value as u32),
            "cpu_power_consumption_watts" => cpu_info.power_consumption = Some(value),
            "cpu_s_core_count" => {
                self.ensure_apple_silicon_info(cpu_info);
                crate::update_optional_field!(
                    cpu_info,
                    apple_silicon_info,
                    s_core_count,
                    value as u32
                );
            }
            "cpu_p_core_count" => {
                self.ensure_apple_silicon_info(cpu_info);
                crate::update_optional_field!(
                    cpu_info,
                    apple_silicon_info,
                    p_core_count,
                    value as u32
                );
            }
            "cpu_e_core_count" => {
                self.ensure_apple_silicon_info(cpu_info);
                crate::update_optional_field!(
                    cpu_info,
                    apple_silicon_info,
                    e_core_count,
                    value as u32
                );
            }
            "cpu_s_core_utilization" => {
                self.ensure_apple_silicon_info(cpu_info);
                crate::update_optional_field!(
                    cpu_info,
                    apple_silicon_info,
                    s_core_utilization,
                    value
                );
            }
            "cpu_p_core_utilization" => {
                self.ensure_apple_silicon_info(cpu_info);
                crate::update_optional_field!(
                    cpu_info,
                    apple_silicon_info,
                    p_core_utilization,
                    value
                );
            }
            "cpu_e_core_utilization" => {
                self.ensure_apple_silicon_info(cpu_info);
                crate::update_optional_field!(
                    cpu_info,
                    apple_silicon_info,
                    e_core_utilization,
                    value
                );
            }
            "cpu_core_utilization" => {
                // Parse per-core utilization
                if let (Some(core_id_str), Some(core_type_str)) =
                    (labels.get("core_id"), labels.get("core_type"))
                    && let Ok(core_id) = core_id_str.parse::<u32>()
                {
                    // Reject out-of-range core_ids to prevent OOM from a
                    // malicious upstream sending e.g. core_id="4294967295".
                    if core_id as usize >= MAX_CPU_CORES {
                        return;
                    }

                    let core_type = match core_type_str.as_str() {
                        "S" => crate::device::CoreType::Super,
                        "P" => crate::device::CoreType::Performance,
                        "E" => crate::device::CoreType::Efficiency,
                        _ => crate::device::CoreType::Standard,
                    };

                    // Ensure vector is large enough
                    while cpu_info.per_core_utilization.len() <= core_id as usize {
                        cpu_info
                            .per_core_utilization
                            .push(crate::device::CoreUtilization {
                                core_id: cpu_info.per_core_utilization.len() as u32,
                                core_type: crate::device::CoreType::Standard,
                                utilization: 0.0,
                            });
                    }

                    // Update the specific core
                    cpu_info.per_core_utilization[core_id as usize] =
                        crate::device::CoreUtilization {
                            core_id,
                            core_type,
                            utilization: value,
                        };
                }
            }
            "cpu_info" => {
                // Extract architecture and platform type from cpu_info metric
                if let Some(architecture) = labels.get("architecture") {
                    cpu_info.architecture = architecture.clone();
                }
                if let Some(platform_type_str) = labels.get("platform_type") {
                    // Parse the platform type from the Debug format
                    cpu_info.platform_type = if platform_type_str.contains("AppleSilicon") {
                        CpuPlatformType::AppleSilicon
                    } else if platform_type_str.contains("Intel") {
                        CpuPlatformType::Intel
                    } else if platform_type_str.contains("Amd") {
                        CpuPlatformType::Amd
                    } else {
                        CpuPlatformType::Other(platform_type_str.clone())
                    };
                }
            }
            _ => {}
        }
    }

    fn process_memory_metrics(
        &self,
        memory_info_map: &mut HashMap<String, MemoryInfo>,
        metric_name: &str,
        labels: &HashMap<String, String>,
        value: f64,
        host: &str,
    ) {
        // Keep the full host address including port
        let memory_index = crate::get_label_or_default!(labels, "index", "0");
        let memory_key = format!("{host}:{memory_index}");

        let memory_info = memory_info_map
            .entry(memory_key)
            .or_insert_with(|| MemoryInfo {
                index: 0, // Network-parsed MemoryInfo: no local AllSmi indexing, default to 0
                host_id: host.to_string(), // Host identifier (e.g., "10.82.128.41:9090")
                hostname: crate::get_label_or_default!(labels, "instance", host), // DNS hostname from instance label
                instance: crate::get_label_or_default!(labels, "instance", host),
                total_bytes: 0,
                used_bytes: 0,
                available_bytes: 0,
                free_bytes: 0,
                buffers_bytes: 0,
                cached_bytes: 0,
                swap_total_bytes: 0,
                swap_used_bytes: 0,
                swap_free_bytes: 0,
                utilization: 0.0,
                time: Local::now().format("%Y-%m-%d %H:%M:%S").to_string(),
            });

        crate::update_metric_field!(metric_name, value, memory_info, {
            "memory_total_bytes" => total_bytes as u64,
            "memory_used_bytes" => used_bytes as u64,
            "memory_available_bytes" => available_bytes as u64,
            "memory_buffers_bytes" => buffers_bytes as u64,
            "memory_cached_bytes" => cached_bytes as u64,
            "memory_utilization" => utilization as f64,
            // Swap metrics (issue #220). The API exporter only emits
            // these series when `swap_total_bytes > 0`, so absent
            // metrics leave the corresponding fields at their default
            // zero — which is exactly what the renderer needs to
            // decide whether to show the Swap row.
            "swap_total_bytes" => swap_total_bytes as u64,
            "swap_used_bytes" => swap_used_bytes as u64,
            "swap_free_bytes" => swap_free_bytes as u64
        });
    }

    fn process_storage_metrics(
        &self,
        storage_info_map: &mut HashMap<String, StorageInfo>,
        metric_name: &str,
        labels: &HashMap<String, String>,
        value: f64,
        host: &str,
    ) {
        // Keep the full host address including port
        let mount_point = crate::get_label_or_default!(labels, "mount_point");
        let storage_index = crate::get_label_or_default!(labels, "index", "0");

        if mount_point.is_empty() {
            return;
        }

        let storage_key = format!("{host}:{mount_point}");
        let storage_info = storage_info_map
            .entry(storage_key)
            .or_insert_with(|| StorageInfo {
                host_id: host.to_string(), // Host identifier (e.g., "10.82.128.41:9090")
                hostname: labels
                    .get("instance")
                    .cloned()
                    .unwrap_or_else(|| host.to_string()), // DNS hostname from instance label
                mount_point: mount_point.clone(),
                total_bytes: 0,
                available_bytes: 0,
                index: storage_index.parse().unwrap_or(0),
            });

        crate::update_metric_field!(metric_name, value, storage_info, {
            "disk_total_bytes" => total_bytes as u64,
            "disk_available_bytes" => available_bytes as u64
        });
    }

    fn ensure_apple_silicon_info(&self, cpu_info: &mut CpuInfo) {
        if cpu_info.apple_silicon_info.is_none() {
            cpu_info.apple_silicon_info = Some(AppleSiliconCpuInfo {
                s_core_count: 0,
                p_core_count: 0,
                e_core_count: 0,
                gpu_core_count: 0,
                s_core_utilization: 0.0,
                p_core_utilization: 0.0,
                e_core_utilization: 0.0,
                ane_ops_per_second: None,
                s_cluster_frequency_mhz: None,
                p_cluster_frequency_mhz: None,
                e_cluster_frequency_mhz: None,
                s_core_l2_cache_mb: None,
                p_core_l2_cache_mb: None,
                e_core_l2_cache_mb: None,
            });
        }
    }

    fn update_instance_names(
        &self,
        gpu_info_map: &mut HashMap<String, GpuInfo>,
        cpu_info_map: &mut HashMap<String, CpuInfo>,
        memory_info_map: &mut HashMap<String, MemoryInfo>,
        storage_info_map: &mut HashMap<String, StorageInfo>,
        instance_name: &str,
    ) {
        // Store instance name in detail field but keep hostname as the host address
        for gpu_info in gpu_info_map.values_mut() {
            gpu_info
                .detail
                .insert("instance_name".to_string(), instance_name.to_string());
        }
        for _cpu_info in cpu_info_map.values_mut() {
            // For CPU info, we may want to store instance name differently
            // since it doesn't have a detail field by default
        }
        for _memory_info in memory_info_map.values_mut() {
            // Similarly for memory info
        }
        for _storage_info in storage_info_map.values_mut() {
            // And storage info
        }
    }
}

impl Default for MetricsParser {
    fn default() -> Self {
        Self::new()
    }
}

/// Split a Prometheus label list on top-level commas only — commas inside a
/// double-quoted value are preserved as literal content. Backslash escapes
/// inside quoted values are honoured so an author cannot terminate the quote
/// by writing `\"`.
///
/// This is a security-critical counterpart to `MetricBuilder::metric`'s
/// escaping: if we split naively on `,` a malicious label value like
/// `vgpu_vm_id="evil\", fake=\"x"` breaks out and injects an attacker-
/// controlled label, enabling cross-host metric spoofing.
/// Saturating cast of a Prometheus `f64` value to `Option<u32>` for the
/// thermal-threshold / P-state fields.
///
/// * Negative values yield `None` (the exporter emits `-1` for "not
///   reported", and non-reading clients should not pretend otherwise).
/// * Values above `u32::MAX` saturate to `u32::MAX`.
/// * `NaN` yields `None`.
fn saturating_u32(value: f64) -> Option<u32> {
    if value.is_nan() || value < 0.0 {
        return None;
    }
    if value >= u32::MAX as f64 {
        return Some(u32::MAX);
    }
    Some(value as u32)
}

/// Saturating cast of a Prometheus `f64` value to `Option<i32>` for the
/// NUMA node id field.
///
/// * Values above `i32::MAX` saturate to `i32::MAX`.
/// * Values below `i32::MIN` saturate to `i32::MIN`.
/// * `NaN` yields `None`.
fn saturating_i32(value: f64) -> Option<i32> {
    if value.is_nan() {
        return None;
    }
    if value >= i32::MAX as f64 {
        return Some(i32::MAX);
    }
    if value <= i32::MIN as f64 {
        return Some(i32::MIN);
    }
    Some(value as i32)
}

/// Maximum CPU core_id accepted from a remote scrape. No real system
/// exposes more than a few hundred cores; 1024 provides generous headroom
/// while preventing OOM from a malicious upstream sending core_id=4294967295.
const MAX_CPU_CORES: usize = 1024;

/// Maximum NvLinks per GPU accepted from a remote scrape. Current NVIDIA
/// hardware caps at 18 physical links; 32 leaves headroom for future
/// generations while still rejecting absurd input.
pub(crate) const MAX_NVLINK_PER_GPU: u32 = 32;

/// Maximum per-link bandwidth in MB/s accepted from a remote scrape.
/// NvLink 5 is ~900 GB/s per direction on H200/B200 boards (≈900 000
/// MB/s); 2 000 000 (2 TB/s) leaves generous headroom for future
/// generations while still rejecting obviously malicious input like
/// `u32::MAX`.
const MAX_NVLINK_BANDWIDTH_MB_S: u32 = 2_000_000;

/// Maximum NUMA node id accepted from a remote scrape. No real system
/// exposes more than a few hundred NUMA nodes; 4096 is paranoid.
const MAX_NUMA_NODE_ID: i32 = 4096;

/// Maximum GSP firmware version string length accepted from a remote
/// scrape. NVIDIA's GSP version strings are well under 32 bytes; 128
/// truncates any obviously pathological label.
const MAX_GSP_VERSION_LEN: usize = 128;

/// Label that `all_smi_gpu_info` carries the legacy `Fan Speed` detail
/// under, after `sanitize_label_name` has run over the key. Held in sync
/// with the exporter's spelling by
/// `legacy_fan_speed_label_matches_the_exporter_spelling`.
const FAN_SPEED_LEGACY_LABEL: &str = "fan_speed";

/// Lazily populate the GPM metrics slot on a [`GpuInfo`] so we do not
/// allocate an empty struct unless at least one GPM field was observed.
fn ensure_gpm_metrics(gpu_info: &mut GpuInfo) -> &mut GpmMetrics {
    if gpu_info.gpm_metrics.is_none() {
        gpu_info.gpm_metrics = Some(GpmMetrics::default());
    }
    gpu_info.gpm_metrics.as_mut().expect("just populated above")
}

fn split_labels_respecting_quotes(labels_str: &str) -> Vec<&str> {
    let bytes = labels_str.as_bytes();
    let mut out: Vec<&str> = Vec::with_capacity(16);
    let mut start = 0usize;
    let mut i = 0usize;
    let mut in_quotes = false;
    while i < bytes.len() {
        let b = bytes[i];
        if in_quotes {
            if b == b'\\' && i + 1 < bytes.len() {
                // Skip the escaped byte so `\"` doesn't terminate the quote.
                i += 2;
                continue;
            }
            if b == b'"' {
                in_quotes = false;
            }
            i += 1;
        } else {
            match b {
                b'"' => {
                    in_quotes = true;
                    i += 1;
                }
                b',' => {
                    out.push(&labels_str[start..i]);
                    i += 1;
                    start = i;
                }
                _ => {
                    i += 1;
                }
            }
        }
    }
    out.push(&labels_str[start..]);
    out
}

/// Strip the surrounding double quotes from a Prometheus label value and
/// un-escape the escape sequences emitted by `MetricBuilder::metric`
/// (`\\`, `\"`, `\n`, `\r`). If the value is not quoted, fall back to the
/// shared sanitizer so callers see consistent trimming behaviour for legacy
/// unquoted inputs.
fn unescape_label_value(raw: &str) -> String {
    let trimmed = raw.trim();
    let inner = match (trimmed.strip_prefix('"'), trimmed.ends_with('"')) {
        (Some(s), true) if trimmed.len() >= 2 => &s[..s.len() - 1],
        _ => return sanitize_label_value(raw),
    };

    let mut out = String::with_capacity(inner.len());
    let mut chars = inner.chars();
    while let Some(c) = chars.next() {
        if c != '\\' {
            out.push(c);
            continue;
        }
        match chars.next() {
            Some('\\') => out.push('\\'),
            Some('"') => out.push('"'),
            Some('n') => out.push('\n'),
            Some('r') => out.push('\r'),
            Some(other) => {
                // Unknown escape: keep the backslash and the following char
                // verbatim so we do not silently lose data.
                out.push('\\');
                out.push(other);
            }
            None => out.push('\\'),
        }
    }

    // Strip control characters (including ANSI escape sequences) to defend
    // against TUI escape injection from compromised remote endpoints. This
    // mirrors the stripping done in `sanitize_label_value` for unquoted
    // values, ensuring ALL label values are safe regardless of quoting.
    let out = crate::parsing::common::strip_control_chars(&out);

    // Run through length truncation (the shared sanitizer also does this,
    // but `inner` has already had its quotes removed and any stray
    // leading/trailing whitespace inside the quotes is intentional).
    const MAX_LABEL_VALUE_LENGTH: usize = 1024;
    if out.len() > MAX_LABEL_VALUE_LENGTH {
        let mut end = MAX_LABEL_VALUE_LENGTH;
        while !out.is_char_boundary(end) {
            end -= 1;
        }
        out[..end].to_string()
    } else {
        out
    }
}

/// Accumulator used while parsing vGPU Prometheus lines.
///
/// Per-instance metrics are keyed by `(gpu_uuid, vgpu_id)` so they merge into
/// a single [`VgpuInfo`] even if emitted across multiple metric families.
/// Host-scoped metrics (host mode, scheduler) populate the parent
/// [`VgpuHostInfo`] keyed solely by `gpu_uuid`.
struct VgpuParseState {
    /// `gpu_uuid -> VgpuHostInfo` being assembled.
    hosts: HashMap<String, VgpuHostInfo>,
    /// `(gpu_uuid, vgpu_id) -> VgpuInfo` instance accumulator.
    instances: HashMap<(String, u32), VgpuInfo>,
}

/// Per-scrape cap on distinct physical vGPU-capable GPUs. Matches the
/// `MAX_DEVICES_PER_TYPE` bound used for every other metric family so a
/// malicious remote exporter cannot exhaust memory by advertising unbounded
/// `gpu_uuid` values.
const MAX_VGPU_HOSTS: usize = 256;
/// Per-scrape cap on distinct `(gpu_uuid, vgpu_id)` tuples (~16 vGPUs * 256
/// hosts). New-key insertions past this limit are dropped; updates to
/// already-tracked instances proceed normally.
const MAX_VGPU_INSTANCES: usize = 4096;

impl VgpuParseState {
    fn new() -> Self {
        Self {
            hosts: HashMap::new(),
            instances: HashMap::new(),
        }
    }

    fn process(
        &mut self,
        metric_name: &str,
        labels: &HashMap<String, String>,
        value: f64,
        host: &str,
    ) {
        let gpu_uuid = labels.get("gpu_uuid").cloned().unwrap_or_default();
        if gpu_uuid.is_empty() {
            return;
        }

        // Drop samples for new hosts once the cap is reached. Updates to an
        // already-tracked host UUID always flow through.
        if !self.hosts.contains_key(&gpu_uuid) && self.hosts.len() >= MAX_VGPU_HOSTS {
            return;
        }

        // Ensure the host row exists before we touch either branch.
        let host_entry = self.hosts.entry(gpu_uuid.clone()).or_insert_with(|| {
            let gpu_index = labels
                .get("gpu_index")
                .and_then(|s| s.parse::<u32>().ok())
                .unwrap_or(0);
            let hostname = labels
                .get("host")
                .or_else(|| labels.get("instance"))
                .cloned()
                .unwrap_or_else(|| host.to_string());
            VgpuHostInfo {
                host_id: host.to_string(),
                hostname: hostname.clone(),
                instance: hostname,
                gpu_index,
                gpu_uuid: gpu_uuid.clone(),
                gpu_name: labels.get("gpu").cloned().unwrap_or_default(),
                host_mode: "Disabled".to_string(),
                scheduler_policy: 0,
                scheduler_arr_mode: 0,
                is_arr_supported: false,
                vgpus: Vec::new(),
                detail: HashMap::new(),
            }
        });

        match metric_name {
            "vgpu_host_mode" => {
                if let Some(mode) = labels.get("host_mode") {
                    host_entry.host_mode = mode.clone();
                }
            }
            "vgpu_scheduler_state" => {
                host_entry.scheduler_arr_mode = value as u32;
                if let Some(flag) = labels.get("arr_supported") {
                    host_entry.is_arr_supported = flag == "true";
                }
            }
            "vgpu_scheduler_policy" => {
                host_entry.scheduler_policy = value as u32;
            }
            _ => {
                // Per-instance metric families: require a vgpu_id label.
                let Some(vgpu_id) = labels.get("vgpu_id").and_then(|s| s.parse::<u32>().ok())
                else {
                    return;
                };
                let instance_key = (gpu_uuid.clone(), vgpu_id);
                // Drop new-instance samples once the cap is reached. Updates
                // to an already-tracked instance always flow through.
                if !self.instances.contains_key(&instance_key)
                    && self.instances.len() >= MAX_VGPU_INSTANCES
                {
                    return;
                }
                let entry = self
                    .instances
                    .entry(instance_key)
                    .or_insert_with(|| VgpuInfo {
                        instance_id: vgpu_id,
                        uuid: labels.get("vgpu_uuid").cloned().unwrap_or_default(),
                        // Keep vm_id round-tripping through the Prometheus
                        // exporter so remote mode can display the same `vm=`
                        // column as local mode.
                        vm_id: labels.get("vgpu_vm_id").cloned().unwrap_or_default(),
                        vgpu_type_name: labels.get("vgpu_type").cloned().unwrap_or_default(),
                        fb_used_bytes: 0,
                        fb_total_bytes: 0,
                        gpu_utilization: None,
                        memory_utilization: None,
                        is_active: false,
                    });

                match metric_name {
                    "vgpu_utilization" => entry.gpu_utilization = Some(value as u32),
                    "vgpu_memory_utilization" => entry.memory_utilization = Some(value as u32),
                    "vgpu_memory_used_bytes" => entry.fb_used_bytes = value as u64,
                    "vgpu_memory_total_bytes" => entry.fb_total_bytes = value as u64,
                    "vgpu_active" => entry.is_active = value > 0.0,
                    _ => {}
                }
            }
        }
    }

    fn finish(mut self) -> Vec<VgpuHostInfo> {
        // Attach instances to their owning host rows.
        for ((gpu_uuid, _vgpu_id), vgpu) in self.instances {
            if let Some(host) = self.hosts.get_mut(&gpu_uuid) {
                host.vgpus.push(vgpu);
            }
        }
        // Deterministic order: instance_id ascending inside each host.
        for host in self.hosts.values_mut() {
            host.vgpus.sort_by_key(|v| v.instance_id);
        }
        let mut out: Vec<VgpuHostInfo> = self.hosts.into_values().collect();
        out.sort_by_key(|h| h.gpu_index);
        out
    }
}

/// Accumulator used while parsing MIG Prometheus lines.
///
/// Per-instance metrics are keyed by `(gpu_uuid, mig_instance)` so they merge
/// into a single [`MigInstanceInfo`] even if emitted across multiple metric
/// families. Host-scoped metrics (`gpu_mig_mode`) populate the parent
/// [`MigGpuInfo`] keyed solely by `gpu_uuid`.
struct MigParseState {
    /// `gpu_uuid -> MigGpuInfo` being assembled.
    hosts: HashMap<String, MigGpuInfo>,
    /// `(gpu_uuid, mig_instance) -> MigInstanceInfo` accumulator.
    instances: HashMap<(String, u32), MigInstanceInfo>,
    /// UUIDs that received an explicit `gpu_mig_mode` line during this scrape.
    /// Used by `finish` to retain disabled-MIG rows (mode=0, zero instances)
    /// that were deliberately emitted by the remote exporter, while still
    /// dropping spurious hosts whose `gpu_uuid` appeared only in some other
    /// (non-MIG) family's labels by accident.
    hosts_with_mode: HashSet<String>,
}

/// Per-scrape cap on distinct MIG-capable GPUs. Matches the existing
/// `MAX_DEVICES_PER_TYPE` and `MAX_VGPU_HOSTS` ceilings so a malicious remote
/// exporter cannot exhaust memory by advertising unbounded `gpu_uuid` values.
const MAX_MIG_GPUS: usize = 256;
/// Per-scrape cap on distinct `(gpu_uuid, mig_instance)` tuples. A100/H100
/// support up to 7 instances per GPU, so 4096 leaves plenty of headroom for
/// large clusters without inviting an OOM via crafted input.
const MAX_MIG_INSTANCES: usize = 4096;
/// Per-instance cap on the `mig_instance` index value itself. MIG hardware
/// caps at 7 today; we accept up to 64 to leave headroom for future
/// architectures while still rejecting obviously bogus indices like
/// `mig_instance="9999"` from a hostile remote.
const MAX_MIG_INSTANCE_INDEX: u32 = 64;

impl MigParseState {
    fn new() -> Self {
        Self {
            hosts: HashMap::new(),
            instances: HashMap::new(),
            hosts_with_mode: HashSet::new(),
        }
    }

    fn process(
        &mut self,
        metric_name: &str,
        labels: &HashMap<String, String>,
        value: f64,
        host: &str,
    ) {
        let gpu_uuid = labels.get("gpu_uuid").cloned().unwrap_or_default();
        if gpu_uuid.is_empty() {
            return;
        }

        // Drop samples for new hosts once the cap is reached. Updates to an
        // already-tracked host UUID always flow through.
        if !self.hosts.contains_key(&gpu_uuid) && self.hosts.len() >= MAX_MIG_GPUS {
            return;
        }

        // Ensure the host row exists before we touch either branch.
        let host_entry = self.hosts.entry(gpu_uuid.clone()).or_insert_with(|| {
            let gpu_index = labels
                .get("gpu_index")
                .and_then(|s| s.parse::<u32>().ok())
                .unwrap_or(0);
            let hostname = labels
                .get("host")
                .or_else(|| labels.get("instance"))
                .cloned()
                .unwrap_or_else(|| host.to_string());
            MigGpuInfo {
                host_id: host.to_string(),
                hostname: hostname.clone(),
                instance: hostname,
                gpu_index,
                gpu_uuid: gpu_uuid.clone(),
                gpu_name: labels.get("gpu").cloned().unwrap_or_default(),
                mig_mode: false,
                instances: Vec::new(),
            }
        });

        match metric_name {
            "gpu_mig_mode" => {
                // Exporter encodes 1=enabled, 0=disabled. Treat any positive
                // value defensively as "enabled" rather than panicking on a
                // weird remote payload.
                host_entry.mig_mode = value > 0.0;
                // Remember that this host was explicitly observed via the
                // MIG-mode metric so `finish` retains it even with mode=0
                // and zero instances. Without this, disabled GPUs would be
                // silently dropped and the metric would be unobservable on
                // the consumer side.
                self.hosts_with_mode.insert(gpu_uuid.clone());
            }
            _ => {
                // Per-instance metric families: require a `mig_instance` label
                // and reject indices beyond the defensive cap.
                let Some(mig_instance) = labels
                    .get("mig_instance")
                    .and_then(|s| s.parse::<u32>().ok())
                else {
                    return;
                };
                if mig_instance > MAX_MIG_INSTANCE_INDEX {
                    return;
                }
                let instance_key = (gpu_uuid.clone(), mig_instance);

                // Drop new-instance samples once the cap is reached. Updates
                // to an already-tracked instance always flow through.
                if !self.instances.contains_key(&instance_key)
                    && self.instances.len() >= MAX_MIG_INSTANCES
                {
                    return;
                }

                let entry = self
                    .instances
                    .entry(instance_key)
                    .or_insert_with(|| MigInstanceInfo {
                        instance_id: mig_instance,
                        gpu_instance_id: labels
                            .get("gpu_instance_id")
                            .and_then(|s| s.parse::<u32>().ok()),
                        compute_instance_id: labels
                            .get("compute_instance_id")
                            .and_then(|s| s.parse::<u32>().ok()),
                        uuid: labels.get("mig_uuid").cloned().unwrap_or_default(),
                        profile_name: labels.get("mig_profile").cloned().unwrap_or_default(),
                        utilization_gpu: None,
                        utilization_memory: None,
                        memory_used_bytes: 0,
                        memory_total_bytes: 0,
                    });

                match metric_name {
                    "mig_instance_utilization_gpu" => {
                        entry.utilization_gpu = Some(value as u32);
                    }
                    "mig_instance_utilization_memory" => {
                        entry.utilization_memory = Some(value as u32);
                    }
                    "mig_instance_memory_used_bytes" => {
                        entry.memory_used_bytes = value as u64;
                    }
                    "mig_instance_memory_total_bytes" => {
                        entry.memory_total_bytes = value as u64;
                    }
                    _ => {}
                }
            }
        }
    }

    fn finish(mut self) -> Vec<MigGpuInfo> {
        // Attach instances to their owning host rows.
        for ((gpu_uuid, _mig_instance), instance) in self.instances {
            if let Some(host) = self.hosts.get_mut(&gpu_uuid) {
                host.instances.push(instance);
            }
        }
        // Deterministic order: instance_id ascending inside each host.
        for host in self.hosts.values_mut() {
            host.instances.sort_by_key(|i| i.instance_id);
        }
        // Retain a host when it either:
        //   * received an explicit `gpu_mig_mode` line (enabled or disabled),
        //     so disabled parent GPUs remain observable to consumers, or
        //   * has at least one MIG instance attached.
        // Hosts whose `gpu_uuid` appeared only incidentally on non-MIG
        // metrics still drop out here.
        let hosts_with_mode = &self.hosts_with_mode;
        self.hosts
            .retain(|uuid, h| hosts_with_mode.contains(uuid) || !h.instances.is_empty());
        // Enforce the invariant: mig_mode must be true whenever instances are
        // present. A remote feed may emit `all_smi_mig_instance_*` lines for a
        // UUID without a corresponding `gpu_mig_mode` line. The retain above
        // keeps such hosts alive via the `!h.instances.is_empty()` arm, but
        // they would carry `mig_mode=false`, which is a contradictory "ghost"
        // state that no real exporter produces. Infer the mode from presence.
        for host in self.hosts.values_mut() {
            if !host.instances.is_empty() {
                host.mig_mode = true;
            }
        }
        let mut out: Vec<MigGpuInfo> = self.hosts.into_values().collect();
        out.sort_by_key(|h| h.gpu_index);
        out
    }
}

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

    fn create_test_parser() -> MetricsParser {
        MetricsParser::new()
    }

    fn create_test_regex() -> Regex {
        Regex::new(r"^all_smi_([^\{]+)\{([^}]+)\} ([\d\.]+)$").unwrap()
    }

    /// Minimal agent-side [`GpuInfo`] used to drive genuine
    /// exporter -> parser round-trip assertions.
    fn create_test_gpu_info() -> GpuInfo {
        GpuInfo {
            uuid: "GPU-RT".to_string(),
            time: "2026-01-01 00:00:00".to_string(),
            name: "AMD Radeon RX 7900 XTX".to_string(),
            device_type: "GPU".to_string(),
            host_id: "node-9".to_string(),
            hostname: "node-9".to_string(),
            instance: "node-9".to_string(),
            utilization: 30.0,
            ane_utilization: 0.0,
            dla_utilization: None,
            tensorcore_utilization: None,
            temperature: 61,
            used_memory: 1024,
            total_memory: 8192,
            frequency: 2400,
            power_consumption: 210.0,
            gpu_core_count: None,
            temperature_threshold_slowdown: None,
            temperature_threshold_shutdown: None,
            temperature_threshold_max_operating: None,
            temperature_threshold_acoustic: None,
            performance_state: None,
            fan_speed_rpm: None,
            numa_node_id: None,
            gsp_firmware_mode: None,
            gsp_firmware_version: None,
            nvlink_remote_devices: Vec::new(),
            gpm_metrics: None,
            detail: HashMap::new(),
        }
    }

    #[test]
    fn test_parse_labels() {
        let parser = create_test_parser();

        let labels = parser.parse_labels(r#"instance="node-0058", mount_point="/", index="0""#);
        assert_eq!(labels.get("instance").unwrap(), "node-0058");
        assert_eq!(labels.get("mount_point").unwrap(), "/");
        assert_eq!(labels.get("index").unwrap(), "0");

        let labels = parser.parse_labels(r#"gpu="NVIDIA H200 141GB HBM3", uuid="GPU-12345""#);
        assert_eq!(labels.get("gpu").unwrap(), "NVIDIA H200 141GB HBM3");
        assert_eq!(labels.get("uuid").unwrap(), "GPU-12345");

        let labels = parser.parse_labels("");
        assert!(labels.is_empty());

        let labels = parser.parse_labels("malformed");
        assert!(labels.is_empty());
    }

    #[test]
    fn test_parse_gpu_metrics() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_data = r#"
all_smi_gpu_utilization{gpu="NVIDIA H200 141GB HBM3", instance="node-0058", uuid="GPU-12345", index="0"} 25.5
all_smi_gpu_memory_used_bytes{gpu="NVIDIA H200 141GB HBM3", instance="node-0058", uuid="GPU-12345", index="0"} 8589934592
all_smi_gpu_memory_total_bytes{gpu="NVIDIA H200 141GB HBM3", instance="node-0058", uuid="GPU-12345", index="0"} 34359738368
all_smi_gpu_temperature_celsius{gpu="NVIDIA H200 141GB HBM3", instance="node-0058", uuid="GPU-12345", index="0"} 65
all_smi_gpu_power_consumption_watts{gpu="NVIDIA H200 141GB HBM3", instance="node-0058", uuid="GPU-12345", index="0"} 400.5
all_smi_ane_utilization{gpu="NVIDIA H200 141GB HBM3", instance="node-0058", uuid="GPU-12345", index="0"} 15.2
"#;

        let parsed = parser.parse_metrics(test_data, host, &re);

        assert_eq!(parsed.gpu_info.len(), 1);
        let gpu = &parsed.gpu_info[0];
        assert_eq!(gpu.uuid, "GPU-12345");
        assert_eq!(gpu.name, "NVIDIA H200 141GB HBM3");
        assert_eq!(gpu.host_id, host);
        assert_eq!(gpu.hostname, "node-0058");
        assert_eq!(gpu.instance, "node-0058");
        assert_eq!(gpu.utilization, 25.5);
        assert_eq!(gpu.used_memory, 8589934592);
        assert_eq!(gpu.total_memory, 34359738368);
        assert_eq!(gpu.temperature, 65);
        assert_eq!(gpu.power_consumption, 400.5);
        assert_eq!(gpu.ane_utilization, 15.2);
    }

    /// Issue #325: a remote macOS node with no IOReport omits its GPU value
    /// series. The viewing side must keep them absent rather than defaulting
    /// them to zero, or the fix only works locally.
    #[test]
    fn test_omitted_gpu_series_stay_absent_after_scrape() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        // Exactly what the degraded exporter renders: identity plus memory,
        // no utilization / temperature / power / frequency / ANE.
        let test_data = r#"
all_smi_gpu_memory_used_bytes{gpu="Apple M2 Max GPU", instance="mac-1", gpu_uuid="AppleSiliconGPU", gpu_index="0"} 8589934592
all_smi_gpu_memory_total_bytes{gpu="Apple M2 Max GPU", instance="mac-1", gpu_uuid="AppleSiliconGPU", gpu_index="0"} 34359738368
all_smi_gpu_info{gpu="Apple M2 Max GPU", instance="mac-1", gpu_uuid="AppleSiliconGPU", gpu_index="0", type="GPU", architecture="Apple Silicon", native_metrics="unavailable"} 1
"#;

        let parsed = parser.parse_metrics(test_data, host, &re);
        assert_eq!(parsed.gpu_info.len(), 1);
        let gpu = &parsed.gpu_info[0];

        assert_eq!(gpu.utilization_reading(), None);
        assert_eq!(gpu.power_consumption_reading(), None);
        assert_eq!(gpu.temperature_reading(), None);
        assert_eq!(gpu.frequency_reading(), None);
        assert_eq!(gpu.ane_utilization_reading(), None);

        // Memory and the reason label do survive.
        assert_eq!(gpu.total_memory, 34359738368);
        assert_eq!(
            gpu.detail.get("native_metrics").map(String::as_str),
            Some("unavailable")
        );
    }

    /// A remote node that reports a genuine zero must still land as a zero on
    /// the viewing side, so the two cases stay distinguishable end to end.
    #[test]
    fn test_zero_gpu_series_survives_scrape_as_a_reading() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_data = r#"
all_smi_gpu_utilization{gpu="Apple M2 Max GPU", instance="mac-1", gpu_uuid="AppleSiliconGPU", gpu_index="0"} 0
all_smi_gpu_power_consumption_watts{gpu="Apple M2 Max GPU", instance="mac-1", gpu_uuid="AppleSiliconGPU", gpu_index="0"} 0
"#;

        let parsed = parser.parse_metrics(test_data, host, &re);
        let gpu = &parsed.gpu_info[0];
        assert_eq!(gpu.utilization_reading(), Some(0.0));
        assert_eq!(gpu.power_consumption_reading(), Some(0.0));
    }

    #[test]
    fn test_parse_gpu_thermal_thresholds_and_pstate() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_data = r#"
all_smi_gpu_utilization{gpu="NVIDIA A100", instance="node-1", uuid="GPU-T", index="0"} 30
all_smi_gpu_temperature_celsius{gpu="NVIDIA A100", instance="node-1", uuid="GPU-T", index="0"} 65
all_smi_gpu_temperature_threshold_slowdown_celsius{gpu="NVIDIA A100", instance="node-1", uuid="GPU-T", index="0"} 90
all_smi_gpu_temperature_threshold_shutdown_celsius{gpu="NVIDIA A100", instance="node-1", uuid="GPU-T", index="0"} 95
all_smi_gpu_temperature_threshold_max_operating_celsius{gpu="NVIDIA A100", instance="node-1", uuid="GPU-T", index="0"} 85
all_smi_gpu_temperature_threshold_acoustic_celsius{gpu="NVIDIA A100", instance="node-1", uuid="GPU-T", index="0"} 77
all_smi_gpu_performance_state{gpu="NVIDIA A100", instance="node-1", uuid="GPU-T", index="0"} 2
"#;

        let parsed = parser.parse_metrics(test_data, host, &re);
        assert_eq!(parsed.gpu_info.len(), 1);
        let gpu = &parsed.gpu_info[0];
        assert_eq!(gpu.temperature_threshold_slowdown, Some(90));
        assert_eq!(gpu.temperature_threshold_shutdown, Some(95));
        assert_eq!(gpu.temperature_threshold_max_operating, Some(85));
        assert_eq!(gpu.temperature_threshold_acoustic, Some(77));
        assert_eq!(gpu.performance_state, Some(2));
    }

    #[test]
    fn test_parse_gpu_fan_speed_round_trips_through_the_exporter() {
        // Exporter -> exposition -> parser must return the same RPM the
        // agent read locally, so a remote node's fan speed is not lost the
        // way the old `detail`-only representation lost it.
        use crate::api::metrics::{MetricExporter, gpu::GpuMetricExporter};

        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let mut source = create_test_gpu_info();
        source.fan_speed_rpm = Some(1450);
        let exposition = GpuMetricExporter::new(&[source]).export_metrics();
        assert!(
            exposition.contains("all_smi_gpu_fan_speed_rpm{"),
            "exporter did not emit the series:\n{exposition}"
        );

        let parsed = parser.parse_metrics(&exposition, host, &re);
        assert_eq!(parsed.gpu_info.len(), 1);
        assert_eq!(parsed.gpu_info[0].fan_speed_rpm, Some(1450));
    }

    #[test]
    fn test_parse_gpu_fan_speed_absent_stays_none() {
        // A card with no tachometer, or a remote node running a build that
        // predates the metric, must leave the field `None` rather than 0.
        use crate::api::metrics::{MetricExporter, gpu::GpuMetricExporter};

        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let exposition = GpuMetricExporter::new(&[create_test_gpu_info()]).export_metrics();
        let parsed = parser.parse_metrics(&exposition, host, &re);
        assert_eq!(parsed.gpu_info.len(), 1);
        assert!(parsed.gpu_info[0].fan_speed_rpm.is_none());
    }

    #[test]
    fn parser_rejects_out_of_range_fan_speed() {
        // A hostile upstream must not be able to widen the TUI row or
        // publish a nonsense tachometer reading.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        for bad_value in ["4294967295", "100001", "1450.5"] {
            let test_data = format!(
                "all_smi_gpu_utilization{{gpu=\"GPU\", instance=\"n\", uuid=\"GPU-FAN\", index=\"0\"}} 0\n\
                 all_smi_gpu_fan_speed_rpm{{gpu=\"GPU\", instance=\"n\", uuid=\"GPU-FAN\", index=\"0\"}} {bad_value}\n"
            );
            let parsed = parser.parse_metrics(&test_data, host, &re);
            assert_eq!(parsed.gpu_info.len(), 1);
            assert!(
                parsed.gpu_info[0].fan_speed_rpm.is_none(),
                "expected None for fan speed {bad_value}, got {:?}",
                parsed.gpu_info[0].fan_speed_rpm
            );
        }
    }

    #[test]
    fn legacy_fan_speed_label_matches_the_exporter_spelling() {
        // The recovery below reads a label name the exporter derives from
        // the detail key by sanitizing it. The two spellings must not
        // diverge: if `FAN_SPEED_DETAIL_KEY` is ever renamed, this fails
        // here instead of silently dropping every legacy node's reading.
        use crate::api::metrics::gpu::FAN_SPEED_DETAIL_KEY;
        use crate::parsing::common::sanitize_label_name;

        assert_eq!(
            sanitize_label_name(FAN_SPEED_DETAIL_KEY),
            FAN_SPEED_LEGACY_LABEL
        );
    }

    #[test]
    fn test_parse_recovers_fan_speed_from_a_legacy_node() {
        // A node predating `GpuInfo::fan_speed_rpm` emits no dedicated
        // metric; its reading only exists as a `detail`-derived label on
        // `all_smi_gpu_info`.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_data = "\
all_smi_gpu_utilization{gpu=\"GPU\", instance=\"n\", uuid=\"GPU-FAN\", index=\"0\"} 0\n\
all_smi_gpu_info{gpu=\"GPU\", instance=\"n\", uuid=\"GPU-FAN\", index=\"0\", type=\"GPU\", fan_speed=\"1450 RPM\"} 1\n";

        let parsed = parser.parse_metrics(test_data, host, &re);
        assert_eq!(parsed.gpu_info.len(), 1);
        assert_eq!(parsed.gpu_info[0].fan_speed_rpm, Some(1450));
    }

    #[test]
    fn test_dedicated_fan_speed_metric_wins_over_the_legacy_label() {
        // A node publishing both must be read from the dedicated metric.
        // The exporter emits `all_smi_gpu_info` before the fan metric, so
        // that order is the one that matters in practice, but the reverse
        // is asserted too: neither the recovery's gap-filling guard nor the
        // metric handler's unconditional write may depend on line order.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let fan_metric = "all_smi_gpu_fan_speed_rpm{gpu=\"GPU\", instance=\"n\", uuid=\"GPU-FAN\", index=\"0\"} 1200\n";
        let info_line = "all_smi_gpu_info{gpu=\"GPU\", instance=\"n\", uuid=\"GPU-FAN\", index=\"0\", type=\"GPU\", fan_speed=\"9999 RPM\"} 1\n";

        for (label, test_data) in [
            ("exporter order", format!("{info_line}{fan_metric}")),
            ("reversed", format!("{fan_metric}{info_line}")),
        ] {
            let parsed = parser.parse_metrics(&test_data, host, &re);
            assert_eq!(parsed.gpu_info.len(), 1, "{label}");
            assert_eq!(parsed.gpu_info[0].fan_speed_rpm, Some(1200), "{label}");
        }
    }

    #[test]
    fn test_legacy_fan_speed_label_rejects_non_tachometer_and_out_of_range_values() {
        // The recovery repeats the dedicated metric's bounds, so a legacy
        // label cannot smuggle in a value the metric path would reject.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        for bad_value in ["40%", "4294967295 RPM", "1450.5 RPM", "garbage"] {
            let test_data = format!(
                "all_smi_gpu_utilization{{gpu=\"GPU\", instance=\"n\", uuid=\"GPU-FAN\", index=\"0\"}} 0\n\
                 all_smi_gpu_info{{gpu=\"GPU\", instance=\"n\", uuid=\"GPU-FAN\", index=\"0\", type=\"GPU\", fan_speed=\"{bad_value}\"}} 1\n"
            );
            let parsed = parser.parse_metrics(&test_data, host, &re);
            assert_eq!(parsed.gpu_info.len(), 1);
            assert!(
                parsed.gpu_info[0].fan_speed_rpm.is_none(),
                "expected None for legacy fan speed {bad_value}, got {:?}",
                parsed.gpu_info[0].fan_speed_rpm
            );
        }
    }

    #[test]
    fn test_parse_gpu_round_trip_preserves_absence() {
        // Round-trip: a scrape from an older all-smi node without the new
        // metrics must leave the new fields as `None`, not defaults.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_data = r#"
all_smi_gpu_utilization{gpu="NVIDIA A100", instance="node-1", uuid="GPU-A", index="0"} 40
all_smi_gpu_temperature_celsius{gpu="NVIDIA A100", instance="node-1", uuid="GPU-A", index="0"} 60
"#;
        let parsed = parser.parse_metrics(test_data, host, &re);
        assert_eq!(parsed.gpu_info.len(), 1);
        let gpu = &parsed.gpu_info[0];
        assert!(gpu.temperature_threshold_slowdown.is_none());
        assert!(gpu.temperature_threshold_shutdown.is_none());
        assert!(gpu.performance_state.is_none());
    }

    #[test]
    fn parser_rejects_out_of_range_pstate() {
        // NVML defines P0–P15 only. Values outside [0, 15] from a
        // malicious or buggy upstream must be silently dropped so the TUI
        // never renders "P9999" or similar.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        for bad_value in [16.0_f64, 100.0, 9999.0, -1.0] {
            let test_data = format!(
                "all_smi_gpu_utilization{{gpu=\"GPU\", instance=\"n\", uuid=\"GPU-BAD\", index=\"0\"}} 0\n\
                 all_smi_gpu_performance_state{{gpu=\"GPU\", instance=\"n\", uuid=\"GPU-BAD\", index=\"0\"}} {bad_value}\n"
            );
            let parsed = parser.parse_metrics(&test_data, host, &re);
            assert_eq!(parsed.gpu_info.len(), 1);
            assert!(
                parsed.gpu_info[0].performance_state.is_none(),
                "expected None for out-of-range pstate {bad_value}, got {:?}",
                parsed.gpu_info[0].performance_state
            );
        }
    }

    #[test]
    fn test_saturating_u32_helper() {
        assert_eq!(saturating_u32(-1.0), None);
        assert_eq!(saturating_u32(f64::NAN), None);
        assert_eq!(saturating_u32(0.0), Some(0));
        assert_eq!(saturating_u32(93.0), Some(93));
        assert_eq!(saturating_u32(1e12), Some(u32::MAX));
    }

    #[test]
    fn parser_accepts_npu_labels_for_uuid_and_index() {
        // Issue #177: NPU vendor exporters now emit `npu_uuid` / `npu_index`
        // alongside `gpu_uuid` / `gpu_index` and the legacy `uuid` / `index`.
        // The parser must reconstruct the same device regardless of which
        // label alias a remote node uses.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_data = "\
all_smi_gpu_utilization{gpu=\"Tenstorrent Wormhole n150s\", instance=\"node-7\", npu_uuid=\"NPU-A\", npu_index=\"0\"} 33.3\n\
all_smi_gpu_temperature_celsius{gpu=\"Tenstorrent Wormhole n150s\", instance=\"node-7\", npu_uuid=\"NPU-A\", npu_index=\"0\"} 52\n\
all_smi_npu_firmware_info{npu=\"Tenstorrent Wormhole n150s\", instance=\"node-7\", npu_uuid=\"NPU-A\", npu_index=\"0\", firmware=\"1.2.3\"} 1\n";

        let parsed = parser.parse_metrics(test_data, host, &re);
        assert_eq!(parsed.gpu_info.len(), 1);
        let npu = &parsed.gpu_info[0];
        assert_eq!(npu.uuid, "NPU-A");
        assert!((npu.utilization - 33.3).abs() < 0.1);
        assert_eq!(npu.temperature, 52);
        assert_eq!(npu.detail.get("index").map(String::as_str), Some("0"));
        assert_eq!(
            npu.detail.get("firmware").map(String::as_str),
            Some("1.2.3")
        );
    }

    #[test]
    fn parser_prefers_gpu_uuid_when_both_gpu_and_npu_labels_present() {
        // A hybrid exposition that carries both sets of aliases must not
        // produce a duplicate row — the parser should pick the first
        // present candidate deterministically (gpu_uuid wins over
        // npu_uuid; gpu_index wins over npu_index).
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_data = "\
all_smi_gpu_utilization{gpu=\"Dual Labels\", instance=\"node-7\", gpu_uuid=\"GPU-Q\", gpu_index=\"2\", npu_uuid=\"NPU-Q\", npu_index=\"9\"} 12.0\n";

        let parsed = parser.parse_metrics(test_data, host, &re);
        assert_eq!(parsed.gpu_info.len(), 1);
        assert_eq!(parsed.gpu_info[0].uuid, "GPU-Q");
        assert_eq!(
            parsed.gpu_info[0].detail.get("index").map(String::as_str),
            Some("2")
        );
    }

    #[test]
    fn test_parse_cpu_metrics() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_data = r#"
all_smi_cpu_utilization{cpu_model="Intel Xeon", instance="node-0058", hostname="node-0058", index="0"} 45.2
all_smi_cpu_socket_count{cpu_model="Intel Xeon", instance="node-0058", hostname="node-0058", index="0"} 2
all_smi_cpu_core_count{cpu_model="Intel Xeon", instance="node-0058", hostname="node-0058", index="0"} 16
all_smi_cpu_thread_count{cpu_model="Intel Xeon", instance="node-0058", hostname="node-0058", index="0"} 32
all_smi_cpu_frequency_mhz{cpu_model="Intel Xeon", instance="node-0058", hostname="node-0058", index="0"} 2400
all_smi_cpu_temperature_celsius{cpu_model="Intel Xeon", instance="node-0058", hostname="node-0058", index="0"} 55
all_smi_cpu_power_consumption_watts{cpu_model="Intel Xeon", instance="node-0058", hostname="node-0058", index="0"} 125.5
"#;

        let parsed = parser.parse_metrics(test_data, host, &re);

        assert_eq!(parsed.cpu_info.len(), 1);
        let cpu = &parsed.cpu_info[0];
        assert_eq!(cpu.host_id, host);
        assert_eq!(cpu.hostname, "node-0058");
        assert_eq!(cpu.instance, "node-0058");
        assert_eq!(cpu.cpu_model, "Intel Xeon");
        assert_eq!(cpu.utilization, 45.2);
        assert_eq!(cpu.socket_count, 2);
        assert_eq!(cpu.total_cores, 16);
        assert_eq!(cpu.total_threads, 32);
        assert_eq!(cpu.base_frequency_mhz, 2400);
        assert_eq!(cpu.max_frequency_mhz, 2400);
        assert_eq!(cpu.temperature, Some(55));
        assert_eq!(cpu.power_consumption, Some(125.5));
        assert!(matches!(
            cpu.platform_type,
            crate::device::CpuPlatformType::Intel
        ));
    }

    #[test]
    fn test_parse_apple_silicon_cpu_metrics() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_data = r#"
all_smi_cpu_utilization{cpu_model="Apple M2 Max", instance="node-0058", hostname="node-0058", index="0"} 30.5
all_smi_cpu_p_core_count{cpu_model="Apple M2 Max", instance="node-0058", hostname="node-0058", index="0"} 8
all_smi_cpu_e_core_count{cpu_model="Apple M2 Max", instance="node-0058", hostname="node-0058", index="0"} 4
all_smi_cpu_p_core_utilization{cpu_model="Apple M2 Max", instance="node-0058", hostname="node-0058", index="0"} 25.2
all_smi_cpu_e_core_utilization{cpu_model="Apple M2 Max", instance="node-0058", hostname="node-0058", index="0"} 10.8
"#;

        let parsed = parser.parse_metrics(test_data, host, &re);

        assert_eq!(parsed.cpu_info.len(), 1);
        let cpu = &parsed.cpu_info[0];
        assert_eq!(cpu.cpu_model, "Apple M2 Max");
        assert_eq!(cpu.utilization, 30.5);
        assert!(matches!(
            cpu.platform_type,
            crate::device::CpuPlatformType::AppleSilicon
        ));

        let apple_info = cpu.apple_silicon_info.as_ref().unwrap();
        assert_eq!(apple_info.p_core_count, 8);
        assert_eq!(apple_info.e_core_count, 4);
        assert_eq!(apple_info.p_core_utilization, 25.2);
        assert_eq!(apple_info.e_core_utilization, 10.8);
    }

    #[test]
    fn test_parse_memory_metrics() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_data = r#"
all_smi_memory_total_bytes{instance="node-0058", hostname="node-0058", index="0"} 137438953472
all_smi_memory_used_bytes{instance="node-0058", hostname="node-0058", index="0"} 68719476736
all_smi_memory_available_bytes{instance="node-0058", hostname="node-0058", index="0"} 68719476736
all_smi_memory_utilization{instance="node-0058", hostname="node-0058", index="0"} 50.0
"#;

        let parsed = parser.parse_metrics(test_data, host, &re);

        assert_eq!(parsed.memory_info.len(), 1);
        let memory = &parsed.memory_info[0];
        assert_eq!(memory.host_id, host);
        assert_eq!(memory.hostname, "node-0058");
        assert_eq!(memory.instance, "node-0058");
        assert_eq!(memory.total_bytes, 137438953472);
        assert_eq!(memory.used_bytes, 68719476736);
        assert_eq!(memory.available_bytes, 68719476736);
        assert_eq!(memory.utilization, 50.0);
        // Swap fields default to zero when the exporter omits them
        // (the API guard at `src/api/metrics/memory.rs:76` skips swap
        // series on hosts where `swap_total_bytes == 0`).
        assert_eq!(memory.swap_total_bytes, 0);
        assert_eq!(memory.swap_used_bytes, 0);
        assert_eq!(memory.swap_free_bytes, 0);
    }

    #[test]
    fn test_parse_swap_metrics() {
        // Issue #220: swap series share the per-host memory accumulator.
        // The parser routes `swap_*` lines into the same `MemoryInfo`
        // row that the matching `memory_*` lines populate.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_data = r#"
all_smi_memory_total_bytes{instance="node-0058", hostname="node-0058", index="0"} 137438953472
all_smi_memory_used_bytes{instance="node-0058", hostname="node-0058", index="0"} 68719476736
all_smi_swap_total_bytes{instance="node-0058", hostname="node-0058", index="0"} 4294967296
all_smi_swap_used_bytes{instance="node-0058", hostname="node-0058", index="0"} 536870912
all_smi_swap_free_bytes{instance="node-0058", hostname="node-0058", index="0"} 3758096384
"#;

        let parsed = parser.parse_metrics(test_data, host, &re);

        assert_eq!(parsed.memory_info.len(), 1);
        let memory = &parsed.memory_info[0];
        assert_eq!(memory.total_bytes, 137438953472);
        assert_eq!(memory.used_bytes, 68719476736);
        assert_eq!(memory.swap_total_bytes, 4294967296);
        assert_eq!(memory.swap_used_bytes, 536870912);
        assert_eq!(memory.swap_free_bytes, 3758096384);
    }

    #[test]
    fn test_parse_storage_metrics() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_data = r#"
all_smi_disk_total_bytes{instance="node-0058", mount_point="/", index="0"} 4398046511104
all_smi_disk_available_bytes{instance="node-0058", mount_point="/", index="0"} 891915494941
all_smi_disk_total_bytes{instance="node-0058", mount_point="/home", index="1"} 1099511627776
all_smi_disk_available_bytes{instance="node-0058", mount_point="/home", index="1"} 549755813888
"#;

        let parsed = parser.parse_metrics(test_data, host, &re);

        assert_eq!(parsed.storage_info.len(), 2);

        let root_storage = parsed
            .storage_info
            .iter()
            .find(|s| s.mount_point == "/")
            .unwrap();
        assert_eq!(root_storage.host_id, host);
        assert_eq!(root_storage.hostname, "node-0058");
        assert_eq!(root_storage.total_bytes, 4398046511104);
        assert_eq!(root_storage.available_bytes, 891915494941);
        assert_eq!(root_storage.index, 0);

        let home_storage = parsed
            .storage_info
            .iter()
            .find(|s| s.mount_point == "/home")
            .unwrap();
        assert_eq!(home_storage.host_id, host);
        assert_eq!(home_storage.hostname, "node-0058");
        assert_eq!(home_storage.total_bytes, 1099511627776);
        assert_eq!(home_storage.available_bytes, 549755813888);
        assert_eq!(home_storage.index, 1);
    }

    #[test]
    fn test_parse_mixed_metrics() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_data = r#"
all_smi_gpu_utilization{gpu="NVIDIA RTX 4090", instance="node-0001", uuid="GPU-ABCDE", index="0"} 75.0
all_smi_cpu_utilization{cpu_model="AMD Ryzen", instance="node-0001", hostname="node-0001", index="0"} 60.0
all_smi_memory_total_bytes{instance="node-0001", hostname="node-0001", index="0"} 68719476736
all_smi_disk_total_bytes{instance="node-0001", mount_point="/", index="0"} 2199023255552
"#;

        let parsed = parser.parse_metrics(test_data, host, &re);

        assert_eq!(parsed.gpu_info.len(), 1);
        assert_eq!(parsed.cpu_info.len(), 1);
        assert_eq!(parsed.memory_info.len(), 1);
        assert_eq!(parsed.storage_info.len(), 1);

        assert_eq!(parsed.gpu_info[0].name, "NVIDIA RTX 4090");
        assert_eq!(parsed.gpu_info[0].utilization, 75.0);
        assert_eq!(parsed.gpu_info[0].host_id, host);
        assert_eq!(parsed.gpu_info[0].hostname, "node-0001");
        assert_eq!(parsed.gpu_info[0].instance, "node-0001");

        assert_eq!(parsed.cpu_info[0].cpu_model, "AMD Ryzen");
        assert_eq!(parsed.cpu_info[0].utilization, 60.0);
        assert!(matches!(
            parsed.cpu_info[0].platform_type,
            crate::device::CpuPlatformType::Amd
        ));

        assert_eq!(parsed.memory_info[0].total_bytes, 68719476736);
        assert_eq!(parsed.storage_info[0].total_bytes, 2199023255552);
    }

    #[test]
    fn test_invalid_metrics() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_data = r#"
invalid_metric_format
all_smi_gpu_utilization{malformed labels} invalid_value
all_smi_unknown_metric{instance="test"} 42.0
"#;

        let parsed = parser.parse_metrics(test_data, host, &re);

        assert!(parsed.gpu_info.is_empty());
        assert!(parsed.cpu_info.is_empty());
        assert!(parsed.memory_info.is_empty());
        assert!(parsed.storage_info.is_empty());
    }

    #[test]
    fn test_empty_metrics() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let parsed = parser.parse_metrics("", host, &re);

        assert!(parsed.gpu_info.is_empty());
        assert!(parsed.cpu_info.is_empty());
        assert!(parsed.memory_info.is_empty());
        assert!(parsed.storage_info.is_empty());
    }

    #[test]
    fn test_hostname_update() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_data = r#"
all_smi_gpu_utilization{gpu="Tesla V100", instance="production-node-42", uuid="GPU-XYZ", index="0"} 85.0
all_smi_cpu_utilization{cpu_model="Intel Xeon", instance="production-node-42", hostname="node-0058", index="0"} 55.0
"#;

        let parsed = parser.parse_metrics(test_data, host, &re);

        assert_eq!(parsed.gpu_info[0].host_id, host);
        assert_eq!(parsed.gpu_info[0].hostname, "production-node-42");
        assert_eq!(parsed.gpu_info[0].instance, "production-node-42");
        assert_eq!(parsed.cpu_info[0].host_id, host);
        assert_eq!(parsed.cpu_info[0].hostname, "production-node-42");
        assert_eq!(parsed.cpu_info[0].instance, "production-node-42");
    }

    #[test]
    fn test_cpu_platform_detection() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_cases = [
            ("Apple M1 Pro", crate::device::CpuPlatformType::AppleSilicon),
            ("Intel Core i9", crate::device::CpuPlatformType::Intel),
            ("AMD Ryzen 9", crate::device::CpuPlatformType::Amd),
            (
                "Unknown Processor",
                crate::device::CpuPlatformType::Other("Unknown".to_string()),
            ),
        ];

        for (cpu_model, expected_type) in test_cases {
            let test_data = format!(
                r#"all_smi_cpu_utilization{{cpu_model="{cpu_model}", instance="test", hostname="test", index="0"}} 50.0"#
            );

            let parsed = parser.parse_metrics(&test_data, host, &re);
            assert_eq!(parsed.cpu_info.len(), 1);

            match (&parsed.cpu_info[0].platform_type, &expected_type) {
                (
                    crate::device::CpuPlatformType::AppleSilicon,
                    crate::device::CpuPlatformType::AppleSilicon,
                ) => {}
                (crate::device::CpuPlatformType::Intel, crate::device::CpuPlatformType::Intel) => {}
                (crate::device::CpuPlatformType::Amd, crate::device::CpuPlatformType::Amd) => {}
                (
                    crate::device::CpuPlatformType::Other(actual),
                    crate::device::CpuPlatformType::Other(expected),
                ) => {
                    assert_eq!(actual, expected);
                }
                _ => panic!(
                    "Platform type mismatch for {cpu_model}: expected {expected_type:?}, got {:?}",
                    parsed.cpu_info[0].platform_type
                ),
            }
        }
    }

    #[test]
    fn test_missing_required_fields() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        let test_data = r#"
all_smi_gpu_utilization{instance="node-0058", index="0"} 25.5
all_smi_disk_total_bytes{instance="node-0058", index="0"} 1000000000
"#;

        let parsed = parser.parse_metrics(test_data, host, &re);

        assert!(parsed.gpu_info.is_empty());
        assert!(parsed.storage_info.is_empty());
    }

    #[test]
    fn test_parse_m5_super_core_metrics() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10001";

        // M5 Max has 6 Super cores + 10 Performance cores, no Efficiency cores
        let test_data = r#"
all_smi_cpu_utilization{cpu_model="Apple M5 Max", instance="m5-node", hostname="m5-node", index="0"} 42.0
all_smi_cpu_s_core_count{cpu_model="Apple M5 Max", instance="m5-node", hostname="m5-node", index="0"} 6
all_smi_cpu_p_core_count{cpu_model="Apple M5 Max", instance="m5-node", hostname="m5-node", index="0"} 10
all_smi_cpu_e_core_count{cpu_model="Apple M5 Max", instance="m5-node", hostname="m5-node", index="0"} 0
all_smi_cpu_s_core_utilization{cpu_model="Apple M5 Max", instance="m5-node", hostname="m5-node", index="0"} 38.5
all_smi_cpu_p_core_utilization{cpu_model="Apple M5 Max", instance="m5-node", hostname="m5-node", index="0"} 44.2
all_smi_cpu_e_core_utilization{cpu_model="Apple M5 Max", instance="m5-node", hostname="m5-node", index="0"} 0.0
"#;

        let parsed = parser.parse_metrics(test_data, host, &re);

        assert_eq!(parsed.cpu_info.len(), 1);
        let cpu = &parsed.cpu_info[0];
        assert_eq!(cpu.cpu_model, "Apple M5 Max");
        assert_eq!(cpu.utilization, 42.0);
        assert!(matches!(
            cpu.platform_type,
            crate::device::CpuPlatformType::AppleSilicon
        ));

        let apple_info = cpu.apple_silicon_info.as_ref().unwrap();
        assert_eq!(apple_info.s_core_count, 6);
        assert_eq!(apple_info.p_core_count, 10);
        assert_eq!(apple_info.e_core_count, 0);
        assert_eq!(apple_info.s_core_utilization, 38.5);
        assert_eq!(apple_info.p_core_utilization, 44.2);
        assert_eq!(apple_info.e_core_utilization, 0.0);
    }

    #[test]
    fn test_parse_m5_per_core_super_type() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10001";

        // Per-core metrics for M5 with S-type cores
        let test_data = r#"
all_smi_cpu_utilization{cpu_model="Apple M5 Pro", instance="m5pro-node", hostname="m5pro-node", index="0"} 35.0
all_smi_cpu_core_utilization{cpu_model="Apple M5 Pro", instance="m5pro-node", hostname="m5pro-node", core_id="0", core_type="S", index="0"} 50.0
all_smi_cpu_core_utilization{cpu_model="Apple M5 Pro", instance="m5pro-node", hostname="m5pro-node", core_id="1", core_type="S", index="0"} 40.0
all_smi_cpu_core_utilization{cpu_model="Apple M5 Pro", instance="m5pro-node", hostname="m5pro-node", core_id="2", core_type="P", index="0"} 30.0
all_smi_cpu_core_utilization{cpu_model="Apple M5 Pro", instance="m5pro-node", hostname="m5pro-node", core_id="3", core_type="P", index="0"} 25.0
"#;

        let parsed = parser.parse_metrics(test_data, host, &re);

        assert_eq!(parsed.cpu_info.len(), 1);
        let cpu = &parsed.cpu_info[0];
        assert_eq!(cpu.per_core_utilization.len(), 4);

        let core0 = &cpu.per_core_utilization[0];
        assert_eq!(core0.core_id, 0);
        assert_eq!(core0.core_type, crate::device::CoreType::Super);
        assert_eq!(core0.utilization, 50.0);

        let core1 = &cpu.per_core_utilization[1];
        assert_eq!(core1.core_id, 1);
        assert_eq!(core1.core_type, crate::device::CoreType::Super);
        assert_eq!(core1.utilization, 40.0);

        let core2 = &cpu.per_core_utilization[2];
        assert_eq!(core2.core_id, 2);
        assert_eq!(core2.core_type, crate::device::CoreType::Performance);
        assert_eq!(core2.utilization, 30.0);

        let core3 = &cpu.per_core_utilization[3];
        assert_eq!(core3.core_id, 3);
        assert_eq!(core3.core_type, crate::device::CoreType::Performance);
        assert_eq!(core3.utilization, 25.0);
    }

    #[test]
    fn test_m5_backward_compat_m1_m4_no_super_cores() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10001";

        // M2 Max: no S-cores, standard P+E layout
        let test_data = r#"
all_smi_cpu_utilization{cpu_model="Apple M2 Max", instance="m2-node", hostname="m2-node", index="0"} 20.0
all_smi_cpu_p_core_count{cpu_model="Apple M2 Max", instance="m2-node", hostname="m2-node", index="0"} 8
all_smi_cpu_e_core_count{cpu_model="Apple M2 Max", instance="m2-node", hostname="m2-node", index="0"} 4
all_smi_cpu_p_core_utilization{cpu_model="Apple M2 Max", instance="m2-node", hostname="m2-node", index="0"} 18.0
all_smi_cpu_e_core_utilization{cpu_model="Apple M2 Max", instance="m2-node", hostname="m2-node", index="0"} 5.0
"#;

        let parsed = parser.parse_metrics(test_data, host, &re);

        assert_eq!(parsed.cpu_info.len(), 1);
        let cpu = &parsed.cpu_info[0];
        let apple_info = cpu.apple_silicon_info.as_ref().unwrap();

        // s_core_count defaults to 0 when not present in metrics
        assert_eq!(apple_info.s_core_count, 0);
        assert_eq!(apple_info.s_core_utilization, 0.0);
        assert_eq!(apple_info.p_core_count, 8);
        assert_eq!(apple_info.e_core_count, 4);
    }

    #[test]
    fn test_parse_vgpu_metrics_populates_host_and_instances() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10100";

        let text = r#"
all_smi_vgpu_host_mode{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1", host_mode="Sriov"} 1
all_smi_vgpu_scheduler_state{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1", arr_supported="true"} 2
all_smi_vgpu_scheduler_policy{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1"} 1
all_smi_vgpu_utilization{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1", vgpu_id="0", vgpu_uuid="GRID-1", vgpu_type="GRID A100-8C"} 55
all_smi_vgpu_memory_used_bytes{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1", vgpu_id="0", vgpu_uuid="GRID-1", vgpu_type="GRID A100-8C"} 8589934592
all_smi_vgpu_memory_total_bytes{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1", vgpu_id="0", vgpu_uuid="GRID-1", vgpu_type="GRID A100-8C"} 17179869184
all_smi_vgpu_memory_utilization{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1", vgpu_id="0", vgpu_uuid="GRID-1", vgpu_type="GRID A100-8C"} 30
all_smi_vgpu_active{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1", vgpu_id="0", vgpu_uuid="GRID-1", vgpu_type="GRID A100-8C"} 1
all_smi_vgpu_utilization{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1", vgpu_id="1", vgpu_uuid="GRID-2", vgpu_type="GRID A100-4C"} 10
"#;

        let parsed = parser.parse_metrics(text, host, &re);

        assert_eq!(parsed.vgpu_info.len(), 1, "expected one host record");
        let host0 = &parsed.vgpu_info[0];
        assert_eq!(host0.host_mode, "Sriov");
        assert_eq!(host0.scheduler_policy, 1);
        assert_eq!(host0.scheduler_arr_mode, 2);
        assert!(host0.is_arr_supported);
        assert_eq!(host0.gpu_uuid, "GPU-A");
        assert_eq!(host0.gpu_name, "NVIDIA A100");
        assert_eq!(host0.vgpus.len(), 2);
        assert_eq!(host0.vgpus[0].instance_id, 0);
        assert_eq!(host0.vgpus[0].gpu_utilization, Some(55));
        assert_eq!(host0.vgpus[0].memory_utilization, Some(30));
        assert_eq!(host0.vgpus[0].fb_used_bytes, 8_589_934_592);
        assert_eq!(host0.vgpus[0].fb_total_bytes, 17_179_869_184);
        assert!(host0.vgpus[0].is_active);
        assert_eq!(host0.vgpus[1].instance_id, 1);
        assert_eq!(host0.vgpus[1].gpu_utilization, Some(10));
    }

    #[test]
    fn test_parse_vgpu_metrics_skips_rows_without_gpu_uuid() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10101";
        let text = r#"
all_smi_vgpu_utilization{instance="node1", vgpu_id="0"} 55
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        assert!(parsed.vgpu_info.is_empty());
    }

    #[test]
    fn test_parse_non_vgpu_host_produces_no_vgpu_rows() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10102";

        let text = r#"
all_smi_gpu_utilization{gpu="NVIDIA A100", instance="node1", uuid="GPU-X", index="0"} 50
all_smi_cpu_utilization{cpu_model="AMD", instance="node1", hostname="node1", index="0"} 20
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        assert_eq!(parsed.gpu_info.len(), 1);
        assert!(
            parsed.vgpu_info.is_empty(),
            "No vGPU rows must be emitted for a bare-metal host"
        );
    }

    #[test]
    fn vgpu_parser_caps_host_count() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10103";

        // Feed 2 * MAX_VGPU_HOSTS unique host UUIDs. Only the first
        // MAX_VGPU_HOSTS may survive; the rest must be dropped silently.
        let mut text = String::new();
        for i in 0..(2 * MAX_VGPU_HOSTS) {
            text.push_str(&format!(
                r#"all_smi_vgpu_host_mode{{gpu_index="0", gpu_uuid="GPU-{i}", gpu="NVIDIA A100", instance="node1", host="node1", host_mode="Sriov"}} 1
"#
            ));
        }
        let parsed = parser.parse_metrics(&text, host, &re);
        assert_eq!(
            parsed.vgpu_info.len(),
            MAX_VGPU_HOSTS,
            "host count must not exceed cap"
        );
    }

    #[test]
    fn vgpu_parser_caps_instance_count() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10104";

        // Generate a single host with 2 * MAX_VGPU_INSTANCES unique vGPU ids.
        // Each is emitted via a vgpu_utilization line — we pick a scrape-style
        // metric so the new-instance branch is the one being gated. After
        // parsing, at most MAX_VGPU_INSTANCES instances may appear.
        let mut text = String::new();
        // Establish the host first so the host row exists.
        text.push_str(
            r#"all_smi_vgpu_host_mode{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1", host_mode="Sriov"} 1
"#,
        );
        for i in 0..(2 * MAX_VGPU_INSTANCES) {
            text.push_str(&format!(
                r#"all_smi_vgpu_utilization{{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1", vgpu_id="{i}", vgpu_uuid="GRID-{i}", vgpu_type="GRID A100-8C"}} 1
"#
            ));
        }
        let parsed = parser.parse_metrics(&text, host, &re);
        assert_eq!(parsed.vgpu_info.len(), 1, "single host must still exist");
        assert_eq!(
            parsed.vgpu_info[0].vgpus.len(),
            MAX_VGPU_INSTANCES,
            "instance count must not exceed cap"
        );
    }

    #[test]
    fn parse_labels_is_quote_aware_against_comma_injection() {
        let parser = create_test_parser();
        // A malicious VM-controlled value embeds `",` and `\"` so a naive
        // splitter would turn one label into three. The quote-aware
        // implementation must see exactly three labels.
        let labels_str = r#"gpu_uuid="a",vgpu_vm_id="pwned\", fake=\"evil",vgpu_id="0""#;
        let labels = parser.parse_labels(labels_str);
        assert_eq!(labels.len(), 3, "got labels: {labels:?}");
        assert_eq!(labels.get("gpu_uuid").unwrap(), "a");
        assert_eq!(labels.get("vgpu_id").unwrap(), "0");
        // vm_id must contain the attacker's payload verbatim, but never be
        // interpreted as additional labels.
        assert_eq!(labels.get("vgpu_vm_id").unwrap(), r#"pwned", fake="evil"#);
        assert!(!labels.contains_key("fake"), "no injected label allowed");
    }

    #[test]
    fn parse_labels_unescapes_newline_and_carriage_return_then_strips() {
        let parser = create_test_parser();
        // Prometheus escape sequences are first reversed, then control
        // characters (including the resulting \n and \r) are stripped to
        // defend against TUI escape injection.
        let labels_str = r#"key="line1\nline2\rend""#;
        let labels = parser.parse_labels(labels_str);
        assert_eq!(labels.get("key").unwrap(), "line1line2end");
    }

    #[test]
    fn test_parse_mig_metrics_populates_host_and_instances() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10200";

        let text = r#"
all_smi_gpu_mig_mode{gpu_index="0", gpu_uuid="GPU-M", gpu="NVIDIA A100", instance="node1", host="node1"} 1
all_smi_mig_instance_utilization_gpu{gpu_index="0", gpu_uuid="GPU-M", gpu="NVIDIA A100", instance="node1", host="node1", mig_instance="0", mig_uuid="MIG-1", mig_profile="1g.5gb", gpu_instance_id="7", compute_instance_id="0"} 55
all_smi_mig_instance_utilization_memory{gpu_index="0", gpu_uuid="GPU-M", gpu="NVIDIA A100", instance="node1", host="node1", mig_instance="0", mig_uuid="MIG-1", mig_profile="1g.5gb", gpu_instance_id="7", compute_instance_id="0"} 30
all_smi_mig_instance_memory_used_bytes{gpu_index="0", gpu_uuid="GPU-M", gpu="NVIDIA A100", instance="node1", host="node1", mig_instance="0", mig_uuid="MIG-1", mig_profile="1g.5gb", gpu_instance_id="7", compute_instance_id="0"} 1073741824
all_smi_mig_instance_memory_total_bytes{gpu_index="0", gpu_uuid="GPU-M", gpu="NVIDIA A100", instance="node1", host="node1", mig_instance="0", mig_uuid="MIG-1", mig_profile="1g.5gb", gpu_instance_id="7", compute_instance_id="0"} 5368709120
all_smi_mig_instance_utilization_gpu{gpu_index="0", gpu_uuid="GPU-M", gpu="NVIDIA A100", instance="node1", host="node1", mig_instance="1", mig_uuid="MIG-2", mig_profile="2g.10gb", gpu_instance_id="2", compute_instance_id="0"} 10
"#;

        let parsed = parser.parse_metrics(text, host, &re);

        assert_eq!(parsed.mig_info.len(), 1, "expected one MIG host record");
        let host0 = &parsed.mig_info[0];
        assert!(host0.mig_mode);
        assert_eq!(host0.gpu_uuid, "GPU-M");
        assert_eq!(host0.gpu_name, "NVIDIA A100");
        assert_eq!(host0.gpu_index, 0);
        assert_eq!(host0.instances.len(), 2);

        let inst0 = &host0.instances[0];
        assert_eq!(inst0.instance_id, 0);
        assert_eq!(inst0.uuid, "MIG-1");
        assert_eq!(inst0.profile_name, "1g.5gb");
        assert_eq!(inst0.gpu_instance_id, Some(7));
        assert_eq!(inst0.compute_instance_id, Some(0));
        assert_eq!(inst0.utilization_gpu, Some(55));
        assert_eq!(inst0.utilization_memory, Some(30));
        assert_eq!(inst0.memory_used_bytes, 1_073_741_824);
        assert_eq!(inst0.memory_total_bytes, 5_368_709_120);

        let inst1 = &host0.instances[1];
        assert_eq!(inst1.instance_id, 1);
        assert_eq!(inst1.utilization_gpu, Some(10));
    }

    #[test]
    fn test_parse_mig_metrics_skips_rows_without_gpu_uuid() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10201";
        let text = r#"
all_smi_mig_instance_utilization_gpu{instance="node1", mig_instance="0"} 55
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        assert!(parsed.mig_info.is_empty());
    }

    #[test]
    fn test_parse_non_mig_host_produces_no_mig_rows() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10202";

        let text = r#"
all_smi_gpu_utilization{gpu="NVIDIA A100", instance="node1", uuid="GPU-X", index="0"} 50
all_smi_cpu_utilization{cpu_model="AMD", instance="node1", hostname="node1", index="0"} 20
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        assert_eq!(parsed.gpu_info.len(), 1);
        assert!(
            parsed.mig_info.is_empty(),
            "No MIG rows must be emitted for a bare-metal host"
        );
    }

    #[test]
    fn mig_parser_records_disabled_mode_when_only_mode_metric_present() {
        // A parent GPU with MIG mode disabled and no instances must still
        // surface one row with `mig_mode=false`. Dashboards rely on this to
        // track runtime MIG toggles and cluster-wide MIG enablement rollout;
        // silently dropping disabled rows would make the metric unobservable.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10203";

        let text = r#"
all_smi_gpu_mig_mode{gpu_index="0", gpu_uuid="GPU-X", gpu="NVIDIA A100", instance="node1", host="node1"} 0
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        assert_eq!(
            parsed.mig_info.len(),
            1,
            "Disabled-MIG row must be retained so consumers can see mode=0"
        );
        assert!(!parsed.mig_info[0].mig_mode);
        assert_eq!(parsed.mig_info[0].gpu_uuid, "GPU-X");
        assert!(parsed.mig_info[0].instances.is_empty());
    }

    #[test]
    fn mig_parser_caps_host_count() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10204";

        // Feed 2 * MAX_MIG_GPUS unique host UUIDs. Only the first
        // MAX_MIG_GPUS may survive; the rest must be dropped silently.
        let mut text = String::new();
        for i in 0..(2 * MAX_MIG_GPUS) {
            text.push_str(&format!(
                r#"all_smi_gpu_mig_mode{{gpu_index="0", gpu_uuid="GPU-{i}", gpu="NVIDIA A100", instance="node1", host="node1"}} 1
"#
            ));
        }
        let parsed = parser.parse_metrics(&text, host, &re);
        assert_eq!(
            parsed.mig_info.len(),
            MAX_MIG_GPUS,
            "host count must not exceed cap"
        );
    }

    #[test]
    fn mig_parser_caps_instance_count() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10205";

        // Single host with 2 * MAX_MIG_INSTANCES unique mig_instance ids.
        // Indices beyond MAX_MIG_INSTANCE_INDEX (64) are also dropped by the
        // per-index defensive cap, so this test exercises both layers.
        let mut text = String::new();
        text.push_str(
            r#"all_smi_gpu_mig_mode{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1"} 1
"#,
        );
        for i in 0..(2 * MAX_MIG_INSTANCES) {
            text.push_str(&format!(
                r#"all_smi_mig_instance_utilization_gpu{{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1", mig_instance="{i}", mig_uuid="MIG-{i}", mig_profile="1g.5gb"}} 1
"#
            ));
        }
        let parsed = parser.parse_metrics(&text, host, &re);
        assert_eq!(parsed.mig_info.len(), 1, "single host must still exist");
        // Per-index cap kicks in first (only IDs 0..=64 survive), so the
        // visible instance count is bounded by MAX_MIG_INSTANCE_INDEX + 1.
        assert!(
            parsed.mig_info[0].instances.len() <= (MAX_MIG_INSTANCE_INDEX as usize) + 1,
            "per-index cap must clamp instance count, got {}",
            parsed.mig_info[0].instances.len()
        );
    }

    #[test]
    fn mig_parser_drops_rows_with_oversized_instance_index() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10206";

        // mig_instance="9999" exceeds the defensive cap (64) — must be
        // dropped silently. The parent host record only surfaces because of
        // the gpu_mig_mode line.
        let text = r#"
all_smi_gpu_mig_mode{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1"} 1
all_smi_mig_instance_utilization_gpu{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1", mig_instance="9999", mig_uuid="MIG-X", mig_profile="1g.5gb"} 50
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        assert_eq!(parsed.mig_info.len(), 1, "host record present");
        assert!(
            parsed.mig_info[0].instances.is_empty(),
            "oversized mig_instance must be rejected"
        );
    }

    #[test]
    fn mig_parser_handles_missing_optional_ids() {
        // Older drivers may not emit gpu_instance_id / compute_instance_id.
        // Empty values must round-trip as None rather than panic.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10207";

        let text = r#"
all_smi_gpu_mig_mode{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1"} 1
all_smi_mig_instance_utilization_gpu{gpu_index="0", gpu_uuid="GPU-A", gpu="NVIDIA A100", instance="node1", host="node1", mig_instance="0", mig_uuid="MIG-X", mig_profile="1g.5gb", gpu_instance_id="", compute_instance_id=""} 25
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        assert_eq!(parsed.mig_info.len(), 1);
        assert_eq!(parsed.mig_info[0].instances.len(), 1);
        assert!(parsed.mig_info[0].instances[0].gpu_instance_id.is_none());
        assert!(
            parsed.mig_info[0].instances[0]
                .compute_instance_id
                .is_none()
        );
    }

    #[test]
    fn mig_parser_infers_mig_mode_from_instance_presence() {
        // When a remote feed emits `all_smi_mig_instance_*` lines for a UUID
        // without a `gpu_mig_mode` line, the host would survive the retain
        // (via the `!instances.is_empty()` arm) but carry `mig_mode=false` —
        // a contradictory "ghost" state. The parser must infer `mig_mode=true`
        // whenever instances are present.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10208";

        // No gpu_mig_mode line — only instance metrics.
        let text = r#"
all_smi_mig_instance_utilization_gpu{gpu_index="0", gpu_uuid="GPU-G", gpu="NVIDIA A100", instance="node1", host="node1", mig_instance="0", mig_uuid="MIG-G1", mig_profile="1g.5gb", gpu_instance_id="7", compute_instance_id="0"} 42
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        assert_eq!(
            parsed.mig_info.len(),
            1,
            "host must survive via instance presence"
        );
        assert!(
            parsed.mig_info[0].mig_mode,
            "mig_mode must be inferred as true when instances are present"
        );
        assert_eq!(parsed.mig_info[0].instances.len(), 1);
        assert_eq!(parsed.mig_info[0].instances[0].uuid, "MIG-G1");
    }

    #[test]
    fn parser_rejects_out_of_range_cpu_core_id() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "127.0.0.1:10058";

        // First emit a valid CPU metric so the parsed.cpu_info entry exists, then
        // inject an out-of-range core_id that must be silently dropped.
        let text = r#"
all_smi_cpu_utilization{cpu_model="Intel Xeon", instance="node-1", index="0"} 50
all_smi_cpu_core_utilization{cpu_model="Intel Xeon", instance="node-1", index="0", core_id="2000", core_type="S"} 99
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        assert_eq!(parsed.cpu_info.len(), 1);
        // The out-of-range core_id=2000 must have been rejected — the
        // per_core_utilization vector must remain empty (or at least not
        // grow to 2001 elements).
        assert!(
            parsed.cpu_info[0].per_core_utilization.is_empty(),
            "per_core_utilization must be empty when core_id >= MAX_CPU_CORES, got len={}",
            parsed.cpu_info[0].per_core_utilization.len()
        );
    }

    // ------------------------------------------------------------------
    // Process metric family (issue #189 - Users tab)
    // ------------------------------------------------------------------

    #[test]
    fn parser_collects_process_rows_with_all_label_fields() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "dgx-01:10001";

        let text = r#"
all_smi_process_memory_used_bytes{pid="1234", name="python", user="alice", device_id="0", gpu_index="0", device_uuid="GPU-abc", command="python train.py"} 2000000000
all_smi_process_start_time_seconds{pid="1234", name="python", user="alice", device_id="0", gpu_index="0", device_uuid="GPU-abc", command="python train.py"} 3723
all_smi_process_cpu_percent{pid="1234", name="python", user="alice", device_id="0", gpu_index="0", device_uuid="GPU-abc", command="python train.py"} 12.5
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        assert_eq!(parsed.process_info.len(), 1, "one row expected");
        let row = &parsed.process_info[0];
        assert_eq!(row.host, host);
        assert_eq!(row.pid, 1234);
        assert_eq!(row.user, "alice");
        assert_eq!(row.gpu_index, 0);
        assert_eq!(row.gpu_uuid, "GPU-abc");
        assert_eq!(row.command, "python train.py");
        assert_eq!(row.name, "python");
        assert_eq!(row.gpu_memory_bytes, 2_000_000_000);
        assert_eq!(row.start_time_seconds, 3723);
        assert_eq!(row.cpu_pct_tenths, 125);
    }

    #[test]
    fn parser_groups_process_families_by_pid_and_gpu_index() {
        // A process touching two GPUs should produce two rows, each with
        // a distinct `(pid, gpu_index)` key.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "dgx-01:10001";
        let text = r#"
all_smi_process_memory_used_bytes{pid="42", name="a", user="bob", device_id="0", gpu_index="0", device_uuid="U0", command="a"} 1000
all_smi_process_memory_used_bytes{pid="42", name="a", user="bob", device_id="1", gpu_index="1", device_uuid="U1", command="a"} 2000
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        assert_eq!(parsed.process_info.len(), 2);
        let mut mem_by_index: Vec<(u32, u64)> = parsed
            .process_info
            .iter()
            .map(|p| (p.gpu_index, p.gpu_memory_bytes))
            .collect();
        mem_by_index.sort();
        assert_eq!(mem_by_index, vec![(0, 1000), (1, 2000)]);
    }

    #[test]
    fn parser_tolerates_missing_user_label() {
        // Windows API mode may omit the `user` label. The row still
        // lands with user="" so the aggregator can group it under the
        // synthetic "unattributed" user.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "win-01:10001";
        let text = r#"
all_smi_process_memory_used_bytes{pid="99", name="svchost", device_id="0", gpu_index="0", device_uuid="U0", command="svchost"} 5000
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        assert_eq!(parsed.process_info.len(), 1);
        assert_eq!(parsed.process_info[0].user, "");
        assert_eq!(parsed.process_info[0].pid, 99);
    }

    #[test]
    fn parser_drops_process_rows_without_pid() {
        // A scrape that omits `pid` from the label set is unusable —
        // the parser must drop it silently rather than panic.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "dgx-01:10001";
        let text = r#"
all_smi_process_memory_used_bytes{name="a", user="bob", device_id="0", gpu_index="0", device_uuid="U0", command="a"} 1000
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        assert!(parsed.process_info.is_empty());
    }

    #[test]
    fn parser_falls_back_to_device_id_when_gpu_index_missing() {
        // Backward compatibility: the legacy exporter only emitted
        // `device_id`. The parser must still key the row correctly.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "legacy-01:10001";
        let text = r#"
all_smi_process_memory_used_bytes{pid="5", name="a", user="u", device_id="3", device_uuid="U3", command="c"} 1000
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        assert_eq!(parsed.process_info.len(), 1);
        assert_eq!(parsed.process_info[0].gpu_index, 3);
    }

    #[test]
    fn parser_returns_empty_process_list_when_no_process_metrics() {
        // Hosts without --processes don't emit the family — the parser
        // must return an empty vec rather than synthesising rows.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "no-procs:10001";
        let text = r#"
all_smi_gpu_utilization{gpu="NVIDIA A100", instance="node-1", gpu_uuid="GPU-A", gpu_index="0"} 77
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        assert!(parsed.process_info.is_empty());
    }

    // ------------------------------------------------------------------
    // Issue #190: NvLink `bandwidth_mb_s` label round-trip + backward
    // compatibility with old exporters that omit the label entirely.
    // ------------------------------------------------------------------

    #[test]
    fn nvlink_backward_compat_without_bandwidth_label() {
        // Old exporters (pre-#190) emit `nvlink_remote_device_type` with
        // only `link_index` and `remote_type` labels. The parser must
        // still populate the device list so remote dashboards are not
        // broken by the upgrade.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "legacy:9100";
        let text = r#"
all_smi_nvlink_remote_device_type{gpu="NVIDIA A100", instance="node-1", gpu_uuid="GPU-OLD", gpu_index="0", link_index="0", remote_type="gpu"} 1
all_smi_nvlink_remote_device_type{gpu="NVIDIA A100", instance="node-1", gpu_uuid="GPU-OLD", gpu_index="0", link_index="1", remote_type="switch"} 1
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        assert_eq!(parsed.gpu_info.len(), 1);
        let links = &parsed.gpu_info[0].nvlink_remote_devices;
        assert_eq!(links.len(), 2);
        // Bandwidth is None on every link since the label is absent.
        assert!(links.iter().all(|l| l.bandwidth_mb_s.is_none()));
    }

    #[test]
    fn nvlink_captures_bandwidth_when_present() {
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "h100:9100";
        let text = r#"
all_smi_nvlink_remote_device_type{gpu="NVIDIA H100", instance="node-1", gpu_uuid="GPU-NEW", gpu_index="0", link_index="0", remote_type="gpu", bandwidth_mb_s="50000"} 1
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        let links = &parsed.gpu_info[0].nvlink_remote_devices;
        assert_eq!(links.len(), 1);
        assert_eq!(links[0].bandwidth_mb_s, Some(50_000));
    }

    #[test]
    fn nvlink_rejects_absurd_bandwidth_values() {
        // A malicious upstream could emit `bandwidth_mb_s="4294967295"`.
        // The parser clamps to MAX_NVLINK_BANDWIDTH_MB_S so the topology
        // renderer never classifies against nonsense input.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "attacker:9100";
        let text = r#"
all_smi_nvlink_remote_device_type{gpu="NVIDIA H100", instance="node-1", gpu_uuid="GPU-X", gpu_index="0", link_index="0", remote_type="gpu", bandwidth_mb_s="4294967295"} 1
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        let links = &parsed.gpu_info[0].nvlink_remote_devices;
        assert_eq!(links.len(), 1);
        assert_eq!(
            links[0].bandwidth_mb_s, None,
            "absurd bandwidth must be filtered"
        );
    }

    #[test]
    fn nvlink_coalesces_duplicate_link_indices() {
        // A scrape that contains the same link index twice (e.g. a
        // racey refresh) must produce a single entry whose bandwidth
        // reflects the most-recent sample.
        let parser = create_test_parser();
        let re = create_test_regex();
        let host = "race:9100";
        let text = r#"
all_smi_nvlink_remote_device_type{gpu="NVIDIA H100", instance="node-1", gpu_uuid="GPU-A", gpu_index="0", link_index="0", remote_type="gpu", bandwidth_mb_s="25000"} 1
all_smi_nvlink_remote_device_type{gpu="NVIDIA H100", instance="node-1", gpu_uuid="GPU-A", gpu_index="0", link_index="0", remote_type="gpu", bandwidth_mb_s="50000"} 1
"#;
        let parsed = parser.parse_metrics(text, host, &re);
        let links = &parsed.gpu_info[0].nvlink_remote_devices;
        assert_eq!(links.len(), 1);
        assert_eq!(links[0].bandwidth_mb_s, Some(50_000));
    }
}