eyes-subscriber 0.8.1

//! Boot-declared named metric definitions: what an app measures, expressed
//! as canonical eyes-query IR.
//!
//! A declared metric is a versioned named aggregate query over the unified
//! observation substrate — a request count, a p95 latency, a queue-depth
//! series. The declaration says *what* is measured; the evaluation consumer
//! (dashboard, alert, investigation) supplies the context: range, bucket
//! width, comparison windows.
//!
//! Every declaration is a **singular metric**: exactly one aggregate stage
//! holding exactly one aggregate expression aliased `value`. The server
//! enforces the same contract at manifest ingestion, so a declaration that
//! builds here is accepted there.
//!
//! # Span-creation contract
//!
//! Fields referenced by declared queries (filters, group dimensions, and
//! aggregate arguments under `fields.*`) are the cja cross-repo span
//! vocabulary and must exist **when the span is created**. Fields recorded
//! with `span.record()` after creation are not visible to queries. This is a
//! CLAUDE.md non-negotiable and applies to every declaration.

use eyes_query::{
    BinaryOp, Expr, FieldPath, FiniteF64, Function, NamedExpr, QueryDefinition, QueryDocument,
    QuerySource, Stage, QUERY_SCHEMA_VERSION,
};
use serde::Serialize;

/// Re-exported so consumers of the declaration contract (cja) can name the
/// IR types without adding their own `eyes-query` dependency.
pub use eyes_query;

/// Convenience re-exports for building declarations.
pub use eyes_query::{AggregateFunction, Literal};

/// The value domain of the metric's single `value` column. `duration` is a
/// first-class domain because the executor preserves it through every
/// aggregate: `p95(duration)` returns Duration cells, and no duration→float
/// cast exists in the engine.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum MetricValueType {
    Int,
    Float,
    Duration,
}

/// What executing this metric returns. Derived from the IR by the builder;
/// validated against the IR by the server.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum MetricResultShape {
    Scalar,
    Table,
    TimeSeries,
}

/// A named metric declaration carried by the boot manifest.
///
/// Deliberately `PartialEq` only: it embeds `eyes_query::QueryDocument`,
/// whose float-bearing IR types intentionally stop at `PartialEq` — and
/// that is why `AppManifest` drops its `Eq` derive in the 0.7.0 release.
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct NamedMetric {
    pub id: String,
    pub display_name: Option<String>,
    pub description: Option<String>,
    pub unit: Option<String>,
    pub value_type: MetricValueType,
    pub result_shape: MetricResultShape,
    pub preferred_bucket_seconds: Option<u64>,
    pub query: eyes_query::QueryDocument,
    /// The original DSL text, provenance/debugging only. Never executed.
    pub original_dsl: Option<String>,
}

/// Comparison direction of a metric-threshold severity policy.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum ThresholdComparison {
    Above,
    Below,
}

/// One severity policy of a metric threshold: a comparison direction and
/// the boundary value. The threshold crosses the wire as an `f64` number —
/// serde emits `1.0` for whole values, and the server validates by value,
/// not JSON representation, so `1.0` is accepted for an int metric.
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct ThresholdSeverity {
    pub comparison: ThresholdComparison,
    pub threshold: f64,
}

/// How multi-row results feed a threshold breach: `Any` breaches when at
/// least one numeric cell breaches; `All` when every numeric cell does.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum ThresholdGroupMatch {
    Any,
    All,
}

/// A metric-threshold declaration carried by the boot manifest: references
/// a named metric declared in the same manifest by `metric_id` (no query
/// embedded — the server joins by id). `PartialEq` only (f64 thresholds),
/// mirroring `NamedMetric`.
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct MetricThreshold {
    pub id: String,
    pub metric_id: String,
    pub description: Option<String>,
    pub window_seconds: u64,
    pub evaluation_interval_seconds: u64,
    pub evaluation_delay_seconds: u64,
    pub group_match: ThresholdGroupMatch,
    pub failure_threshold: u32,
    pub recovery_threshold: u32,
    pub enabled: bool,
    pub warning: Option<ThresholdSeverity>,
    pub critical: Option<ThresholdSeverity>,
}

/// The ergonomic Rust surface for declaring a metric threshold. All
/// validation happens in [`MetricThresholdBuilder::build`] — no I/O, no
/// server round trip. The builder does **not** enforce int-range on
/// severity thresholds: the subscriber cannot know the referenced metric's
/// `value_type`; out-of-domain values are rejected server-side at ingest.
#[derive(Debug)]
pub struct MetricThresholdBuilder {
    id: String,
    metric_id: String,
    description: Option<String>,
    window_seconds: Option<u64>,
    evaluation_interval_seconds: Option<u64>,
    evaluation_delay_seconds: u64,
    group_match: ThresholdGroupMatch,
    failure_threshold: u32,
    recovery_threshold: u32,
    enabled: bool,
    warning: Option<ThresholdSeverity>,
    critical: Option<ThresholdSeverity>,
}

