dynamo-llm 1.6.2

// SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

use axum::{
    Router,
    extract::State,
    http::StatusCode,
    response::{IntoResponse, sse::Event},
    routing::get,
};
use dynamo_runtime::{
    config::environment_names::llm::metrics as env_metrics,
    metrics::prometheus_names::{
        frontend_service, name_prefix, sanitize_frontend_prometheus_prefix,
    },
};
use prometheus::{
    Encoder, GaugeVec, HistogramOpts, HistogramVec, IntCounterVec, IntGaugeVec, Opts,
    core::{Collector, Desc},
    proto::{Gauge, LabelPair, Metric, MetricFamily, MetricType},
};
use serde::Serialize;
use std::{
    sync::{Arc, LazyLock, OnceLock},
    time::{Duration, Instant},
};

use crate::discovery::ModelManager;
use crate::local_model::runtime_config::ModelRuntimeConfig;
use crate::model_card::ModelDeploymentCard;
use crate::protocols::{
    common::metrics::{ANNOTATION_LLM_METRICS, LLMMetricAnnotation},
    openai::chat_completions::NvCreateChatCompletionStreamResponse,
};
use crate::reasoning_field::{ReasoningField, RoutedReasoning};
use dynamo_runtime::metrics::prometheus_names::clamp_u64_to_i64;

use dynamo_runtime::error::{DynamoError, ErrorType as DynamoErrorType};

fn new_failure_counter(metrics_prefix: Option<&str>) -> IntCounterVec {
    let prefix =
        sanitize_frontend_prometheus_prefix(metrics_prefix.unwrap_or(name_prefix::FRONTEND));
    IntCounterVec::new(
        Opts::new(
            format!("{}_{}", prefix, frontend_service::FAILURES_TOTAL),
            "Total number of terminal semantic request failures",
        ),
        &["class", "reason"],
    )
    .expect("the fixed failure metric name and labels are valid")
}

/// Process-wide because protocol error renderers can run without a request-scoped `Metrics`.
static FAILURE_METRICS_PREFIX: OnceLock<String> = OnceLock::new();
pub(crate) static DYNAM_FAILURES_TOTAL: LazyLock<IntCounterVec> = LazyLock::new(|| {
    let prefix = FAILURE_METRICS_PREFIX.get_or_init(|| {
        let raw = std::env::var(env_metrics::DYN_METRICS_PREFIX)
            .unwrap_or_else(|_| name_prefix::FRONTEND.to_string());
        sanitize_frontend_prometheus_prefix(&raw)
    });
    new_failure_counter(Some(prefix.as_str()))
});

fn record_failure_into(counter: &IntCounterVec, error: &DynamoError) {
    counter
        .with_label_values(&[error.class().as_str(), error.reason().as_str()])
        .inc();
}

pub(crate) fn record_failure(error: &DynamoError) {
    record_failure_into(&DYNAM_FAILURES_TOTAL, error);
}

/// Check whether an error chain indicates the request was rejected.
pub fn request_was_rejected(err: &(dyn std::error::Error + 'static)) -> bool {
    // Both overload flavors are client-visible rejections (HTTP 529). They differ
    // only in whether migration may retry elsewhere.
    const REJECTION: &[DynamoErrorType] = &[
        DynamoErrorType::ResourceExhausted,
        DynamoErrorType::WorkerOverloaded,
    ];
    const NON_REJECTION: &[DynamoErrorType] = &[];
    dynamo_runtime::error::match_error_chain(err, REJECTION, NON_REJECTION)
}

/// User-facing body for a request whose caller-supplied deadline elapsed
/// before dispatch; shared by every HTTP surface so the wording stays uniform.
pub(crate) const REQUEST_DEADLINE_EXCEEDED_MESSAGE: &str = "request deadline exceeded";

/// Identify a deadline that expired while waiting in the router's queue.
pub fn request_deadline_exceeded(err: &(dyn std::error::Error + 'static)) -> bool {
    queue_deadline_error(err).is_some()
}

pub(super) fn queue_deadline_error<'a>(
    err: &'a (dyn std::error::Error + 'static),
) -> Option<&'a DynamoError> {
    let mut current = Some(err);
    while let Some(error) = current {
        if let Some(error) = error.downcast_ref::<DynamoError>()
            && error.reason().as_str() == "router.queue_deadline_exceeded"
        {
            return Some(error);
        }
        current = error.source();
    }
    None
}

