diff --git a/crates/devtools/src/bin/stage_pipeline.rs b/crates/devtools/src/bin/stage_pipeline.rs
index f586c982..ac9335ca 100644
--- a/crates/devtools/src/bin/stage_pipeline.rs
+++ b/crates/devtools/src/bin/stage_pipeline.rs
@@ -4,6 +4,11 @@
 // planner-layering-3a and planner-layering-3b: #509 Example 3, Patterns A
 // and B; 3a has 972 candidates, all priced, 64 shown by default)
 // cargo run -p asap-devtools --bin stage_pipeline -- \
+//     --workload dashboard.json --out run.json
+// (a `PlanningWorkload` JSON file, e.g. with a repeating query's
+// `"range_evaluation": {"step_ms": 60000, "span_ms": 3600000}`; PromQL, or
+// SQL over `--table` declarations)
+// cargo run -p asap-devtools --bin stage_pipeline -- \
 //     --promql "topk by (job) (10, rate(x[1m]))" --max-error 0.01 --failure-probability 0.001 --out run.json
 // cargo run -p asap-devtools --bin stage_pipeline -- \
 //     --table '{"name": "metrics", "columns": [{"name": "ts", "type": "timestamp"},
@@ -47,7 +52,10 @@
 //     once if materialized, else once per query (or materialized node)
 //     that reads it, so a shared NotMaterialized node is charged once per
 //     query. When selection fails, `selected` is null, with the rejections,
-//     or with `error` for any other failure.
+//     or with `error` for any other failure. `pane_retention_ms` maps each
+//     pane chain the selected plan keeps beyond one evaluation (by node id)
+//     to how far back from the newest evaluation the runtime must keep its
+//     panes: the window plus the longest range-evaluation span reading it.
 // When Pass 1 fails, the document still has the Stage 0 DAG; Stage 1 has
 // `error` and no candidates, and nothing is selected.
 //
@@ -223,6 +231,7 @@ fn document_of(
 
 fn run(args: Vec<String>) -> Result<(), Failure> {
     let mut example = None;
+    let mut workload_path: Option<String> = None;
     // (language, text, name) in workload order.
     let mut queries: Vec<(QueryLanguage, String, Option<String>)> = Vec::new();
     let mut tables = Vec::new();
@@ -242,6 +251,7 @@ fn run(args: Vec<String>) -> Result<(), Failure> {
         let number = |v: String| v.parse::<f64>().map_err(|e| format!("{v}: {e}"));
         match flag.as_str() {
             "--example" => example = Some(value()?),
+            "--workload" => workload_path = Some(value()?),
             "--promql" => queries.push((QueryLanguage::PromQL, value()?, None)),
             "--sql" => queries.push((
                 QueryLanguage::SQL(SqlDialect::DataFusionSQL),
@@ -295,7 +305,7 @@ fn run(args: Vec<String>) -> Result<(), Failure> {
     let has_sql = queries
         .iter()
         .any(|(language, _, _)| *language != QueryLanguage::PromQL);
-    if !tables.is_empty() && !has_sql {
+    if !tables.is_empty() && !has_sql && workload_path.is_none() {
         return Err("--table needs --sql".into());
     }
     let out = out.ok_or("--out is required")?;
@@ -355,7 +365,43 @@ fn run(args: Vec<String>) -> Result<(), Failure> {
         measured_operation_costs,
         ..default_settings
     };
+    let from_file = match &workload_path {
+        Some(path) if example.is_none() && queries.is_empty() => {
+            let text =
+                std::fs::read_to_string(path).map_err(|e| format!("--workload {path}: {e}"))?;
+            let workload: PlanningWorkload =
+                serde_json::from_str(&text).map_err(|e| format!("--workload {path}: {e}"))?;
+            workload
+                .validate()
+                .map_err(|e| format!("--workload {path}: {e}"))?;
+            Some((workload, sql_catalog(&tables)?))
+        }
+        Some(_) => {
+            return Err(
+                "give exactly one of --example, --workload or --promql/--sql queries".into(),
+            )
+        }
+        None => None,
+    };
     let mut document = match (example.as_deref(), queries.is_empty()) {
+        (None, true) if from_file.is_some() => {
+            let (workload, catalog) = from_file.expect("checked");
+            let roots = match workload.query_workload.language {
+                QueryLanguage::PromQL => lower_promql_roots(&workload)?,
+                _ => lower_sql_roots(&workload, &catalog)?,
+            };
+            let entries: Vec<_> = workload.query_workload.entries().collect();
+            let data = workload.data_workload.clone().unwrap_or_default();
+            let result = run_planning_stages_over(
+                roots.clone(),
+                &entries,
+                &data,
+                Some(max_combinations),
+                &plan_selection,
+                &deployment_capabilities,
+            );
+            document_of(workload_queries(&workload), &roots, result, max_displayed)?
+        }
         (Some(example), true) => {
             let (workload, catalog) = match example {
                 "planner-layering-1" => (example1(), SqlCatalog::new()),
@@ -458,7 +504,11 @@ fn run(args: Vec<String>) -> Result<(), Failure> {
             );
             document_of(described, &roots, result, max_displayed)?
         }
-        _ => return Err("give exactly one of --example or --promql/--sql queries".into()),
+        _ => {
+            return Err(
+                "give exactly one of --example, --workload or --promql/--sql queries".into(),
+            )
+        }
     };
     // Read-only: the deployment inputs the run planned with.
     document["deployment_capabilities"] =
@@ -647,6 +697,11 @@ fn plan_selection_json(
         "selected": selection.selected,
         "rejected": rejections_json(&selection.rejected),
         "front": selection.front,
+        "pane_retention_ms": selection
+            .pane_retention_ms
+            .iter()
+            .map(|(node, ms)| (node.to_string(), json!(ms)))
+            .collect::<serde_json::Map<_, _>>(),
     })
 }
 
@@ -699,6 +754,9 @@ fn workload_queries(workload: &PlanningWorkload) -> Vec<Value> {
             {
                 requirements["repeat_interval_ms"] = json!(interval.0);
             }
+            if let Some(range) = entry.range_evaluation {
+                requirements["range_evaluation"] = json!(range);
+            }
             json!({
                 "id": format!("q{}", index + 1),
                 "language": language,
@@ -810,6 +868,7 @@ fn query_batch(
                         predictability: Default::default(),
                         invocations: 1,
                         execute_at: None,
+                        range_evaluation: None,
                     })
                     .collect(),
             ),
diff --git a/crates/devtools/src/candidate_labels.rs b/crates/devtools/src/candidate_labels.rs
index 9af35316..2ff21c82 100644
--- a/crates/devtools/src/candidate_labels.rs
+++ b/crates/devtools/src/candidate_labels.rs
@@ -485,6 +485,7 @@ mod tests {
                 max_latency_ms: None,
             },
             predictability: Predictability::Predictable { known_at: None },
+            range_evaluation: None,
         };
         let workload = PlanningWorkload {
             query_workload: QueryWorkload {
@@ -551,6 +552,7 @@ mod tests {
                             max_latency_ms: None,
                         },
                         predictability: Predictability::Predictable { known_at: None },
+                        range_evaluation: None,
                     }]),
                 },
                 data_workload: Some(shared_data_workload(
diff --git a/crates/devtools/src/lib.rs b/crates/devtools/src/lib.rs
index 85f20653..8086c513 100644
--- a/crates/devtools/src/lib.rs
+++ b/crates/devtools/src/lib.rs
@@ -55,6 +55,7 @@ pub fn lower_promql_with_sample_interval(
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },
diff --git a/crates/devtools/src/planner_layering_examples.rs b/crates/devtools/src/planner_layering_examples.rs
index a584dcab..047a6145 100644
--- a/crates/devtools/src/planner_layering_examples.rs
+++ b/crates/devtools/src/planner_layering_examples.rs
@@ -53,6 +53,7 @@ pub fn example1() -> PlanningWorkload {
             max_latency_ms,
         },
         predictability: Predictability::Predictable { known_at: None },
+        range_evaluation: None,
     };
     PlanningWorkload {
         query_workload: QueryWorkload {
@@ -130,6 +131,7 @@ pub fn example2() -> (PlanningWorkload, asap_frontend_sql::SqlCatalog) {
                             max_latency_ms: None,
                         },
                         predictability: Predictability::Predictable { known_at: None },
+                        range_evaluation: None,
                     })
                     .collect(),
             ),
@@ -239,6 +241,7 @@ pub fn pattern_a(recurrence: PatternARecurrence, arrival: DataArrival) -> Planni
                         predictability: Predictability::AdHoc,
                         invocations: 1,
                         execute_at: Some(TimestampMs(T_MS)),
+                        range_evaluation: None,
                     })
                     .collect(),
             ),
@@ -258,6 +261,7 @@ pub fn pattern_a(recurrence: PatternARecurrence, arrival: DataArrival) -> Planni
                         predictability: Predictability::Predictable {
                             known_at: Some(TimestampMs(T_MS)),
                         },
+                        range_evaluation: None,
                     })
                     .collect(),
             ),
@@ -295,6 +299,7 @@ pub fn pattern_b() -> PlanningWorkload {
                     max_latency_ms: Some(200.0),
                 },
                 predictability: Predictability::Predictable { known_at: None },
