diff --git a/nightlight/crates/nightlight-server/src/query.rs b/nightlight/crates/nightlight-server/src/query.rs
index 2d7a353..a7a3b52 100644
--- a/nightlight/crates/nightlight-server/src/query.rs
+++ b/nightlight/crates/nightlight-server/src/query.rs
@@ -10,23 +10,24 @@ use axum::{
     routing::get,
     Router,
 };
-use parking_lot::Mutex;
 use nightlight_types::{Error, Label, Result, Sample, SeriesId, TimeSeries};
+use parking_lot::Mutex;
 use promql_parser::{
-    label::Matchers,
+    label::{MatchOp, Matchers},
     parser::{
-        AggregateExpr, BinaryExpr, Call, Expr, MatrixSelector, VectorSelector,
+        AggregateExpr, BinModifier, BinaryExpr, Call, Expr, LabelModifier, MatrixSelector,
+        NumberLiteral, UnaryExpr, VectorMatchCardinality, VectorSelector,
     },
 };
 use serde::{Deserialize, Serialize};
-use std::collections::{HashMap, VecDeque};
+use std::collections::{BTreeMap, HashMap, VecDeque};
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::Instant;
 use tokio::sync::RwLock;
-use tracing::{debug, error};
 #[cfg(test)]
 use tracing::info;
+use tracing::{debug, error};
 
 const QUERY_DURATION_HISTORY_LIMIT: usize = 512;
 
@@ -65,6 +66,12 @@ pub struct QueryMetricsSnapshot {
     pub query_duration_p99: f64,
 }
 