/// Check whether an error chain indicates that no backend worker is available
/// to this request. Both flavors are HTTP 503; they differ only in whether
/// migration may retry elsewhere.
pub fn request_was_unavailable(err: &(dyn std::error::Error + 'static)) -> bool {
    const UNAVAILABLE: &[DynamoErrorType] = &[
        DynamoErrorType::Unavailable,
        DynamoErrorType::WorkerUnavailable,
    ];
    const AVAILABLE: &[DynamoErrorType] = &[];
    dynamo_runtime::error::match_error_chain(err, UNAVAILABLE, AVAILABLE)
}

/// Check whether an error chain indicates the request was cancelled.
pub fn request_was_cancelled(err: &(dyn std::error::Error + 'static)) -> bool {
    const CANCELLATION: &[DynamoErrorType] = &[
        DynamoErrorType::Cancelled,
        DynamoErrorType::Backend(dynamo_runtime::error::BackendError::Cancelled),
    ];
    const NON_CANCELLATION: &[DynamoErrorType] = &[];
    dynamo_runtime::error::match_error_chain(err, CANCELLATION, NON_CANCELLATION)
}

pub fn request_was_timed_out(err: &(dyn std::error::Error + 'static)) -> bool {
    use dynamo_runtime::error::BackendError;

    const TIMEOUT: &[DynamoErrorType] = &[
        DynamoErrorType::ResponseTimeout,
        DynamoErrorType::ConnectionTimeout,
        DynamoErrorType::Backend(BackendError::ResponseTimeout),
        DynamoErrorType::Backend(BackendError::ConnectionTimeout),
    ];
    const NON_TIMEOUT: &[DynamoErrorType] = &[];
    dynamo_runtime::error::match_error_chain(err, TIMEOUT, NON_TIMEOUT)
}

pub use prometheus::Registry;

use super::RouteDoc;

/// Worker type label values for Prometheus timing metrics
pub use crate::discovery::{WORKER_TYPE_DECODE, WORKER_TYPE_PREFILL};
const UNSET_DP_RANK_LABEL: &str = "none";
const ITL_LOCAL_FLUSH_TOKENS: u64 = 64;

const MODEL_READY_HELP: &str = "Whether the frontend can route at least one inference request for the model (1 = ready, 0 = not ready)";

fn model_ready_metric_name(metrics_prefix: Option<&str>) -> String {
    let prefix =
        sanitize_frontend_prometheus_prefix(metrics_prefix.unwrap_or(name_prefix::FRONTEND));
    format!("{}_{}", prefix, frontend_service::MODEL_READY)
}

/// Collects current model readiness directly from the frontend's routing catalog.
///
/// This is evaluated at scrape time so worker registration and removal are
/// reflected without maintaining a second, potentially stale readiness state.
struct ModelReadyCollector {
    manager: Arc<ModelManager>,
    desc: Desc,
    metric_name: String,
}

impl ModelReadyCollector {
    fn new(
        manager: Arc<ModelManager>,
        metrics_prefix: Option<String>,
    ) -> Result<Self, prometheus::Error> {
        let metric_name = model_ready_metric_name(metrics_prefix.as_deref());
        let desc = Desc::new(
            metric_name.clone(),
            MODEL_READY_HELP.to_string(),
            vec!["model".to_string()],
            Default::default(),
        )?;
        Ok(Self {
            manager,
            desc,
            metric_name,
        })
    }
}

impl Collector for ModelReadyCollector {
    fn desc(&self) -> Vec<&Desc> {
        vec![&self.desc]
    }

    fn collect(&self) -> Vec<MetricFamily> {
        let readiness = self.manager.registered_model_readiness();
        if readiness.is_empty() {
            return Vec::new();
        }

        let mut metrics = Vec::with_capacity(readiness.len());
        for (model, ready) in readiness {
            let mut model_label = LabelPair::default();
            model_label.set_name("model".to_string());
            model_label.set_value(model);

            let mut gauge = Gauge::default();
            gauge.set_value(if ready { 1.0 } else { 0.0 });

            let mut metric = Metric::default();
            metric.set_label(vec![model_label]);
            metric.set_gauge(gauge);
            metrics.push(metric);
        }

        let mut family = MetricFamily::default();
        family.set_name(self.metric_name.clone());
        family.set_help(MODEL_READY_HELP.to_string());
        family.set_field_type(MetricType::GAUGE);
        family.set_metric(metrics);
        vec![family]
    }
}

/// Register the scrape-time model readiness collector with the frontend registry.
pub fn register_model_ready_metric(
    registry: &Registry,
    manager: Arc<ModelManager>,
    metrics_prefix: Option<String>,
) -> Result<(), prometheus::Error> {
    registry.register(Box::new(ModelReadyCollector::new(manager, metrics_prefix)?))
}

/// Global Prometheus gauge for last observed TTFT per worker (in seconds)
/// Labels: worker_id, dp_rank, worker_type
pub static WORKER_LAST_TIME_TO_FIRST_TOKEN_GAUGE: LazyLock<GaugeVec> = LazyLock::new(|| {
    GaugeVec::new(
        Opts::new(
            format!(
                "{}_{}",
                name_prefix::FRONTEND,
                frontend_service::WORKER_LAST_TIME_TO_FIRST_TOKEN_SECONDS
            ),
            "Last observed time to first token per worker (seconds)",
        ),
        &["worker_id", "dp_rank", "worker_type"],
    )
    .expect("Failed to create worker_last_time_to_first_token gauge")
});

/// Global Prometheus gauge for last observed input sequence tokens per worker
/// Labels: worker_id, dp_rank, worker_type
/// Updated atomically with TTFT - represents the input token count from the same request
pub static WORKER_LAST_INPUT_SEQUENCE_TOKENS_GAUGE: LazyLock<IntGaugeVec> = LazyLock::new(|| {
    IntGaugeVec::new(
        Opts::new(
            format!(
                "{}_{}",
                name_prefix::FRONTEND,
                frontend_service::WORKER_LAST_INPUT_SEQUENCE_TOKENS
            ),
            "Last observed input sequence tokens per worker",
        ),
        &["worker_id", "dp_rank", "worker_type"],
    )
    .expect("Failed to create worker_last_input_sequence_tokens gauge")
});

/// Global Prometheus gauge for last observed ITL per worker (in seconds)
/// Labels: worker_id, dp_rank, worker_type
pub static WORKER_LAST_INTER_TOKEN_LATENCY_GAUGE: LazyLock<GaugeVec> = LazyLock::new(|| {
    GaugeVec::new(
        Opts::new(
            format!(
                "{}_{}",
                name_prefix::FRONTEND,
                frontend_service::WORKER_LAST_INTER_TOKEN_LATENCY_SECONDS
            ),
            "Last observed inter-token latency per worker (seconds)",
        ),
        &["worker_id", "dp_rank", "worker_type"],
    )
    .expect("Failed to create worker_last_inter_token_latency gauge")
});

/// Register the global per-worker TTFT/ITL/input-tokens Prometheus metrics with the given registry.
///
/// This should be called once during HTTP service setup to expose the metrics
/// via the `/metrics` endpoint.
///
/// # Errors
/// Returns an error if the metrics are already registered with the registry.
pub fn register_worker_timing_metrics(registry: &Registry) -> Result<(), prometheus::Error> {
    registry.register(Box::new(WORKER_LAST_TIME_TO_FIRST_TOKEN_GAUGE.clone()))?;
    registry.register(Box::new(WORKER_LAST_INPUT_SEQUENCE_TOKENS_GAUGE.clone()))?;
    registry.register(Box::new(WORKER_LAST_INTER_TOKEN_LATENCY_GAUGE.clone()))?;
    Ok(())
}

// ---------------------------------------------------------------------------
// LoRA allocation metrics (updated by LoraController each tick)
// ---------------------------------------------------------------------------

pub static LORA_REPLICA_FACTOR_GAUGE: LazyLock<IntGaugeVec> = LazyLock::new(|| {
    IntGaugeVec::new(
        Opts::new(
            format!("dynamo_frontend_{}", frontend_service::LORA_REPLICA_FACTOR),
            "Number of replicas allocated for a LoRA adapter",
        ),
        &["endpoint", "lora"],
    )
    .expect("Failed to create lora_replica_factor gauge")
});

pub static LORA_IS_ACTIVE_GAUGE: LazyLock<IntGaugeVec> = LazyLock::new(|| {
    IntGaugeVec::new(
        Opts::new(
            format!("dynamo_frontend_{}", frontend_service::LORA_IS_ACTIVE),
            "Whether a LoRA adapter is active (1) or inactive (0)",
        ),
        &["endpoint", "lora"],
    )
    .expect("Failed to create lora_is_active gauge")
});

pub static LORA_RAW_ARRIVAL_COUNT_GAUGE: LazyLock<IntGaugeVec> = LazyLock::new(|| {
    IntGaugeVec::new(
        Opts::new(
            format!(
                "dynamo_frontend_{}",
                frontend_service::LORA_RAW_ARRIVAL_COUNT
            ),
            "Raw arrival count (windowed rate counter) for a LoRA adapter",
        ),
        &["endpoint", "lora"],
    )
    .expect("Failed to create lora_raw_arrival_count gauge")
});

pub static LORA_ESTIMATED_LOAD_GAUGE: LazyLock<IntGaugeVec> = LazyLock::new(|| {
    IntGaugeVec::new(
        Opts::new(
            format!("dynamo_frontend_{}", frontend_service::LORA_ESTIMATED_LOAD),
            "Estimated load (windowed request count) for a LoRA adapter",
        ),
        &["endpoint", "lora"],
    )
    .expect("Failed to create lora_estimated_load gauge")
});

pub static LORA_ACTIVE_REQUESTS_GAUGE: LazyLock<IntGaugeVec> = LazyLock::new(|| {
    IntGaugeVec::new(
        Opts::new(
            format!("dynamo_frontend_{}", frontend_service::LORA_ACTIVE_REQUESTS),
            "Number of in-flight requests for a LoRA adapter",
        ),
        &["endpoint", "lora"],
    )
    .expect("Failed to create lora_active_requests gauge")
});

pub static LORA_CHURN_LOADS_GAUGE: LazyLock<IntGaugeVec> = LazyLock::new(|| {
    IntGaugeVec::new(
        Opts::new(
            format!(
                "dynamo_frontend_{}",
                frontend_service::LORA_CHURN_LOADS_TOTAL
            ),
            "Total LoRA loads (new placements) this tick",
        ),
        &["endpoint"],
    )
    .expect("Failed to create lora_churn_loads gauge")
});

pub static LORA_CHURN_UNLOADS_GAUGE: LazyLock<IntGaugeVec> = LazyLock::new(|| {
    IntGaugeVec::new(
        Opts::new(
            format!(
                "dynamo_frontend_{}",
                frontend_service::LORA_CHURN_UNLOADS_TOTAL
            ),
            "Total LoRA unloads (removed placements) this tick",
        ),
        &["endpoint"],
    )
    .expect("Failed to create lora_churn_unloads gauge")
});

pub static LORA_OVERFLOW_COUNT_GAUGE: LazyLock<IntGaugeVec> = LazyLock::new(|| {
    IntGaugeVec::new(
        Opts::new(
            format!("dynamo_frontend_{}", frontend_service::LORA_OVERFLOW_COUNT),
            "MCF solver overflow count (unplaceable replicas)",
        ),
        &["endpoint"],
    )
    .expect("Failed to create lora_overflow_count gauge")
});

pub fn register_lora_allocation_metrics(registry: &Registry) -> Result<(), prometheus::Error> {
    registry.register(Box::new(LORA_REPLICA_FACTOR_GAUGE.clone()))?;
    registry.register(Box::new(LORA_IS_ACTIVE_GAUGE.clone()))?;
    registry.register(Box::new(LORA_RAW_ARRIVAL_COUNT_GAUGE.clone()))?;
    registry.register(Box::new(LORA_ESTIMATED_LOAD_GAUGE.clone()))?;
    registry.register(Box::new(LORA_ACTIVE_REQUESTS_GAUGE.clone()))?;
    registry.register(Box::new(LORA_CHURN_LOADS_GAUGE.clone()))?;
    registry.register(Box::new(LORA_CHURN_UNLOADS_GAUGE.clone()))?;
    registry.register(Box::new(LORA_OVERFLOW_COUNT_GAUGE.clone()))?;
    Ok(())
}

/// Generate log-spaced histogram buckets with values rounded to 2 significant figures.
///
/// # Arguments
/// * `min` - Minimum value for the buckets (must be > 0 for log spacing)
/// * `max` - Maximum value for the buckets (must be > min)
/// * `count` - Number of buckets to generate
///
/// # Returns
/// A vector of log-spaced values, always starting with 0.0 and ending with the rounded max value.
/// Duplicates created by rounding are removed, so the final count may be less than requested.
///
/// # Note
/// With 2 significant figures, there are roughly 90 unique values per order of magnitude.
/// Requesting more buckets than can be uniquely represented will result in deduplication.
pub fn generate_log_buckets(min: f64, max: f64, count: usize) -> Vec<f64> {
    if count == 0 {
        return vec![];
    }
    if count == 1 {
        return vec![0.0];
    }

    let requested_count = count;
    let mut buckets = Vec::with_capacity(count);
    buckets.push(0.0);

    // Generate log-spaced values from min to max
    for i in 1..count {
        let log_min = min.ln();
        let log_max = max.ln();
        let log_value = log_min + (log_max - log_min) * (i as f64) / ((count - 1) as f64);
        let value = log_value.exp();
        buckets.push(round_to_sig_figs(value, 2));
    }

    // Remove consecutive duplicates (buckets are already sorted)
    let original_len = buckets.len();
    buckets.dedup();

    // Warn if significant deduplication occurred
    if buckets.len() < original_len && (original_len - buckets.len()) > original_len / 10 {
        tracing::warn!(
            requested = requested_count,
            unique = buckets.len(),
            duplicates = original_len - buckets.len(),
            min = min,
            max = max,
            "Histogram bucket generation: Significant duplicate values after rounding to 2 sig figs. \
             Consider reducing bucket count or increasing range."
        );
    }

    buckets
}

/// Round a number to a specified number of significant figures
pub fn round_to_sig_figs(value: f64, sig_figs: u32) -> f64 {
    if value == 0.0 {
        return 0.0;
    }

    let magnitude = value.abs().log10().floor();
    let scale = 10_f64.powf(sig_figs as f64 - 1.0 - magnitude);
    (value * scale).round() / scale
}

const MAX_BUCKET_COUNT: usize = 512;

fn validate_bucket_config(min: f64, max: f64, count: usize) -> bool {
    min.is_finite()
        && max.is_finite()
        && min > 0.0
        && min < max
        && count > 0
        && count <= MAX_BUCKET_COUNT
}

/// How bucket settings are looked up. Injected rather than calling `std::env::var`
/// directly so tests can supply values without mutating the process environment:
/// concurrent `setenv`/`getenv` is undefined behavior on Unix, and many other tests in
/// this module construct `Metrics` and read these same variables.
type EnvLookup<'a> = &'a dyn Fn(&str) -> Option<String>;

/// The real lookup, used everywhere outside tests.
fn system_env(name: &str) -> Option<String> {
    std::env::var(name).ok()
}

/// Read one histogram bucket setting, e.g. `DYN_METRICS_ITL_MAX`, from the environment.
///
/// Falls back to the deprecated doubled form (`DYN_HISTOGRAM_DYN_METRICS_ITL_MAX`),
/// warning so operators can migrate off it.
///
/// Returns `default` when neither name is set. A name that *is* set but cannot be
/// parsed also returns `default`, but warns first: silently ignoring a typo makes
/// the whole knob look like it does not exist.
fn bucket_env_var<T: std::str::FromStr>(
    env: EnvLookup<'_>,
    prefix: &str,
    suffix: &str,
    default: T,
) -> T {
    let name = format!("{prefix}_{suffix}");
    let found = match env(&name) {
        Some(value) => Some((name, value)),
        None => {
            let deprecated = format!(
                "{}{prefix}_{suffix}",
                env_metrics::DEPRECATED_HISTOGRAM_PREFIX
            );
            env(&deprecated).map(|value| {
                tracing::warn!(
                    deprecated = %deprecated,
                    replacement = %name,
                    "Deprecated histogram bucket environment variable; rename it, \
                     support for the old name will be removed in a future release"
                );
                (deprecated, value)
            })
        }
    };

    match found {
        None => default,
        Some((name, value)) => value.parse::<T>().unwrap_or_else(|_| {
            tracing::warn!(
                env_var = %name,
                value = %value,
                "Could not parse histogram bucket environment variable, using default"
            );
            default
        }),
    }
}

/// Parse histogram bucket configuration from environment variables
/// Returns (min, max, count) with defaults if not specified
fn parse_bucket_config(
    env: EnvLookup<'_>,
    env_prefix: &str,
    default_min: f64,
    default_max: f64,
    default_count: usize,
) -> (f64, f64, usize) {
    if !validate_bucket_config(default_min, default_max, default_count) {
        tracing::error!(
            default_min,
            default_max,
            default_count,
            "Invalid default histogram configuration"
        );
        return (1.0, 10.0, 10);
    }
    let mut min = bucket_env_var(env, env_prefix, "MIN", default_min);
    let mut max = bucket_env_var(env, env_prefix, "MAX", default_max);
    let mut count = bucket_env_var(env, env_prefix, "COUNT", default_count);

    if !validate_bucket_config(min, max, count) {
        tracing::warn!(
            min=%min,
            max=%max,
            count=%count,
            "Invalid histogram configuration given, using defaults"
        );
        min = default_min;
        max = default_max;
        count = default_count;
    }

    (min, max, count)
}

/// State for metrics handler.
/// Optionally holds a reference to the DRT's [`MetricsRegistry`] so that
/// metrics created via `metrics().create*()` anywhere in the process are
/// automatically included in the `/metrics` response.
struct MetricsHandlerState {
    registry: Arc<Registry>,
    drt_metrics: Option<dynamo_runtime::metrics::MetricsRegistry>,
}

pub struct Metrics {
    request_started_counter: IntCounterVec,
    request_counter: IntCounterVec,
    /// Deprecated: use `active_requests_gauge`. Kept for backwards compatibility.
    inflight_gauge: IntGaugeVec,
    active_requests_gauge: IntGaugeVec,
    client_disconnect_gauge: prometheus::IntGauge,
    http_queue_gauge: IntGaugeVec,
    request_duration: HistogramVec,
    input_sequence_length: HistogramVec,
    output_sequence_length: HistogramVec,
    cached_tokens: HistogramVec,
    tokenizer_latency: HistogramVec,
    output_tokens_counter: IntCounterVec,
    time_to_first_token: HistogramVec,
    inter_token_latency: HistogramVec,
    /// End-to-end latency of an OpenAI `/v1/embeddings` request. Distinct from
    /// `request_duration` so the buckets can be sized for sub-second pooling
    /// inference rather than the 1-512s LLM-generation range. Labeled by
    /// `model`.
    embedding_latency: HistogramVec,

    // Per-request multimodal content-part count histograms (labeled by `model`).
    images_per_request: HistogramVec,
    videos_per_request: HistogramVec,
    audio_per_request: HistogramVec,
    image_tokens_per_request: HistogramVec,

    // Runtime configuration metrics. Note: Some of these metrics represent counter-like values from
    // source systems, but are implemented as gauges because they are copied/synchronized from upstream
    // counter values rather than being directly incremented.
    model_total_kv_blocks: IntGaugeVec,
    model_max_num_seqs: IntGaugeVec,
    model_max_num_batched_tokens: IntGaugeVec,
    model_context_length: IntGaugeVec,
    model_kv_cache_block_size: IntGaugeVec,
    model_migration_limit: IntGaugeVec,
    model_migration_total: IntCounterVec,
    model_migration_duration_seconds: HistogramVec,
    model_migration_max_seq_len_exceeded_total: IntCounterVec,
    model_cancellation_total: IntCounterVec,
    model_rejection_total: IntCounterVec,
}

// Inflight tracks requests from HTTP handler start until complete response is finished.
// HTTP queue tracks requests from HTTP handler start until first token generation begins (including prefill time).
// HTTP queue time is a subset of inflight time. For detailed explanation, see:
// deploy/metrics/README.md - "Request Processing Flow" section

/// RAII object for HTTP queue gauge
/// Tracks requests from HTTP handler start until metrics processing begins
pub struct HttpQueueGuard {
    metrics: Arc<Metrics>,
    model: String,
}

/// RAII object for inflight gauge and request counters
/// If this object is dropped without calling `mark_ok`, then the request will increment
/// the request counter with the `status` label with [`frontend_service::status::ERROR`]; otherwise, it will increment
/// the counter with `status` label [`frontend_service::status::SUCCESS`]
pub struct InflightGuard {
    metrics: Arc<Metrics>,
    model: String,
    endpoint: Endpoint,
    request_type: RequestType,
    status: Status,
    error_type: ErrorType,
    timer: Instant,
    request_id: String,
    span: tracing::Span,
}

/// Requests will be logged by the type of endpoint hit
/// This will include llamastack in the future
#[derive(Clone, Copy)]
pub enum Endpoint {
    /// OAI Completions
    Completions,

    /// OAI Chat Completions
    ChatCompletions,

    /// OAI Embeddings
    Embeddings,

    /// Classification (sequence classification / cross-encoder pooling)
    Classify,

    /// Cross-encoder reranking
    Rerank,

    /// Pooling (raw pooler output)
    Pooling,

    /// OAI Images
    Images,

    /// OAI Videos
    Videos,

    /// OAI Audio Speech
    Audios,

    /// OAI Responses
    Responses,

    /// Anthropic Messages
    AnthropicMessages,

    /// Tensor
    Tensor,

    /// Generate (token-in/token-out)
    Generate,
}

/// Metrics for the HTTP service
pub enum RequestType {
    /// SingleIn / SingleOut
    Unary,

    /// SingleIn / ManyOut
    Stream,
}

/// Labels for cancellation metrics
pub struct CancellationLabels {
    pub model: String,
    pub endpoint: String,
    pub request_type: String,
}

/// Status
#[derive(PartialEq)]
pub enum Status {
    Success,
    Error,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum RequestCompletion {
    Success,
    Cancelled,
    Error,
}

impl RequestCompletion {
    fn as_str(self) -> &'static str {
        match self {
            Self::Success => "success",
            Self::Cancelled => "cancelled",
            Self::Error => "error",
        }
    }
}

/// Error type classification for fine-grained observability
#[derive(PartialEq, Clone, Debug)]
pub enum ErrorType {
    /// No error (for successful requests)
    None,
    /// Client validation error (4xx with "Validation:" prefix)
    Validation,
    /// Model or resource not found (404)
    NotFound,
    /// Service overloaded or rate limited (429 or 529)
    Overload,
    /// Service unavailable because no backend worker can serve the request
    Unavailable,
    /// Request cancelled by client or timeout
    Cancelled,
    /// Backend accepted the request but stopped responding (response inactivity timeout)
    ResponseTimeout,
    /// Internal server error (500 and other unexpected errors)
    Internal,
    /// Feature not implemented (501)
    NotImplemented,
}

/// Track response-specific metrics
pub struct ResponseMetricCollector {
    metrics: Arc<Metrics>,
    model: String,
    // Per-model metric handles cached for the request. Most are resolved at construction;
    // ITL is resolved lazily on its first observation so requests that never produce ITL
    // do not allocate a local histogram. Caching avoids re-hashing the `model` label on
    // every chunk or output token. Each handle shares the underlying metric with its vec,
    // so observations are equivalent.
    output_tokens_counter: prometheus::IntCounter,
    time_to_first_token: prometheus::Histogram,
    inter_token_latency: Option<prometheus::local::LocalHistogram>,
    itl_pending_tokens: u64,
    input_sequence_length: prometheus::Histogram,
    cached_tokens: prometheus::Histogram,
    // Per-model multimodal count histogram handles, resolved once at construction.
    images_per_request: prometheus::Histogram,
    videos_per_request: prometheus::Histogram,
    audio_per_request: prometheus::Histogram,
    image_tokens_per_request: prometheus::Histogram,
    // Latched per-request counts (for the tracing span fields recorded in `Drop`).
    image_count_val: usize,
    video_count_val: usize,
    audio_count_val: usize,
    multimodal_counts_observed: bool,
    start_time: Instant,
    // we use is_first_token to distinguish TTFT from ITL. It is true by default and
    // flipped to false when the first token is returned and TTFT is published.
    is_first_token: bool,
    // we track the last response time so that ITL for the newly returned tokens can
    // be computed.
    last_response_time: Option<Duration>,
    osl: usize,
    isl: usize,
    ttft_ms: Option<f64>,
    itl_sum_secs: f64,
    itl_count: u64,
    // we track if cached_tokens has been observed to ensure we only increment once per request
    cached_tokens_observed: bool,
    // we track if tokenize latency has been observed to ensure we only increment once per request
    tokenize_latency_observed: bool,
    // latest accumulated detokenize latency and sample count reported by tracker
    detokenize_latency_total: Duration,
    detokenize_count_total: u64,
    // Prefill worker info for TTFT attribution (set from LLMMetricAnnotation)
    prefill_worker_id: Option<u64>,
    prefill_dp_rank: Option<u32>,
    // Prefill worker type for Prometheus labeling - stored at routing time to avoid MDC lookup
    prefill_worker_type: Option<String>,
    // Decode worker info for ITL attribution (set from LLMMetricAnnotation)
    decode_worker_id: Option<u64>,
    decode_dp_rank: Option<u32>,
    // Decode worker type for Prometheus labeling - stored at routing time to avoid MDC lookup
    decode_worker_type: Option<String>,
    // Cached per-worker ITL gauge handle. The decode-worker labels are latched once at
    // routing time (`set_worker_info` only sets when unset), so this GaugeVec handle is
    // resolved a single time and reused — the per-token observe path then does no label
    // formatting (`worker_id.to_string()`) or label hashing, only `set`.
    decode_itl_gauge: Option<prometheus::Gauge>,
}

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

impl Metrics {
    /// Create Metrics with the standard prefix defined by [`name_prefix::FRONTEND`] or specify custom prefix via the following environment variable:
    /// - `DYN_METRICS_PREFIX`: Override the default metrics prefix
    ///
    /// The following metrics will be created with the configured prefix:
    /// - `{prefix}_requests_total` - IntCounterVec for the total number of requests processed
    /// - `{prefix}_inflight_requests` - IntGaugeVec for the number of inflight/concurrent requests
    /// - `{prefix}_disconnected_clients` - IntGauge for the number of disconnected clients
    /// - `{prefix}_request_duration_seconds` - HistogramVec for the duration of requests
    /// - `{prefix}_input_sequence_tokens` - HistogramVec for input sequence length in tokens
    /// - `{prefix}_output_sequence_tokens` - HistogramVec for output sequence length in tokens
    /// - `{prefix}_tokenizer_latency_ms` - HistogramVec for tokenizer latency in milliseconds
    /// - `{prefix}_output_tokens_total` - IntCounterVec for total output tokens generated (real-time updates)
    /// - `{prefix}_time_to_first_token_seconds` - HistogramVec for time to first token in seconds
    /// - `{prefix}_inter_token_latency_seconds` - HistogramVec for inter-token latency in seconds
    ///
    /// ## Histogram Bucket Configuration
    ///
    /// All histograms use log-spaced buckets rounded to 2 significant figures. Bucket configuration
    /// can be customized via environment variables (MIN: minimum value, MAX: maximum value, COUNT: number of buckets):
    ///
    /// - `DYN_METRICS_REQUEST_DURATION_{MIN,MAX,COUNT}` - Request duration histogram (defaults: 1.0, 512.0, 10)
    /// - `DYN_METRICS_INPUT_SEQUENCE_{MIN,MAX,COUNT}` - Input sequence length histogram (defaults: 50.0, 128000.0, 12)
    /// - `DYN_METRICS_OUTPUT_SEQUENCE_{MIN,MAX,COUNT}` - Output sequence length histogram (defaults: 50.0, 32000.0, 10)
    /// - `DYN_METRICS_TTFT_{MIN,MAX,COUNT}` - Time to first token histogram (defaults: 0.001, 480.0, 18)
    /// - `DYN_METRICS_ITL_{MIN,MAX,COUNT}` - Inter-token latency histogram (defaults: 0.001, 80.0, 20)
    /// - `DYN_METRICS_EMBEDDING_LATENCY_{MIN,MAX,COUNT}` - End-to-end `/v1/embeddings` latency histogram (defaults: 0.001, 10.0, 14)
    ///
    /// ## Model Configuration Metrics
    ///
    /// Runtime config metrics (from ModelRuntimeConfig):
    /// - `{prefix}_model_total_kv_blocks` - IntGaugeVec for total KV cache blocks available for a worker serving the model
    /// - `{prefix}_model_max_num_seqs` - IntGaugeVec for maximum sequences for a worker serving the model
    /// - `{prefix}_model_max_num_batched_tokens` - IntGaugeVec for maximum batched tokens for a worker serving the model
    ///
    /// MDC metrics (from ModelDeploymentCard):
    /// - `{prefix}_model_context_length` - IntGaugeVec for maximum context length for a worker serving the model
    /// - `{prefix}_model_kv_cache_block_size` - IntGaugeVec for KV cache block size for a worker serving the model
    /// - `{prefix}_model_migration_limit` - IntGaugeVec for request migration limit for a worker serving the model
    ///
    /// ## Runtime Config Polling Configuration
    ///
    /// The polling behavior can be configured via environment variables:
    /// - `DYN_HTTP_SVC_CONFIG_METRICS_POLL_INTERVAL_SECS`: Poll interval in seconds (must be > 0, supports fractional seconds, defaults to 8)
    ///
    /// Metrics are never removed to preserve historical data. Runtime config and MDC
    /// metrics are updated when models are discovered and their configurations are available.
    pub fn new() -> Self {
        Self::new_with_prefix(std::env::var(env_metrics::DYN_METRICS_PREFIX).ok())
    }

    /// Create Metrics with an explicit optional prefix. `None` uses the standard
    /// frontend prefix and does not read environment variables.
    pub fn new_with_prefix(metrics_prefix: Option<String>) -> Self {
        Self::build(metrics_prefix, &system_env)
    }

    /// Real constructor. `env` is injected so tests can pin bucket configuration
    /// without touching the process environment.
    fn build(metrics_prefix: Option<String>, env: EnvLookup<'_>) -> Self {
        // TODO: Remove DYN_METRICS_PREFIX env-var override (added in PR #2432 for
        // NIM compatibility with the old "nv_llm_http_service_" prefix). No longer
        // needed — hardcode name_prefix::FRONTEND and drop the sanitize function.
        let raw_prefix = metrics_prefix.unwrap_or_else(|| name_prefix::FRONTEND.to_string());
        let prefix = sanitize_frontend_prometheus_prefix(&raw_prefix);
        if let Err(configured_prefix) = FAILURE_METRICS_PREFIX.set(prefix.clone())
            && configured_prefix != prefix
        {
            tracing::warn!(
                configured=%configured_prefix,
                requested=%prefix,
                "Semantic failure metrics already use a different process-wide prefix"
            );
        }
        if prefix != raw_prefix {
            tracing::warn!(
                raw=%raw_prefix,
                sanitized=%prefix,
                env=%frontend_service::METRICS_PREFIX_ENV,
                "Sanitized HTTP metrics prefix"
            );
        }
        let frontend_metric_name = |suffix: &str| format!("{}_{}", &prefix, suffix);

        let request_counter = IntCounterVec::new(
            Opts::new(
                frontend_metric_name(frontend_service::REQUESTS_TOTAL),
                "Total number of LLM requests processed",
            ),
            &["model", "endpoint", "request_type", "status", "error_type"],
        )
        .unwrap();

        let request_started_counter = IntCounterVec::new(
            Opts::new(
                frontend_metric_name(frontend_service::REQUESTS_STARTED_TOTAL),
                "Total number of LLM requests accepted by the frontend handler",
            ),
            &["model", "endpoint", "request_type"],
        )
        .unwrap();

        let inflight_gauge = IntGaugeVec::new(
            Opts::new(
                frontend_metric_name(frontend_service::INFLIGHT_REQUESTS),
                "Number of inflight requests",
            ),
            &["model"],
        )
        .unwrap();

        let active_requests_gauge = IntGaugeVec::new(
            Opts::new(
                frontend_metric_name(frontend_service::ACTIVE_REQUESTS),
                "Number of requests currently being handled by the frontend, from HTTP handler entry to response completion",
            ),
            &["model"],
        )
        .unwrap();

        let client_disconnect_gauge = prometheus::IntGauge::new(
            frontend_metric_name(frontend_service::DISCONNECTED_CLIENTS),
            "Number of disconnected clients",
        )
        .unwrap();

        let http_queue_gauge = IntGaugeVec::new(
            Opts::new(
                frontend_metric_name(frontend_service::QUEUED_REQUESTS),
                "Number of requests in HTTP processing queue",
            ),
            &["model"],
        )
        .unwrap();

        // Request duration buckets: configurable via DYN_METRICS_REQUEST_DURATION_{MIN,MAX,COUNT}
        let (req_dur_min, req_dur_max, req_dur_count) = parse_bucket_config(
            env,
            env_metrics::DYN_METRICS_REQUEST_DURATION,
            1.0,
            512.0,
            10,
        );
        let request_duration_buckets =
            generate_log_buckets(req_dur_min, req_dur_max, req_dur_count);

        let request_duration = HistogramVec::new(
            HistogramOpts::new(
                frontend_metric_name(frontend_service::REQUEST_DURATION_SECONDS),
                "Duration of LLM requests",
            )
            .buckets(request_duration_buckets),
            &["model"],
        )
        .unwrap();

        // Input sequence length buckets: configurable via DYN_METRICS_INPUT_SEQUENCE_{MIN,MAX,COUNT}
        let (isl_min, isl_max, isl_count) = parse_bucket_config(
            env,
            env_metrics::DYN_METRICS_INPUT_SEQUENCE,
            50.0,
            128000.0,
            12,
        );
        let input_sequence_buckets = generate_log_buckets(isl_min, isl_max, isl_count);

        let input_sequence_length = HistogramVec::new(
            HistogramOpts::new(
                frontend_metric_name(frontend_service::INPUT_SEQUENCE_TOKENS),
                "Input sequence length in tokens",
            )
            .buckets(input_sequence_buckets.clone()),
            &["model"],
        )
        .unwrap();

        // Output sequence length buckets: configurable via DYN_METRICS_OUTPUT_SEQUENCE_{MIN,MAX,COUNT}
        let (osl_min, osl_max, osl_count) = parse_bucket_config(
            env,
            env_metrics::DYN_METRICS_OUTPUT_SEQUENCE,
            50.0,
            32000.0,
            10,
        );
        let output_sequence_buckets = generate_log_buckets(osl_min, osl_max, osl_count);

        let output_sequence_length = HistogramVec::new(
            HistogramOpts::new(
                frontend_metric_name(frontend_service::OUTPUT_SEQUENCE_TOKENS),
                "Output sequence length in tokens",
            )
            .buckets(output_sequence_buckets),
            &["model"],
        )
        .unwrap();

        let output_tokens_counter = IntCounterVec::new(
            Opts::new(
                frontend_metric_name(frontend_service::OUTPUT_TOKENS_TOTAL),
                "Total number of output tokens generated (updates in real-time)",
            ),
            &["model"],
        )
        .unwrap();

        // Time to first token buckets: configurable via DYN_METRICS_TTFT_{MIN,MAX,COUNT}
        let (ttft_min, ttft_max, ttft_count) =
            parse_bucket_config(env, env_metrics::DYN_METRICS_TTFT, 0.001, 480.0, 18);
        let time_to_first_token_buckets = generate_log_buckets(ttft_min, ttft_max, ttft_count);

        let time_to_first_token = HistogramVec::new(
            HistogramOpts::new(
                frontend_metric_name(frontend_service::TIME_TO_FIRST_TOKEN_SECONDS),
                "Time to first token in seconds",
            )
            .buckets(time_to_first_token_buckets),
            &["model"],
        )
        .unwrap();

        // Inter-token latency buckets: configurable via DYN_METRICS_ITL_{MIN,MAX,COUNT}
        let (itl_min, itl_max, itl_count) =
            parse_bucket_config(env, env_metrics::DYN_METRICS_ITL, 0.001, 80.0, 20);
        let inter_token_latency_buckets = generate_log_buckets(itl_min, itl_max, itl_count);

        let inter_token_latency = HistogramVec::new(
            HistogramOpts::new(
                frontend_metric_name(frontend_service::INTER_TOKEN_LATENCY_SECONDS),
                "Inter-token latency in seconds",
            )
            .buckets(inter_token_latency_buckets),
            &["model"],
        )
        .unwrap();

        // Embedding latency buckets: pooling-model inference is typically
        // sub-second (60-200 token inputs on L4-class GPUs land in the 5-50ms
        // range), so 1ms..10s on a log scale gives p50/p99 resolution that
        // the 1..512s `request_duration` buckets cannot.
        let (emb_min, emb_max, emb_count) = parse_bucket_config(
            env,
            env_metrics::DYN_METRICS_EMBEDDING_LATENCY,
            0.001,
            10.0,
            14,
        );
        let embedding_latency_buckets = generate_log_buckets(emb_min, emb_max, emb_count);

        let embedding_latency = HistogramVec::new(
            HistogramOpts::new(
                frontend_metric_name(frontend_service::EMBEDDING_LATENCY_SECONDS),
                "End-to-end latency of /v1/embeddings requests, in seconds. \
                 Distinct from request_duration_seconds so the bucket range can \
                 cover sub-second pooling-model inference.",
            )
            .buckets(embedding_latency_buckets),
            &["model"],
        )
        .unwrap();

        // Per-request media-count buckets: every integer through 10 (where most
        // requests land), then coarse steps for the large-batch tail.
        let multimodal_count_buckets = vec![
            0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0, 20.0, 30.0, 40.0, 50.0, 100.0,
        ];

        let images_per_request = HistogramVec::new(
            HistogramOpts::new(
                frontend_metric_name(frontend_service::IMAGES_PER_REQUEST),
                "Number of image_url content parts per request",
            )
            .buckets(multimodal_count_buckets.clone()),
            &["model"],
        )
        .unwrap();

        let videos_per_request = HistogramVec::new(
            HistogramOpts::new(
                frontend_metric_name(frontend_service::VIDEOS_PER_REQUEST),
                "Number of video_url content parts per request",
            )
            .buckets(multimodal_count_buckets.clone()),
            &["model"],
        )
        .unwrap();

        let audio_per_request = HistogramVec::new(
            HistogramOpts::new(
                frontend_metric_name(frontend_service::AUDIO_PER_REQUEST),
                "Number of audio_url content parts per request",
            )
            .buckets(multimodal_count_buckets),
            &["model"],
        )
        .unwrap();

        // Image placeholder-token buckets: powers of two cover the common
        // single-image range and multi-image request totals.
        let image_token_buckets = vec![
            4.0, 8.0, 16.0, 32.0, 64.0, 128.0, 256.0, 512.0, 1024.0, 2048.0, 4096.0, 8192.0,
            16384.0, 32768.0, 65536.0,
        ];
        let image_tokens_per_request = HistogramVec::new(
            HistogramOpts::new(
                frontend_metric_name(frontend_service::IMAGE_TOKENS_PER_REQUEST),
                "Calculated image-placeholder token count per image-bearing request; \
                 recorded from the response path only when every image resolves and \
                 processor overrides are absent",
            )
            .buckets(image_token_buckets),
            &["model"],
        )
        .unwrap();

        let cached_tokens = HistogramVec::new(
            HistogramOpts::new(
                frontend_metric_name(frontend_service::CACHED_TOKENS),
                "Number of cached tokens (prefix cache hits) per request",
            )
            .buckets(input_sequence_buckets.clone()),
            &["model"],
        )
        .unwrap();

        let tokenizer_latency = HistogramVec::new(
            HistogramOpts::new(
                frontend_metric_name(frontend_service::TOKENIZER_LATENCY_MS),
                "Tokenizer latency in milliseconds",
            )
            .buckets(vec![
                0.5, 1.0, 2.0, 4.0, 8.0, 16.0, 32.0, 64.0, 128.0, 256.0, 512.0,
            ]),
            &[frontend_service::OPERATION_LABEL],
        )
        .unwrap();

        // Runtime configuration metrics
        // Note: Some of these metrics represent counter-like values from source systems,
        // but are implemented as gauges because they are copied/synchronized from upstream
        // counter values rather than being directly incremented.
        let model_total_kv_blocks = IntGaugeVec::new(
            Opts::new(
                frontend_metric_name(frontend_service::MODEL_TOTAL_KV_BLOCKS),
                "Total KV cache blocks available for a worker serving the model",
            ),
            &["model"],
        )
        .unwrap();

        let model_max_num_seqs = IntGaugeVec::new(
            Opts::new(
                frontend_metric_name(frontend_service::MODEL_MAX_NUM_SEQS),
                "Maximum number of sequences for a worker serving the model",
            ),
            &["model"],
        )
        .unwrap();

        let model_max_num_batched_tokens = IntGaugeVec::new(
            Opts::new(
                frontend_metric_name(frontend_service::MODEL_MAX_NUM_BATCHED_TOKENS),
                "Maximum number of batched tokens for a worker serving the model",
            ),
            &["model"],
        )
        .unwrap();

        let model_context_length = IntGaugeVec::new(
            Opts::new(
                frontend_metric_name(frontend_service::MODEL_CONTEXT_LENGTH),
                "Maximum context length in tokens for a worker serving the model",
            ),
            &["model"],
        )
        .unwrap();

        let model_kv_cache_block_size = IntGaugeVec::new(
            Opts::new(
                frontend_metric_name(frontend_service::MODEL_KV_CACHE_BLOCK_SIZE),
                "KV cache block size in tokens for a worker serving the model",
            ),
            &["model"],
        )
        .unwrap();

        let model_migration_limit = IntGaugeVec::new(
            Opts::new(
                frontend_metric_name(frontend_service::MODEL_MIGRATION_LIMIT),
                "Maximum number of request migrations allowed for the model",
            ),
            &["model"],
        )
        .unwrap();

        let model_migration_total = IntCounterVec::new(
            Opts::new(
                frontend_metric_name(frontend_service::MODEL_MIGRATION_TOTAL),
                "Total number of request migrations due to worker unavailability",
            ),
            &["model", frontend_service::MIGRATION_TYPE_LABEL],
        )
        .unwrap();

        let model_migration_duration_seconds = HistogramVec::new(
            HistogramOpts::new(
                frontend_metric_name(frontend_service::MODEL_MIGRATION_DURATION_SECONDS),
                "Time from detecting a migratable failure until recovery, terminal failure, or cancellation",
            )
            .buckets(vec![
                0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 20.0, 40.0, 60.0,
            ]),
            &[
                "model",
                frontend_service::MIGRATION_TYPE_LABEL,
                frontend_service::MIGRATION_OUTCOME_LABEL,
            ],
        )
        .unwrap();

        let model_migration_max_seq_len_exceeded_total = IntCounterVec::new(
            Opts::new(
                frontend_metric_name(frontend_service::MODEL_MIGRATION_MAX_SEQ_LEN_EXCEEDED_TOTAL),
                "Total number of times migration was disabled by max_seq_len limit",
            ),
            &["model"],
        )
        .unwrap();

        let model_cancellation_total = IntCounterVec::new(
            Opts::new(
                frontend_metric_name(frontend_service::MODEL_CANCELLATION_TOTAL),
                "Total number of request cancellations",
            ),
            &["model", "endpoint", "request_type"],
        )
        .unwrap();

        let model_rejection_total = IntCounterVec::new(
            Opts::new(
                frontend_metric_name(frontend_service::MODEL_REJECTION_TOTAL),
                "Total number of requests rejected due to resource exhaustion",
            ),
            &["model", "endpoint"],
        )
        .unwrap();

        Metrics {
            request_started_counter,
            request_counter,
            inflight_gauge,
            active_requests_gauge,
            client_disconnect_gauge,
            http_queue_gauge,
            request_duration,
            input_sequence_length,
            output_sequence_length,
            cached_tokens,
            tokenizer_latency,
            output_tokens_counter,
            time_to_first_token,
            inter_token_latency,
            embedding_latency,
            images_per_request,
            videos_per_request,
            audio_per_request,
            image_tokens_per_request,
            model_total_kv_blocks,
            model_max_num_seqs,
            model_max_num_batched_tokens,
            model_context_length,
            model_kv_cache_block_size,
            model_migration_limit,
            model_migration_total,
            model_migration_duration_seconds,
            model_migration_max_seq_len_exceeded_total,
            model_cancellation_total,
            model_rejection_total,
        }
    }

    /// Get the number of successful requests for the given dimensions:
    /// - model
    /// - endpoint (completions/chat_completions)
    /// - request type (unary/stream)
    /// - status (success/error)
    pub fn get_request_counter(
        &self,
        model: &str,
        endpoint: &Endpoint,
        request_type: &RequestType,
        status: &Status,
        error_type: &ErrorType,
    ) -> u64 {
        self.request_counter
            .with_label_values(&[
                model,
                endpoint.as_str(),
                request_type.as_str(),
                status.as_str(),
                error_type.as_str(),
            ])
            .get()
    }

    /// Increment the counter for requests for the given dimensions:
    /// - model
    /// - endpoint (completions/chat_completions)
    /// - request type (unary/stream)
    /// - status (success/error)
    fn inc_request_counter(
        &self,
        model: &str,
        endpoint: &Endpoint,
        request_type: &RequestType,
        status: &Status,
        error_type: &ErrorType,
    ) {
        self.request_counter
            .with_label_values(&[
                model,
                endpoint.as_str(),
                request_type.as_str(),
                status.as_str(),
                error_type.as_str(),
            ])
            .inc()
    }

    /// Increment the counter for requests accepted by the frontend handler.
    fn inc_request_started_counter(
        &self,
        model: &str,
        endpoint: &Endpoint,
        request_type: &RequestType,
    ) {
        self.request_started_counter
            .with_label_values(&[model, endpoint.as_str(), request_type.as_str()])
            .inc()
    }

    /// Get the number if inflight requests for the given model
    pub fn get_inflight_count(&self, model: &str) -> i64 {
        self.inflight_gauge.with_label_values(&[model]).get()
    }

    /// Record one observation of the end-to-end latency of an
    /// `/v1/embeddings` request, in seconds. Labeled by model.
    ///
    /// Negative values are dropped (they can only happen from a clock skew or
    /// programming bug; recording them would corrupt percentile estimates).
    pub fn observe_embedding_latency(&self, model: &str, seconds: f64) {
        if !seconds.is_finite() || seconds < 0.0 {
            return;
        }
        self.embedding_latency
            .with_label_values(&[model])
            .observe(seconds);
    }

    /// Get the cumulative count of embedding latency observations for the given
    /// model. Test helper.
    #[cfg(test)]
    pub fn embedding_latency_count(&self, model: &str) -> u64 {
        self.embedding_latency
            .with_label_values(&[model])
            .get_sample_count()
    }

    /// Get the cumulative sum (seconds) of embedding latency observations for
    /// the given model. Test helper.
    #[cfg(test)]
    pub fn embedding_latency_sum(&self, model: &str) -> f64 {
        self.embedding_latency
            .with_label_values(&[model])
            .get_sample_sum()
    }

    fn inc_inflight_gauge(&self, model: &str) {
        self.inflight_gauge.with_label_values(&[model]).inc();
        self.active_requests_gauge.with_label_values(&[model]).inc();
    }

    fn dec_inflight_gauge(&self, model: &str) {
        self.inflight_gauge.with_label_values(&[model]).dec();
        self.active_requests_gauge.with_label_values(&[model]).dec();
    }

    /// Increment the gauge for client disconnections
    pub fn inc_client_disconnect(&self) {
        self.client_disconnect_gauge.inc();
    }

    /// Get the count of client disconnections
    pub fn get_client_disconnect_count(&self) -> i64 {
        self.client_disconnect_gauge.get()
    }

    fn inc_http_queue_gauge(&self, model: &str) {
        self.http_queue_gauge.with_label_values(&[model]).inc()
    }

    fn dec_http_queue_gauge(&self, model: &str) {
        self.http_queue_gauge.with_label_values(&[model]).dec()
    }

    pub fn register(&self, registry: &Registry) -> Result<(), prometheus::Error> {
        registry.register(Box::new(self.request_started_counter.clone()))?;
        registry.register(Box::new(self.request_counter.clone()))?;
        registry.register(Box::new(self.inflight_gauge.clone()))?;
        registry.register(Box::new(self.active_requests_gauge.clone()))?;
        registry.register(Box::new(self.client_disconnect_gauge.clone()))?;
        registry.register(Box::new(self.http_queue_gauge.clone()))?;
        registry.register(Box::new(self.request_duration.clone()))?;
        registry.register(Box::new(self.input_sequence_length.clone()))?;
        registry.register(Box::new(self.output_sequence_length.clone()))?;
        registry.register(Box::new(self.cached_tokens.clone()))?;
        registry.register(Box::new(self.tokenizer_latency.clone()))?;
        registry.register(Box::new(self.output_tokens_counter.clone()))?;
        registry.register(Box::new(self.time_to_first_token.clone()))?;
        registry.register(Box::new(self.inter_token_latency.clone()))?;
        registry.register(Box::new(self.embedding_latency.clone()))?;
        registry.register(Box::new(self.images_per_request.clone()))?;
        registry.register(Box::new(self.videos_per_request.clone()))?;
        registry.register(Box::new(DYNAM_FAILURES_TOTAL.clone()))?;
        registry.register(Box::new(self.audio_per_request.clone()))?;
        registry.register(Box::new(self.image_tokens_per_request.clone()))?;

        // Register runtime configuration metrics
        registry.register(Box::new(self.model_total_kv_blocks.clone()))?;
        registry.register(Box::new(self.model_max_num_seqs.clone()))?;
        registry.register(Box::new(self.model_max_num_batched_tokens.clone()))?;
        registry.register(Box::new(self.model_context_length.clone()))?;
        registry.register(Box::new(self.model_kv_cache_block_size.clone()))?;
        registry.register(Box::new(self.model_migration_limit.clone()))?;
        registry.register(Box::new(self.model_migration_total.clone()))?;
        registry.register(Box::new(self.model_migration_duration_seconds.clone()))?;
        registry.register(Box::new(
            self.model_migration_max_seq_len_exceeded_total.clone(),
        ))?;
        registry.register(Box::new(self.model_cancellation_total.clone()))?;
        registry.register(Box::new(self.model_rejection_total.clone()))?;

        Ok(())
    }

    /// Update runtime configuration metrics for a model
    /// This should be called when model runtime configuration is available or updated
    pub fn update_runtime_config_metrics(
        &self,
        model_name: &str,
        runtime_config: &ModelRuntimeConfig,
    ) {
        if let Some(total_kv_blocks) = runtime_config.total_kv_blocks {
            self.model_total_kv_blocks
                .with_label_values(&[model_name])
                .set(clamp_u64_to_i64(total_kv_blocks));
        }

        if let Some(max_num_seqs) = runtime_config.max_num_seqs {
            self.model_max_num_seqs
                .with_label_values(&[model_name])
                .set(clamp_u64_to_i64(max_num_seqs));
        }

        if let Some(max_batched_tokens) = runtime_config.max_num_batched_tokens {
            self.model_max_num_batched_tokens
                .with_label_values(&[model_name])
                .set(clamp_u64_to_i64(max_batched_tokens));
        }
    }

    /// Update metrics from a ModelDeploymentCard
    /// This updates both runtime config metrics and MDC-specific metrics
    pub fn update_metrics_from_mdc(&self, card: &ModelDeploymentCard) -> anyhow::Result<()> {
        self.update_runtime_config_metrics(&card.display_name, &card.runtime_config);

        self.model_context_length
            .with_label_values(&[&card.display_name])
            .set(card.effective_context_length() as i64);

        self.model_kv_cache_block_size
            .with_label_values(&[&card.display_name])
            .set(card.kv_cache_block_size as i64);

        self.model_migration_limit
            .with_label_values(&[&card.display_name])
            .set(card.migration_limit as i64);

        tracing::debug!(
            model = %card.display_name,
            "Successfully updated MDC metrics"
        );

        Ok(())
    }

    /// Increment the migration counter for a new request migration
    pub fn inc_migration_new_request(&self, model: &str) {
        self.model_migration_total
            .with_label_values(&[model, frontend_service::migration_type::NEW_REQUEST])
            .inc();
    }

    /// Increment the migration counter for an ongoing request migration
    pub fn inc_migration_ongoing_request(&self, model: &str) {
        self.model_migration_total
            .with_label_values(&[model, frontend_service::migration_type::ONGOING_REQUEST])
            .inc();
    }

    /// Get the current count of new request migrations for a model
    pub fn get_migration_new_request_count(&self, model: &str) -> u64 {
        self.model_migration_total
            .with_label_values(&[model, frontend_service::migration_type::NEW_REQUEST])
            .get()
    }

    /// Get the current count of ongoing request migrations for a model
    pub fn get_migration_ongoing_request_count(&self, model: &str) -> u64 {
        self.model_migration_total
            .with_label_values(&[model, frontend_service::migration_type::ONGOING_REQUEST])
            .get()
    }

    /// Observe the elapsed time for a completed migration event.
    pub fn observe_migration_duration(
        &self,
        model: &str,
        migration_type: &str,
        outcome: &str,
        duration: Duration,
    ) {
        self.model_migration_duration_seconds
            .with_label_values(&[model, migration_type, outcome])
            .observe(duration.as_secs_f64());
    }

    /// Get the number of observed migration durations for the given dimensions.
    pub fn get_migration_duration_sample_count(
        &self,
        model: &str,
        migration_type: &str,
        outcome: &str,
    ) -> u64 {
        self.model_migration_duration_seconds
            .with_label_values(&[model, migration_type, outcome])
            .get_sample_count()
    }

    /// Increment the counter for migrations disabled by max_seq_len being exceeded
    pub fn inc_migration_max_seq_len_exceeded(&self, model: &str) {
        self.model_migration_max_seq_len_exceeded_total
            .with_label_values(&[model])
            .inc();
    }

    /// Get the current count of migrations disabled by max_seq_len being exceeded
    pub fn get_migration_max_seq_len_exceeded_count(&self, model: &str) -> u64 {
        self.model_migration_max_seq_len_exceeded_total
            .with_label_values(&[model])
            .get()
    }

    /// Increment the cancellation counter
    pub fn inc_cancellation(&self, labels: &CancellationLabels) {
        self.model_cancellation_total
            .with_label_values(&[&labels.model, &labels.endpoint, &labels.request_type])
            .inc();
    }

    /// Get the current cancellation count
    pub fn get_cancellation_count(&self, labels: &CancellationLabels) -> u64 {
        self.model_cancellation_total
            .with_label_values(&[&labels.model, &labels.endpoint, &labels.request_type])
            .get()
    }

    /// Increment the rejection counter for a request rejected due to resource exhaustion
    pub fn inc_rejection(&self, model: &str, endpoint: Endpoint) {
        self.model_rejection_total
            .with_label_values(&[model, &endpoint.to_string()])
            .inc();
    }

    /// Get the current rejection count for a model and endpoint
    pub fn get_rejection_count(&self, model: &str, endpoint: Endpoint) -> u64 {
        self.model_rejection_total
            .with_label_values(&[model, &endpoint.to_string()])
            .get()
    }

    /// Create a new [`InflightGuard`] for the given model and annotate if its a streaming request,
    /// and the kind of endpoint that was hit
    ///
    /// The [`InflightGuard`] is an RAII object will handle incrementing the inflight gauge and
    /// request counters.
    ///
    /// # Metrics Distinction
    ///
    /// This method creates an inflight guard  t tracks requests actively being processed by the LLM engine.
    /// This is distinct from [`HttpQueueGuard`] which tracks requests from HTTP handler start until
    /// first token generation (including prefill time). The separation allows monitoring both HTTP processing queue time
    /// and actual LLM processing time.
    pub fn create_inflight_guard(
        self: Arc<Self>,
        model: &str,
        endpoint: Endpoint,
        streaming: bool,
        request_id: &str,
    ) -> InflightGuard {
        let request_type = if streaming {
            RequestType::Stream
        } else {
            RequestType::Unary
        };

        InflightGuard::new(
            self.clone(),
            model.to_string(),
            endpoint,
            request_type,
            request_id.to_string(),
        )
    }

    /// Create a new [`ResponseMetricCollector`] for collecting per-response metrics (i.e., TTFT, ITL)
    pub fn create_response_collector(self: Arc<Self>, model: &str) -> ResponseMetricCollector {
        ResponseMetricCollector::new(self, model.to_string())
    }

    /// Create a new [`HttpQueueGuard`] for tracking HTTP processing queue
    ///
    /// This guard tracks requests from HTTP handler start until first token generation,
    /// providing visibility into HTTP processing queue time before actual LLM processing begins.
    pub fn create_http_queue_guard(self: Arc<Self>, model: &str) -> HttpQueueGuard {
        HttpQueueGuard::new(self, model.to_string())
    }
}

impl HttpQueueGuard {
    fn new(metrics: Arc<Metrics>, model: String) -> Self {
        // Increment the HTTP queue gauge when the guard is created
        metrics.inc_http_queue_gauge(&model);

        HttpQueueGuard { metrics, model }
    }
}

impl Drop for HttpQueueGuard {
    fn drop(&mut self) {
        // Decrement the HTTP queue gauge when the guard is dropped
        self.metrics.dec_http_queue_gauge(&self.model);
    }
}

impl InflightGuard {
    fn new(
        metrics: Arc<Metrics>,
        model: String,
        endpoint: Endpoint,
        request_type: RequestType,
        request_id: String,
    ) -> Self {
        let timer = Instant::now();
        metrics.inc_inflight_gauge(&model);
        metrics.inc_request_started_counter(&model, &endpoint, &request_type);

        tracing::Span::current().record("model", model.as_str());

        tracing::info!(
            request_id = %request_id,
            model = %model,
            endpoint = %endpoint,
            request_type = %request_type,
            "request received"
        );

        InflightGuard {
            metrics,
            model,
            endpoint,
            request_type,
            status: Status::Error,
            error_type: ErrorType::Internal,
            timer,
            request_id,
            span: tracing::Span::current(),
        }
    }

    pub fn request_id(&self) -> &str {
        &self.request_id
    }
    pub fn model(&self) -> &str {
        &self.model
    }
    pub fn endpoint(&self) -> &Endpoint {
        &self.endpoint
    }
    pub fn request_type(&self) -> &RequestType {
        &self.request_type
    }
    pub fn error_type(&self) -> &ErrorType {
        &self.error_type
    }
    pub fn elapsed_ms(&self) -> u128 {
        self.timer.elapsed().as_millis()
    }

    pub(crate) fn mark_ok(&mut self) {
        self.status = Status::Success;
        self.error_type = ErrorType::None;
    }

    pub(crate) fn mark_error(&mut self, error_type: ErrorType) {
        self.status = Status::Error;
        self.error_type = error_type;
    }

    fn request_completion(&self) -> RequestCompletion {
        match (&self.status, &self.error_type) {
            (Status::Success, _) => RequestCompletion::Success,
            (Status::Error, ErrorType::Cancelled) => RequestCompletion::Cancelled,
            (Status::Error, _) => RequestCompletion::Error,
        }
    }
}

impl Drop for InflightGuard {
    fn drop(&mut self) {
        let _enter = self.span.enter();
        let duration = self.timer.elapsed().as_secs_f64();
        self.metrics.dec_inflight_gauge(&self.model);
        self.metrics.inc_request_counter(
            &self.model,
            &self.endpoint,
            &self.request_type,
            &self.status,
            &self.error_type,
        );
        self.metrics
            .request_duration
            .with_label_values(&[&self.model])
            .observe(duration);

        let completion = self.request_completion();
        self.span.record("request.outcome", completion.as_str());

        let elapsed_ms = (duration * 1000.0) as u64;
        let status_str = self.status.as_str();
        match completion {
            RequestCompletion::Error => {
                let detail = match self.error_type {
                    ErrorType::ResponseTimeout => "backend stream inactivity timeout",
                    ErrorType::Internal => "internal server error during processing",
                    ErrorType::Validation => "invalid request parameters",
                    ErrorType::NotFound => "model or resource not found",
                    ErrorType::Overload => "service overloaded or rate limited",
                    ErrorType::Unavailable => "no backend worker available",
                    ErrorType::NotImplemented => "requested feature not implemented",
                    // `request_completion()` routes `(Error, Cancelled)` to
                    // `RequestCompletion::Cancelled`, so only `None` reaches
                    // here. `Cancelled` is listed to keep the match total
                    // without a panic in a `Drop` impl.
                    ErrorType::None | ErrorType::Cancelled => "unknown error",
                };
                tracing::error!(
                    request_id = %self.request_id,
                    model = %self.model,
                    endpoint = %self.endpoint,
                    request_type = %self.request_type,
                    status = %status_str,
                    error_type = %self.error_type,
                    error_detail = %detail,
                    elapsed_ms = %elapsed_ms,
                    "request completed"
                );
            }
            RequestCompletion::Cancelled => {
                tracing::info!(
                    request_id = %self.request_id,
                    model = %self.model,
                    endpoint = %self.endpoint,
                    request_type = %self.request_type,
                    status = %completion.as_str(),
                    error_type = %self.error_type,
                    error_detail = "cancelled before completion",
                    elapsed_ms = %elapsed_ms,
                    "request completed"
                );
            }
            RequestCompletion::Success => {
                tracing::info!(
                    request_id = %self.request_id,
                    model = %self.model,
                    endpoint = %self.endpoint,
                    request_type = %self.request_type,
                    status = %status_str,
                    elapsed_ms = %elapsed_ms,
                    "request completed"
                );
            }
        }
    }
}

impl std::fmt::Display for Endpoint {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Endpoint::Completions => write!(f, "completions"),
            Endpoint::ChatCompletions => write!(f, "chat_completions"),
            Endpoint::Embeddings => write!(f, "embeddings"),
            Endpoint::Classify => write!(f, "classify"),
            Endpoint::Rerank => write!(f, "rerank"),
            Endpoint::Pooling => write!(f, "pooling"),
            Endpoint::Images => write!(f, "images"),
            Endpoint::Videos => write!(f, "videos"),
            Endpoint::Audios => write!(f, "audios"),
            Endpoint::Responses => write!(f, "responses"),
            Endpoint::AnthropicMessages => write!(f, "anthropic_messages"),
            Endpoint::Tensor => write!(f, "tensor"),
            Endpoint::Generate => write!(f, "generate"),
        }
    }
}

impl Endpoint {
    pub fn as_str(&self) -> &'static str {
        match self {
            Endpoint::Completions => "completions",
            Endpoint::ChatCompletions => "chat_completions",
            Endpoint::Embeddings => "embeddings",
            Endpoint::Classify => "classify",
            Endpoint::Rerank => "rerank",
            Endpoint::Pooling => "pooling",
            Endpoint::Images => "images",
            Endpoint::Videos => "videos",
            Endpoint::Audios => "audios",
            Endpoint::Responses => "responses",
            Endpoint::AnthropicMessages => "anthropic_messages",
            Endpoint::Tensor => "tensor",
            Endpoint::Generate => "generate",
        }
    }
}