impl MetricThresholdBuilder {
    /// Start a declaration referencing `metric_id` (which must be declared
    /// in the same manifest's `metrics`).
    pub fn new(id: impl Into<String>, metric_id: impl Into<String>) -> Self {
        Self {
            id: id.into(),
            metric_id: metric_id.into(),
            description: None,
            window_seconds: None,
            evaluation_interval_seconds: None,
            evaluation_delay_seconds: 0,
            group_match: ThresholdGroupMatch::Any,
            failure_threshold: 3,
            recovery_threshold: 1,
            enabled: true,
            warning: None,
            critical: None,
        }
    }

    pub fn window_seconds(mut self, v: u64) -> Self {
        self.window_seconds = Some(v);
        self
    }
    pub fn evaluation_interval_seconds(mut self, v: u64) -> Self {
        self.evaluation_interval_seconds = Some(v);
        self
    }
    pub fn evaluation_delay_seconds(mut self, v: u64) -> Self {
        self.evaluation_delay_seconds = v;
        self
    }
    pub fn group_match(mut self, v: ThresholdGroupMatch) -> Self {
        self.group_match = v;
        self
    }
    pub fn failure_threshold(mut self, v: u32) -> Self {
        self.failure_threshold = v;
        self
    }
    pub fn recovery_threshold(mut self, v: u32) -> Self {
        self.recovery_threshold = v;
        self
    }
    pub fn enabled(mut self, v: bool) -> Self {
        self.enabled = v;
        self
    }
    pub fn description(mut self, v: impl Into<String>) -> Self {
        self.description = Some(v.into());
        self
    }
    /// Declare (or replace) the warning severity policy.
    pub fn warning(mut self, comparison: ThresholdComparison, threshold: f64) -> Self {
        self.warning = Some(ThresholdSeverity {
            comparison,
            threshold,
        });
        self
    }
    /// Declare (or replace) the critical severity policy.
    pub fn critical(mut self, comparison: ThresholdComparison, threshold: f64) -> Self {
        self.critical = Some(ThresholdSeverity {
            comparison,
            threshold,
        });
        self
    }

    /// Assemble + validate: window 1..=2_678_400, interval 10..=86_400,
    /// delay 0..=600, failure/recovery thresholds 1..=100, at least one
    /// severity, finite severity thresholds. Out-of-domain (int-range)
    /// thresholds are the server's job — see the struct docs.
    pub fn build(self) -> Result<MetricThreshold, String> {
        let Self {
            id,
            metric_id,
            description,
            window_seconds,
            evaluation_interval_seconds,
            evaluation_delay_seconds,
            group_match,
            failure_threshold,
            recovery_threshold,
            enabled,
            warning,
            critical,
        } = self;
        let Some(window_seconds) = window_seconds else {
            return Err(format!("metric threshold {id}: window_seconds is required"));
        };
        if !(1..=2_678_400).contains(&window_seconds) {
            return Err(format!(
                "metric threshold {id}: window_seconds must be 1..=2678400"
            ));
        }
        let Some(evaluation_interval_seconds) = evaluation_interval_seconds else {
            return Err(format!(
                "metric threshold {id}: evaluation_interval_seconds is required"
            ));
        };
        if !(10..=86_400).contains(&evaluation_interval_seconds) {
            return Err(format!(
                "metric threshold {id}: evaluation_interval_seconds must be 10..=86400"
            ));
        }
        if evaluation_delay_seconds > 600 {
            return Err(format!(
                "metric threshold {id}: evaluation_delay_seconds must be 0..=600"
            ));
        }
        for (label, value) in [
            ("failure_threshold", failure_threshold),
            ("recovery_threshold", recovery_threshold),
        ] {
            if !(1..=100).contains(&value) {
                return Err(format!("metric threshold {id}: {label} must be 1..=100"));
            }
        }
        if warning.is_none() && critical.is_none() {
            return Err(format!(
                "metric threshold {id}: at least one of warning or critical is required"
            ));
        }
        for (label, policy) in [("warning", &warning), ("critical", &critical)] {
            if let Some(policy) = policy {
                if !policy.threshold.is_finite() {
                    return Err(format!(
                        "metric threshold {id}: {label} threshold must be finite"
                    ));
                }
            }
        }
        Ok(MetricThreshold {
            id,
            metric_id,
            description,
            window_seconds,
            evaluation_interval_seconds,
            evaluation_delay_seconds,
            group_match,
            failure_threshold,
            recovery_threshold,
            enabled,
            warning,
            critical,
        })
    }
}