+                range_evaluation: None,
             }]),
         },
         data_workload: Some(shared_data_workload(
diff --git a/crates/devtools/src/planning_stages.rs b/crates/devtools/src/planning_stages.rs
index b94136df..421c47de 100644
--- a/crates/devtools/src/planning_stages.rs
+++ b/crates/devtools/src/planning_stages.rs
@@ -173,9 +173,8 @@ pub fn run_planning_stages_over(
             .without_unsupported_summaries(deployment);
     // Count Stage 1 candidates as they are enumerated and stop as soon as
     // the count passes the limit; never truncate.
-    let recurrence: Vec<_> = entries.iter().map(|e| e.recurrence.clone()).collect();
     let mut stage1 = Vec::new();
-    for candidate in stage1_logical_asap_optimization(&per_computation, &recurrence) {
+    for candidate in stage1_logical_asap_optimization(&per_computation, entries) {
         if max_combinations.is_some_and(|limit| stage1.len() == limit) {
             return Err(PlanningStagesError::TooManyCandidates {
                 limit: stage1.len(),
diff --git a/crates/devtools/tests/stage_pipeline.rs b/crates/devtools/tests/stage_pipeline.rs
index e7a4d514..d1b6fb19 100644
--- a/crates/devtools/tests/stage_pipeline.rs
+++ b/crates/devtools/tests/stage_pipeline.rs
@@ -885,3 +885,67 @@ fn logical_candidate_labels_are_distinct() {
     assert_eq!(labels.len(), 9, "{labels:#?}");
     assert_eq!(distinct.len(), 9, "{labels:#?}");
 }
+
+/// `--workload` plans a `PlanningWorkload` JSON file. A dashboard panel
+/// (`max_over_time(m[5m])` every minute over 1M series) answered as the
+/// last hour at 1-minute steps (`range_evaluation`) is priced as 61
+/// evaluations per request: the kept panes (`P2-m2`, 70.51 cost/s) beat
+/// rebuilding the window at every step (`P1`, 127.08), and the document
+/// reports their retention, 1 h + 5 min. The workload section shows the
+/// range. An invalid range is refused.
+#[test]
+fn workload_file_with_a_range_evaluation_is_planned() {
+    let workload = serde_json::json!({
+        "query_workload": {
+            "language": "promql",
+            "query_batch": null,
+            "repeating_queries": [{
+                "query": "max_over_time(m[5m])",
+                "demand": {"fixed_interval": 60000},
+                "requirements": {},
+                "predictability": {"predictable": {"known_at": null}},
+                "range_evaluation": {"step_ms": 60000, "span_ms": 3600000},
+            }],
+        },
+        "data_workload": {
+            "arrival": {"ingesting": {"rows_per_second": 66666.67}},
+            "sources": {"m": {"series_count": 1000000, "sample_interval": 15000}},
+        },
+    });
+    let path = std::env::temp_dir().join(format!(
+        "stage_pipeline_{}_range_workload.json",
+        std::process::id()
+    ));
+    std::fs::write(&path, workload.to_string()).unwrap();
+    let document = document_of(&[
+        "--workload",
+        path.to_str().unwrap(),
+        "--deployment-capabilities",
+        r#"{"query_time_results_retained": true}"#,
+    ]);
+    assert_eq!(
+        document["workload"]["queries"][0]["requirements"]["range_evaluation"],
+        serde_json::json!({"step_ms": 60000, "span_ms": 3600000})
+    );
+    let selection = &document["stage3_plan_selection"];
+    assert_eq!(selection["selected"], "P2-m2");
+    let cost = |id: &str| selection["costs"][id]["cost_per_second"].as_f64().unwrap();
+    assert!((cost("P2-m2") - 70.5133).abs() < 1e-3, "{}", cost("P2-m2"));
+    assert!((cost("P1") - 127.0833).abs() < 1e-3, "{}", cost("P1"));
+    assert_eq!(
+        selection["pane_retention_ms"],
+        serde_json::json!({"1": 3_900_000})
+    );
+
+    let mut invalid = workload.clone();
+    invalid["query_workload"]["repeating_queries"][0]["range_evaluation"]["step_ms"] =
+        serde_json::json!(0);
+    std::fs::write(&path, invalid.to_string()).unwrap();
+    let (status, stderr) = failure(&["--workload", path.to_str().unwrap()]);
+    assert_ne!(status, Some(0));
+    assert!(
+        stderr.contains("step_ms must be greater than zero"),
+        "{stderr}"
+    );
+    std::fs::remove_file(path).unwrap();
+}
diff --git a/crates/executor/src/physical_planner/candidates.rs b/crates/executor/src/physical_planner/candidates.rs
index 141cb850..57eba3d4 100644
--- a/crates/executor/src/physical_planner/candidates.rs
+++ b/crates/executor/src/physical_planner/candidates.rs
@@ -356,6 +356,7 @@ mod tests {
                     predictability: Predictability::Unknown,
                     invocations: 1,
                     execute_at: None,
+                    range_evaluation: None,
                 }]),
                 repeating_queries: None,
             },
diff --git a/crates/executor/tests/common/mod.rs b/crates/executor/tests/common/mod.rs
index a4feec86..87472344 100644
--- a/crates/executor/tests/common/mod.rs
+++ b/crates/executor/tests/common/mod.rs
@@ -75,6 +75,7 @@ fn run_stage1_to_stage3(
             },
             predictability: Default::default(),
             recurrence: Default::default(),
+            range_evaluation: None,
         }],
         &data,
         asap_plan_selection::PlanningModels::builtin(),
diff --git a/crates/executor/tests/current_series_heap.rs b/crates/executor/tests/current_series_heap.rs
index 09d1abfa..3d3e7725 100644
--- a/crates/executor/tests/current_series_heap.rs
+++ b/crates/executor/tests/current_series_heap.rs
@@ -318,6 +318,7 @@ fn planner_current_series_candidate_compiles_with_dynamic_identity() {
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },
diff --git a/crates/executor/tests/deployment_computation.rs b/crates/executor/tests/deployment_computation.rs
index 6fe112fd..f5649bf7 100644
--- a/crates/executor/tests/deployment_computation.rs
+++ b/crates/executor/tests/deployment_computation.rs
@@ -33,6 +33,7 @@ fn lower_with(query: &str, accuracy: AccuracyTarget) -> Rc<planner_types::ir::Op
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },
diff --git a/crates/executor/tests/planspace_series_identity_heap.rs b/crates/executor/tests/planspace_series_identity_heap.rs
index c3271e5c..e1d2cdbb 100644
--- a/crates/executor/tests/planspace_series_identity_heap.rs
+++ b/crates/executor/tests/planspace_series_identity_heap.rs
@@ -26,6 +26,7 @@ fn lower(query: &str, accuracy: &AccuracyTarget) -> Rc<OperatorNode> {
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },
diff --git a/crates/executor/tests/precompute_candidates.rs b/crates/executor/tests/precompute_candidates.rs
index ba88821d..c242a4ec 100644
--- a/crates/executor/tests/precompute_candidates.rs
+++ b/crates/executor/tests/precompute_candidates.rs
@@ -39,6 +39,7 @@ fn exact_promql_root(query: &str) -> std::rc::Rc<planner_types::ir::OperatorNode
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },
@@ -632,6 +633,7 @@ fn population_topk_cuts_equal_per_frontier_compilation() {
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },
diff --git a/crates/executor/tests/weighted_topk_binding.rs b/crates/executor/tests/weighted_topk_binding.rs
index a0f271f5..62c81f19 100644
--- a/crates/executor/tests/weighted_topk_binding.rs
+++ b/crates/executor/tests/weighted_topk_binding.rs
@@ -196,6 +196,7 @@ pub fn lower_promql(
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },
diff --git a/crates/frontend-metricsql/tests/lowering_semantics.rs b/crates/frontend-metricsql/tests/lowering_semantics.rs
index 3a2e28a5..37f0dddd 100644
--- a/crates/frontend-metricsql/tests/lowering_semantics.rs
+++ b/crates/frontend-metricsql/tests/lowering_semantics.rs
@@ -29,6 +29,7 @@ fn workload(language: QueryLanguage, query: &str) -> PlanningWorkload {
                 predictability: Default::default(),
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },
diff --git a/crates/frontend-promql/src/lib.rs b/crates/frontend-promql/src/lib.rs
index c4ec52e8..ea794b82 100644
--- a/crates/frontend-promql/src/lib.rs
+++ b/crates/frontend-promql/src/lib.rs
@@ -139,6 +139,7 @@ mod tests {
                     predictability: Default::default(),
                     invocations: 1,
                     execute_at: None,
+                    range_evaluation: None,
                 }]),
                 repeating_queries: None,
             },
diff --git a/crates/frontend-promql/tests/promql_lowering.rs b/crates/frontend-promql/tests/promql_lowering.rs
index 1569cbdd..6c0f39eb 100644
--- a/crates/frontend-promql/tests/promql_lowering.rs
+++ b/crates/frontend-promql/tests/promql_lowering.rs
@@ -1173,6 +1173,7 @@ fn batch_lowers_each_entry_and_reads_per_query_accuracy() {
                     predictability: Predictability::Unknown,
                     invocations: 1,
                     execute_at: None,
+                    range_evaluation: None,
                 },
                 BatchEntry {
                     query: Query("quantile_over_time(0.9, b[5m])".into()),
@@ -1183,6 +1184,7 @@ fn batch_lowers_each_entry_and_reads_per_query_accuracy() {
                     predictability: Predictability::Unknown,
                     invocations: 1,
                     execute_at: None,
+                    range_evaluation: None,
                 },
             ]),
             repeating_queries: None,
@@ -1206,6 +1208,7 @@ fn batch_rejects_non_promql_language() {
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },
diff --git a/crates/frontend-promql/tests/support.rs b/crates/frontend-promql/tests/support.rs
index 23b0ea0f..38d2182d 100644
--- a/crates/frontend-promql/tests/support.rs
+++ b/crates/frontend-promql/tests/support.rs
@@ -24,6 +24,7 @@ pub fn workload(query: &str, accuracy: AccuracyTarget) -> PlanningWorkload {
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },
@@ -123,6 +124,7 @@ pub fn run_stage1_to_stage3(
             },
             predictability: Default::default(),
             recurrence: Default::default(),
+            range_evaluation: None,
         }],
         &DataWorkload::default(),
         asap_plan_selection::PlanningModels::builtin(),
@@ -182,6 +184,7 @@ pub fn logical_plan(root: &Rc<OperatorNode>) -> Result<Rc<OperatorNode>, String>
             },
             predictability: Default::default(),
             recurrence: Default::default(),
+            range_evaluation: None,
         }],
         &DataWorkload::default(),
         asap_plan_selection::PlanningModels::builtin(),
diff --git a/crates/frontend-sql/tests/frequency_l2.rs b/crates/frontend-sql/tests/frequency_l2.rs
index 604af0a3..5254e331 100644
--- a/crates/frontend-sql/tests/frequency_l2.rs
+++ b/crates/frontend-sql/tests/frequency_l2.rs
@@ -137,6 +137,7 @@ async fn batch_lowering_recognizes_frequency_l2() {
             predictability: Predictability::Unknown,
             invocations: 1,
             execute_at: None,
+            range_evaluation: None,
         }]),
         repeating_queries: None,
     };
diff --git a/crates/frontend-sql/tests/pearson_corr.rs b/crates/frontend-sql/tests/pearson_corr.rs
index f082277e..3fe2b9c9 100644
--- a/crates/frontend-sql/tests/pearson_corr.rs
+++ b/crates/frontend-sql/tests/pearson_corr.rs
@@ -159,6 +159,7 @@ async fn corr_survives_exact_plan_compilation() {
             },
             predictability: Default::default(),
             recurrence: Default::default(),
+            range_evaluation: None,
         }],
         &Default::default(),
         asap_plan_selection::PlanningModels::builtin(),
diff --git a/crates/integration-tests/src/lib.rs b/crates/integration-tests/src/lib.rs
index df9f6142..23d8c66e 100644
--- a/crates/integration-tests/src/lib.rs
+++ b/crates/integration-tests/src/lib.rs
@@ -52,6 +52,7 @@ pub mod fixtures {
                     predictability: Predictability::Unknown,
                     invocations: 1,
                     execute_at: None,
+                    range_evaluation: None,
                 }]),
                 repeating_queries: None,
             },
@@ -161,6 +162,7 @@ pub mod planning {
                 },
                 predictability: Default::default(),
                 recurrence: Default::default(),
+                range_evaluation: None,
             }],
             &DataWorkload::default(),
             asap_plan_selection::PlanningModels::builtin(),
diff --git a/crates/integration-tests/tests/operator_design_examples.rs b/crates/integration-tests/tests/operator_design_examples.rs
index ac07a786..5fa8ff8c 100644
--- a/crates/integration-tests/tests/operator_design_examples.rs
+++ b/crates/integration-tests/tests/operator_design_examples.rs
@@ -291,6 +291,7 @@ async fn batch_planning_replaces_summary_operators() {
                         predictability: Predictability::Unknown,
                         invocations: 2,
                         execute_at: None,
+                        range_evaluation: None,
                     })
                     .collect(),
             ),
diff --git a/crates/integration-tests/tests/planner_layering_example1.rs b/crates/integration-tests/tests/planner_layering_example1.rs
index 05481ed8..3db57eb6 100644
--- a/crates/integration-tests/tests/planner_layering_example1.rs
+++ b/crates/integration-tests/tests/planner_layering_example1.rs
@@ -138,6 +138,7 @@ fn exact_every_10s() -> asap_types::workload::QueryWorkloadEntry {
         requirements: Default::default(),
         predictability: Default::default(),
         recurrence: every_10s(),
+        range_evaluation: None,
     }
 }
 
diff --git a/crates/integration-tests/tests/planner_layering_example2.rs b/crates/integration-tests/tests/planner_layering_example2.rs
index 0a6be821..899ab86b 100644
--- a/crates/integration-tests/tests/planner_layering_example2.rs
+++ b/crates/integration-tests/tests/planner_layering_example2.rs
@@ -66,6 +66,7 @@ fn panel(query: &str, (epsilon, delta): (f64, f64), scope: AccuracyScope) -> Rep
             max_latency_ms: None,
         },
         predictability: Predictability::Predictable { known_at: None },
+        range_evaluation: None,
     }
 }
 
diff --git a/crates/integration-tests/tests/promql_to_post_asap.rs b/crates/integration-tests/tests/promql_to_post_asap.rs
index 0a8f74c3..f8666f78 100644
--- a/crates/integration-tests/tests/promql_to_post_asap.rs
+++ b/crates/integration-tests/tests/promql_to_post_asap.rs
@@ -716,6 +716,7 @@ fn ddsketch_ratio_selection(
             },
             predictability: Default::default(),
             recurrence: Default::default(),
+            range_evaluation: None,
         }],
         &DataWorkload::default(),
         PlanningModels::builtin().with_evidence(evidence),