impl RequestType {
    pub fn as_str(&self) -> &'static str {
        match self {
            RequestType::Unary => frontend_service::request_type::UNARY,
            RequestType::Stream => frontend_service::request_type::STREAM,
        }
    }
}

impl std::fmt::Display for RequestType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}

impl Status {
    pub fn as_str(&self) -> &'static str {
        match self {
            Status::Success => frontend_service::status::SUCCESS,
            Status::Error => frontend_service::status::ERROR,
        }
    }
}

impl ErrorType {
    pub fn as_str(&self) -> &'static str {
        match self {
            ErrorType::None => frontend_service::error_type::NONE,
            ErrorType::Validation => frontend_service::error_type::VALIDATION,
            ErrorType::NotFound => frontend_service::error_type::NOT_FOUND,
            ErrorType::Overload => frontend_service::error_type::OVERLOAD,
            ErrorType::Unavailable => frontend_service::error_type::UNAVAILABLE,
            ErrorType::Cancelled => frontend_service::error_type::CANCELLED,
            ErrorType::ResponseTimeout => frontend_service::error_type::RESPONSE_TIMEOUT,
            ErrorType::Internal => frontend_service::error_type::INTERNAL,
            ErrorType::NotImplemented => frontend_service::error_type::NOT_IMPLEMENTED,
        }
    }
}

