diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs
index 6582b79b..5e9b62ad 100644
--- a/control_plane/src/emit/stage_config.rs
+++ b/control_plane/src/emit/stage_config.rs
@@ -2991,6 +2991,11 @@ pub(crate) fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonVa
             obj.insert("weight_mode".into(), JsonValue::String(mode.into()));
         }
     }
+    // PromQL range selectors are (start, end]. Encode the boundary convention
+    // in state identity so legacy half-open panes cannot satisfy this binding.
+    if matches!(agg.aggregation_input, AggregationInput::Raw) {
+        parameters["promql_right_closed"] = json!(true);
+    }
     let aggregation_input = match agg.aggregation_input {
         AggregationInput::SketchEnvelope => "sketch_envelope",
         AggregationInput::Raw => "raw",
diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs
index be6dc44d..cf08aa43 100644
--- a/control_plane/src/main.rs
+++ b/control_plane/src/main.rs
@@ -826,6 +826,7 @@ fn compile_physical_plan_request(
     }
 
     let planning_request = physical::compiler::PlanningRequest {
+        query_workload: None,
         queries,
         evidence: request.evidence,
         planner_revision: request.planner_revision,
diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index 890e33c7..b5e9278c 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -126,6 +126,8 @@ pub struct LifecyclePlanningInput {
 
 #[derive(Debug, Clone, Default)]
 pub struct PlanningRequest {
+    /// Original dashboard demand, in the same order as queries. None is legacy input.
+    pub query_workload: Option<QueryWorkload>,
     pub queries: Vec<PlanningQuery>,
     pub evidence: HashMap<String, TopKMembershipEvidence>,
     pub planner_revision: String,
@@ -180,6 +182,9 @@ pub struct BackendLocalPlanningSnapshot {
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 #[serde(deny_unknown_fields)]
 pub struct BackendLocalImplementation {
+    /// Provider-priced concrete pane choices keyed by the registered PromQL text.
+    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
+    pub window_candidates: HashMap<String, Vec<WindowImplementationCandidate>>,
     pub lifecycle_costs: LifecycleCostEvidence,
     pub evidence_observed_at_unix_ms: u64,
     pub evidence_valid_for_ms: u64,
@@ -1600,6 +1605,13 @@ impl BackendLocalPlanningSnapshot {
                 "QueryWorkload must contain at least one query".into(),
             ));
         }
+        for query in self.implementation.window_candidates.keys() {
+            if !entries.iter().any(|entry| &entry.query.0 == query) {
+                return Err(CompileError::Snapshot(format!(
+                    "window candidates reference unregistered query `{query}`"
+                )));
+            }
+        }
         let mut queries = Vec::with_capacity(entries.len());
         let mut canonical_roots = Vec::with_capacity(entries.len());
         let mut topk_evidence_by_id = HashMap::new();
@@ -1661,7 +1673,7 @@ impl BackendLocalPlanningSnapshot {
             }
             queries.push(PlanningQuery {
                 query_id,
-                query_string,
+                query_string: query_string.clone(),
                 post_asap,
                 source: Source::TimeSeries {
                     metric: metadata.metric_name,
@@ -1670,14 +1682,21 @@ impl BackendLocalPlanningSnapshot {
                 group_by: metadata.group_by_labels,
                 accuracy,
                 lifecycle,
-                window_implementations: vec![WindowImplementationCandidate {
-                    implementation_id: self.implementation.window_implementation_id.clone(),
-                    framework: SummaryWindowFramework::Tumbling,
-                    window_secs: lookback_ms / 1_000,
-                    pane_secs: lookback_ms / 1_000,
-                    state_layout: self.implementation.state_layout.clone(),
-                    cost,
-                }],
+                window_implementations: self
+                    .implementation
+                    .window_candidates
+                    .get(&query_string)
+                    .cloned()
+                    .unwrap_or_else(|| {
+                        vec![WindowImplementationCandidate {
+                            implementation_id: self.implementation.window_implementation_id.clone(),
+                            framework: SummaryWindowFramework::Tumbling,
+                            window_secs: lookback_ms / 1_000,
+                            pane_secs: lookback_ms / 1_000,
+                            state_layout: self.implementation.state_layout.clone(),
+                            cost,
+                        }]
+                    }),
                 runtime_policy: RuntimeRulePolicy::default(),
             });
         }
@@ -1685,6 +1704,7 @@ impl BackendLocalPlanningSnapshot {
         preserve_native_unsafe_raw_roots(&mut queries)?;
         Ok((
             PlanningRequest {
+                query_workload: Some(workload),
                 queries,
                 evidence: topk_evidence_by_id,
                 planner_revision: PLANNER_REVISION.into(),
@@ -1808,6 +1828,18 @@ impl PhysicalCompiler {
             });
         }
 
+        if let Some(workload) = &request.query_workload {
+            let entries = workload.entries().collect::<Vec<_>>();
+            if entries.len() != request.queries.len()
+                || entries
+                    .iter()
+                    .zip(&request.queries)
+                    .any(|(entry, query)| entry.query.0 != query.query_string)
+            {
+                return Err(CompileError::Snapshot("original workload and planning queries must have identical order and query text".into()));
+            }
+        }
+
         if environment.target == PhysicalDeploymentTarget::BackendLocalRemoteWrite {
             preserve_native_unsafe_raw_roots(&mut request.queries)?;
         }
@@ -1833,7 +1865,8 @@ impl PhysicalCompiler {
         // BackendPlan candidates to rediscover this decision.
         let mut node_bindings = HashMap::<usize, asap_types::PolicyFingerprint>::new();
         let consumers = materialization_consumers(&request.queries, environment.target)?;
-        let mut lifecycle_estimates = BTreeMap::new();
+        let mut lifecycle_estimates =
+            BTreeMap::<asap_types::PolicyFingerprint, MaterializationLifecycleEstimate>::new();
 
         for query in &request.queries {
             let evidence = request.evidence.get(&query.query_id);
@@ -1868,10 +1901,12 @@ impl PhysicalCompiler {
                     SummaryMaintenanceCapabilities {
                         incremental_update: true,
                         merge: true,
-                        delete: false,
+                        // Expired whole panes are excluded when reading a moving scope.
+                        delete: environment.target
+                            == PhysicalDeploymentTarget::BackendLocalRemoteWrite,
                     },
                 )
-                .with_window_framework_costs(window_costs);
+                .with_window_implementation_costs(window_costs);
             if environment.target == PhysicalDeploymentTarget::DistributedCollectors
                 && selected.iter().any(|state| {
                     matches!(
@@ -1911,7 +1946,7 @@ impl PhysicalCompiler {
                     }
                     _ => selected.algorithm.clone(),
                 };
-                let aggregation = physical_aggregation(
+                let mut aggregation = physical_aggregation(
                     query,
                     &selected,
                     aggregation_id.clone(),
@@ -1930,22 +1965,47 @@ impl PhysicalCompiler {
                     &model,
                     &environment,
                     &state_consumers,
+                    request.query_workload.as_ref().map(|workload| {
+                        (
+                            workload,
+                            consumers[&materialization].iter().copied().collect(),
+                        )
+                    }),
                 )?;
                 let window_implementation = query.window_implementations.iter()
-                    .filter(|candidate| candidate.framework == planner_selection.window_framework)
-                    .min_by(|left, right| left.cost.weighted_cost.total_cmp(&right.cost.weighted_cost))
+                    .find(|candidate| candidate.implementation_id == planner_selection.window_implementation_id
+                        && candidate.framework == planner_selection.window_framework)
                     .ok_or_else(|| CompileError::Lifecycle {
                         query_id: query.query_id.clone(),
                         reason: "Planner selected a window framework without a retained concrete implementation".into(),
                     })?;
+                // The logical consumer group is identified above. The installed state
+                // identity includes the actual pane width selected by Planner.
+                aggregation.window_secs = window_implementation.pane_secs;
+                let materialization =
+                    backend_plan::aggregation_config_for_materialization(&aggregation)?
+                        .policy_fingerprint();
+                let consumer_query_ids = state_consumers
+                    .iter()
+                    .map(|query| query.query_id.clone())
+                    .collect::<Vec<_>>();
+                // Lifecycle demand was priced before choosing pane width. Two
+                // distinct logical cohorts can now collide on one physical
+                // fingerprint; retaining either quote would omit consumers.
+                if lifecycle_estimates
+                    .get(&materialization)
+                    .is_some_and(|existing| existing.consumer_query_ids != consumer_query_ids)
+                {
+                    return Err(CompileError::Lifecycle {
+                        query_id: query.query_id.clone(),
+                        reason: "selected pane coalesces distinct logical consumer cohorts; joint physical-pane lifecycle evidence is required".into(),
+                    });
+                }
                 lifecycle_estimates
                     .entry(materialization)
                     .or_insert_with(|| MaterializationLifecycleEstimate {
                         materialization,
-                        consumer_query_ids: state_consumers
-                            .iter()
-                            .map(|query| query.query_id.clone())
-                            .collect(),
+                        consumer_query_ids,
                         window_implementation_id: window_implementation.implementation_id.clone(),
                         horizon_seconds: query.lifecycle.horizon_seconds,
                         expected_reads: planner_selection.expected_reads,
@@ -1987,7 +2047,7 @@ impl PhysicalCompiler {
                     algorithm: physical_algorithm,
                     parameters: selected.parameters,
                     group_by: query.group_by.clone(),
-                    window_secs: query.window_secs,
+                    window_secs: window_implementation.pane_secs,
                     abstract_window_framework: planner_selection.window_framework.clone(),
                     window_implementation_id: window_implementation.implementation_id.clone(),
                     pane_secs: window_implementation.pane_secs,
@@ -2151,7 +2211,9 @@ impl PhysicalCompiler {
                             ))
                         })?;
                     if materialization.family != physical_materialization_family(node_family)
-                        || materialization.window.size_ms != query.window_secs.saturating_mul(1_000)
+                        || materialization.window.size_ms == 0
+                        || query.window_secs.saturating_mul(1_000) % materialization.window.size_ms
+                            != 0
                         || materialization.group_by != query.group_by
                     {
                         return Err(crate::query_plan::QueryPlanError::Invalid(format!(
@@ -2164,7 +2226,7 @@ impl PhysicalCompiler {
                         metric: planned_metric,
                         sid_grouping: query.group_by.clone(),
                         output_grouping: PhysicalGrouping::Reduce(query.group_by.clone()),
-                        window_ms: query.window_secs.saturating_mul(1_000),
+                        window_ms: materialization.window.size_ms,
                     })
                 },
             )?;
@@ -2394,9 +2456,9 @@ fn validate_lifecycle_input(
 fn validate_window_implementations(
     query: &PlanningQuery,
     environment: &DeploymentEnvironment,
-) -> Result<Vec<(SummaryWindowFramework, Cost)>, CompileError> {
+) -> Result<Vec<(String, SummaryWindowFramework, Cost)>, CompileError> {
     let mut ids = BTreeSet::new();
-    let mut cheapest = BTreeMap::<SummaryWindowFramework, f64>::new();
+    let mut candidates = Vec::new();
     for candidate in &query.window_implementations {
         let evidence = &candidate.cost;
         let age = environment
@@ -2420,11 +2482,15 @@ fn validate_window_implementations(
             && candidate.pane_secs != 0
             && candidate.pane_secs <= candidate.window_secs
             && candidate.window_secs % candidate.pane_secs == 0
-            // Current Collector runtime contract is the MVP's anchored,
-            // tumbling implementation. Other Planner primitives become
-            // candidates only when an executor advertises full semantics.
             && candidate.framework == SummaryWindowFramework::Tumbling
-            && candidate.pane_secs == candidate.window_secs;
+            && match environment.target {
+                PhysicalDeploymentTarget::DistributedCollectors => {
+                    candidate.pane_secs == candidate.window_secs
+                }
+                // Cadence does not establish phase alignment. Serving checks each
+                // actual interval and falls back when whole panes cannot cover it.
+                PhysicalDeploymentTarget::BackendLocalRemoteWrite => true,
+            };
         if !valid {
             return Err(CompileError::Lifecycle {
                 query_id: query.query_id.clone(),
@@ -2434,24 +2500,23 @@ fn validate_window_implementations(
                 ),
             });
         }
-        cheapest
-            .entry(candidate.framework.clone())
-            .and_modify(|cost| *cost = cost.min(evidence.weighted_cost))
-            .or_insert(evidence.weighted_cost);
+        candidates.push((
+            candidate.implementation_id.clone(),
+            candidate.framework.clone(),
+            Cost(evidence.weighted_cost),
+        ));
     }
-    if cheapest.is_empty() {
+    if candidates.is_empty() {
         return Err(CompileError::Lifecycle {
             query_id: query.query_id.clone(),
             reason: "no complete executor-feasible window implementation evidence".into(),
         });
     }
-    Ok(cheapest
-        .into_iter()
-        .map(|(framework, cost)| (framework, Cost(cost)))
-        .collect())
+    Ok(candidates)
 }
 
 struct PlannerPhysicalSelection {
+    window_implementation_id: String,
     lifecycle: CollectorLifecycle,
     window_framework: SummaryWindowFramework,
     expected_reads: f64,
@@ -2465,6 +2530,7 @@ fn select_lifecycle(
     model: &ControlPlaneCostModel,
     environment: &DeploymentEnvironment,
     consumers: &[&PlanningQuery],
+    original_workload: Option<(&QueryWorkload, Vec<usize>)>,
 ) -> Result<PlannerPhysicalSelection, CompileError> {
     // Current lifecycle evidence is per producer with one unit read cost.
     // Conflicting source/rate/horizon/cost snapshots cannot be averaged into
@@ -2518,9 +2584,11 @@ fn select_lifecycle(
             ..DataWorkload::default()
         }),
     };
+    let (workload, indices) =
+        original_workload.unwrap_or((&workload, (0..consumers.len()).collect()));
     let plan = plan_summary_maintenance_lifecycles(
         Rc::new(node.clone()),
-        WorkloadDemand::new(&workload, &(0..consumers.len()).collect::<Vec<_>>()),
+        WorkloadDemand::new(workload, &indices),
         environment.observed_at_unix_ms,
         Some(Horizon(query.lifecycle.horizon_seconds)),
         SummaryMaintenanceLifecycleCapabilities {
@@ -2552,6 +2620,12 @@ fn select_lifecycle(
             reason: "latest ASAPPlanner selected no window framework from the supplied physical evidence".into(),
         })?;
     Ok(PlannerPhysicalSelection {
+        window_implementation_id: plan.selected_physical_plan_id.clone().ok_or_else(|| {
+            CompileError::Lifecycle {
+                query_id: query.query_id.clone(),
+                reason: "Planner returned no concrete window implementation identity".into(),
+            }
+        })?,
         expected_reads: plan.expected_reads.ok_or_else(|| CompileError::Lifecycle {
             query_id: query.query_id.clone(),
             reason: "missing joint read demand".into(),
@@ -3045,6 +3119,7 @@ mod tests {
             evidence_by_query.insert(query_id.to_string(), evidence);
         }
         Ok(PlanningRequest {
+            query_workload: None,
             queries: vec![PlanningQuery {
                 query_id: query_id.into(),
                 query_string: promql.into(),
@@ -3584,6 +3659,7 @@ mod tests {
             query_workload,
             data_workload,
             implementation: BackendLocalImplementation {
+                window_candidates: HashMap::new(),
                 lifecycle_costs: template.lifecycle.costs,
                 evidence_observed_at_unix_ms: 9_500,
                 evidence_valid_for_ms: 60_000,
@@ -3595,6 +3671,16 @@ mod tests {
             },
             environment,
         };
+        assert_eq!(
+            snapshot
+                .clone()
+                .planning_request()
+                .unwrap()
+                .0
+                .query_workload
+                .as_ref(),
+            Some(&snapshot.query_workload)
+        );
         let first = snapshot
             .clone()
             .compile()
@@ -3833,6 +3919,81 @@ mod tests {
         ));
     }
 
+    // Concrete candidates with the same framework survive selection; changing
+    // their quoted costs changes installed state, not the query's lookback.
+    #[test]
+    fn tumbling_sizes_are_selected_by_cost_and_installed() {
+        for (small_cost, expected_secs, expected_id) in [(0.1, 10, "small"), (10.0, 60, "large")] {
+            let mut request = request("q", "sum(sum_over_time(m[1m]))");
+            let query = &mut request.queries[0];
+            let mut small = query.window_implementations[0].clone();
+            small.implementation_id = "small".into();
+            small.pane_secs = 10;
+            small.cost.weighted_cost = small_cost;
+            query.window_implementations[0].implementation_id = "large".into();
+            query.window_implementations.push(small);
+            let mut env = environment(10_000);
+            env.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite;
+            env.collector_ids.clear();
+            let bundle = PhysicalCompiler.compile(request, env).unwrap();
+            let materialization = bundle
+                .backend_plan
+                .materializations
+                .values()
+                .next()
+                .unwrap();
+            assert_eq!(materialization.window.size_ms, expected_secs * 1000);
+            let entry = bundle
+                .query_plan
+                .lookup("sum(sum_over_time(m[1m]))")
+                .unwrap();
+            assert_eq!(entry.instant.lookback_ms, 60_000);
+            assert_eq!(
+                entry.materialization_bindings()[0].window_ms,
+                expected_secs * 1000
+            );
+            assert_eq!(
+                bundle.lifecycle_estimates[0].window_implementation_id,
+                expected_id
+            );
+        }
+    }
+
+    // Distinct logical cohorts cannot silently coalesce using only the first quote.
+    #[test]
+    fn selected_panes_reject_unpriced_cross_cohort_coalescing() {
+        let mut workload = request("q20", "sum(sum_over_time(m[20s]))");
+        let mut second = request("q40", "sum(sum_over_time(m[40s]))")
+            .queries
+            .remove(0);
+        workload.queries[0].window_secs = 20;
+        workload.queries[0].window_implementations[0].window_secs = 20;
+        second.window_secs = 40;
+        second.lifecycle.evaluation_interval_ms = 20_000;
+        second.window_implementations[0].window_secs = 40;
+        workload.queries.push(second);
+        for query in &mut workload.queries {
+            query.window_implementations[0].pane_secs = 10;
+            query.window_implementations[0].implementation_id = "shared-ten-second-pane".into();
+        }
+        let mut env = environment(10_000);
+        env.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite;
+        env.collector_ids.clear();
+        assert!(
+            matches!(PhysicalCompiler.compile(workload, env), Err(CompileError::Lifecycle { reason, .. }) if reason.contains("distinct logical consumer cohorts"))
+        );
+    }
+
+    // Non-divisor panes cannot reconstruct a lookback from whole states.
+    #[test]
+    fn tumbling_sizes_reject_non_divisors() {
+        let mut request = request("q", "sum(sum_over_time(m[1m]))");
+        request.queries[0].window_implementations[0].pane_secs = 7;
+        let mut env = environment(10_000);
+        env.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite;
+        assert!(PhysicalCompiler.compile(request, env).is_err());
+    }
+
     #[test]
     fn missing_window_implementation_evidence_fails_closed() {
         let mut request = request("q-window", "quantile_over_time(0.99, m[1m])");
diff --git a/control_plane/src/physical/post_asap/cost_model.rs b/control_plane/src/physical/post_asap/cost_model.rs
index 9cdf0624..67ba378f 100644
--- a/control_plane/src/physical/post_asap/cost_model.rs
+++ b/control_plane/src/physical/post_asap/cost_model.rs
@@ -63,7 +63,7 @@ pub struct ControlPlaneCostModel {
     pub workload_accuracy: AccuracyTarget,
     lifecycle_costs: SummaryMaintenanceLifecycleCostInputs,
     summary_maintenance: SummaryMaintenanceCapabilities,
-    window_framework_costs: Vec<(SummaryWindowFramework, Cost)>,
+    window_framework_costs: Vec<(Option<String>, SummaryWindowFramework, Cost)>,
     offline_evidence: Option<EmpiricalEvidenceProvider>,
     offline_frequency_comparison: Option<(OfflineComparisonEvidence, OfflineComparisonRequest)>,
 }
@@ -213,15 +213,27 @@ impl ControlPlaneCostModel {
         costs.into_iter().map(|(algorithm, _)| algorithm).collect()
     }
 
-    /// Bind the cheapest complete, workload-scoped physical realization for
-    /// each Planner-owned abstract window framework. Concrete implementation
-    /// identities stay in the physical compiler; only framework and cost
-    /// cross into Planner's candidate comparison.
+    /// Keep concrete physical identities in Planner's complete-candidate estimate,
+    /// including distinct pane sizes using the same abstract window framework.
+    pub fn with_window_implementation_costs(
+        mut self,
+        costs: Vec<(String, SummaryWindowFramework, Cost)>,
+    ) -> Self {
+        self.window_framework_costs = costs
+            .into_iter()
+            .map(|(id, framework, cost)| (Some(id), framework, cost))
+            .collect();
+        self
+    }
+
     pub fn with_window_framework_costs(
         mut self,
         costs: Vec<(SummaryWindowFramework, Cost)>,
     ) -> Self {
-        self.window_framework_costs = costs;
+        self.window_framework_costs = costs
+            .into_iter()
+            .map(|(framework, cost)| (None, framework, cost))
+            .collect();
         self
     }
 
@@ -387,14 +399,19 @@ impl CostModel for ControlPlaneCostModel {
             // GOS/error propagation is introduced by the later adaptation
             // slice. Until then, do not claim an approximate exponential
             // histogram window is exact.
-            .filter(|(framework, _)| {
+            .filter(|(_, framework, _)| {
                 !matches!(framework, SummaryWindowFramework::ExponentialHistogram)
             })
-            .filter(|(_, cost)| cost.0.is_finite() && cost.0 >= 0.0)
-            .min_by(|left, right| left.1 .0.total_cmp(&right.1 .0))
+            .filter(|(_, _, cost)| cost.0.is_finite() && cost.0 >= 0.0)
+            .min_by(|left, right| {
+                left.2
+                     .0
+                    .total_cmp(&right.2 .0)
+                    .then_with(|| left.0.cmp(&right.0))
+            })
             .map(
-                |(framework, physical_cost)| CompleteSummaryCandidateEstimate {
-                    physical_plan_id: None,
+                |(id, framework, physical_cost)| CompleteSummaryCandidateEstimate {
+                    physical_plan_id: id.clone(),
                     cost: Cost(lifecycle_cost + physical_cost.0),
                     window_frameworks: vec![Some(framework.clone()); deployments.len()],
                     window_accuracy_guarantee: Some(
diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs
index 54feda38..98fb0d2c 100644
--- a/control_plane/src/physical/workload_cost.rs
+++ b/control_plane/src/physical/workload_cost.rs
@@ -518,6 +518,14 @@ mod tests {
         ] {
             let (mut request, env) = fixture().planning_request().unwrap();
             request.queries[0].query_string = query.into();
+            request
+                .query_workload
+                .as_mut()
+                .unwrap()
+                .repeating_queries
+                .as_mut()
+                .unwrap()[0]
+                .query = planner_types::workload::Query(query.into());
             let exact = with_exact_alternative(request).unwrap().pop().unwrap();
             let plan = PhysicalCompiler
                 .compile(exact.clone(), env.clone())
@@ -659,6 +667,16 @@ mod tests {
         let mut second = shared.queries[0].clone();
         second.query_id = "second-consumer".into();
         second.query_string = "sum(count_over_time(m[1m]))".into();
+        let entries = shared
+            .query_workload
+            .as_mut()
+            .unwrap()
+            .repeating_queries
+            .as_mut()
+            .unwrap();
+        let mut demand = entries[0].clone();
+        demand.query = planner_types::workload::Query(second.query_string.clone());
+        entries.push(demand);
         shared.queries.push(second);
         let roots = shared
             .queries
diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs
index 3949c268..eda87aac 100644
--- a/data_plane/src/precompute_engine/worker.rs
+++ b/data_plane/src/precompute_engine/worker.rs
@@ -407,18 +407,33 @@ impl Worker {
         }
         let state = self.group_states.get_mut(&sid).unwrap();
 
+        // Keep original timestamps inside accumulators (notably rate/increase),
+        // shifting only pane membership and closure watermark for PromQL (a,b].
+        let right_closed = state
+            .config
+            .parameters
+            .get("promql_right_closed")
+            .and_then(serde_json::Value::as_bool)
+            .unwrap_or(false);
+        let pane_timestamp = |ts: i64| {
+            if right_closed {
+                ts.saturating_sub(1)
+            } else {
+                ts
+            }
+        };
         // Find the timestamp span in this batch. A first batch may contain
         // several windows (Prometheus commonly sends catch-up samples after
         // startup), so its minimum timestamp is also the initial closure
         // scan boundary.
         let batch_min_ts = samples
             .iter()
-            .map(|(_, ts, _)| *ts)
+            .map(|(_, ts, _)| pane_timestamp(*ts))
             .min()
             .unwrap_or(i64::MIN);
         let batch_max_ts = samples
             .iter()
-            .map(|(_, ts, _)| *ts)
+            .map(|(_, ts, _)| pane_timestamp(*ts))
             .max()
             .unwrap_or(i64::MIN);
         let previous_event_time = state.max_event_time_ms;
@@ -435,8 +450,9 @@ impl Worker {
         // Route each sample to its pane
         for (series_key, ts, val) in &samples {
             let too_late = previous_event_time != i64::MIN
-                && *ts < watermark_for_event_time(previous_event_time, allowed_lateness_ms);
-            let pane_start = state.window_manager.pane_start_for(*ts);
+                && pane_timestamp(*ts)
+                    < watermark_for_event_time(previous_event_time, allowed_lateness_ms);
+            let pane_start = state.window_manager.pane_start_for(pane_timestamp(*ts));
             let pane_end = pane_start + state.window_manager.slide_interval_ms();
             let pane_closed = !state.active_panes.contains_key(&pane_start)
                 && previous_closure_watermark >= pane_start + state.window_manager.window_size_ms();
diff --git a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs
index de53bf96..1a0cc1dd 100644
--- a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs
+++ b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs
@@ -110,6 +110,16 @@ pub fn execute_query_plan_instant(
     } else {
         now_ms.saturating_sub(entry.instant.lookback_ms)
     };
+    for binding in entry.materialization_bindings() {
+        if binding.window_ms == 0
+            || t0_ms % binding.window_ms != 0
+            || now_ms % binding.window_ms != 0
+        {
+            return Err(LoweringSkip::MaterializationNotReady(
+                "evaluation interval cuts a materialized pane".into(),
+            ));
+        }
+    }
     let outcome = execute_physical_query_plan(
         index,
         entry,
@@ -896,6 +906,101 @@ mod tests {
         assert_eq!(outcome.coverage, Some((2_000, 2_000)));
     }
 
+    #[test]
+    fn repeated_multi_pane_reads_exclude_expired_state_and_reject_gaps() {
+        let idx = SketchStore::new();
+        let policy = asap_types::PolicyFingerprint(777);
+        idx.register(SketchInstanceMetadata {
+            sid: 7,
+            metric_name: "requests_total".into(),
+            group_by_keys: std::collections::BTreeSet::new(),
+            capability: Some(Capability::ExactAgg(asap_types::AggregationType::Sum)),
+            agg_kind: AggKind::ExactAgg {
+                agg_type: asap_types::AggregationType::Sum,
+                parameters_canonical: String::new(),
+                spatial_filter_canonical: String::new(),
+            },
+            accuracy: None,
+            first_seen_unix_ms: 0,
+            retired_at_ms: None,
+            expires_at_ms: None,
+            policy_fp: policy,
+        });
+        // Old panes remain stored; each advancing query must select only its lookback.
+        for pane in 0..8 {
+            idx.append_precompute(
+                7,
+                BTreeMap::new(),
+                (pane * 10_000, (pane + 1) * 10_000),
+                Box::new(
+                    crate::precompute_engine::operators::SumAccumulator::with_sum(
+                        (pane + 1) as f64,
+                    ),
+                ),
+            );
+        }
+
+        let entry = control_plane::query_plan::QueryPlanEntry {
+            query_id: "q-rate".into(),
+            canonical_promql: "rate(requests_total[1m])".into(),
+            root: control_plane::query_plan::QueryNodeId(0),
+            nodes: BTreeMap::from([
+                (
+                    control_plane::query_plan::QueryNodeId(0),
+                    QueryPlanNode::ExactReadout {
+                        input: control_plane::query_plan::QueryNodeId(1),
+                        readout: control_plane::query_plan::ExactReadout::Sum,
+                    },
+                ),
+                (
+                    control_plane::query_plan::QueryNodeId(1),
+                    QueryPlanNode::ReadMaterialization {
+                        binding: control_plane::query_plan::MaterializationBinding {
+                            materialization: policy,
+                            metric: "requests_total".into(),
+                            sid_grouping: vec![],
+                            output_grouping: control_plane::query_plan::PhysicalGrouping::PerEntity,
+                            window_ms: 10_000,
+                        },
+                    },
+                ),
+            ]),
+            instant: control_plane::query_plan::InstantExecution {
+                lookback_ms: 60_000,
+                full_history: false,
+                cumulative_readout: true,
+            },
+            fallback: control_plane::query_plan::FallbackPolicy::ExactBackend,
+        };
+        for (now, expected) in [(60_000, 21.0), (70_000, 27.0), (80_000, 33.0)] {
+            let (outcome, _) = execute_query_plan_instant(&idx, &entry, now).unwrap();
+            assert_eq!(outcome.series[0].1[0].1, expected);
+        }
+        assert!(
+            execute_query_plan_instant(&idx, &entry, 70_001).is_err(),
+            "partial pane must fall back"
+        );
+        assert!(
+            execute_query_plan_instant(&idx, &entry, 90_000).is_err(),
+            "open/missing trailing pane must fall back"
+        );
+        // Both endpoints exist, but the missing interior pane is not evidence of zero samples.
+        let gap_idx = SketchStore::new();
+        idx.with_instance(7, |meta| gap_idx.register(meta.clone()));
+        for pane in [0, 1, 3, 4, 5] {
+            gap_idx.append_precompute(
+                7,
+                BTreeMap::new(),
+                (pane * 10_000, (pane + 1) * 10_000),
+                Box::new(crate::precompute_engine::operators::SumAccumulator::with_sum(1.0)),
+            );
+        }
+        assert!(
+            execute_query_plan_instant(&gap_idx, &entry, 60_000).is_err(),
+            "interior gap must fall back"
+        );
+    }
+
     #[test]
     fn exact_query_plan_rate_uses_reset_aware_readout() {
         let idx = SketchStore::new();
diff --git a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs
index c20c8e9b..92666632 100644
--- a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs
+++ b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs
@@ -372,6 +372,42 @@ impl QueryExecutionContext<'_> {
             ExactAgg(AggregationType),
         }
 
+        // A missing pane can mean delayed ingestion, not an empty interval.
+        // Until explicit empty-pane completion exists, multi-pane reads require
+        // every pane for each stored series, before any cross-series merge.
+        let check_panes = |ends: Vec<i64>| -> Result<(), SummaryExecutorError> {
+            let width = binding.window_ms;
+            if width == 0 {
+                return Err(SummaryExecutorError::Unsupported("zero pane width"));
+            }
+            if self.t1_ms.saturating_sub(self.t0_ms) <= width {
+                return Ok(());
+            }
+            if self.t0_ms % width != 0 || self.t1_ms % width != 0 {
+                return Err(SummaryExecutorError::Unsupported("partial pane interval"));
+            }
+            let mut expected = self.t0_ms.checked_add(width);
+            for end in ends {
+                let Ok(end) = u64::try_from(end) else {
+                    continue;
+                };
+                if end <= self.t0_ms {
+                    continue;
+                } // delta decoding carry-in is not an answer pane
+                if Some(end) != expected {
+                    return Err(SummaryExecutorError::Unsupported(
+                        "missing materialized pane",
+                    ));
+                }
+                expected = end.checked_add(width);
+            }
+            if expected != self.t1_ms.checked_add(width) {
+                return Err(SummaryExecutorError::Unsupported(
+                    "incomplete materialized panes",
+                ));
+            }
+            Ok(())
+        };
         let required_keys: BTreeSet<_> = binding.sid_grouping.iter().cloned().collect();
         let mut sids = self.index.sids_for_policy(binding.materialization);
         sids.sort_unstable();
@@ -407,8 +443,10 @@ impl QueryExecutionContext<'_> {
                         .into_iter()
                         .next()
                     else {
+                        check_panes(Vec::new())?;
                         continue;
                     };
+                    check_panes(series.samples.keys().copied().collect())?;
                     let key = match &binding.output_grouping {
                         PhysicalGrouping::PerEntity => series.series_label_values.clone(),
                         PhysicalGrouping::Reduce(keys) => {
@@ -427,8 +465,10 @@ impl QueryExecutionContext<'_> {
                         .into_iter()
                         .next()
                     else {
+                        check_panes(Vec::new())?;
                         continue;
                     };
+                    check_panes(windows.keys().copied().collect())?;
                     let key = match &binding.output_grouping {
                         PhysicalGrouping::PerEntity => labels,
                         PhysicalGrouping::Reduce(keys) => project_group_key(keys, &labels),
diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs
index b245b32d..24ed0e0d 100644
--- a/data_plane/tests/asapquery_compatibility_process_e2e.rs
+++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs
@@ -381,6 +381,16 @@ async fn registered_temporal_topk(algorithm: planner_types::post_asap::SketchAlg
 // uneven instance sample counts and Remote Write retries.
 #[tokio::test]
 async fn shared_exact_dashboard_executes_selected_workload() {
+    run_shared_dashboard(false).await;
+}
+
+// A selected 5s pane serves advancing 10s lookbacks through the production HTTP path.
+#[tokio::test]
+async fn repeated_dashboard_executes_multiple_selected_panes() {
+    run_shared_dashboard(true).await;
+}
+
+async fn run_shared_dashboard(multi_pane: bool) {
     let fallback_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
     let fallback_address = fallback_listener.local_addr().unwrap();
     tokio::spawn(async move {
@@ -394,8 +404,16 @@ async fn shared_exact_dashboard_executes_selected_workload() {
         "../../docs/examples/asapquery-compatibility-demo-snapshot.json"
     ))
     .unwrap();
-    let sum = "sum by (service) (sum_over_time(asap_demo_gauge[5s]))";
-    let count = "sum by (service) (count_over_time(asap_demo_gauge[5s]))";
+    let sum = if multi_pane {
+        "sum by (service) (sum_over_time(asap_demo_gauge[10s]))"
+    } else {
+        "sum by (service) (sum_over_time(asap_demo_gauge[5s]))"
+    };
+    let count = if multi_pane {
+        "sum by (service) (count_over_time(asap_demo_gauge[10s]))"
+    } else {
+        "sum by (service) (count_over_time(asap_demo_gauge[5s]))"
+    };
     let mean = format!("{sum} / {count}");
     let mut entry = snapshot["query_workload"]["repeating_queries"][2].clone();
     snapshot["query_workload"]["repeating_queries"] = Value::Array(
@@ -409,6 +427,25 @@ async fn shared_exact_dashboard_executes_selected_workload() {
     );
     let mut typed: control_plane::physical::compiler::BackendLocalPlanningSnapshot =
         serde_json::from_value(snapshot.clone()).unwrap();
+    if multi_pane {
+        for entry in typed.query_workload.repeating_queries.as_mut().unwrap() {
+            entry.time_selection.lookback = Some(planner_types::workload::DurationMs(10_000));
+        }
+        let (request, _) = typed.clone().planning_request().unwrap();
+        for query in request.queries {
+            let mut candidates = query.window_implementations;
+            let mut small = candidates[0].clone();
+            small.implementation_id = "five-second-pane".into();
+            small.pane_secs = 5;
+            small.cost.weighted_cost = 0.0;
+            candidates[0].cost.weighted_cost = 10.0;
+            candidates.push(small);
+            typed
+                .implementation
+                .window_candidates
+                .insert(query.query_string, candidates);
+        }
+    }
     let (request, environment) = typed.clone().planning_request().unwrap();
     let candidates =
         control_plane::physical::workload_cost::with_exact_alternative(request).unwrap();
@@ -449,6 +486,16 @@ async fn shared_exact_dashboard_executes_selected_workload() {
     assert!(plan.cost_comparison.is_some());
     assert_eq!(plan.precompute_plan.materializations.len(), 1);
     assert_eq!(plan.query_plan.entries.len(), 3);
+    if multi_pane {
+        assert_eq!(
+            plan.lifecycle_estimates[0].window_implementation_id,
+            "five-second-pane"
+        );
+        for entry in plan.query_plan.entries.values() {
+            assert_eq!(entry.instant.lookback_ms, 10_000);
+            assert_eq!(entry.materialization_bindings()[0].window_ms, 5_000);
+        }
+    }
     let snapshot_path = output_dir.path().join("snapshot.json");
     std::fs::write(&snapshot_path, serde_json::to_vec(&snapshot).unwrap()).unwrap();
     let port = unused_port();
@@ -497,7 +544,15 @@ async fn shared_exact_dashboard_executes_selected_workload() {
     };
     let request = WriteRequest {
         timeseries: vec![
-            labeled("api", "a", &[(base + 500, 10.0)]),
+            if multi_pane {
+                labeled(
+                    "api",
+                    "a",
+                    &[(base, 10000.0), (base + 500, 10.0), (base + 5000, 6.0)],
+                )
+            } else {
+                labeled("api", "a", &[(base + 500, 10.0)])
+            },
             labeled(
                 "api",
                 "b",
@@ -518,22 +573,51 @@ async fn shared_exact_dashboard_executes_selected_workload() {
     assert_eq!(remote_write(&client, &backend, &advance).await, 204);
     let final_advance = WriteRequest {
         timeseries: vec![
-            labeled("api", "a", &[(base + 10500, 1000.0)]),
+            if multi_pane {
+                labeled("api", "a", &[(base + 10000, 7.0), (base + 10500, 1000.0)])
+            } else {
+                labeled("api", "a", &[(base + 10500, 1000.0)])
+            },
             labeled("api", "b", &[(base + 10500, 1000.0)]),
             labeled("worker", "c", &[(base + 10500, 1000.0)]),
         ],
     };
     assert_eq!(remote_write(&client, &backend, &final_advance).await, 204);
+    if multi_pane {
+        let close_third = WriteRequest {
+            timeseries: vec![
+                labeled("api", "a", &[(base + 15500, 9999.0)]),
+                labeled("api", "b", &[(base + 15500, 9999.0)]),
+                labeled("worker", "c", &[(base + 15500, 9999.0)]),
+            ],
+        };
+        assert_eq!(remote_write(&client, &backend, &close_third).await, 204);
+    }
+    let evaluation = base + if multi_pane { 10000 } else { 5000 };
     for (query, expected) in [
-        (sum, [24.0, 24.0]),
-        (count, [4.0, 2.0]),
-        (mean.as_str(), [6.0, 12.0]),
+        (
+            sum,
+            if multi_pane {
+                [237.0, 124.0]
+            } else {
+                [24.0, 24.0]
+            },
+        ),
+        (count, if multi_pane { [8.0, 3.0] } else { [4.0, 2.0] }),
+        (
+            mean.as_str(),
+            if multi_pane {
+                [237.0 / 8.0, 124.0 / 3.0]
+            } else {
+                [6.0, 12.0]
+            },
+        ),
     ] {
         let result = wait_for_warm_instant(
             &client,
             &backend,
             query,
-            (base + 5000) as f64 / 1000.0,
+            evaluation as f64 / 1000.0,
             &output_dir.path().join("query_engine.log"),
         )
         .await;
@@ -550,7 +634,7 @@ async fn shared_exact_dashboard_executes_selected_workload() {
             );
             assert_eq!(
                 row["value"][0].as_f64().unwrap(),
-                (base + 5000) as f64 / 1000.0
+                evaluation as f64 / 1000.0
             );
         }
     }
@@ -558,8 +642,8 @@ async fn shared_exact_dashboard_executes_selected_workload() {
         .get(format!("{backend}/api/v1/query_range"))
         .query(&[
             ("query", mean.clone()),
-            ("start", ((base + 5000) as f64 / 1000.0).to_string()),
-            ("end", ((base + 10000) as f64 / 1000.0).to_string()),
+            ("start", (evaluation as f64 / 1000.0).to_string()),
+            ("end", ((evaluation + 5000) as f64 / 1000.0).to_string()),
             ("step", "5".into()),
         ])
         .send()
@@ -572,14 +656,26 @@ async fn shared_exact_dashboard_executes_selected_workload() {
     for row in result["data"]["result"].as_array().unwrap() {
         let values = row["values"].as_array().unwrap();
         assert_eq!(values.len(), 2, "{result}");
-        assert_eq!(values[1][1], "100", "{result}");
+        assert_eq!(
+            values[1][1],
+            if multi_pane {
+                if row["metric"]["service"] == "api" {
+                    "441.4"
+                } else {
+                    "550"
+                }
+            } else {
+                "100"
+            },
+            "{result}"
+        );
     }
     // Unaligned intervals cannot be answered by whole tumbling states.
     let result: Value = client
         .get(format!("{backend}/api/v1/query"))
         .query(&[
             ("query", mean),
-            ("time", ((base + 5001) as f64 / 1000.0).to_string()),
+            ("time", ((evaluation + 1) as f64 / 1000.0).to_string()),
         ])
         .send()
         .await
@@ -1157,6 +1253,11 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
         .unwrap();
     snapshot.query_workload.repeating_queries.as_mut().unwrap()[0].query =
         planner_types::workload::Query("rate(asap_demo_counter_total[1m])".into());
+    // This fixture constructs an old envelope counter over whole retired panes,
+    // not a real-time retracting raw counter (which Planner correctly refuses).
+    snapshot.query_workload.repeating_queries.as_mut().unwrap()[0]
+        .time_selection
+        .scope = planner_types::workload::QueryTimeScope::Unknown;
     let (mut legacy_request, mut environment) = snapshot.planning_request().unwrap();
     let query = &mut legacy_request.queries[0];
     let parsed = control_plane::query_parser::parse_query_expr_canonical(
diff --git a/docs/developer_docs/planning/repeated-dashboard-panes.md b/docs/developer_docs/planning/repeated-dashboard-panes.md
new file mode 100644
index 00000000..995ae04a
--- /dev/null
+++ b/docs/developer_docs/planning/repeated-dashboard-panes.md
@@ -0,0 +1,67 @@
+# Repeated dashboard pane selection
+
+The backend-local planning snapshot accepts `implementation.window_candidates`,
+a map from the original registered PromQL text to a list of
+`WindowImplementationCandidate` objects. Omitting a query keeps the existing
+single lookback-sized candidate. Unknown query keys and empty candidate lists
+are rejected. Each candidate supplies its own implementation ID, Tumbling
+framework, query `window_secs`, `pane_secs`, state layout, and complete fresh
+`ImplementationCostEvidence`. Pane sizes must be positive divisors of lookback.
+Distributed Collector deployments still require pane size equal to lookback.
+
+For example, a 60-second query may offer 10-, 20-, and 60-second panes, with
+separate provider quotes for each. The cost model returns the concrete ID in
+Planner's `CompleteSummaryCandidateEstimate`; the compiler installs the ID
+returned by Planner. It does not choose another size after planning. Actual
+pane width enters the state fingerprint, precompute configuration, and query
+binding; the query's 60-second lookback remains unchanged. Shared DAG consumers
+must agree on the physical deployment contract. Distinct logical-window cohorts
+that collide only after selecting a smaller shared pane are rejected until
+joint lifecycle evidence is available for the resulting physical state. The
+compiler never keeps only the first cohort's consumer count or cost quote.
+Sharing within one already-priced logical cohort remains supported.
+
+The original QueryWorkload, including recurrence, time scope, predictability,
+requirements and data evidence, reaches lifecycle planning. Shared consumers'
+reads are combined without multiplying state updates. Legacy direct requests
+without a QueryWorkload retain their previous synthesized demand. Refresh cadence
+alone does not prove evaluation phase alignment, so it is not a configuration
+rejection condition. Candidate costs must account for the supplied workload;
+the backend does not manufacture measured costs or an automatic pane-size cost
+formula. Version 2 still requires complete workload-versus-exact quotes.
+
+## Execution guarantees and limits
+
+Backend-local raw plans encode `promql_right_closed: true` in state parameters
+and therefore in the fingerprint. Raw workers assign boundary samples to
+PromQL's `(start, end]` panes, while preserving original timestamps inside
+accumulators. Legacy half-open states have a different identity.
+
+At every instant/range evaluation, serving checks each binding's pane alignment.
+Multi-pane reads require contiguous stored pane ends for every matched stored
+series before merging state. Partial, missing/open, or interior missing panes
+fail closed to exact fallback. Without explicit empty-pane completion evidence,
+a sparse interval is conservatively a fallback rather than an assumed zero.
+Whole panes outside the current lookback are excluded even when retained in
+storage; this is logical window expiration, not a claim of immediate physical
+state reclamation. Normal retention remains responsible for reclaiming state.
+
+This change does not add operators or min/max-specific paths. Existing operator,
+source, and predicate limitations continue to apply.
+
+## Reproduction
+
+From the repository root:
+
+```sh
+cargo test -p control_plane --lib
+cargo test -p data_plane --lib -- --test-threads=1
+cargo test -p data_plane --test asapquery_compatibility_process_e2e -- --test-threads=1
+```
+
+The conformance tests change candidate costs and assert the selected ID and
+installed pane width; execute advancing stored-state queries; reject missing and
+partial panes; and run production RemoteWrite, planning, installation, SUM,
+COUNT, and a ratio DAG with 5-second panes and 10-second lookbacks. Boundary
+samples verify left exclusion and right inclusion. These are synthetic
+correctness tests, not an o11ybench performance or benefit report.