diff --git a/crates/executor/src/capability.rs b/crates/executor/src/capability.rs
index 9d535dd8..c5d64623 100644
--- a/crates/executor/src/capability.rs
+++ b/crates/executor/src/capability.rs
@@ -146,8 +146,29 @@ fn validate_hydra_cms(
     };
     match (&input.weight, proof) {
         (SummaryInputExpr::Constant(w), NonNegativeWeightProof::UnitCount) if *w == 1.0 => Ok(()),
+        (SummaryInputExpr::Constant(w), NonNegativeWeightProof::PositiveConstant) => {
+            hydra_constant_weight(*w).map(|_| ())
+        }
         (SummaryInputExpr::Column(_), _) => Ok(()),
-        _ => Err("HydraCms weight must be the unit count or a non-negative weight column".into()),
+        _ => Err(
+            "HydraCms weight must be the unit count, a positive constant or a non-negative \
+             weight column"
+                .into(),
+        ),
+    }
+}
+
+/// A HydraCms constant weight as the `i32` sketchlib's Hydra adds: a
+/// positive integer within `i32`; an error otherwise.
+pub(crate) fn hydra_constant_weight(weight: f64) -> Result<i32, String> {
+    if weight.is_finite() && weight > 0.0 && weight.fract() == 0.0 && weight <= f64::from(i32::MAX)
+    {
+        Ok(weight as i32)
+    } else {
+        Err(format!(
+            "HydraCms constant weight must be a positive integer within i32 (asap_sketchlib's \
+             Hydra adds i32 weights), got {weight}"
+        ))
     }
 }
 
diff --git a/crates/executor/src/operators/mod.rs b/crates/executor/src/operators/mod.rs
index 147475c1..5059582e 100644
--- a/crates/executor/src/operators/mod.rs
+++ b/crates/executor/src/operators/mod.rs
@@ -35,6 +35,13 @@ pub use aggregate::Reduction;
 pub use series_window::SubquerySteps;
 pub use sort::SortKey;
 pub use summary::SummaryEvaluation;