impl std::fmt::Display for ErrorType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}

impl ResponseMetricCollector {
    fn new(metrics: Arc<Metrics>, model: String) -> Self {
        // Resolve the per-model handles once (cheap clones of the vec entries) so the
        // per-chunk / per-token hot path in `observe_response` does no label hashing.
        let output_tokens_counter = metrics.output_tokens_counter.with_label_values(&[&model]);
        let time_to_first_token = metrics.time_to_first_token.with_label_values(&[&model]);
        let input_sequence_length = metrics.input_sequence_length.with_label_values(&[&model]);
        let cached_tokens = metrics.cached_tokens.with_label_values(&[&model]);
        let images_per_request = metrics.images_per_request.with_label_values(&[&model]);
        let videos_per_request = metrics.videos_per_request.with_label_values(&[&model]);
        let audio_per_request = metrics.audio_per_request.with_label_values(&[&model]);
        let image_tokens_per_request = metrics
            .image_tokens_per_request
            .with_label_values(&[&model]);
        ResponseMetricCollector {
            metrics,
            model,
            output_tokens_counter,
            time_to_first_token,
            inter_token_latency: None,
            itl_pending_tokens: 0,
            input_sequence_length,
            cached_tokens,
            images_per_request,
            videos_per_request,
            audio_per_request,
            image_tokens_per_request,
            image_count_val: 0,
            video_count_val: 0,
            audio_count_val: 0,
            multimodal_counts_observed: false,
            is_first_token: true,
            last_response_time: None,
            start_time: Instant::now(),
            osl: 0,
            isl: 0,
            ttft_ms: None,
            itl_sum_secs: 0.0,
            itl_count: 0,
            cached_tokens_observed: false,
            tokenize_latency_observed: false,
            detokenize_latency_total: Duration::ZERO,
            detokenize_count_total: 0,
            prefill_worker_id: None,
            prefill_dp_rank: None,
            prefill_worker_type: None,
            decode_worker_id: None,
            decode_dp_rank: None,
            decode_worker_type: None,
            decode_itl_gauge: None,
        }
    }

    /// Set the worker info for per-worker TTFT/ITL metrics.
    /// In disaggregated mode, TTFT is attributed to prefill worker, ITL to decode worker.
    /// Worker types are stored at routing time to avoid expensive MDC lookup when updating metrics.
    pub fn set_worker_info(
        &mut self,
        prefill_worker_id: Option<u64>,
        prefill_dp_rank: Option<u32>,
        prefill_worker_type: Option<String>,
        decode_worker_id: Option<u64>,
        decode_dp_rank: Option<u32>,
        decode_worker_type: Option<String>,
    ) {
        if self.prefill_worker_id.is_none() {
            self.prefill_worker_id = prefill_worker_id;
        }
        if self.prefill_dp_rank.is_none() {
            self.prefill_dp_rank = prefill_dp_rank;
        }
        if self.prefill_worker_type.is_none() {
            self.prefill_worker_type = prefill_worker_type;
        }
        if self.decode_worker_id.is_none() {
            self.decode_worker_id = decode_worker_id;
        }
        if self.decode_dp_rank.is_none() {
            self.decode_dp_rank = decode_dp_rank;
        }
        if self.decode_worker_type.is_none() {
            self.decode_worker_type = decode_worker_type;
        }
    }

    fn set_worker_info_from_metrics(&mut self, metrics: &LLMMetricAnnotation) {
        let prefill_worker_type = self
            .prefill_worker_type
            .is_none()
            .then(|| metrics.prefill_worker_type.clone())
            .flatten();
        let decode_worker_type = self
            .decode_worker_type
            .is_none()
            .then(|| metrics.decode_worker_type.clone())
            .flatten();

        self.set_worker_info(
            metrics.prefill_worker_id,
            metrics.prefill_dp_rank,
            prefill_worker_type,
            metrics.decode_worker_id,
            metrics.decode_dp_rank,
            decode_worker_type,
        );
    }

    fn local_inter_token_latency(&mut self) -> &mut prometheus::local::LocalHistogram {
        let metrics = &self.metrics;
        let model = self.model.as_str();
        self.inter_token_latency.get_or_insert_with(|| {
            metrics
                .inter_token_latency
                .with_label_values(&[model])
                .local()
        })
    }

    /// Observe the current output sequence length
    pub fn observe_current_osl(&mut self, osl: usize) {
        self.osl = osl;
    }

    /// Check if this will be the first token (before calling observe_response)
    pub fn is_first_token(&self) -> bool {
        self.is_first_token
    }

    /// Observe cached tokens (prefix cache hits), observing only once per request when value is available
    pub fn observe_cached_tokens(&mut self, cached_tokens: Option<usize>) {
        if let Some(tokens) = cached_tokens
            && !self.cached_tokens_observed
        {
            self.cached_tokens_observed = true;
            self.cached_tokens.observe(tokens as f64);
        }
    }

    /// Observe per-request multimodal content-part counts, latched to run
    /// exactly once per request (the counts are constant across chunks).
    pub fn observe_multimodal_counts(
        &mut self,
        image_count: usize,
        video_count: usize,
        audio_count: usize,
    ) {
        self.observe_multimodal_metrics(image_count, video_count, audio_count, None);
    }

    fn observe_multimodal_metrics(
        &mut self,
        image_count: usize,
        video_count: usize,
        audio_count: usize,
        image_tokens: Option<usize>,
    ) {
        if self.multimodal_counts_observed {
            return;
        }
        self.multimodal_counts_observed = true;

        self.image_count_val = image_count;
        self.video_count_val = video_count;
        self.audio_count_val = audio_count;

        self.images_per_request.observe(image_count as f64);
        self.videos_per_request.observe(video_count as f64);
        self.audio_per_request.observe(audio_count as f64);
        if let Some(image_tokens) = image_tokens {
            self.image_tokens_per_request.observe(image_tokens as f64);
        }
    }

    /// Observe tokenize/detokenize latencies in milliseconds.
    /// Tokenize is observed once per request; detokenize is accumulated and observed at request end.
    pub fn observe_tokenize_latencies(
        &mut self,
        tokenize_latency: Option<Duration>,
        detokenize_latency: Option<Duration>,
        detokenize_count: Option<u64>,
    ) {
        if let Some(latency) = tokenize_latency
            && !self.tokenize_latency_observed
        {
            self.tokenize_latency_observed = true;
            self.metrics
                .tokenizer_latency
                .with_label_values(&[frontend_service::operation::TOKENIZE])
                .observe(latency.as_secs_f64() * 1000.0);
        }

        if let Some(latency) = detokenize_latency {
            self.detokenize_latency_total = latency;
        }
        if let Some(count) = detokenize_count {
            self.detokenize_count_total = count;
        }
    }

    /// Observe a response with input sequence length and number of new tokens
    pub fn observe_response(&mut self, isl: usize, num_tokens: usize) {
        if num_tokens == 0 {
            return;
        }

        // Store ISL for span recording on drop
        self.isl = isl;

        // Increment the real-time output tokens counter
        self.output_tokens_counter.inc_by(num_tokens as u64);

        if self.is_first_token {
            // NOTE: when there are multiple tokens in the first response,
            // we use the full response time as TTFT and ignore the ITL
            self.is_first_token = false;

            // Publish TTFT and store for span recording
            let ttft = self.start_time.elapsed().as_secs_f64();
            self.ttft_ms = Some(ttft * 1000.0);
            self.time_to_first_token.observe(ttft);

            // Update per-worker TTFT and input sequence tokens gauges - attributed to prefill worker.
            // Both gauges are updated atomically from the same request to correlate latency with input size.
            // Use stored worker_type (from routing time) to avoid MDC lookup.
            // Falls back to WORKER_TYPE_PREFILL if not available.
            if let Some(worker_id) = self.prefill_worker_id {
                let worker_id_str = worker_id.to_string();
                let dp_rank_str = self
                    .prefill_dp_rank
                    .map_or(UNSET_DP_RANK_LABEL.to_string(), |r| r.to_string());
                let worker_type = self
                    .prefill_worker_type
                    .as_deref()
                    .unwrap_or(WORKER_TYPE_PREFILL);
                let labels = &[worker_id_str.as_str(), dp_rank_str.as_str(), worker_type];
                WORKER_LAST_TIME_TO_FIRST_TOKEN_GAUGE
                    .with_label_values(labels)
                    .set(ttft);
                WORKER_LAST_INPUT_SEQUENCE_TOKENS_GAUGE
                    .with_label_values(labels)
                    .set(isl as i64);
            }

            // Publish ISL
            // TODO: publish ISL as soon as the tokenization process completes
            self.input_sequence_length.observe(isl as f64);
        }

        let current_duration = self.start_time.elapsed();

        if let Some(last_response_time) = self.last_response_time {
            let response_duration = current_duration - last_response_time;
            let itl = response_duration.as_secs_f64() / num_tokens as f64;
            self.itl_sum_secs += itl * num_tokens as f64;
            self.itl_count += num_tokens as u64;
            self.itl_pending_tokens = self.itl_pending_tokens.saturating_add(num_tokens as u64);
            let should_flush = self.itl_pending_tokens >= ITL_LOCAL_FLUSH_TOKENS;
            {
                // Resolve the request-local histogram on the first ITL only, then reuse
                // it without re-hashing the model label for each output token.
                let histogram = self.local_inter_token_latency();
                for _ in 0..num_tokens {
                    histogram.observe(itl);
                }
                if should_flush {
                    histogram.flush();
                }
            }
            if should_flush {
                self.itl_pending_tokens = 0;
            }

            // Update per-worker ITL gauge - attributed to decode worker.
            // Use stored worker_type (from routing time) to avoid MDC lookup.
            // Falls back to WORKER_TYPE_DECODE if not available.
            //
            // The worker labels are fixed for the request's lifetime (latched in
            // `set_worker_info`), so resolve the GaugeVec handle once and cache it.
            // This keeps the per-token path free of `to_string()` allocations and
            // label hashing; subsequent tokens only call `set`.
            if self.decode_itl_gauge.is_none()
                && let Some(worker_id) = self.decode_worker_id
            {
                let worker_id_str = worker_id.to_string();
                let dp_rank_str = self
                    .decode_dp_rank
                    .map_or(UNSET_DP_RANK_LABEL.to_string(), |r| r.to_string());
                let worker_type = self
                    .decode_worker_type
                    .as_deref()
                    .unwrap_or(WORKER_TYPE_DECODE);
                self.decode_itl_gauge =
                    Some(WORKER_LAST_INTER_TOKEN_LATENCY_GAUGE.with_label_values(&[
                        worker_id_str.as_str(),
                        dp_rank_str.as_str(),
                        worker_type,
                    ]));
            }
            if let Some(gauge) = &self.decode_itl_gauge {
                gauge.set(itl);
            }
        }

        self.last_response_time = Some(current_duration);
    }
}

impl Drop for ResponseMetricCollector {
    fn drop(&mut self) {
        if let Some(histogram) = &self.inter_token_latency {
            histogram.flush();
        }
        self.itl_pending_tokens = 0;

        if !self.detokenize_latency_total.is_zero() && self.detokenize_count_total > 0 {
            let avg_detokenize_latency_ms = (self.detokenize_latency_total.as_secs_f64() * 1000.0)
                / self.detokenize_count_total as f64;
            self.metrics
                .tokenizer_latency
                .with_label_values(&[frontend_service::operation::DETOKENIZE])
                .observe(avg_detokenize_latency_ms);
        }

        // Publish final OSL when the collector is dropped, but only for
        // requests that actually produced output tokens. Recording zero for
        // failed/cancelled requests would corrupt histogram averages (sum/count)
        // and percentiles. Failures are already tracked by requests_total with
        // status and error_type labels.
        if self.osl > 0 {
            self.metrics
                .output_sequence_length
                .with_label_values(&[&self.model])
                .observe(self.osl as f64);
        }

        // Record request summary on the enclosing span.
        // InflightGuard::Drop and on_response logs will inherit these.
        let span = tracing::Span::current();
        span.record("input_tokens", self.isl as u32);
        span.record("output_tokens", self.osl as u32);
        // Only record once observed, so requests that never carried a metrics
        // annotation (e.g. errored before the first chunk) don't stamp a false zero.
        if self.multimodal_counts_observed {
            span.record("image_count", self.image_count_val as u32);
            span.record("video_count", self.video_count_val as u32);
            span.record("audio_count", self.audio_count_val as u32);
        }
        if let Some(ttft_ms) = self.ttft_ms {
            span.record("ttft_ms", format!("{:.2}", ttft_ms).as_str());
        }
        if self.itl_count > 0 {
            let avg_ms = (self.itl_sum_secs / self.itl_count as f64) * 1000.0;
            span.record("avg_itl_ms", format!("{:.2}", avg_ms).as_str());
        }
        if let Some(worker_id) = self.prefill_worker_id {
            span.record("prefill_worker_id", worker_id);
        }
        if let Some(worker_id) = self.decode_worker_id {
            span.record("decode_worker_id", worker_id);
        }
    }
}

/// Process streaming metrics for annotated responses
///
/// This function handles metrics collection and http_queue_guard management for streaming responses.
/// It observes the current output sequence length, drops the http_queue_guard on the first token,
/// and records response metrics.
pub fn process_response_and_observe_metrics<T>(
    annotated: &crate::types::Annotated<T>,
    response_collector: &mut ResponseMetricCollector,
    http_queue_guard: &mut Option<HttpQueueGuard>,
) {
    if let Ok(Some(metrics)) = LLMMetricAnnotation::from_annotation(annotated) {
        observe_llm_metrics(&metrics, response_collector, http_queue_guard);
    }
}

/// Process streaming metrics for chat-derived responses.
///
/// The typed metrics field is the hot path. Annotation parsing remains as a compatibility
/// fallback for legacy/generated annotation frames.
pub fn process_chat_response_and_observe_metrics(
    annotated: &crate::types::Annotated<NvCreateChatCompletionStreamResponse>,
    response_collector: &mut ResponseMetricCollector,
    http_queue_guard: &mut Option<HttpQueueGuard>,
) {
    if let Some(metrics) = annotated
        .data
        .as_ref()
        .and_then(|data| data.llm_metrics.as_ref())
    {
        observe_llm_metrics(metrics, response_collector, http_queue_guard);
    } else {
        process_response_and_observe_metrics(annotated, response_collector, http_queue_guard);
    }
}

/// Event converter wrapper for streaming responses
pub struct EventConverter<T>(pub crate::types::Annotated<T>);

impl<T> From<crate::types::Annotated<T>> for EventConverter<T> {
    fn from(annotated: crate::types::Annotated<T>) -> Self {
        EventConverter(annotated)
    }
}

fn sse_json_data<T: Serialize>(event: Event, data: &T) -> Result<Event, axum::Error> {
    // serde_json escapes literal LF/CR in string content, so the resulting JSON
    // is one SSE data line and can avoid Axum json_data's per-write filtering.
    let json = serde_json::to_string(data).map_err(axum::Error::new)?;
    Ok(event.data(json))
}

fn observe_llm_metrics(
    metrics: &LLMMetricAnnotation,
    response_collector: &mut ResponseMetricCollector,
    http_queue_guard: &mut Option<HttpQueueGuard>,
) {
    response_collector.observe_current_osl(metrics.output_tokens);
    response_collector.observe_cached_tokens(metrics.cached_tokens);
    response_collector.observe_multimodal_metrics(
        metrics.image_count,
        metrics.video_count,
        metrics.audio_count,
        metrics.image_tokens,
    );
    response_collector.observe_tokenize_latencies(
        metrics.tokenize_latency,
        metrics.detokenize_total_latency,
        metrics.detokenize_count,
    );
    response_collector.set_worker_info_from_metrics(metrics);

    if response_collector.is_first_token()
        && metrics.chunk_tokens > 0
        && let Some(guard) = http_queue_guard.take()
    {
        drop(guard);
    }

    response_collector.observe_response(metrics.input_tokens, metrics.chunk_tokens);
}

fn observe_annotation_metrics<T>(
    annotated: &crate::types::Annotated<T>,
    response_collector: &mut ResponseMetricCollector,
    http_queue_guard: &mut Option<HttpQueueGuard>,
) -> bool {
    if let Ok(Some(metrics)) = LLMMetricAnnotation::from_annotation(annotated) {
        observe_llm_metrics(&metrics, response_collector, http_queue_guard);
        true
    } else {
        false
    }
}

fn annotated_to_sse_event<T: Serialize>(
    annotated: crate::types::Annotated<T>,
) -> Result<Option<Event>, axum::Error> {
    let mut event = Event::default();

    if let Some(ref data) = annotated.data {
        event = sse_json_data(event, data)?;
    }

    if let Some(ref msg) = annotated.event {
        if msg == "error" {
            let error_message = if let Some(ref dynamo_err) = annotated.error {
                format!(
                    "semantic stream error: class={} reason={}",
                    dynamo_err.class(),
                    dynamo_err.reason()
                )
            } else {
                "backend stream error".to_string()
            };
            return Err(axum::Error::new(error_message));
        }
        event = event.event(msg);
    }

    if let Some(comments) = annotated.comment {
        for comment in comments {
            // Axum's Event::comment() panics on \n / \r
            event = event.comment(comment.replace(['\n', '\r'], " "));
        }
    }

    // Filter out metrics annotation events (events without SSE data payload)
    if annotated.data.is_none() && annotated.event.is_none() {
        Ok(None)
    } else {
        Ok(Some(event))
    }
}

/// Process streaming response with event conversion for SSE
///
/// This function handles metrics collection, http_queue_guard management, and converts
/// annotated responses to SSE events for streaming responses.
///
/// Returns None for metrics annotation events (events without SSE data payload).
pub fn process_response_using_event_converter_and_observe_metrics<T: Serialize>(
    annotated: EventConverter<T>,
    response_collector: &mut ResponseMetricCollector,
    http_queue_guard: &mut Option<HttpQueueGuard>,
) -> Result<Option<Event>, axum::Error> {
    let mut annotated = annotated.0;

    if observe_annotation_metrics(&annotated, response_collector, http_queue_guard) {
        // Preserve the previous SSE behavior: observe legacy metrics annotation
        // frames internally, then strip them before building the outbound event.
        if annotated.event.as_deref() == Some(ANNOTATION_LLM_METRICS) {
            annotated.event = None;
            annotated.comment = None;
        }
    }

    // ANNOTATION_PAYLOAD_USAGE is payload-only and must never reach the client SSE
    // stream (the payload DeltaAggregator consumed its usage upstream).
    if annotated.event.as_deref() == Some(crate::preprocessor::ANNOTATION_PAYLOAD_USAGE) {
        annotated.event = None;
        annotated.comment = None;
        annotated.data = None;
    }

    annotated_to_sse_event(annotated)
}