+#[derive(Debug, Clone)]
+enum EvalValue {
+    Vector(Vec<TimeSeries>),
+    Scalar(f64),
+}
+
 impl QueryService {
     pub fn new() -> Self {
         Self {
@@ -88,7 +95,10 @@ impl QueryService {
     #[cfg(test)]
     pub fn new_with_persistence(data_path: &std::path::Path) -> Result<Self> {
         let storage = QueryableStorage::load_from_file(data_path)?;
-        info!("Loaded {} series from persistent storage", storage.series.len());
+        info!(
+            "Loaded {} series from persistent storage",
+            storage.series.len()
+        );
 
         Ok(Self {
             storage: Arc::new(RwLock::new(storage)),
@@ -101,7 +111,10 @@ impl QueryService {
     pub async fn save_to_disk(&self, data_path: &std::path::Path) -> Result<()> {
         let storage = self.storage.read().await;
         storage.save_to_file(data_path)?;
-        debug!("Saved {} series to persistent storage", storage.series.len());
+        debug!(
+            "Saved {} series to persistent storage",
+            storage.series.len()
+        );
         Ok(())
     }
 
@@ -137,26 +150,12 @@ impl QueryService {
 
         // Execute the expression
         let storage = self.storage.read().await;
-        let result = self.evaluate_expr(&expr, time, time, 0, &storage).await;
+        let result = self.evaluate_value(&expr, time, time, 0, &storage).await;
         let success = result.is_ok();
         self.metrics.finish_query(started, success);
         let result = result?;
 
-        Ok(QueryResult {
-            result_type: "vector".to_string(),
-            result: result
-                .into_iter()
-                .map(|ts| {
-                    let metric = ts
-                        .labels
-                        .into_iter()
-                        .map(|l| (l.name, l.value))
-                        .collect();
-                    let value = ts.samples.last().map(|s| (s.timestamp, s.value));
-                    InstantQueryResult { metric, value }
-                })
-                .collect(),
-        })
+        Ok(instant_query_from_eval_value(result, time))
     }
 
     /// Execute a range query over a time range with step
@@ -204,7 +203,7 @@ impl QueryService {
         let mut current_time = start;
         while current_time <= end {
             let step_result = self
-                .evaluate_expr(&expr, current_time, end, step, &storage)
+                .evaluate_value(&expr, current_time, end, step, &storage)
                 .await;
             let step_result = match step_result {
                 Ok(step_result) => step_result,
@@ -214,33 +213,7 @@ impl QueryService {
                 }
             };
 
-            for ts in step_result {
-                // Create a unique key for this series based on labels
-                let series_key = ts
-                    .labels
-                    .iter()
-                    .map(|l| format!("{}={}", l.name, l.value))
-                    .collect::<Vec<_>>()
-                    .join(",");
-
-                // Find or create result entry for this series
-                let entry = results.entry(series_key).or_insert_with(|| {
-                    let metric = ts
-                        .labels
-                        .iter()
-                        .map(|l| (l.name.clone(), l.value.clone()))
-                        .collect();
-                    RangeResult {
-                        metric,
-                        values: Vec::new(),
-                    }
-                });
-
-                // Append samples to existing series
-                for sample in ts.samples {
-                    entry.values.push((sample.timestamp, sample.value));
-                }
-            }
+            append_range_step_result(&mut results, step_result, current_time);
 
             current_time += step;
         }
@@ -254,28 +227,31 @@ impl QueryService {
     }
 
     /// Evaluate a PromQL expression (recursive with boxing for async)
-    fn evaluate_expr<'a>(
+    fn evaluate_value<'a>(
         &'a self,
         expr: &'a Expr,
         time: i64,
         end_time: i64,
         step: i64,
         storage: &'a QueryableStorage,
-    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<Vec<TimeSeries>>> + Send + 'a>> {
+    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<EvalValue>> + Send + 'a>> {
         Box::pin(async move {
             match expr {
                 Expr::VectorSelector(selector) => {
                     // Vector selector: metric_name{label="value"}
                     self.evaluate_vector_selector(selector, time, storage)
+                        .map(EvalValue::Vector)
                 }
                 Expr::MatrixSelector(selector) => {
                     // Range selector: metric_name[5m]
                     self.evaluate_matrix_selector(selector, time, storage)
+                        .map(EvalValue::Vector)
                 }
                 Expr::Aggregate(agg) => {
                     // Aggregation: sum(metric), avg(metric), etc.
                     self.evaluate_aggregation(agg, time, end_time, step, storage)
                         .await
+                        .map(EvalValue::Vector)
                 }
                 Expr::Call(call) => {
                     // Function call: rate(metric[5m]), etc.
@@ -287,6 +263,15 @@ impl QueryService {
                     self.evaluate_binary(binary, time, end_time, step, storage)
                         .await
                 }
+                Expr::NumberLiteral(NumberLiteral { val }) => Ok(EvalValue::Scalar(*val)),
+                Expr::Paren(paren) => {
+                    self.evaluate_value(&paren.expr, time, end_time, step, storage)
+                        .await
+                }
+                Expr::Unary(unary) => {
+                    self.evaluate_unary(unary, time, end_time, step, storage)
+                        .await
+                }
                 _ => Err(Error::Query(format!(
                     "Unsupported expression type: {:?}",
                     expr
@@ -306,7 +291,7 @@ impl QueryService {
         let matching_series: Vec<TimeSeries> = storage
             .series
             .values()
-            .filter(|ts| self.matches_selectors(ts, &selector.matchers))
+            .filter(|ts| self.matches_selector(ts, selector))
             .cloned()
             .map(|mut ts| {
                 // Filter to get sample closest to query time
@@ -338,7 +323,7 @@ impl QueryService {
         let mut series = storage
             .series
             .values()
-            .filter(|ts| self.matches_selectors(ts, &selector.vs.matchers))
+            .filter(|ts| self.matches_selector(ts, &selector.vs))
             .cloned()
             .collect::<Vec<_>>();
 
@@ -363,62 +348,47 @@ impl QueryService {
         step: i64,
         storage: &QueryableStorage,
     ) -> Result<Vec<TimeSeries>> {
-        // Evaluate the input expression
-        let input_series = self
-            .evaluate_expr(&agg.expr, time, end_time, step, storage)
-            .await?;
+        let input_series = expect_vector(
+            self.evaluate_value(&agg.expr, time, end_time, step, storage)
+                .await?,
+            "aggregation input must be a vector",
+        )?;
 
         if input_series.is_empty() {
             return Ok(vec![]);
         }
 
-        // Perform aggregation
-        let aggregated_value = match agg.op.to_string().to_lowercase().as_str() {
-            "sum" => input_series
-                .iter()
-                .flat_map(|ts| ts.samples.iter().map(|s| s.value))
-                .sum(),
-            "avg" => {
-                let sum: f64 = input_series
-                    .iter()
-                    .flat_map(|ts| ts.samples.iter().map(|s| s.value))
-                    .sum();
-                let count = input_series
-                    .iter()
-                    .map(|ts| ts.samples.len())
-                    .sum::<usize>() as f64;
-                if count > 0.0 {
-                    sum / count
-                } else {
-                    0.0
-                }
-            }
-            "min" => input_series
-                .iter()
-                .flat_map(|ts| ts.samples.iter().map(|s| s.value))
-                .fold(f64::INFINITY, f64::min),
-            "max" => input_series
-                .iter()
-                .flat_map(|ts| ts.samples.iter().map(|s| s.value))
-                .fold(f64::NEG_INFINITY, f64::max),
-            "count" => input_series.len() as f64,
-            op => {
-                return Err(Error::Query(format!("Unsupported aggregation: {}", op)));
-            }
-        };
+        let mut groups: BTreeMap<String, (Vec<Label>, Vec<f64>)> = BTreeMap::new();
+        for series in input_series {
+            let Some(sample) = series.samples.last() else {
+                continue;
+            };
 
-        // Create result time series
-        Ok(vec![TimeSeries {
-            id: SeriesId(0), // Aggregated result has no series ID
-            labels: vec![Label {
-                name: "__name__".to_string(),
-                value: format!("{}()", agg.op),
-            }],
-            samples: vec![Sample {
-                timestamp: time,
-                value: aggregated_value,
-            }],
-        }])
+            let labels = aggregation_output_labels(&series.labels, agg.modifier.as_ref());
+            let key = labels_key(&labels);
+            groups
+                .entry(key)
+                .or_insert_with(|| (labels, Vec::new()))
+                .1
+                .push(sample.value);
+        }
+
+        let op = agg.op.to_string().to_lowercase();
+        let mut results = Vec::new();
+        for (labels, values) in groups.into_values() {
+            if values.is_empty() {
+                continue;
+            }
+
+            let aggregated_value = aggregate_values(&op, &values)?;
+            results.push(TimeSeries {
+                id: SeriesId(results.len() as u64),
+                labels,
+                samples: vec![Sample::new(time, aggregated_value)],
+            });
+        }
+
+        Ok(results)
     }
 
     /// Evaluate function call (rate, irate, increase, histogram_quantile)
@@ -429,23 +399,26 @@ impl QueryService {
         end_time: i64,
         step: i64,
         storage: &QueryableStorage,
-    ) -> Result<Vec<TimeSeries>> {
+    ) -> Result<EvalValue> {
         let func_name = &call.func.name;
 
         match func_name.to_string().as_str() {
-            "rate" => self.evaluate_rate(call, time, end_time, step, storage).await,
-            "irate" => {
-                self.evaluate_irate(call, time, end_time, step, storage)
-                    .await
-            }
-            "increase" => {
-                self.evaluate_increase(call, time, end_time, step, storage)
-                    .await
-            }
-            "histogram_quantile" => {
-                self.evaluate_histogram_quantile(call, time, end_time, step, storage)
-                    .await
-            }
+            "rate" => self
+                .evaluate_rate(call, time, end_time, step, storage)
+                .await
+                .map(EvalValue::Vector),
+            "irate" => self
+                .evaluate_irate(call, time, end_time, step, storage)
+                .await
+                .map(EvalValue::Vector),
+            "increase" => self
+                .evaluate_increase(call, time, end_time, step, storage)
+                .await
+                .map(EvalValue::Vector),
+            "histogram_quantile" => self
+                .evaluate_histogram_quantile(call, time, end_time, step, storage)
+                .await
+                .map(EvalValue::Vector),
             _ => Err(Error::Query(format!("Unsupported function: {}", func_name))),
         }
     }
@@ -461,14 +434,17 @@ impl QueryService {
     ) -> Result<Vec<TimeSeries>> {
         // rate() expects a range vector (MatrixSelector) as argument
         if call.args.args.is_empty() {
-            return Err(Error::Query("rate() requires a range vector argument".into()));
+            return Err(Error::Query(
+                "rate() requires a range vector argument".into(),
+            ));
         }
 
-        // Get the first argument - should be a MatrixSelector
         let arg = &call.args.args[0];
-
-        // Evaluate the argument to get range data
-        let series_list = Box::pin(self.evaluate_expr(arg, time, end_time, step, storage)).await?;
+        let series_list = expect_vector(
+            self.evaluate_value(arg, time, end_time, step, storage)
+                .await?,
+            "rate() requires a range vector argument",
+        )?;
 
         // Apply rate calculation to each series
         let mut result = Vec::new();
@@ -517,14 +493,17 @@ impl QueryService {
     ) -> Result<Vec<TimeSeries>> {
         // irate() expects a range vector (MatrixSelector) as argument
         if call.args.args.is_empty() {
-            return Err(Error::Query("irate() requires a range vector argument".into()));
+            return Err(Error::Query(
+                "irate() requires a range vector argument".into(),
+            ));
         }
 
-        // Get the first argument - should be a MatrixSelector
         let arg = &call.args.args[0];
-
-        // Evaluate the argument to get range data
-        let series_list = Box::pin(self.evaluate_expr(arg, time, end_time, step, storage)).await?;
+        let series_list = expect_vector(
+            self.evaluate_value(arg, time, end_time, step, storage)
+                .await?,
+            "irate() requires a range vector argument",
+        )?;
 
         // Apply irate calculation to each series
         let mut result = Vec::new();
@@ -572,14 +551,17 @@ impl QueryService {
     ) -> Result<Vec<TimeSeries>> {
         // increase() expects a range vector (MatrixSelector) as argument
         if call.args.args.is_empty() {
-            return Err(Error::Query("increase() requires a range vector argument".into()));
+            return Err(Error::Query(
+                "increase() requires a range vector argument".into(),
+            ));
         }
 
-        // Get the first argument - should be a MatrixSelector
         let arg = &call.args.args[0];
-
-        // Evaluate the argument to get range data
-        let series_list = Box::pin(self.evaluate_expr(arg, time, end_time, step, storage)).await?;
+        let series_list = expect_vector(
+            self.evaluate_value(arg, time, end_time, step, storage)
+                .await?,
+            "increase() requires a range vector argument",
+        )?;
 
         // Apply increase calculation to each series
         let mut result = Vec::new();
@@ -610,61 +592,179 @@ impl QueryService {
         Ok(result)
     }
 
-    /// Evaluate histogram_quantile() function (simplified - stub for now)
+    /// Evaluate histogram_quantile() from Prometheus-style cumulative buckets.
     async fn evaluate_histogram_quantile(
         &self,
-        _call: &Call,
-        _time: i64,
-        _end_time: i64,
-        _step: i64,
-        _storage: &QueryableStorage,
+        call: &Call,
+        time: i64,
+        end_time: i64,
+        step: i64,
+        storage: &QueryableStorage,
     ) -> Result<Vec<TimeSeries>> {
-        // Simplified implementation - full histogram quantile is complex
-        // This would require proper histogram bucket handling
-        Err(Error::Query(
-            "histogram_quantile() not yet implemented".into(),
-        ))
+        if call.args.args.len() != 2 {
+            return Err(Error::Query(
+                "histogram_quantile() requires a scalar quantile and a bucket vector".into(),
+            ));
+        }
+
+        let quantile = expect_scalar(
+            self.evaluate_value(&call.args.args[0], time, end_time, step, storage)
+                .await?,
+            "histogram_quantile() requires a scalar quantile",
+        )?;
+        if !(0.0..=1.0).contains(&quantile) {
+            return Err(Error::Query(format!(
+                "histogram_quantile() quantile must be within [0, 1], got {quantile}"
+            )));
+        }
+
+        let buckets = expect_vector(
+            self.evaluate_value(&call.args.args[1], time, end_time, step, storage)
+                .await?,
+            "histogram_quantile() requires a bucket vector",
+        )?;
+
+        let mut grouped: BTreeMap<String, (Vec<Label>, Vec<(f64, f64)>)> = BTreeMap::new();
+        for series in buckets {
+            let Some(sample) = series.samples.last() else {
+                continue;
+            };
+            let Some(le) = series.get_label("le") else {
+                continue;
+            };
+            let upper_bound = parse_histogram_bound(le)?;
+            let labels = histogram_output_labels(&series.labels);
+            let key = labels_key(&labels);
+            grouped
+                .entry(key)
+                .or_insert_with(|| (labels, Vec::new()))
+                .1
+                .push((upper_bound, sample.value));
+        }
+
+        let mut results = Vec::new();
+        for (labels, mut buckets) in grouped.into_values() {
+            buckets.sort_by(|(lhs, _), (rhs, _)| lhs.total_cmp(rhs));
+            let value = histogram_quantile_value(quantile, &buckets)?;
+            results.push(TimeSeries {
+                id: SeriesId(results.len() as u64),
+                labels,
+                samples: vec![Sample::new(time, value)],
+            });
+        }
+
+        Ok(results)
     }
 
-    /// Evaluate binary operation (simplified - stub for now)
+    /// Evaluate binary operations for scalar/vector arithmetic and one-to-one vector matching.
     async fn evaluate_binary(
         &self,
-        _binary: &BinaryExpr,
-        _time: i64,
-        _end_time: i64,
-        _step: i64,
-        _storage: &QueryableStorage,
-    ) -> Result<Vec<TimeSeries>> {
-        // Simplified binary operation evaluation
-        // Full implementation would require proper vector matching
-        Err(Error::Query(
-            "Binary operations not yet fully implemented".into(),
-        ))
+        binary: &BinaryExpr,
+        time: i64,
+        end_time: i64,
+        step: i64,
+        storage: &QueryableStorage,
+    ) -> Result<EvalValue> {
+        if let Some(modifier) = &binary.modifier {
+            if !matches!(modifier.card, VectorMatchCardinality::OneToOne) {
+                return Err(Error::Query(
+                    "many-to-one and many-to-many vector matching are not supported yet".into(),
+                ));
+            }
+        }
+
+        let lhs = self
+            .evaluate_value(&binary.lhs, time, end_time, step, storage)
+            .await?;
+        let rhs = self
+            .evaluate_value(&binary.rhs, time, end_time, step, storage)
+            .await?;
+
+        match (lhs, rhs) {
+            (EvalValue::Scalar(lhs), EvalValue::Scalar(rhs)) => {
+                evaluate_scalar_binary(binary, lhs, rhs).map(EvalValue::Scalar)
+            }
+            (EvalValue::Vector(lhs), EvalValue::Scalar(rhs)) => {
+                evaluate_vector_scalar_binary(binary, lhs, rhs, true).map(EvalValue::Vector)
+            }
+            (EvalValue::Scalar(lhs), EvalValue::Vector(rhs)) => {
+                evaluate_vector_scalar_binary(binary, rhs, lhs, false).map(EvalValue::Vector)
+            }
+            (EvalValue::Vector(lhs), EvalValue::Vector(rhs)) => {
+                evaluate_vector_vector_binary(binary, lhs, rhs).map(EvalValue::Vector)
+            }
+        }
     }
 
-    /// Check if a time series matches label matchers
-    fn matches_selectors(&self, ts: &TimeSeries, matchers: &Matchers) -> bool {
-        for matcher in &matchers.matchers {
-            let label_name = &matcher.name;
-            let label_value = ts
-                .labels
-                .iter()
-                .find(|l| &l.name == label_name)
-                .map(|l| l.value.as_str());
+    async fn evaluate_unary(
+        &self,
+        unary: &UnaryExpr,
+        time: i64,
+        end_time: i64,
+        step: i64,
+        storage: &QueryableStorage,
+    ) -> Result<EvalValue> {
+        match self
+            .evaluate_value(&unary.expr, time, end_time, step, storage)
+            .await?
+        {
+            EvalValue::Scalar(value) => Ok(EvalValue::Scalar(-value)),
+            EvalValue::Vector(mut series) => {
+                for entry in &mut series {
+                    for sample in &mut entry.samples {
+                        sample.value = -sample.value;
+                    }
+                }
+                Ok(EvalValue::Vector(series))
+            }
+        }
+    }
 
-            // For now, simple exact match (can be enhanced)
-            let matches = if let Some(value) = label_value {
-                matcher.value == value
-            } else {
-                false
-            };
-
-            if !matches {
+    fn matches_selector(&self, ts: &TimeSeries, selector: &VectorSelector) -> bool {
+        if let Some(name) = selector.name.as_deref() {
+            if ts.name() != Some(name) {
                 return false;
             }
         }
 
-        true
+        self.matches_matchers(ts, &selector.matchers)
+    }
+
+    fn matches_matchers(&self, ts: &TimeSeries, matchers: &Matchers) -> bool {
+        if !matchers
+            .matchers
+            .iter()
+            .all(|matcher| self.matcher_matches(ts, matcher))
+        {
+            return false;
+        }
+
+        if matchers.or_matchers.is_empty() {
+            return true;
+        }
+
+        matchers.or_matchers.iter().any(|group| {
+            group
+                .iter()
+                .all(|matcher| self.matcher_matches(ts, matcher))
+        })
+    }
+
+    fn matcher_matches(&self, ts: &TimeSeries, matcher: &promql_parser::label::Matcher) -> bool {
+        let label_value = ts.get_label(&matcher.name);
+
+        match &matcher.op {
+            MatchOp::Equal => label_value == Some(matcher.value.as_str()),
+            MatchOp::NotEqual => match label_value {
+                Some(value) => value != matcher.value,
+                None => true,
+            },
+            MatchOp::Re(regex) => label_value.is_some_and(|value| regex.is_match(value)),
+            MatchOp::NotRe(regex) => match label_value {
+                Some(value) => !regex.is_match(value),
+                None => true,
+            },
+        }
     }
 
     pub async fn series_metadata(
@@ -842,6 +942,385 @@ impl QueryableStorage {
     }
 }
 
+fn expect_vector(value: EvalValue, message: &str) -> Result<Vec<TimeSeries>> {
+    match value {
+        EvalValue::Vector(series) => Ok(series),
+        EvalValue::Scalar(_) => Err(Error::Query(message.to_string())),
+    }
+}
+
+fn expect_scalar(value: EvalValue, message: &str) -> Result<f64> {
+    match value {
+        EvalValue::Scalar(value) => Ok(value),
+        EvalValue::Vector(_) => Err(Error::Query(message.to_string())),
+    }
+}
+
+fn instant_query_from_eval_value(value: EvalValue, time: i64) -> QueryResult {
+    match value {
+        EvalValue::Scalar(value) => QueryResult {
+            result_type: "scalar".to_string(),
+            result: vec![InstantQueryResult {
+                metric: HashMap::new(),
+                value: Some((time, value)),
+            }],
+        },
+        EvalValue::Vector(series) => QueryResult {
+            result_type: "vector".to_string(),
+            result: series
+                .into_iter()
+                .map(|ts| InstantQueryResult {
+                    metric: labels_to_map(&ts.labels),
+                    value: ts
+                        .samples
+                        .last()
+                        .map(|sample| (sample.timestamp, sample.value)),
+                })
+                .collect(),
+        },
+    }
+}
+
+fn append_range_step_result(
+    results: &mut HashMap<String, RangeResult>,
+    step_result: EvalValue,
+    time: i64,
+) {
+    match step_result {
+        EvalValue::Scalar(value) => {
+            let entry = results
+                .entry("__scalar__".to_string())
+                .or_insert_with(|| RangeResult {
+                    metric: HashMap::new(),
+                    values: Vec::new(),
+                });
+            entry.values.push((time, value));
+        }
+        EvalValue::Vector(series) => {
+            for ts in series {
+                let key = labels_key(&ts.labels);
+                let metric = labels_to_map(&ts.labels);
+                let entry = results.entry(key).or_insert_with(|| RangeResult {
+                    metric,
+                    values: Vec::new(),
+                });
+                for sample in ts.samples {
+                    entry.values.push((sample.timestamp, sample.value));
+                }
+            }
+        }
+    }
+}
+
+fn labels_to_map(labels: &[Label]) -> HashMap<String, String> {
+    labels
+        .iter()
+        .map(|label| (label.name.clone(), label.value.clone()))
+        .collect()
+}
+
+fn normalize_labels(mut labels: Vec<Label>) -> Vec<Label> {
+    labels.sort_by(|lhs, rhs| lhs.name.cmp(&rhs.name).then(lhs.value.cmp(&rhs.value)));
+    labels
+}
+
+fn labels_key(labels: &[Label]) -> String {
+    let mut pairs: Vec<(String, String)> = labels
+        .iter()
+        .map(|label| (label.name.clone(), label.value.clone()))
+        .collect();
+    pairs.sort();
+    pairs
+        .into_iter()
+        .map(|(name, value)| format!("{name}={value}"))
+        .collect::<Vec<_>>()
+        .join(",")
+}
+
+fn aggregate_values(op: &str, values: &[f64]) -> Result<f64> {
+    match op {
+        "sum" => Ok(values.iter().sum()),
+        "avg" => Ok(values.iter().sum::<f64>() / values.len() as f64),
+        "min" => Ok(values.iter().copied().fold(f64::INFINITY, f64::min)),
+        "max" => Ok(values.iter().copied().fold(f64::NEG_INFINITY, f64::max)),
+        "count" => Ok(values.len() as f64),
+        other => Err(Error::Query(format!("Unsupported aggregation: {other}"))),
+    }
+}
+
+fn aggregation_output_labels(labels: &[Label], modifier: Option<&LabelModifier>) -> Vec<Label> {
+    match modifier {
+        None => Vec::new(),
+        Some(LabelModifier::Include(group)) => normalize_labels(
+            labels
+                .iter()
+                .filter(|label| group.labels.iter().any(|name| name == &label.name))
+                .cloned()
+                .collect(),
+        ),
+        Some(LabelModifier::Exclude(group)) => normalize_labels(
+            labels
+                .iter()
+                .filter(|label| label.name != "__name__")
+                .filter(|label| !group.labels.iter().any(|name| name == &label.name))
+                .cloned()
+                .collect(),
+        ),
+    }
+}
+
+fn histogram_output_labels(labels: &[Label]) -> Vec<Label> {
+    normalize_labels(
+        labels
+            .iter()
+            .filter(|label| label.name != "__name__" && label.name != "le")
+            .cloned()
+            .collect(),
+    )
+}
+
+fn parse_histogram_bound(value: &str) -> Result<f64> {
+    match value {
+        "+Inf" | "Inf" | "+inf" | "inf" => Ok(f64::INFINITY),
+        _ => value.parse::<f64>().map_err(|error| {
+            Error::Query(format!("invalid histogram bucket bound {value:?}: {error}"))
+        }),
+    }
+}
+
+fn histogram_quantile_value(quantile: f64, buckets: &[(f64, f64)]) -> Result<f64> {
+    let Some((_, total_count)) = buckets.last() else {
+        return Err(Error::Query(
+            "histogram_quantile() requires at least one bucket".into(),
+        ));
+    };
+    if *total_count <= 0.0 {
+        return Ok(0.0);
+    }
+
+    let target = quantile * total_count;
+    let mut lower_bound = 0.0;
+    let mut previous_count = 0.0;
+
+    for (upper_bound, count) in buckets {
+        if *count >= target {
+            if upper_bound.is_infinite() {
+                return Ok(lower_bound);
+            }
+
+            let bucket_count = (*count - previous_count).max(0.0);
+            if bucket_count == 0.0 {
+                return Ok(*upper_bound);
+            }
+
+            let fraction = ((target - previous_count) / bucket_count).clamp(0.0, 1.0);
+            return Ok(lower_bound + (upper_bound - lower_bound) * fraction);
+        }
+
+        lower_bound = *upper_bound;
+        previous_count = *count;
+    }
+
+    Ok(lower_bound)
+}
+
+fn evaluate_scalar_binary(binary: &BinaryExpr, lhs: f64, rhs: f64) -> Result<f64> {
+    let op = binary.op.to_string();
+    if is_set_operator(&op) {
+        return Err(Error::Query(format!("Unsupported set operator: {op}")));
+    }
+
+    if is_comparison_operator(&op) {
+        if !binary.return_bool() {
+            return Err(Error::Query(
+                "scalar comparisons require the bool modifier".into(),
+            ));
+        }
+
+        return Ok(if compare_values(&op, lhs, rhs)? {
+            1.0
+        } else {
+            0.0
+        });
+    }
+
+    arithmetic_value(&op, lhs, rhs)
+}
+
+fn evaluate_vector_scalar_binary(
+    binary: &BinaryExpr,
+    series_list: Vec<TimeSeries>,
+    scalar: f64,
+    vector_on_lhs: bool,
+) -> Result<Vec<TimeSeries>> {
+    let mut results = Vec::new();
+    for mut series in series_list {
+        let labels = binary_result_labels(&series.labels, binary.modifier.as_ref());
+        let mut samples = Vec::new();
+
+        for sample in &mut series.samples {
+            let (lhs, rhs) = if vector_on_lhs {
+                (sample.value, scalar)
+            } else {
+                (scalar, sample.value)
+            };
+            let retained = sample.value;
+            if let Some(value) = apply_binary_sample(binary, lhs, rhs, retained)? {
+                samples.push(Sample::new(sample.timestamp, value));
+            }
+        }
+
+        if !samples.is_empty() {
+            results.push(TimeSeries {
+                id: series.id,
+                labels,
+                samples,
+            });
+        }
+    }
+
+    Ok(results)
+}
+
+fn evaluate_vector_vector_binary(
+    binary: &BinaryExpr,
+    lhs_series: Vec<TimeSeries>,
+    rhs_series: Vec<TimeSeries>,
+) -> Result<Vec<TimeSeries>> {
+    let lhs_index = build_vector_match_index(&lhs_series, binary.modifier.as_ref())?;
+    let rhs_index = build_vector_match_index(&rhs_series, binary.modifier.as_ref())?;
+    let mut results = Vec::new();
+
+    for (key, lhs) in lhs_index {
+        let Some(rhs) = rhs_index.get(&key) else {
+            continue;
+        };
+        let Some(lhs_sample) = lhs.samples.last() else {
+            continue;
+        };
+        let Some(rhs_sample) = rhs.samples.last() else {
+            continue;
+        };
+
+        if let Some(value) =
+            apply_binary_sample(binary, lhs_sample.value, rhs_sample.value, lhs_sample.value)?
+        {
+            results.push(TimeSeries {
+                id: lhs.id,
+                labels: binary_result_labels(&lhs.labels, binary.modifier.as_ref()),
+                samples: vec![Sample::new(
+                    lhs_sample.timestamp.max(rhs_sample.timestamp),
+                    value,
+                )],
+            });
+        }
+    }
+
+    Ok(results)
+}
+
+fn build_vector_match_index<'a>(
+    series_list: &'a [TimeSeries],
+    modifier: Option<&BinModifier>,
+) -> Result<HashMap<String, &'a TimeSeries>> {
+    let mut index = HashMap::new();
+    for series in series_list {
+        let key = vector_match_key(series, modifier);
+        if index.insert(key.clone(), series).is_some() {
+            return Err(Error::Query(format!(
+                "duplicate vector match key {key:?}; many-to-one matching is not supported"
+            )));
+        }
+    }
+    Ok(index)
+}
+
+fn vector_match_key(series: &TimeSeries, modifier: Option<&BinModifier>) -> String {
+    labels_key(&binary_result_labels(&series.labels, modifier))
+}
+
+fn binary_result_labels(labels: &[Label], modifier: Option<&BinModifier>) -> Vec<Label> {
+    let selected = match modifier.and_then(|modifier| modifier.matching.as_ref()) {
+        Some(LabelModifier::Include(group)) => labels
+            .iter()
+            .filter(|label| group.labels.iter().any(|name| name == &label.name))
+            .cloned()
+            .collect(),
+        Some(LabelModifier::Exclude(group)) => labels
+            .iter()
+            .filter(|label| label.name != "__name__")
+            .filter(|label| !group.labels.iter().any(|name| name == &label.name))
+            .cloned()
+            .collect(),
+        None => labels
+            .iter()
+            .filter(|label| label.name != "__name__")
+            .cloned()
+            .collect(),
+    };
+
+    normalize_labels(selected)
+}
+
+fn apply_binary_sample(
+    binary: &BinaryExpr,
+    lhs: f64,
+    rhs: f64,
+    retained: f64,
+) -> Result<Option<f64>> {
+    let op = binary.op.to_string();
+    if is_set_operator(&op) {
+        return Err(Error::Query(format!("Unsupported set operator: {op}")));
+    }
+
+    if is_comparison_operator(&op) {
+        let matched = compare_values(&op, lhs, rhs)?;
+        if binary.return_bool() {
+            return Ok(Some(if matched { 1.0 } else { 0.0 }));
+        }
+        return Ok(matched.then_some(retained));
+    }
+
+    Ok(Some(arithmetic_value(&op, lhs, rhs)?))
+}
+
+fn compare_values(op: &str, lhs: f64, rhs: f64) -> Result<bool> {
+    match op {
+        "==" => Ok(lhs == rhs),
+        "!=" => Ok(lhs != rhs),
+        ">" => Ok(lhs > rhs),
+        ">=" => Ok(lhs >= rhs),
+        "<" => Ok(lhs < rhs),
+        "<=" => Ok(lhs <= rhs),
+        other => Err(Error::Query(format!(
+            "Unsupported comparison operator: {other}"
+        ))),
+    }
+}
+
+fn arithmetic_value(op: &str, lhs: f64, rhs: f64) -> Result<f64> {
+    match op {
+        "+" => Ok(lhs + rhs),
+        "-" => Ok(lhs - rhs),
+        "*" => Ok(lhs * rhs),
+        "/" => Ok(lhs / rhs),
+        "%" => Ok(lhs % rhs),
+        "^" => Ok(lhs.powf(rhs)),
+        "atan2" => Ok(lhs.atan2(rhs)),
+        other => Err(Error::Query(format!(
+            "Unsupported arithmetic operator: {other}"
+        ))),
+    }
+}
+
+fn is_comparison_operator(op: &str) -> bool {
+    matches!(op, "==" | "!=" | ">" | ">=" | "<" | "<=")
+}
+
+fn is_set_operator(op: &str) -> bool {
+    matches!(op, "and" | "or" | "unless")
+}
+
 fn percentile(values: &[u64], quantile: f64) -> f64 {
     if values.is_empty() {
         return 0.0;
@@ -1137,8 +1616,8 @@ impl QueryableStorage {
         }
 
         // Read file
-        let mut file = File::open(path)
-            .map_err(|e| Error::Storage(format!("Failed to open file: {}", e)))?;
+        let mut file =
+            File::open(path).map_err(|e| Error::Storage(format!("Failed to open file: {}", e)))?;
 
         let mut buffer = Vec::new();
         file.read_to_end(&mut buffer)
@@ -1162,6 +1641,27 @@ mod tests {
     };
     use tower::ServiceExt;
 
+    fn test_series(id: u64, labels: &[(&str, &str)], samples: &[(i64, f64)]) -> TimeSeries {
+        TimeSeries {
+            id: SeriesId(id),
+            labels: labels
+                .iter()
+                .map(|(name, value)| Label::new(*name, *value))
+                .collect(),
+            samples: samples
+                .iter()
+                .map(|(timestamp, value)| Sample::new(*timestamp, *value))
+                .collect(),
+        }
+    }
+
+    async fn seed_series(service: &QueryService, series: Vec<TimeSeries>) {
+        let mut storage = service.storage.write().await;
+        for entry in series {
+            storage.upsert_series(entry);
+        }
+    }
+
     #[tokio::test]
     async fn test_query_service_creation() {
         let service = QueryService::new();
@@ -1210,12 +1710,256 @@ mod tests {
     #[tokio::test]
     async fn test_range_query_empty_storage() {
         let service = QueryService::new();
-        let result = service
-            .execute_range_query("up", 1000, 2000, 100)
-            .await;
+        let result = service.execute_range_query("up", 1000, 2000, 100).await;
         assert!(result.is_ok());
     }
 
+    #[tokio::test]
+    async fn test_scalar_instant_query_returns_scalar_result() {
+        let service = QueryService::new();
+        let result = service.execute_instant_query("1", 1000).await.unwrap();
+        assert_eq!(result.result_type, "scalar");
+        assert_eq!(result.result.len(), 1);
+        assert_eq!(result.result[0].value, Some((1000, 1.0)));
+    }
+
+    #[tokio::test]
+    async fn test_label_matchers_support_regex_and_negative_match() {
+        let service = QueryService::new();
+        seed_series(
+            &service,
+            vec![
+                test_series(
+                    1,
+                    &[
+                        ("__name__", "http_requests_total"),
+                        ("method", "GET"),
+                        ("status", "200"),
+                    ],
+                    &[(1000, 10.0)],
+                ),
+                test_series(
+                    2,
+                    &[
+                        ("__name__", "http_requests_total"),
+                        ("method", "POST"),
+                        ("status", "500"),
+                    ],
+                    &[(1000, 20.0)],
+                ),
+            ],
+        )
+        .await;
+
+        let result = service
+            .execute_instant_query("http_requests_total{method=~\"G.*\",status!=\"500\"}", 1000)
+            .await
+            .unwrap();
+
+        assert_eq!(result.result_type, "vector");
+        assert_eq!(result.result.len(), 1);
+        assert_eq!(
+            result.result[0].metric.get("method"),
+            Some(&"GET".to_string())
+        );
+    }
+
+    #[tokio::test]
+    async fn test_aggregation_groups_by_labels() {
+        let service = QueryService::new();
+        seed_series(
+            &service,
+            vec![
+                test_series(
+                    1,
+                    &[("__name__", "cpu_usage"), ("job", "api"), ("instance", "a")],
+                    &[(1000, 2.0)],
+                ),
+                test_series(
+                    2,
+                    &[("__name__", "cpu_usage"), ("job", "api"), ("instance", "b")],
+                    &[(1000, 3.0)],
+                ),
+            ],
+        )
+        .await;
+
+        let result = service
+            .execute_instant_query("sum by (job)(cpu_usage)", 1000)
+            .await
+            .unwrap();
+
+        assert_eq!(result.result.len(), 1);
+        assert_eq!(result.result[0].metric.get("job"), Some(&"api".to_string()));
+        assert_eq!(result.result[0].value, Some((1000, 5.0)));
+    }
+
+    #[tokio::test]
+    async fn test_binary_queries_support_vector_scalar_and_vector_vector() {
+        let service = QueryService::new();
+        seed_series(
+            &service,
+            vec![
+                test_series(
+                    1,
+                    &[("__name__", "requests_total"), ("service", "api")],
+                    &[(1000, 20.0)],
+                ),
+                test_series(
+                    2,
+                    &[("__name__", "errors_total"), ("service", "api")],
+                    &[(1000, 5.0)],
+                ),
+            ],
+        )
+        .await;
+
+        let vector_scalar = service
+            .execute_instant_query("requests_total / 2", 1000)
+            .await
+            .unwrap();
+        assert_eq!(vector_scalar.result.len(), 1);
+        assert_eq!(
+            vector_scalar.result[0].metric.get("service"),
+            Some(&"api".to_string())
+        );
+        assert_eq!(vector_scalar.result[0].value, Some((1000, 10.0)));
+        assert!(!vector_scalar.result[0].metric.contains_key("__name__"));
+
+        let ratio = service
+            .execute_instant_query("errors_total / requests_total", 1000)
+            .await
+            .unwrap();
+        assert_eq!(ratio.result.len(), 1);
+        assert_eq!(ratio.result[0].value, Some((1000, 0.25)));
+        assert_eq!(ratio.result[0].metric.len(), 1);
+        assert_eq!(
+            ratio.result[0].metric.get("service"),
+            Some(&"api".to_string())
+        );
+    }
+
+    #[tokio::test]
+    async fn test_histogram_quantile_with_grouped_buckets() {
+        let service = QueryService::new();
+        seed_series(
+            &service,
+            vec![
+                test_series(
+                    1,
+                    &[
+                        ("__name__", "request_duration_seconds_bucket"),
+                        ("job", "api"),
+                        ("instance", "a"),
+                        ("le", "0.1"),
+                    ],
+                    &[(1000, 5.0)],
+                ),
+                test_series(
+                    2,
+                    &[
+                        ("__name__", "request_duration_seconds_bucket"),
+                        ("job", "api"),
+                        ("instance", "a"),
+                        ("le", "0.2"),
+                    ],
+                    &[(1000, 10.0)],
+                ),
+                test_series(
+                    3,
+                    &[
+                        ("__name__", "request_duration_seconds_bucket"),
+                        ("job", "api"),
+                        ("instance", "a"),
+                        ("le", "+Inf"),
+                    ],
+                    &[(1000, 20.0)],
+                ),
+                test_series(
+                    4,
+                    &[
+                        ("__name__", "request_duration_seconds_bucket"),
+                        ("job", "api"),
+                        ("instance", "b"),
+                        ("le", "0.1"),
+                    ],
+                    &[(1000, 5.0)],
+                ),
+                test_series(
+                    5,
+                    &[
+                        ("__name__", "request_duration_seconds_bucket"),
+                        ("job", "api"),
+                        ("instance", "b"),
+                        ("le", "0.2"),
+                    ],
+                    &[(1000, 10.0)],
+                ),
+                test_series(
+                    6,
+                    &[
+                        ("__name__", "request_duration_seconds_bucket"),
+                        ("job", "api"),
+                        ("instance", "b"),
+                        ("le", "+Inf"),
+                    ],
+                    &[(1000, 20.0)],
+                ),
+            ],
+        )
+        .await;
+
+        let result = service
+            .execute_instant_query(
+                "histogram_quantile(0.5, sum by (job, le)(request_duration_seconds_bucket{job=\"api\"}))",
+                1000,
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(result.result.len(), 1);
+        assert_eq!(result.result[0].metric.get("job"), Some(&"api".to_string()));
+        assert!(!result.result[0].metric.contains_key("le"));
+        assert_eq!(result.result[0].value, Some((1000, 0.2)));
+    }
+
+    #[tokio::test]
+    async fn test_query_range_route_supports_binary_expression() {
+        let service = QueryService::new();
+        seed_series(
+            &service,
+            vec![test_series(
+                1,
+                &[("__name__", "requests_total"), ("service", "api")],
+                &[(1000, 10.0), (2000, 20.0)],
+            )],
+        )
+        .await;
+
+        let app: axum::Router = service.router();
+        let request = Request::builder()
+            .method(Method::GET)
+            .uri("/api/v1/query_range?query=requests_total%20/%202&start=1000&end=2000&step=1000")
+            .body(Body::empty())
+            .unwrap();
+        let response = app.oneshot(request).await.unwrap();
+        assert_eq!(response.status(), StatusCode::OK);
+
+        let body = to_bytes(response.into_body(), 1024 * 1024).await.unwrap();
+        let payload: serde_json::Value = serde_json::from_slice(&body).unwrap();
+        assert_eq!(payload["status"], "success");
+        assert_eq!(payload["data"]["resultType"], "matrix");
+        assert_eq!(payload["data"]["result"][0]["metric"]["service"], "api");
+        assert_eq!(
+            payload["data"]["result"][0]["values"][0],
+            serde_json::json!([1000, 5.0])
+        );
+        assert_eq!(
+            payload["data"]["result"][0]["values"][1],
+            serde_json::json!([2000, 10.0])
+        );
+    }
+
     #[tokio::test]
     async fn test_storage_upsert() {
         let service = QueryService::new();
@@ -1276,10 +2020,7 @@ mod tests {
             .uri("/api/v1/label/job/values")
             .body(Body::empty())
             .unwrap();
-        let response = app
-            .oneshot(request)
-            .await
-            .unwrap();
+        let response = app.oneshot(request).await.unwrap();
         assert_eq!(response.status(), StatusCode::OK);
 
         let body = to_bytes(response.into_body(), 1024 * 1024).await.unwrap();
@@ -1330,19 +2071,13 @@ mod tests {
 
         let series1 = TimeSeries {
             id: SeriesId(1),
-            labels: vec![
-                Label::new("__name__", "metric1"),
-                Label::new("job", "test"),
-            ],
+            labels: vec![Label::new("__name__", "metric1"), Label::new("job", "test")],
             samples: vec![Sample::new(1000, 10.0), Sample::new(2000, 20.0)],
         };
 
         let series2 = TimeSeries {
             id: SeriesId(2),
-            labels: vec![
-                Label::new("__name__", "metric2"),
-                Label::new("job", "prod"),
-            ],
+            labels: vec![Label::new("__name__", "metric2"), Label::new("job", "prod")],
             samples: vec![Sample::new(1000, 30.0)],
         };
 
diff --git a/nix/test-cluster/run-cluster.sh b/nix/test-cluster/run-cluster.sh
index 5e37065..87dc0b1 100755
--- a/nix/test-cluster/run-cluster.sh
+++ b/nix/test-cluster/run-cluster.sh
@@ -4985,10 +4985,7 @@ validate_nightlight_flow_with_base() {
   local metric_value
   metric_value="$(awk 'BEGIN{srand(); printf "%.3f\n", (rand()*100)+1}')"
 
-  python3 "${REPO_ROOT}/nix/test-cluster/nightlight_remote_write.py" \
-    --url "${base_url}/write" \
-    --metric "${metric_name}" \
-    --value "${metric_value}" \
+  nightlight_remote_write_sample "${base_url}" "${metric_name}" "${metric_value}" auto \
     --label source=smoke \
     --label cluster=photoncloud
 
@@ -4998,6 +4995,34 @@ validate_nightlight_flow_with_base() {
     | jq -e --arg name "${metric_name}" '.status == "success" and (.data | index($name)) != null' >/dev/null
   curl -fsS "${base_url}/series" \
     | jq -e --arg name "${metric_name}" '.status == "success" and (.data | any(.__name__ == $name))' >/dev/null
+
+  validate_nightlight_promql_features "${base_url}" "${flow_name}"
+}
+
+nightlight_remote_write_sample() {
+  local base_url="$1"
+  local metric_name="$2"
+  local metric_value="$3"
+  local timestamp_ms="$4"
+  shift 4
+
+  local -a cmd=(
+    python3 "${REPO_ROOT}/nix/test-cluster/nightlight_remote_write.py"
+    --url "${base_url}/write"
+    --metric "${metric_name}"
+    --value "${metric_value}"
+  )
+
+  if [[ "${timestamp_ms}" != "auto" ]]; then
+    cmd+=(--timestamp-ms "${timestamp_ms}")
+  fi
+
+  while (($#)); do
+    cmd+=("$1")
+    shift
+  done
+
+  "${cmd[@]}"
 }
 
 wait_for_nightlight_query_result() {
@@ -5024,6 +5049,136 @@ wait_for_nightlight_query_result() {
   done
 }
 
+validate_nightlight_promql_features() {
+  local base_url="$1"
+  local flow_name="$2"
+  local suffix="promql_$(date +%s)"
+  local requests_metric="nightlight_promql_requests_${suffix}"
+  local errors_metric="nightlight_promql_errors_${suffix}"
+  local cpu_metric="nightlight_promql_cpu_${suffix}"
+  local latency_metric="nightlight_promql_latency_${suffix}"
+  local timestamp_end timestamp_start step_ms
+
+  timestamp_end="$(date +%s%3N)"
+  step_ms=60000
+  timestamp_start=$((timestamp_end - step_ms))
+
+  log "Validating ${flow_name} PromQL arithmetic, grouping, histogram quantile, and range queries"
+
+  nightlight_remote_write_sample "${base_url}" "${requests_metric}" 20 "${timestamp_start}" \
+    --label suite=promql \
+    --label service=api
+  nightlight_remote_write_sample "${base_url}" "${requests_metric}" 40 "${timestamp_end}" \
+    --label suite=promql \
+    --label service=api
+  nightlight_remote_write_sample "${base_url}" "${errors_metric}" 4 "${timestamp_start}" \
+    --label suite=promql \
+    --label service=api
+  nightlight_remote_write_sample "${base_url}" "${errors_metric}" 10 "${timestamp_end}" \
+    --label suite=promql \
+    --label service=api
+
+  nightlight_remote_write_sample "${base_url}" "${cpu_metric}" 2 "${timestamp_end}" \
+    --label suite=promql \
+    --label job=api \
+    --label instance=a
+  nightlight_remote_write_sample "${base_url}" "${cpu_metric}" 3 "${timestamp_end}" \
+    --label suite=promql \
+    --label job=api \
+    --label instance=b
+
+  nightlight_remote_write_sample "${base_url}" "${latency_metric}" 5 "${timestamp_end}" \
+    --label suite=promql \
+    --label job=api \
+    --label instance=a \
+    --label le=0.1
+  nightlight_remote_write_sample "${base_url}" "${latency_metric}" 10 "${timestamp_end}" \
+    --label suite=promql \
+    --label job=api \
+    --label instance=a \
+    --label le=0.2
+  nightlight_remote_write_sample "${base_url}" "${latency_metric}" 20 "${timestamp_end}" \
+    --label suite=promql \
+    --label job=api \
+    --label instance=a \
+    --label le=+Inf
+  nightlight_remote_write_sample "${base_url}" "${latency_metric}" 5 "${timestamp_end}" \
+    --label suite=promql \
+    --label job=api \
+    --label instance=b \
+    --label le=0.1
+  nightlight_remote_write_sample "${base_url}" "${latency_metric}" 10 "${timestamp_end}" \
+    --label suite=promql \
+    --label job=api \
+    --label instance=b \
+    --label le=0.2
+  nightlight_remote_write_sample "${base_url}" "${latency_metric}" 20 "${timestamp_end}" \
+    --label suite=promql \
+    --label job=api \
+    --label instance=b \
+    --label le=+Inf
+
+  wait_for_nightlight_query_result \
+    "${base_url}" \
+    "${flow_name} PromQL latest sample" \
+    "${requests_metric}" \
+    40 \
+    "suite=\"promql\",service=\"api\""
+
+  curl -fsS --get "${base_url}/query" \
+    --data-urlencode "query=${errors_metric}{suite=\"promql\",service=\"api\"} / ${requests_metric}{suite=\"promql\",service=\"api\"}" \
+    | jq -e '
+      .status == "success"
+      and .data.resultType == "vector"
+      and (.data.result | length) == 1
+      and .data.result[0].metric.service == "api"
+      and (.data.result[0].metric | has("__name__") | not)
+      and (.data.result[0].value[1] >= 0.249)
+      and (.data.result[0].value[1] <= 0.251)
+    ' >/dev/null
+
+  curl -fsS --get "${base_url}/query" \
+    --data-urlencode "query=sum by (job)(${cpu_metric}{suite=\"promql\"})" \
+    | jq -e '
+      .status == "success"
+      and .data.resultType == "vector"
+      and (.data.result | length) == 1
+      and .data.result[0].metric.job == "api"
+      and (.data.result[0].value[1] >= 4.999)
+      and (.data.result[0].value[1] <= 5.001)
+    ' >/dev/null
+
+  curl -fsS --get "${base_url}/query" \
+    --data-urlencode "query=histogram_quantile(0.5, sum by (job, le)(${latency_metric}{suite=\"promql\",job=\"api\"}))" \
+    | jq -e '
+      .status == "success"
+      and .data.resultType == "vector"
+      and (.data.result | length) == 1
+      and .data.result[0].metric.job == "api"
+      and (.data.result[0].metric | has("le") | not)
+      and (.data.result[0].value[1] >= 0.199)
+      and (.data.result[0].value[1] <= 0.201)
+    ' >/dev/null
+
+  curl -fsS --get "${base_url}/query_range" \
+    --data-urlencode "query=${requests_metric}{suite=\"promql\",service=\"api\"} / 2" \
+    --data-urlencode "start=${timestamp_start}" \
+    --data-urlencode "end=${timestamp_end}" \
+    --data-urlencode "step=${step_ms}" \
+    | jq -e --argjson start "${timestamp_start}" --argjson end "${timestamp_end}" '
+      .status == "success"
+      and .data.resultType == "matrix"
+      and (.data.result | length) == 1
+      and .data.result[0].metric.service == "api"
+      and .data.result[0].values[0][0] == $start
+      and (.data.result[0].values[0][1] >= 9.999)
+      and (.data.result[0].values[0][1] <= 10.001)
+      and .data.result[0].values[1][0] == $end
+      and (.data.result[0].values[1][1] >= 19.999)
+      and (.data.result[0].values[1][1] <= 20.001)
+    ' >/dev/null
+}
+
 validate_nightlight_flow() {
   validate_nightlight_flow_with_base "http://127.0.0.1:9090/api/v1" "NightLight"
   validate_nightlight_grpc_and_persistence