diff --git a/crates/devtools/tests/stage_pipeline.rs b/crates/devtools/tests/stage_pipeline.rs
index 10321daf..5d50eef1 100644
--- a/crates/devtools/tests/stage_pipeline.rs
+++ b/crates/devtools/tests/stage_pipeline.rs
@@ -9,6 +9,7 @@ use asap_devtools::planner_layering_examples::{
 use asap_devtools::planning_stages::{
     lower_promql_roots, run_planning_stages, PlanningStagesError,
 };
+use asap_plan_selection::RejectionCause;
 use asap_types::ir::flat::FlatDag;
 use asap_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPDAGDocument};
 use asap_types::ir::QueryRoot;
@@ -135,14 +136,8 @@ fn example1_document_is_valid_and_committed_fixture_is_current() {
         assert!(!rejection["reason"].as_str().unwrap().is_empty());
         // A candidate that failed a check is not priced; one that lost
         // under the cost model is.
-        let failed_a_check = [
-            "failure_probability_resize_failed",
-            "accuracy_target_missed",
-            "raw_data_not_retained",
-            "latency_bound_exceeded",
-            "over_cluster_capacity",
-        ]
-        .contains(&rejection["cause"].as_str().unwrap());
+        let cause: RejectionCause = serde_json::from_value(rejection["cause"].clone()).unwrap();
+        let failed_a_check = cause.failed_a_check();
         assert_eq!(failed_a_check, !costs.contains_key(id), "{rejection}");
     }
     assert!(costs.contains_key(selected));
diff --git a/crates/executor/src/physical_planner/precompute.rs b/crates/executor/src/physical_planner/precompute.rs
index 7bcb3b20..8afaed70 100644
--- a/crates/executor/src/physical_planner/precompute.rs
+++ b/crates/executor/src/physical_planner/precompute.rs
@@ -82,8 +82,8 @@ pub fn raw_sample_row(
 }
 
 /// Input contract of a precompute boundary: raw sample rows for a raw time
-/// series scan (or a `TimeRange`, optionally over a `TimeShift`, of one),
-/// otherwise the stored population of its summary state.
+/// series scan (or a `TimeRange`, optionally over stacked `TimeShift`s, of
+/// one), otherwise the stored population of its summary state.
 pub fn boundary_schema(
     node: &PhysicalASAPDAGNode,
     nodes: &BTreeMap<u64, &PhysicalASAPDAGNode>,
@@ -93,7 +93,7 @@ pub fn boundary_schema(
         scan = nodes
             .get(&(*child as u64))
             .ok_or_else(|| invalid("missing precompute node"))?;
-        if let Payload::NonASAP(NonASAPOp::TimeShift { child, .. }) = &scan.payload {
+        while let Payload::NonASAP(NonASAPOp::TimeShift { child, .. }) = &scan.payload {
             scan = nodes
                 .get(&(*child as u64))
                 .ok_or_else(|| invalid("missing precompute node"))?;
diff --git a/crates/executor/tests/precompute.rs b/crates/executor/tests/precompute.rs
index 70b0359c..db790675 100644
--- a/crates/executor/tests/precompute.rs
+++ b/crates/executor/tests/precompute.rs
@@ -2,7 +2,7 @@
 mod common;
 use asap_executor::physical_planner::precompute::{boundary_schema, raw_sample_schema};
 use common::compile_physical_asap_dag;
-use planner_types::ir::operator::Source;
+use planner_types::ir::operator::{Source, TimeShift};
 use planner_types::ir::schema::{DataType, Field, Schema};
 use planner_types::ir::{NonASAPOp, Operator, OperatorNode, TimeRangeKind};
 use std::{collections::BTreeMap, rc::Rc, time::Duration};
@@ -53,3 +53,35 @@ fn only_a_time_series_range_reads_raw_samples() {
     // A table range is not raw samples: it must be a stored population.
     assert_ne!(table.ok(), Some(raw_sample_schema()));
 }
+
+/// A range over stacked time shifts of a time series scan still reads raw
+/// samples: the boundary looks through every shift, not only the first.
+#[test]
+fn a_range_over_stacked_shifts_reads_raw_samples() {
+    let range = time_range_over(Source::TimeSeries { metric: "m".into() });
+    let NonASAPOp::TimeRange {
+        range: window,
+        kind,
+        child: scan,
+    } = range.non_asap().unwrap().clone()
+    else {
+        unreachable!()
+    };
+    let shifted = |offset_ms, child| {
+        OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeShift {
+            shift: TimeShift {
+                offset_ms,
+                at: None,
+            },
+            child,
+        }))
+        .unwrap()
+    };
+    let stacked = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeRange {
+        range: window,
+        kind,
+        child: shifted(60_000, shifted(300_000, scan)),
+    }))
+    .unwrap();
+    assert_eq!(boundary(&stacked).unwrap(), raw_sample_schema());
+}
diff --git a/crates/logical-asap-optimization/Cargo.toml b/crates/logical-asap-optimization/Cargo.toml
index c7708aed..db0b9e26 100644
--- a/crates/logical-asap-optimization/Cargo.toml
+++ b/crates/logical-asap-optimization/Cargo.toml
@@ -5,8 +5,8 @@ edition = "2021"
 
 # #509 Stage 1, logical ASAP-aware optimization: Pass 1 per-computation
 # candidates and Pass 2 ASAP-aware CSE over asap-types' operator DAG, with