/// Process chat-derived streaming responses with typed metrics before SSE conversion.
///
/// If the typed field is absent, this falls back to the same legacy annotation
/// frame parsing and stripping behavior as `process_response_using_event_converter_and_observe_metrics`.
pub fn process_chat_response_using_event_converter_and_observe_metrics(
    annotated: EventConverter<NvCreateChatCompletionStreamResponse>,
    response_collector: &mut ResponseMetricCollector,
    http_queue_guard: &mut Option<HttpQueueGuard>,
    reasoning_field: ReasoningField,
) -> Result<Option<Event>, axum::Error> {
    let mut annotated = annotated.0;
    if let Some(data) = annotated.data.as_mut() {
        data.tool_call_completion.clear();
    }

    if let Some(metrics) = annotated
        .data
        .as_ref()
        .and_then(|data| data.llm_metrics.as_ref())
    {
        observe_llm_metrics(metrics, response_collector, http_queue_guard);
    } else if observe_annotation_metrics(&annotated, response_collector, http_queue_guard)
        && annotated.event.as_deref() == Some(ANNOTATION_LLM_METRICS)
    {
        // Legacy compatibility path: annotation frames were never emitted as
        // client-visible SSE payloads after metrics collection.
        annotated.event = None;
        annotated.comment = None;
    }

    // ANNOTATION_PAYLOAD_USAGE is payload-only: the payload DeltaAggregator already
    // consumed its usage upstream, so strip the whole chunk; it must never
    // reach the client SSE stream.
    if annotated.event.as_deref() == Some(crate::preprocessor::ANNOTATION_PAYLOAD_USAGE) {
        annotated.event = None;
        annotated.comment = None;
        annotated.data = None;
    }

    // Route reasoning at the SSE boundary. Internal representation stays
    // `reasoning_content`.
    let annotated =
        annotated.map_data(|response| Ok(RoutedReasoning::new(response, reasoning_field)));
    annotated_to_sse_event(annotated)
}

/// Create a new router with optional DRT metrics integration.
///
/// When `drt_metrics` is provided, the `/metrics` handler will also include
/// metrics from the DRT's registry tree (anything created via `metrics().create*()`).
pub fn router(
    registry: Registry,
    path: Option<String>,
    drt_metrics: Option<dynamo_runtime::metrics::MetricsRegistry>,
) -> (Vec<RouteDoc>, Router) {
    let path = path.unwrap_or_else(|| "/metrics".to_string());
    let doc = RouteDoc::new(axum::http::Method::GET, &path);

    let metrics_state = MetricsHandlerState {
        registry: Arc::new(registry),
        drt_metrics,
    };

    let route = Router::new()
        .route(&path, get(handler_metrics))
        .with_state(Arc::new(metrics_state));
    (vec![doc], route)
}