+/// A shared summary build's default row weight: one per row, a count.
+fn unit_weight() -> i32 {
+    1
+}
+fn is_unit_weight(weight: &i32) -> bool {
+    *weight == 1
+}
 #[derive(Clone, serde::Serialize, serde::Deserialize)]
 enum Kind {
     #[serde(skip)]
@@ -157,9 +164,13 @@ enum Kind {
     SharedSummaryBuild {
         family: SummaryFamilyType,
         item: usize,
-        /// The weight column; each row adds 1 without one.
+        /// The weight column; each row adds `constant_weight` without one.
         #[serde(default, skip_serializing_if = "Option::is_none")]
         weight: Option<usize>,
+        /// What each row adds when there is no weight column: 1 for a count,
+        /// or a positive integer constant weight.
+        #[serde(default = "unit_weight", skip_serializing_if = "is_unit_weight")]
+        constant_weight: i32,
         groups: Vec<usize>,
         /// Rows for which this is not true update no state; their group is kept.
         #[serde(default, skip_serializing_if = "Option::is_none")]
diff --git a/crates/executor/src/operators/summary/mod.rs b/crates/executor/src/operators/summary/mod.rs
index 9ee33f2c..44de52ec 100644
--- a/crates/executor/src/operators/summary/mod.rs
+++ b/crates/executor/src/operators/summary/mod.rs
@@ -63,15 +63,20 @@ impl Operator {
         })
     }
     /// One HydraCms state shared by every group, emitted as one row per group
-    /// that reads it. Each row adds its `weight` column's value (1 without
-    /// one) to its `item` value; a row whose weight is NULL adds nothing.
+    /// that reads it. Each row adds its `weight` column's value (or
+    /// `constant_weight` without one, which must be positive) to its `item`
+    /// value; a row whose weight is NULL adds nothing.
     pub fn shared_summary_build(
         input: SchemaRef,
         family: SummaryFamilyType,
         item: usize,
         weight: Option<usize>,
+        constant_weight: i32,
         groups: Vec<usize>,
     ) -> Result<Self, Error> {
+        if constant_weight <= 0 {
+            return Err(invalid("HydraCms constant weight must be positive"));
+        }
         crate::values::validate_family(&family)?;
         crate::factory::create_hydra_cms(&family).map_err(Error::Invalid)?;
         validate_groups(&input, &groups)?;
@@ -106,6 +111,7 @@ impl Operator {
                 family,
                 item,
                 weight,
+                constant_weight,
                 groups,
                 filter: None,
             },
@@ -340,13 +346,22 @@ pub(super) fn execute<'a>(
             family,
             item,
             weight,
+            constant_weight,
             groups,
             filter,
         } => Ok(futures::stream::once(async move {
             Batch::try_new(
                 output,
-                build_shared_summary(input, family, *item, *weight, groups, filter.as_deref(), &context)
-                    .await?,
+                build_shared_summary(
+                    input,
+                    family,
+                    *item,
+                    (*weight, *constant_weight),
+                    groups,
+                    filter.as_deref(),
+                    &context,
+                )
+                .await?,
             )
         })
         .boxed_local()),
@@ -785,7 +800,7 @@ async fn build_shared_summary(
     mut input: Input<'_, Batch>,
     family: &SummaryFamilyType,
     item: usize,
-    weight: Option<usize>,
+    (weight, constant_weight): (Option<usize>, i32),
     groups: &[usize],
     filter: Option<&Expression>,
     context: &RunContext,
@@ -820,7 +835,7 @@ async fn build_shared_summary(
                 continue;
             }
             let weight = match weight.map(|i| &row[i]) {
-                None => 1,
+                None => constant_weight,
                 Some(Value::Null) => continue,
                 Some(value) => hydra_weight(value)?,
             };
diff --git a/crates/executor/src/operators/unchecked.rs b/crates/executor/src/operators/unchecked.rs
index 9f6bd079..1d6980fe 100644
--- a/crates/executor/src/operators/unchecked.rs
+++ b/crates/executor/src/operators/unchecked.rs
@@ -170,10 +170,18 @@ impl TryFrom<UncheckedOperator> for Operator {
                 family,
                 item,
                 weight,
+                constant_weight,
                 groups,
                 filter,
             } => with_filter(
-                Operator::shared_summary_build(input(0)?, family, item, weight, groups)?,
+                Operator::shared_summary_build(
+                    input(0)?,
+                    family,
+                    item,
+                    weight,
+                    constant_weight,
+                    groups,
+                )?,
                 filter,
             )?,
             Kind::KeyedEvaluation { state, k } => {
diff --git a/crates/executor/src/physical_planner/mod.rs b/crates/executor/src/physical_planner/mod.rs
index 0d0ce4de..1dc798aa 100644
--- a/crates/executor/src/physical_planner/mod.rs
+++ b/crates/executor/src/physical_planner/mod.rs
@@ -430,12 +430,9 @@ fn compile_internal(
                         "per-entity summary requires a resolved raw time range",
                     ));
                 };
-                // A tumbling pane reads its window through a `TimeShift`; the
+                // A tumbling pane reads its window through `TimeShift`s; the
                 // deployment supplies the shifted raw rows.
-                let source = match child.non_asap() {
-                    Some(NonASAPOp::TimeShift { child, .. }) => child,
-                    _ => child,
-                };
+                let source = below_time_shifts(child, |node| node.non_asap(), Ok)?;
                 let Some(NonASAPOp::Scan { schema, .. }) = source.non_asap() else {
                     return Err(invalid("per-entity summary requires a resolved source"));
                 };
@@ -598,6 +595,21 @@ fn compile_internal(
     Ok(physical_dag)
 }
 
+/// The first node at or below `node` that is not a `TimeShift`, following
+/// each shift's child: a range reads its raw source through any number of
+/// stacked shifts. `operator` gives a node's non-ASAP operator (`None` for an
+/// ASAP one) and `resolve` a child's node; an error when a child is missing.
+pub(crate) fn below_time_shifts<'a, N, C: 'a>(
+    mut node: &'a N,
+    operator: impl Fn(&'a N) -> Option<&'a NonASAPOp<C>>,
+    resolve: impl Fn(&'a C) -> Result<&'a N, Error>,
+) -> Result<&'a N, Error> {
+    while let Some(NonASAPOp::TimeShift { child, .. }) = operator(node) {
+        node = resolve(child)?;
+    }
+    Ok(node)
+}
+
 /// The logical window of the counter evaluation `id`, as `(offset,
 /// lookback)`: it ends `offset` ms before the evaluation time and spans
 /// `lookback` ms. It is the union of the range windows the evaluation
@@ -1013,16 +1025,22 @@ fn summary_build(
         let Some(SummaryInputExpr::Column(item)) = &update.item else {
             unreachable!("validated HydraCms item column")
         };
-        // Validated: the unit count, or a non-negative weight column.
-        let weight = match &update.weight {
-            SummaryInputExpr::Column(weight) => Some(named_column(input, weight)?),
-            _ => None,
+        // Validated: the unit count, a positive integer constant or a
+        // non-negative weight column.
+        let (weight, constant_weight) = match &update.weight {
+            SummaryInputExpr::Column(weight) => (Some(named_column(input, weight)?), 1),
+            SummaryInputExpr::Constant(weight) => (
+                None,
+                crate::capability::hydra_constant_weight(*weight).map_err(Error::Invalid)?,
+            ),
+            _ => (None, 1),
         };
         return Operator::shared_summary_build(
             input.clone(),
             family.clone(),
             named_column(input, item)?,
             weight,
+            constant_weight,
             groups(input, keys)?,
         );
     }
diff --git a/crates/executor/src/physical_planner/precompute.rs b/crates/executor/src/physical_planner/precompute.rs
index 8afaed70..a20a2f7e 100644
--- a/crates/executor/src/physical_planner/precompute.rs
+++ b/crates/executor/src/physical_planner/precompute.rs
@@ -88,16 +88,22 @@ pub fn boundary_schema(
     node: &PhysicalASAPDAGNode,
     nodes: &BTreeMap<u64, &PhysicalASAPDAGNode>,
 ) -> Result<SchemaRef, Error> {
+    let resolve = |child: &usize| {
+        nodes
+            .get(&(*child as u64))
+            .copied()
+            .ok_or_else(|| invalid("missing precompute node"))
+    };
     let mut scan = node;
     if let Payload::NonASAP(NonASAPOp::TimeRange { child, .. }) = &scan.payload {
-        scan = nodes
-            .get(&(*child as u64))
-            .ok_or_else(|| invalid("missing precompute node"))?;
-        while let Payload::NonASAP(NonASAPOp::TimeShift { child, .. }) = &scan.payload {
-            scan = nodes
-                .get(&(*child as u64))
-                .ok_or_else(|| invalid("missing precompute node"))?;
-        }
+        scan = below_time_shifts(
+            resolve(child)?,
+            |node| match &node.payload {
+                Payload::NonASAP(operator) => Some(operator),
+                Payload::ASAP(_) => None,
+            },
+            resolve,
+        )?;
     }
     if !matches!(
         &scan.payload,
diff --git a/crates/executor/src/summary_kernels/factory.rs b/crates/executor/src/summary_kernels/factory.rs
index be9c21d5..b09c5e0a 100644
--- a/crates/executor/src/summary_kernels/factory.rs
+++ b/crates/executor/src/summary_kernels/factory.rs
@@ -59,15 +59,24 @@ pub trait AccumulatorUpdater: Send {
             return Err("summary update requires Float64".into());
         };
         self.validate_single_input(*value)?;
-        self.update_single(*value, timestamp_ms);
-        Ok(())
+        self.update_single(*value, timestamp_ms)
     }
 
-    /// Feed a single (value, timestamp_ms) pair — for SingleSubpopulation types.
-    fn update_single(&mut self, value: f64, timestamp_ms: i64);
+    /// Feed a single (value, timestamp_ms) pair — for SingleSubpopulation
+    /// types. An error when the state cannot take the value (a counter or a
+    /// sum past its range) or when the updater is keyed; the state is then
+    /// unchanged.
+    fn update_single(&mut self, value: f64, timestamp_ms: i64) -> Result<(), String>;
 
-    /// Feed a keyed (key, value, timestamp_ms) triple, e.g. a frequency item or an exact keyed state.
-    fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, timestamp_ms: i64);
+    /// Feed a keyed (key, value, timestamp_ms) triple, e.g. a frequency item
+    /// or an exact keyed state. An error when the state cannot take the
+    /// value (a sum past `f64::MAX`); the state is then unchanged.
+    fn update_keyed(
+        &mut self,
+        key: &KeyByLabelValues,
+        value: f64,
+        timestamp_ms: i64,
+    ) -> Result<(), String>;
 
     /// Extract the final accumulator as a boxed `AggregateCore`.
     fn take_accumulator(&mut self) -> Box<dyn AggregateCore>;
@@ -112,12 +121,18 @@ impl KllAccumulatorUpdater {
 }
 
 impl AccumulatorUpdater for KllAccumulatorUpdater {
-    fn update_single(&mut self, value: f64, _timestamp_ms: i64) {
+    fn update_single(&mut self, value: f64, _timestamp_ms: i64) -> Result<(), String> {
         self.acc.update(value);
+        Ok(())
     }
 
-    fn update_keyed(&mut self, _key: &KeyByLabelValues, value: f64, timestamp_ms: i64) {
-        self.update_single(value, timestamp_ms);
+    fn update_keyed(
+        &mut self,
+        _key: &KeyByLabelValues,
+        value: f64,
+        timestamp_ms: i64,
+    ) -> Result<(), String> {
+        self.update_single(value, timestamp_ms)
     }
 
     impl_clone_accumulator_methods!(acc);
@@ -164,12 +179,18 @@ impl AccumulatorUpdater for DDSketchAccumulatorUpdater {
         }
     }
 
-    fn update_single(&mut self, value: f64, _timestamp_ms: i64) {
+    fn update_single(&mut self, value: f64, _timestamp_ms: i64) -> Result<(), String> {
         self.acc.inner.update(value);
+        Ok(())
     }
 
-    fn update_keyed(&mut self, _key: &KeyByLabelValues, value: f64, timestamp_ms: i64) {
-        self.update_single(value, timestamp_ms);
+    fn update_keyed(
+        &mut self,
+        _key: &KeyByLabelValues,
+        value: f64,
+        timestamp_ms: i64,
+    ) -> Result<(), String> {
+        self.update_single(value, timestamp_ms)
     }
 
     impl_clone_accumulator_methods!(acc);
@@ -215,15 +236,18 @@ impl CmsAccumulatorUpdater {
 }
 
 impl AccumulatorUpdater for CmsAccumulatorUpdater {
-    fn update_single(&mut self, _value: f64, _timestamp_ms: i64) {
-        debug_assert!(
-            false,
-            "update_single called on keyed updater; use update_keyed"
-        );
+    fn update_single(&mut self, _value: f64, _timestamp_ms: i64) -> Result<(), String> {
+        Err("update_single called on a keyed updater; use update_keyed".into())
     }
 
-    fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) {
+    fn update_keyed(
+        &mut self,
+        key: &KeyByLabelValues,
+        value: f64,
+        _timestamp_ms: i64,
+    ) -> Result<(), String> {
         self.acc.inner.update(&key.to_semicolon_str(), value);
+        Ok(())
     }
 
     impl_clone_accumulator_methods!(acc);
@@ -284,19 +308,21 @@ impl AccumulatorUpdater for WeightedFrequencyUpdater {
             .map_err(|error| error.to_string())
     }
 
-    fn update_single(&mut self, _value: f64, _timestamp_ms: i64) {
-        debug_assert!(
-            false,
-            "update_single called on keyed updater; use update_keyed"
-        );
+    fn update_single(&mut self, _value: f64, _timestamp_ms: i64) -> Result<(), String> {
+        Err("update_single called on a keyed updater; use update_keyed".into())
     }
 
-    fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) {
+    fn update_keyed(
+        &mut self,
+        key: &KeyByLabelValues,
+        value: f64,
+        _timestamp_ms: i64,
+    ) -> Result<(), String> {
         // Inputs pass `validate_single_input` first; only a sum past
         // `f64::MAX` can fail here.
         self.acc
             .update_labels(&key.labels, value)
-            .expect("weighted frequency counter overflow");
+            .map_err(|error| error.to_string())
     }
 
     impl_clone_accumulator_methods!(acc);
@@ -337,15 +363,18 @@ impl HydraKllAccumulatorUpdater {
 }
 
 impl AccumulatorUpdater for HydraKllAccumulatorUpdater {
-    fn update_single(&mut self, _value: f64, _timestamp_ms: i64) {
-        debug_assert!(
-            false,
-            "update_single called on keyed updater; use update_keyed"
-        );
+    fn update_single(&mut self, _value: f64, _timestamp_ms: i64) -> Result<(), String> {
+        Err("update_single called on a keyed updater; use update_keyed".into())
     }
 
-    fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) {
+    fn update_keyed(
+        &mut self,
+        key: &KeyByLabelValues,
+        value: f64,
+        _timestamp_ms: i64,
+    ) -> Result<(), String> {
         self.acc.update(key, value);
+        Ok(())
     }
 
     impl_clone_accumulator_methods!(acc);
@@ -503,11 +532,18 @@ struct PlannerExactUpdater {
     acc: crate::summary_kernels::exact::ExactAccumulator,
 }
 impl AccumulatorUpdater for PlannerExactUpdater {
-    fn update_single(&mut self, value: f64, timestamp: i64) {
+    fn update_single(&mut self, value: f64, timestamp: i64) -> Result<(), String> {
         self.acc.update(None, value, timestamp);
+        Ok(())
     }
-    fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, timestamp: i64) {
+    fn update_keyed(
+        &mut self,
+        key: &KeyByLabelValues,
+        value: f64,
+        timestamp: i64,
+    ) -> Result<(), String> {
         self.acc.update(Some(key), value, timestamp);
+        Ok(())
     }
     impl_clone_accumulator_methods!(acc);
     fn reset(&mut self) {
@@ -540,14 +576,20 @@ impl AccumulatorUpdater for HllUpdater {
     fn memory_usage_bytes(&self) -> usize {
         self.acc.approx_memory_bytes()
     }
-    fn update_single(&mut self, value: f64, _: i64) {
+    fn update_single(&mut self, value: f64, _: i64) -> Result<(), String> {
         if !value.is_nan() {
             let bits = if value == 0.0 { 0 } else { value.to_bits() };
             self.acc.inner.update(&bits.to_le_bytes());
         }
+        Ok(())
     }
-    fn update_keyed(&mut self, _: &KeyByLabelValues, value: f64, timestamp_ms: i64) {
-        self.update_single(value, timestamp_ms);
+    fn update_keyed(
+        &mut self,
+        _: &KeyByLabelValues,
+        value: f64,
+        timestamp_ms: i64,
+    ) -> Result<(), String> {
+        self.update_single(value, timestamp_ms)
     }
     impl_clone_accumulator_methods!(acc);
     fn reset(&mut self) {
@@ -568,13 +610,18 @@ impl AccumulatorUpdater for UnivMonUpdater {
     fn memory_usage_bytes(&self) -> usize {
         self.acc.approx_memory_bytes()
     }
-    fn update_single(&mut self, value: f64, _: i64) {
+    fn update_single(&mut self, value: f64, _: i64) -> Result<(), String> {
         self.acc
             .insert_sample(value)
-            .expect("UnivMon sample counter overflow");
+            .map_err(|error| error.to_string())
     }
-    fn update_keyed(&mut self, _: &KeyByLabelValues, value: f64, timestamp_ms: i64) {
-        self.update_single(value, timestamp_ms);
+    fn update_keyed(
+        &mut self,
+        _: &KeyByLabelValues,
+        value: f64,
+        timestamp_ms: i64,
+    ) -> Result<(), String> {
+        self.update_single(value, timestamp_ms)
     }
     impl_clone_accumulator_methods!(acc);
     fn reset(&mut self) {
diff --git a/crates/executor/src/summary_kernels/kmv.rs b/crates/executor/src/summary_kernels/kmv.rs
index 8ca1a089..6f6ab50a 100644
--- a/crates/executor/src/summary_kernels/kmv.rs
+++ b/crates/executor/src/summary_kernels/kmv.rs
@@ -85,11 +85,13 @@ impl AccumulatorUpdater for KmvAccumulator {
     fn memory_usage_bytes(&self) -> usize {
         self.approx_memory_bytes()
     }
-    fn update_single(&mut self, value: f64, _: i64) {
+    fn update_single(&mut self, value: f64, _: i64) -> Result<(), String> {
         self.insert_sample(value);
+        Ok(())
     }
-    fn update_keyed(&mut self, _: &KeyByLabelValues, value: f64, _: i64) {
+    fn update_keyed(&mut self, _: &KeyByLabelValues, value: f64, _: i64) -> Result<(), String> {
         self.insert_sample(value);
+        Ok(())
     }
     fn take_accumulator(&mut self) -> Box<dyn AggregateCore> {
         let result = Box::new(self.clone());
@@ -202,7 +204,7 @@ mod tests {
     fn negative_zero_is_zero_and_nan_is_not_an_item() {
         let mut kmv = KmvAccumulator::new(16);
         for v in [0.0, -0.0, f64::NAN, 1.0] {
-            kmv.update_single(v, 0);
+            kmv.update_single(v, 0).unwrap();
         }
         assert_eq!(cardinality(&kmv), 2.0);
         assert_eq!(cardinality(kmv.take_accumulator().as_ref()), 2.0);
diff --git a/crates/executor/tests/common/mod.rs b/crates/executor/tests/common/mod.rs
index 9bd25ef4..a4feec86 100644
--- a/crates/executor/tests/common/mod.rs
+++ b/crates/executor/tests/common/mod.rs
@@ -68,11 +68,13 @@ fn run_stage1_to_stage3(
     asap_plan_selection::run_stage1_to_stage3(
         vec![(0, QueryRoot::Operator(root))],
         &[planner_types::workload::QueryWorkloadEntry {
+            query: Default::default(),
             requirements: planner_types::workload::QueryRequirements {
                 accuracy,
                 max_latency_ms: None,
             },
-            ..Default::default()
+            predictability: Default::default(),
+            recurrence: Default::default(),
         }],
         &data,
         asap_plan_selection::PlanningModels::builtin(),
diff --git a/crates/executor/tests/deployment.rs b/crates/executor/tests/deployment.rs
index 857cb5d5..4e43b2c4 100644
--- a/crates/executor/tests/deployment.rs
+++ b/crates/executor/tests/deployment.rs
@@ -28,7 +28,7 @@ fn build(values: &[f64]) -> Box<dyn AggregateCore> {
     .unwrap();
     for (at, value) in values.iter().enumerate() {
         operator.validate_single_input(*value).unwrap();
-        operator.update_single(*value, at as i64);
+        operator.update_single(*value, at as i64).unwrap();
     }
     operator.into_accumulator()
 }
@@ -85,7 +85,7 @@ fn native_count_sketch_dimensions_are_not_packed_wire_dimensions() {
     update.item = Some(SummaryInputExpr::Column(ColumnRef::Named("host".into())));
     let mut operator = create_planner_accumulator(&family, &update, &Default::default()).unwrap();
     let key = KeyByLabelValues::new_with_labels(vec!["a".into()]);
-    operator.update_keyed(&key, 7.0, 1000);
+    operator.update_keyed(&key, 7.0, 1000).unwrap();
     let state = operator.into_accumulator();
     let state = state
         .as_any()
@@ -132,7 +132,7 @@ fn build_weighted_frequency(
     let mut operator = create_planner_accumulator(family, &update, &Default::default()).unwrap();
     for (at, (i, weight)) in updates.into_iter().enumerate() {
         operator.validate_single_input(weight).unwrap();
-        operator.update_keyed(&host(i), weight, at as i64);
+        operator.update_keyed(&host(i), weight, at as i64).unwrap();
     }
     operator.into_accumulator()
 }
@@ -398,3 +398,21 @@ fn plain_count_sketch_merge_equals_union_build() {
         ))
         .is_err());
 }
+
+// A keyed weighted-frequency update whose sum would pass `f64::MAX` returns
+// an error from `update_keyed` instead of panicking; the state keeps the
+// last good sum.
+#[test]
+fn keyed_update_overflow_is_an_error_not_a_panic() {
+    use asap_executor::planner::ir::schema::SummaryInputExpr;
+    let mut update = SummaryUpdate::column(ColumnRef::SampleValue);
+    update.item = Some(SummaryInputExpr::Column(ColumnRef::Named("host".into())));
+    let family = plain_count_sketch(64, 3);
+    let mut operator = create_planner_accumulator(&family, &update, &Default::default()).unwrap();
+    operator.validate_single_input(f64::MAX).unwrap();
+    operator.update_keyed(&host(0), f64::MAX, 0).unwrap();
+    let error = operator.update_keyed(&host(0), f64::MAX, 1).unwrap_err();
+    assert!(error.contains("overflow"), "{error}");
+    let state = operator.into_accumulator();
+    assert_eq!(plain_estimate(&*state, 0), f64::MAX);
+}
diff --git a/crates/executor/tests/hydra_cms_execution.rs b/crates/executor/tests/hydra_cms_execution.rs
index 99c0d758..3a8d895b 100644
--- a/crates/executor/tests/hydra_cms_execution.rs
+++ b/crates/executor/tests/hydra_cms_execution.rs
@@ -329,6 +329,32 @@ fn weighted_hydra_cms_sums_item_weights_per_group() {
     );
 }
 
+// A positive constant weight, `Constant(2.0)` with its `PositiveConstant`
+// proof, adds 2 per row: each job's point count is twice the unit count's.
+// A non-integer constant is refused at binding, since sketchlib's Hydra adds
+// integer weights.
+#[test]
+fn positive_constant_weight_adds_the_constant_per_row() {
+    let constant = |weight: f64| SummaryUpdate {
+        weight: SummaryInputExpr::Constant(weight),
+        weight_domain: WeightDomain::NonNegative {
+            proof: NonNegativeWeightProof::PositiveConstant,
+        },
+        ..unit_count()
+    };
+    let unit = run(unit_count(), None, vec![batch(&ROWS)]);
+    let doubled = run(constant(2.0), None, vec![batch(&ROWS)]);
+    for job in ["api", "batch", "db"] {
+        assert_eq!(doubled[job], 2.0 * unit[job], "{job}");
+    }
+    for weight in [0.0, -2.0, 2.5, f64::INFINITY] {
+        let error = compile_node(&summary_agg(constant(weight), None), &[input_schema()])
+            .err()
+            .unwrap_or_else(|| panic!("constant weight {weight} accepted"));
+        assert!(error.to_string().contains("HydraCms"), "{weight}: {error}");
+    }
+}
+
 // A weight column of unknown sign is rejected at binding: a Count-Min point
 // count only overestimates when no weight is negative.
 #[test]
diff --git a/crates/executor/tests/precompute_candidates.rs b/crates/executor/tests/precompute_candidates.rs
index d34506b1..ba88821d 100644
--- a/crates/executor/tests/precompute_candidates.rs
+++ b/crates/executor/tests/precompute_candidates.rs
@@ -125,7 +125,7 @@ fn grouped_rate_can_be_materialized_before_or_after_grouped_sum() {
         .map(|(index, values)| {
             let mut accumulator = create_planner_accumulator(family, update, grouping).unwrap();
             for (i, value) in values.into_iter().enumerate() {
-                accumulator.update_single(value, i as i64 * 1000);
+                accumulator.update_single(value, i as i64 * 1000).unwrap();
             }
             let state = accumulator.into_accumulator();
             expected_rate_sum += state
@@ -349,7 +349,7 @@ fn grouped_rate_can_be_materialized_before_or_after_grouped_sum() {
     assert_eq!(results[1], results[2]);
     let mut wrong_order = create_planner_accumulator(family, update, grouping).unwrap();
     for (i, value) in [200., 200., 100.].into_iter().enumerate() {
-        wrong_order.update_single(value, i as i64 * 1000);
+        wrong_order.update_single(value, i as i64 * 1000).unwrap();
     }
     let rate_of_sum = wrong_order
         .into_accumulator()
@@ -459,7 +459,7 @@ fn enumerated_grouped_rate_candidates_execute_numeric_query_outputs() {
             .map(|instance| {
                 let mut accumulator = create_planner_accumulator(family, input, grouping).unwrap();
                 for (timestamp, value) in [(1_000, 1.), (31_000, 31.), (59_000, 59.)] {
-                    accumulator.update_single(value, timestamp);
+                    accumulator.update_single(value, timestamp).unwrap();
                 }
                 let summary = Value::Summary {
                     family: family.clone(),
diff --git a/crates/executor/tests/precompute_population.rs b/crates/executor/tests/precompute_population.rs
index a1751c46..77b97b9c 100644
--- a/crates/executor/tests/precompute_population.rs
+++ b/crates/executor/tests/precompute_population.rs
@@ -196,7 +196,7 @@ fn finalized_shared_panes_rebuild_one_global_summary_after_recovery() {
                     &GroupingStrategy::PerSubpopulationInstance,
                 )
                 .unwrap();
-                state.update_single(value, time);
+                state.update_single(value, time).unwrap();
                 vec![
                     Value::Map(
                         vec![(Value::Utf8("instance".into()), Value::Utf8(group.into()))].into(),
diff --git a/crates/executor/tests/promql_values.rs b/crates/executor/tests/promql_values.rs
index 554bddb6..22d8cc8b 100644
--- a/crates/executor/tests/promql_values.rs
+++ b/crates/executor/tests/promql_values.rs
@@ -422,7 +422,7 @@ fn exact_state_evaluations_recover_and_finalize_panes() {
                     )
                     .unwrap();
                     for sample in samples {
-                        state.update_single(sample, 0);
+                        state.update_single(sample, 0).unwrap();
                     }
                     let labels = row(&[("__name__", "m"), ("instance", "a")], 0.).remove(0);
                     vec![
@@ -469,9 +469,9 @@ fn recovered_exact_counter_uses_window_and_omits_insufficient_samples() {
                     &GroupingStrategy::PerSubpopulationInstance,
                 )
                 .unwrap();
-                state.update_single(100., -50_000);
+                state.update_single(100., -50_000).unwrap();
                 if count == 2 {
-                    state.update_single(140., -10_000);
+                    state.update_single(140., -10_000).unwrap();
                 }
                 vec![
                     row(&[("instance", if count == 1 { "one" } else { "two" })], 0.).remove(0),
diff --git a/crates/executor/tests/summary_projection.rs b/crates/executor/tests/summary_projection.rs
index c88d4b52..89802262 100644
--- a/crates/executor/tests/summary_projection.rs
+++ b/crates/executor/tests/summary_projection.rs
@@ -137,7 +137,7 @@ fn post_asap_summary_projection_survives_recovery() {
         &GroupingStrategy::PerSubpopulationInstance,
     )
     .unwrap();
-    accumulator.update_single(7., 1);
+    accumulator.update_single(7., 1).unwrap();
     let state = Arc::from(accumulator.into_accumulator());
     let batch = Batch::try_new(
         schema.clone(),
diff --git a/crates/executor/tests/weighted_topk_binding.rs b/crates/executor/tests/weighted_topk_binding.rs
index 183781df..7a69de72 100644
--- a/crates/executor/tests/weighted_topk_binding.rs
+++ b/crates/executor/tests/weighted_topk_binding.rs
@@ -670,7 +670,9 @@ fn maintained_rate_heap_compiles_fixed_window_precompute() {
                         asap_executor::factory::create_planner_accumulator(family, input, grouping)
                             .unwrap();
                     for (offset, value) in [10_000, 30_000, 50_000].into_iter().zip(samples) {
-                        accumulator.update_single(value, end - 60_000 + offset);
+                        accumulator
+                            .update_single(value, end - 60_000 + offset)
+                            .unwrap();
                     }
                     let summary = Value::Summary {
                         family: family.clone(),
diff --git a/crates/frontend-promql/tests/support.rs b/crates/frontend-promql/tests/support.rs
index 445f3387..23b0ea0f 100644
--- a/crates/frontend-promql/tests/support.rs
+++ b/crates/frontend-promql/tests/support.rs
@@ -116,11 +116,13 @@ pub fn run_stage1_to_stage3(
     asap_plan_selection::run_stage1_to_stage3(
         vec![(0, asap_types::ir::QueryRoot::Operator(root))],
         &[asap_types::workload::QueryWorkloadEntry {
+            query: Default::default(),
             requirements: asap_types::workload::QueryRequirements {
                 accuracy,
                 max_latency_ms: None,
             },
-            ..Default::default()
+            predictability: Default::default(),
+            recurrence: Default::default(),
         }],
         &DataWorkload::default(),
         asap_plan_selection::PlanningModels::builtin(),
@@ -173,11 +175,13 @@ pub fn logical_plan(root: &Rc<OperatorNode>) -> Result<Rc<OperatorNode>, String>
     let run = asap_plan_selection::run_stage1_to_stage3(
         vec![(0, asap_types::ir::QueryRoot::Operator(Rc::clone(root)))],
         &[asap_types::workload::QueryWorkloadEntry {
+            query: Default::default(),
             requirements: asap_types::workload::QueryRequirements {
                 accuracy: asap_types::ir::properties::AccuracyTarget::exact(),
                 max_latency_ms: None,
             },
-            ..Default::default()
+            predictability: Default::default(),
+            recurrence: Default::default(),
         }],
         &DataWorkload::default(),
         asap_plan_selection::PlanningModels::builtin(),
diff --git a/crates/frontend-sql/tests/pearson_corr.rs b/crates/frontend-sql/tests/pearson_corr.rs
index df8f81fa..f082277e 100644
--- a/crates/frontend-sql/tests/pearson_corr.rs
+++ b/crates/frontend-sql/tests/pearson_corr.rs
@@ -152,11 +152,13 @@ async fn corr_survives_exact_plan_compilation() {
     let run = asap_plan_selection::run_stage1_to_stage3(
         vec![(0, asap_types::ir::QueryRoot::Operator(query.clone()))],
         &[asap_types::workload::QueryWorkloadEntry {
+            query: Default::default(),
             requirements: asap_types::workload::QueryRequirements {
                 accuracy: asap_types::ir::properties::AccuracyTarget::exact(),
                 max_latency_ms: None,
             },
-            ..Default::default()
+            predictability: Default::default(),
+            recurrence: Default::default(),
         }],
         &Default::default(),
         asap_plan_selection::PlanningModels::builtin(),
diff --git a/crates/integration-tests/src/lib.rs b/crates/integration-tests/src/lib.rs
index a37ce7cc..df9f6142 100644
--- a/crates/integration-tests/src/lib.rs
+++ b/crates/integration-tests/src/lib.rs
@@ -143,7 +143,7 @@ pub mod post_asap {
 pub mod planning {
     use asap_types::ir::properties::AccuracyTarget;
     use asap_types::ir::{OperatorNode, QueryRoot};
-    use asap_types::workload::{DataWorkload, QueryRequirements, QueryWorkloadEntry};
+    use asap_types::workload::DataWorkload;
     use std::rc::Rc;
 
     /// Every planning stage's output for the single query `root`.
@@ -153,12 +153,14 @@ pub mod planning {
     ) -> asap_plan_selection::Stage1ToStage3Run<usize> {
         asap_plan_selection::run_stage1_to_stage3(
             vec![(0, QueryRoot::Operator(root))],
-            &[QueryWorkloadEntry {
-                requirements: QueryRequirements {
+            &[asap_types::workload::QueryWorkloadEntry {
+                query: Default::default(),
+                requirements: asap_types::workload::QueryRequirements {
                     accuracy,
                     max_latency_ms: None,
                 },
-                ..Default::default()
+                predictability: Default::default(),
+                recurrence: Default::default(),
             }],
             &DataWorkload::default(),
             asap_plan_selection::PlanningModels::builtin(),
diff --git a/crates/integration-tests/tests/planner_layering_example1.rs b/crates/integration-tests/tests/planner_layering_example1.rs
index 8a58806d..05481ed8 100644
--- a/crates/integration-tests/tests/planner_layering_example1.rs
+++ b/crates/integration-tests/tests/planner_layering_example1.rs
@@ -134,8 +134,10 @@ fn every_10s() -> asap_types::workload::QueryRecurrence {
 /// The workload entry of an exact query evaluated every 10 s.
 fn exact_every_10s() -> asap_types::workload::QueryWorkloadEntry {
     asap_types::workload::QueryWorkloadEntry {
+        query: Default::default(),
+        requirements: Default::default(),
+        predictability: Default::default(),
         recurrence: every_10s(),
-        ..Default::default()
     }
 }
 
diff --git a/crates/integration-tests/tests/precompute_raw_samples.rs b/crates/integration-tests/tests/precompute_raw_samples.rs
index f14bc886..ccfe1f58 100644
--- a/crates/integration-tests/tests/precompute_raw_samples.rs
+++ b/crates/integration-tests/tests/precompute_raw_samples.rs
@@ -383,7 +383,9 @@ fn check(
             .entry(population(reduction, &keys, labels))
             .or_insert_with(|| create_planner_accumulator(family, input, grouping).unwrap());
         let unit = input.item.is_some();
-        updater.update_single(if unit { *value } else { weight(input, *value) }, *time);
+        updater
+            .update_single(if unit { *value } else { weight(input, *value) }, *time)
+            .unwrap();
     }
     assert_eq!(actual.len(), expected.len(), "{query}");
     for (labels, state) in actual {
diff --git a/crates/integration-tests/tests/promql_to_post_asap.rs b/crates/integration-tests/tests/promql_to_post_asap.rs
index 1aa326df..3830f8a1 100644
--- a/crates/integration-tests/tests/promql_to_post_asap.rs
+++ b/crates/integration-tests/tests/promql_to_post_asap.rs
@@ -26,7 +26,7 @@ use asap_types::ir::schema::{
     SketchAlgorithm, SketchKind, SketchParams, SketchStatistic, SummaryInputExpr, SummaryUpdate,
 };
 use asap_types::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode, QueryRoot, ScalarExpr};
-use asap_types::workload::{DataWorkload, QueryRequirements, QueryWorkloadEntry};
+use asap_types::workload::DataWorkload;
 
 /// The first logical candidate (Stage 1, in enumeration order) of `root` as
 /// a query at `accuracy` that uses an ASAP operator, or the plain exact one
@@ -707,12 +707,14 @@ fn ddsketch_ratio_selection(
     let root = lower_promql(query, planned).unwrap();
     let run = asap_plan_selection::run_stage1_to_stage3(
         vec![(0, QueryRoot::Operator(root))],
-        &[QueryWorkloadEntry {
-            requirements: QueryRequirements {
+        &[asap_types::workload::QueryWorkloadEntry {
+            query: Default::default(),
+            requirements: asap_types::workload::QueryRequirements {
                 accuracy,
                 max_latency_ms: None,
             },
-            ..Default::default()
+            predictability: Default::default(),
+            recurrence: Default::default(),
         }],
         &DataWorkload::default(),
         PlanningModels::builtin().with_evidence(evidence),
diff --git a/crates/integration-tests/tests/reuse_rules.rs b/crates/integration-tests/tests/reuse_rules.rs
index cd589206..9fa03c48 100644
--- a/crates/integration-tests/tests/reuse_rules.rs
+++ b/crates/integration-tests/tests/reuse_rules.rs
@@ -57,8 +57,10 @@ fn plan(queries: &[&str], recurrence: QueryRecurrence) -> Stage1ToStage3Run<usiz
 /// The workload entry of an exact query evaluated as `recurrence` says.
 fn exact(recurrence: QueryRecurrence) -> QueryWorkloadEntry {
     QueryWorkloadEntry {
+        query: Default::default(),
+        requirements: Default::default(),
+        predictability: Default::default(),
         recurrence,
-        ..Default::default()
     }
 }
 
diff --git a/crates/integration-tests/tests/tumbling_pane_merge.rs b/crates/integration-tests/tests/tumbling_pane_merge.rs
index b2b2f48f..bc87e617 100644
--- a/crates/integration-tests/tests/tumbling_pane_merge.rs
+++ b/crates/integration-tests/tests/tumbling_pane_merge.rs
@@ -537,6 +537,62 @@ fn offset_counter_window_ends_at_the_offset() {
     }
 }
 
+/// A counter over `[5m] @ 300` reads the window ending at 300 s whatever the
+/// evaluation time, and `@ 300 offset 1m` the one ending at 240 s: the same
+/// values as the unanchored counter evaluated at 300 s and at 240 s. Today
+/// it reads the window ending at the evaluation time (#748).
+#[test]
+#[ignore = "the counter window drops an @ anchor and its offset: #748"]
+fn anchored_counter_window_ends_at_the_anchor() {
+    let samples = samples(7_000);
+    for function in ["rate", "increase"] {
+        let at_300 = whole_build_values_at(&format!("{function}(m[5m])"), EVALUATION_MS, &samples);
+        let at_240 = whole_build_values_at(
+            &format!("{function}(m[5m])"),
+            EVALUATION_MS - PANE_MS,
+            &samples,
+        );
+        assert_ne!(at_300, at_240, "{function}: the two windows differ");
+        let anchored = whole_build_values_at(
+            &format!("{function}(m[5m] @ 300)"),
+            EVALUATION_MS + PANE_MS,
+            &samples,
+        );
+        assert_eq!(anchored, at_300, "{function} @ 300");
+        let anchored_offset = whole_build_values_at(
+            &format!("{function}(m[5m] @ 300 offset 1m)"),
+            EVALUATION_MS + PANE_MS,
+            &samples,
+        );
+        assert_eq!(anchored_offset, at_240, "{function} @ 300 offset 1m");
+    }
+}
+
+/// Panes whose ranges read the scan through two stacked `TimeShift`s (the
+/// pane's shift over a zero shift) compile and answer as the single build:
+/// the per-entity build looks through every shift to its raw source.
+#[test]
+fn panes_over_stacked_time_shifts_match_single_build() {
+    let single = single_build("increase(m[5m])", AccuracyTarget::exact());
+    let (panes, _) = tumbling_panes(&single);
+    let agg = summary_agg(&single);
+    let Some(NonASAPOp::TimeRange { child: scan, .. }) = agg.children()[0].non_asap() else {
+        panic!("expected a raw time range under the summary");
+    };
+    let zero_shift = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeShift {
+        shift: TimeShift {
+            offset_ms: 0,
+            at: None,
+        },
+        child: scan.clone(),
+    }))
+    .unwrap();
+    let stacked = replace(&panes, scan, &zero_shift);
+    let expected = render(&execute(&single, &samples(0)));
+    assert_eq!(expected.len(), 2, "{expected:?}");
+    assert_eq!(render(&execute(&stacked, &samples(0))), expected);
+}
+
 /// Exact counter Increase over 5 merged 1-minute panes equals one 5-minute
 /// build: the merge joins consecutive panes across their boundaries.
 #[test]
diff --git a/crates/logical-asap-optimization/src/asap_aware_cse.rs b/crates/logical-asap-optimization/src/asap_aware_cse.rs
index 1e65cf35..9113d974 100644
--- a/crates/logical-asap-optimization/src/asap_aware_cse.rs
+++ b/crates/logical-asap-optimization/src/asap_aware_cse.rs
@@ -517,9 +517,10 @@ pub fn summary_capability_pass2_candidates(
         .iter()
         .any(|candidate| candidate.members.len() > MAX_CAPABILITY_SHARING_MEMBERS)
     {
-        return Err(PerComputationCandidateError::Unsupported(
-            "a capability sharing candidate of more than 20 summary builds (2^n sharing subsets)",
-        ));
+        return Err(PerComputationCandidateError::TooManyCandidates {
+            candidates: "summary builds in one capability sharing candidate",
+            max: MAX_CAPABILITY_SHARING_MEMBERS,
+        });
     }
     // Each capability sharing candidate's options: `None` (no sharing),
     // then its subsets.
@@ -787,9 +788,10 @@ pub fn window_composition_pass2_candidates(
     candidates: &[PaneCandidate],
 ) -> Result<Vec<WindowCompositionPass2Candidate>, PerComputationCandidateError> {
     if candidates.len() > MAX_PANE_CANDIDATES {
-        return Err(PerComputationCandidateError::Unsupported(
-            "more than 20 summary builds that could be built in tumbling panes (2^n candidates)",
-        ));
+        return Err(PerComputationCandidateError::TooManyCandidates {
+            candidates: "summary builds that could be built in tumbling panes",
+            max: MAX_PANE_CANDIDATES,
+        });
     }
     let mut pass2_candidates = Vec::new();
     for mask in 1u64..(1u64 << candidates.len()) {
@@ -1271,10 +1273,13 @@ fn subsets<T: Clone>(
 ) -> Result<Vec<Vec<T>>, PerComputationCandidateError> {
     let n = candidates.len();
     if n > MAX_REUSE_CANDIDATES {
-        return Err(PerComputationCandidateError::Unsupported(match rule {
-            "rollup" => "more than 20 rollup candidates (2^n Pass 2 candidates)",
-            _ => "more than 20 top-k reuse candidates (2^n Pass 2 candidates)",
-        }));
+        return Err(PerComputationCandidateError::TooManyCandidates {
+            candidates: match rule {
+                "rollup" => "rollup candidates",
+                _ => "top-k reuse candidates",
+            },
+            max: MAX_REUSE_CANDIDATES,
+        });
     }
     Ok((1u64..(1u64 << n))
         .map(|mask| {
@@ -1942,10 +1947,18 @@ mod tests {
             .take(MAX_CAPABILITY_SHARING_MEMBERS + 1)
             .cloned()
             .collect();
+        let error = summary_capability_pass2_candidates(&roots, &sharing).unwrap_err();
         assert!(matches!(
-            summary_capability_pass2_candidates(&roots, &sharing),
-            Err(PerComputationCandidateError::Unsupported(_))
+            error,
+            PerComputationCandidateError::TooManyCandidates {
+                max: MAX_CAPABILITY_SHARING_MEMBERS,
+                ..
+            }
         ));
+        // The message names the cap from its constant, not a copy of it.
+        assert!(error
+            .to_string()
+            .contains(&format!("more than {MAX_CAPABILITY_SHARING_MEMBERS} ")));
     }
 
     /// Equal targets give identical builds, which are the
diff --git a/crates/logical-asap-optimization/src/per_computation_candidates.rs b/crates/logical-asap-optimization/src/per_computation_candidates.rs
index 257d9464..3722ff69 100644
--- a/crates/logical-asap-optimization/src/per_computation_candidates.rs
+++ b/crates/logical-asap-optimization/src/per_computation_candidates.rs
@@ -426,6 +426,16 @@ pub enum PerComputationCandidateError {
     InvalidChoice,
     #[error("unsupported substitution in logical ASAP-aware optimization: {0}")]
     Unsupported(&'static str),
+    /// A Pass 2 rule found more than `max` candidates of one kind, whose
+    /// 2^n subsets it would enumerate: the workload is too large to plan
+    /// this way, not bad input.
+    #[error("more than {max} {candidates} (2^n Pass 2 candidates)")]
+    TooManyCandidates {
+        /// What was counted, e.g. "rollup candidates".
+        candidates: &'static str,
+        /// The cap that was exceeded.
+        max: usize,
+    },
 }
 
 /// Enumerate exact, rewrite and summary choices of one measure in stable catalog
diff --git a/crates/logical-asap-optimization/src/test_support.rs b/crates/logical-asap-optimization/src/test_support.rs
index 98e5e9bb..d598492f 100644
--- a/crates/logical-asap-optimization/src/test_support.rs
+++ b/crates/logical-asap-optimization/src/test_support.rs
@@ -274,11 +274,12 @@ pub fn entry_with(
     recurrence: asap_types::workload::QueryRecurrence,
 ) -> asap_types::workload::QueryWorkloadEntry {
     asap_types::workload::QueryWorkloadEntry {
+        query: Default::default(),
         requirements: QueryRequirements {
             accuracy,
             max_latency_ms: None,
         },
+        predictability: Default::default(),
         recurrence,
-        ..Default::default()
     }
 }
diff --git a/crates/logical-asap-optimization/src/workload_candidates.rs b/crates/logical-asap-optimization/src/workload_candidates.rs
index 540f051c..4f85af28 100644
--- a/crates/logical-asap-optimization/src/workload_candidates.rs
+++ b/crates/logical-asap-optimization/src/workload_candidates.rs
@@ -287,9 +287,10 @@ impl Pass2List {
         if candidates.len() + topk_reuse_candidates(roots).len() > MAX_REUSE_CANDIDATES {
             self.out.push(Pass2::failed(
                 rules,
-                PerComputationCandidateError::Unsupported(
-                    "more than 20 rollup and top-k reuse candidates together (2^n Pass 2 candidates)",
-                ),
+                PerComputationCandidateError::TooManyCandidates {
+                    candidates: "rollup and top-k reuse candidates together",
+                    max: MAX_REUSE_CANDIDATES,
+                },
             ));
             return;
         }
@@ -414,7 +415,20 @@ mod tests {
         };
         assert!(matches!(
             failed.roots,
-            Err(PerComputationCandidateError::Unsupported(_))
+            Err(PerComputationCandidateError::TooManyCandidates {
+                max: MAX_REUSE_CANDIDATES,
+                ..
+            })
         ));
+        let Err(error) = &failed.roots else {
+            unreachable!()
+        };
+        assert_eq!(
+            error.to_string(),
+            format!(
+                "more than {MAX_REUSE_CANDIDATES} rollup and top-k reuse candidates together \
+                 (2^n Pass 2 candidates)"
+            )
+        );
     }
 }
diff --git a/crates/types/src/ir/schema/sketch.rs b/crates/types/src/ir/schema/sketch.rs
index 8900e795..78c80569 100644
--- a/crates/types/src/ir/schema/sketch.rs
+++ b/crates/types/src/ir/schema/sketch.rs
@@ -727,6 +727,9 @@ pub enum NonNegativeWeightProof {
     /// gives it on confirming a
     /// [`WeightDomain::SamplesOfMetric`] requirement.
     CounterSamples,
+    /// A constant weight greater than 0, such as `Constant(2.0)`: every row
+    /// adds the same positive amount, so no weight is negative.
+    PositiveConstant,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
diff --git a/crates/types/src/workload/mod.rs b/crates/types/src/workload/mod.rs
index 9ad7e0a2..1a51c005 100644
--- a/crates/types/src/workload/mod.rs
+++ b/crates/types/src/workload/mod.rs
@@ -269,9 +269,7 @@ pub struct RepeatingEntry {
 /// as separate axes; the event time a query reads is in its text. Planning takes one per query, in
 /// root order: Stage 2 (physical ASAP-aware optimization) reads the
 /// recurrence and predictability, Stage 3 (plan selection) the accuracy
-/// target, the latency bound and the recurrence. The default is an empty
-/// query, exact, with no latency bound and unknown predictability and
-/// recurrence.
+/// target, the latency bound and the recurrence.
 ///
 /// # Example
 ///
@@ -287,7 +285,7 @@ pub struct RepeatingEntry {
 ///     ),
 /// };
 /// ```
-#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct QueryWorkloadEntry {
     /// The query text in the workload's language.
diff --git a/tools/dag-viewer/stages.js b/tools/dag-viewer/stages.js
index d6730dec..db5a0473 100644
--- a/tools/dag-viewer/stages.js
+++ b/tools/dag-viewer/stages.js
@@ -329,7 +329,9 @@ function stageCandidateRows(doc) {
   const order = { selected: 0, costlier: 1, rejected: 2, 'not judged': 2, pruned: 3, 'composition error': 4 };
   rows.sort((a, b) => order[a.status] - order[b.status] || (a.total ?? 0) - (b.total ?? 0));
   let rank = 0;
-  rows.forEach((row) => { if (row.total !== null) row.rank = ++rank; });
+  // Only valid candidates are ranked: the selected one and those that lost
+  // under the cost model, whether or not the document carries their cost.
+  rows.forEach((row) => { if (row.status === 'selected' || row.status === 'costlier') row.rank = ++rank; });
   return rows;
 }
 
diff --git a/tools/dag-viewer/test_viewer.py b/tools/dag-viewer/test_viewer.py
index 839d82b8..d76a39a9 100644
--- a/tools/dag-viewer/test_viewer.py
+++ b/tools/dag-viewer/test_viewer.py
@@ -180,8 +180,14 @@ def test_status_follows_the_rejection_cause_not_the_cost(self):
         lost = next(r for r in selection["rejected"] if r["cause"] == "costlier")
         selection["costs"][failed["id"]] = selection["costs"][lost["id"]]
         del selection["costs"][lost["id"]]
-        status = {row["id"]: row["status"] for row in self.js.call("stageCandidateRows", self.doc)}
+        rows = self.js.call("stageCandidateRows", self.doc)
+        status = {row["id"]: row["status"] for row in rows}
         self.assertEqual((status[failed["id"]], status[lost["id"]]), ("rejected", "costlier"))
+        # Ranks follow the status too: the costlier row is ranked without a
+        # cost, the rejected one is not ranked although it has one.
+        rank = {row["id"]: row["rank"] for row in rows}
+        self.assertIsNone(rank[failed["id"]])
+        self.assertIsInstance(rank[lost["id"]], int)
 
     def test_pruned_and_composition_errors_come_last_with_their_reasons(self):
         """A Stage 2 pruned candidate and a Stage 1 candidate that did not