diff --git a/Cargo.lock b/Cargo.lock
index c008c0f5..02a0ffdc 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -307,6 +307,7 @@ dependencies = [
 name = "asap-aware-mapping"
 version = "0.1.0"
 dependencies = [
+ "asap-frontend-promql",
  "asap-types",
  "serde",
  "serde_json",
@@ -349,6 +350,7 @@ dependencies = [
 name = "asap-frontend-sql"
 version = "0.1.0"
 dependencies = [
+ "asap-aware-mapping",
  "asap-sql-function-catalog",
  "asap-types",
  "datafusion",
diff --git a/crates/asap-aware-mapping/Cargo.toml b/crates/asap-aware-mapping/Cargo.toml
index 352ed1cf..7f158edc 100644
--- a/crates/asap-aware-mapping/Cargo.toml
+++ b/crates/asap-aware-mapping/Cargo.toml
@@ -12,3 +12,6 @@ asap-types = { path = "../types" }
 thiserror = "2"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
+
+[dev-dependencies]
+asap-frontend-promql = { path = "../frontend-promql" }
diff --git a/crates/asap-aware-mapping/src/accuracy.rs b/crates/asap-aware-mapping/src/accuracy.rs
index 45a03e44..a132eb39 100644
--- a/crates/asap-aware-mapping/src/accuracy.rs
+++ b/crates/asap-aware-mapping/src/accuracy.rs
@@ -716,6 +716,43 @@ impl AccuracyModel for DefaultAccuracyModel {
         let same_metric = |metric: ErrorMetric| approximate.iter().all(|g| g.metric == metric);
 
         match op {
+            CompositionOperator::CheckedRelativeDivision => {
+                if inputs.len() != 2 || local.is_some() || !same_metric(ErrorMetric::RelativeValue)
+                {
+                    return Err(unsupported(
+                        "checked division requires two exact/relative-value operands".into(),
+                    ));
+                }
+                let a = inputs[0]
+                    .bound
+                    .evaluate()
+                    .ok_or_else(|| unsupported("unknown numerator bound".into()))?;
+                let b = inputs[1]
+                    .bound
+                    .evaluate()
+                    .ok_or_else(|| unsupported("unknown denominator bound".into()))?;
+                if !(0.0..1.0).contains(&b) || a < 0.0 || !a.is_finite() {
+                    return Err(unsupported("invalid relative division bounds".into()));
+                }
+                Ok(ResultGuarantee {
+                    metric: ErrorMetric::RelativeValue,
+                    bound: BoundExpr::Constant {
+                        value: (a + b) / (1.0 - b) + 4.0 * f64::EPSILON,
+                    },
+                    failure_probability: ProbabilityExpr::UnionBound {
+                        terms: inputs
+                            .iter()
+                            .map(|g| g.failure_probability.clone())
+                            .collect(),
+                    },
+                    provenance: composed_provenance(
+                        op,
+                        inputs,
+                        &ResultGuarantee::exact("checked floating-point division"),
+                        "checked_relative_division_union_bound",
+                    ),
+                })
+            }
             CompositionOperator::ApproximateAggregate => {
                 let local = local.ok_or_else(|| {
                     unsupported("approximate operator has no local guarantee to compose".into())
@@ -954,6 +991,25 @@ mod tests {
     use asap_types::post_asap::{GroupingStrategy, SketchKind};
     use asap_types::workload::{DataDistribution, DataWorkload, Evidence, EvidenceSource};
 
+    // Rank error cannot certify a numeric ratio; a same-sketch identity is not cancellation evidence.
+    #[test]
+    fn checked_division_propagates_value_bounds_and_rejects_rank_bounds() {
+        let op = CompositionOperator::CheckedRelativeDivision;
+        let inputs = [rel(0.01), rel(0.01)];
+        let g = DefaultAccuracyModel
+            .propagate(&op, &inputs, None, &Default::default())
+            .unwrap();
+        assert!((g.bound.evaluate().unwrap() - 0.02 / 0.99).abs() < 1e-14);
+        let mut rank = inputs[0].clone();
+        rank.metric = ErrorMetric::Rank;
+        assert!(DefaultAccuracyModel
+            .propagate(&op, &[rank.clone(), rank], None, &Default::default())
+            .is_err());
+        assert!(DefaultAccuracyModel
+            .propagate(&op, &[rel(0.01), rel(1.0)], None, &Default::default())
+            .is_err());
+    }
+
     fn abs(bound: f64, delta: f64) -> ResultGuarantee {
         ResultGuarantee {
             metric: ErrorMetric::AbsoluteValue,
diff --git a/crates/asap-aware-mapping/src/function_rules.rs b/crates/asap-aware-mapping/src/function_rules.rs
index 82ad4e20..e5c5f7ba 100644
--- a/crates/asap-aware-mapping/src/function_rules.rs
+++ b/crates/asap-aware-mapping/src/function_rules.rs
@@ -15,7 +15,11 @@ pub(crate) fn function_rules(intent: &AggIntent) -> Option<FunctionRules> {
             CompositionOperator::ExactSum,
             Some((ExactKind::Sum, ExactParams::Sum)),
         ),
-        AggIntent::Min { .. } | AggIntent::Max { .. } => (
+        AggIntent::Min { .. } => (
+            CompositionOperator::ExactExtremum,
+            Some((ExactKind::Min, ExactParams::Min)),
+        ),
+        AggIntent::Max { .. } => (
             CompositionOperator::ExactExtremum,
             Some((ExactKind::MinMax, ExactParams::MinMax)),
         ),
@@ -39,3 +43,20 @@ pub(crate) fn function_rules(intent: &AggIntent) -> Option<FunctionRules> {
         accumulator,
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    // The maintained extrema state must encode the direction independently of query text.
+    #[test]
+    fn minimum_and_maximum_have_distinct_accumulator_contracts() {
+        assert_ne!(
+            function_rules(&AggIntent::Min { col: None })
+                .unwrap()
+                .accumulator,
+            function_rules(&AggIntent::Max { col: None })
+                .unwrap()
+                .accumulator
+        );
+    }
+}
diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs
index da96958a..b3262c28 100644
--- a/crates/asap-aware-mapping/src/lib.rs
+++ b/crates/asap-aware-mapping/src/lib.rs
@@ -254,3 +254,5 @@ pub use summary_maintenance_lifecycle::{
     WorkloadDemand,
 };
 pub use topk_reuse::TopKLimitReuseStrategy;
+
+pub mod maintained_population;
diff --git a/crates/asap-aware-mapping/src/maintained_population.rs b/crates/asap-aware-mapping/src/maintained_population.rs
new file mode 100644
index 00000000..ba9de4f6
--- /dev/null
+++ b/crates/asap-aware-mapping/src/maintained_population.rs
@@ -0,0 +1,459 @@
+//! Shared maintained-population candidates over canonical relational IR.
+use crate::replacement::{
+    Replacement, ReplacementProvenance, ReplacementStrategy, ReplacementSubDAG, TargetSubDAG,
+};
+use asap_types::post_asap::{
+    maintained_population::*, ExecutionTiming, ResultGuarantee, SummaryExpr, SummaryFamilyType,
+    SummaryField, SummaryNode, SummarySchema, ValueOperation,
+};
+use asap_types::pre_asap::{
+    AggIntent, CompareOpKind, DataType, QueryExpr, Reduction, ScalarValue, Schema, Source,
+};
+use std::rc::Rc;
+
+fn plain(schema: Schema) -> SummarySchema {
+    SummarySchema {
+        time_index: schema.time_index,
+        fields: schema
+            .columns
+            .into_iter()
+            .map(|c| SummaryField {
+                name: c.name,
+                dtype: SummaryFamilyType::Plain(c.dtype),
+                nullable: c.nullable,
+            })
+            .collect(),
+    }
+}
+
+fn strip_projection(mut root: &QueryExpr) -> &QueryExpr {
+    while let QueryExpr::Project { child, .. } = root {
+        root = child;
+    }
+    root
+}
+
+fn recognize(root: &QueryExpr) -> Option<(MaintainedPopulation, PopulationReadout, Rc<QueryExpr>)> {
+    let root = strip_projection(root);
+    let (source, grouping, readout, value_column) = match root {
+        QueryExpr::Aggregate {
+            child,
+            reduction: Reduction::Reduce(grouping),
+            measures,
+            having: None,
+            ..
+        } => {
+            let [intent] = measures.as_slice() else {
+                return None;
+            };
+            let (col, readout) = match intent {
+                AggIntent::Quantile { q, col, .. } if q.is_finite() => {
+                    (*col, PopulationReadout::Quantile { q: *q })
+                }
+                AggIntent::Sum { col } => (*col, PopulationReadout::Sum),
+                AggIntent::Count { .. } => (None, PopulationReadout::Count),
+                AggIntent::Avg { col } => (*col, PopulationReadout::Average),
+                _ => return None,
+            };
+            let schema = child.output_schema().ok()?;
+            if col.is_some_and(|c| schema.columns.get(c).is_none()) {
+                return None;
+            }
+            (child, grouping, readout, col)
+        }
+        QueryExpr::Limit {
+            n,
+            offset: 0,
+            child,
+        } => {
+            let QueryExpr::Sort {
+                child,
+                keys,
+                partition_by,
+            } = child.as_ref()
+            else {
+                return None;
+            };
+            let [key] = keys.as_slice() else {
+                return None;
+            };
+            let QueryExpr::Column(col) = &key.expr else {
+                return None;
+            };
+            if key.ascending {
+                return None;
+            }
+            (
+                child,
+                partition_by,
+                PopulationReadout::TopK { k: *n },
+                Some(*col),
+            )
+        }
+        _ => return None,
+    };
+    if let QueryExpr::Scan {
+        source: Source::Table { .. },
+        schema,
+        ..
+    } = source.as_ref()
+    {
+        let value_column = value_column.or_else(|| {
+            schema
+                .columns
+                .iter()
+                .position(|c| c.dtype == DataType::Float64 && !c.nullable)
+        })?;
+        let population = MaintainedPopulation {
+            input: PopulationInput::Rows {
+                input: Rc::clone(source),
+                value_column,
+                grouping: grouping.clone(),
+            },
+            max_k: 0,
+            quantiles: false,
+        };
+        if !schema.closed || !population.matches_input(source) {
+            return None;
+        }
+        return Some((population, readout, Rc::clone(source)));
+    }
+    let QueryExpr::Scan {
+        source: Source::TimeSeries { metric },
+        predicates,
+        schema,
+    } = source.as_ref()
+    else {
+        return None;
+    };
+    if value_column.is_some_and(|c| schema.columns.get(c).is_none_or(|c| c.name != "value")) {
+        return None;
+    }
+    // Open time-series schemas distinguish instant PromQL populations from table rows.
+    if metric.is_empty() || schema.closed || schema.time_index.is_none() {
+        return None;
+    }
+    let label = |col: usize| -> Option<String> {
+        let c = schema.columns.get(col)?;
+        (c.dtype == DataType::Utf8).then(|| c.name.clone())
+    };
+    let mut matchers = Vec::new();
+    for predicate in predicates {
+        let QueryExpr::Compare { left, op, right } = predicate.0.as_ref() else {
+            return None;
+        };
+        let (QueryExpr::Column(col), QueryExpr::Literal(ScalarValue::Utf8(value))) =
+            (left.as_ref(), right.as_ref())
+        else {
+            return None;
+        };
+        let operation = match op {
+            CompareOpKind::Eq => CurrentSeriesMatch::Equal,
+            CompareOpKind::Ne => CurrentSeriesMatch::NotEqual,
+            CompareOpKind::Regex => CurrentSeriesMatch::Regex,
+            CompareOpKind::NotRegex => CurrentSeriesMatch::NotRegex,
+            _ => return None,
+        };
+        matchers.push(CurrentSeriesMatcher {
+            label: label(*col)?,
+            value: value.clone(),
+            operation,
+        });
+    }
+    matchers.sort();
+    matchers.dedup();
+    let mut labels = grouping
+        .keys()
+        .iter()
+        .map(|c| label(*c))
+        .collect::<Option<Vec<_>>>()?;
+    labels.sort();
+    labels.dedup();
+    Some((
+        MaintainedPopulation {
+            input: PopulationInput::CurrentSeries(CurrentSeriesInput {
+                metric: metric.clone(),
+                matchers,
+                grouping: labels,
+                without: grouping.is_without(),
+                lookback_ms: 300_000,
+            }),
+            max_k: 0,
+            quantiles: false,
+        },
+        readout,
+        Rc::clone(source),
+    ))
+}
+
+/// Workload-aware rule: compatible readouts share one retractable population.
+/// Deployments opt in by registering this strategy when they can maintain complete
+/// population updates and price the maintenance/readout boundary.
+/// The population is exact; max_k bounds the shared readout cache, not its members.
+pub struct MaintainedPopulationStrategy {
+    roots: Vec<Rc<QueryExpr>>,
+}
+impl MaintainedPopulationStrategy {
+    pub fn new(roots: &[Rc<QueryExpr>]) -> Self {
+        Self {
+            roots: roots.to_vec(),
+        }
+    }
+    pub fn candidate(&self, root: &Rc<QueryExpr>) -> Option<Rc<SummaryNode>> {
+        if let QueryExpr::Project {
+            cols,
+            qualifier,
+            child,
+        } = root.as_ref()
+        {
+            let child = self.candidate(child)?;
+            return Some(Rc::new(SummaryNode {
+                guarantee: child.guarantee.clone(),
+                schema: plain(root.output_schema().ok()?),
+                expr: SummaryExpr::ValueOperation {
+                    child,
+                    operation: ValueOperation::Project {
+                        cols: cols.clone(),
+                        qualifier: qualifier.clone(),
+                    },
+                    timing: ExecutionTiming::ReadTime,
+                },
+            }));
+        }
+        let (mut population, readout, source) = recognize(root)?;
+        let identity = population.clone();
+        for other in self.roots.iter().chain(std::iter::once(root)) {
+            if let Some((p, r, _)) = recognize(other) {
+                if p == identity {
+                    match r {
+                        PopulationReadout::Quantile { .. } => population.quantiles = true,
+                        PopulationReadout::TopK { k } => population.max_k = population.max_k.max(k),
+                        PopulationReadout::Sum
+                        | PopulationReadout::Count
+                        | PopulationReadout::Average => {}
+                    }
+                }
+            }
+        }
+        let input_schema = plain(source.output_schema().ok()?);
+        let scan = Rc::new(SummaryNode {
+            expr: SummaryExpr::KeepPreAsap(source),
+            schema: input_schema.clone(),
+            guarantee: Some(ResultGuarantee::exact("source samples")),
+        });
+        let maintained = Rc::new(SummaryNode {
+            expr: SummaryExpr::ValueOperation {
+                child: scan,
+                operation: ValueOperation::MaintainPopulation { population },
+                timing: ExecutionTiming::MaintenanceTime,
+            },
+            schema: input_schema,
+            guarantee: Some(ResultGuarantee::exact(
+                "exact members under the declared population semantics",
+            )),
+        });
+        Some(Rc::new(SummaryNode {
+            expr: SummaryExpr::ValueOperation {
+                child: maintained,
+                operation: ValueOperation::ReadPopulation { readout },
+                timing: ExecutionTiming::ReadTime,
+            },
+            schema: plain(root.output_schema().ok()?),
+            guarantee: Some(ResultGuarantee::exact("exact current-population readout")),
+        }))
+    }
+}
+impl ReplacementStrategy for MaintainedPopulationStrategy {
+    fn matches(&self, target: &TargetSubDAG<'_>) -> bool {
+        recognize(target.root).is_some()
+    }
+    fn replacements(&self, target: &TargetSubDAG<'_>) -> Vec<ReplacementSubDAG> {
+        self.candidate(target.root)
+            .map(|node| ReplacementSubDAG {
+                strategy: "MaintainedPopulationStrategy",
+                replacement: Replacement::Summary(node),
+                provenance: ReplacementProvenance::SummaryImplementation,
+                rationale:
+                    "share an exact maintained population across compatible aggregate readouts"
+                        .into(),
+            })
+            .into_iter()
+            .collect()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use asap_types::post_asap::{compile_executable_dag, share_common_summary_subtrees};
+    fn lower(q: &str) -> Rc<QueryExpr> {
+        Rc::new(
+            asap_frontend_promql::lower_promql(q, asap_types::types::AccuracyTarget::Exact)
+                .unwrap(),
+        )
+    }
+
+    // Instant scalar aggregations share the same retractable series population.
+    #[test]
+    fn instant_sum_count_average_are_typed_current_series_candidates() {
+        let roots: Vec<_> = [
+            "sum(a)",
+            "count(a)",
+            "avg(a)",
+            "sum by(job)(a)",
+            "count by(job)(a)",
+            "avg by(job)(a)",
+        ]
+        .map(lower)
+        .into();
+        let rule = MaintainedPopulationStrategy::new(&roots);
+        for root in roots {
+            let candidate = rule
+                .candidate(&root)
+                .expect("current-series rule candidate");
+            compile_executable_dag(&candidate).expect("typed executable DAG");
+        }
+    }
+
+    // Different readout parameters retain one shared maintenance producer in the DAG.
+    #[test]
+    fn quantiles_and_topk_share_a_planner_population() {
+        let roots: Vec<_> = [
+            "quantile by(job)(0.5,a)",
+            "quantile by(job)(0.99,a)",
+            "topk by(job)(1,a)",
+            "topk by(job)(5,a)",
+        ]
+        .map(lower)
+        .into();
+        let strategy = MaintainedPopulationStrategy::new(&roots);
+        let space = crate::search_workload_with(
+            roots
+                .iter()
+                .enumerate()
+                .map(|(i, r)| (i, Rc::clone(r)))
+                .collect(),
+            &[Box::new(MaintainedPopulationStrategy::new(&roots))],
+        );
+        assert!(space
+            .groups()
+            .flat_map(|g| &g.candidates)
+            .any(|c| c.strategy == "MaintainedPopulationStrategy"));
+        let plans = share_common_summary_subtrees(
+            roots
+                .iter()
+                .enumerate()
+                .map(|(i, r)| (i, strategy.candidate(r).unwrap()))
+                .collect(),
+        );
+        let mut producers = Vec::new();
+        for (_, plan) in &plans {
+            compile_executable_dag(plan).unwrap();
+            let SummaryExpr::ValueOperation {
+                child,
+                operation: ValueOperation::ReadPopulation { .. },
+                ..
+            } = &plan.expr
+            else {
+                panic!("missing typed readout")
+            };
+            let SummaryExpr::ValueOperation {
+                operation: ValueOperation::MaintainPopulation { population },
+                ..
+            } = &child.expr
+            else {
+                panic!("missing maintained population")
+            };
+            assert_eq!(population.max_k, 5);
+            assert!(population.quantiles);
+            producers.push(Rc::as_ptr(child));
+        }
+        assert!(producers.iter().all(|p| *p == producers[0]));
+    }
+
+    // Source/group/matcher identity separates populations; temporal/nested operations are not instant populations.
+    #[test]
+    fn rule_respects_population_semantics() {
+        let roots: Vec<_> = [
+            "topk(5,a)",
+            "topk(10,b)",
+            "quantile by(job)(0.5,a)",
+            "quantile(0.9,a{job=\"api\"})",
+        ]
+        .map(lower)
+        .into();
+        let strategy = MaintainedPopulationStrategy::new(&roots);
+        let (p, _, _) = recognize(&roots[0]).unwrap();
+        assert!(matches!(p.input, PopulationInput::CurrentSeries(ref s) if s.grouping.is_empty()));
+        let candidate = strategy.candidate(&roots[0]).unwrap();
+        let SummaryExpr::ValueOperation { child, .. } = &candidate.expr else {
+            unreachable!()
+        };
+        let SummaryExpr::ValueOperation {
+            operation: ValueOperation::MaintainPopulation { population },
+            ..
+        } = &child.expr
+        else {
+            unreachable!()
+        };
+        assert_eq!(population.max_k, 5);
+        assert!(!population.quantiles);
+        for q in [
+            "quantile_over_time(0.5,a[1m])",
+            "quantile(0.5,sum by(job)(a))",
+            "topk(5,a offset 1m)",
+            "topk(5,a @ 100)",
+            "bottomk(5,a)",
+        ] {
+            assert!(
+                strategy.candidate(&lower(q)).is_none(),
+                "unexpected current population for {q}"
+            );
+        }
+        let q = lower("quantile without(instance)(0.5,a{job=~\"api.*\"})");
+        let (p, _, _) = recognize(&q).unwrap();
+        let PopulationInput::CurrentSeries(p) = p.input else {
+            panic!("series input")
+        };
+        assert!(p.without);
+        assert_eq!(p.grouping, ["instance"]);
+        assert_eq!(p.matchers[0].operation, CurrentSeriesMatch::Regex);
+    }
+    // A readout cannot reinterpret arbitrary rows as maintained state or exceed its producer's contract.
+    #[test]
+    fn malformed_population_dags_fail_closed() {
+        let root = lower("topk(5,a)");
+        let strategy = MaintainedPopulationStrategy::new(std::slice::from_ref(&root));
+        let candidate = strategy.candidate(&root).unwrap();
+        let mut bad = (*candidate).clone();
+        let SummaryExpr::ValueOperation { operation, .. } = &mut bad.expr else {
+            unreachable!()
+        };
+        *operation = ValueOperation::ReadPopulation {
+            readout: PopulationReadout::TopK { k: 6 },
+        };
+        assert!(compile_executable_dag(&Rc::new(bad.clone())).is_err());
+        let SummaryExpr::ValueOperation {
+            child, operation, ..
+        } = &mut bad.expr
+        else {
+            unreachable!()
+        };
+        *operation = ValueOperation::ReadPopulation {
+            readout: PopulationReadout::TopK { k: 5 },
+        };
+        let producer = Rc::make_mut(child);
+        let SummaryExpr::ValueOperation {
+            operation: ValueOperation::MaintainPopulation { population },
+            ..
+        } = &mut producer.expr
+        else {
+            unreachable!()
+        };
+        let PopulationInput::CurrentSeries(spec) = &mut population.input else {
+            unreachable!()
+        };
+        spec.metric = "b".into();
+        assert!(compile_executable_dag(&Rc::new(bad)).is_err());
+    }
+}
diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs
index fccc188d..deef4095 100644
--- a/crates/asap-aware-mapping/src/replacement.rs
+++ b/crates/asap-aware-mapping/src/replacement.rs
@@ -1366,6 +1366,25 @@ impl<'a> SketchAlgorithmStrategy<'a> {
     /// differs — see [`realize_child_with`]).
     fn propose_with(&self, root: &Rc<QueryExpr>, intent_override: Option<&AggIntent>) -> Proposals {
         let mut proposals = Proposals::default();
+        if let Ok(Some(node)) = exact_topk_over_temporal_values(root, self.models) {
+            proposals.candidates.push(ReplacementSubDAG {
+                replacement: Replacement::Summary(node),
+                strategy: "SketchAlgorithmStrategy",
+                provenance: ReplacementProvenance::SummaryImplementation,
+                rationale: "select exact Top-K from independently maintained temporal values"
+                    .into(),
+            });
+        }
+        if intent_override.is_none() {
+            if let Ok(Some(node)) = realize_temporal_average(root, self.models, None) {
+                proposals.candidates.push(ReplacementSubDAG {
+                    replacement: Replacement::Summary(node),
+                    strategy: "SketchAlgorithmStrategy",
+                    provenance: ReplacementProvenance::SummaryImplementation,
+                    rationale: "read temporal average from sum/count only within the finite arithmetic domain; otherwise execute the original average".into(),
+                });
+            }
+        }
         if intent_override.is_none() && is_supported_exact_binary(root) {
             if let Ok(Some(node)) = realize_binary(root, self.models, None) {
                 proposals.candidates.push(ReplacementSubDAG {
@@ -1663,11 +1682,93 @@ pub(crate) fn realize_child(
 /// re-splitting for its own approximate children) under the allocated
 /// budget. A child whose declared target is `Exact` keeps it: an allocation
 /// never approximates something the caller declared exact.
+fn exact_topk_over_temporal_values(
+    root: &Rc<QueryExpr>,
+    models: Models<'_>,
+) -> Result<Option<Rc<SummaryNode>>, ImplementError> {
+    let QueryExpr::Aggregate {
+        reduction,
+        measures,
+        output_names,
+        having: None,
+        child,
+    } = root.as_ref()
+    else {
+        return Ok(None);
+    };
+    if !matches!(
+        measures.as_slice(),
+        [AggIntent::TopK {
+            accuracy: AccuracyTarget::Exact,
+            ..
+        }]
+    ) {
+        return Ok(None);
+    }
+    let QueryExpr::Aggregate {
+        reduction: Reduction::PerEntity,
+        child: input,
+        ..
+    } = child.as_ref()
+    else {
+        return Ok(None);
+    };
+    if !matches!(input.as_ref(), QueryExpr::TimeRange { .. }) {
+        return Ok(None);
+    }
+    let values = realize_child_with(child, models, Some(&AccuracyTarget::Exact))?;
+    if matches!(values.expr, SummaryExpr::KeepPreAsap(_))
+        || !values
+            .guarantee
+            .as_ref()
+            .is_some_and(ResultGuarantee::is_exact)
+    {
+        return Ok(None);
+    }
+    let values = finalize_exact_accumulator(values, child)?;
+    let node = Rc::new(SummaryNode {
+        guarantee: values.guarantee.clone(),
+        schema: lift(&root.output_schema()?),
+        expr: SummaryExpr::ValueOperation {
+            child: values,
+            operation: ValueOperation::Exact(ExactOperation::Aggregate {
+                reduction: reduction.clone(),
+                measures: measures.clone(),
+                output_names: output_names.clone(),
+                having: None,
+            }),
+            timing: ExecutionTiming::ReadTime,
+        },
+    });
+    validate_execution_data_states_at(&node, ExecutionDataState::READ_ROWS)?;
+    Ok(Some(node))
+}
+
+fn realize_temporal_average(
+    root: &Rc<QueryExpr>,
+    models: Models<'_>,
+    target: Option<&AccuracyTarget>,
+) -> Result<Option<Rc<SummaryNode>>, ImplementError> {
+    let Some(components) = crate::rewrite::temporal_average_components(root) else {
+        return Ok(None);
+    };
+    let mut node = realize_child_with(&components, models, target)?;
+    let SummaryExpr::BinaryOp { operator, .. } = &mut Rc::make_mut(&mut node).expr else {
+        return Ok(None);
+    };
+    operator.checked_finite_division = true;
+    validate_execution_data_states_at(&node, ExecutionDataState::READ_ROWS)?;
+    Ok(Some(node))
+}
+
 pub(crate) fn realize_child_with(
     root: &Rc<QueryExpr>,
     models: Models<'_>,
     end_to_end_target: Option<&AccuracyTarget>,
 ) -> Result<Rc<SummaryNode>, ImplementError> {
+    if let Some(node) = realize_temporal_average(root, models, end_to_end_target)? {
+        return Ok(node);
+    }
     if let Some(composed) = realize_binary(root, models, end_to_end_target)? {
         return Ok(composed);
     }
@@ -1704,6 +1805,119 @@ pub(crate) fn realize_child_with(
     }
 }
 
+// The logical rule sizes DDSketch operands against the final expression budget.
+// It never converts a rank certificate into a value certificate.
+fn relative_division_candidate(
+    root: &Rc<QueryExpr>,
+    models: Models<'_>,
+    target: Option<&AccuracyTarget>,
+) -> Result<Option<Rc<SummaryNode>>, ImplementError> {
+    let QueryExpr::BinaryOp {
+        op: BinaryOpKind::Arithmetic(asap_types::pre_asap::ArithmeticOpKind::Div),
+        lhs,
+        rhs,
+        vector_match: None,
+    } = root.as_ref()
+    else {
+        return Ok(None);
+    };
+    let target = target
+        .or_else(|| bindable_intent(lhs).and_then(accuracy_target))
+        .or_else(|| bindable_intent(rhs).and_then(accuracy_target));
+    let Some(target) = target else {
+        return Ok(None);
+    };
+    let epsilon = match target {
+        AccuracyTarget::Exact => return Ok(None),
+        AccuracyTarget::Epsilon(e) => *e,
+        AccuracyTarget::EpsilonDelta { epsilon, .. } => *epsilon,
+    };
+    if !epsilon.is_finite() || epsilon <= 1e-12 {
+        return Ok(None);
+    }
+    let alpha = (epsilon - 8.0 * f64::EPSILON) / (2.0 + epsilon);
+    let local_target = AccuracyTarget::Epsilon(alpha);
+    let operand = |expr: &Rc<QueryExpr>,
+                   layer: usize|
+     -> Result<Option<Rc<SummaryNode>>, ImplementError> {
+        if let Some(intent @ AggIntent::Quantile { .. }) = bindable_intent(expr) {
+            if matches!(accuracy_target(intent), Some(AccuracyTarget::Exact)) {
+                return Ok(None);
+            }
+            let intent = override_accuracy(intent, &local_target);
+            for implementation in implementations_for_with(&intent, models.cost) {
+                if !matches!(&implementation, Implementation::Sketch(kind) if kind.algorithm() == &SketchAlgorithm::DDSketch)
+                {
+                    continue;
+                }
+                if let Ok(node) = construct_summary_with(
+                    expr,
+                    &intent,
+                    implementation,
+                    models,
+                    None,
+                    Some(GuaranteeSource::BudgetAllocation {
+                        allocator: "RelativeDivisionAllocator".into(),
+                        layer,
+                        layer_count: 2,
+                        local_target: local_target.clone(),
+                        end_to_end_target: target.clone(),
+                    }),
+                ) {
+                    return Ok(Some(node));
+                }
+            }
+            return Ok(None);
+        }
+        let node = realize_child_with(expr, models, None)?;
+        Ok(node
+            .guarantee
+            .as_ref()
+            .is_some_and(ResultGuarantee::is_exact)
+            .then_some(node))
+    };
+    let (Some(left), Some(right)) = (operand(lhs, 0)?, operand(rhs, 1)?) else {
+        return Ok(None);
+    };
+    let left = finalize_exact_accumulator(left, lhs)?;
+    let right = finalize_exact_accumulator(right, rhs)?;
+    let Some(inputs) = [left.guarantee.clone(), right.guarantee.clone()]
+        .into_iter()
+        .collect::<Option<Vec<_>>>()
+    else {
+        return Ok(None);
+    };
+    if inputs.iter().all(ResultGuarantee::is_exact) {
+        return Ok(None);
+    }
+    let Ok(guarantee) = models.accuracy.propagate(
+        &CompositionOperator::CheckedRelativeDivision,
+        &inputs,
+        None,
+        &Default::default(),
+    ) else {
+        return Ok(None);
+    };
+    if !models.accuracy.satisfies(&guarantee, target) {
+        return Ok(None);
+    }
+    Ok(Some(Rc::new(SummaryNode {
+        expr: SummaryExpr::BinaryOp {
+            timing: ExecutionTiming::ReadTime,
+            lhs: left,
+            rhs: right,
+            operator: asap_types::post_asap::BinaryOperator {
+                checked_relative_division: true,
+                checked_finite_division: false,
+                kind: BinaryOpKind::Arithmetic(asap_types::pre_asap::ArithmeticOpKind::Div),
+                vector_match: None,
+            },
+        },
+        schema: lift(&root.output_schema()?),
+        guarantee: Some(guarantee),
+    })))
+}
+
 /// Preserve an exact arithmetic root while allowing each vector operand to
 /// select its own summary implementation. If either vector arm cannot be
 /// accelerated, return `None` so the caller keeps the whole query exact;
@@ -1726,6 +1940,9 @@ fn realize_binary(
         return Ok(None);
     }
 
+    if let Some(candidate) = relative_division_candidate(root, models, end_to_end_target)? {
+        return Ok(Some(candidate));
+    }
     let lhs_scalar = is_promql_scalar(lhs);
     let rhs_scalar = is_promql_scalar(rhs);
     if lhs_scalar && rhs_scalar {
@@ -1800,6 +2017,8 @@ fn realize_binary(
             lhs: lhs_node,
             rhs: rhs_node,
             operator: asap_types::post_asap::BinaryOperator {
+                checked_relative_division: false,
+                checked_finite_division: false,
                 kind: op.clone(),
                 vector_match: vector_match.clone(),
             },
@@ -5512,6 +5731,81 @@ mod tests {
         }))
     }
 
+    // Finite samples can overflow a sum although their native average is finite.
+    #[test]
+    fn temporal_average_requires_finite_division_guard() {
+        let root = Rc::new(
+            asap_frontend_promql::lower_promql("avg_over_time(a[5m])", AccuracyTarget::Exact)
+                .unwrap(),
+        );
+        let candidates =
+            SketchAlgorithmStrategy::default_cost_model().replacements(&TargetSubDAG::new(&root));
+        let operator = candidates
+            .iter()
+            .find_map(|c| match &c.replacement {
+                Replacement::Summary(node) => match &node.expr {
+                    SummaryExpr::BinaryOp { operator, .. } => Some(operator),
+                    _ => None,
+                },
+                _ => None,
+            })
+            .expect("maintained average candidate");
+        assert!(operator.checked_finite_division);
+        assert!(
+            crate::rewrite::SemanticEquivalentRewriteStrategy
+                .replacements(&TargetSubDAG::new(&root))
+                .is_empty(),
+            "an unconditional pre-ASAP rewrite would bypass the runtime guard"
+        );
+    }
+
+    // A ratio needs a value-error certificate for the expression, not two rank bounds.
+    // Exact Top-K consumes the Planner's maintained temporal values.
+    #[test]
+    fn exact_temporal_topk_has_a_maintained_value_candidate() {
+        for query in [
+            "topk(5, sum_over_time(a[5m]))",
+            "topk by(job)(5, count_over_time(a[5m]))",
+        ] {
+            let root =
+                Rc::new(asap_frontend_promql::lower_promql(query, AccuracyTarget::Exact).unwrap());
+            let models = Models::with_default_accuracy(&crate::cost_model::DefaultCostModel);
+            let node = exact_topk_over_temporal_values(&root, models)
+                .unwrap()
+                .expect("exact Top-K candidate");
+            assert!(node.guarantee.as_ref().unwrap().is_exact());
+            assert!(matches!(
+                node.expr,
+                SummaryExpr::ValueOperation {
+                    operation: ValueOperation::Exact(ExactOperation::Aggregate { .. }),
+                    ..
+                }
+            ));
+            asap_types::post_asap::compile_executable_dag(&node).unwrap();
+        }
+    }
+
+    #[test]
+    fn quantile_ratio_has_a_sized_relative_value_candidate() {
+        let target = AccuracyTarget::EpsilonDelta {
+            epsilon: 0.01,
+            delta: 0.01,
+        };
+        for query in [
+            "quantile_over_time(0.5,a[5m]) / quantile_over_time(0.9,a[5m])",
+            "avg_over_time(a[5m]) / quantile_over_time(0.5,a[5m])",
+        ] {
+            let root = Rc::new(asap_frontend_promql::lower_promql(query, target.clone()).unwrap());
+            let models = Models::with_default_accuracy(&crate::cost_model::DefaultCostModel);
+            let node = realize_binary(&root, models, Some(&target))
+                .unwrap()
+                .expect("ratio candidate");
+            let guarantee = node.guarantee.as_ref().expect("ratio guarantee");
+            assert_eq!(guarantee.metric, ErrorMetric::RelativeValue);
+            assert!(DefaultAccuracyModel.satisfies(guarantee, &target));
+        }
+    }
+
     #[test]
     fn relational_join_predicate_requires_and_normalizes_cross_input_columns() {
         let forward = normalize_cross_input_equi_predicate(&equi_pred(1, 3), 2, 4)
@@ -5626,7 +5920,7 @@ mod tests {
             ),
             // exact mergeable accumulators
             (A::Sum { col: None }, Acc(E::Sum)),
-            (A::Min { col: None }, Acc(E::MinMax)),
+            (A::Min { col: None }, Acc(E::Min)),
             (A::Max { col: None }, Acc(E::MinMax)),
             (A::Rate, Acc(E::Rate)),
             (A::IRate, Acc(E::IRate)),
diff --git a/crates/asap-aware-mapping/src/rewrite.rs b/crates/asap-aware-mapping/src/rewrite.rs
index 459f2e8a..1c481d86 100644
--- a/crates/asap-aware-mapping/src/rewrite.rs
+++ b/crates/asap-aware-mapping/src/rewrite.rs
@@ -26,20 +26,12 @@
 //! that reshaping — see "Non-goals" below for why it does not also decide
 //! whether the reshaping is worth it.
 //!
-//! ## Scope: `by(...)` grouping only (issue #253's own scope note)
+//! ## Scope
 //!
-//! [`AvgToSumOverCountStrategy::matches`] additionally requires
-//! `Reduction::Reduce(by)` with `by` an ordinary (non-`without`) grouping —
-//! narrower than [`SketchAlgorithmStrategy`]'s `bindable_intent`, which is
-//! `Reduction`-agnostic. Two concrete reasons, not stylistic ones:
+//! Ordinary `by(...)` averages use a schema-preserving projection. Temporal
+//! Float64 temporal averages require a typed finite-division guard; their
+//! sum/count components are never exported as an unconditional logical rewrite.
 //!
-//! - **`Reduction::PerEntity`** (`rate`/`increase`/`*_over_time`) is
-//!   single-measure by construction —
-//!   [`aggregate_output_schema`](asap_types::pre_asap::query_expr::aggregate_output_schema)
-//!   `debug_assert!`s exactly one measure for it. This rewrite's entire
-//!   point is introducing a *second* measure (`Count` alongside `Sum`)
-//!   under the same node, which would violate that invariant outright, not
-//!   just drift a schema detail.
 //! - **`without(...)` grouping** leaves an `Aggregate`'s own output schema
 //!   *open* (`closed: false`, see `without_output_schema`), while the
 //!   `Project` this strategy always wraps the rewrite in forces
@@ -135,6 +127,52 @@ fn avg_rewrite_target(node: &QueryExpr) -> Option<(usize, Option<ColumnId>)> {
 /// types a `Div` of two `Int64` operands as `Int64` — the explicit operand
 /// `Cast` is what keeps both the division and rewritten `avg` column
 /// `Float64` the way the original always was, not an incidental extra step).
+// These are conditional physical components, never an unconditional Rewrite.
+// The caller must attach the finite-division execution guard before admission.
+pub(crate) fn temporal_average_components(root: &Rc<QueryExpr>) -> Option<Rc<QueryExpr>> {
+    let QueryExpr::Aggregate {
+        reduction: Reduction::PerEntity,
+        measures,
+        child,
+        having: None,
+        ..
+    } = root.as_ref()
+    else {
+        return None;
+    };
+    let [AggIntent::Avg { col }] = measures.as_slice() else {
+        return None;
+    };
+    if !matches!(child.as_ref(), QueryExpr::TimeRange { .. }) {
+        return None;
+    }
+    let schema = child.output_schema().ok()?;
+    let value = schema
+        .columns
+        .get(col.or_else(|| schema.column_id("value"))?)?;
+    if value.nullable || value.dtype != DataType::Float64 {
+        return None;
+    }
+    let aggregate = |intent| {
+        Rc::new(QueryExpr::Aggregate {
+            reduction: Reduction::PerEntity,
+            measures: vec![intent],
+            output_names: vec![],
+            having: None,
+            child: Rc::clone(child),
+        })
+    };
+    let rewritten = Rc::new(QueryExpr::BinaryOp {
+        op: BinaryOpKind::Arithmetic(ArithmeticOpKind::Div),
+        lhs: aggregate(AggIntent::Sum { col: *col }),
+        rhs: aggregate(AggIntent::Count {
+            accuracy: AccuracyTarget::Exact,
+        }),
+        vector_match: None,
+    });
+    (root.output_schema().ok()? == rewritten.output_schema().ok()?).then_some(rewritten)
+}
+
 fn build_rewrite(root: &Rc<QueryExpr>) -> Option<Rc<QueryExpr>> {
     let (group_count, col) = avg_rewrite_target(root)?;
     let QueryExpr::Aggregate {
@@ -378,6 +416,28 @@ mod tests {
         }
     }
 
+    // Temporal averages expose two single-measure children without closing labels.
+    #[test]
+    fn temporal_average_components_preserves_schema_and_exposes_sum_count() {
+        let root = Rc::new(
+            asap_frontend_promql::lower_promql(
+                "avg_over_time(a{job=\"api\"}[5m])",
+                AccuracyTarget::Exact,
+            )
+            .unwrap(),
+        );
+        assert!(SemanticEquivalentRewriteStrategy
+            .replacements(&TargetSubDAG::new(&root))
+            .is_empty());
+        let rewritten =
+            temporal_average_components(&root).expect("conditional sum/count components");
+        assert_eq!(
+            root.output_schema().unwrap(),
+            rewritten.output_schema().unwrap()
+        );
+        assert!(matches!(rewritten.as_ref(), QueryExpr::BinaryOp { .. }));
+    }
+
     // ── matches ──────────────────────────────────────────────────────────
 
     #[test]
diff --git a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs
index 00ef21d8..9cfc370d 100644
--- a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs
+++ b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs
@@ -3004,6 +3004,8 @@ mod tests {
                 lhs: Rc::clone(&operand),
                 rhs: operand,
                 operator: asap_types::post_asap::BinaryOperator {
+                    checked_relative_division: false,
+                    checked_finite_division: false,
                     kind: asap_types::pre_asap::BinaryOpKind::Arithmetic(
                         asap_types::pre_asap::ArithmeticOpKind::Add,
                     ),
diff --git a/crates/frontend-promql/src/promql.rs b/crates/frontend-promql/src/promql.rs
index 6310f129..4a5861fa 100644
--- a/crates/frontend-promql/src/promql.rs
+++ b/crates/frontend-promql/src/promql.rs
@@ -45,7 +45,7 @@
 //! | `info(v, [selector])` | `PromqlInfoEnrich{selector}` — label-enrichment join against the info metric(s); join keys resolved during post-ASAP binding (issue #84) |
 //! | `group` / `offset` / `@` / `info` | **rejected** — distinct semantics with no intent-algebra representation yet (`info` label-join → #84) |
 //! | `OUTER by (dims) (…)` | `Aggregate.reduction = Reduce(by = dims)` (generic `topk by`/`bottomk` grouping → `Sort.partition_by`) |
-//! | `count by (d) (…)` | `Aggregate{[Cardinality], …}` |
+//! | `count by (d) (…)` | `Aggregate{[Count], …}` |
 //! | `group(v)` / `count_values("l", v)` | `Aggregate{[Group]}` (constant 1) / `Aggregate{[CountValues{l}]}` (group-by-value + count, new label `l`) — issue #49 |
 //! | `limitk(k, v)` / `limit_ratio(r, v)` | `PromqlSeriesSample{LimitK(k) \| LimitRatio(r)}` — series-sampling selection, whole series kept unchanged (issue #86) |
 //! | `topk(k, count_over_time(…))` / `topk(k, sum_over_time(…))` | `Aggregate{[TopK{k}]}` (heavy-hitter intent) over the explicit inner `Aggregate{[Count/Sum]}` |
@@ -592,8 +592,7 @@ fn build_over_subtree(outer: Outer, keys: Vec<ColumnRef>, child: Unresolved) ->
         Outer::Plain(intent) => outer_aggregate(keys, outer_intent(&intent), child),
         Outer::Count => outer_aggregate(
             keys,
-            AggIntent::Cardinality {
-                col: None,
+            AggIntent::Count {
                 accuracy: current_accuracy(),
             },
             child,
@@ -1448,11 +1447,23 @@ fn build(inner: Inner, keys: Vec<ColumnRef>, outer: Outer) -> Result<Unresolved>
             }
         }),
         Outer::Count => Ok(match &inner.func {
-            None => windowed_aggregate(inner, keys, cardinality()),
+            None => windowed_aggregate(
+                inner,
+                keys,
+                AggIntent::Count {
+                    accuracy: current_accuracy(),
+                },
+            ),
             Some(f) => {
                 let inner_i = inner_intent(f);
                 let inner_agg = windowed_aggregate(inner, vec![], inner_i);
-                outer_aggregate(keys, cardinality(), inner_agg)
+                outer_aggregate(
+                    keys,
+                    AggIntent::Count {
+                        accuracy: current_accuracy(),
+                    },
+                    inner_agg,
+                )
             }
         }),
         Outer::CountValues { label } => Ok(match &inner.func {
@@ -1648,15 +1659,6 @@ fn filtered_source(metric: String, matchers: Vec<Unresolved>, shift: TimeShift)
     }
 }
 
-/// `count(v)` / `count by (…) (v)` — SQL `COUNT(DISTINCT col)`'s PromQL
-/// counterpart, over the (always implicit) sample value.
-fn cardinality() -> AggIntent<ColumnRef> {
-    AggIntent::Cardinality {
-        col: None,
-        accuracy: current_accuracy(),
-    }
-}
-
 fn inner_intent(f: &InnerFunc) -> AggIntent<ColumnRef> {
     match f {
         InnerFunc::FrequencyL2 => AggIntent::FrequencyL2 {
diff --git a/crates/frontend-promql/tests/promql_conformance.rs b/crates/frontend-promql/tests/promql_conformance.rs
index 1d6830f5..d346b077 100644
--- a/crates/frontend-promql/tests/promql_conformance.rs
+++ b/crates/frontend-promql/tests/promql_conformance.rs
@@ -328,10 +328,10 @@ fn sum_by_groups_via_positional_aggregate() {
 }
 
 #[test]
-fn count_is_cardinality() {
+fn count_counts_series() {
     assert!(has(&ok("count(up)"), |i| matches!(
         i,
-        AggIntent::Cardinality { .. }
+        AggIntent::Count { .. }
     )));
 }
 
@@ -718,22 +718,21 @@ fn double_unary_negation_nests() {
 }
 
 #[test]
-fn count_maps_to_cardinality_and_inherits_accuracy() {
-    // SEMANTICS (review #2): PromQL `count by (...)` counts distinct series → the
-    // `Cardinality` intent. The workload's AccuracyTarget threads onto it:
+fn count_maps_to_count_and_inherits_accuracy() {
+    // PromQL counts vector elements, including series with identical values.
+    // The workload accuracy target is preserved on the Count intent:
     // `Exact` stays exact (no silent HLL substitution); an approximate target is
     // carried through for post-ASAP binding to honor. This pins the
-    // intentional count→Cardinality mapping and its accuracy gating.
+    // count mapping and its accuracy gating.
     let exact = lower_promql("count by (job) (up)", AccuracyTarget::Exact).unwrap();
     assert!(
         has(&exact, |i| matches!(
             i,
-            AggIntent::Cardinality {
-                col: None,
+            AggIntent::Count {
                 accuracy: AccuracyTarget::Exact
             }
         )),
-        "count→Cardinality must stay Exact under AccuracyTarget::Exact, got {:?}",
+        "count→Count must stay Exact under AccuracyTarget::Exact, got {:?}",
         intents(&exact)
     );
 
@@ -741,12 +740,11 @@ fn count_maps_to_cardinality_and_inherits_accuracy() {
     assert!(
         has(&approx, |i| matches!(
             i,
-            AggIntent::Cardinality {
-                col: None,
+            AggIntent::Count {
                 accuracy: AccuracyTarget::Epsilon(e)
             } if (*e - 0.01).abs() < 1e-9
         )),
-        "count→Cardinality must carry the approximate target, got {:?}",
+        "count→Count must carry the approximate target, got {:?}",
         intents(&approx)
     );
 }
@@ -2072,7 +2070,7 @@ fn limitk_by_carries_the_grouping_and_composes_in_a_set_op() {
     // the PromqlSeriesSample must be preserved under the set op (it must lower, not reject).
     assert!(has(
         &ok("count(limitk(2, http_requests) and http_requests)"),
-        |i| matches!(i, AggIntent::Cardinality { .. })
+        |i| matches!(i, AggIntent::Count { .. })
     ));
 }
 
diff --git a/crates/frontend-promql/tests/promql_lowering.rs b/crates/frontend-promql/tests/promql_lowering.rs
index 5cfa32cc..128783f9 100644
--- a/crates/frontend-promql/tests/promql_lowering.rs
+++ b/crates/frontend-promql/tests/promql_lowering.rs
@@ -373,10 +373,7 @@ fn count_over_rate_keeps_both_levels() {
     else {
         panic!("expected outer Aggregate{{Cardinality}}, got {qe:?}");
     };
-    assert!(matches!(
-        measures.as_slice(),
-        [AggIntent::Cardinality { .. }]
-    ));
+    assert!(matches!(measures.as_slice(), [AggIntent::Count { .. }]));
     assert!(matches!(
         child.as_ref(),
         QueryExpr::Aggregate { measures, .. } if matches!(measures.as_slice(), [AggIntent::Rate])
@@ -399,7 +396,7 @@ fn count_over_time_is_count_intent() {
 }
 
 #[test]
-fn outer_count_is_cardinality() {
+fn outer_count_counts_vector_elements() {
     // `count by (symbol) (count_over_time(...))`: inner per-series sample count
     // over the window (label-preserving), outer cross-series cardinality grouped
     // on a positional `Aggregate.by`. Leaf = [ts, value, symbol] → symbol = col 2.
@@ -414,10 +411,7 @@ fn outer_count_is_cardinality() {
         panic!("expected outer Aggregate grouped by symbol, got {qe:?}");
     };
     assert_eq!(reduction, &Reduction::by(vec![2]));
-    assert!(matches!(
-        measures.as_slice(),
-        [AggIntent::Cardinality { .. }]
-    ));
+    assert!(matches!(measures.as_slice(), [AggIntent::Count { .. }]));
     // Inner: Aggregate{Count} over TimeRange (per-series count_over_time).
     let QueryExpr::Aggregate {
         measures, child, ..
diff --git a/crates/frontend-sql/Cargo.toml b/crates/frontend-sql/Cargo.toml
index 57062f5d..66179346 100644
--- a/crates/frontend-sql/Cargo.toml
+++ b/crates/frontend-sql/Cargo.toml
@@ -16,6 +16,7 @@ datafusion = "43"
 serde_json = "1"
 
 [dev-dependencies]
+asap-aware-mapping = { path = "../asap-aware-mapping" }
 tokio = { version = "1", features = ["rt", "macros", "rt-multi-thread"] }
 # bgp_jan2024_workload corpus is sourced verbatim as YAML (ASAPQuery PR #561)
 # rather than transcribed into the flat .sql shape the other corpora use.
diff --git a/crates/frontend-sql/tests/maintained_population.rs b/crates/frontend-sql/tests/maintained_population.rs
new file mode 100644
index 00000000..6ad7e987
--- /dev/null
+++ b/crates/frontend-sql/tests/maintained_population.rs
@@ -0,0 +1,182 @@
+//! SQL and PromQL use the same shared-state rule without sharing membership semantics.
+use asap_aware_mapping::maintained_population::MaintainedPopulationStrategy;
+use asap_frontend_sql::{lower_sql, SqlCatalog};
+use asap_types::{
+    post_asap::{
+        compile_executable_dag,
+        maintained_population::{MaintainedPopulation, PopulationInput},
+        share_common_summary_subtrees, SummaryExpr, ValueOperation,
+    },
+    pre_asap::{Column, DataType, QueryExpr, Schema},
+    types::AccuracyTarget,
+};
+use std::rc::Rc;
+
+async fn aggregate(q: &str) -> Rc<QueryExpr> {
+    let catalog = SqlCatalog::new().with_table(
+        "samples",
+        Schema::new(vec![
+            Column::new("latency", DataType::Float64, false),
+            Column::new("job", DataType::Utf8, false),
+        ]),
+    );
+    let root = lower_sql(q, &catalog, AccuracyTarget::Exact).await.unwrap();
+    Rc::new(root)
+}
+
+fn population(
+    mut node: &asap_types::post_asap::SummaryNode,
+) -> (
+    &Rc<asap_types::post_asap::SummaryNode>,
+    &MaintainedPopulation,
+) {
+    while let SummaryExpr::ValueOperation {
+        child,
+        operation: ValueOperation::Project { .. },
+        ..
+    } = &node.expr
+    {
+        node = child;
+    }
+    let SummaryExpr::ValueOperation { child, .. } = &node.expr else {
+        panic!("readout")
+    };
+    let SummaryExpr::ValueOperation {
+        operation: ValueOperation::MaintainPopulation { population },
+        ..
+    } = &child.expr
+    else {
+        panic!("state")
+    };
+    (child, population)
+}
+
+// Quantile parameters are readout identity, while source, value column and grouping are state identity.
+#[tokio::test]
+async fn sql_quantiles_share_rows_without_promql_lookback() {
+    let roots = vec![
+        aggregate("SELECT median(latency) FROM samples").await,
+        aggregate("SELECT approx_percentile_cont(latency, 0.99) FROM samples").await,
+    ];
+    let rule = MaintainedPopulationStrategy::new(&roots);
+    let plans = share_common_summary_subtrees(
+        roots
+            .iter()
+            .enumerate()
+            .map(|(i, r)| (i, rule.candidate(r).expect("table population")))
+            .collect(),
+    );
+    for (_, plan) in &plans {
+        compile_executable_dag(plan).unwrap();
+    }
+    let (a, spec) = population(&plans[0].1);
+    let (b, _) = population(&plans[1].1);
+    assert!(Rc::ptr_eq(a, b));
+    assert!(matches!(
+        spec.input,
+        PopulationInput::Rows {
+            value_column: 0,
+            ..
+        }
+    ));
+}
+
+// Different GROUP BY populations must not be merged just because they read the same table.
+#[tokio::test]
+async fn sql_grouping_separates_populations() {
+    let roots = vec![
+        aggregate("SELECT median(latency) FROM samples").await,
+        aggregate("SELECT job, median(latency) FROM samples GROUP BY job").await,
+    ];
+    let rule = MaintainedPopulationStrategy::new(&roots);
+    let a = rule.candidate(&roots[0]).unwrap();
+    let b = rule.candidate(&roots[1]).unwrap();
+    assert_ne!(population(&a).1.input, population(&b).1.input);
+}
+
+// Input predicates and value expressions remain part of sharing identity.
+#[tokio::test]
+async fn sql_filters_separate_populations() {
+    let roots = vec![
+        aggregate("SELECT median(latency) FROM samples WHERE job = 'api'").await,
+        aggregate("SELECT median(latency) FROM samples WHERE job = 'db'").await,
+    ];
+    let rule = MaintainedPopulationStrategy::new(&roots);
+    let a = rule.candidate(&roots[0]).expect("filtered table input");
+    let b = rule.candidate(&roots[1]).expect("filtered table input");
+    assert_ne!(population(&a).1.input, population(&b).1.input);
+}
+
+// All four scalar readouts can share the same non-null numeric SQL population.
+#[tokio::test]
+async fn sql_scalar_readouts_share_membership() {
+    let mut roots = Vec::new();
+    for function in [
+        "median(latency)",
+        "sum(latency)",
+        "avg(latency)",
+        "count(*)",
+    ] {
+        roots.push(aggregate(&format!("SELECT {function} FROM samples")).await);
+    }
+    let rule = MaintainedPopulationStrategy::new(&roots);
+    let plans = share_common_summary_subtrees(
+        roots
+            .iter()
+            .enumerate()
+            .map(|(i, r)| (i, rule.candidate(r).expect("scalar population")))
+            .collect(),
+    );
+    for (_, plan) in &plans {
+        compile_executable_dag(plan).unwrap();
+        assert!(Rc::ptr_eq(population(&plans[0].1).0, population(plan).0));
+    }
+}
+
+// A readout cannot reinterpret a label column as its numeric population.
+#[tokio::test]
+async fn malformed_table_population_fails_validation() {
+    let root = aggregate("SELECT median(latency) FROM samples").await;
+    let rule = MaintainedPopulationStrategy::new(std::slice::from_ref(&root));
+    let mut candidate = rule.candidate(&root).unwrap();
+    let SummaryExpr::ValueOperation { child, .. } = &mut Rc::make_mut(&mut candidate).expr else {
+        unreachable!()
+    };
+    let SummaryExpr::ValueOperation { child, .. } = &mut Rc::make_mut(child).expr else {
+        unreachable!()
+    };
+    let SummaryExpr::ValueOperation {
+        operation: ValueOperation::MaintainPopulation { population },
+        ..
+    } = &mut Rc::make_mut(child).expr
+    else {
+        unreachable!()
+    };
+    let PopulationInput::Rows { value_column, .. } = &mut population.input else {
+        unreachable!()
+    };
+    *value_column = 1;
+    assert!(compile_executable_dag(&candidate).is_err());
+}
+
+// SQL ORDER BY value DESC LIMIT k uses the same maximum-k state contract.
+#[tokio::test]
+async fn sql_topk_limits_share_maximum_k() {
+    let roots = vec![
+        aggregate("SELECT * FROM samples ORDER BY latency DESC LIMIT 1").await,
+        aggregate("SELECT * FROM samples ORDER BY latency DESC LIMIT 5").await,
+    ];
+    let rule = MaintainedPopulationStrategy::new(&roots);
+    let plans = share_common_summary_subtrees(
+        roots
+            .iter()
+            .enumerate()
+            .map(|(i, r)| (i, rule.candidate(r).expect("SQL topk")))
+            .collect(),
+    );
+    for (_, plan) in &plans {
+        compile_executable_dag(plan).unwrap();
+        assert_eq!(population(plan).1.max_k, 5);
+        assert!(Rc::ptr_eq(population(&plans[0].1).0, population(plan).0));
+    }
+}
diff --git a/crates/integration-tests/tests/aggregate.rs b/crates/integration-tests/tests/aggregate.rs
index 1c9ffb8b..1a1e0ca2 100644
--- a/crates/integration-tests/tests/aggregate.rs
+++ b/crates/integration-tests/tests/aggregate.rs
@@ -65,15 +65,14 @@ fn q06_sum_by_job() {
     );
 }
 
-// #7 — PromQL `count` is cross-series cardinality, not per-sample Count
+// #7 — PromQL `count` counts current vector rows, including repeated sample values.
 #[test]
-fn q07_count_is_cardinality() {
+fn q07_count_counts_series_rows() {
     assert_eq!(
         lower("count(http_requests_total)"),
         agg(
             vec![],
-            AggIntent::Cardinality {
-                col: None,
+            AggIntent::Count {
                 accuracy: AccuracyTarget::Exact
             },
             scan("http_requests_total", &[]),
diff --git a/crates/integration-tests/tests/exact_composition.rs b/crates/integration-tests/tests/exact_composition.rs
index fe32e568..86b5ad1f 100644
--- a/crates/integration-tests/tests/exact_composition.rs
+++ b/crates/integration-tests/tests/exact_composition.rs
@@ -253,7 +253,7 @@ fn every_exact_accumulator_is_finalized_before_an_outer_sketch() {
                 AggIntent::Min { col: None },
                 Rc::new(metric_scan(&["zone"])),
             ),
-            ExactKind::MinMax,
+            ExactKind::Min,
         ),
         (
             agg(
diff --git a/crates/integration-tests/tests/promql_to_post_asap.rs b/crates/integration-tests/tests/promql_to_post_asap.rs
index 0668a4d3..3945a2b6 100644
--- a/crates/integration-tests/tests/promql_to_post_asap.rs
+++ b/crates/integration-tests/tests/promql_to_post_asap.rs
@@ -414,7 +414,7 @@ fn promql_binary_arithmetic_preserves_both_scalar_operand_orders() {
 
 #[test]
 fn promql_binary_arithmetic_falls_back_as_a_whole_for_unsupported_arm() {
-    let root = lower_and_realize("rate(a[1m]) + avg_over_time(b[1m])");
+    let root = lower_and_realize("rate(a[1m]) + stddev_over_time(b[1m])");
     assert!(matches!(root.expr, SummaryExpr::KeepPreAsap(_)));
 }
 
diff --git a/crates/types/src/post_asap/cse.rs b/crates/types/src/post_asap/cse.rs
index 92e29913..a8077f74 100644
--- a/crates/types/src/post_asap/cse.rs
+++ b/crates/types/src/post_asap/cse.rs
@@ -417,6 +417,8 @@ mod tests {
                             lhs: Rc::clone(&current),
                             rhs: current,
                             operator: super::super::BinaryOperator {
+                                checked_relative_division: false,
+                                checked_finite_division: false,
                                 kind: crate::pre_asap::BinaryOpKind::Arithmetic(
                                     crate::pre_asap::ArithmeticOpKind::Add,
                                 ),
diff --git a/crates/types/src/post_asap/execution_data_state.rs b/crates/types/src/post_asap/execution_data_state.rs
index fba174b0..2dcd15b3 100644
--- a/crates/types/src/post_asap/execution_data_state.rs
+++ b/crates/types/src/post_asap/execution_data_state.rs
@@ -153,6 +153,8 @@ impl ExecutionDataStateEdge {
 /// it expects, and so tests can assert the *reason* a plan was rejected.
 #[derive(Debug, Clone, PartialEq, Eq, Error)]
 pub enum ExecutionDataStateError {
+    #[error("invalid maintained-population maintenance/readout contract")]
+    InvalidMaintainedPopulation,
     /// A query-time value (`SummaryEstimate` / read-time `ValueOperation` output)
     /// placed beneath a maintained summary — the one shape issue #171's
     /// data_state split exists to make unrepresentable.
@@ -196,6 +198,8 @@ pub enum ExecutionDataStateError {
     MaintenanceRowsAtRoot,
     #[error("unsupported maintenance binary schema or operator")]
     InvalidMaintenanceBinary,
+    #[error("checked division requires one valid guard on a read-time division operator")]
+    InvalidCheckedDivision,
     /// An `ExactOperation` whose input columns are not all `Plain` at its
     /// declared data_state.
     #[error("exact operator consumes non-plain column {column:?} ({dtype})")]
@@ -330,6 +334,18 @@ fn visit(
             timing,
             operator,
         } => {
+            if (operator.checked_relative_division && operator.checked_finite_division)
+                || (operator.checked_relative_division || operator.checked_finite_division)
+                    && (*timing != ExecutionTiming::ReadTime
+                        || !matches!(
+                            operator.kind,
+                            crate::pre_asap::BinaryOpKind::Arithmetic(
+                                crate::pre_asap::ArithmeticOpKind::Div
+                            )
+                        ))
+            {
+                return Err(ExecutionDataStateError::InvalidCheckedDivision);
+            }
             if *timing == ExecutionTiming::MaintenanceTime {
                 use crate::pre_asap::{BinaryOpKind, DataType};
                 if operator.vector_match.is_some()
@@ -462,6 +478,20 @@ fn visit(
             operation,
             timing,
         } => {
+            let valid_population = match operation {
+                ValueOperation::MaintainPopulation { population } => {
+                    *timing == ExecutionTiming::MaintenanceTime
+                        && matches!(&child.expr, SummaryExpr::KeepPreAsap(input) if population.matches_input(input))
+                }
+                ValueOperation::ReadPopulation { readout } => {
+                    *timing == ExecutionTiming::ReadTime
+                        && matches!(&child.expr, SummaryExpr::ValueOperation { operation: ValueOperation::MaintainPopulation { population }, timing: ExecutionTiming::MaintenanceTime, .. } if population.supports(readout))
+                }
+                _ => true,
+            };
+            if !valid_population {
+                return Err(ExecutionDataStateError::InvalidMaintainedPopulation);
+            }
             let required = match timing {
                 ExecutionTiming::MaintenanceTime => ExecutionDataState::MAINTENANCE_ROWS,
                 ExecutionTiming::ReadTime => ExecutionDataState::READ_ROWS,
@@ -471,7 +501,17 @@ fn visit(
                 || matches!(operation, ValueOperation::FinalizeExactAccumulator))
                 && s == ExecutionDataState::MAINTENANCE_SUMMARY
                 && is_exact_accumulator_state(&child.schema).is_ok();
-            if s != required && !exact_readout {
+            let population_readout = matches!(operation, ValueOperation::ReadPopulation { .. })
+                && *timing == ExecutionTiming::ReadTime
+                && matches!(
+                    &child.expr,
+                    SummaryExpr::ValueOperation {
+                        operation: ValueOperation::MaintainPopulation { .. },
+                        timing: ExecutionTiming::MaintenanceTime,
+                        ..
+                    }
+                );
+            if s != required && !exact_readout && !population_readout {
                 return Err(ExecutionDataStateError::IllegalChildDataState {
                     edge: ExecutionDataStateEdge::ValueOperationChild.describe(),
                     child: s,
diff --git a/crates/types/src/post_asap/expr.rs b/crates/types/src/post_asap/expr.rs
index f0110717..c5c878f9 100644
--- a/crates/types/src/post_asap/expr.rs
+++ b/crates/types/src/post_asap/expr.rs
@@ -25,6 +25,15 @@ pub enum ExactOperation {
 #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
 #[non_exhaustive]
 pub enum ValueOperation {
+    /// Maintain the full declared population, including membership changes,
+    /// so removing a TopK member can promote another.
+    MaintainPopulation {
+        population: super::maintained_population::MaintainedPopulation,
+    },
+    /// Read an aggregate or TopK prefix from the maintained population.
+    ReadPopulation {
+        readout: super::maintained_population::PopulationReadout,
+    },
     Exact(ExactOperation),
     /// Read an exact accumulator's state as its finalized scalar value.
     ///
@@ -249,6 +258,16 @@ pub enum SummaryExpr {
 /// All semantics owned by a post-ASAP binary operator.
 #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
 pub struct BinaryOperator {
+    /// Execute division only for finite operands, a nonzero divisor, and a
+    /// normal finite result; otherwise use exact execution. Required by the
+    /// relative-value division certificate, including floating-point range.
+    #[serde(default)]
+    pub checked_relative_division: bool,
+    /// Conditional exact rewrites (such as temporal average from sum/count)
+    /// require finite operands and quotient. Zero/subnormal results are valid;
+    /// overflow must fall back to the original query rather than emit infinity.
+    #[serde(default)]
+    pub checked_finite_division: bool,
     pub kind: BinaryOpKind,
     /// `None` is the only currently supported vector/vector matching mode.
     /// The field is retained so execution never has to recover semantics by
diff --git a/crates/types/src/post_asap/guarantee.rs b/crates/types/src/post_asap/guarantee.rs
index 770cd295..46b8a0c1 100644
--- a/crates/types/src/post_asap/guarantee.rs
+++ b/crates/types/src/post_asap/guarantee.rs
@@ -212,6 +212,9 @@ impl ProbabilityExpr {
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[serde(tag = "op", rename_all = "snake_case")]
 pub enum CompositionOperator {
+    /// Relative division with runtime finite/nonzero/range checks. For operand
+    /// bounds a,b the ratio bound is (a+b)/(1-b), with b < 1.
+    CheckedRelativeDivision,
     /// An approximate summary built over its inputs' (approximate) values
     /// — the sketch-over-sketch case. Its own `local` guarantee composes
     /// with the inputs' under a same-metric rule.
diff --git a/crates/types/src/post_asap/maintained_population.rs b/crates/types/src/post_asap/maintained_population.rs
new file mode 100644
index 00000000..fc953d53
--- /dev/null
+++ b/crates/types/src/post_asap/maintained_population.rs
@@ -0,0 +1,145 @@
+//! Language-independent maintained populations and their readouts.
+//! Resource limits, ingestion placement and data structures belong to the executor.
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CurrentSeriesInput {
+    pub metric: String,
+    pub matchers: Vec<CurrentSeriesMatcher>,
+    pub grouping: Vec<String>,
+    pub without: bool,
+    pub lookback_ms: u64,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
+pub struct CurrentSeriesMatcher {
+    pub label: String,
+    pub value: String,
+    pub operation: CurrentSeriesMatch,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
+pub enum CurrentSeriesMatch {
+    Equal,
+    NotEqual,
+    Regex,
+    NotRegex,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub enum PopulationReadout {
+    Quantile { q: f64 },
+    TopK { k: usize },
+    Sum,
+    Count,
+    Average,
+}
+
+impl CurrentSeriesInput {
+    /// Verify the named contract against the canonical maintenance input.
+    pub fn matches_input(&self, input: &crate::pre_asap::QueryExpr) -> bool {
+        use crate::pre_asap::{CompareOpKind, DataType, QueryExpr, ScalarValue, Source};
+        let QueryExpr::Scan {
+            source: Source::TimeSeries { metric },
+            predicates,
+            schema,
+        } = input
+        else {
+            return false;
+        };
+        if self.metric.is_empty()
+            || *metric != self.metric
+            || schema.closed
+            || schema.time_index.is_none()
+            || self.lookback_ms != 300_000
+        {
+            return false;
+        }
+        if self.grouping.iter().any(|label| {
+            !schema
+                .columns
+                .iter()
+                .any(|c| c.name == *label && c.dtype == DataType::Utf8)
+        }) {
+            return false;
+        }
+        let mut matchers = Vec::new();
+        for predicate in predicates {
+            let QueryExpr::Compare { left, op, right } = predicate.0.as_ref() else {
+                return false;
+            };
+            let (QueryExpr::Column(col), QueryExpr::Literal(ScalarValue::Utf8(value))) =
+                (left.as_ref(), right.as_ref())
+            else {
+                return false;
+            };
+            let Some(column) = schema.columns.get(*col) else {
+                return false;
+            };
+            if column.dtype != DataType::Utf8 {
+                return false;
+            }
+            let operation = match op {
+                CompareOpKind::Eq => CurrentSeriesMatch::Equal,
+                CompareOpKind::Ne => CurrentSeriesMatch::NotEqual,
+                CompareOpKind::Regex => CurrentSeriesMatch::Regex,
+                CompareOpKind::NotRegex => CurrentSeriesMatch::NotRegex,
+                _ => return false,
+            };
+            matchers.push(CurrentSeriesMatcher {
+                label: column.name.clone(),
+                value: value.clone(),
+                operation,
+            });
+        }
+        matchers.sort();
+        matchers.dedup();
+        self.matchers == matchers && self.grouping.windows(2).all(|w| w[0] < w[1])
+    }
+}
+
+/// Membership is part of state identity. Table rows must never acquire implicit
+/// latest-per-series selection, stale markers, or a PromQL lookback.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub enum PopulationInput {
+    CurrentSeries(CurrentSeriesInput),
+    Rows {
+        input: std::rc::Rc<crate::pre_asap::QueryExpr>,
+        value_column: usize,
+        grouping: crate::pre_asap::GroupKeys,
+    },
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct MaintainedPopulation {
+    pub input: PopulationInput,
+    pub max_k: usize,
+    pub quantiles: bool,
+}
+
+impl MaintainedPopulation {
+    pub fn matches_input(&self, input: &crate::pre_asap::QueryExpr) -> bool {
+        match &self.input {
+            PopulationInput::CurrentSeries(spec) => spec.matches_input(input),
+            PopulationInput::Rows {
+                input: expected,
+                value_column,
+                grouping,
+            } => {
+                use crate::pre_asap::{DataType, QueryExpr, Source};
+                expected.as_ref() == input
+                    && matches!(input, QueryExpr::Scan { source: Source::Table { .. }, schema, .. }
+                        if schema.closed && schema.columns.get(*value_column).is_some_and(|c| c.dtype == DataType::Float64 && !c.nullable)
+                            && !grouping.is_without() && grouping.keys().iter().all(|k| *k < schema.columns.len()))
+            }
+        }
+    }
+
+    pub fn supports(&self, readout: &PopulationReadout) -> bool {
+        match readout {
+            PopulationReadout::Quantile { q } => self.quantiles && q.is_finite(),
+            PopulationReadout::TopK { k } => *k <= self.max_k,
+            PopulationReadout::Sum | PopulationReadout::Count | PopulationReadout::Average => true,
+        }
+    }
+}
diff --git a/crates/types/src/post_asap/mod.rs b/crates/types/src/post_asap/mod.rs
index 52e9f690..5b7ba43b 100644
--- a/crates/types/src/post_asap/mod.rs
+++ b/crates/types/src/post_asap/mod.rs
@@ -32,6 +32,7 @@ pub mod executable_dag;
 pub mod execution_data_state;
 pub mod expr;
 pub mod guarantee;
+pub mod maintained_population;
 pub mod query_time;
 pub mod schema;
 pub mod sketch;
diff --git a/crates/types/src/post_asap/sketch.rs b/crates/types/src/post_asap/sketch.rs
index b6f43826..6a14e126 100644
--- a/crates/types/src/post_asap/sketch.rs
+++ b/crates/types/src/post_asap/sketch.rs
@@ -15,6 +15,8 @@ pub enum ExactKind {
     Count,
     /// Exact min/max accumulator (mergeable by comparison).
     MinMax,
+    /// Exact minimum, distinct from the legacy maximum accumulator.
+    Min,
     /// Exact increase accumulator (counter-reset-aware delta).
     Increase,
     /// Rate accumulator (increase / time window duration).
@@ -32,6 +34,8 @@ pub enum ExactParams {
     Sum,
     Count,
     MinMax,
+    /// Exact minimum, distinct from the legacy maximum accumulator.
+    Min,
     Increase,
     Rate,
     IRate,
diff --git a/docs/design_docs/asap-aware-mapping/README.md b/docs/design_docs/asap-aware-mapping/README.md
index ab8a1925..534561e7 100644
--- a/docs/design_docs/asap-aware-mapping/README.md
+++ b/docs/design_docs/asap-aware-mapping/README.md
@@ -41,6 +41,7 @@ budgets; deployment belongs to a later stage.
 - **Replacement Sub-DAG**: A candidate post-ASAP sub-DAG to replace a target sub-DAG. For example, a quantile aggregation may have KLL, DDSketch, and exact aggregation as alternatives.
 - **ReplacementStrategy**: A rule to recognize a target Sub-DAG and produces one or more valid replacement Sub-DAGs.
 - **Candidate Plan**: A complete post-ASAP plan formed by choosing compatible ReplacementStrategies across the plan.
+- **Maintained population**: A multiset of qualifying records retained across evaluations and updated as members enter, change, leave or expire; multiple readouts can share this state.
 - **Cost Model**: A model used to compare valid candidate plans according to criteria such as storage, update cost, query latency, and accuracy.
 
 The distinction between **ReplacementStrategy** and **Candidate Plan** is important. A ReplacementStrategy is a local choice at one decision point, while a candidate plan is a complete plan that combines choices across all relevant decision points.
@@ -88,6 +89,8 @@ The design is split into focused documents:
   combines, checks, costs, and ranks alternatives across a workload.
 - [Optimizations](optimizations.md) describes summary selection, parameterization,
   subpopulation and time organization, roll-ups, sharing, semantic rewrites, and hybrid execution.
+- [Shared maintained population rule](maintained-populations.md) defines population membership,
+  SQL/PromQL input contracts, sharing preconditions, the replacement DAG, and deployment obligations.
 - [Summary properties](summary_properties.md) lists the capabilities used to determine whether
   summaries and optimizations can be composed safely.
 - [End-to-end accuracy guarantees](end-to-end-accuracy-guarantees.md) specifies the typed
diff --git a/docs/design_docs/asap-aware-mapping/maintained-populations.md b/docs/design_docs/asap-aware-mapping/maintained-populations.md
new file mode 100644
index 00000000..74469111
--- /dev/null
+++ b/docs/design_docs/asap-aware-mapping/maintained-populations.md
@@ -0,0 +1,165 @@
+# Shared maintained population rule
+
+## Definition and motivation
+
+A **population** is the multiset of records that an aggregation is defined over,
+after applying its source selection, predicates, membership semantics and grouping.
+A **maintained population** is that multiset represented by state which is kept
+across evaluations and updated when members enter, change, leave or expire.
+A **readout** computes a result from the maintained state at an admitted evaluation.
+
+For group `g` and evaluation `t`, write this multiset as `P_g(t)`. The contract is
+that `ReadPopulation(f, t)` returns `f(P_g(t))`; it must not read a partial or stale
+population outside the deployment's admitted coverage/freshness contract. The
+population describes **which records count**. The physical data structure describes
+**how those records are retained and read**.
+
+For example, suppose two live PromQL series currently have values `7` and `7`.
+Their population contains two members: `count(a)` is `2`, not `1`. If the first
+series changes to `9`, the population becomes `{9, 7}`, not `{7, 7, 9}`. Its previous
+value is replaced. This distinction requires an explicit rule and state contract:
+an append-only quantile sketch cannot by itself implement current-series updates.
+
+The current rule retains an exact population. It does not prescribe a particular
+tree, heap or sketch implementation, and it does not imply a deletable DDSketch.
+
+## Membership semantics
+
+| Input contract | Members | Membership changes |
+| --- | --- | --- |
+| `CurrentSeries` | Latest live sample for each matching series at `t`, partitioned by the declared labels | A newer sample replaces that series' member; a stale marker removes it; lookback expiry removes it |
+| `Rows` | Every row of the declared table input, preserving duplicate multiplicity and applying its predicates/grouping | Inserts add members, updates replace affected members, deletes remove the corresponding occurrences; a complete snapshot can atomically replace the multiset |
+
+The canonical PromQL contract uses a five-minute lookback. For Prometheus 3.5,
+the valid sample interval is `(t - 5m, t]`: a sample exactly at the lower boundary
+is expired. This is a membership requirement, not a configurable sketch window.
+SQL table rows do not inherit this lookback, series identity, or stale-marker behavior.
+For example, two historical rows belonging to one device still count as two SQL
+rows unless the SQL plan explicitly selects the latest row per device.
+
+`CurrentSeriesInput` carries the metric, label matchers, grouping and lookback.
+`Rows` carries the canonical input (including predicates/schema), value-column
+index and grouping. Source identity, predicates, membership semantics, value
+column and grouping determine whether consumers refer to the same population.
+
+## Rule: share one population across compatible readouts
+
+**Implementation:** `MaintainedPopulationStrategy`, an opt-in `ReplacementStrategy`
+in [maintained_population.rs](../../../crates/asap-aware-mapping/src/maintained_population.rs).
+
+**Target sub-DAGs:**
+
+- A single-measure `Aggregate(Reduce(grouping), input)` with Quantile, Sum, Count
+  or Average intent and no HAVING clause.
+- A descending single-column `Sort(input)` followed by `Limit(k, offset=0)`.
+- SQL projections above these targets are preserved in the replacement.
+
+The supported input is a canonical direct scan with one of the membership
+contracts above. Current-series scans must have the canonical open time-series
+schema and supported label predicates. Table scans require a closed schema and
+a non-null Float64 value column. Arbitrary relational inputs, nullable value
+columns and multi-measure aggregates need additional rules.
+
+**Replacement sub-DAG:**
+
+```text
+KeepPreAsap(input)
+  -> MaintainPopulation { input, max_k, quantiles }  [maintenance]
+       -> ReadPopulation { Quantile(q1) }           [read]
+       -> ReadPopulation { Quantile(q2) }           [read]
+       -> ReadPopulation { TopK(k1) }               [read]
+       -> ReadPopulation { TopK(k2) }               [read]
+       -> ReadPopulation { Sum | Count | Average }  [read]
+```
+
+The rule examines compatible workload roots, sets `max_k` to the largest requested
+k and enables quantile readout if any consumer needs it. It emits a candidate for
+each root; canonical summary CSE interns their identical maintenance producers.
+The readout rank `q` and requested prefix `k` do not identify different input
+populations. The union of readout requirements does affect the shared producer's
+configuration, retained memory and cost.
+
+**Concrete transformation:**
+
+```promql
+quantile(0.5, a)
+quantile(0.99, a)
+topk(1, a)
+topk(5, a)
+```
+
+These queries can use one `CurrentSeries` producer with `max_k=5` and quantile
+readout enabled. The full population remains available: deleting a TopK member
+must allow a previously lower-ranked member to be promoted. Retaining only the
+largest five values would not preserve that behavior.
+
+The same rule can represent these SQL consumers using a `Rows` producer:
+
+```sql
+SELECT median(latency) FROM samples;
+SELECT approx_percentile_cont(latency, 0.99) FROM samples;
+SELECT * FROM samples ORDER BY latency DESC LIMIT 1;
+SELECT * FROM samples ORDER BY latency DESC LIMIT 5;
+```
+
+By contrast, `a{job="api"}` and `a{job="db"}`, different value columns, and
+`by(job)` versus `by(region)` identify different populations and are not shared
+by this rule. SQL rows and PromQL current-series members never share state merely
+because their source names or numeric values happen to agree.
+
+## Validation, selection and execution responsibilities
+
+Planner validates the declared input, maintenance/read phases and readout
+compatibility. Its intended guarantee is exact membership and exact readout;
+a physical implementation still must preserve the language's numeric and empty-input
+semantics. In particular, SQL global COUNT over an empty population returns a row
+with zero, while PromQL COUNT over an empty vector returns an empty vector.
+
+The rule proposes a candidate; it does not select it unconditionally. A compiler
+must lower the typed DAG only if its executor supports that membership contract.
+Installation requires complete cost evidence for population construction, updates,
+retention, readouts, retirement and any required raw-data work. Shared state is
+not automatically cheaper than independent or native execution.
+
+The executor owns record identity, input completeness, replacement/retraction,
+coverage, freshness, atomic publication and resource limits. Missing coverage,
+unsupported semantics or exhausted resources must not produce a partial result
+advertised as exact. The backend's current-series implementation rejects evaluations
+older than retained state and falls back while coverage is insufficient.
+
+At the PR #404/#700 implementation boundary, current-series populations are deployable;
+SQL `Rows` candidates are representable but require a table-update/deletion executor.
+Existing SQL window-summary compilation is separate. The SQL executor work is being
+implemented separately; this design does not treat it as already shipped.
+
+## Relation to sketch rules and other optimizations
+
+This rule adds an exact maintained-state alternative. It is distinct from choosing
+a sketch family or merging temporal panes, and can coexist with those alternatives
+in the same workload. Temporal sketch rules continue to emit
+`SummaryAgg -> SummaryEstimate` DAGs.
+
+For example, `distinct_over_time(a[5m])`, `l2_over_time(a[5m])` and
+`entropy_over_time(a[5m])` can read one UnivMon frequency summary when input,
+partitioning, window and sketch parameters match. Here L2 is
+`sqrt(sum_v count(v)^2)`, and entropy is computed from the same value frequencies.
+Each readout still needs its own accuracy evidence: sharing an entropy certificate
+does not establish a cardinality or L2 bound. This is the same separation of
+population/state identity from readout identity, implemented by the existing sketch
+rules rather than by converting UnivMon into an exact `MaintainPopulation` node.
+
+## Acceptance evidence
+
+- PromQL quantiles, TopK limits and scalar readouts share only compatible populations.
+- SQL frontend tests cover shared quantiles/scalar readouts/maximum k, separation
+  by grouping and filters, preservation of projections, and invalid value-column rejection.
+- Backend admission rejects a table-row producer when only a current-series executor
+  is available.
+- Process tests compare current-series replacements and expiry with Prometheus 3.5.
+- The UnivMon process test installs one compatible materialization for all three
+  readouts and checks that missing entropy evidence does not disable the L2 path.
+
+These tests establish the covered semantic and sharing behavior, not measured
+end-to-end speedups or universal floating-point equivalence. The review regressions
+for the exact lookback boundary and temporal-average overflow are separate checks;
+passing the ordinary workload examples alone does not establish those edge cases.
diff --git a/docs/design_docs/asap-aware-mapping/optimizations.md b/docs/design_docs/asap-aware-mapping/optimizations.md
index 4491ed8c..51c217ff 100644
--- a/docs/design_docs/asap-aware-mapping/optimizations.md
+++ b/docs/design_docs/asap-aware-mapping/optimizations.md
@@ -14,6 +14,24 @@ ASAP-aware mapping should support several largely orthogonal dimensions of optim
 
 Some of these are described below with examples.
 
+## Shared maintained population rule
+
+A maintained population is the multiset of qualifying input records represented
+by state retained across query evaluations. Membership updates and aggregate
+readouts are separate operations. This lets different quantiles, TopK limits and
+scalar aggregates share one producer when their input semantics agree.
+
+`MaintainedPopulationStrategy` recognizes supported Aggregate or Sort/Limit
+sub-DAGs and emits `MaintainPopulation -> ReadPopulation` candidates. For example,
+`quantile(0.5, a)`, `quantile(0.99, a)`, `topk(1, a)` and `topk(5, a)` can share one
+current-series population and a maximum-k cache of five. SQL table-row consumers
+use the same rule with a different membership contract; they do not become
+latest-series queries.
+
+See [the rule specification](maintained-populations.md) for the definition,
+matching conditions, sharing identity, exactness requirements, SQL examples,
+compiler capability checks and relation to UnivMon/sketch readouts.
+
 ## Using a subpopulation sketch
 
 Queries often compute the same statistic over many subpopulations:
diff --git a/docs/design_docs/asap-aware-mapping/physical-plan-integration.md b/docs/design_docs/asap-aware-mapping/physical-plan-integration.md
index 330c8d4b..827336e8 100644
--- a/docs/design_docs/asap-aware-mapping/physical-plan-integration.md
+++ b/docs/design_docs/asap-aware-mapping/physical-plan-integration.md
@@ -404,3 +404,19 @@ replace those failures with zero cost or structural node counting.
 
 See [Analytical resource cost](analytical-resource-cost.md) for the resource
 formulas, evidence validation, comparison-scope rules, and calibration model.
+
+## Conditional temporal-average lowering
+
+`avg_over_time(a[5m])` can expose independently maintained sum and count
+components, but their division is conditional. Two finite samples of `1e308`
+have a finite average even though their sum overflows. Planner therefore emits
+a read-time `BinaryOperator` with `checked_finite_division=true` and never exports
+this temporal transformation as an unconditional pre-ASAP rewrite.
+
+The backend lowers the guard to `FiniteDiv`: operands and quotient must be finite,
+and the divisor must be nonzero. Failure executes the original average query.
+Zero and subnormal averages remain valid accelerated results. This guard is
+distinct from `checked_relative_division`, whose relative-error certificate also
+requires a normal result; setting both guards or attaching a guard to a non-division
+operator is invalid. Compilers must preserve this typed condition rather than
+recovering average semantics from query text.