/// Filter literal for [`NamedMetricBuilder::filter_eq`].
#[derive(Debug, Clone, PartialEq)]
pub enum MetricFilterValue {
    Bool(bool),
    Int(i64),
    Float(f64),
    Str(String),
}

impl From<bool> for MetricFilterValue {
    fn from(value: bool) -> Self {
        Self::Bool(value)
    }
}
impl From<i64> for MetricFilterValue {
    fn from(value: i64) -> Self {
        Self::Int(value)
    }
}
impl From<f64> for MetricFilterValue {
    fn from(value: f64) -> Self {
        Self::Float(value)
    }
}
impl From<&str> for MetricFilterValue {
    fn from(value: &str) -> Self {
        Self::Str(value.to_owned())
    }
}
impl From<String> for MetricFilterValue {
    fn from(value: String) -> Self {
        Self::Str(value)
    }
}

impl MetricFilterValue {
    fn into_literal(self) -> Result<Literal, String> {
        match self {
            Self::Bool(value) => Ok(Literal::Bool(value)),
            Self::Int(value) => Ok(Literal::Int(value)),
            Self::Float(value) => {
                let finite = FiniteF64::new(value)
                    .map_err(|error| format!("float filter value {value}: {error}"))?;
                Ok(Literal::Float(finite))
            }
            Self::Str(value) => Ok(Literal::String(value)),
        }
    }
}

/// One accumulated filter, in call order.
#[derive(Debug)]
enum FilterEntry {
    /// `path = literal` (a Where stage).
    Eq(FieldPath, MetricFilterValue),
    /// Numeric-only population restriction: `path >= f64::MIN`.
    Numeric(FieldPath),
}

/// The ergonomic Rust surface for declaring a named metric.
///
/// All validation happens in [`NamedMetricBuilder::build`] — no I/O, no
/// server round trip — so a bad declaration fails at boot in the app, and
/// the transmitted IR is exactly what the server validates again at ingest.
#[derive(Debug)]
pub struct NamedMetricBuilder {
    id: String,
    function: AggregateFunction,
    argument: Option<FieldPath>,
    /// Filters in call order: equality predicates and numeric restrictions.
    filters: Vec<FilterEntry>,
    /// Group paths; aliases are assigned positionally at build time.
    groups: Vec<FieldPath>,
    bucket_seconds: Option<u64>,
    display_name: Option<String>,
    description: Option<String>,
    unit: Option<String>,
    value_type: Option<MetricValueType>,
    preferred_bucket_seconds: Option<u64>,
    original_dsl: Option<String>,
}

impl NamedMetricBuilder {
    /// Start a declaration. `count()` has no argument; every other function
    /// requires one — a field path such as `"duration"`,
    /// `"measurement.value"`, or `"fields.queue_depth"`.
    pub fn new(
        id: impl Into<String>,
        function: AggregateFunction,
        argument: Option<&str>,
    ) -> Result<Self, String> {
        let argument = match (function, argument) {
            (AggregateFunction::Count, Some(_)) => return Err("count takes no argument".to_owned()),
            (AggregateFunction::Count, None) => None,
            (_, Some(text)) => Some(parse_path(text)?),
            (_, None) => {
                return Err(format!(
                    "{} requires an argument (a field path)",
                    function.as_str()
                ))
            }
        };
        Ok(Self {
            id: id.into(),
            function,
            argument,
            filters: Vec::new(),
            groups: Vec::new(),
            bucket_seconds: None,
            display_name: None,
            description: None,
            unit: None,
            value_type: None,
            preferred_bucket_seconds: None,
            original_dsl: None,
        })
    }

    /// Equality filter: `path = literal` (a Where stage, appended in call
    /// order).
    pub fn filter_eq(
        mut self,
        path: &str,
        value: impl Into<MetricFilterValue>,
    ) -> Result<Self, String> {
        self.filters
            .push(FilterEntry::Eq(parse_path(path)?, value.into()));
        Ok(self)
    }

    /// Numeric-only population restriction for `path`: bakes
    /// `path >= f64::MIN` into the transmitted IR — the same stage shape the
    /// legacy metric preset injects. Explicit and visible in `query_json`;
    /// never injected silently by the server. Without it, non-numeric
    /// occurrences produce the engine's canonical row errors.
    pub fn filter_numeric(mut self, path: &str) -> Result<Self, String> {
        self.filters.push(FilterEntry::Numeric(parse_path(path)?));
        Ok(self)
    }