-# the analytical accuracy model. Depends on no front end and on no later
-# stage.
+# the analytical accuracy model. Depends on no later stage, and on no
+# query-language frontend except through the `test-support` feature.
 [dependencies]
 asap_sketchlib = { workspace = true }
 asap-types = { path = "../types" }
diff --git a/crates/logical-asap-optimization/src/accuracy/estimators/hll.rs b/crates/logical-asap-optimization/src/accuracy/estimators/hll.rs
index 76b59d83..083e43fc 100644
--- a/crates/logical-asap-optimization/src/accuracy/estimators/hll.rs
+++ b/crates/logical-asap-optimization/src/accuracy/estimators/hll.rs
@@ -47,14 +47,18 @@ pub fn classic_relative_error(precision: u8, relative_error: Option<f64>) -> f64
 /// `relative_error` over `source` is left as ([`classic_guarantee`]).
 /// `relative_error` is written in Rust's shortest round-trip form, so it
 /// parses back to the same value.
-pub fn classic_failure_statistic(precision: u8, relative_error: f64, source: &str) -> String {
+pub(crate) fn classic_failure_statistic(
+    precision: u8,
+    relative_error: f64,
+    source: &str,
+) -> String {
     format!(
         "{CLASSIC_FAILURE_PROBABILITY}[p={precision},eps={relative_error}]({MAX_DISTINCT}[{source}])"
     )
 }
 
-/// The precision, relative error and source a [`classic_failure_statistic`]
-/// names.
+/// The precision, relative error and source a classic HLL failure
+/// statistic (`classic_failure_statistic`) names.
 pub fn parse_classic_failure_statistic(statistic: &str) -> Option<(u8, f64, &str)> {
     let rest = statistic
         .strip_prefix(CLASSIC_FAILURE_PROBABILITY)?
@@ -112,7 +116,8 @@ pub(crate) fn classic_guarantee(
     ))
 }
 
-/// The failure probability a [`classic_failure_statistic`] names, given the
+/// The failure probability a classic HLL failure statistic
+/// (`classic_failure_statistic`) names, given the
 /// source's distinct bound: `1` (no certificate) when the bound is outside
 /// the contract's range for this precision and relative error.
 pub fn classic_failure_probability(precision: u8, relative_error: f64, max_distinct: u64) -> f64 {
diff --git a/crates/logical-asap-optimization/src/asap_aware_cse.rs b/crates/logical-asap-optimization/src/asap_aware_cse.rs
index c6cb094f..1e65cf35 100644
--- a/crates/logical-asap-optimization/src/asap_aware_cse.rs
+++ b/crates/logical-asap-optimization/src/asap_aware_cse.rs
@@ -393,24 +393,24 @@ pub fn capability_sharing_candidates(roots: &[QueryRoot]) -> Vec<CapabilityShari
             readers.entry(Rc::as_ptr(child)).or_default().push(node);
         }
     }