/// Unified metrics handler.
///
/// Gathers from the local HTTP-service registry first, then appends any
/// metrics from the DRT's registry tree (if configured).
async fn handler_metrics(State(state): State<Arc<MetricsHandlerState>>) -> impl IntoResponse {
    let encoder = prometheus::TextEncoder::new();
    let metric_families = state.registry.gather();
    let mut buffer = vec![];
    if encoder.encode(&metric_families, &mut buffer).is_err() {
        return (
            StatusCode::INTERNAL_SERVER_ERROR,
            "Failed to encode metrics",
        )
            .into_response();
    }

    let mut metrics = match String::from_utf8(buffer) {
        Ok(metrics) => metrics,
        Err(_) => {
            return (
                StatusCode::INTERNAL_SERVER_ERROR,
                "Failed to encode metrics",
            )
                .into_response();
        }
    };

    // Append DRT registry tree metrics (anything created via metrics().create*()).
    if let Some(ref drt_metrics) = state.drt_metrics {
        match drt_metrics.prometheus_expfmt_combined() {
            Ok(drt_text) => {
                if !drt_text.is_empty() {
                    if !metrics.is_empty() && !metrics.ends_with('\n') {
                        metrics.push('\n');
                    }
                    metrics.push_str(&drt_text);
                }
            }
            Err(e) => {
                tracing::warn!("Failed to gather DRT metrics: {}", e);
            }
        }
    }

    (StatusCode::OK, metrics).into_response()
}

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

    fn model_ready_value_with_name(
        registry: &Registry,
        metric_name: &str,
        model: &str,
    ) -> Option<f64> {
        registry
            .gather()
            .into_iter()
            .find(|family| family.name() == metric_name)?
            .get_metric()
            .iter()
            .find(|metric| {
                metric
                    .get_label()
                    .iter()
                    .any(|label| label.name() == "model" && label.value() == model)
            })
            .map(|metric| metric.get_gauge().value())
    }

    fn model_ready_value(registry: &Registry, model: &str) -> Option<f64> {
        model_ready_value_with_name(registry, &model_ready_metric_name(None), model)
    }

    #[test]
    fn model_ready_metric_tracks_live_routing_catalog() {
        let manager = Arc::new(ModelManager::new());
        let registry = Registry::new();
        register_model_ready_metric(&registry, manager.clone(), None).unwrap();

        assert_eq!(model_ready_value(&registry, "test-model"), None);

        let mut card = ModelDeploymentCard::default();
        card.worker_type = Some(crate::worker_type::WorkerType::Aggregated);
        let mut worker_set = crate::discovery::WorkerSet::new(
            "watched".to_string(),
            "watched-mdc".to_string(),
            card,
        );
        let (worker_tx, worker_rx) = tokio::sync::watch::channel(Vec::new());
        worker_set.set_instance_watcher(worker_rx);
        worker_set.chat_engine = Some(Arc::new(crate::engines::StreamingEngineAdapter::new(
            crate::engines::make_echo_engine(),
        )));
        assert!(manager.add_worker_set("test-model", "watched", worker_set));
        assert_eq!(model_ready_value(&registry, "test-model"), Some(0.0));

        let custom_registry = Registry::new();
        register_model_ready_metric(
            &custom_registry,
            manager.clone(),
            Some("custom_frontend".to_string()),
        )
        .unwrap();
        assert_eq!(
            model_ready_value_with_name(
                &custom_registry,
                "custom_frontend_model_ready",
                "test-model"
            ),
            Some(0.0)
        );
        assert_eq!(model_ready_value(&custom_registry, "test-model"), None);

        worker_tx.send(vec![1]).unwrap();
        assert_eq!(model_ready_value(&registry, "test-model"), Some(1.0));

        worker_tx.send(Vec::new()).unwrap();
        assert_eq!(model_ready_value(&registry, "test-model"), Some(0.0));
    }

    #[test]
    fn model_ready_metric_omits_alias_names() {
        let manager = Arc::new(ModelManager::new());
        let registry = Registry::new();
        register_model_ready_metric(&registry, manager.clone(), None).unwrap();

        let mut card = ModelDeploymentCard::default();
        card.worker_type = Some(crate::worker_type::WorkerType::Aggregated);
        let mut worker_set = crate::discovery::WorkerSet::new(
            "watched".to_string(),
            "watched-mdc".to_string(),
            card,
        );
        let (worker_tx, worker_rx) = tokio::sync::watch::channel(Vec::new());
        worker_set.set_instance_watcher(worker_rx);
        worker_set.chat_engine = Some(Arc::new(crate::engines::StreamingEngineAdapter::new(
            crate::engines::make_echo_engine(),
        )));
        let worker_set = Arc::new(worker_set);

        // `register_alias` refuses a name that is already a live primary, so it must
        // run before the alias name gains its own WorkerSet.
        assert!(manager.add_worker_set_arc("test-model", "watched", worker_set.clone()));
        assert!(manager.register_alias("test-model-alias", "test-model"));
        assert!(manager.add_worker_set_arc("test-model-alias", "watched", worker_set));

        worker_tx.send(vec![1]).unwrap();

        assert_eq!(model_ready_value(&registry, "test-model"), Some(1.0));
        assert_eq!(model_ready_value(&registry, "test-model-alias"), None);

        let series = registry
            .gather()
            .into_iter()
            .find(|family| family.name() == model_ready_metric_name(None))
            .map(|family| family.get_metric().len());
        assert_eq!(series, Some(1));
    }

    #[test]
    fn test_round_to_sig_figs() {
        // Test rounding to 2 significant figures
        assert_eq!(round_to_sig_figs(0.0026, 2), 0.0026);
        assert_eq!(round_to_sig_figs(0.26, 2), 0.26);
        assert_eq!(round_to_sig_figs(0.2356, 2), 0.24);
        assert_eq!(round_to_sig_figs(1.234, 2), 1.2);
        assert_eq!(round_to_sig_figs(12.34, 2), 12.0);
        assert_eq!(round_to_sig_figs(123.4, 2), 120.0);
        assert_eq!(round_to_sig_figs(1234.0, 2), 1200.0);
        assert_eq!(round_to_sig_figs(0.0, 2), 0.0);

        // Test edge cases
        assert_eq!(round_to_sig_figs(0.999, 2), 1.0);
        assert_eq!(round_to_sig_figs(9.99, 2), 10.0);
        assert_eq!(round_to_sig_figs(99.9, 2), 100.0);
    }

    #[test]
    fn test_generate_log_buckets_basic() {
        // Test basic properties
        let buckets = generate_log_buckets(1.0, 100.0, 5);

        // Check length
        assert_eq!(buckets.len(), 5);

        // Check first value is 0
        assert_eq!(buckets[0], 0.0);

        // Check last value is approximately max (rounded to 2 sig figs)
        assert_eq!(buckets[buckets.len() - 1], 100.0);

        // Check values are increasing
        for i in 1..buckets.len() {
            assert!(
                buckets[i] > buckets[i - 1],
                "Bucket values should be increasing: {} <= {}",
                buckets[i - 1],
                buckets[i]
            );
        }
    }

    /// A stand-in for the process environment. Tests inject this instead of calling
    /// `temp_env`: mutating the real environment races the many other tests in this
    /// module that construct `Metrics` and read these same variables, which on Unix is
    /// an unsafe `setenv`/`getenv` race, not merely an ordering hazard.
    fn fake_env<'a>(pairs: &'a [(&'a str, &'a str)]) -> impl Fn(&str) -> Option<String> + 'a {
        move |name| {
            pairs
                .iter()
                .find(|(key, _)| *key == name)
                .map(|(_, value)| (*value).to_string())
        }
    }

    #[test]
    fn bucket_config_reads_the_documented_env_var_names() {
        // These names were once read under an extra DYN_HISTOGRAM_ prefix, which
        // silently stopped them working; they must resolve as documented.
        let env = fake_env(&[
            ("DYN_METRICS_ITL_MIN", "0.002"),
            ("DYN_METRICS_ITL_MAX", "80"),
            ("DYN_METRICS_ITL_COUNT", "20"),
        ]);
        let cfg = parse_bucket_config(&env, env_metrics::DYN_METRICS_ITL, 0.001, 2.0, 13);
        assert_eq!(cfg, (0.002, 80.0, 20));
    }

    #[test]
    fn bucket_config_falls_back_to_the_deprecated_doubled_name() {
        // The doubled form was the only name that worked for several releases, so it
        // stays supported for one more rather than silently reverting to defaults.
        let env = fake_env(&[("DYN_HISTOGRAM_DYN_METRICS_ITL_MAX", "80")]);
        let (_, max, _) = parse_bucket_config(&env, env_metrics::DYN_METRICS_ITL, 0.001, 2.0, 13);
        assert_eq!(max, 80.0);
    }

    #[test]
    fn bucket_config_prefers_the_new_name_over_the_deprecated_one() {
        let env = fake_env(&[
            ("DYN_METRICS_ITL_MAX", "80"),
            ("DYN_HISTOGRAM_DYN_METRICS_ITL_MAX", "30"),
        ]);
        let (_, max, _) = parse_bucket_config(&env, env_metrics::DYN_METRICS_ITL, 0.001, 2.0, 13);
        assert_eq!(max, 80.0);
    }

    #[test]
    fn bucket_config_falls_back_to_defaults_for_unset_and_unparseable_values() {
        let env = fake_env(&[
            ("DYN_METRICS_ITL_MIN", "not-a-number"),
            ("DYN_METRICS_ITL_COUNT", "12.5"),
        ]);
        let cfg = parse_bucket_config(&env, env_metrics::DYN_METRICS_ITL, 0.001, 2.0, 13);
        assert_eq!(cfg, (0.001, 2.0, 13));
    }

    #[test]
    fn bucket_config_rejects_a_parseable_but_invalid_combination() {
        // min >= max parses fine but cannot produce buckets, so the whole set reverts —
        // including a COUNT that was valid on its own.
        let env = fake_env(&[
            ("DYN_METRICS_ITL_MIN", "5"),
            ("DYN_METRICS_ITL_MAX", "1"),
            ("DYN_METRICS_ITL_COUNT", "20"),
        ]);
        let cfg = parse_bucket_config(&env, env_metrics::DYN_METRICS_ITL, 0.001, 2.0, 13);
        assert_eq!(cfg, (0.001, 2.0, 13));
    }

    /// Upper bounds of a histogram's buckets, i.e. the `le` label values that a
    /// dashboard's `histogram_quantile` query sees on `/metrics`.
    fn bucket_upper_bounds(registry: &Registry, metric_name: &str) -> Vec<f64> {
        registry
            .gather()
            .into_iter()
            .find(|family| family.name() == metric_name)
            .expect("histogram not registered")
            .get_metric()[0]
            .get_histogram()
            .get_bucket()
            .iter()
            .map(|b| b.upper_bound())
            .collect()
    }

    #[test]
    fn itl_ceiling_env_var_reaches_the_exported_le_labels() {
        // Deliberately not the default (80.0, 20): a test that sets the env to the
        // default value would pass even if the variable were ignored entirely.
        let env = fake_env(&[
            ("DYN_METRICS_ITL_MAX", "30"),
            ("DYN_METRICS_ITL_COUNT", "8"),
        ]);
        let registry = Registry::new();
        let metrics = Metrics::build(None, &env);
        metrics.register(&registry).unwrap();
        metrics
            .inter_token_latency
            .with_label_values(&["m"])
            .observe(0.5);

        let bounds = bucket_upper_bounds(
            &registry,
            &format!(
                "{}_{}",
                name_prefix::FRONTEND,
                frontend_service::INTER_TOKEN_LATENCY_SECONDS
            ),
        );
        assert_eq!(
            bounds.last().copied(),
            Some(30.0),
            "top finite bucket should follow DYN_METRICS_ITL_MAX, got {bounds:?}"
        );
        assert_eq!(
            bounds.len(),
            8,
            "bucket count should follow DYN_METRICS_ITL_COUNT"
        );
    }

    #[test]
    fn itl_default_buckets_reach_80_seconds() {
        // The ceiling was 2.0s, so any inter-token latency above it landed in `+Inf` and
        // `histogram_quantile` pinned p99 at exactly 2.0 -- indistinguishable from a real
        // 2s measurement. Guard the shipped default, not just the env-var override.
        let registry = Registry::new();
        let metrics = Metrics::build(None, &fake_env(&[]));
        metrics.register(&registry).unwrap();
        metrics
            .inter_token_latency
            .with_label_values(&["m"])
            .observe(0.5);

        let bounds = bucket_upper_bounds(
            &registry,
            &format!(
                "{}_{}",
                name_prefix::FRONTEND,
                frontend_service::INTER_TOKEN_LATENCY_SECONDS
            ),
        );
        // The exact set is the contract. Note the bottom edge of 0.0018: vLLM's equivalent
        // starts at 0.01 and cannot resolve anything faster than 10ms per token, so raising
        // the ceiling must not cost the low-end resolution Dynamo has and vLLM does not.
        assert_eq!(
            bounds,
            vec![
                0.0, 0.0018, 0.0033, 0.0059, 0.011, 0.02, 0.035, 0.064, 0.12, 0.21, 0.38, 0.69,
                1.2, 2.3, 4.1, 7.4, 13.0, 24.0, 44.0, 80.0
            ],
        );
    }

    #[test]
    fn test_generate_log_buckets_edge_cases() {
        // Test empty buckets
        let buckets = generate_log_buckets(1.0, 100.0, 0);
        assert_eq!(buckets.len(), 0);

        // Test single bucket
        let buckets = generate_log_buckets(1.0, 100.0, 1);
        assert_eq!(buckets.len(), 1);
        assert_eq!(buckets[0], 0.0);

        // Test two buckets
        let buckets = generate_log_buckets(1.0, 100.0, 2);
        assert_eq!(buckets.len(), 2);
        assert_eq!(buckets[0], 0.0);
        assert_eq!(buckets[1], 100.0);
    }

    #[test]
    fn test_generate_log_buckets_always_includes_zero() {
        // Test various configurations
        for count in 1..=20 {
            let buckets = generate_log_buckets(0.1, 1000.0, count);
            assert_eq!(
                buckets[0], 0.0,
                "First bucket should always be 0.0 for count={}",
                count
            );
        }
    }

    #[test]
    fn test_all_buckets_are_two_sig_figs() {
        let test_cases = vec![
            (1.0, 512.0, 10),
            (50.0, 128000.0, 12),
            (50.0, 32000.0, 10),
            (0.001, 480.0, 18),
            (0.001, 80.0, 20),
            (0.001, 10.0, 14),
        ];

        for (min, max, count) in test_cases {
            let buckets = generate_log_buckets(min, max, count);
            for &value in buckets.iter().skip(1) {
                let rounded = round_to_sig_figs(value, 2);
                assert_eq!(
                    value, rounded,
                    "Value {} should be rounded to 2 sig figs (min={}, max={}, count={})",
                    value, min, max, count
                );
            }
        }
    }

    #[test]
    fn test_sig_fig_limitation_with_many_buckets() {
        // This test demonstrates that 2 sig figs limits the number of unique bucket values
        // With 1000 requested buckets but only 2 sig figs, we'll get automatic deduplication
        let buckets = generate_log_buckets(0.0001, 1.0, 1000);

        println!(
            "Requested 1000 buckets, got {} total values (including 0.0)",
            buckets.len()
        );

        // With 2 sig figs across 4 orders of magnitude (0.0001 to 1.0),
        // we can have roughly 90 unique values per order of magnitude
        // So we expect around 360 unique values maximum
        assert!(
            buckets.len() < 500,
            "Expected fewer than 500 unique buckets due to 2 sig fig limitation, got {}",
            buckets.len()
        );

        // Verify all values are unique (no duplicates remain after deduplication)
        let mut sorted_buckets = buckets.clone();
        sorted_buckets.sort_by(|a, b| a.partial_cmp(b).unwrap());
        sorted_buckets.dedup();
        assert_eq!(
            buckets.len(),
            sorted_buckets.len(),
            "All buckets should be unique after deduplication"
        );

        // Verify first is still 0.0
        assert_eq!(buckets[0], 0.0);

        // Verify values are still in increasing order
        for i in 1..buckets.len() {
            assert!(
                buckets[i] > buckets[i - 1],
                "Buckets should be in increasing order"
            );
        }
    }

    #[test]
    fn test_deduplication_preserves_order() {
        // Test that deduplication maintains increasing order
        let buckets = generate_log_buckets(0.01, 1.0, 50);

        // Verify all values are unique
        let mut unique_check = std::collections::HashSet::new();
        for &bucket in &buckets {
            assert!(
                unique_check.insert(bucket.to_bits()),
                "Duplicate value {} found after deduplication",
                bucket
            );
        }

        // Verify order is maintained
        for i in 1..buckets.len() {
            assert!(
                buckets[i] > buckets[i - 1],
                "Bucket values should be in increasing order after deduplication"
            );
        }
    }

    /// A chunk whose delta renders as nothing is still an accounting event: the
    /// engine generated those tokens. The chat SSE loop drops such chunks before
    /// forwarding them (multi-byte token assembly, and the Nemotron
    /// force_nonempty_content deferral, both produce them), so it observes their
    /// metrics explicitly before discarding. This pins the helper it calls: an
    /// empty delta carrying `llm_metrics` must still count.
    #[test]
    fn test_empty_delta_with_metrics_is_still_counted() {
        use crate::protocols::common::metrics::LLMMetricAnnotation;
        use dynamo_protocols::types::{
            ChatChoiceStream, ChatCompletionStreamResponseDelta, CreateChatCompletionStreamResponse,
        };

        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();
        let model = "test-model";
        let mut collector = metrics.clone().create_response_collector(model);
        let mut guard = None;

        #[allow(deprecated)]
        let choice = ChatChoiceStream {
            index: 0,
            delta: ChatCompletionStreamResponseDelta {
                role: None,
                content: None,
                tool_calls: None,
                function_call: None,
                refusal: None,
                reasoning_content: None,
            },
            finish_reason: None,
            logprobs: None,
        };
        let data = NvCreateChatCompletionStreamResponse {
            inner: CreateChatCompletionStreamResponse {
                id: "test".to_string(),
                choices: vec![choice],
                created: 0,
                model: model.to_string(),
                system_fingerprint: None,
                object: "chat.completion.chunk".to_string(),
                usage: None,
                service_tier: None,
            },
            nvext: None,
            prompt_logprobs: None,
            llm_metrics: Some(LLMMetricAnnotation {
                input_tokens: 10,
                output_tokens: 4,
                chunk_tokens: 4,
                ..Default::default()
            }),
            tool_call_completion: Vec::new(),
        };
        let annotated = crate::types::Annotated::from_data(data);

        process_chat_response_and_observe_metrics(&annotated, &mut collector, &mut guard);

        assert_eq!(
            metrics
                .output_tokens_counter
                .with_label_values(&[model])
                .get(),
            4,
            "tokens on a non-renderable chunk must still be counted"
        );
    }

    #[test]
    fn test_output_tokens_counter_increments() {
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model = "test-model";

        // Create response collector
        let mut collector = metrics.clone().create_response_collector(model);

        // Simulate first chunk (5 tokens)
        collector.observe_response(100, 5);

        // Verify counter incremented by 5
        let counter_value = metrics
            .output_tokens_counter
            .with_label_values(&[model])
            .get();
        assert_eq!(counter_value, 5);

        // Simulate second chunk (10 tokens)
        collector.observe_response(100, 10);

        // Verify counter incremented to 15
        let counter_value = metrics
            .output_tokens_counter
            .with_label_values(&[model])
            .get();
        assert_eq!(counter_value, 15);

        // Simulate third chunk (7 tokens)
        collector.observe_response(100, 7);

        // Verify counter incremented to 22
        let counter_value = metrics
            .output_tokens_counter
            .with_label_values(&[model])
            .get();
        assert_eq!(counter_value, 22);
    }

    #[test]
    fn test_local_itl_histogram_is_initialized_lazily() {
        let metrics = Arc::new(Metrics::new());
        let model = "lazy-local-itl-model";
        let global = metrics.inter_token_latency.with_label_values(&[model]);
        let mut collector = metrics.create_response_collector(model);

        assert!(collector.inter_token_latency.is_none());

        collector.observe_response(10, 0);
        assert!(
            collector.inter_token_latency.is_none(),
            "zero-token responses must not allocate a local histogram"
        );

        collector.observe_response(10, 1);
        assert!(
            collector.inter_token_latency.is_none(),
            "the TTFT-only response must not allocate a local histogram"
        );

        collector.observe_response(10, 1);
        assert!(
            collector.inter_token_latency.is_some(),
            "the first ITL observation must allocate a local histogram"
        );
        assert_eq!(
            global.get_sample_count(),
            0,
            "the first ITL observation must remain local until flush or drop"
        );

        drop(collector);
        assert_eq!(global.get_sample_count(), 1);
    }

    #[test]
    fn test_cached_handles_record_through_vec() {
        // The collector caches per-model handles; observing through them must update
        // the same metric the vec exposes. Regression guard for the handle cache
        // (including the lazily resolved handle used by the per-token ITL loop).
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();
        let model = "cached-handle-model";

        let mut collector = metrics.clone().create_response_collector(model);
        // First chunk (3 tokens): TTFT + ISL observed once; no ITL yet.
        collector.observe_response(42, 3);
        // Second chunk (4 tokens): ITL observed once per token via the cached handle.
        collector.observe_response(42, 4);

        // ITL observations are request-local until the batch threshold or drop.
        assert_eq!(
            metrics
                .inter_token_latency
                .with_label_values(&[model])
                .get_sample_count(),
            0
        );
        drop(collector);

        let ttft = metrics.time_to_first_token.with_label_values(&[model]);
        assert_eq!(ttft.get_sample_count(), 1, "TTFT observed once");

        let isl = metrics.input_sequence_length.with_label_values(&[model]);
        assert_eq!(isl.get_sample_count(), 1, "ISL observed once");
        assert_eq!(isl.get_sample_sum(), 42.0);

        let itl = metrics.inter_token_latency.with_label_values(&[model]);
        assert_eq!(
            itl.get_sample_count(),
            4,
            "ITL observed once per token of 2nd chunk"
        );

        let out = metrics
            .output_tokens_counter
            .with_label_values(&[model])
            .get();
        assert_eq!(out, 7, "output tokens = 3 + 4 via cached counter handle");
    }

    #[test]
    fn test_local_itl_histogram_flushes_at_threshold_and_drop() {
        use prometheus::core::Collector;

        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();
        let model = "local-itl-flush-model";
        let global = metrics.inter_token_latency.with_label_values(&[model]);

        let mut collector = metrics.clone().create_response_collector(model);
        collector.observe_response(10, 1); // TTFT only.
        assert!(collector.inter_token_latency.is_none());
        collector.observe_response(10, 63);
        assert!(collector.inter_token_latency.is_some());
        assert_eq!(
            global.get_sample_count(),
            0,
            "a partial local batch must remain unpublished"
        );

        collector.observe_response(10, 1);
        assert_eq!(
            global.get_sample_count(),
            ITL_LOCAL_FLUSH_TOKENS,
            "the 64th pending ITL sample must flush the local histogram"
        );

        collector.observe_response(10, 7);
        assert_eq!(
            global.get_sample_count(),
            ITL_LOCAL_FLUSH_TOKENS,
            "the next partial batch must remain local"
        );
        drop(collector);

        assert_eq!(global.get_sample_count(), 71);
        assert!(global.get_sample_sum() > 0.0);

        let families = global.collect();
        let histogram = families[0].get_metric()[0].get_histogram();
        let last_bucket = histogram.get_bucket().last().unwrap();
        assert_eq!(
            last_bucket.cumulative_count(),
            global.get_sample_count(),
            "flushing must update histogram buckets as well as count and sum"
        );
    }

    #[test]
    fn test_worker_metadata_is_latched_without_replacement() {
        let metrics = Arc::new(Metrics::new());
        let mut collector = metrics.create_response_collector("worker-latch-model");
        let first = LLMMetricAnnotation {
            prefill_worker_id: Some(11),
            prefill_dp_rank: Some(1),
            prefill_worker_type: Some("prefill-first".to_string()),
            decode_worker_id: Some(22),
            decode_dp_rank: Some(2),
            decode_worker_type: Some("decode-first".to_string()),
            ..Default::default()
        };
        collector.set_worker_info_from_metrics(&first);

        let later = LLMMetricAnnotation {
            prefill_worker_id: Some(33),
            prefill_dp_rank: Some(3),
            prefill_worker_type: Some("prefill-later".to_string()),
            decode_worker_id: Some(44),
            decode_dp_rank: Some(4),
            decode_worker_type: Some("decode-later".to_string()),
            ..Default::default()
        };
        collector.set_worker_info_from_metrics(&later);

        assert_eq!(collector.prefill_worker_id, Some(11));
        assert_eq!(collector.prefill_dp_rank, Some(1));
        assert_eq!(
            collector.prefill_worker_type.as_deref(),
            Some("prefill-first")
        );
        assert_eq!(collector.decode_worker_id, Some(22));
        assert_eq!(collector.decode_dp_rank, Some(2));
        assert_eq!(
            collector.decode_worker_type.as_deref(),
            Some("decode-first")
        );
    }

    #[test]
    fn test_multimodal_counts_observed_once() {
        // Counts are request-level constants carried on every chunk's annotation.
        // The collector must latch them on the first observation: each histogram
        // records exactly one per-request sample, and later (differing) values are
        // ignored. The histogram `_sum` doubles as the cumulative volume.
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();
        let model = "mm-counts-model";

        let mut collector = metrics.clone().create_response_collector(model);
        // Repeated calls simulate the same counts arriving on each streamed chunk.
        collector.observe_multimodal_metrics(2, 1, 0, Some(1290));
        collector.observe_multimodal_metrics(2, 1, 0, Some(1290));
        // A later, differing value must not override the latched counts.
        collector.observe_multimodal_metrics(99, 99, 99, Some(9999));

        let img = metrics.images_per_request.with_label_values(&[model]);
        assert_eq!(img.get_sample_count(), 1, "image histogram observed once");
        assert_eq!(img.get_sample_sum(), 2.0, "image _sum == cumulative volume");
        let vid = metrics.videos_per_request.with_label_values(&[model]);
        assert_eq!(vid.get_sample_count(), 1, "video histogram observed once");
        assert_eq!(vid.get_sample_sum(), 1.0);
        let aud = metrics.audio_per_request.with_label_values(&[model]);
        assert_eq!(
            aud.get_sample_count(),
            1,
            "audio histogram observes even a 0 (text-only distribution)"
        );
        assert_eq!(aud.get_sample_sum(), 0.0);
        let image_tokens = metrics.image_tokens_per_request.with_label_values(&[model]);
        assert_eq!(
            image_tokens.get_sample_count(),
            1,
            "image-token histogram observed once"
        );
        assert_eq!(image_tokens.get_sample_sum(), 1290.0);
    }

    #[test]
    fn test_multimodal_counts_text_only_zeros() {
        // A text-only request carries zero counts; the histograms still get one
        // sample each (at 0), so `_count` is 1 and `_sum` is 0.
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();
        let model = "text-only-model";

        let mut collector = metrics.clone().create_response_collector(model);
        collector.observe_multimodal_counts(0, 0, 0);

        let img = metrics.images_per_request.with_label_values(&[model]);
        assert_eq!(img.get_sample_count(), 1);
        assert_eq!(img.get_sample_sum(), 0.0);
        let image_tokens = metrics.image_tokens_per_request.with_label_values(&[model]);
        assert_eq!(
            image_tokens.get_sample_count(),
            0,
            "unavailable image-token count must not emit a sample"
        );
    }

    #[test]
    fn test_cached_decode_itl_gauge_records_for_worker() {
        // The per-worker ITL gauge handle is resolved once (worker labels are latched
        // at routing time via `set_worker_info`) and reused on every output token.
        // Verify observations through the cached handle land on the same GaugeVec
        // series the labels address. Regression guard for the per-token handle cache.
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).ok();
        let model = "itl-gauge-model";

        // Distinctive worker id so this never collides with other tests on the
        // process-global WORKER_LAST_INTER_TOKEN_LATENCY_GAUGE.
        let worker_id: u64 = 9_876_543_210;
        let mut collector = metrics.clone().create_response_collector(model);
        collector.set_worker_info(
            None,
            None,
            None,
            Some(worker_id),
            Some(0),
            Some("decode".to_string()),
        );

        // First chunk: TTFT only, no ITL yet (no prior response time) -> gauge unset.
        collector.observe_response(10, 1);
        // Elapse measurable time so ITL > 0, then a second chunk drives the cached
        // per-worker ITL gauge.
        std::thread::sleep(std::time::Duration::from_millis(2));
        collector.observe_response(10, 1);

        let worker_id_str = worker_id.to_string();
        let gauge = WORKER_LAST_INTER_TOKEN_LATENCY_GAUGE.with_label_values(&[
            worker_id_str.as_str(),
            "0",
            "decode",
        ]);
        assert!(
            gauge.get() > 0.0,
            "cached per-worker ITL gauge must record a positive ITL through the vec, got {}",
            gauge.get()
        );
    }

    #[test]
    fn test_output_tokens_counter_zero_tokens() {
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model = "test-model";
        let mut collector = metrics.clone().create_response_collector(model);

        // Simulate chunk with zero tokens (should not increment)
        collector.observe_response(100, 0);

        // Verify counter remains 0
        let counter_value = metrics
            .output_tokens_counter
            .with_label_values(&[model])
            .get();
        assert_eq!(counter_value, 0);

        // Add some tokens
        collector.observe_response(100, 5);
        assert_eq!(
            metrics
                .output_tokens_counter
                .with_label_values(&[model])
                .get(),
            5
        );

        // Try zero tokens again (should not change counter)
        collector.observe_response(100, 0);
        assert_eq!(
            metrics
                .output_tokens_counter
                .with_label_values(&[model])
                .get(),
            5
        );
    }

    #[test]
    fn test_output_tokens_counter_multiple_models() {
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model1 = "model-1";
        let model2 = "model-2";

        // Create collectors for different models
        let mut collector1 = metrics.clone().create_response_collector(model1);
        let mut collector2 = metrics.clone().create_response_collector(model2);

        // Increment model1
        collector1.observe_response(100, 10);
        assert_eq!(
            metrics
                .output_tokens_counter
                .with_label_values(&[model1])
                .get(),
            10
        );
        assert_eq!(
            metrics
                .output_tokens_counter
                .with_label_values(&[model2])
                .get(),
            0
        );

        // Increment model2
        collector2.observe_response(200, 20);
        assert_eq!(
            metrics
                .output_tokens_counter
                .with_label_values(&[model1])
                .get(),
            10
        );
        assert_eq!(
            metrics
                .output_tokens_counter
                .with_label_values(&[model2])
                .get(),
            20
        );

        // Increment model1 again
        collector1.observe_response(100, 5);
        assert_eq!(
            metrics
                .output_tokens_counter
                .with_label_values(&[model1])
                .get(),
            15
        );
        assert_eq!(
            metrics
                .output_tokens_counter
                .with_label_values(&[model2])
                .get(),
            20
        );
    }

    #[test]
    fn test_cached_tokens_once_per_request() {
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model = "test-model";
        let expected_metric_name = "dynamo_frontend_cached_tokens";
        let mut collector = metrics.clone().create_response_collector(model);

        // Create histogram handle first
        let _histogram = metrics.cached_tokens.with_label_values(&[model]);

        // First call should observe and record 1 sample
        collector.observe_cached_tokens(Some(100));
        let metric_families = registry.gather();
        let histogram_family = metric_families
            .iter()
            .find(|mf| mf.name() == expected_metric_name)
            .expect("histogram should be registered");
        assert_eq!(
            histogram_family.get_metric()[0]
                .get_histogram()
                .get_sample_count(),
            1
        );

        // Second call with same collector should not observe again (idempotent)
        collector.observe_cached_tokens(Some(50));
        let metric_families = registry.gather();
        let histogram_family = metric_families
            .iter()
            .find(|mf| mf.name() == expected_metric_name)
            .expect("histogram should be registered");
        assert_eq!(
            histogram_family.get_metric()[0]
                .get_histogram()
                .get_sample_count(),
            1
        );

        // Third call with different value should still be idempotent
        collector.observe_cached_tokens(Some(75));
        let metric_families = registry.gather();
        let histogram_family = metric_families
            .iter()
            .find(|mf| mf.name() == expected_metric_name)
            .expect("histogram should be registered");
        assert_eq!(
            histogram_family.get_metric()[0]
                .get_histogram()
                .get_sample_count(),
            1
        );
    }

    #[test]
    fn test_metrics_annotation_event_handling() {
        use crate::preprocessor::LLMMetricAnnotation;
        use crate::types::Annotated;

        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model = "test-model";
        let expected_metric_name = "dynamo_frontend_cached_tokens";
        let expected_tokenizer_metric_name = "dynamo_frontend_tokenizer_latency_ms";
        let mut collector = metrics.clone().create_response_collector(model);

        // Create a metrics annotation event (event without SSE data payload)
        let mut annotated = Annotated::<
            crate::protocols::openai::chat_completions::NvCreateChatCompletionStreamResponse,
        > {
            id: None,
            data: None,
            event: Some(crate::preprocessor::ANNOTATION_LLM_METRICS.to_string()),
            comment: None,
            error: None,
        };

        // Add metrics annotation with cached_tokens
        let llm_metrics = LLMMetricAnnotation {
            input_tokens: 10,
            output_tokens: 20,
            chunk_tokens: 5,
            cached_tokens: Some(15),
            prefill_worker_id: None,
            prefill_dp_rank: None,
            prefill_worker_type: None,
            decode_worker_id: None,
            decode_dp_rank: None,
            decode_worker_type: None,
            tokenize_latency: Some(Duration::from_millis(8)),
            detokenize_total_latency: Some(Duration::from_micros(100)),
            detokenize_count: Some(2),
            ..Default::default()
        };

        let annotation = llm_metrics.to_annotation::<()>().unwrap();
        annotated.event = annotation.event;
        annotated.comment = annotation.comment;

        // Process the event
        let mut http_queue_guard = None;
        let result = process_response_using_event_converter_and_observe_metrics(
            EventConverter::from(annotated),
            &mut collector,
            &mut http_queue_guard,
        );

        // Should return Ok(None) for metrics annotation events
        assert!(matches!(result, Ok(None)));

        // Drop collector so the detokenize observation fires in Drop
        drop(collector);

        // Should have observed the cached tokens from the metrics annotation event
        let metric_families = registry.gather();
        let histogram_family = metric_families
            .iter()
            .find(|mf| mf.name() == expected_metric_name)
            .expect("histogram should be registered");
        assert_eq!(
            histogram_family.get_metric()[0]
                .get_histogram()
                .get_sample_count(),
            1
        );

        let histogram_family = metric_families
            .iter()
            .find(|mf| mf.name() == expected_tokenizer_metric_name)
            .expect("histogram should be registered");

        // Find the tokenize and detokenize observations by label
        let tokenize_metric = histogram_family
            .get_metric()
            .iter()
            .find(|m| m.get_label().iter().any(|l| l.value() == "tokenize"))
            .expect("tokenize metric should exist");
        assert_eq!(tokenize_metric.get_histogram().get_sample_count(), 1);
        // 8ms
        assert!(
            (tokenize_metric.get_histogram().get_sample_sum() - 8.0).abs() < 0.001,
            "tokenize latency should be 8.0ms"
        );

        let detokenize_metric = histogram_family
            .get_metric()
            .iter()
            .find(|m| m.get_label().iter().any(|l| l.value() == "detokenize"))
            .expect("detokenize metric should exist");
        assert_eq!(detokenize_metric.get_histogram().get_sample_count(), 1);
        // Average: 100us total / 2 samples = 50us = 0.05ms
        assert!(
            (detokenize_metric.get_histogram().get_sample_sum() - 0.05).abs() < 0.001,
            "detokenize average latency should be 0.05ms, got {}",
            detokenize_metric.get_histogram().get_sample_sum()
        );
    }

    #[tokio::test]
    async fn metrics_annotations_stripped_from_client_sse() {
        // PR #9390: two metric annotations must never leak to the client SSE.
        // (1) Per-chunk `llm_metrics` rides on a content chunk: the content delta
        //     must reach the client, but the metrics event/comment must be stripped.
        // (2) The payload-only `payload_usage` chunk must be dropped entirely (the payload
        //     DeltaAggregator already consumed its usage upstream).
        use crate::preprocessor::{ANNOTATION_PAYLOAD_USAGE, LLMMetricAnnotation};
        use crate::types::Annotated;
        use axum::response::IntoResponse;
        use axum::response::sse::Sse;

        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();
        let mut collector = metrics.clone().create_response_collector("test-model");

        let llm_metrics = LLMMetricAnnotation {
            input_tokens: 7,
            output_tokens: 3,
            chunk_tokens: 1,
            cached_tokens: Some(2),
            prefill_worker_id: None,
            prefill_dp_rank: None,
            prefill_worker_type: None,
            decode_worker_id: None,
            decode_dp_rank: None,
            decode_worker_type: None,
            tokenize_latency: None,
            detokenize_total_latency: None,
            detokenize_count: None,
            ..Default::default()
        };
        let metrics_comment = llm_metrics.to_annotation::<()>().unwrap().comment;
        let metrics_json = metrics_comment.as_ref().expect("metrics comment present")[0].clone();

        // (1) Per-chunk metrics on a content chunk (event = llm_metrics).
        let content: crate::protocols::openai::chat_completions::NvCreateChatCompletionStreamResponse =
            serde_json::from_value(serde_json::json!({
                "id": "chatcmpl-x", "object": "chat.completion.chunk", "created": 1,
                "model": "test-model",
                "choices": [{"index": 0, "delta": {"content": "hello"}}]
            }))
            .unwrap();
        let per_chunk = Annotated {
            id: None,
            data: Some(content),
            event: Some(crate::preprocessor::ANNOTATION_LLM_METRICS.to_string()),
            comment: metrics_comment.clone(),
            error: None,
        };

        let mut http_queue_guard = None;
        let event = process_response_using_event_converter_and_observe_metrics(
            EventConverter::from(per_chunk),
            &mut collector,
            &mut http_queue_guard,
        )
        .expect("conversion ok")
        .expect("content chunk should yield a client event");

        let sse = Sse::new(futures::stream::once(async move {
            Ok::<_, std::convert::Infallible>(event)
        }));
        let body = sse.into_response().into_body();
        let bytes = axum::body::to_bytes(body, usize::MAX).await.unwrap();
        let wire = String::from_utf8_lossy(&bytes);

        assert!(
            wire.contains("hello"),
            "content delta should reach the client: {wire}"
        );
        assert!(
            !wire.contains(&metrics_json)
                && !wire.contains("chunk_tokens")
                && !wire.contains("input_tokens"),
            "internal metrics leaked to client SSE: {wire}"
        );

        // A choice-less Dynamo metadata frame is client-visible.
        let metadata: crate::protocols::openai::chat_completions::NvCreateChatCompletionStreamResponse =
            serde_json::from_value(serde_json::json!({
                "id": "chatcmpl-x", "object": "chat.completion.chunk", "created": 1,
                "model": "test-model", "choices": [],
                "nvext": {"engine_data": {"prompt_token_ids": [1, 2]}}
            }))
            .unwrap();
        let metadata = Annotated {
            id: None,
            data: Some(metadata),
            event: None,
            comment: None,
            error: None,
        };
        let mut http_queue_guard = None;
        let event = process_chat_response_using_event_converter_and_observe_metrics(
            EventConverter::from(metadata),
            &mut collector,
            &mut http_queue_guard,
            ReasoningField::default(),
        )
        .expect("conversion ok")
        .expect("nvext chunk should yield a client event");
        let sse = Sse::new(futures::stream::once(async move {
            Ok::<_, std::convert::Infallible>(event)
        }));
        let body = sse.into_response().into_body();
        let bytes = axum::body::to_bytes(body, usize::MAX).await.unwrap();
        let wire = String::from_utf8_lossy(&bytes);
        assert!(
            wire.contains("engine_data") && wire.contains("prompt_token_ids"),
            "nvext metadata did not reach client SSE: {wire}"
        );

        // (2) Payload-only usage chunk (event = payload_usage, carries usage data).
        let usage: crate::protocols::openai::chat_completions::NvCreateChatCompletionStreamResponse =
            serde_json::from_value(serde_json::json!({
                "id": "chatcmpl-x", "object": "chat.completion.chunk", "created": 1,
                "model": "test-model", "choices": [],
                "usage": {"prompt_tokens": 7, "completion_tokens": 3, "total_tokens": 10}
            }))
            .unwrap();
        let payload_usage = Annotated {
            id: None,
            data: Some(usage),
            event: Some(ANNOTATION_PAYLOAD_USAGE.to_string()),
            comment: metrics_comment,
            error: None,
        };

        let mut http_queue_guard = None;
        let result = process_chat_response_using_event_converter_and_observe_metrics(
            EventConverter::from(payload_usage),
            &mut collector,
            &mut http_queue_guard,
            ReasoningField::default(),
        )
        .expect("conversion ok");
        assert!(
            result.is_none(),
            "payload_usage chunk must not be forwarded to the client SSE stream"
        );
    }

    #[test]
    fn test_chat_typed_metrics_fast_path_observes_without_annotation() {
        use crate::protocols::common::metrics::LLMMetricAnnotation;

        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model = "test-model";
        let expected_metric_name = "dynamo_frontend_cached_tokens";
        let expected_tokenizer_metric_name = "dynamo_frontend_tokenizer_latency_ms";
        let mut collector = metrics.clone().create_response_collector(model);
        let mut annotated = make_chat_stream_annotated("hello");
        annotated.data.as_mut().unwrap().llm_metrics = Some(LLMMetricAnnotation {
            input_tokens: 10,
            output_tokens: 20,
            chunk_tokens: 5,
            cached_tokens: Some(15),
            prefill_worker_id: Some(11),
            prefill_dp_rank: Some(1),
            prefill_worker_type: Some(WORKER_TYPE_PREFILL.to_string()),
            decode_worker_id: Some(22),
            decode_dp_rank: Some(2),
            decode_worker_type: Some(WORKER_TYPE_DECODE.to_string()),
            tokenize_latency: Some(Duration::from_millis(8)),
            detokenize_total_latency: Some(Duration::from_micros(100)),
            detokenize_count: Some(2),
            ..Default::default()
        });

        assert!(annotated.event.is_none());
        assert!(annotated.comment.is_none());

        annotated.data.as_mut().unwrap().tool_call_completion = vec![
            crate::protocols::openai::chat_completions::ToolCallCompletion {
                choice_index: 0,
                tool_index: 0,
                complete: true,
            },
        ];
        let transported: crate::types::Annotated<NvCreateChatCompletionStreamResponse> =
            serde_json::from_value(serde_json::to_value(&annotated).unwrap()).unwrap();
        assert_eq!(
            transported.data.as_ref().unwrap().tool_call_completion,
            annotated.data.as_ref().unwrap().tool_call_completion
        );

        let mut http_queue_guard = Some(metrics.clone().create_http_queue_guard(model));
        let result = process_chat_response_using_event_converter_and_observe_metrics(
            EventConverter::from(annotated),
            &mut collector,
            &mut http_queue_guard,
            ReasoningField::default(),
        );

        assert!(
            http_queue_guard.is_none(),
            "first positive chunk should release HTTP queue guard"
        );
        let event = result
            .unwrap()
            .expect("typed metrics data chunk should still produce an SSE event");
        let json = extract_sse_data_json(event);
        assert!(
            json.get("llm_metrics").is_none(),
            "typed metrics must be skipped on the SSE wire"
        );
        assert!(json.get("tool_call_completion").is_none());

        drop(collector);

        let metric_families = registry.gather();
        let histogram_family = metric_families
            .iter()
            .find(|mf| mf.name() == expected_metric_name)
            .expect("histogram should be registered");
        assert_eq!(
            histogram_family.get_metric()[0]
                .get_histogram()
                .get_sample_count(),
            1
        );

        let histogram_family = metric_families
            .iter()
            .find(|mf| mf.name() == expected_tokenizer_metric_name)
            .expect("histogram should be registered");
        let tokenize_metric = histogram_family
            .get_metric()
            .iter()
            .find(|m| m.get_label().iter().any(|l| l.value() == "tokenize"))
            .expect("tokenize metric should exist");
        assert_eq!(tokenize_metric.get_histogram().get_sample_count(), 1);
        assert!(
            (tokenize_metric.get_histogram().get_sample_sum() - 8.0).abs() < 0.001,
            "tokenize latency should be 8.0ms"
        );
    }

    #[test]
    fn test_chat_stream_typed_metrics_are_skipped_in_json() {
        use crate::protocols::common::metrics::LLMMetricAnnotation;

        let without_metrics = make_chat_stream_annotated("hello").data.unwrap();
        let mut with_metrics = without_metrics.clone();
        with_metrics.llm_metrics = Some(LLMMetricAnnotation {
            input_tokens: 1,
            output_tokens: 2,
            chunk_tokens: 1,
            cached_tokens: Some(1),
            image_tokens: Some(300),
            prefill_worker_id: None,
            prefill_dp_rank: None,
            prefill_worker_type: None,
            decode_worker_id: None,
            decode_dp_rank: None,
            decode_worker_type: None,
            tokenize_latency: None,
            detokenize_total_latency: None,
            detokenize_count: None,
            ..Default::default()
        });

        let without_json = serde_json::to_value(&without_metrics).unwrap();
        let with_json = serde_json::to_value(&with_metrics).unwrap();

        assert_eq!(with_json, without_json);
        assert!(with_json.get("llm_metrics").is_none());

        let mut inbound_json = without_json;
        inbound_json["llm_metrics"] = serde_json::json!({
            "input_tokens": 1,
            "output_tokens": 2,
            "chunk_tokens": 1,
            "cached_tokens": 1,
            "image_tokens": 300
        });
        let inbound: NvCreateChatCompletionStreamResponse =
            serde_json::from_value(inbound_json).unwrap();
        assert_eq!(inbound.llm_metrics, with_metrics.llm_metrics);
    }

    #[test]
    fn test_non_streaming_path_observes_cached_tokens() {
        use crate::preprocessor::LLMMetricAnnotation;
        use crate::types::Annotated;

        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model = "test-model";
        let expected_metric_name = "dynamo_frontend_cached_tokens";
        let expected_tokenizer_metric_name = "dynamo_frontend_tokenizer_latency_ms";
        let mut collector = metrics.clone().create_response_collector(model);

        // Create a metrics annotation event
        let mut annotated = Annotated::<
            crate::protocols::openai::chat_completions::NvCreateChatCompletionStreamResponse,
        > {
            id: None,
            data: None,
            event: Some(crate::preprocessor::ANNOTATION_LLM_METRICS.to_string()),
            comment: None,
            error: None,
        };

        let llm_metrics = LLMMetricAnnotation {
            input_tokens: 10,
            output_tokens: 20,
            chunk_tokens: 5,
            cached_tokens: Some(15),
            prefill_worker_id: None,
            prefill_dp_rank: None,
            prefill_worker_type: None,
            decode_worker_id: None,
            decode_dp_rank: None,
            decode_worker_type: None,
            tokenize_latency: Some(Duration::from_millis(8)),
            detokenize_total_latency: Some(Duration::from_micros(100)),
            detokenize_count: Some(2),
            ..Default::default()
        };

        let annotation = llm_metrics.to_annotation::<()>().unwrap();
        annotated.event = annotation.event;
        annotated.comment = annotation.comment;

        // Process via the non-streaming metrics hook
        let mut http_queue_guard = None;
        process_response_and_observe_metrics(&annotated, &mut collector, &mut http_queue_guard);

        // Drop collector so the detokenize observation fires in Drop
        drop(collector);

        // Should have observed the cached tokens from the metrics annotation event
        let metric_families = registry.gather();
        let histogram_family = metric_families
            .iter()
            .find(|mf| mf.name() == expected_metric_name)
            .expect("histogram should be registered");
        assert_eq!(
            histogram_family.get_metric()[0]
                .get_histogram()
                .get_sample_count(),
            1
        );

        let histogram_family = metric_families
            .iter()
            .find(|mf| mf.name() == expected_tokenizer_metric_name)
            .expect("histogram should be registered");

        // Find the tokenize and detokenize observations by label
        let tokenize_metric = histogram_family
            .get_metric()
            .iter()
            .find(|m| m.get_label().iter().any(|l| l.value() == "tokenize"))
            .expect("tokenize metric should exist");
        assert_eq!(tokenize_metric.get_histogram().get_sample_count(), 1);

        let detokenize_metric = histogram_family
            .get_metric()
            .iter()
            .find(|m| m.get_label().iter().any(|l| l.value() == "detokenize"))
            .expect("detokenize metric should exist");
        assert_eq!(detokenize_metric.get_histogram().get_sample_count(), 1);
        // Average: 100us total / 2 samples = 50us = 0.05ms
        assert!(
            (detokenize_metric.get_histogram().get_sample_sum() - 0.05).abs() < 0.001,
            "detokenize average latency should be 0.05ms, got {}",
            detokenize_metric.get_histogram().get_sample_sum()
        );
    }

    #[test]
    fn test_error_type_as_str() {
        assert_eq!(ErrorType::None.as_str(), "");
        assert_eq!(ErrorType::Validation.as_str(), "validation");
        assert_eq!(ErrorType::NotFound.as_str(), "not_found");
        assert_eq!(ErrorType::Overload.as_str(), "overload");
        assert_eq!(ErrorType::Unavailable.as_str(), "unavailable");
        assert_eq!(ErrorType::Cancelled.as_str(), "cancelled");
        assert_eq!(ErrorType::ResponseTimeout.as_str(), "response_timeout");
        assert_eq!(ErrorType::Internal.as_str(), "internal");
        assert_eq!(ErrorType::NotImplemented.as_str(), "not_implemented");
    }

    #[test]
    fn test_inflight_guard_marks_success_with_correct_error_type() {
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model = "test-model";

        {
            let mut guard =
                metrics
                    .clone()
                    .create_inflight_guard(model, Endpoint::ChatCompletions, false, "");
            guard.mark_ok();
        } // guard drops here

        // Verify counter incremented with status=success, error_type=""
        let counter_value = metrics
            .request_counter
            .with_label_values(&[
                model,
                Endpoint::ChatCompletions.as_str(),
                RequestType::Unary.as_str(),
                Status::Success.as_str(),
                ErrorType::None.as_str(),
            ])
            .get();
        assert_eq!(counter_value, 1);

        let started_counter_value = metrics
            .request_started_counter
            .with_label_values(&[
                model,
                Endpoint::ChatCompletions.as_str(),
                RequestType::Unary.as_str(),
            ])
            .get();
        assert_eq!(started_counter_value, 1);
    }

    #[test]
    fn test_inflight_guard_marks_validation_error() {
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model = "test-model";

        {
            let mut guard =
                metrics
                    .clone()
                    .create_inflight_guard(model, Endpoint::ChatCompletions, false, "");
            guard.mark_error(ErrorType::Validation);
        } // guard drops here

        // Verify counter incremented with status=error, error_type=validation
        let counter_value = metrics
            .request_counter
            .with_label_values(&[
                model,
                Endpoint::ChatCompletions.as_str(),
                RequestType::Unary.as_str(),
                Status::Error.as_str(),
                ErrorType::Validation.as_str(),
            ])
            .get();
        assert_eq!(counter_value, 1);
    }

    #[test]
    fn test_inflight_guard_classifies_cancellation_separately_for_tracing() {
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model = "test-model";
        let mut guard = metrics.clone().create_inflight_guard(
            model,
            Endpoint::ChatCompletions,
            true,
            "cancelled-request",
        );
        guard.mark_error(ErrorType::Cancelled);

        assert_eq!(guard.request_completion(), RequestCompletion::Cancelled);
        drop(guard);

        // Keep the existing Prometheus contract while tracing reports cancellation
        // as an expected request outcome rather than a span error.
        let counter_value = metrics
            .request_counter
            .with_label_values(&[
                model,
                Endpoint::ChatCompletions.as_str(),
                RequestType::Stream.as_str(),
                Status::Error.as_str(),
                ErrorType::Cancelled.as_str(),
            ])
            .get();
        assert_eq!(counter_value, 1);
    }

    #[test]
    fn test_inflight_guard_defaults_to_internal_error_on_drop() {
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model = "test-model";

        {
            let _guard =
                metrics
                    .clone()
                    .create_inflight_guard(model, Endpoint::ChatCompletions, false, "");
            // Don't call mark_ok() or mark_error() - simulate panic/unhandled error
        } // guard drops with default error_type=Internal

        // Verify counter incremented with status=error, error_type=internal
        let counter_value = metrics
            .request_counter
            .with_label_values(&[
                model,
                Endpoint::ChatCompletions.as_str(),
                RequestType::Unary.as_str(),
                Status::Error.as_str(),
                ErrorType::Internal.as_str(),
            ])
            .get();
        assert_eq!(counter_value, 1);
    }

    #[test]
    fn test_active_requests_tracks_inflight_guard() {
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model = "test-model-active";

        // Both gauges start at 0
        assert_eq!(metrics.inflight_gauge.with_label_values(&[model]).get(), 0);
        assert_eq!(
            metrics
                .active_requests_gauge
                .with_label_values(&[model])
                .get(),
            0
        );

        {
            let _guard = metrics.clone().create_inflight_guard(
                model,
                Endpoint::ChatCompletions,
                true,
                "req-1",
            );

            // Both gauges increment together
            assert_eq!(metrics.inflight_gauge.with_label_values(&[model]).get(), 1);
            assert_eq!(
                metrics
                    .active_requests_gauge
                    .with_label_values(&[model])
                    .get(),
                1
            );

            {
                let _guard2 = metrics.clone().create_inflight_guard(
                    model,
                    Endpoint::ChatCompletions,
                    true,
                    "req-2",
                );
                assert_eq!(metrics.inflight_gauge.with_label_values(&[model]).get(), 2);
                assert_eq!(
                    metrics
                        .active_requests_gauge
                        .with_label_values(&[model])
                        .get(),
                    2
                );
            }
            // guard2 dropped
            assert_eq!(metrics.inflight_gauge.with_label_values(&[model]).get(), 1);
            assert_eq!(
                metrics
                    .active_requests_gauge
                    .with_label_values(&[model])
                    .get(),
                1
            );
        }
        // guard dropped — both back to 0
        assert_eq!(metrics.inflight_gauge.with_label_values(&[model]).get(), 0);
        assert_eq!(
            metrics
                .active_requests_gauge
                .with_label_values(&[model])
                .get(),
            0
        );
    }

    #[test]
    fn test_lifecycle_constructors_preserve_model_label_casing() {
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model = "Llama-3.1-8B-Instruct";

        let _inflight = metrics.clone().create_inflight_guard(
            model,
            Endpoint::ChatCompletions,
            true,
            "req-case",
        );
        let mut collector = metrics.clone().create_response_collector(model);
        collector.observe_response(100, 1);
        let _queue = metrics.clone().create_http_queue_guard(model);

        // Drop the lifecycle guards so their Drop impls record
        // completion-time metrics (request_counter, request_duration,
        // detokenize observations) before we gather. Without this the
        // regression only covers constructor-time labels.
        drop(_inflight);
        drop(collector);
        drop(_queue);

        let metric_families = registry.gather();
        let observed: Vec<String> = metric_families
            .iter()
            .flat_map(|mf| mf.get_metric())
            .flat_map(|m| m.get_label())
            .filter(|l| l.name() == "model")
            .map(|l| l.value().to_string())
            .collect();

        assert!(
            !observed.is_empty(),
            "expected at least one metric to carry a model label"
        );
        for value in &observed {
            assert_eq!(
                value, model,
                "model label was modified; expected original casing to be preserved"
            );
        }
    }

    #[test]
    fn test_all_error_types_recorded_correctly() {
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model = "test-model";
        let endpoint = Endpoint::ChatCompletions;

        // Test each error type
        let error_types = vec![
            ErrorType::Validation,
            ErrorType::NotFound,
            ErrorType::Overload,
            ErrorType::Unavailable,
            ErrorType::Cancelled,
            ErrorType::ResponseTimeout,
            ErrorType::Internal,
            ErrorType::NotImplemented,
        ];

        for error_type in &error_types {
            let mut guard = metrics
                .clone()
                .create_inflight_guard(model, endpoint, false, "");
            guard.mark_error(error_type.clone());
            drop(guard);
        }

        // Verify each error type recorded correctly
        for error_type in &error_types {
            let counter_value = metrics
                .request_counter
                .with_label_values(&[
                    model,
                    endpoint.as_str(),
                    RequestType::Unary.as_str(),
                    Status::Error.as_str(),
                    error_type.as_str(),
                ])
                .get();
            assert_eq!(
                counter_value,
                1,
                "Should have 1 request for error_type={}",
                error_type.as_str()
            );
        }
    }

    #[test]
    fn semantic_failure_metric_honors_custom_prefix() {
        let registry = prometheus::Registry::new();
        let counter = new_failure_counter(Some("nv_llm_http_service"));
        counter
            .with_label_values(&["Internal", "runtime.internal"])
            .inc();
        registry.register(Box::new(counter)).unwrap();

        assert!(
            registry
                .gather()
                .iter()
                .any(|family| family.name() == "nv_llm_http_service_failures_total")
        );
    }

    #[test]
    fn semantic_failure_metric_has_only_class_and_reason_labels() {
        use dynamo_runtime::error::{DynamoError, ErrorClass, ErrorReason};

        let registry = prometheus::Registry::new();
        let counter = new_failure_counter(None);
        registry.register(Box::new(counter.clone())).unwrap();
        let error = DynamoError::builder()
            .class(ErrorClass::InvalidRequest)
            .reason(ErrorReason::new("request.invalid").unwrap())
            .build();

        record_failure_into(&counter, &error);

        let family = registry
            .gather()
            .into_iter()
            .find(|family| family.name() == "dynamo_frontend_failures_total")
            .expect("dynamo_frontend_failures_total metric");
        let metric = family
            .get_metric()
            .iter()
            .find(|metric| {
                metric
                    .get_label()
                    .iter()
                    .any(|label| label.name() == "reason" && label.value() == "request.invalid")
            })
            .expect("request.invalid sample");
        let labels: Vec<_> = metric
            .get_label()
            .iter()
            .map(|label| (label.name(), label.value()))
            .collect();
        assert_eq!(
            labels,
            vec![("class", "InvalidRequest"), ("reason", "request.invalid")]
        );
        assert_eq!(metric.get_counter().value(), 1.0);
    }

    #[test]
    fn test_multiple_requests_different_error_types() {
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model = "test-model";

        // Record 2 validation errors, 3 internal errors, 1 success
        for _ in 0..2 {
            let mut guard =
                metrics
                    .clone()
                    .create_inflight_guard(model, Endpoint::ChatCompletions, false, "");
            guard.mark_error(ErrorType::Validation);
            drop(guard);
        }

        for _ in 0..3 {
            let mut guard =
                metrics
                    .clone()
                    .create_inflight_guard(model, Endpoint::Completions, false, "");
            guard.mark_error(ErrorType::Internal);
            drop(guard);
        }

        {
            let mut guard =
                metrics
                    .clone()
                    .create_inflight_guard(model, Endpoint::Embeddings, false, "");
            guard.mark_ok();
            drop(guard);
        }

        // Check validation errors (2 from ChatCompletions)
        let validation_count = metrics
            .request_counter
            .with_label_values(&[
                model,
                Endpoint::ChatCompletions.as_str(),
                RequestType::Unary.as_str(),
                Status::Error.as_str(),
                ErrorType::Validation.as_str(),
            ])
            .get();
        assert_eq!(validation_count, 2);

        // Check internal errors (3 from Completions)
        let internal_count = metrics
            .request_counter
            .with_label_values(&[
                model,
                Endpoint::Completions.as_str(),
                RequestType::Unary.as_str(),
                Status::Error.as_str(),
                ErrorType::Internal.as_str(),
            ])
            .get();
        assert_eq!(internal_count, 3);

        // Check success (1 from Embeddings)
        let success_count = metrics
            .request_counter
            .with_label_values(&[
                model,
                Endpoint::Embeddings.as_str(),
                RequestType::Unary.as_str(),
                Status::Success.as_str(),
                ErrorType::None.as_str(),
            ])
            .get();
        assert_eq!(success_count, 1);
    }

    fn run_event_converter(
        annotated: crate::types::Annotated<String>,
    ) -> Result<Option<Event>, axum::Error> {
        let metrics = Arc::new(Metrics::new());
        let mut collector = ResponseMetricCollector::new(metrics, "test-model".to_string());
        let mut http_queue_guard: Option<HttpQueueGuard> = None;
        process_response_using_event_converter_and_observe_metrics(
            EventConverter::from(annotated),
            &mut collector,
            &mut http_queue_guard,
        )
    }

    fn run_chat_stream_event_converter(
        annotated: crate::types::Annotated<
            crate::protocols::openai::chat_completions::NvCreateChatCompletionStreamResponse,
        >,
    ) -> Result<Option<Event>, axum::Error> {
        let metrics = Arc::new(Metrics::new());
        let mut collector = ResponseMetricCollector::new(metrics, "test-model".to_string());
        let mut http_queue_guard: Option<HttpQueueGuard> = None;
        process_chat_response_using_event_converter_and_observe_metrics(
            EventConverter::from(annotated),
            &mut collector,
            &mut http_queue_guard,
            ReasoningField::default(),
        )
    }

    fn make_chat_stream_annotated(
        content: &str,
    ) -> crate::types::Annotated<
        crate::protocols::openai::chat_completions::NvCreateChatCompletionStreamResponse,
    > {
        use dynamo_protocols::types::{
            ChatChoiceStream, ChatCompletionMessageContent, ChatCompletionStreamResponseDelta,
            CreateChatCompletionStreamResponse,
        };

        #[allow(deprecated)]
        let choice = ChatChoiceStream {
            index: 0,
            delta: ChatCompletionStreamResponseDelta {
                content: Some(ChatCompletionMessageContent::Text(content.to_string())),
                function_call: None,
                tool_calls: None,
                role: None,
                refusal: None,
                reasoning_content: None,
            },
            finish_reason: None,
            logprobs: None,
        };

        crate::types::Annotated {
            id: Some("test-id".to_string()),
            data: Some(
                crate::protocols::openai::chat_completions::NvCreateChatCompletionStreamResponse {
                    inner: CreateChatCompletionStreamResponse {
                        id: "test-id".to_string(),
                        choices: vec![choice],
                        created: 0,
                        model: "test-model".to_string(),
                        system_fingerprint: None,
                        object: "chat.completion.chunk".to_string(),
                        usage: None,
                        service_tier: None,
                    },
                    nvext: None,
                    prompt_logprobs: None,
                    llm_metrics: None,
                    tool_call_completion: Vec::new(),
                },
            ),
            event: None,
            comment: None,
            error: None,
        }
    }

    fn extract_sse_data_json(event: Event) -> serde_json::Value {
        let body = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("tokio runtime")
            .block_on(async move {
                use axum::{
                    body::to_bytes,
                    response::{IntoResponse, Sse},
                };
                let stream = futures::stream::iter(vec![Ok::<Event, axum::Error>(event)]);
                let response = Sse::new(stream).into_response();
                let bytes = to_bytes(response.into_body(), 1 << 20)
                    .await
                    .expect("body bytes");
                String::from_utf8(bytes.to_vec()).expect("utf8 body")
            });

        let data = body
            .lines()
            .find_map(|line| line.strip_prefix("data: "))
            .expect("SSE data line");
        serde_json::from_str(data).unwrap_or_else(|e| {
            panic!("failed to parse JSON from SSE data: {e}\nbody: {body}\ndata: {data}")
        })
    }

    fn error_annotated(
        error: Option<dynamo_runtime::error::DynamoError>,
        comment: Option<Vec<String>>,
    ) -> crate::types::Annotated<String> {
        crate::types::Annotated {
            data: None,
            id: None,
            event: Some("error".to_string()),
            comment,
            error,
        }
    }

    #[test]
    fn test_error_event_uses_semantic_identity_without_diagnostic() {
        use dynamo_runtime::error::DynamoError;
        let result = run_event_converter(error_annotated(
            Some(DynamoError::msg("image load failed: 403 Forbidden")),
            None,
        ));
        assert!(result.is_err());
        let message = result.unwrap_err().to_string();
        assert!(message.contains("class=Internal"));
        assert!(message.contains("reason=runtime.unclassified"));
        assert!(!message.contains("403 Forbidden"));
    }

    #[test]
    fn test_error_event_does_not_expose_comment() {
        let result =
            run_event_converter(error_annotated(None, Some(vec!["connection lost".into()])));
        assert!(result.is_err());
        let message = result.unwrap_err().to_string();
        assert_eq!(message, "backend stream error");
        assert!(!message.contains("connection lost"));
    }

    #[test]
    fn test_error_event_without_identity_is_generic() {
        let result = run_event_converter(error_annotated(None, None));
        assert!(result.is_err());
        assert_eq!(result.unwrap_err().to_string(), "backend stream error");
    }

    #[test]
    fn test_event_converter_serializes_chat_stream_response_as_json_data() {
        let event = run_chat_stream_event_converter(make_chat_stream_annotated("hello"))
            .unwrap()
            .expect("chat stream response should produce an SSE event");

        let json = extract_sse_data_json(event);
        assert_eq!(json["id"], "test-id");
        assert_eq!(json["object"], "chat.completion.chunk");
        assert_eq!(json["choices"][0]["delta"]["content"], "hello");
    }

    #[test]
    fn test_event_converter_json_data_round_trips_escaped_content() {
        let content = "line1\nline2\r\"quoted\" \\\\ slash 你好 🚀";
        let event = run_chat_stream_event_converter(make_chat_stream_annotated(content))
            .unwrap()
            .expect("chat stream response should produce an SSE event");

        let json = extract_sse_data_json(event);
        assert_eq!(json["choices"][0]["delta"]["content"], content);
    }

    #[test]
    fn test_comment_newlines_sanitized() {
        let annotated = crate::types::Annotated::<String> {
            data: Some("test".to_string()),
            id: None,
            event: Some("metrics".to_string()),
            comment: Some(vec!["line1\nline2\r\nline3".into()]),
            error: None,
        };
        let event = run_event_converter(annotated)
            .unwrap()
            .expect("data event with comment should be returned");
        let debug = format!("{:?}", event);
        assert!(
            debug.contains(": line1 line2  line3\\n"),
            "comment newlines should be replaced before Event::comment: {debug}"
        );
    }

    #[test]
    fn test_observe_embedding_latency_records_observation() {
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model = "embed-test";
        assert_eq!(metrics.embedding_latency_count(model), 0);
        assert_eq!(metrics.embedding_latency_sum(model), 0.0);

        metrics.observe_embedding_latency(model, 0.005);
        metrics.observe_embedding_latency(model, 0.020);

        assert_eq!(metrics.embedding_latency_count(model), 2);
        assert!((metrics.embedding_latency_sum(model) - 0.025).abs() < 1e-9);
    }

    #[test]
    fn test_observe_embedding_latency_drops_invalid_values() {
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let model = "embed-test";

        metrics.observe_embedding_latency(model, -0.5);
        metrics.observe_embedding_latency(model, f64::NAN);
        metrics.observe_embedding_latency(model, f64::INFINITY);
        metrics.observe_embedding_latency(model, f64::NEG_INFINITY);

        assert_eq!(metrics.embedding_latency_count(model), 0);
        assert_eq!(metrics.embedding_latency_sum(model), 0.0);

        // Zero is allowed (boundary case for finite, non-negative).
        metrics.observe_embedding_latency(model, 0.0);
        assert_eq!(metrics.embedding_latency_count(model), 1);
    }

    #[test]
    fn test_observe_embedding_latency_partitions_by_model() {
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        let m1 = "embed-a";
        let m2 = "embed-b";

        metrics.observe_embedding_latency(m1, 0.010);
        metrics.observe_embedding_latency(m1, 0.030);
        metrics.observe_embedding_latency(m2, 0.005);

        assert_eq!(metrics.embedding_latency_count(m1), 2);
        assert_eq!(metrics.embedding_latency_count(m2), 1);
        assert!((metrics.embedding_latency_sum(m1) - 0.040).abs() < 1e-9);
        assert!((metrics.embedding_latency_sum(m2) - 0.005).abs() < 1e-9);
    }

    #[test]
    fn test_embedding_latency_registered_in_registry() {
        let metrics = Arc::new(Metrics::new());
        let registry = prometheus::Registry::new();
        metrics.register(&registry).unwrap();

        metrics.observe_embedding_latency("embed-test", 0.012);

        let metric_families = registry.gather();
        let found = metric_families.iter().any(|mf| {
            mf.name()
                .ends_with(frontend_service::EMBEDDING_LATENCY_SECONDS)
        });
        assert!(
            found,
            "embedding_latency_seconds histogram must be registered with the registry"
        );
    }

    /// Request payload capture must be transparent: the collector, the client
    /// response and the handler's error path behave identically with capture on
    /// (`scan_aggregate_with_future`) and off (#11349).
    mod capture_transparency {
        use super::*;
        use crate::http::service::openai::check_for_backend_error;
        use crate::http::service::service_v2::BackendErrorCheck;
        use crate::preprocessor::OpenAIPreprocessor;
        use crate::protocols::common::llm_backend::{BackendOutput, FinishReason};
        use crate::protocols::common::metrics::ANNOTATION_PAYLOAD_USAGE;
        use crate::protocols::openai::ParsingOptions;
        use crate::protocols::openai::chat_completions::aggregator::ChatCompletionAggregator;
        use crate::protocols::openai::chat_completions::{
            NvCreateChatCompletionRequest, NvCreateChatCompletionResponse,
        };
        use crate::request_trace::payload_stream::scan_aggregate_with_future;
        use crate::types::Annotated;
        use dynamo_runtime::engine::AsyncEngineContext;
        use futures::{Stream, StreamExt};

        const MODEL: &str = "test-model";
        const REQUEST_ID: &str = "test-id";
        const INPUT_TOKENS: usize = 7;
        const TAIL_CACHED_TOKENS: usize = 2;

        #[derive(Debug)]
        struct TestContext;

        #[async_trait::async_trait]
        impl AsyncEngineContext for TestContext {
            fn id(&self) -> &str {
                REQUEST_ID
            }
            fn stop_generating(&self) {}
            fn is_stopped(&self) -> bool {
                false
            }
            fn is_killed(&self) -> bool {
                false
            }
            async fn stopped(&self) {}
            async fn killed(&self) {}
            fn stop(&self) {}
            fn kill(&self) {}
            fn link_child(&self, _: Arc<dyn AsyncEngineContext>) {}
        }

        fn backend_output(
            text: &str,
            token_ids: Vec<u32>,
            finish_reason: Option<FinishReason>,
        ) -> BackendOutput {
            BackendOutput {
                token_ids,
                tokens: vec![],
                text: Some(text.to_string()),
                cum_log_probs: None,
                log_probs: None,
                top_logprobs: None,
                finish_reason,
                stop_reason: None,
                index: Some(0),
                completion_usage: None,
                disaggregated_params: None,
                encoder_result: None,
                worker_trace_link: None,
                engine_data: None,
                routing_data: None,
                jailed_text: None,
            }
        }

        /// A backend that answers "Hello world" in two chunks and reports cached
        /// prompt tokens on the final one.
        fn backend_success() -> Vec<Annotated<BackendOutput>> {
            let mut last = backend_output("world", vec![2, 3], Some(FinishReason::Stop));
            last.completion_usage = Some(dynamo_protocols::types::CompletionUsage {
                prompt_tokens: INPUT_TOKENS as u32,
                completion_tokens: 3,
                total_tokens: (INPUT_TOKENS + 3) as u32,
                prompt_tokens_details: Some(dynamo_protocols::types::PromptTokensDetails {
                    cached_tokens: Some(TAIL_CACHED_TOKENS as u32),
                    ..Default::default()
                }),
                ..Default::default()
            });
            vec![
                Annotated::from_data(backend_output("Hello ", vec![1], None)),
                Annotated::from_data(last),
            ]
        }

        /// What the frontend hands the non-streaming chat handler: the real
        /// preprocessor output for `outputs`, with payload capture off or on.
        /// The two differ in shape (capture on ends with a `payload_usage`
        /// chunk), which is exactly what the handler chain must not notice.
        fn preprocessed(
            capture: bool,
            outputs: Vec<Annotated<BackendOutput>>,
        ) -> impl Stream<Item = Annotated<NvCreateChatCompletionStreamResponse>> + Send + 'static
        {
            let request = NvCreateChatCompletionRequest {
                inner: dynamo_protocols::types::CreateChatCompletionRequest {
                    model: MODEL.to_string(),
                    messages: vec![dynamo_protocols::types::ChatCompletionRequestMessage::User(
                        dynamo_protocols::types::ChatCompletionRequestUserMessage {
                            content:
                                dynamo_protocols::types::ChatCompletionRequestUserMessageContent::Text(
                                    "Hello".to_string(),
                                ),
                            name: None,
                        },
                    )],
                    ..Default::default()
                },
                ..Default::default()
            };
            // The non-streaming handler forces usage on before preprocessing
            // (`force_include_usage`), so the client always gets a usage block.
            let mut request = request;
            request.inner.stream_options =
                Some(dynamo_protocols::types::ChatCompletionStreamOptions {
                    include_usage: true,
                    continuous_usage_stats: false,
                });
            let mut generator = request.response_generator(REQUEST_ID.to_string());
            generator.update_isl(INPUT_TOKENS as u32);
            OpenAIPreprocessor::transform_postprocessor_stream(
                futures::stream::iter(outputs),
                Box::new(generator),
                Arc::new(TestContext),
                capture,
                false,
                None,
                Default::default(),
            )
        }

        /// Run `outputs` through the handler chain with capture off, and with
        /// capture on through the scan; return both registries and results plus
        /// the capture record future.
        async fn both_legs(
            outputs: Vec<Annotated<BackendOutput>>,
        ) -> (
            (Registry, Result<NvCreateChatCompletionResponse, Rejected>),
            (Registry, Result<NvCreateChatCompletionResponse, Rejected>),
            impl std::future::Future<Output = crate::request_trace::payload_stream::PayloadOutcome>,
        ) {
            let plain = observe_and_aggregate(preprocessed(false, outputs.clone())).await;
            let (captured, future) = scan_aggregate_with_future(
                Box::pin(preprocessed(true, outputs)),
                crate::protocols::openai::ParsingOptions::default(),
            );
            let capture = observe_and_aggregate(captured).await;
            (plain, capture, future)
        }

        fn chat_chunk(
            content: Option<&str>,
            finish: Option<dynamo_protocols::types::FinishReason>,
            llm_metrics: Option<LLMMetricAnnotation>,
        ) -> Annotated<NvCreateChatCompletionStreamResponse> {
            #[allow(deprecated)]
            let delta = dynamo_protocols::types::ChatCompletionStreamResponseDelta {
                role: Some(dynamo_protocols::types::Role::Assistant),
                content: content.map(|c| {
                    dynamo_protocols::types::ChatCompletionMessageContent::Text(c.to_string())
                }),
                tool_calls: None,
                function_call: None,
                refusal: None,
                reasoning_content: None,
            };
            let choice = dynamo_protocols::types::ChatChoiceStream {
                index: 0,
                delta,
                finish_reason: finish,
                logprobs: None,
            };
            let response = NvCreateChatCompletionStreamResponse {
                inner: dynamo_protocols::types::CreateChatCompletionStreamResponse {
                    id: REQUEST_ID.to_string(),
                    choices: vec![choice],
                    created: 0,
                    model: MODEL.to_string(),
                    system_fingerprint: None,
                    object: "chat.completion.chunk".to_string(),
                    usage: None,
                    service_tier: None,
                },
                nvext: None,
                llm_metrics,
                prompt_logprobs: None,
                tool_call_completion: Vec::new(),
            };
            Annotated {
                data: Some(response),
                id: None,
                event: None,
                comment: None,
                error: None,
            }
        }

        /// Per-chunk metrics as the preprocessor attaches them to content
        /// chunks: no `cached_tokens`, which only the usage tail carries.
        fn chunk_metrics(chunk_tokens: usize, output_tokens: usize) -> LLMMetricAnnotation {
            LLMMetricAnnotation {
                input_tokens: INPUT_TOKENS,
                output_tokens,
                chunk_tokens,
                cached_tokens: None,
                ..Default::default()
            }
        }

        /// The tool-call-jail tail shape: usage data plus cumulative metrics as a
        /// `payload_usage` annotation. The tag is set explicitly because
        /// `to_annotation` always emits `llm_metrics`.
        fn payload_usage_tail(
            output_tokens: usize,
        ) -> Annotated<NvCreateChatCompletionStreamResponse> {
            let tail_metrics = LLMMetricAnnotation {
                cached_tokens: Some(TAIL_CACHED_TOKENS),
                ..chunk_metrics(0, output_tokens)
            };
            let annotation = tail_metrics.to_annotation::<()>().unwrap();
            let mut tail = chat_chunk(None, None, None);
            {
                let data = tail.data.as_mut().unwrap();
                data.inner.choices = vec![];
                data.inner.usage = Some(dynamo_protocols::types::CompletionUsage {
                    prompt_tokens: INPUT_TOKENS as u32,
                    completion_tokens: output_tokens as u32,
                    total_tokens: (INPUT_TOKENS + output_tokens) as u32,
                    ..Default::default()
                });
            }
            tail.event = Some(ANNOTATION_PAYLOAD_USAGE.to_string());
            tail.comment = annotation.comment;
            tail
        }

        /// What the collector wrote, minus durations: TTFT and ITL sums are
        /// wall-clock and cannot be compared across runs, so only their
        /// sample counts are part of the signature.
        #[derive(Debug, PartialEq)]
        struct MetricSignature {
            output_tokens_total: u64,
            /// (sample count, sample sum)
            isl: (u64, u64),
            osl: (u64, u64),
            cached_tokens: (u64, u64),
            ttft_samples: u64,
            itl_samples: u64,
        }

        fn signature(registry: &Registry) -> MetricSignature {
            let families = registry.gather();
            let histogram = |name: &str| -> (u64, u64) {
                families
                    .iter()
                    .find(|mf| mf.name() == name)
                    .map(|mf| {
                        let h = mf.get_metric()[0].get_histogram();
                        (h.get_sample_count(), h.get_sample_sum() as u64)
                    })
                    .unwrap_or((0, 0))
            };
            let counter = |name: &str| -> u64 {
                families
                    .iter()
                    .find(|mf| mf.name() == name)
                    .map(|mf| mf.get_metric()[0].get_counter().value() as u64)
                    .unwrap_or(0)
            };
            MetricSignature {
                output_tokens_total: counter("dynamo_frontend_output_tokens_total"),
                isl: histogram("dynamo_frontend_input_sequence_tokens"),
                osl: histogram("dynamo_frontend_output_sequence_tokens"),
                cached_tokens: histogram("dynamo_frontend_cached_tokens"),
                ttft_samples: histogram("dynamo_frontend_time_to_first_token_seconds").0,
                itl_samples: histogram("dynamo_frontend_inter_token_latency_seconds").0,
            }
        }

        /// Where the non-streaming handler chain rejected the stream, if it did.
        #[derive(Debug, PartialEq)]
        enum Rejected {
            Preflight,
            Aggregation,
        }

        /// Drive `stream` through the non-streaming HTTP handler's chain (observe,
        /// backend-error preflight, aggregate) against a private registry.
        /// Consuming the stream drops the collector, which flushes ITL and OSL.
        async fn observe_and_aggregate(
            stream: impl Stream<Item = Annotated<NvCreateChatCompletionStreamResponse>> + Send + 'static,
        ) -> (Registry, Result<NvCreateChatCompletionResponse, Rejected>) {
            let metrics = Arc::new(Metrics::new_with_prefix(None));
            let registry = Registry::new();
            metrics.register(&registry).unwrap();
            let mut collector = metrics.create_response_collector(MODEL);
            let mut http_queue_guard = None;

            let observed = stream.inspect(move |response| {
                process_chat_response_and_observe_metrics(
                    response,
                    &mut collector,
                    &mut http_queue_guard,
                );
            });
            let checked =
                match check_for_backend_error(observed, BackendErrorCheck::UntilFirstEvent).await {
                    Ok(checked) => checked,
                    Err(_) => return (registry, Err(Rejected::Preflight)),
                };
            let response = NvCreateChatCompletionResponse::from_annotated_stream(
                checked,
                ParsingOptions::default(),
            )
            .await
            .map_err(|_| Rejected::Aggregation);
            (registry, response)
        }

        fn ok(
            result: Result<NvCreateChatCompletionResponse, Rejected>,
        ) -> NvCreateChatCompletionResponse {
            result.expect("production-shaped stream must aggregate")
        }

        /// The invariant #11349 is about: the same backend output, through the
        /// real preprocessor with capture off and on, gives the collector the
        /// same numbers and the client the same response.
        #[tokio::test]
        async fn test_capture_does_not_change_metrics_or_response() {
            let ((plain_registry, plain), (capture_registry, capture), future) =
                both_legs(backend_success()).await;

            let signature_plain = signature(&plain_registry);
            assert_eq!(
                signature(&capture_registry),
                signature_plain,
                "capture changed the collector output"
            );
            // Not agreeing zeros: the real request.
            assert_eq!(
                signature_plain,
                MetricSignature {
                    output_tokens_total: 3,
                    isl: (1, INPUT_TOKENS as u64),
                    osl: (1, 3),
                    cached_tokens: (1, TAIL_CACHED_TOKENS as u64),
                    ttft_samples: 1,
                    itl_samples: 2,
                }
            );

            let (mut plain, mut capture) = (ok(plain), ok(capture));
            // `created` is wall-clock per generator; everything else must match.
            plain.inner.created = 0;
            capture.inner.created = 0;
            assert_eq!(plain, capture, "capture changed the client response");
            assert_eq!(plain.inner.model, MODEL);
            assert_eq!(
                plain.inner.choices[0].message.content.as_ref().unwrap(),
                &dynamo_protocols::types::ChatCompletionMessageContent::Text(
                    "Hello world".to_string()
                )
            );

            let outcome = future.await;
            assert!(outcome.drop_reason.is_none());
            let record = outcome.response.expect("capture must produce a record");
            assert_eq!(record.inner.model, MODEL);
            assert_eq!(record.inner.usage.as_ref().unwrap().completion_tokens, 3);
        }

        /// A chunk carrying both typed `llm_metrics` and a `payload_usage`
        /// annotation (tool-call jail shape) is observed once, typed form
        /// winning, with capture on or off.
        #[tokio::test]
        async fn test_capture_observes_dual_carrier_chunk_once() {
            let dual_carrier = || {
                let mut tail = payload_usage_tail(3);
                tail.data.as_mut().unwrap().llm_metrics = Some(chunk_metrics(5, 5));
                vec![tail]
            };
            let (plain_registry, _) =
                observe_and_aggregate(futures::stream::iter(dual_carrier())).await;
            let (captured, _future) = scan_aggregate_with_future(
                futures::stream::iter(dual_carrier()),
                crate::protocols::openai::ParsingOptions::default(),
            );
            let (capture_registry, _) = observe_and_aggregate(captured).await;

            let expected = MetricSignature {
                output_tokens_total: 5,
                isl: (1, INPUT_TOKENS as u64),
                osl: (1, 5),
                cached_tokens: (0, 0),
                ttft_samples: 1,
                itl_samples: 0,
            };
            assert_eq!(signature(&plain_registry), expected);
            assert_eq!(signature(&capture_registry), expected);
        }

        /// A backend error after content is surfaced to the client as an error
        /// (not an empty success) with capture on, exactly as with capture off,
        /// and the metrics observed before the error agree.
        #[tokio::test]
        async fn test_capture_surfaces_mid_stream_error_identically() {
            let outputs = vec![
                Annotated::from_data(backend_output("Hello ", vec![1], None)),
                Annotated::<BackendOutput>::from_error("invalid sampling parameter"),
            ];
            let ((plain_registry, plain), (capture_registry, capture), future) =
                both_legs(outputs).await;

            assert_eq!(plain.unwrap_err(), Rejected::Aggregation);
            assert_eq!(
                capture.unwrap_err(),
                Rejected::Aggregation,
                "capture must not turn an error into a success"
            );
            assert_eq!(signature(&capture_registry), signature(&plain_registry));

            let outcome = future.await;
            assert!(
                outcome
                    .drop_reason
                    .as_deref()
                    .unwrap()
                    .contains("invalid sampling parameter")
            );
            assert!(
                outcome.response.is_some(),
                "the pre-error prefix is recorded"
            );
        }

        /// A backend error before any data chunk is rejected by the preflight,
        /// with capture on or off. Without a leading frame nothing is observed;
        /// with a leading metrics frame that frame is observed (the observer runs
        /// ahead of the preflight, as on the streaming path) — identically in
        /// both modes.
        #[tokio::test]
        async fn test_capture_rejects_leading_error_identically() {
            // Legacy engines send a data-less `llm_metrics` frame; the
            // preprocessor forwards it untouched.
            let frame = || {
                chunk_metrics(1, 1)
                    .to_annotation::<BackendOutput>()
                    .unwrap()
            };
            let error = || Annotated::<BackendOutput>::from_error("backend failed");
            let nothing = MetricSignature {
                output_tokens_total: 0,
                isl: (0, 0),
                osl: (0, 0),
                cached_tokens: (0, 0),
                ttft_samples: 0,
                itl_samples: 0,
            };
            let one_frame = MetricSignature {
                output_tokens_total: 1,
                isl: (1, INPUT_TOKENS as u64),
                osl: (1, 1),
                cached_tokens: (0, 0),
                ttft_samples: 1,
                itl_samples: 0,
            };
            for (outputs, expected) in [
                (vec![error()], nothing),
                (vec![frame(), error()], one_frame),
            ] {
                let ((plain_registry, plain), (capture_registry, capture), future) =
                    both_legs(outputs).await;

                assert_eq!(plain.unwrap_err(), Rejected::Preflight);
                assert_eq!(capture.unwrap_err(), Rejected::Preflight);
                assert_eq!(signature(&plain_registry), expected);
                assert_eq!(signature(&capture_registry), expected);
                assert!(
                    future
                        .await
                        .drop_reason
                        .as_deref()
                        .unwrap()
                        .contains("backend failed")
                );
            }
        }
    }
}