    /// Group by a field path. The group *alias* is assigned positionally at
    /// build time (`_dim_0`, `_dim_1`, …) — the same convention as the
    /// metric preset — because a path like `fields.route` is not a valid IR
    /// identifier. Duplicate paths are rejected.
    pub fn group_by(mut self, path: &str) -> Result<Self, String> {
        let path = parse_path(path)?;
        if self.groups.contains(&path) {
            return Err(format!("duplicate group path {path:?}"));
        }
        self.groups.push(path);
        Ok(self)
    }

    /// Leading time-bucket group, alias `bucket`. Also defaults
    /// `preferred_bucket_seconds` to this width when not set explicitly.
    pub fn time_bucket(mut self, seconds: u64) -> Self {
        self.bucket_seconds = Some(seconds);
        self
    }

    pub fn display_name(mut self, v: impl Into<String>) -> Self {
        self.display_name = Some(v.into());
        self
    }
    pub fn description(mut self, v: impl Into<String>) -> Self {
        self.description = Some(v.into());
        self
    }
    pub fn unit(mut self, v: impl Into<String>) -> Self {
        self.unit = Some(v.into());
        self
    }
    /// Overrides the derived default; checked against the same policy the
    /// server enforces — an inconsistent override fails `build()`.
    pub fn value_type(mut self, v: MetricValueType) -> Self {
        self.value_type = Some(v);
        self
    }
    pub fn preferred_bucket_seconds(mut self, v: u64) -> Self {
        self.preferred_bucket_seconds = Some(v);
        self
    }
    /// Provenance only; never executed.
    pub fn original_dsl(mut self, v: impl Into<String>) -> Self {
        self.original_dsl = Some(v.into());
        self
    }

    /// Assemble + validate. Fails on unparseable paths (already rejected at
    /// each setter), duplicate groups, an inconsistent explicit
    /// `value_type`, or any structural violation of the IR — the document
    /// is round-tripped through `QueryDocument::from_json_str` so alias and
    /// structural validation runs here, at boot, not at the server.
    pub fn build(self) -> Result<NamedMetric, String> {
        let Self {
            id,
            function,
            argument,
            filters,
            groups,
            bucket_seconds,
            display_name,
            description,
            unit,
            value_type,
            preferred_bucket_seconds,
            original_dsl,
        } = self;

        // Value-type policy: the derived default, or the caller's explicit
        // choice checked against the same rules the server enforces.
        let value_type = match value_type {
            Some(explicit) => {
                if !value_type_allowed(function, argument.as_ref(), explicit) {
                    let required = derive_value_type(function, argument.as_ref())?;
                    return Err(format!(
                        "value_type must be {} for {}({})",
                        match required {
                            MetricValueType::Int => "int",
                            MetricValueType::Float => "float",
                            MetricValueType::Duration => "duration",
                        },
                        function.as_str(),
                        argument_name(argument.as_ref())
                    ));
                }
                explicit
            }
            None => derive_value_type(function, argument.as_ref())?,
        };

        // Stages in canonical order: one Where per filter in call order,
        // then the single Aggregate.
        let mut stages: Vec<Stage> = Vec::with_capacity(filters.len() + 1);
        for filter in &filters {
            let predicate = match filter {
                FilterEntry::Eq(path, value) => Expr::Binary {
                    op: BinaryOp::Eq,
                    left: Box::new(Expr::Path { path: path.clone() }),
                    right: Box::new(Expr::Literal {
                        value: value.clone().into_literal()?,
                    }),
                },
                FilterEntry::Numeric(path) => Expr::Binary {
                    op: BinaryOp::Gte,
                    left: Box::new(Expr::Path { path: path.clone() }),
                    right: Box::new(Expr::Literal {
                        value: Literal::Float(
                            FiniteF64::new(f64::MIN)
                                .expect("f64::MIN is finite and is the numeric domain's floor"),
                        ),
                    }),
                },
            };
            stages.push(Stage::Where { predicate });
        }

        // The bucket group is first when present, then one group per path
        // with a positional `_dim_{index}` alias.
        let mut group_exprs: Vec<NamedExpr> = Vec::with_capacity(groups.len() + 1);
        if let Some(seconds) = bucket_seconds {
            let micros = i64::try_from(seconds)
                .ok()
                .and_then(|s| s.checked_mul(1_000_000))
                .ok_or_else(|| format!("bucket width {seconds}s overflows microseconds"))?;
            group_exprs.push(NamedExpr {
                expr: Expr::Call {
                    function: Function::TimeBucket,
                    args: vec![
                        Expr::Path {
                            path: FieldPath::bare("timestamp"),
                        },
                        Expr::Literal {
                            value: Literal::Duration(micros),
                        },
                    ],
                },
                alias: "bucket".to_owned(),
            });
        }
        for (index, path) in groups.iter().enumerate() {
            group_exprs.push(NamedExpr {
                expr: Expr::Path { path: path.clone() },
                alias: format!("_dim_{index}"),
            });
        }

        stages.push(Stage::Aggregate {
            aggregates: vec![eyes_query::AggregateExpr {
                function,
                argument: argument.map(|path| Expr::Path { path }),
                alias: "value".to_owned(),
            }],
            groups: group_exprs,
        });

        let document = QueryDocument {
            version: QUERY_SCHEMA_VERSION,
            query: QueryDefinition {
                source: QuerySource::Telemetry,
                stages,
            },
        };

        // Round-trip through the canonical constructor so alias,
        // reserved-name, duplicate-alias and structural violations fail
        // here, at boot, not at the server. Load-bearing: the path parser
        // only lexes/parses; alias validation runs inside `from_json_str`.
        let serialized =
            serde_json::to_string(&document).map_err(|e| format!("query is invalid: {e}"))?;
        let query = QueryDocument::from_json_str(&serialized).map_err(|diagnostics| {
            format!(
                "metric {id} query is invalid: {}",
                diagnostics
                    .iter()
                    .map(|d| format!("{}: {}", d.code, d.message))
                    .collect::<Vec<_>>()
                    .join("; ")
            )
        })?;

        let result_shape = if bucket_seconds.is_some() {
            MetricResultShape::TimeSeries
        } else if groups.is_empty() {
            MetricResultShape::Scalar
        } else {
            MetricResultShape::Table
        };

        let preferred_bucket_seconds = match (preferred_bucket_seconds, bucket_seconds) {
            (Some(explicit), _) => Some(explicit),
            (None, bucket) => bucket,
        };

        Ok(NamedMetric {
            id,
            display_name,
            description,
            unit,
            value_type,
            result_shape,
            preferred_bucket_seconds,
            query,
            original_dsl,
        })
    }
}