-    let mut groups: Vec<CapabilitySharingCandidate> = Vec::new();
+    let mut candidates: Vec<CapabilitySharingCandidate> = Vec::new();
     for node in &nodes {
         let Some(family) = shareable_family(node, readers.get(&Rc::as_ptr(node))) else {
             continue;
         };
-        match groups
+        match candidates
             .iter_mut()
-            .find(|g| g.family == family && same_input_data(&g.members[0], node))
+            .find(|c| c.family == family && same_input_data(&c.members[0], node))
         {
-            Some(group) => group.members.push(Rc::clone(node)),
-            None => groups.push(CapabilitySharingCandidate {
+            Some(candidate) => candidate.members.push(Rc::clone(node)),
+            None => candidates.push(CapabilitySharingCandidate {
                 family,
                 members: vec![Rc::clone(node)],
             }),
         }
     }
-    groups.retain(|g| !all_equal(g.members.iter()));
-    groups
+    candidates.retain(|c| !all_equal(c.members.iter()));
+    candidates
 }
 
 fn collect_children_first(
@@ -510,25 +510,28 @@ pub fn summary_capability_pass2_candidates(
     roots: &[QueryRoot],
     candidates: &[CapabilitySharingCandidate],
 ) -> Result<Vec<SummaryCapabilityPass2Candidate>, PerComputationCandidateError> {
-    // Every subset of a group is a candidate: 2^n of them. Past
-    // MAX_CAPABILITY_SHARING_MEMBERS that is no longer a list plan selection can price.
+    // Every subset of a capability sharing candidate is a Pass 2
+    // candidate: 2^n of them. Past MAX_CAPABILITY_SHARING_MEMBERS that is
+    // no longer a list plan selection can price.
     if candidates
         .iter()
-        .any(|group| group.members.len() > MAX_CAPABILITY_SHARING_MEMBERS)
+        .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)",
         ));
     }
-    // Each group's options: `None` (no sharing), then its subsets.
+    // Each capability sharing candidate's options: `None` (no sharing),
+    // then its subsets.
     let options: Vec<Vec<Option<Vec<usize>>>> = candidates
         .iter()