diff --git a/crates/integration-tests/tests/range_evaluation.rs b/crates/integration-tests/tests/range_evaluation.rs
new file mode 100644
index 00000000..a161e165
--- /dev/null
+++ b/crates/integration-tests/tests/range_evaluation.rs
@@ -0,0 +1,293 @@
+//! #757: a query answered as a range request (`RangeEvaluation`), e.g. a
+//! dashboard panel that evaluates the last hour at 1-minute steps on every
+//! refresh. Stage 1 and Stage 2 keep pane widths on the step grid, and
+//! Stage 3 prices `span_ms / step_ms + 1` evaluations per request, credits
+//! the panes kept across steps and requests, and retains kept panes for
+//! the span plus the window.
+
+mod planner_layering_common;
+
+use asap_logical_asap_optimization::DeploymentCapabilities;
+use asap_plan_selection::RejectionCause;
+use asap_types::ir::physical_export::PhysicalASAPOperatorPayload;
+use asap_types::ir::ASAPOp;
+use asap_types::physical::{Materialization, Medium};
+use asap_types::workload::{
+    DataArrival, DataWorkload, DurationMs, PlanningWorkload, Predictability, Query, QueryLanguage,
+    QueryRequirements, QueryWorkload, RangeEvaluation, RepeatedDemand, RepeatingEntry,
+    RepetitionInterval, SourceFacts,
+};
+use planner_layering_common::*;
+
+/// The dashboard's range: the last hour at 1-minute steps, 61 evaluations
+/// per request.
+const LAST_HOUR_BY_MINUTE: RangeEvaluation = RangeEvaluation {
+    step_ms: 60_000,
+    span_ms: 3_600_000,
+};
+
+/// Pattern B (#509 Example 3: `quantile_over_time(0.99, latency_ms[5m])`
+/// every minute over 1M series) with each request answered as `range`.
+fn pattern_b_with(range: Option<RangeEvaluation>) -> PlanningWorkload {
+    let mut workload = pattern_b();
+    for entry in workload
+        .query_workload
+        .repeating_queries
+        .iter_mut()
+        .flatten()
+    {
+        entry.range_evaluation = range;
+    }
+    workload
+}
+
+/// A deployment that keeps query-time results across evaluations (B3).
+fn keeping_query_time_results() -> DeploymentCapabilities {
+    DeploymentCapabilities {
+        query_time_results_retained: true,
+        ..DeploymentCapabilities::default()
+    }
+}
+
+/// `query` every `interval_ms` over 1M series of `m` sampled every 15 s,
+/// arriving continuously, exact, answered as `range`.
+fn exact_panel(query: &str, interval_ms: u32, range: Option<RangeEvaluation>) -> PlanningWorkload {
+    PlanningWorkload {
+        query_workload: QueryWorkload {
+            language: QueryLanguage::PromQL,
+            query_batch: None,
+            repeating_queries: Some(vec![RepeatingEntry {
+                query: Query(query.into()),
+                demand: RepeatedDemand::FixedInterval(RepetitionInterval(interval_ms)),
+                requirements: QueryRequirements::default(),
+                predictability: Predictability::Predictable { known_at: None },
+                range_evaluation: range,
+            }]),
+        },
+        data_workload: Some(DataWorkload {
+            arrival: DataArrival::Ingesting {
+                rows_per_second: SHARED_ROWS_PER_SECOND,
+            },
+            sources: [(
+                "m".to_string(),
+                SourceFacts {
+                    series_count: Some(1_000_000),
+                    sample_interval: Some(DurationMs(15_000)),
+                    ..SourceFacts::default()
+                },
+            )]
+            .into(),
+        }),
+    }
+}
+
+fn cost(run: &Run, id: &str) -> f64 {
+    run.cost(id).unwrap_or_else(|| panic!("{id} is not priced"))
+}
+
+fn close(actual: f64, expected: f64) -> bool {
+    (actual - expected).abs() <= 1e-6 * expected.abs().max(1.0)
+}
+
+/// Each physical candidate's pane chains: (pane width, materialization).
+fn pane_chains(run: &Run) -> Vec<(String, u64, Materialization)> {
+    run.physical
+        .iter()
+        .flat_map(|p| {
+            p.dag().nodes.iter().filter_map(|n| match &n.payload {
+                PhysicalASAPOperatorPayload::ASAP(ASAPOp::PaneChain { pane_width_ms, .. }) => {
+                    Some((p.id.clone(), *pane_width_ms, n.materialization))
+                }
+                _ => None,
+            })
+        })
+        .collect()
+}
+
+/// The candidate whose single pane chain is materialized as `wanted`.
+fn with_chain(run: &Run, wanted: Materialization) -> String {
+    pane_chains(run)
+        .into_iter()
+        .find(|(_, _, m)| *m == wanted)
+        .unwrap_or_else(|| panic!("no candidate with a {wanted:?} chain"))
+        .0
+}
+
+/// (a) The issue's acceptance example: Pattern B as a dashboard over the
+/// last hour at 1-minute steps prices 61 evaluations per request. Every
+/// query-time computation runs at every step, so the whole-window KLL
+/// (2.1 → 128.1 cost/s), the panes rebuilt per step (2.18333 → 133.18333)
+/// and the exact plan (2.4 → 146.4) cost 61 times as much. The kept panes
+/// (B3) are built once per request (one new 1-min pane) but merged at every
+/// step, and keep 64 older panes instead of 4 (1 h + 5 min of panes): 8,884.53
+/// → 142,151.53. B1 keeps 66 instead of 6: 13,326.53 → 146,593.53. The
+/// whole-window KLL is still selected.
+#[test]
+fn stage3_plan_selection_prices_61_evaluations_per_dashboard_request() {
+    let deployment = keeping_query_time_results();
+    let none = run_promql_for(&pattern_b_with(None), &deployment);
+    let range = run_promql_for(&pattern_b_with(Some(LAST_HOUR_BY_MINUTE)), &deployment);
+    assert_eq!(LAST_HOUR_BY_MINUTE.evaluations_per_request(), 61);
+    for (id, before, after) in [
+        ("P1", 2.4, 146.4),
+        ("P2", 2.1, 128.1),
+        ("P3", 2.183_333_333_333_333, 133.183_333_333_333_37),
+        ("P3-m1", 13_326.530_002_066_667, 146_593.530_002_066_7),
+        ("P3-m2", 8_884.53, 142_151.53),
+    ] {
+        assert!(close(cost(&none, id), before), "{id}: {}", cost(&none, id));
+        assert!(close(cost(&range, id), after), "{id}: {}", cost(&range, id));
+    }
+    // Every query-time computation that is not kept runs once per step.
+    for id in ["P1", "P2", "P3"] {
+        assert!(close(cost(&range, id), 61.0 * cost(&none, id)), "{id}");
+    }
+    let memory = |run: &Run, id: &str| run.selection.costs[id].usage.average_memory_bytes;
+    // B3: 4 → 64 panes kept between requests; B1: 6 → 66 panes kept.
+    assert_eq!(memory(&none, "P3-m2"), 71_072_000_000.0);
+    assert_eq!(memory(&range, "P3-m2"), 71_072_000_000.0 / 4.0 * 64.0);
+    assert_eq!(memory(&none, "P3-m1"), 106_608_000_000.0);
+    assert_eq!(memory(&range, "P3-m1"), 106_608_000_000.0 / 6.0 * 66.0);
+    assert_eq!(range.selection.selected, "P2");
+}
+
+/// (b) Refreshing every minute at 30-second steps, Stage 1 offers panes of
+/// `gcd(5 min, 1 min, 30 s) = 30 s`, on the step grid, and Stage 2 keeps
+/// them; without the range, the panes are 1 min wide. Stage 2 never offers
+/// a chain whose width does not divide the step.
+#[test]
+fn stage2_physical_asap_optimization_offers_only_step_aligned_pane_widths() {
+    let step_30s = RangeEvaluation {
+        step_ms: 30_000,
+        span_ms: 3_600_000,
+    };
+    let range = run_promql(&exact_panel("max_over_time(m[5m])", 60_000, Some(step_30s)));
+    let widths: Vec<u64> = pane_chains(&range).into_iter().map(|(_, w, _)| w).collect();
+    assert!(!widths.is_empty(), "panes are offered");
+    assert!(widths.iter().all(|&w| w == 30_000), "{widths:?}");
+    assert!(range.pruned.is_empty(), "{:?}", range.pruned);
+
+    let none = run_promql(&exact_panel("max_over_time(m[5m])", 60_000, None));
+    let widths: Vec<u64> = pane_chains(&none).into_iter().map(|(_, w, _)| w).collect();
+    assert!(
+        !widths.is_empty() && widths.iter().all(|&w| w == 60_000),
+        "{widths:?}"
+    );
+}
+
+/// (c) A plan that keeps panes reports how long the runtime must keep
+/// them: the window plus the range-evaluation span, 5 min + 1 h =
+/// 3,900,000 ms, so the next refresh's earliest step finds its panes.
+/// Without the range it is the window, 300,000 ms.
+#[test]
+fn stage3_plan_selection_reports_the_retention_of_kept_panes() {
+    let deployment = keeping_query_time_results();
+    let query = "max_over_time(m[5m])";
+    let range = run_promql_for(
+        &exact_panel(query, 60_000, Some(LAST_HOUR_BY_MINUTE)),
+        &deployment,
+    );
+    let kept = with_chain(
+        &range,
+        Materialization::MatAtQueryKeptAcrossEvaluations(Medium::Memory),
+    );
+    assert_eq!(range.selection.selected, kept);
+    let retention: Vec<u64> = range
+        .selection
+        .pane_retention_ms
+        .values()
+        .copied()
+        .collect();
+    assert_eq!(retention, [3_900_000]);
+
+    // Pattern B without a range selects the whole-window KLL, which keeps
+    // nothing; B1 reports its window when it is the plan (no raw data).
+    let none = run_promql(&pattern_b_with(None));
+    assert!(none.selection.pane_retention_ms.is_empty());
+    let b1 = run_promql_for(
+        &pattern_b_with(None),
+        &DeploymentCapabilities {
+            raw_data_retained: false,
+            ..DeploymentCapabilities::default()
+        },
+    );
+    let retention: Vec<u64> = b1.selection.pane_retention_ms.values().copied().collect();
+    assert_eq!(retention, [PATTERN_B_WINDOW_MS]);
+}
+
+/// (d) Without a range evaluation nothing changes: Pattern B's pinned
+/// costs (#509 Example 3 and 4: whole-window KLL 2.1, panes 2.18333, B3
+/// 8,884.53, B1 106.6 GB kept) are as before, and a range of one step per
+/// request (`span_ms: 0`) prices exactly as no range.
+#[test]
+fn no_range_evaluation_leaves_every_price_unchanged() {
+    let deployment = keeping_query_time_results();
+    let none = run_promql_for(&pattern_b_with(None), &deployment);
+    assert!(close(cost(&none, "P2"), 2.1));
+    assert!(close(cost(&none, "P3"), 2.1 + 5e6 * 1e-6 / 60.0));
+    assert!(close(cost(&none, "P3-m2"), 8_884.53));
+    assert_eq!(
+        none.selection.costs["P3-m1"].usage.average_memory_bytes,
+        106_608_000_000.0
+    );
+    let one_step = run_promql_for(
+        &pattern_b_with(Some(RangeEvaluation {
+            step_ms: 60_000,
+            span_ms: 0,
+        })),
+        &deployment,
+    );
+    assert_eq!(one_step.selection.costs, none.selection.costs);
+    assert_eq!(one_step.selection.selected, none.selection.selected);
+}
+
+/// The example where declaring the range makes Stage 3 prefer sharing:
+/// `max_over_time(m[5m])` every minute over 1M series. As one evaluation
+/// per minute, rebuilding the window (P1, 2.0833 cost/s) beats the kept
+/// panes (P2-m2, 4.5133: 32 MB of per-series maxima kept) and the panes
+/// maintained at ingestion (P2-m1, 6.5133). As a dashboard over the last
+/// hour at 1-minute steps, rebuilding costs 61 times as much (127.0833),
+/// while the kept panes add only the per-step merges and 1 h of panes
+/// (70.5133), and B1 72.5133: the kept panes are selected, and without
+/// query-time retention, the ingestion-time panes.
+#[test]
+fn stage3_plan_selection_prefers_kept_panes_once_the_range_is_declared() {
+    let query = "max_over_time(m[5m])";
+    let deployment = keeping_query_time_results();
+    let none = run_promql_for(&exact_panel(query, 60_000, None), &deployment);
+    let range = run_promql_for(
+        &exact_panel(query, 60_000, Some(LAST_HOUR_BY_MINUTE)),
+        &deployment,
+    );
+    let kept = with_chain(
+        &range,
+        Materialization::MatAtQueryKeptAcrossEvaluations(Medium::Memory),
+    );
+    let ingestion = with_chain(&range, Materialization::MatAtIngestion(Medium::Memory));
+    for (run, id, expected) in [
+        (&none, "P1", 2.083_333),
+        (&none, kept.as_str(), 4.513_333),
+        (&none, ingestion.as_str(), 6.513_333),
+        (&range, "P1", 127.083_333),
+        (&range, kept.as_str(), 70.513_333),
+        (&range, ingestion.as_str(), 72.513_333),
+    ] {
+        assert!(
+            (cost(run, id) - expected).abs() < 1e-5,
+            "{id}: {}",
+            cost(run, id)
+        );
+    }
+    assert_eq!(none.selection.selected, "P1");
+    assert_eq!(range.selection.selected, kept);
+    let p1 = range
+        .selection
+        .rejected
+        .iter()
+        .find(|r| r.id == "P1")
+        .unwrap();
+    assert_eq!(p1.cause, RejectionCause::Costlier);
+
+    let without_retention = run_promql(&exact_panel(query, 60_000, Some(LAST_HOUR_BY_MINUTE)));
+    assert_eq!(without_retention.selection.selected, ingestion);
+}
diff --git a/crates/integration-tests/tests/reuse_rules.rs b/crates/integration-tests/tests/reuse_rules.rs
index 9fa03c48..4fd4694b 100644
--- a/crates/integration-tests/tests/reuse_rules.rs
+++ b/crates/integration-tests/tests/reuse_rules.rs
@@ -61,6 +61,7 @@ fn exact(recurrence: QueryRecurrence) -> QueryWorkloadEntry {
         requirements: Default::default(),
         predictability: Default::default(),
         recurrence,
+        range_evaluation: None,
     }
 }
 