/// The statically derived value-type policy: the same rules the server's
/// `validate_singular_metric` enforces.
fn derive_value_type(
    function: AggregateFunction,
    argument: Option<&FieldPath>,
) -> Result<MetricValueType, String> {
    match function {
        AggregateFunction::Count => Ok(MetricValueType::Int),
        _ => {
            let Some(path) = argument else {
                return Err(format!("{} requires an argument", function.as_str()));
            };
            if path.root == "duration" && path.segments.is_empty() {
                Ok(MetricValueType::Duration)
            } else if path.root == "fields"
                || (path.root == "measurement"
                    && matches!(&path.segments[..], [eyes_query::PathSegment::Key(key)] if key == "value"))
            {
                Ok(MetricValueType::Float)
            } else {
                Err(format!(
                    "aggregate argument must be duration, a fields.* path, or measurement.value; got {}",
                    argument_name(Some(path))
                ))
            }
        }
    }
}

/// Whether a declared value type satisfies the static policy — the derived
/// type, or the Int-or-Float allowance for `sum`/`min`/`max` over dynamic
/// numeric domains.
fn value_type_allowed(
    function: AggregateFunction,
    argument: Option<&FieldPath>,
    declared: MetricValueType,
) -> bool {
    let Ok(derived) = derive_value_type(function, argument) else {
        return false;
    };
    derived == declared
        || matches!(
            (function, argument.is_some(), derived, declared),
            (
                AggregateFunction::Sum | AggregateFunction::Min | AggregateFunction::Max,
                true,
                MetricValueType::Float,
                MetricValueType::Int
            )
        )
}

/// Renders an argument for error messages, or "()" when absent.
fn argument_name(argument: Option<&FieldPath>) -> String {
    match argument {
        None => "()".to_owned(),
        Some(path) => render_path(path),
    }
}

fn render_path(path: &FieldPath) -> String {
    let mut text = path.root.clone();
    for segment in &path.segments {
        match segment {
            eyes_query::PathSegment::Key(key) => {
                text.push('.');
                text.push_str(key);
            }
            eyes_query::PathSegment::Index(index) => {
                text.push_str(&format!("[{index}]"));
            }
        }
    }
    text
}