-        .map(|group| {
-            let n = group.members.len();
+        .map(|candidate| {
+            let n = candidate.members.len();
             let mut options = vec![None];
             for mask in 1u64..(1u64 << n) {
                 let members: Vec<usize> = (0..n).filter(|&m| mask >> m & 1 == 1).collect();
-                if members.len() >= 2 && !all_equal(members.iter().map(|&m| &group.members[m])) {
+                if members.len() >= 2 && !all_equal(members.iter().map(|&m| &candidate.members[m]))
+                {
                     options.push(Some(members));
                 }
             }
@@ -538,11 +541,11 @@ pub fn summary_capability_pass2_candidates(
     let mut choice = vec![0usize; candidates.len()];
     let mut pass2_candidates = Vec::new();
     loop {
-        // Mixed-radix increment, the last group varying fastest.
+        // Mixed-radix increment, the last candidate varying fastest.
         let mut carried = true;
-        for (digit, group_options) in choice.iter_mut().zip(&options).rev() {
+        for (digit, candidate_options) in choice.iter_mut().zip(&options).rev() {
             *digit += 1;
-            if *digit < group_options.len() {
+            if *digit < candidate_options.len() {
                 carried = false;
                 break;
             }
@@ -554,12 +557,12 @@ pub fn summary_capability_pass2_candidates(
         let shared_builds: Vec<Vec<Rc<OperatorNode>>> = choice
             .iter()
             .zip(candidates.iter().zip(&options))
-            .filter_map(|(&option, (group, group_options))| {
-                let members = group_options[option].as_ref()?;
+            .filter_map(|(&option, (candidate, candidate_options))| {
+                let members = candidate_options[option].as_ref()?;
                 Some(
                     members
                         .iter()
-                        .map(|&m| Rc::clone(&group.members[m]))
+                        .map(|&m| Rc::clone(&candidate.members[m]))
                         .collect(),
                 )
             })
diff --git a/crates/logical-asap-optimization/src/lib.rs b/crates/logical-asap-optimization/src/lib.rs
index 863171f5..e8cece06 100644
--- a/crates/logical-asap-optimization/src/lib.rs
+++ b/crates/logical-asap-optimization/src/lib.rs
@@ -7,10 +7,13 @@
 //! invisible at the logical level ([`asap_aware_cse`], Pass 2), e.g.
 //! `Quantile(x, 0.99)` and `Quantile(x, 0.95)` reading one built sketch.
 //!
-//! It depends only on `asap-types`, never on a front end or a later stage:
-//! Stage 2 (`asap-physical-asap-optimization`) and Stage 3
-//! (`asap-plan-selection`) depend on it, and Cargo enforces that one-way
-//! flow.
+//! It depends only on `asap-types`, never on a later stage: Stage 2
+//! (`asap-physical-asap-optimization`) and Stage 3 (`asap-plan-selection`)
+//! depend on it, and Cargo enforces that one-way flow. It depends on no
+//! query-language frontend either, except with the `test-support` feature,
+//! whose test helpers (`test_support`) lower PromQL test queries through
+//! `asap-frontend-promql`; the later stages enable that feature only for
+//! their tests.
 //!
 //! Post-lowering **canonicalization** is *not* here: it lives in
 //! `asap_types::ir::canonicalize`, run inside `asap_frontend_common`'s shared
diff --git a/crates/logical-asap-optimization/src/test_support.rs b/crates/logical-asap-optimization/src/test_support.rs
index 214581fb..98e5e9bb 100644
--- a/crates/logical-asap-optimization/src/test_support.rs
+++ b/crates/logical-asap-optimization/src/test_support.rs
@@ -11,6 +11,13 @@ use asap_types::workload::{
     QueryRequirements, QueryWorkload,
 };
 
+/// The statistic classic HLL's failure probability at `precision` and
+/// `relative_error` over `source` is left as, for Stage 3 tests that bind
+/// it. The estimator keeps the format crate-private.
+pub fn hll_classic_failure_statistic(precision: u8, relative_error: f64, source: &str) -> String {
+    crate::accuracy::estimators::hll::classic_failure_statistic(precision, relative_error, source)
+}
+
 pub fn lower_promql(query: &str, accuracy: AccuracyTarget) -> Rc<OperatorNode> {
     let workload = PlanningWorkload {
         query_workload: QueryWorkload {
diff --git a/crates/logical-asap-optimization/src/workload_candidates.rs b/crates/logical-asap-optimization/src/workload_candidates.rs
index e8451889..540f051c 100644
--- a/crates/logical-asap-optimization/src/workload_candidates.rs
+++ b/crates/logical-asap-optimization/src/workload_candidates.rs
@@ -8,7 +8,7 @@ use std::rc::Rc;
 
 use asap_types::ir::flat::flatten;
 use asap_types::ir::{OperatorNode, QueryRoot};
-use asap_types::workload::{QueryRecurrence, RepeatedDemand};
+use asap_types::workload::QueryRecurrence;
 
 use crate::asap_aware_cse::{
     capability_sharing_candidates, identical_expression_pass2_candidates, pane_candidates,
@@ -79,7 +79,10 @@ pub fn stage1_logical_asap_optimization<'a, Id: Clone>(
     per_computation: &'a PerComputationCandidates<Id>,
     recurrence: &[QueryRecurrence],
 ) -> impl Iterator<Item = Stage1Candidate> + 'a {
-    let interval_ms: Vec<Option<u64>> = recurrence.iter().map(fixed_interval_ms).collect();
+    let interval_ms: Vec<Option<u64>> = recurrence
+        .iter()
+        .map(QueryRecurrence::fixed_interval_ms)
+        .collect();
     let mut choice = Some(vec![0; per_computation.targets.len()]);
     let mut queue = std::collections::VecDeque::new();
     let mut position = 0usize;
@@ -355,17 +358,6 @@ fn next_choice<Id>(inventory: &PerComputationCandidates<Id>, choice: &mut [usize
     false
 }
 
-/// The fixed interval at which a query repeats, in milliseconds.
-pub fn fixed_interval_ms(recurrence: &QueryRecurrence) -> Option<u64> {
-    match recurrence {
-        QueryRecurrence::Repeated(
-            RepeatedDemand::FixedInterval(interval)
-            | RepeatedDemand::FixedIntervalAt { interval, .. },
-        ) => Some(u64::from(interval.0)).filter(|&ms| ms > 0),
-        _ => None,
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/crates/physical-asap-optimization/src/lib.rs b/crates/physical-asap-optimization/src/lib.rs
index 26ff57bd..806cdd18 100644
--- a/crates/physical-asap-optimization/src/lib.rs
+++ b/crates/physical-asap-optimization/src/lib.rs
@@ -93,7 +93,6 @@ use asap_types::workload::{
 use thiserror::Error;
 
 use asap_logical_asap_optimization::deployment_capabilities::DeploymentCapabilities;
-use asap_logical_asap_optimization::workload_candidates::fixed_interval_ms;
 
 /// One CandidatePhysicalASAPDAG (Stage 2 output), derived from exactly one
 /// CandidateLogicalASAPDAG (Stage 1 output). It holds planning data only;
@@ -241,7 +240,7 @@ pub(crate) fn builds_one_pane_per_evaluation<'a>(
     let mut any_reader = false;
     for recurrence in readers {
         any_reader = true;
-        if fixed_interval_ms(recurrence) != Some(pane_width_ms) {
+        if recurrence.fixed_interval_ms() != Some(pane_width_ms) {
             return false;
         }
         if let QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt {
diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs
index f81c871c..4c827270 100644
--- a/crates/plan-selection/src/lib.rs
+++ b/crates/plan-selection/src/lib.rs
@@ -571,7 +571,7 @@ pub struct Rejection {
 /// candidate failed a check and is invalid ([`Self::failed_a_check`]); the
 /// rest mean it was valid and lost under the cost model. JSON: the
 /// snake_case variant name, e.g. `"costlier"`.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum RejectionCause {
     /// Its summaries, resized for the queries' failure probabilities, do
@@ -2092,7 +2092,7 @@ fn price(
     let mut held: HashMap<PhysicalASAPNodeId, (u64, f64)> = HashMap::new();
     // The raw data query-time scans read, kept on disk: per source, the
     // largest scan's (bytes, node).
-    let mut raw_data: HashMap<String, (u64, PhysicalASAPNodeId)> = HashMap::new();
+    let mut raw_data: HashMap<&Source, (u64, PhysicalASAPNodeId)> = HashMap::new();
     // Each in-memory ingestion-time state's bytes per window (what it keeps
     // for one window, its output or an exact aggregate's values).
     let mut maintained: Vec<(PhysicalASAPNodeId, u64)> = Vec::new();
@@ -2554,9 +2554,7 @@ fn price(
         }
         if let Payload::NonASAP(NonASAPOp::Scan { source, .. }) = &node.payload {
             if !ingestion {
-                let largest = raw_data
-                    .entry(format!("{source:?}"))
-                    .or_insert((out.bytes, node.id));
+                let largest = raw_data.entry(source).or_insert((out.bytes, node.id));
                 if out.bytes > largest.0 {
                     *largest = (out.bytes, node.id);
                 }
@@ -3186,7 +3184,9 @@ mod tests {
     /// unknown.
     #[test]
     fn classic_hll_failure_probability_binds_only_its_own_sources_bound() {
-        let statistic = hll::classic_failure_statistic(15, 0.05, "a");
+        let statistic = asap_logical_asap_optimization::test_support::hll_classic_failure_statistic(
+            15, 0.05, "a",
+        );
         let bounded = |source: &str| {
             let mut data = data();
             data.sources
diff --git a/crates/plan-selection/src/measured_operation_costs.rs b/crates/plan-selection/src/measured_operation_costs.rs
index 2d8054bc..934d73a3 100644
--- a/crates/plan-selection/src/measured_operation_costs.rs
+++ b/crates/plan-selection/src/measured_operation_costs.rs
@@ -5,11 +5,11 @@
 //! with given parameters costs (e.g. sketch-bench's `rqe_atomic_costs.json`,
 //! converted by `asap-devtools`' `convert_rqe_atomic_costs`) and puts the
 //! table in [`PlanSelectionSettings::measured_operation_costs`]. Stage 3
-//! prices a sketch node (an operator node of the physical plan DAG) from
-//! the entry for its sketch parameters and operation when there is one, and
-//! from its formulas otherwise, and reports which per node
-//! ([`NodeCost::price_source`]). The planner reads
-//! only this table, never a benchmark file.
+//! prices a sketch node's (an operator node of the physical plan DAG) CPU
+//! from the entry for its sketch parameters and operation when there is
+//! one, and from its formulas otherwise, and reports which per node
+//! ([`NodeCost::price_source`]). The planner reads only this table, never a
+//! benchmark file.
 //!
 //! A measured time enters [`ResourceUsage`] as CPU operations: `t` ns is
 //! `t · 10⁻⁶ / ms_per_cpu_op` operations of [`ResourceSpeed`], so latency
diff --git a/crates/types/src/ir/operator/operator_properties.rs b/crates/types/src/ir/operator/operator_properties.rs
index 1403ca55..0dfcc9d7 100644
--- a/crates/types/src/ir/operator/operator_properties.rs
+++ b/crates/types/src/ir/operator/operator_properties.rs
@@ -178,7 +178,7 @@ pub enum DataModel {
 /// The leaf data source of a `Scan`. The schema itself rides on the
 /// `Scan.schema` field (SchemaResolver-built); `Source` carries only the leaf's
 /// identity.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub enum Source {
     /// Time-series leaf — PromQL / DC lifecycle. Produces `(ts, value, *labels)`.
     TimeSeries { metric: String },
diff --git a/crates/types/src/workload/mod.rs b/crates/types/src/workload/mod.rs
index 7a072e84..9ad7e0a2 100644
--- a/crates/types/src/workload/mod.rs
+++ b/crates/types/src/workload/mod.rs
@@ -197,6 +197,20 @@ pub enum QueryRecurrence {
     Unknown,
 }
 
+impl QueryRecurrence {
+    /// The fixed interval at which the query repeats, in milliseconds:
+    /// `Some` only for a fixed-interval repetition with a positive interval.
+    pub fn fixed_interval_ms(&self) -> Option<u64> {
+        match self {
+            QueryRecurrence::Repeated(
+                RepeatedDemand::FixedInterval(interval)
+                | RepeatedDemand::FixedIntervalAt { interval, .. },
+            ) => Some(u64::from(interval.0)).filter(|&ms| ms > 0),
+            _ => None,
+        }
+    }
+}
+
 /// One query submitted as part of a finite batch, including how many times it
 /// will run and, when known, its execution time.
 ///
@@ -666,6 +680,43 @@ pub enum WorkloadError {
 mod tests {
     use super::*;
 
+    /// Only a fixed-interval repetition with a positive interval has a
+    /// fixed interval; a zero interval, a schedule, an estimated rate, a
+    /// one-time query and an unknown recurrence have none.
+    #[test]
+    fn fixed_interval_ms_is_set_only_for_a_positive_fixed_interval() {
+        let repeated = |demand| QueryRecurrence::Repeated(demand);
+        assert_eq!(
+            repeated(RepeatedDemand::FixedInterval(RepetitionInterval(10_000))).fixed_interval_ms(),
+            Some(10_000)
+        );
+        assert_eq!(
+            repeated(RepeatedDemand::FixedIntervalAt {
+                interval: RepetitionInterval(60_000),
+                evaluation_phase: TimestampMs(5_000),
+            })
+            .fixed_interval_ms(),
+            Some(60_000)
+        );
+        assert_eq!(
+            repeated(RepeatedDemand::FixedInterval(RepetitionInterval(0))).fixed_interval_ms(),
+            None
+        );
+        assert_eq!(
+            repeated(RepeatedDemand::Scheduled(vec![TimestampMs(1_000)])).fixed_interval_ms(),
+            None
+        );
+        assert_eq!(
+            QueryRecurrence::OneTime {
+                invocations: 1,
+                execute_at: None,
+            }
+            .fixed_interval_ms(),
+            None
+        );
+        assert_eq!(QueryRecurrence::Unknown.fixed_interval_ms(), None);
+    }
+
     /// Query requirements: an accuracy target (omitted: exact) and an
     /// optional latency bound in ms (omitted: none). The old tagged forms
     /// and unknown fields are rejected.
diff --git a/tools/dag-viewer/stages.js b/tools/dag-viewer/stages.js
index c72c6be7..d6730dec 100644
--- a/tools/dag-viewer/stages.js
+++ b/tools/dag-viewer/stages.js
@@ -164,15 +164,19 @@ function uniqueIds(list, where, errors) {
   return ids;
 }
 
-// Every `stage3_plan_selection.rejected[].cause` Stage 3 writes
-// (`RejectionCause`): the first five are failed checks, the rest losses
-// under the cost model.
-const REJECTION_CAUSES = [
+// The `stage3_plan_selection.rejected[].cause` values (`RejectionCause`)
+// of a candidate that failed a check (`RejectionCause::failed_a_check`):
+// it is not priced.
+const FAILED_CHECK_CAUSES = [
   'failure_probability_resize_failed', 'accuracy_target_missed', 'raw_data_not_retained',
   'latency_bound_exceeded', 'over_cluster_capacity',
+];
+// Every `stage3_plan_selection.rejected[].cause` Stage 3 writes: the failed
+// checks, then the losses under the cost model (priced, not selected).
+const REJECTION_CAUSES = FAILED_CHECK_CAUSES.concat([
   'costlier', 'slower', 'more_memory', 'not_lowest_latency_on_pareto_front', 'dominated',
   'tied_lost_on_enumeration_order',
-];
+]);
 
 function validateSelection(selection, physicalById) {
   const errors = [];
@@ -299,9 +303,11 @@ function stageCandidateRows(doc) {
   const rows = (physical.candidates || []).map((candidate) => {
     const cost = costs[candidate.id];
     const rejection = rejected.get(candidate.id);
-    // A failed selection (`error`) judged no candidate.
+    // A rejection's cause says whether it failed a check or lost under
+    // the cost model; a failed selection (`error`) judged no candidate.
     const status = candidate.id === selection.selected ? 'selected'
-      : cost ? 'costlier' : rejection || !selection.error ? 'rejected' : 'not judged';
+      : rejection ? (FAILED_CHECK_CAUSES.includes(rejection.cause) ? 'rejected' : 'costlier')
+        : selection.error ? 'not judged' : 'rejected';
     return {
       id: candidate.id,
       logical: candidate.from_logical,
diff --git a/tools/dag-viewer/test_viewer.py b/tools/dag-viewer/test_viewer.py
index 1a2896fd..839d82b8 100644
--- a/tools/dag-viewer/test_viewer.py
+++ b/tools/dag-viewer/test_viewer.py
@@ -161,7 +161,7 @@ def test_selected_first_then_costlier_by_cost_then_rejected(self):
         costlier = [row for row in rows if row["status"] == "costlier"]
         self.assertEqual([r["total"] for r in costlier], sorted(r["total"] for r in costlier))
         self.assertEqual([r["rank"] for r in costlier], list(range(2, len(costlier) + 2)))
-        failed_a_check = {"failure_probability_resize_failed", "accuracy_target_missed", "raw_data_not_retained", "latency_bound_exceeded", "over_cluster_capacity"}
+        failed_a_check = set(json.loads(self.js.eval("JSON.stringify(FAILED_CHECK_CAUSES)")))
         rejected = {r["id"]: r["reason"] for r in selection["rejected"] if r["cause"] in failed_a_check}
         self.assertTrue(rejected)
         for row in rows:
@@ -171,6 +171,18 @@ def test_selected_first_then_costlier_by_cost_then_rejected(self):
         self.assertEqual(len(rows), len(physical))
         self.assertTrue(all(row["display_name"] and row["logical"] for row in rows))
 
+    def test_status_follows_the_rejection_cause_not_the_cost(self):
+        """A candidate's row says "costlier" when it lost under the cost
+        model and "rejected" when it failed a check, whether or not the
+        document carries its cost."""
+        selection = self.doc["stage3_plan_selection"]
+        failed = next(r for r in selection["rejected"] if r["cause"] == "accuracy_target_missed")
+        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)}
+        self.assertEqual((status[failed["id"]], status[lost["id"]]), ("rejected", "costlier"))
+
     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
         compose are rows too, after the Stage 3 ones."""