diff --git a/crates/integration-tests/tests/stage1_logical_asap_optimization.rs b/crates/integration-tests/tests/stage1_logical_asap_optimization.rs
index 485e90dc..6c46f5da 100644
--- a/crates/integration-tests/tests/stage1_logical_asap_optimization.rs
+++ b/crates/integration-tests/tests/stage1_logical_asap_optimization.rs
@@ -64,6 +64,7 @@ fn promql(query: &str) -> QueryRoot {
                 predictability: Default::default(),
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },
diff --git a/crates/logical-asap-optimization/src/asap_aware_cse.rs b/crates/logical-asap-optimization/src/asap_aware_cse.rs
index 66ad12e1..a9ce3c23 100644
--- a/crates/logical-asap-optimization/src/asap_aware_cse.rs
+++ b/crates/logical-asap-optimization/src/asap_aware_cse.rs
@@ -679,7 +679,8 @@ fn sized_for_all(
 /// (the family-capability table: exact accumulators included, and families
 /// whose merged readouts are only approximate), read only by queries that
 /// repeat at a fixed interval. Its panes are `pane_count = W / w` back-to-back
-/// windows of `w = gcd(W, E)` over the queries' intervals `E`, at least two.
+/// windows of `w = gcd(W, E)` over the queries' intervals `E` (and, for a
+/// query with a range evaluation, its step), at least two.
 #[derive(Debug, Clone)]
 pub struct PaneCandidate {
     /// The `SummaryAgg` over `TimeRange(W)` built as panes.
@@ -707,9 +708,16 @@ pub struct WindowCompositionPass2Candidate {
 pub const MAX_PANE_CANDIDATES: usize = 20;
 
 /// The pane candidates of the candidate with query roots `roots`, children
-/// first. `interval_ms[q]` is how often query `q` repeats, `None` when it
-/// does not repeat at a fixed interval.
-pub fn pane_candidates(roots: &[QueryRoot], interval_ms: &[Option<u64>]) -> Vec<PaneCandidate> {
+/// first. `evaluation_grid_ms[q]` is the grid query `q`'s evaluations fall
+/// on: how often it repeats, or with a range evaluation the greatest common
+/// divisor of that and its step
+/// (`QueryWorkloadEntry::evaluation_grid_ms`); `None` when it does not
+/// repeat at a fixed interval. Each pane width divides every reader's grid,
+/// so every evaluation, every range step included, ends on a pane boundary.
+pub fn pane_candidates(
+    roots: &[QueryRoot],
+    evaluation_grid_ms: &[Option<u64>],
+) -> Vec<PaneCandidate> {
     let mut nodes = Vec::new();
     let mut seen = std::collections::HashSet::new();
     for node in roots.iter().flat_map(root_operators) {
@@ -734,7 +742,10 @@ pub fn pane_candidates(roots: &[QueryRoot], interval_ms: &[Option<u64>]) -> Vec<
                 .collect();
             let mut width = window_ms;
             for &q in &readers {
-                width = gcd(width, interval_ms.get(q).copied().flatten()?);
+                width = asap_types::workload::greatest_common_divisor(
+                    width,
+                    evaluation_grid_ms.get(q).copied().flatten()?,
+                );
             }
             let pane_count = u32::try_from(window_ms / width).ok()?;
             (!readers.is_empty() && pane_count >= 2).then_some(PaneCandidate {
@@ -777,14 +788,6 @@ fn pane_window_ms(build: &OperatorNode) -> Option<u64> {
     u64::try_from(range.as_millis()).ok().filter(|&ms| ms > 0)
 }
 
-fn gcd(a: u64, b: u64) -> u64 {
-    if b == 0 {
-        a
-    } else {
-        gcd(b, a % b)
-    }
-}
-
 /// Every Pass 2 candidate of the window-composition rule for the candidate
 /// with query roots `roots` and pane candidates `candidates`
 /// ([`pane_candidates`]): one per non-empty subset of them, in
diff --git a/crates/logical-asap-optimization/src/test_support.rs b/crates/logical-asap-optimization/src/test_support.rs
index d598492f..7644a8a0 100644
--- a/crates/logical-asap-optimization/src/test_support.rs
+++ b/crates/logical-asap-optimization/src/test_support.rs
@@ -31,6 +31,7 @@ pub fn lower_promql(query: &str, accuracy: AccuracyTarget) -> Rc<OperatorNode> {
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },
@@ -205,7 +206,7 @@ pub fn cms_summary(query: &str) -> OperatorNode {
         .without_unsupported_summaries(&DeploymentCapabilities::default());
     let summary = stage1_logical_asap_optimization(
         &per_computation,
-        &[asap_types::workload::QueryRecurrence::Unknown],
+        &[entry(asap_types::workload::QueryRecurrence::Unknown)],
     )
     .filter_map(|candidate| candidate.roots.ok())
     .flat_map(|roots| match &roots[0] {
@@ -281,5 +282,6 @@ pub fn entry_with(
         },
         predictability: Default::default(),
         recurrence,
+        range_evaluation: None,
     }
 }
diff --git a/crates/logical-asap-optimization/src/workload_candidates.rs b/crates/logical-asap-optimization/src/workload_candidates.rs
index 640a78bf..b104d9d7 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;
+use asap_types::workload::QueryWorkloadEntry;
 
 use crate::asap_aware_cse::{
     capability_sharing_candidates, identical_expression_pass2_candidates, pane_candidates,
@@ -69,19 +69,21 @@ pub struct Stage1Candidate {
 /// (exact top-k reuse and rollup), then each summary-capability
 /// candidate followed by its own identical-expression and
 /// window-composition candidates; every window-composition candidate is
-/// followed by its own identical-expression candidates. `recurrence[q]` is
-/// how often query `q` is evaluated: the window-composition rule splits
-/// only builds of queries that repeat at a fixed interval. A Pass 2
+/// followed by its own identical-expression candidates. `entries[q]` is
+/// query `q`'s workload entry: the window-composition rule splits only
+/// builds of queries that repeat at a fixed interval, into panes on every
+/// reader's evaluation grid ([`QueryWorkloadEntry::evaluation_grid_ms`]:
+/// the interval, or with a range evaluation also its step). A Pass 2
 /// candidate equal to an earlier one of the same Pass 1 candidate (same
 /// nodes, same sharing) is listed once. Lazy and uncapped: a caller that
 /// limits the candidates stops consuming it.
 pub fn stage1_logical_asap_optimization<'a, Id: Clone>(
     per_computation: &'a PerComputationCandidates<Id>,
-    recurrence: &[QueryRecurrence],
+    entries: &[QueryWorkloadEntry],
 ) -> impl Iterator<Item = Stage1Candidate> + 'a {
-    let interval_ms: Vec<Option<u64>> = recurrence
+    let evaluation_grid_ms: Vec<Option<u64>> = entries
         .iter()
-        .map(QueryRecurrence::fixed_interval_ms)
+        .map(QueryWorkloadEntry::evaluation_grid_ms)
         .collect();
     let mut choice = Some(vec![0; per_computation.targets.len()]);
     let mut queue = std::collections::VecDeque::new();
@@ -101,7 +103,7 @@ pub fn stage1_logical_asap_optimization<'a, Id: Clone>(
                 .map(|roots| roots.into_iter().map(|(_, root)| root).collect::<Vec<_>>());
             let from = position + queue.len();
             let pass2 = match &roots {
-                Ok(roots) => pass2_candidates(roots, &interval_ms),
+                Ok(roots) => pass2_candidates(roots, &evaluation_grid_ms),
                 Err(_) => Vec::new(),
             };
             queue.push_back(Stage1Candidate {
@@ -165,16 +167,17 @@ impl Pass2 {
 }
 
 /// The Pass 2 candidates of the composed candidate with roots `roots`, in
-/// enumeration order, each distinct candidate once. `interval_ms[q]` is the
-/// fixed interval of query `q`, if it repeats at one.
-fn pass2_candidates(roots: &[QueryRoot], interval_ms: &[Option<u64>]) -> Vec<Pass2> {
+/// enumeration order, each distinct candidate once. `evaluation_grid_ms[q]`
+/// is the grid query `q`'s evaluations fall on, if it repeats at a fixed
+/// interval ([`QueryWorkloadEntry::evaluation_grid_ms`]).
+fn pass2_candidates(roots: &[QueryRoot], evaluation_grid_ms: &[Option<u64>]) -> Vec<Pass2> {
     let mut found = Pass2List {
         seen: vec![flatten(roots).0],
         out: Vec::new(),
     };
     let none = Rules::default();
     found.identical_expression(&none, roots);
-    found.window_composition(&none, roots, interval_ms);
+    found.window_composition(&none, roots, evaluation_grid_ms);
     found.reuse(&none, roots);
     let candidates = capability_sharing_candidates(roots);
     let pass2_candidates = match summary_capability_pass2_candidates(roots, &candidates) {
@@ -193,7 +196,7 @@ fn pass2_candidates(roots: &[QueryRoot], interval_ms: &[Option<u64>]) -> Vec<Pas
             Ok(roots) => {
                 found.push(&rules, Vec::new(), &roots);
                 found.identical_expression(&rules, &roots);
-                found.window_composition(&rules, &roots, interval_ms);
+                found.window_composition(&rules, &roots, evaluation_grid_ms);
             }
             // Listed as a candidate that cannot be composed; the other
             // Pass 2 candidates go on.
@@ -249,9 +252,9 @@ impl Pass2List {
         &mut self,
         rules: &Rules,
         roots: &[QueryRoot],
-        interval_ms: &[Option<u64>],
+        evaluation_grid_ms: &[Option<u64>],
     ) {
-        let candidates = pane_candidates(roots, interval_ms);
+        let candidates = pane_candidates(roots, evaluation_grid_ms);
         let pass2_candidates = match window_composition_pass2_candidates(roots, &candidates) {
             Ok(pass2_candidates) => pass2_candidates,
             Err(error) => {
diff --git a/crates/logical-asap-optimization/tests/per_computation_candidates.rs b/crates/logical-asap-optimization/tests/per_computation_candidates.rs
index e2203915..c099e512 100644
--- a/crates/logical-asap-optimization/tests/per_computation_candidates.rs
+++ b/crates/logical-asap-optimization/tests/per_computation_candidates.rs
@@ -195,6 +195,7 @@ fn exact_promql(query: &str) -> Vec<QueryRoot> {
                 predictability: Default::default(),
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },
diff --git a/crates/physical-asap-optimization/src/lib.rs b/crates/physical-asap-optimization/src/lib.rs
index 806cdd18..ab32d3b6 100644
--- a/crates/physical-asap-optimization/src/lib.rs
+++ b/crates/physical-asap-optimization/src/lib.rs
@@ -47,9 +47,24 @@
 //!   not taken to be ordered;
 //! * it is a tumbling-pane chain, or a whole-window build whose readers
 //!   all run at one fixed interval with a known phase, over a window no
-//!   longer than that interval and not shifted;
+//!   longer than that interval and not shifted, and none is a range
+//!   evaluation of more than one step per request (one window per request
+//!   cannot serve every step);
 //! * everything below it can run at ingestion time.
 //!
+//! ## Range evaluations
+//!
+//! A query whose workload entry declares a
+//! [`RangeEvaluation`](asap_types::workload::RangeEvaluation) evaluates at
+//! every `step_ms` from `end − span_ms` forward to each request's end.
+//! Every step reads the
+//! panes of a tumbling-pane chain, so each pane width must divide the step
+//! of every range-evaluated reader. Stage 1 offers only such widths
+//! (`QueryWorkloadEntry::evaluation_grid_ms`); a candidate with a chain off
+//! a reader's step grid fails with
+//! [`PhysicalASAPOptimizationError::PaneWidthOffRangeStep`] and the caller
+//! prunes it with that reason.
+//!
 //! ## Physical operator implementation
 //!
 //! The runtime has one implementation per logical operator except exact
@@ -131,6 +146,21 @@ pub enum PhysicalASAPOptimizationError {
     Timing(#[from] ExecutionDataStateError),
     #[error("exact top-k has no Sort + Limit implementation: {0}")]
     ExactTopK(String),
+    /// Bad candidate for the workload: a tumbling-pane chain whose pane
+    /// width does not divide the step of a query that reads it as a range
+    /// request, so some step would end inside a pane and could not read
+    /// the panes. Stage 1 offers only panes on every reader's step grid;
+    /// the candidate is pruned with this reason.
+    #[error(
+        "pane width {pane_width_ms} ms does not divide the {step_ms} ms step of \
+         a range evaluation that reads it"
+    )]
+    PaneWidthOffRangeStep {
+        /// The chain's pane width.
+        pane_width_ms: u64,
+        /// The reader's range-evaluation step.
+        step_ms: u64,
+    },
 }
 
 /// Why a summary build cannot be materialized at ingestion time (S4).
@@ -153,6 +183,9 @@ pub(crate) enum IngestionPruneReason {
     /// interval with a known phase, or whose window is longer than that
     /// interval, shifted, or not a range.
     WindowNotFixed,
+    /// A whole-window build read by a range evaluation of more than one
+    /// step per request: it keeps one window per request, not one per step.
+    RangeEvaluatedWholeWindow,
     /// Something below it can only run at query time.
     UpstreamAtQueryTime(String),
 }
@@ -275,6 +308,7 @@ pub fn optimize_physical_asap(
     let mut memo = HashMap::new();
     let implemented = map_roots(roots, &mut |node| implement(node, &mut memo))?;
     let mat_candidates = materialization_candidates(&implemented, entries, data, deployment);
+    check_panes_on_range_steps(&mat_candidates, entries)?;
     let mut candidates = Vec::new();
     for assignment in materialization_assignments(&mat_candidates) {
         let materialized: Vec<_> = mat_candidates
@@ -339,9 +373,42 @@ pub fn physical_candidate_count(
     let mut memo = HashMap::new();
     let implemented = map_roots(roots, &mut |node| implement(node, &mut memo))?;
     let mat_candidates = materialization_candidates(&implemented, entries, data, deployment);
+    check_panes_on_range_steps(&mat_candidates, entries)?;
     Ok(materialization_assignments(&mat_candidates).count())
 }
 
+/// `Err` for the first tumbling-pane chain whose pane width does not
+/// divide the step of a reader's range evaluation
+/// ([`PhysicalASAPOptimizationError::PaneWidthOffRangeStep`]): that step
+/// would end inside a pane.
+fn check_panes_on_range_steps(
+    mat_candidates: &[MaterializationCandidate],
+    entries: &[QueryWorkloadEntry],
+) -> Result<(), PhysicalASAPOptimizationError> {
+    for mat_candidate in mat_candidates {
+        let Some(ASAPOp::PaneChain { pane_width_ms, .. }) = mat_candidate.node.asap() else {
+            continue;
+        };
+        for &query in &mat_candidate.readers {
+            // A one-step range evaluates only at the request's end.
+            let Some(range) = entries
+                .get(query)
+                .and_then(|e| e.range_evaluation)
+                .filter(|range| range.evaluations_per_request() > 1)
+            else {
+                continue;
+            };
+            if range.step_ms % pane_width_ms.max(&1) != 0 {
+                return Err(PhysicalASAPOptimizationError::PaneWidthOffRangeStep {
+                    pane_width_ms: *pane_width_ms,
+                    step_ms: range.step_ms,
+                });
+            }
+        }
+    }
+    Ok(())
+}
+
 /// The operator nodes `root` reads.
 fn root_operators(root: &QueryRoot) -> Vec<&Rc<OperatorNode>> {
     match root {
@@ -479,6 +546,12 @@ fn ingestion_eligibility(
     if chain {
         return Ok(());
     }
+    if demands.iter().flatten().any(|d| {
+        d.range_evaluation
+            .is_some_and(|range| range.evaluations_per_request() > 1)
+    }) {
+        return Err(Why::RangeEvaluatedWholeWindow);
+    }
     // A whole-window build: one window per evaluation, at one known phase,
     // that ends at the evaluation and fits in the interval.
     let mut cadence = None;
@@ -1357,6 +1430,106 @@ mod tests {
         );
     }
 
+    /// A whole-window build read by a range evaluation of several steps per
+    /// request is not maintained at ingestion time: it would hold one
+    /// window per request, not one per step. A one-step range (`span_ms`
+    /// 0) is one evaluation per request and stays eligible.
+    #[test]
+    fn a_range_evaluated_whole_window_build_stays_at_query_time() {
+        use asap_types::workload::RangeEvaluation;
+        let root = summary_root("quantile_over_time(0.99, m[1m])");
+        let at_minute = |span_ms: u64| QueryWorkloadEntry {
+            recurrence: QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt {
+                interval: RepetitionInterval(MINUTE as u32),
+                evaluation_phase: TimestampMs(0),
+            }),
+            range_evaluation: Some(RangeEvaluation {
+                step_ms: MINUTE,
+                span_ms,
+            }),
+            ..every_minute()
+        };
+        let at_ingestion = |entry: QueryWorkloadEntry| {
+            let mut memo = HashMap::new();
+            let implemented = map_roots(std::slice::from_ref(&root), &mut |n| {
+                implement(n, &mut memo)
+            })
+            .unwrap();
+            materialization_candidates(
+                &implemented,
+                &[entry],
+                &arriving(),
+                &DeploymentCapabilities::default(),
+            )
+            .remove(0)
+            .at_ingestion
+        };
+        assert_eq!(
+            at_ingestion(at_minute(60 * MINUTE)),
+            Err(IngestionPruneReason::RangeEvaluatedWholeWindow)
+        );
+        assert_eq!(at_ingestion(at_minute(0)), Ok(()));
+    }
+
+    /// A pane chain whose width does not divide a range-evaluated reader's
+    /// step is not offered: the candidate fails with the widths, and the
+    /// caller prunes it. 1-min panes serve 2-min steps.
+    #[test]
+    fn a_pane_chain_off_the_range_step_grid_is_rejected() {
+        use asap_types::workload::RangeEvaluation;
+        let root = in_panes(summary_root("quantile_over_time(0.99, m[5m])"));
+        let stepping = |step_ms: u64| QueryWorkloadEntry {
+            range_evaluation: Some(RangeEvaluation {
+                step_ms,
+                span_ms: 60 * MINUTE,
+            }),
+            ..every_minute()
+        };
+        let implement_for = |entry: QueryWorkloadEntry| {
+            optimize_physical_asap(
+                "L1",
+                std::slice::from_ref(&root),
+                &[entry],
+                &arriving(),
+                &DeploymentCapabilities::default(),
+            )
+        };
+        let error = implement_for(stepping(30_000)).unwrap_err();
+        assert!(
+            matches!(
+                error,
+                PhysicalASAPOptimizationError::PaneWidthOffRangeStep {
+                    pane_width_ms: MINUTE,
+                    step_ms: 30_000,
+                }
+            ),
+            "{error}"
+        );
+        assert_eq!(
+            error.to_string(),
+            "pane width 60000 ms does not divide the 30000 ms step of a range \
+             evaluation that reads it"
+        );
+        assert!(physical_candidate_count(
+            std::slice::from_ref(&root),
+            &[stepping(30_000)],
+            &arriving(),
+            &DeploymentCapabilities::default(),
+        )
+        .is_err());
+        assert_eq!(implement_for(stepping(2 * MINUTE)).unwrap().len(), 2);
+        // A one-step range (span 0) is one evaluation per request, at the
+        // refresh: its step is not read.
+        let one_step = QueryWorkloadEntry {
+            range_evaluation: Some(RangeEvaluation {
+                step_ms: 30_000,
+                span_ms: 0,
+            }),
+            ..every_minute()
+        };
+        assert_eq!(implement_for(one_step).unwrap().len(), 2);
+    }
+
     /// A build two queries read is also offered built once per evaluation
     /// for both (`MatAtQuery`); a build one query reads is not.
     #[test]
diff --git a/crates/plan-selection/src/cost/query_physical_lowering.rs b/crates/plan-selection/src/cost/query_physical_lowering.rs
index 6e3a2651..f5833494 100644
--- a/crates/plan-selection/src/cost/query_physical_lowering.rs
+++ b/crates/plan-selection/src/cost/query_physical_lowering.rs
@@ -2897,6 +2897,7 @@ mod tests {
                     predictability: Predictability::Unknown,
                     invocations: 1,
                     execute_at: None,
+                    range_evaluation: None,
                 }]),
                 repeating_queries: None,
             },
diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs
index c0f1e55e..967a33d4 100644
--- a/crates/plan-selection/src/lib.rs
+++ b/crates/plan-selection/src/lib.rs
@@ -72,6 +72,29 @@
 //! look-back, are kept every second (in memory, all `N` at the peak). When
 //! Stage 2 offers both query-time options for a chain, plan selection
 //! prices each and the cheaper wins.
+//!
+//! A query whose workload entry declares a range evaluation
+//! ([`RangeEvaluation`]: each request evaluates it at every `step_ms` over
+//! the last `span_ms`, as a dashboard's `query_range` does) is priced as
+//! `S = span_ms / step_ms + 1` evaluations per request (61 for a 1 h span
+//! at 1 min steps):
+//!
+//! * every query-time computation that is not kept across evaluations
+//!   (`NotMaterialized`, `MatAtQuery`) runs once per step, `S` times per
+//!   request; a `MatAtIngestion` state is maintained as before, and the
+//!   merge and estimate over it run once per step;
+//! * a chain kept across evaluations is built once per request: Stage 2
+//!   offers it only when the requests repeat at its pane width, so in the
+//!   steady state each request builds its newest pane and every other pane
+//!   any of its steps reads was built by an earlier request or step;
+//! * a pane chain kept beyond one evaluation (ingestion time or kept
+//!   across evaluations) retains its window plus the longest
+//!   `span_ms` among its readers ([`Selection::pane_retention_ms`]),
+//!   so the earliest step of the next request still finds its panes: that
+//!   many more panes are kept every second.
+//!
+//! Latency stays per evaluation (one step). Without a range evaluation
+//! `S = 1` and the retention is the window, so every price is as above.
 //! Memory a computation
 //! holds only while it runs is not charged under cost model 1: it counts in
 //! [`ResourceUsage::peak_memory_bytes`], which cost models 2 and 3 read. A
@@ -187,7 +210,7 @@ use asap_types::ir::{
 };
 use asap_types::physical::{Materialization, Medium};
 use asap_types::workload::{
-    DataWorkload, MetricType, QueryRecurrence, QueryWorkloadEntry, RepeatedDemand,
+    DataWorkload, MetricType, QueryRecurrence, QueryWorkloadEntry, RangeEvaluation, RepeatedDemand,
 };
 use thiserror::Error;
 
@@ -778,6 +801,15 @@ pub struct Selection {
     /// other beats on peak memory, disk and worst-query latency at once, in
     /// candidate order; empty under every other objective and cost model.
     pub front: Vec<String>,
+    /// For each tumbling-pane chain of the selected candidate kept beyond
+    /// one evaluation (maintained at ingestion time, or built at query time
+    /// and kept across evaluations), by node id in its physical DAG: how
+    /// far back from the newest evaluation the runtime must keep its
+    /// completed panes, in milliseconds. It is the chain's window plus the
+    /// longest range-evaluation span of its readers, so the earliest step
+    /// of the next request still finds its panes; just the window without
+    /// range evaluations. Empty when the plan keeps no panes.
+    pub pane_retention_ms: BTreeMap<PhysicalASAPNodeId, u64>,
 }
 
 #[derive(Debug, Error)]
@@ -835,9 +867,11 @@ pub fn select_plan(
     let mut accepted = BTreeMap::new();
     let mut rejected = Vec::new();
     // Valid candidates in candidate order: the index into `cands` and the
-    // roots sized for their failure probabilities.
-    let mut valid: Vec<(usize, Vec<QueryRoot>)> = Vec::new();
+    // candidate sized for its failure probabilities.
+    let mut valid: Vec<(usize, PhysicalCandidate)> = Vec::new();
     let recurrence: Vec<QueryRecurrence> = entries.iter().map(|e| e.recurrence.clone()).collect();
+    let range_evaluations: Vec<Option<RangeEvaluation>> =
+        entries.iter().map(|e| e.range_evaluation).collect();
     for (index, candidate) in cands.iter().enumerate() {
         if candidate.roots.len() != entries.len() {
             return Err(SelectionError::EntryCount {
@@ -886,6 +920,7 @@ pub fn select_plan(
             &nodes,
             &kept_chains,
             &recurrence,
+            &range_evaluations,
             data,
             settings,
         )
@@ -911,7 +946,7 @@ pub fn select_plan(
         }
         costs.insert(candidate.id.clone(), cost);
         accepted.insert(candidate.id.clone(), accepted_accuracy);
-        valid.push((index, candidate.roots));
+        valid.push((index, candidate));
     }
     let id = |index: usize| cands[index].id.as_str();
     let ranking = Ranking::of(&settings.cost_model);
@@ -935,7 +970,13 @@ pub fn select_plan(
             });
         }
     }
-    let plan = with_confirmed_weights(&valid.swap_remove(best).1, data);
+    let (_, chosen) = valid.swap_remove(best);
+    let plan = with_confirmed_weights(&chosen.roots, data);
+    let pane_retention_ms = pane_retention_ms(
+        &chosen.dag,
+        &query_reach(&chosen.dag, &node_map(&chosen.dag)),
+        &range_evaluations,
+    );
     Ok(Selection {
         selected,
         plan,
@@ -943,6 +984,7 @@ pub fn select_plan(
         accepted,
         rejected,
         front,
+        pane_retention_ms,
     })
 }
 
@@ -2041,19 +2083,29 @@ fn priced_series(sources: &BTreeSet<&str>, data: &DataWorkload) -> u64 {
         .max(1)
 }
 
-/// Evaluations per second of a node computed once for `queries` together.
-/// Repeating queries with equal intervals are evaluated together, so each
-/// interval counts once; estimated rates add. One-off, scheduled and unknown
-/// queries run as one batch: the most invocations among them, spread over
+/// Evaluations per second of a node computed once for `queries` together,
+/// each a recurrence with its range evaluation: a request with a range
+/// evaluation is [`RangeEvaluation::evaluations_per_request`] evaluations,
+/// one without it one. Repeating queries with equal intervals are
+/// evaluated together, every request ending at the same instant: a node
+/// they share is computed once per distinct instant their steps fall on,
+/// `end − k · step_ms` for `k · step_ms ≤ span_ms`. A query evaluated once
+/// per request (no range, or one step) evaluates only at the end. Queries
+/// on different intervals add; estimated rates add. One-off, scheduled and unknown queries run as one
+/// batch: the most evaluations among them, spread over
 /// `non_repeating_query_horizon_seconds`.
 fn evaluation_rate<'r>(
-    queries: impl IntoIterator<Item = &'r QueryRecurrence>,
+    queries: impl IntoIterator<Item = (&'r QueryRecurrence, Option<RangeEvaluation>)>,
     non_repeating_query_horizon_seconds: f64,
 ) -> Result<f64, AnalyticalCostError> {
-    let mut intervals = BTreeSet::new();
+    // Per interval in ms, each step in ms its readers use and the longest
+    // span in ms at that step; once per request is the interval with span 0.
+    let mut longest_span_ms_by_step_ms_by_interval_ms: BTreeMap<u32, BTreeMap<u64, u64>> =
+        BTreeMap::new();
     let mut estimated = 0.0;
     let mut invocations = 0u64;
-    for recurrence in queries {
+    for (recurrence, range_evaluation) in queries {
+        let steps = range_evaluation.map_or(1, |range| range.evaluations_per_request());
         match recurrence {
             QueryRecurrence::Repeated(
                 RepeatedDemand::FixedInterval(interval)
@@ -2062,28 +2114,60 @@ fn evaluation_rate<'r>(
                 if interval.0 == 0 {
                     return Err(AnalyticalCostError::InvalidRecurrence);
                 }
-                intervals.insert(interval.0);
+                let (step_ms, span_ms) = range_evaluation
+                    .filter(|_| steps > 1)
+                    .map_or((u64::from(interval.0), 0), |range| {
+                        (range.step_ms, range.span_ms)
+                    });
+                let longest_span_ms = longest_span_ms_by_step_ms_by_interval_ms
+                    .entry(interval.0)
+                    .or_default()
+                    .entry(step_ms)
+                    .or_default();
+                *longest_span_ms = (*longest_span_ms).max(span_ms);
             }
             QueryRecurrence::Repeated(RepeatedDemand::EstimatedRate(estimate)) => {
                 let rate = estimate.expected_rate.0;
                 if !rate.is_finite() || rate < 0.0 {
                     return Err(AnalyticalCostError::InvalidRecurrence);
                 }
-                estimated += rate;
+                estimated += rate * steps as f64;
             }
             QueryRecurrence::Repeated(RepeatedDemand::Scheduled(times)) => {
-                invocations = invocations.max(times.len() as u64);
+                invocations = invocations.max(times.len() as u64 * steps);
             }
             QueryRecurrence::OneTime {
                 invocations: count, ..
-            } => invocations = invocations.max(*count),
-            QueryRecurrence::Unknown => invocations = invocations.max(1),
+            } => invocations = invocations.max(*count * steps),
+            QueryRecurrence::Unknown => invocations = invocations.max(steps),
         }
     }
-    let repeated: f64 = intervals.into_iter().map(|ms| 1_000.0 / ms as f64).sum();
+    let repeated: f64 = longest_span_ms_by_step_ms_by_interval_ms
+        .into_iter()
+        .map(|(interval_ms, longest_span_ms_by_step_ms)| {
+            1_000.0 / f64::from(interval_ms)
+                * distinct_instants_per_request(&longest_span_ms_by_step_ms) as f64
+        })
+        .sum();
     Ok(repeated + estimated + invocations as f64 / non_repeating_query_horizon_seconds)
 }
 
+/// The distinct instants one request evaluates at, for steps in ms each with
+/// the longest span in ms at that step: the union of the offsets
+/// `k · step_ms ≤ span_ms` back from the request's end. One step needs no
+/// union: `span_ms / step_ms + 1`.
+fn distinct_instants_per_request(longest_span_ms_by_step_ms: &BTreeMap<u64, u64>) -> u64 {
+    if longest_span_ms_by_step_ms.len() == 1 {
+        let (&step_ms, &span_ms) = longest_span_ms_by_step_ms.iter().next().expect("one step");
+        return span_ms / step_ms.max(1) + 1;
+    }
+    let mut offsets_ms = BTreeSet::new();
+    for (&step_ms, &span_ms) in longest_span_ms_by_step_ms {
+        offsets_ms.extend((0..=span_ms / step_ms.max(1)).map(|k| k * step_ms));
+    }
+    offsets_ms.len() as u64
+}
+
 /// The nodes each query reads first, in query order: query `i` is the scalar
 /// root keyed `i` when there is one, else the next operator root.
 fn query_starts(dag: &PhysicalASAPDAG) -> Vec<Vec<PhysicalASAPNodeId>> {
@@ -2104,9 +2188,12 @@ fn query_starts(dag: &PhysicalASAPDAG) -> Vec<Vec<PhysicalASAPNodeId>> {
 
 /// How often each node of `dag` is computed: `(multiplier, per_second)`
 /// (module doc, "Cost model"). A materialized node is computed once for all
-/// its readers: once per second at ingestion time (`MatAtIngestion`), or at
+/// its readers: once per second at ingestion time (`MatAtIngestion`), at
 /// the combined rate of the queries reaching it (`MatAtQuery`,
-/// [`evaluation_rate`]). A `NotMaterialized` node is computed once per
+/// [`evaluation_rate`]), or once per request of those queries for a chain
+/// kept across evaluations, whose request builds only its newest pane
+/// however many range steps it evaluates. `range_evaluations[q]` is query
+/// `q`'s range evaluation (none when missing). A `NotMaterialized` node is computed once per
 /// consumer that reaches it, at that consumer's rate. A consumer is a query
 /// (operator or scalar root), at its own rate, or a materialized node, at
 /// that node's rate; a walk stops at the first materialized node it meets,
@@ -2115,8 +2202,10 @@ fn query_starts(dag: &PhysicalASAPDAG) -> Vec<Vec<PhysicalASAPNodeId>> {
 fn computations(
     dag: &PhysicalASAPDAG,
     recurrence: &[QueryRecurrence],
+    range_evaluations: &[Option<RangeEvaluation>],
     non_repeating_query_horizon_seconds: f64,
 ) -> Result<HashMap<PhysicalASAPNodeId, (u64, f64)>, AnalyticalCostError> {
+    let range_of = |query: usize| range_evaluations.get(query).copied().flatten();
     let materialized: HashMap<_, _> = dag
         .nodes
         .iter()
@@ -2145,14 +2234,18 @@ fn computations(
     }
     let mut rate_of_materialized = HashMap::new();
     for (&id, materialization) in &materialized {
+        // A chain kept across evaluations is built once per request.
+        let per_request = matches!(
+            materialization,
+            Materialization::MatAtQueryKeptAcrossEvaluations(_)
+        );
         let rate = match materialization {
             Materialization::MatAtIngestion(_) => 1.0,
             _ => evaluation_rate(
-                reached
-                    .get(&id)
-                    .into_iter()
-                    .flatten()
-                    .filter_map(|&q| recurrence.get(q)),
+                reached.get(&id).into_iter().flatten().filter_map(|&q| {
+                    let range = if per_request { None } else { range_of(q) };
+                    recurrence.get(q).map(|r| (r, range))
+                }),
                 non_repeating_query_horizon_seconds,
             )?,
         };
@@ -2161,7 +2254,10 @@ fn computations(
     // Each consumer's starting nodes and its rate.
     let mut consumers = Vec::new();
     for (query, start) in starts.into_iter().enumerate() {
-        let rate = evaluation_rate(recurrence.get(query), non_repeating_query_horizon_seconds)?;
+        let rate = evaluation_rate(
+            recurrence.get(query).map(|r| (r, range_of(query))),
+            non_repeating_query_horizon_seconds,
+        )?;
         consumers.push((start, rate));
     }
     for (&id, &rate) in &rate_of_materialized {
@@ -2201,13 +2297,17 @@ fn computations(
 /// time ranges keep, times how often it is computed per second
 /// ([`computations`]); an ingestion-time node over one second of arriving
 /// rows. A materialized node's kept state is counted in byte-seconds per
-/// second. `nodes` and `kept_chains` are `dag`'s ([`node_map`],
-/// [`kept_pane_chains`]).
+/// second; a pane chain kept beyond one evaluation keeps the panes its
+/// retention covers ([`pane_retention_ms`]). `nodes` and `kept_chains` are
+/// `dag`'s ([`node_map`], [`kept_pane_chains`]); `recurrence[q]` and
+/// `range_evaluations[q]` say how often query `q` runs and how many steps
+/// each run evaluates (one when missing).
 fn price(
     dag: &PhysicalASAPDAG,
     nodes: &NodeMap<'_>,
     kept_chains: &HashSet<PhysicalASAPNodeId>,
     recurrence: &[QueryRecurrence],
+    range_evaluations: &[Option<RangeEvaluation>],
     data: &DataWorkload,
     settings: &PlanSelectionSettings,
 ) -> Result<CandidateCost, (PhysicalASAPNodeId, AnalyticalCostError)> {
@@ -2228,9 +2328,12 @@ fn price(
     let multipliers = computations(
         dag,
         recurrence,
+        range_evaluations,
         settings.non_repeating_query_horizon_seconds,
     )
     .map_err(|error| (first, error))?;
+    let reach = query_reach(dag, nodes);
+    let retention_ms = pane_retention_ms(dag, &reach, range_evaluations);
     let mut output: HashMap<PhysicalASAPNodeId, EdgeStatistics> = HashMap::new();
     let mut per_node = BTreeMap::new();
     // The candidate's usage, accumulated node by node; peak cores and
@@ -2656,16 +2759,21 @@ fn price(
                 peak_memory_bytes = estimate.peak_memory_bytes().saturating_mul(multiplier);
                 (0, Medium::Memory)
             }
-            // A pane chain kept across evaluations holds its N − 1 older
-            // panes between evaluations, for the next ones, and all N
-            // while one runs.
+            // A pane chain kept across evaluations holds the panes its
+            // retention covers ([`pane_retention_ms`]: N, plus one per pane
+            // width of the readers' range-evaluation span), all but the
+            // newest between evaluations and all while one runs.
             (Materialization::MatAtQueryKeptAcrossEvaluations(medium), _) => {
                 let panes = panes_of(node);
-                let between = state / panes * (panes - 1);
+                let retained = retained_panes(node, &retention_ms);
+                let between = state / panes * (retained - 1);
                 match medium {
                     Medium::Memory => {
                         average_memory_bytes = between as f64;
-                        peak_memory_bytes = state;
+                        peak_memory_bytes = match retained == panes {
+                            true => state,
+                            false => state / panes * retained,
+                        };
                     }
                     Medium::Disk => disk_bytes = between,
                 }
@@ -2682,12 +2790,13 @@ fn price(
                 }
                 (0, medium)
             }
-            (
-                Materialization::MatAtIngestion(medium),
-                Payload::ASAP(ASAPOp::PaneChain { pane_count, .. }),
-            ) => {
-                let panes = u64::from(*pane_count).max(1);
-                (state / panes * (panes + 1), medium)
+            // The retained panes and the one being built.
+            (Materialization::MatAtIngestion(medium), Payload::ASAP(ASAPOp::PaneChain { .. })) => {
+                let panes = panes_of(node);
+                (
+                    state / panes * (retained_panes(node, &retention_ms) + 1),
+                    medium,
+                )
             }
             (Materialization::MatAtIngestion(medium), _) => (2 * state, medium),
         };
@@ -2772,7 +2881,6 @@ fn price(
     let latency = query_latencies(dag, nodes, &output, &work_ms, &merge_ms, &speed);
     // A MatAtQuery state is held for one evaluation: from its build until
     // the longest query reading it finishes, about that query's latency.
-    let reach = query_reach(dag, nodes);
     for (id, (state, per_second)) in held {
         let evaluation_ms = reach
             .iter()
@@ -2933,6 +3041,60 @@ fn ingest_workers(
     (ingest_ms_per_s / 1_000.0).ceil().max(1.0) as u64
 }
 
+/// For each tumbling-pane chain of `dag` kept beyond one evaluation (at
+/// ingestion time, or at query time across evaluations), how far back
+/// from the newest evaluation its completed panes must be kept, in
+/// milliseconds: its window `pane_count · pane_width_ms`, plus the longest
+/// `span_ms` ([`RangeEvaluation`]) among the queries reaching it,
+/// whose earliest step reads that much further back. `reach` is
+/// [`query_reach`] of `dag`; `range_evaluations[q]` is query `q`'s (none
+/// when missing). Without range evaluations, each chain's window.
+fn pane_retention_ms(
+    dag: &PhysicalASAPDAG,
+    reach: &[HashSet<PhysicalASAPNodeId>],
+    range_evaluations: &[Option<RangeEvaluation>],
+) -> BTreeMap<PhysicalASAPNodeId, u64> {
+    dag.nodes
+        .iter()
+        .filter_map(|node| match (&node.payload, node.materialization) {
+            (
+                Payload::ASAP(ASAPOp::PaneChain {
+                    pane_width_ms,
+                    pane_count,
+                    ..
+                }),
+                Materialization::MatAtIngestion(_)
+                | Materialization::MatAtQueryKeptAcrossEvaluations(_),
+            ) => {
+                let span_ms = reach
+                    .iter()
+                    .enumerate()
+                    .filter(|(_, reached)| reached.contains(&node.id))
+                    .filter_map(|(q, _)| range_evaluations.get(q).copied().flatten())
+                    .map(|range| range.span_ms)
+                    .max()
+                    .unwrap_or(0);
+                Some((node.id, pane_width_ms * u64::from(*pane_count) + span_ms))
+            }
+            _ => None,
+        })
+        .collect()
+}
+
+/// The panes a chain in `retention_ms` ([`pane_retention_ms`]) keeps
+/// completed: its retention over its pane width, at least its pane count.
+fn retained_panes(
+    node: &asap_types::ir::physical_export::PhysicalASAPDAGNode,
+    retention_ms: &BTreeMap<PhysicalASAPNodeId, u64>,
+) -> u64 {
+    match (&node.payload, retention_ms.get(&node.id)) {
+        (Payload::ASAP(ASAPOp::PaneChain { pane_width_ms, .. }), Some(&retention)) => {
+            (retention / pane_width_ms.max(&1)).max(panes_of(node))
+        }
+        _ => panes_of(node),
+    }
+}
+
 /// A node's panes: a pane chain's count, 1 for any other node.
 fn panes_of(node: &asap_types::ir::physical_export::PhysicalASAPDAGNode) -> u64 {
     match &node.payload {
@@ -3251,6 +3413,7 @@ mod tests {
             &node_map(dag),
             &kept_pane_chains(dag),
             recurrence,
+            &[],
             data,
             settings,
         )
@@ -5796,6 +5959,40 @@ mod tests {
         assert!(close(total(&two, &mixed), 1.5 * single));
     }
 
+    /// Queries that refresh together at one interval and step evaluate at
+    /// the same instants: a node they share runs at the most steps among
+    /// them, not their sum. One over the last hour (61 steps) and one over
+    /// the last 10 min (11 steps), both every minute at 1-min steps, and a
+    /// third with no range (its one evaluation is the others' last step):
+    /// 61 evaluations a minute. A different step adds only the instants
+    /// the others do not evaluate at.
+    #[test]
+    fn readers_at_one_interval_and_step_share_their_steps() {
+        let every_minute = repeating(60_000);
+        let range = |span_ms: u64, step_ms: u64| Some(RangeEvaluation { step_ms, span_ms });
+        let rate = |queries: &[Option<RangeEvaluation>]| {
+            evaluation_rate(queries.iter().map(|r| (&every_minute, *r)), 3_600.0).unwrap()
+        };
+        let close = |a: f64, b: f64| (a - b).abs() < 1e-12;
+        assert!(close(rate(&[range(3_600_000, 60_000)]), 61.0 / 60.0));
+        assert!(close(
+            rate(&[range(3_600_000, 60_000), range(600_000, 60_000), None]),
+            61.0 / 60.0
+        ));
+        assert!(close(rate(&[None, None]), 1.0 / 60.0));
+        // A one-step range evaluates only at the request's end.
+        assert!(close(
+            rate(&[range(3_600_000, 60_000), range(0, 30_000)]),
+            61.0 / 60.0
+        ));
+        // A 30 s step over the last 10 min shares 11 of its 21 instants
+        // with the 1-min steps over the last hour: 71 distinct instants.
+        assert!(close(
+            rate(&[range(3_600_000, 60_000), range(600_000, 30_000)]),
+            71.0 / 60.0
+        ));
+    }
+
     /// One-off and unknown recurrence are spread over the horizon:
     /// invocations × per-evaluation cost / non_repeating_query_horizon_seconds.
     #[test]
diff --git a/crates/plan-selection/src/pass/stage1_to_stage3.rs b/crates/plan-selection/src/pass/stage1_to_stage3.rs
index 13982334..e73976a4 100644
--- a/crates/plan-selection/src/pass/stage1_to_stage3.rs
+++ b/crates/plan-selection/src/pass/stage1_to_stage3.rs
@@ -163,8 +163,7 @@ pub fn run_stage1_to_stage3_from_pass1<Id: Clone + std::fmt::Debug>(
 ) -> Stage1ToStage3Run<Id> {
     let per_computation =
         per_computation.without_unsupported_summaries(models.deployment_capabilities);
-    let recurrence: Vec<_> = entries.iter().map(|e| e.recurrence.clone()).collect();
-    let stage1 = stage1_logical_asap_optimization(&per_computation, &recurrence).collect();
+    let stage1 = stage1_logical_asap_optimization(&per_computation, entries).collect();
     run_stage2_and_stage3(per_computation, stage1, entries, data, models)
 }
 
diff --git a/crates/planner/tests/deployment_accuracy_model.rs b/crates/planner/tests/deployment_accuracy_model.rs
index 0a432945..1417611a 100644
--- a/crates/planner/tests/deployment_accuracy_model.rs
+++ b/crates/planner/tests/deployment_accuracy_model.rs
@@ -44,6 +44,7 @@ fn rank_target_workload(query: &str) -> PlanningWorkload {
                     max_latency_ms: None,
                 },
                 predictability: Predictability::Unknown,
+                range_evaluation: None,
             }]),
         },
         data_workload: Some(DataWorkload {
diff --git a/crates/planner/tests/e2e_plan.rs b/crates/planner/tests/e2e_plan.rs
index c0b6a510..fe9b866a 100644
--- a/crates/planner/tests/e2e_plan.rs
+++ b/crates/planner/tests/e2e_plan.rs
@@ -42,6 +42,7 @@ fn batch(sql: &str) -> BatchEntry {
         predictability: Predictability::AdHoc,
         invocations: 1,
         execute_at: None,
+        range_evaluation: None,
     }
 }
 
@@ -173,6 +174,7 @@ async fn lowers_repeating_sql_entries_too() {
             demand: RepeatedDemand::FixedInterval(RepetitionInterval(60_000)),
             requirements: approximate_for("approx_percentile_cont"),
             predictability: Predictability::Unknown,
+            range_evaluation: None,
         }]),
     );
     let catalog = lineitem_catalog();
@@ -310,6 +312,7 @@ async fn each_plan_carries_its_selected_root() {
                 demand: RepeatedDemand::FixedInterval(RepetitionInterval(60_000)),
                 requirements: approximate(),
                 predictability: Predictability::Unknown,
+                range_evaluation: None,
             }]),
         },
         // A range selector only: no sample interval is needed.
diff --git a/crates/planner/tests/plan_stages.rs b/crates/planner/tests/plan_stages.rs
index 4d22c85d..aa6b89a7 100644
--- a/crates/planner/tests/plan_stages.rs
+++ b/crates/planner/tests/plan_stages.rs
@@ -23,6 +23,7 @@ fn promql_stages(query: &str, epsilon: f64) -> Stage1ToStage3Run<usize> {
                 predictability: Default::default(),
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },
diff --git a/crates/planner/tests/summary_sharing.rs b/crates/planner/tests/summary_sharing.rs
index c4c2317c..1f777109 100644
--- a/crates/planner/tests/summary_sharing.rs
+++ b/crates/planner/tests/summary_sharing.rs
@@ -50,6 +50,7 @@ fn repeating(query: &str, epsilon: f64) -> RepeatingEntry {
         demand: RepeatedDemand::FixedInterval(RepetitionInterval(600_000)),
         requirements: requirements(query, epsilon),
         predictability: Predictability::Unknown,
+        range_evaluation: None,
     }
 }
 
diff --git a/crates/types/src/workload/mod.rs b/crates/types/src/workload/mod.rs
index 1a51c005..4b5092f2 100644
--- a/crates/types/src/workload/mod.rs
+++ b/crates/types/src/workload/mod.rs
@@ -211,6 +211,57 @@ impl QueryRecurrence {
     }
 }
 
+/// The most evaluation steps one range request may have, as Prometheus
+/// allows for `/api/v1/query_range` (it rejects a request with more).
+pub const MAX_RANGE_EVALUATION_STEPS: u64 = 11_000;
+
+/// How one request of a query is answered as a range request, as
+/// Prometheus' `/api/v1/query_range` does: it evaluates the query at
+/// `start = end − span_ms` and every `step_ms` after, up to and including
+/// the request's `end`, e.g. a dashboard panel over the last hour at
+/// 1-minute steps (`...&start=now-1h&end=now&step=60`). `span_ms` must be a
+/// whole number of steps, so the last step is the request's end.
+/// [`QueryRecurrence`] still says how often a request runs.
+///
+/// JSON: `{"step_ms": 60000, "span_ms": 3600000}`.
+///
+/// # Example
+///
+/// ```
+/// use asap_types::workload::RangeEvaluation;
+///
+/// let last_hour_by_minute = RangeEvaluation { step_ms: 60_000, span_ms: 3_600_000 };
+/// assert_eq!(last_hour_by_minute.evaluations_per_request(), 61);
+/// ```
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct RangeEvaluation {
+    /// Time between consecutive evaluation steps of one request, in
+    /// milliseconds; must be positive.
+    pub step_ms: u64,
+    /// How far before the request's end its first step is, in
+    /// milliseconds: a multiple of `step_ms`, 0 for one step per request.
+    pub span_ms: u64,
+}
+
+impl RangeEvaluation {
+    /// Steps one request evaluates, `span_ms / step_ms + 1`: 61 for a 1 h
+    /// span at 1 min steps. Stage 3 prices each step as one evaluation.
+    pub fn evaluations_per_request(&self) -> u64 {
+        self.span_ms / self.step_ms.max(1) + 1
+    }
+}
+
+/// The greatest common divisor of `a` and `b` (`a` when `b` is 0): the
+/// grid two cadences in milliseconds share.
+pub fn greatest_common_divisor(a: u64, b: u64) -> u64 {
+    if b == 0 {
+        a
+    } else {
+        greatest_common_divisor(b, a % b)
+    }
+}
+
 /// One query submitted as part of a finite batch, including how many times it
 /// will run and, when known, its execution time.
 ///
@@ -227,6 +278,7 @@ impl QueryRecurrence {
 ///     },
 ///     invocations: 1,
 ///     execute_at: Some(TimestampMs(10_000)),
+///     range_evaluation: None,
 /// };
 /// ```
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -237,6 +289,10 @@ pub struct BatchEntry {
     pub predictability: Predictability,
     pub invocations: u64,
     pub execute_at: Option<TimestampMs>,
+    /// How each invocation is answered as a range request: `None` (the
+    /// default) for one evaluation per invocation.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub range_evaluation: Option<RangeEvaluation>,
 }
 
 /// One query with repeated demand. Its recurrence does not imply that the
@@ -253,6 +309,8 @@ pub struct BatchEntry {
 ///     demand: RepeatedDemand::FixedInterval(RepetitionInterval(10_000)),
 ///     requirements: QueryRequirements::default(),
 ///     predictability: Predictability::Predictable { known_at: None },
+///     // Each refresh evaluates the last hour at 1-minute steps.
+///     range_evaluation: Some(RangeEvaluation { step_ms: 60_000, span_ms: 3_600_000 }),
 /// };
 /// ```
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -262,6 +320,10 @@ pub struct RepeatingEntry {
     pub demand: RepeatedDemand,
     pub requirements: QueryRequirements,
     pub predictability: Predictability,
+    /// How each request is answered as a range request: `None` (the
+    /// default) for one evaluation per request.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub range_evaluation: Option<RangeEvaluation>,
 }
 
 /// Planner-facing normalized form of either [`BatchEntry`] or
@@ -283,6 +345,7 @@ pub struct RepeatingEntry {
 ///     recurrence: QueryRecurrence::Repeated(
 ///         RepeatedDemand::FixedInterval(RepetitionInterval(10_000)),
 ///     ),
+///     range_evaluation: None,
 /// };
 /// ```
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -298,6 +361,30 @@ pub struct QueryWorkloadEntry {
     /// How often it runs: sets how often each node is priced, and only
     /// repeated queries may keep state from ingestion time (default: unknown).
     pub recurrence: QueryRecurrence,
+    /// How each run is answered as a range request, one evaluation per
+    /// step; `None` (the default) for one evaluation per run. Stage 1 and
+    /// Stage 2 keep pane widths on its step grid, and Stage 3 prices
+    /// [`RangeEvaluation::evaluations_per_request`] evaluations per run.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub range_evaluation: Option<RangeEvaluation>,
+}
+
+impl QueryWorkloadEntry {
+    /// The grid every evaluation of this query falls on, in milliseconds:
+    /// its fixed interval ([`QueryRecurrence::fixed_interval_ms`]), or with
+    /// a range evaluation of more than one step the greatest common divisor
+    /// of the interval and the step, since each run evaluates at every step
+    /// up to its end. `None` when it does not repeat at a positive fixed
+    /// interval.
+    pub fn evaluation_grid_ms(&self) -> Option<u64> {
+        let interval_ms = self.recurrence.fixed_interval_ms()?;
+        Some(match self.range_evaluation {
+            Some(range) if range.step_ms > 0 && range.evaluations_per_request() > 1 => {
+                greatest_common_divisor(interval_ms, range.step_ms)
+            }
+            _ => interval_ms,
+        })
+    }
 }
 
 impl From<&BatchEntry> for QueryWorkloadEntry {
@@ -310,6 +397,7 @@ impl From<&BatchEntry> for QueryWorkloadEntry {
                 invocations: entry.invocations,
                 execute_at: entry.execute_at,
             },
+            range_evaluation: entry.range_evaluation,
         }
     }
 }
@@ -321,6 +409,7 @@ impl From<&RepeatingEntry> for QueryWorkloadEntry {
             requirements: entry.requirements.clone(),
             predictability: entry.predictability.clone(),
             recurrence: QueryRecurrence::Repeated(entry.demand.clone()),
+            range_evaluation: entry.range_evaluation,
         }
     }
 }
@@ -614,6 +703,21 @@ fn validate_entry(entry: &QueryWorkloadEntry) -> Result<(), WorkloadError> {
         }
         _ => {}
     }
+    if let Some(range) = entry.range_evaluation {
+        if range.step_ms == 0 {
+            return Err(WorkloadError::ZeroRangeEvaluationStep);
+        }
+        if range.span_ms % range.step_ms != 0 {
+            return Err(WorkloadError::RangeEvaluationSpanNotMultipleOfStep {
+                span_ms: range.span_ms,
+                step_ms: range.step_ms,
+            });
+        }
+        let steps = range.evaluations_per_request();
+        if steps > MAX_RANGE_EVALUATION_STEPS {
+            return Err(WorkloadError::TooManyRangeEvaluationSteps { steps });
+        }
+    }
     Ok(())
 }
 
@@ -649,6 +753,32 @@ pub enum WorkloadError {
     /// Bad input: a repeating query whose fixed interval is zero.
     #[error("a fixed repetition interval must be greater than zero")]
     ZeroRepetitionInterval,
+    /// Bad input: a range evaluation whose `step_ms` is zero.
+    #[error("a range evaluation's step_ms must be greater than zero")]
+    ZeroRangeEvaluationStep,
+    /// Bad input: a range evaluation whose `span_ms` is not a whole number
+    /// of steps. Prometheus places steps forward from the start, so its
+    /// last step would not be the request's end and would fall off the
+    /// end's grid.
+    #[error(
+        "a range evaluation's span_ms ({span_ms}) must be a multiple of its step_ms ({step_ms})"
+    )]
+    RangeEvaluationSpanNotMultipleOfStep {
+        /// The declared span.
+        span_ms: u64,
+        /// The declared step.
+        step_ms: u64,
+    },
+    /// Bad input: a range evaluation with more steps per request than
+    /// [`MAX_RANGE_EVALUATION_STEPS`], which Prometheus would reject.
+    #[error(
+        "a range evaluation has {steps} steps per request, more than {}",
+        MAX_RANGE_EVALUATION_STEPS
+    )]
+    TooManyRangeEvaluationSteps {
+        /// The steps one request would evaluate, `span_ms / step_ms + 1`.
+        steps: u64,
+    },
     /// Bad input: a scheduled repeating query with no timestamps.
     #[error("a repeated-query schedule must not be empty")]
     EmptySchedule,
@@ -850,6 +980,7 @@ mod tests {
             demand: RepeatedDemand::FixedInterval(RepetitionInterval(0)),
             requirements: QueryRequirements::default(),
             predictability: Predictability::Predictable { known_at: None },
+            range_evaluation: None,
         }]);
         assert_eq!(
             workload.validate(),
@@ -857,6 +988,109 @@ mod tests {
         );
     }
 
+    /// A range evaluation is optional JSON on a repeating entry, carried to
+    /// its normalized entry; its step must be positive and a request may
+    /// have at most `MAX_RANGE_EVALUATION_STEPS` steps. An entry without
+    /// one serializes as before, with no `range_evaluation` key.
+    #[test]
+    fn range_evaluation_is_optional_carried_and_validated() {
+        let dashboard = |range: serde_json::Value| {
+            let mut entry = serde_json::json!({
+                "query": "rate(requests_total[5m])",
+                "demand": {"fixed_interval": 30000},
+                "requirements": {},
+                "predictability": {"predictable": {"known_at": null}},
+            });
+            if !range.is_null() {
+                entry["range_evaluation"] = range;
+            }
+            serde_json::from_value::<RepeatingEntry>(entry).unwrap()
+        };
+        let plain = dashboard(serde_json::Value::Null);
+        assert_eq!(plain.range_evaluation, None);
+        assert!(serde_json::to_value(&plain)
+            .unwrap()
+            .get("range_evaluation")
+            .is_none());
+
+        let hour = dashboard(serde_json::json!({"step_ms": 60000, "span_ms": 3600000}));
+        let range = RangeEvaluation {
+            step_ms: 60_000,
+            span_ms: 3_600_000,
+        };
+        assert_eq!(hour.range_evaluation, Some(range));
+        assert_eq!(range.evaluations_per_request(), 61);
+        let entry = QueryWorkloadEntry::from(&hour);
+        assert_eq!(entry.range_evaluation, Some(range));
+        // Refreshes every 30 s and steps every 60 s: evaluations on a 30 s grid.
+        assert_eq!(entry.evaluation_grid_ms(), Some(30_000));
+        assert_eq!(
+            QueryWorkloadEntry::from(&plain).evaluation_grid_ms(),
+            Some(30_000)
+        );
+        // A one-step range (span 0) is one evaluation per request: its
+        // step does not narrow the grid (refreshing every 60 s at a 45 s
+        // step would otherwise give a 15 s grid for nothing).
+        let one_step = QueryWorkloadEntry {
+            range_evaluation: Some(RangeEvaluation {
+                step_ms: 45_000,
+                span_ms: 0,
+            }),
+            recurrence: QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(
+                RepetitionInterval(60_000),
+            )),
+            ..entry.clone()
+        };
+        assert_eq!(one_step.evaluation_grid_ms(), Some(60_000));
+
+        let validate = |range: RangeEvaluation| {
+            let mut workload = base_workload();
+            workload.repeating_queries = Some(vec![RepeatingEntry {
+                range_evaluation: Some(range),
+                ..hour.clone()
+            }]);
+            workload.validate()
+        };
+        assert_eq!(validate(range), Ok(()));
+        assert_eq!(
+            validate(RangeEvaluation {
+                step_ms: 0,
+                span_ms: 60_000
+            }),
+            Err(WorkloadError::ZeroRangeEvaluationStep)
+        );
+        assert_eq!(
+            validate(RangeEvaluation {
+                step_ms: 1_000,
+                span_ms: 11_000_000
+            }),
+            Err(WorkloadError::TooManyRangeEvaluationSteps { steps: 11_001 })
+        );
+        assert_eq!(
+            validate(RangeEvaluation {
+                step_ms: 1_000,
+                span_ms: 10_999_000
+            }),
+            Ok(())
+        );
+        // Prometheus places steps forward from the start, so a span that is
+        // not a whole number of steps would put the last step off the end's
+        // grid: refreshing every 60 s at 60 s steps over 90 s evaluates at
+        // E − 90 s and E − 30 s. It is rejected.
+        assert_eq!(
+            validate(RangeEvaluation {
+                step_ms: 60_000,
+                span_ms: 90_000
+            }),
+            Err(WorkloadError::RangeEvaluationSpanNotMultipleOfStep {
+                span_ms: 90_000,
+                step_ms: 60_000
+            })
+        );
+        let unknown = serde_json::json!({"step_ms": 1, "span_ms": 0, "start": 0});
+        assert!(serde_json::from_value::<RangeEvaluation>(unknown).is_err());
+    }
+
     fn base_workload() -> QueryWorkload {
         QueryWorkload {
             language: QueryLanguage::PromQL,
@@ -874,12 +1108,14 @@ mod tests {
             predictability: Predictability::AdHoc,
             invocations: 1,
             execute_at: None,
+            range_evaluation: None,
         }]);
         workload.repeating_queries = Some(vec![RepeatingEntry {
             query: Query("dashboard".into()),
             demand: RepeatedDemand::FixedInterval(RepetitionInterval(10_000)),
             requirements: QueryRequirements::default(),
             predictability: Predictability::Predictable { known_at: None },
+            range_evaluation: None,
         }]);
 
         let entries: Vec<_> = workload.entries().collect();
@@ -920,6 +1156,7 @@ mod tests {
                         max_latency_ms: Some(100.0),
                     },
                     predictability: Predictability::Predictable { known_at: None },
+                    range_evaluation: None,
                 }]),
             },
             data_workload: Some(DataWorkload {
@@ -1062,6 +1299,7 @@ mod tests {
             }),
             requirements: QueryRequirements::default(),
             predictability: Predictability::Unknown,
+            range_evaluation: None,
         }]);
         assert_eq!(workload.validate(), Err(WorkloadError::InvalidRate(-1.0)));
     }
diff --git a/tools/dag-viewer/examples/planner-layering-example1.json b/tools/dag-viewer/examples/planner-layering-example1.json
index 709134cf..3241349d 100644
--- a/tools/dag-viewer/examples/planner-layering-example1.json
+++ b/tools/dag-viewer/examples/planner-layering-example1.json
@@ -105774,6 +105774,7 @@
       }
     },
     "front": [],
+    "pane_retention_ms": {},
     "rejected": [
       {
         "cause": "no_accuracy_guarantee",