fn parse_path(text: &str) -> Result<FieldPath, String> {
    eyes_query::dsl::parse_field_path(text).map_err(|diagnostics| {
        format!(
            "invalid field path {text:?}: {}",
            diagnostics
                .iter()
                .map(|d| d.message.clone())
                .collect::<Vec<_>>()
                .join("; ")
        )
    })
}

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

    #[test]
    fn count_with_bucket_builds_a_time_series() {
        let metric = NamedMetricBuilder::new("http.request_count", AggregateFunction::Count, None)
            .unwrap()
            .filter_eq("semantic_kind", "http.request")
            .unwrap()
            .time_bucket(300)
            .unit("requests")
            .build()
            .unwrap();

        assert_eq!(metric.result_shape, MetricResultShape::TimeSeries);
        assert_eq!(metric.value_type, MetricValueType::Int);
        assert_eq!(metric.preferred_bucket_seconds, Some(300));
        assert_eq!(metric.unit.as_deref(), Some("requests"));

        // The query round-trips through the canonical constructor.
        let serialized = serde_json::to_string(&metric.query).unwrap();
        QueryDocument::from_json_str(&serialized).unwrap();

        // Structure: one where, one aggregate; bucket group first, aliased
        // `bucket`; aggregate aliased `value`.
        let stages = &metric.query.query.stages;
        assert_eq!(stages.len(), 2);
        assert!(matches!(stages[0], Stage::Where { .. }));
        let Stage::Aggregate { aggregates, groups } = &stages[1] else {
            panic!("second stage must be the aggregate");
        };
        assert_eq!(aggregates.len(), 1);
        assert_eq!(aggregates[0].alias, "value");
        assert_eq!(groups.len(), 1);
        assert_eq!(groups[0].alias, "bucket");
        assert!(matches!(
            &groups[0].expr,
            Expr::Call {
                function: Function::TimeBucket,
                ..
            }
        ));
    }

    #[test]
    fn p95_over_duration_is_a_duration_scalar() {
        let metric = NamedMetricBuilder::new(
            "http.request_latency",
            AggregateFunction::P95,
            Some("duration"),
        )
        .unwrap()
        .filter_eq("semantic_kind", "http.request")
        .unwrap()
        .build()
        .unwrap();

        assert_eq!(metric.result_shape, MetricResultShape::Scalar);
        assert_eq!(metric.value_type, MetricValueType::Duration);
        assert_eq!(metric.preferred_bucket_seconds, None);
    }

    #[test]
    fn dynamic_arguments_default_to_float_and_sum_accepts_int() {
        let avg =
            NamedMetricBuilder::new("lat.avg", AggregateFunction::Avg, Some("fields.latency_ms"))
                .unwrap()
                .build()
                .unwrap();
        assert_eq!(avg.value_type, MetricValueType::Float);

        let sum =
            NamedMetricBuilder::new("q.sum", AggregateFunction::Sum, Some("fields.queue_depth"))
                .unwrap()
                .build()
                .unwrap();
        assert_eq!(sum.value_type, MetricValueType::Float);

        let sum_int =
            NamedMetricBuilder::new("q.sum", AggregateFunction::Sum, Some("fields.queue_depth"))
                .unwrap()
                .value_type(MetricValueType::Int)
                .build()
                .unwrap();
        assert_eq!(sum_int.value_type, MetricValueType::Int);

        // Int-or-Float policy does not extend to the percentile/avg family.
        let err =
            NamedMetricBuilder::new("q.p95", AggregateFunction::P95, Some("fields.queue_depth"))
                .unwrap()
                .value_type(MetricValueType::Int)
                .build()
                .unwrap_err();
        assert!(
            err.contains("float"),
            "error should name the required type: {err}"
        );
    }

    #[test]
    fn measurement_value_is_a_dynamic_numeric_domain() {
        let metric = NamedMetricBuilder::new(
            "cpu.usage",
            AggregateFunction::Max,
            Some("measurement.value"),
        )
        .unwrap()
        .build()
        .unwrap();
        assert_eq!(metric.value_type, MetricValueType::Float);
    }

    #[test]
    fn group_by_assigns_positional_aliases() {
        let metric = NamedMetricBuilder::new("by.route", AggregateFunction::Count, None)
            .unwrap()
            .group_by("fields.route")
            .unwrap()
            .build()
            .unwrap();

        assert_eq!(metric.result_shape, MetricResultShape::Table);
        let Stage::Aggregate { groups, .. } = metric
            .query
            .query
            .stages
            .iter()
            .find(|s| matches!(s, Stage::Aggregate { .. }))
            .unwrap()
        else {
            unreachable!();
        };
        assert_eq!(groups.len(), 1);
        assert_eq!(groups[0].alias, "_dim_0");
        assert!(matches!(&groups[0].expr, Expr::Path { path } if path.root == "fields"));
    }

    #[test]
    fn bucket_plus_groups_keeps_bucket_first() {
        let metric = NamedMetricBuilder::new("req.by_route", AggregateFunction::Count, None)
            .unwrap()
            .group_by("fields.route")
            .unwrap()
            .time_bucket(60)
            .build()
            .unwrap();
        assert_eq!(metric.result_shape, MetricResultShape::TimeSeries);
        let Stage::Aggregate { groups, .. } = metric
            .query
            .query
            .stages
            .iter()
            .find(|s| matches!(s, Stage::Aggregate { .. }))
            .unwrap()
        else {
            unreachable!();
        };
        let aliases: Vec<&str> = groups.iter().map(|g| g.alias.as_str()).collect();
        assert_eq!(aliases, vec!["bucket", "_dim_0"]);
    }

    #[test]
    fn filter_numeric_bakes_the_gte_stage_into_the_ir() {
        let metric = NamedMetricBuilder::new(
            "q.depth",
            AggregateFunction::Max,
            Some("fields.queue_depth"),
        )
        .unwrap()
        .filter_numeric("fields.queue_depth")
        .unwrap()
        .build()
        .unwrap();

        let serialized = serde_json::to_value(&metric.query).unwrap();
        assert_eq!(
            serialized["query"]["stages"][0],
            json!({
                "stage": "where",
                "predicate": {
                    "type": "binary",
                    "op": "gte",
                    "left": { "type": "path", "path": { "root": "fields", "segments": [{ "type": "key", "value": "queue_depth" }] } },
                    "right": { "type": "literal", "value": { "type": "float", "value": -1.7976931348623157e308 } }
                }
            })
        );
    }

    #[test]
    fn invalid_builds_fail_loudly() {
        // count with an argument
        let err =
            NamedMetricBuilder::new("a", AggregateFunction::Count, Some("duration")).unwrap_err();
        assert_eq!(err, "count takes no argument");

        // p95 without one
        let err = NamedMetricBuilder::new("a", AggregateFunction::P95, None).unwrap_err();
        assert!(err.contains("requires an argument"), "{err}");

        // unparseable path
        let err = NamedMetricBuilder::new("a", AggregateFunction::Count, None)
            .unwrap()
            .filter_eq("not a path!", "x")
            .unwrap_err();
        assert!(err.contains("invalid field path"), "{err}");

        // duplicate group path
        let err = NamedMetricBuilder::new("a", AggregateFunction::Count, None)
            .unwrap()
            .group_by("fields.route")
            .unwrap()
            .group_by("fields.route")
            .unwrap_err();
        assert!(err.contains("duplicate group path"), "{err}");

        // inconsistent explicit value type: p95(duration) must be duration
        let err = NamedMetricBuilder::new("a", AggregateFunction::P95, Some("duration"))
            .unwrap()
            .value_type(MetricValueType::Float)
            .build()
            .unwrap_err();
        assert!(err.contains("duration"), "{err}");

        // count must be int
        let err = NamedMetricBuilder::new("a", AggregateFunction::Count, None)
            .unwrap()
            .value_type(MetricValueType::Float)
            .build()
            .unwrap_err();
        assert!(err.contains("int"), "{err}");

        // a string envelope root is not a measurement
        let err = NamedMetricBuilder::new("a", AggregateFunction::Max, Some("name"))
            .unwrap()
            .build()
            .unwrap_err();
        assert!(err.contains("measurement.value"), "{err}");

        // non-finite float filter value
        let err = NamedMetricBuilder::new("a", AggregateFunction::Count, None)
            .unwrap()
            .filter_eq("fields.x", f64::NAN)
            .unwrap()
            .build()
            .unwrap_err();
        assert!(err.contains("finite"), "{err}");
    }

    #[test]
    fn fixed_aliases_are_not_reserved_roots() {
        // `value` and `bucket` must pass IR validation: a future IR change
        // that makes them reserved should fail *here*, loudly, instead of
        // every built metric 400ing at ingest.
        let metric = NamedMetricBuilder::new("a", AggregateFunction::Count, None)
            .unwrap()
            .time_bucket(60)
            .build()
            .unwrap();
        let serialized = serde_json::to_string(&metric.query).unwrap();
        QueryDocument::from_json_str(&serialized)
            .expect("the fixed aliases value/bucket must remain valid IR");
        assert!(!eyes_query::ENVELOPE_SCALAR_ROOTS.contains(&"value"));
        assert!(!eyes_query::ENVELOPE_SCALAR_ROOTS.contains(&"bucket"));
        assert!(!eyes_query::NAMESPACE_ROOTS.contains(&"value"));
        assert!(!eyes_query::NAMESPACE_ROOTS.contains(&"bucket"));
        // `_dim_*` cannot collide with a reserved root: a leading underscore
        // is a valid identifier start and never a reserved name.
        assert!(!eyes_query::ENVELOPE_SCALAR_ROOTS.contains(&"_dim_0"));
        assert!(!eyes_query::NAMESPACE_ROOTS.contains(&"_dim_0"));
    }

    #[test]
    fn preferred_bucket_defaults_but_an_explicit_value_wins() {
        let defaulted = NamedMetricBuilder::new("a", AggregateFunction::Count, None)
            .unwrap()
            .time_bucket(120)
            .build()
            .unwrap();
        assert_eq!(defaulted.preferred_bucket_seconds, Some(120));

        let explicit = NamedMetricBuilder::new("a", AggregateFunction::Count, None)
            .unwrap()
            .time_bucket(120)
            .preferred_bucket_seconds(300)
            .build()
            .unwrap();
        assert_eq!(explicit.preferred_bucket_seconds, Some(300));

        // A scalar metric has no bucket guidance at all.
        let scalar = NamedMetricBuilder::new("a", AggregateFunction::Count, None)
            .unwrap()
            .preferred_bucket_seconds(300)
            .build()
            .unwrap();
        assert_eq!(scalar.preferred_bucket_seconds, Some(300));
    }

    #[test]
    fn threshold_builder_serializes_exact_wire_json_with_defaults() {
        let threshold = MetricThresholdBuilder::new("err_rate_high", "http.request_count")
            .window_seconds(300)
            .evaluation_interval_seconds(60)
            .warning(ThresholdComparison::Above, 50.0)
            .critical(ThresholdComparison::Above, 100.0)
            .build()
            .unwrap();
        assert_eq!(
            serde_json::to_value(&threshold).unwrap(),
            json!({
                "id": "err_rate_high",
                "metric_id": "http.request_count",
                "description": null,
                "window_seconds": 300,
                "evaluation_interval_seconds": 60,
                "evaluation_delay_seconds": 0,
                "group_match": "any",
                "failure_threshold": 3,
                "recovery_threshold": 1,
                "enabled": true,
                "warning": { "comparison": "above", "threshold": 50.0 },
                "critical": { "comparison": "above", "threshold": 100.0 },
            })
        );
    }

    #[test]
    fn threshold_builder_validates_every_failure_case() {
        let base = || {
            MetricThresholdBuilder::new("t", "m")
                .window_seconds(300)
                .evaluation_interval_seconds(60)
                .critical(ThresholdComparison::Above, 1.0)
        };
        // Missing required fields.
        assert!(MetricThresholdBuilder::new("t", "m")
            .evaluation_interval_seconds(60)
            .critical(ThresholdComparison::Above, 1.0)
            .build()
            .is_err());
        assert!(MetricThresholdBuilder::new("t", "m")
            .window_seconds(300)
            .critical(ThresholdComparison::Above, 1.0)
            .build()
            .is_err());
        // Bounds.
        assert!(base().window_seconds(0).build().is_err());
        assert!(base().window_seconds(2_678_401).build().is_err());
        assert!(base().evaluation_interval_seconds(9).build().is_err());
        assert!(base().evaluation_interval_seconds(86_401).build().is_err());
        assert!(base().evaluation_delay_seconds(601).build().is_err());
        assert!(base().failure_threshold(0).build().is_err());
        assert!(base().failure_threshold(101).build().is_err());
        assert!(base().recovery_threshold(0).build().is_err());
        assert!(base().recovery_threshold(101).build().is_err());
        // No severity at all.
        assert!(MetricThresholdBuilder::new("t", "m")
            .window_seconds(300)
            .evaluation_interval_seconds(60)
            .build()
            .is_err());
        // Non-finite thresholds.
        assert!(base()
            .critical(ThresholdComparison::Above, f64::NAN)
            .build()
            .is_err());
        assert!(base()
            .critical(ThresholdComparison::Above, f64::INFINITY)
            .build()
            .is_err());
        // Happy path sanity, including the delay/group/enabled setters.
        let built = base()
            .evaluation_delay_seconds(30)
            .group_match(ThresholdGroupMatch::All)
            .enabled(false)
            .warning(ThresholdComparison::Below, 0.5)
            .description("p95 latency too high")
            .build()
            .unwrap();
        assert_eq!(built.evaluation_delay_seconds, 30);
        assert_eq!(built.group_match, ThresholdGroupMatch::All);
        assert!(!built.enabled);
        assert_eq!(
            built.warning.as_ref().unwrap().comparison,
            ThresholdComparison::Below
        );
    }
}