From f6d4ce402fcd35262b633a6962e725d39266adb4 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Sat, 12 Sep 2026 10:00:34 -0600
Subject: [PATCH 1/6] fix(planning): derive the fallback window shape from the
 evaluation cadence

A snapshot that prices no window implementation for a query got one
synthesized here, hardcoded to a lookback-wide tumbling window with a
lookback-wide pane. The query's own evaluation cadence was parsed a few
lines above, used for lifecycle costing, and then dropped.

The result is a plan whose answer only changes once per window. A
workload of two 5m-lookback quantiles evaluated every 30s planned as
window 300 / slide 300 / tumbling / Pane{300}: the state advances every
five minutes while the queries run every thirty seconds. The retained
count followed it down to 2, which is arithmetically right for a
300s pane and useless for a 30s cadence.

Derive the shape instead. When the cadence is shorter than the window
and divides it, slide by one evaluation interval and store panes of that
width; otherwise keep the previous tumbling shape. The same workload now
plans as window 300 / slide 30 / sliding / Pane{30}, retaining 11 states
-- ten panes covering the lookback plus the one still filling.

This derives a shape, never a price. `ImplementationCostEvidence` is
measured evidence: its `weighted_cost` doc puts pricing update CPU,
query-time merges, retained memory, storage, scans and network on the
evidence producer. So this does not synthesize a second candidate to
rank against the first -- a lone candidate is chosen by a `min_by` over
one element, where the cost cannot change the outcome. Ranking Pane
against FullWindow still requires a snapshot that supplies both with
their own priced evidence, and `window_candidates` already carries them
untouched when it does.

Two existing assertions encoded the old default's pane width. The
snapshot they load evaluates every 10s over a 1m lookback, so its stored
pane is now 10s; both readouts keep their 1m `readout_lookback_ms`, and
every other property those tests assert is unchanged.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 control_plane/src/physical/compiler.rs | 249 +++++++++++++++++++++++--
 1 file changed, 238 insertions(+), 11 deletions(-)

diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index b57f00e8..9622eb6c 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -636,16 +636,12 @@ impl BackendLocalPlanningSnapshot {
                     .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,
-                            slide_secs: lookback_ms / 1_000,
-                            layout: asap_types::WindowMaterializationLayout::Pane {
-                                pane_secs: lookback_ms / 1_000,
-                            },
+                        vec![derived_window_candidate(
+                            self.implementation.window_implementation_id.clone(),
+                            lookback_ms,
+                            evaluation_interval_ms,
                             cost,
-                        }]
+                        )]
                     }),
                 runtime_policy: RuntimeRulePolicy::default(),
             });
@@ -2188,6 +2184,78 @@ fn validate_lifecycle_input(
     Ok(())
 }
 
+/// The one window implementation to plan with when the snapshot priced none
+/// for this query.
+///
+/// This is a *shape*, not a cost quote. `ImplementationCostEvidence` is
+/// measured evidence an evidence producer supplies — its `weighted_cost` doc
+/// is explicit that the producer, not this compiler, prices update CPU,
+/// query-time merges, retained memory, storage, scans and network. So this
+/// function never synthesizes a second candidate to rank: a single candidate
+/// is selected by `complete_summary_candidate_estimate`'s `min_by` over a
+/// one-element list, where the cost value cannot change the outcome. Ranking
+/// Pane against FullWindow requires the snapshot to supply both in
+/// `window_candidates`, each with its own priced evidence.
+///
+/// What the shape must respect is the query's own demand. A workload that
+/// evaluates every 30s over a 5m lookback needs its state to advance every
+/// 30s; planning it as one 5m tumbling window answers with results that only
+/// change every 5 minutes. `evaluation_interval_ms` already reaches this
+/// function — it was previously read for lifecycle costing and then dropped
+/// on the floor here.
+///
+/// The derivation, and why each guard exists
+/// (`validate_window_implementations` rejects a candidate that breaks any of
+/// them, so a bad shape would surface as a compile error, never a silent
+/// plan):
+///
+/// - `window_secs` is the semantic lookback, which `PlanningQuery::window_secs`
+///   also uses; the validator requires the two to be equal.
+/// - The slide advances one evaluation interval, so consecutive evaluations
+///   share state, but only when that interval is shorter than the window and
+///   divides it. A non-dividing interval (45s into 300s) has no pane width
+///   that divides both, and `WindowMaterializationLayout::validate` would
+///   reject it, so keep the tumbling shape.
+/// - `Pane { pane_secs: slide_secs }` divides the slide trivially and divides
+///   the window by the same guard. Each sample then updates exactly one pane
+///   (`worker.rs`'s `stores_full_windows` branch), and a read composes
+///   `window / slide` of them. `FullWindow` is the other legal Sliding
+///   layout and is deliberately not chosen here: preferring it over panes is
+///   a cost comparison, and this function has no second quote to compare.
+/// - Tumbling pairs only with `Pane` in the validator's framework/layout
+///   table, so the degenerate `pane_secs == window_secs` case stays as it was.
+fn derived_window_candidate(
+    implementation_id: String,
+    lookback_ms: u64,
+    evaluation_interval_ms: u32,
+    cost: ImplementationCostEvidence,
+) -> WindowImplementationCandidate {
+    let window_secs = lookback_ms / 1_000;
+    let evaluation_secs = u64::from(evaluation_interval_ms) / 1_000;
+    let advances_within_window = evaluation_secs != 0
+        && evaluation_secs < window_secs
+        && window_secs.is_multiple_of(evaluation_secs);
+    let slide_secs = if advances_within_window {
+        evaluation_secs
+    } else {
+        window_secs
+    };
+    WindowImplementationCandidate {
+        implementation_id,
+        framework: if advances_within_window {
+            SummaryWindowFramework::Sliding
+        } else {
+            SummaryWindowFramework::Tumbling
+        },
+        window_secs,
+        slide_secs,
+        layout: asap_types::WindowMaterializationLayout::Pane {
+            pane_secs: slide_secs,
+        },
+        cost,
+    }
+}
+
 pub(super) fn validate_window_implementations(
     query: &PlanningQuery,
     environment: &DeploymentEnvironment,
@@ -4890,13 +4958,170 @@ mod tests {
                 )
             })
             .collect::<BTreeSet<_>>();
+        // `window_ms` is the stored pane width, `readout_lookback_ms` the
+        // semantic range. The snapshot evaluates every 10s, so `a`'s derived
+        // candidate stores 10s panes and composes six of them for its 1m
+        // readout; `b` keeps the 60s pane its explicitly supplied candidate
+        // priced. Both readouts are unchanged.
         assert_eq!(
             actual,
-            BTreeSet::from([("a", 60_000, Some(60_000)), ("b", 60_000, Some(300_000)),])
+            BTreeSet::from([("a", 10_000, Some(60_000)), ("b", 60_000, Some(300_000)),])
         );
         assert_eq!(plan.precompute_plan.materializations.len(), 2);
     }
 
+    fn planning_snapshot() -> BackendLocalPlanningSnapshot {
+        serde_json::from_str(include_str!(
+            "../../../docs/examples/asapquery-planning-snapshot.json"
+        ))
+        .unwrap()
+    }
+
+    // A workload evaluated more often than its window is wide must advance its
+    // state at that cadence. Planning it as one lookback-wide tumbling window
+    // answers with results that only change once per window.
+    #[test]
+    fn derived_window_candidate_follows_the_evaluation_cadence() {
+        let cost = planning_snapshot().implementation.implementation_cost;
+        let candidate = derived_window_candidate("id".into(), 300_000, 30_000, cost);
+        assert_eq!(candidate.framework, SummaryWindowFramework::Sliding);
+        assert_eq!((candidate.window_secs, candidate.slide_secs), (300, 30));
+        assert_eq!(
+            candidate.layout,
+            asap_types::WindowMaterializationLayout::Pane { pane_secs: 30 }
+        );
+    }
+
+    // Every shape this function can emit must survive the validator, or a bad
+    // derivation would reach a plan instead of a compile error.
+    #[test]
+    fn derived_window_candidate_shapes_are_accepted_by_validation() {
+        let snapshot = planning_snapshot();
+        let (request, environment) = snapshot.planning_request().unwrap();
+        let cost = planning_snapshot().implementation.implementation_cost;
+        for (lookback_ms, evaluation_ms) in [
+            (300_000, 30_000),
+            (300_000, 300_000),
+            (300_000, 45_000),
+            (60_000, 90_000),
+        ] {
+            let mut query = request.queries[0].clone();
+            query.window_secs = lookback_ms / 1_000;
+            query.window_implementations = vec![derived_window_candidate(
+                "derived".into(),
+                lookback_ms,
+                evaluation_ms,
+                cost.clone(),
+            )];
+            validate_window_implementations(&query, &environment).unwrap_or_else(|error| {
+                panic!("lookback {lookback_ms} cadence {evaluation_ms}: {error:?}")
+            });
+        }
+    }
+
+    // A cadence that cannot divide the window has no pane width dividing both,
+    // and one at or above the window has nothing to slide within. Both keep the
+    // previous tumbling shape rather than emitting something unschedulable.
+    #[test]
+    fn derived_window_candidate_stays_tumbling_without_a_dividing_cadence() {
+        let cost = planning_snapshot().implementation.implementation_cost;
+        for evaluation_ms in [300_000, 450_000, 45_000, 0] {
+            let candidate =
+                derived_window_candidate("id".into(), 300_000, evaluation_ms, cost.clone());
+            assert_eq!(
+                (
+                    candidate.framework.clone(),
+                    candidate.slide_secs,
+                    candidate.layout.clone()
+                ),
+                (
+                    SummaryWindowFramework::Tumbling,
+                    300,
+                    asap_types::WindowMaterializationLayout::Pane { pane_secs: 300 }
+                ),
+                "cadence {evaluation_ms}"
+            );
+        }
+    }
+
+    // Priced evidence is the evidence producer's to supply. A snapshot that
+    // carries its own candidates keeps them verbatim.
+    #[test]
+    fn supplied_window_candidates_are_not_replaced_by_the_derivation() {
+        let mut snapshot = planning_snapshot();
+        let query_string = snapshot.query_workload.repeating_queries.as_ref().unwrap()[0]
+            .query
+            .0
+            .clone();
+        let mut supplied = derived_window_candidate(
+            "supplied".into(),
+            60_000,
+            60_000,
+            snapshot.implementation.implementation_cost.clone(),
+        );
+        supplied.framework = SummaryWindowFramework::Sliding;
+        supplied.slide_secs = 20;
+        supplied.layout = asap_types::WindowMaterializationLayout::Pane { pane_secs: 20 };
+        snapshot
+            .implementation
+            .window_candidates
+            .insert(query_string, vec![supplied.clone()]);
+        let (request, _) = snapshot.planning_request().unwrap();
+        assert_eq!(request.queries[0].window_implementations, vec![supplied]);
+    }
+
+    // End to end: the retained-state count is derived from the pane width, so
+    // fixing the shape fixes it too. Six 10s panes cover the 1m lookback, plus
+    // the one still being filled.
+    #[test]
+    fn retained_state_count_follows_the_derived_pane_width() {
+        let snapshot = planning_snapshot();
+        let (request, environment) = snapshot.planning_request().unwrap();
+        let plan = PhysicalCompiler.compile(request, environment).unwrap();
+        assert_eq!(
+            plan.precompute_plan.materializations[0].num_aggregates_to_retain,
+            Some(7)
+        );
+    }
+
+    // The reported case: two 5m-lookback quantiles evaluated every 30s. The
+    // whole chain has to land — sliding framework, 30s panes, and the retained
+    // count that falls out of the pane width — or the answer only changes once
+    // every five minutes.
+    #[test]
+    fn five_minute_lookback_evaluated_every_thirty_seconds_slides_by_thirty() {
+        let mut snapshot = planning_snapshot();
+        {
+            let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0];
+            entry.query = Query("quantile_over_time(0.5, data[5m])".into());
+            entry.time_selection.lookback = Some(DurationMs(300_000));
+            entry.demand = RepeatedDemand::FixedIntervalAt {
+                interval: RepetitionInterval(30_000),
+                evaluation_phase: planner_types::workload::TimestampMs(0),
+            };
+        }
+        let (request, environment) = snapshot.planning_request().unwrap();
+        let plan = PhysicalCompiler.compile(request, environment).unwrap();
+        let materialization = &plan.precompute_plan.materializations[0];
+        assert_eq!(
+            (
+                materialization.window_size,
+                materialization.slide_interval,
+                materialization.window_type,
+                materialization.window_layout.clone(),
+                materialization.num_aggregates_to_retain,
+            ),
+            (
+                300,
+                30,
+                asap_types::WindowKind::Sliding,
+                asap_types::WindowMaterializationLayout::Pane { pane_secs: 30 },
+                // Ten 30s panes cover the 5m lookback, plus the one still filling.
+                Some(11),
+            )
+        );
+    }
+
     // A filtered denominator is a typed residual while its summary sibling remains installed.
     #[test]
     fn composable_binary_retains_summary_sibling_of_prometheus_filtered_subtree() {
@@ -4916,9 +5141,11 @@ mod tests {
         assert_eq!(bindings.len(), 1);
         let identity = &plan.summary_catalog.materializations[&bindings[0].materialization];
         let data = &plan.summary_catalog.data_descriptors[&identity.data_descriptor_id];
+        // 10s panes for a 10s evaluation cadence; the 1m readout range is
+        // carried by `readout_lookback_ms`, not by the stored pane width.
         assert_eq!(
             (data.time_series_metric().unwrap(), bindings[0].window_ms),
-            ("a", 60_000)
+            ("a", 10_000)
         );
         assert!(!query
             .nodes

From 0149328ab55b5eb68b605e2d09999522532f4a67 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Sat, 12 Sep 2026 10:15:50 -0600
Subject: [PATCH 2/6] fix(planning): derive one fallback window per range
 selector

The derived fallback built a single candidate from
`time_selection.lookback`. A query carries one range selector per
operand, and they need not agree with the declared lookback or with each
other.

The consequence is silent. Under `hybrid_execution`, a selected state
whose window has no candidate is filtered out of selection entirely, and
the surviving candidates are then narrowed to the selected window before
validation. So `sum(sum_over_time(a[1m])) / sum(sum_over_time(b[5m]))`
under a 1m lookback kept `a` and dropped `b` to exact execution, with
nothing reported and no error raised.

Derive one candidate per distinct range-selector window instead, each
shaped by the evaluation cadence as before. `window_implementation_id`
reaches lifecycle estimates and cost manifests, so a query with several
windows suffixes it per window; a single-window query keeps the
snapshot's identity untouched.

Three assertions changed, all of them pinning the previous artifact
rather than intended behavior:

- `composable_binary_binds_independent_source_windows` hand-supplied a
  5m candidate so `b` would survive. The derivation now covers it, so
  the workaround is gone and both operands keep their own range.
- `composable_binary_retains_summary_sibling_of_prometheus_filtered_subtree`
  asserted a filtered denominator stays a typed residual. Nothing about
  the filter forced that -- the missing 5m candidate did. It now gets a
  summary over its own filtered population, which the renamed
  `composable_binary_summarizes_each_prometheus_filtered_operand` pins,
  filter and range together.
- `counter_materialization_manifest_prices_owned_state_and_distinct_native_alternative`
  required an `exact_backend` component. Its 5m operand landed there for
  the same reason and is now summarized.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 control_plane/src/physical/compiler.rs      | 298 +++++++++++++-------
 control_plane/src/physical/workload_cost.rs |   5 +-
 2 files changed, 200 insertions(+), 103 deletions(-)

diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index 9622eb6c..9123c9ff 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -636,12 +636,13 @@ impl BackendLocalPlanningSnapshot {
                     .get(&query_string)
                     .cloned()
                     .unwrap_or_else(|| {
-                        vec![derived_window_candidate(
-                            self.implementation.window_implementation_id.clone(),
+                        derived_window_candidates(
+                            &self.implementation.window_implementation_id,
+                            canonical_roots.last().expect("root pushed above"),
                             lookback_ms,
                             evaluation_interval_ms,
                             cost,
-                        )]
+                        )
                     }),
                 runtime_policy: RuntimeRulePolicy::default(),
             });
@@ -2184,76 +2185,136 @@ fn validate_lifecycle_input(
     Ok(())
 }
 
-/// The one window implementation to plan with when the snapshot priced none
-/// for this query.
+/// Every distinct range-selector window in `expr`, as seconds.
+///
+/// A single query can carry several. `sum(sum_over_time(a[1m])) / sum(sum_over_time(b[5m]))`
+/// has two, and each one becomes its own materialization with its own window.
+/// `time_selection.lookback` is the workload's declared range and is not
+/// required to equal any of them.
+fn range_selector_windows_secs(expr: &QueryExpr) -> BTreeSet<u64> {
+    fn visit(expr: &QueryExpr, windows: &mut BTreeSet<u64>) {
+        if let QueryExpr::TimeRange { range, .. } = expr {
+            let secs = range.as_secs();
+            if secs != 0 {
+                windows.insert(secs);
+            }
+        }
+        match expr {
+            QueryExpr::PromqlScalarBridge(child)
+            | QueryExpr::PromqlVectorFromScalar(child)
+            | QueryExpr::PromqlScalarFromVector(child)
+            | QueryExpr::PromqlRelabel { child, .. }
+            | QueryExpr::PromqlSeriesSample { child, .. }
+            | QueryExpr::Filter { child, .. }
+            | QueryExpr::Project { child, .. }
+            | QueryExpr::Aggregate { child, .. }
+            | QueryExpr::Dedup { child, .. }
+            | QueryExpr::Sort { child, .. }
+            | QueryExpr::Limit { child, .. }
+            | QueryExpr::PromqlSubquery { child, .. }
+            | QueryExpr::TimeRange { child, .. }
+            | QueryExpr::TimeShift { child, .. } => visit(child, windows),
+            QueryExpr::BinaryOp {
+                lhs: left,
+                rhs: right,
+                ..
+            }
+            | QueryExpr::Join { left, right, .. }
+            | QueryExpr::SetOp { left, right, .. } => {
+                visit(left, windows);
+                visit(right, windows);
+            }
+            _ => {}
+        }
+    }
+    let mut windows = BTreeSet::new();
+    visit(expr, &mut windows);
+    windows
+}
+
+/// The window implementations to plan with when the snapshot priced none for
+/// this query.
 ///
-/// This is a *shape*, not a cost quote. `ImplementationCostEvidence` is
-/// measured evidence an evidence producer supplies — its `weighted_cost` doc
-/// is explicit that the producer, not this compiler, prices update CPU,
-/// query-time merges, retained memory, storage, scans and network. So this
-/// function never synthesizes a second candidate to rank: a single candidate
-/// is selected by `complete_summary_candidate_estimate`'s `min_by` over a
-/// one-element list, where the cost value cannot change the outcome. Ranking
-/// Pane against FullWindow requires the snapshot to supply both in
-/// `window_candidates`, each with its own priced evidence.
+/// These are *shapes*, not cost quotes. `ImplementationCostEvidence` is
+/// measured evidence: its `weighted_cost` doc puts pricing update CPU,
+/// query-time merges, retained memory, storage, scans and network on the
+/// evidence producer. So this never synthesizes competing candidates for one
+/// window to rank against each other — one shape per window, selected by a
+/// `min_by` over a one-element list where the cost cannot change the outcome.
+/// Ranking `Pane` against `FullWindow` requires a snapshot supplying both with
+/// their own priced evidence in `window_candidates`.
 ///
-/// What the shape must respect is the query's own demand. A workload that
-/// evaluates every 30s over a 5m lookback needs its state to advance every
-/// 30s; planning it as one 5m tumbling window answers with results that only
-/// change every 5 minutes. `evaluation_interval_ms` already reaches this
-/// function — it was previously read for lifecycle costing and then dropped
-/// on the floor here.
+/// **One candidate per range-selector window, not one per query.** A state
+/// whose window has no candidate is dropped from selection outright
+/// (`hybrid_execution`'s filter), and the surviving candidates are narrowed to
+/// the selected window before validation. Deriving a single candidate from
+/// `time_selection.lookback` therefore silently costs every operand whose own
+/// range differs from it its summary: `sum(sum_over_time(a[1m])) /
+/// sum(sum_over_time(b[5m]))` under a 1m lookback kept `a` and dropped `b` to
+/// exact execution, with nothing reported.
 ///
-/// The derivation, and why each guard exists
-/// (`validate_window_implementations` rejects a candidate that breaks any of
-/// them, so a bad shape would surface as a compile error, never a silent
-/// plan):
+/// Within one window, the shape follows the query's evaluation cadence. A
+/// workload evaluated every 30s over a 5m window needs its state to advance
+/// every 30s; one 5m tumbling window answers with results that only change
+/// once every five minutes. `evaluation_interval_ms` already reaches this
+/// function — it was read for lifecycle costing and then dropped here.
 ///
-/// - `window_secs` is the semantic lookback, which `PlanningQuery::window_secs`
-///   also uses; the validator requires the two to be equal.
-/// - The slide advances one evaluation interval, so consecutive evaluations
-///   share state, but only when that interval is shorter than the window and
-///   divides it. A non-dividing interval (45s into 300s) has no pane width
-///   that divides both, and `WindowMaterializationLayout::validate` would
-///   reject it, so keep the tumbling shape.
-/// - `Pane { pane_secs: slide_secs }` divides the slide trivially and divides
-///   the window by the same guard. Each sample then updates exactly one pane
-///   (`worker.rs`'s `stores_full_windows` branch), and a read composes
-///   `window / slide` of them. `FullWindow` is the other legal Sliding
-///   layout and is deliberately not chosen here: preferring it over panes is
-///   a cost comparison, and this function has no second quote to compare.
-/// - Tumbling pairs only with `Pane` in the validator's framework/layout
-///   table, so the degenerate `pane_secs == window_secs` case stays as it was.
-fn derived_window_candidate(
-    implementation_id: String,
+/// The guards are the validator's own rules, so a bad shape is a compile error
+/// rather than a silent plan: `WindowMaterializationLayout::validate` requires
+/// the pane to divide both window and slide (45s into 300s has no such pane),
+/// and the framework/layout table admits `Tumbling + Pane` and `Sliding + Pane`.
+/// `pane_secs == slide_secs` is the coarsest legal pane for a cadence, so it
+/// is the one with the fewest query-time merges. `FullWindow` is the other
+/// legal `Sliding` layout and is deliberately not emitted alongside it:
+/// preferring it is a write-amplification-versus-read-amplification tradeoff,
+/// which is a cost comparison, and there is no second quote to compare.
+fn derived_window_candidates(
+    implementation_id: &str,
+    expr: &QueryExpr,
     lookback_ms: u64,
     evaluation_interval_ms: u32,
     cost: ImplementationCostEvidence,
-) -> WindowImplementationCandidate {
-    let window_secs = lookback_ms / 1_000;
-    let evaluation_secs = u64::from(evaluation_interval_ms) / 1_000;
-    let advances_within_window = evaluation_secs != 0
-        && evaluation_secs < window_secs
-        && window_secs.is_multiple_of(evaluation_secs);
-    let slide_secs = if advances_within_window {
-        evaluation_secs
-    } else {
-        window_secs
-    };
-    WindowImplementationCandidate {
-        implementation_id,
-        framework: if advances_within_window {
-            SummaryWindowFramework::Sliding
-        } else {
-            SummaryWindowFramework::Tumbling
-        },
-        window_secs,
-        slide_secs,
-        layout: asap_types::WindowMaterializationLayout::Pane {
-            pane_secs: slide_secs,
-        },
-        cost,
-    }
+) -> Vec<WindowImplementationCandidate> {
+    let mut windows = range_selector_windows_secs(expr);
+    if windows.is_empty() {
+        windows.insert(lookback_ms / 1_000);
+    }
+    // `window_implementation_id` reaches lifecycle estimates and cost
+    // manifests, so one label must not describe several shapes. A query with a
+    // single window keeps the snapshot's identity untouched.
+    let distinct = windows.len() > 1;
+    windows
+        .into_iter()
+        .map(|window_secs| {
+            let evaluation_secs = u64::from(evaluation_interval_ms) / 1_000;
+            let advances_within_window = evaluation_secs != 0
+                && evaluation_secs < window_secs
+                && window_secs.is_multiple_of(evaluation_secs);
+            let slide_secs = if advances_within_window {
+                evaluation_secs
+            } else {
+                window_secs
+            };
+            WindowImplementationCandidate {
+                implementation_id: if distinct {
+                    format!("{implementation_id}-{window_secs}s")
+                } else {
+                    implementation_id.to_string()
+                },
+                framework: if advances_within_window {
+                    SummaryWindowFramework::Sliding
+                } else {
+                    SummaryWindowFramework::Tumbling
+                },
+                window_secs,
+                slide_secs,
+                layout: asap_types::WindowMaterializationLayout::Pane {
+                    pane_secs: slide_secs,
+                },
+                cost: cost.clone(),
+            }
+        })
+        .collect()
 }
 
 pub(super) fn validate_window_implementations(
@@ -4929,15 +4990,9 @@ mod tests {
         let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0];
         entry.query = Query("sum(sum_over_time(a[1m])) / sum(sum_over_time(b[5m]))".into());
         entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact);
-        let (mut request, env) = snapshot.planning_request().unwrap();
-        let query = &mut request.queries[0];
-        let mut five_minutes = query.window_implementations[0].clone();
-        five_minutes.implementation_id = "five-minute-evidence".into();
-        five_minutes.window_secs = 300;
-        five_minutes.framework = SummaryWindowFramework::Sliding;
-        five_minutes.slide_secs = 60;
-        five_minutes.layout = asap_types::WindowMaterializationLayout::Pane { pane_secs: 60 };
-        query.window_implementations.push(five_minutes);
+        // The derivation now covers both range selectors, so this no longer
+        // needs a hand-supplied 5m candidate to keep `b` from falling back.
+        let (request, env) = snapshot.planning_request().unwrap();
         let plan = PhysicalCompiler.compile(request, env).unwrap();
         let bindings = plan
             .query_plan
@@ -4959,13 +5014,12 @@ mod tests {
             })
             .collect::<BTreeSet<_>>();
         // `window_ms` is the stored pane width, `readout_lookback_ms` the
-        // semantic range. The snapshot evaluates every 10s, so `a`'s derived
-        // candidate stores 10s panes and composes six of them for its 1m
-        // readout; `b` keeps the 60s pane its explicitly supplied candidate
-        // priced. Both readouts are unchanged.
+        // semantic range. Each operand keeps its own range -- 1m for `a`, 5m
+        // for `b` -- while both store 10s panes, because the snapshot
+        // evaluates every 10s and the derivation now covers both selectors.
         assert_eq!(
             actual,
-            BTreeSet::from([("a", 10_000, Some(60_000)), ("b", 60_000, Some(300_000)),])
+            BTreeSet::from([("a", 10_000, Some(60_000)), ("b", 10_000, Some(300_000)),])
         );
         assert_eq!(plan.precompute_plan.materializations.len(), 2);
     }
@@ -4983,7 +5037,14 @@ mod tests {
     #[test]
     fn derived_window_candidate_follows_the_evaluation_cadence() {
         let cost = planning_snapshot().implementation.implementation_cost;
-        let candidate = derived_window_candidate("id".into(), 300_000, 30_000, cost);
+        let expr = crate::query_parser::parse_query_expr_canonical(
+            "quantile_over_time(0.5, data[5m])",
+            AccuracyTarget::Exact,
+        )
+        .unwrap();
+        let derived = derived_window_candidates("id", &expr, 300_000, 30_000, cost);
+        assert_eq!(derived.len(), 1);
+        let candidate = &derived[0];
         assert_eq!(candidate.framework, SummaryWindowFramework::Sliding);
         assert_eq!((candidate.window_secs, candidate.slide_secs), (300, 30));
         assert_eq!(
@@ -5007,12 +5068,18 @@ mod tests {
         ] {
             let mut query = request.queries[0].clone();
             query.window_secs = lookback_ms / 1_000;
-            query.window_implementations = vec![derived_window_candidate(
-                "derived".into(),
+            let expr = crate::query_parser::parse_query_expr_canonical(
+                &format!("quantile_over_time(0.5, data[{}s])", lookback_ms / 1_000),
+                AccuracyTarget::Exact,
+            )
+            .unwrap();
+            query.window_implementations = derived_window_candidates(
+                "derived",
+                &expr,
                 lookback_ms,
                 evaluation_ms,
                 cost.clone(),
-            )];
+            );
             validate_window_implementations(&query, &environment).unwrap_or_else(|error| {
                 panic!("lookback {lookback_ms} cadence {evaluation_ms}: {error:?}")
             });
@@ -5025,9 +5092,15 @@ mod tests {
     #[test]
     fn derived_window_candidate_stays_tumbling_without_a_dividing_cadence() {
         let cost = planning_snapshot().implementation.implementation_cost;
+        let expr = crate::query_parser::parse_query_expr_canonical(
+            "quantile_over_time(0.5, data[5m])",
+            AccuracyTarget::Exact,
+        )
+        .unwrap();
         for evaluation_ms in [300_000, 450_000, 45_000, 0] {
-            let candidate =
-                derived_window_candidate("id".into(), 300_000, evaluation_ms, cost.clone());
+            let derived =
+                derived_window_candidates("id", &expr, 300_000, evaluation_ms, cost.clone());
+            let candidate = &derived[0];
             assert_eq!(
                 (
                     candidate.framework.clone(),
@@ -5053,12 +5126,19 @@ mod tests {
             .query
             .0
             .clone();
-        let mut supplied = derived_window_candidate(
-            "supplied".into(),
+        let expr = crate::query_parser::parse_query_expr_canonical(
+            "quantile_over_time(0.99, m[1m])",
+            AccuracyTarget::Exact,
+        )
+        .unwrap();
+        let mut supplied = derived_window_candidates(
+            "supplied",
+            &expr,
             60_000,
             60_000,
             snapshot.implementation.implementation_cost.clone(),
-        );
+        )
+        .remove(0);
         supplied.framework = SummaryWindowFramework::Sliding;
         supplied.slide_secs = 20;
         supplied.layout = asap_types::WindowMaterializationLayout::Pane { pane_secs: 20 };
@@ -5122,9 +5202,10 @@ mod tests {
         );
     }
 
-    // A filtered denominator is a typed residual while its summary sibling remains installed.
+    // A filtered operand gets a summary over its own filtered population,
+    // with each operand keeping its own range.
     #[test]
-    fn composable_binary_retains_summary_sibling_of_prometheus_filtered_subtree() {
+    fn composable_binary_summarizes_each_prometheus_filtered_operand() {
         use crate::query_plan::{logical::LogicalOperator, QueryPlanNode};
         let mut snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!(
             "../../../docs/examples/asapquery-planning-snapshot.json"
@@ -5138,22 +5219,35 @@ mod tests {
         let plan = PhysicalCompiler.compile(request, env).unwrap();
         let query = plan.query_plan.entries.values().next().unwrap();
         let bindings = query.materialization_bindings();
-        assert_eq!(bindings.len(), 1);
-        let identity = &plan.summary_catalog.materializations[&bindings[0].materialization];
-        let data = &plan.summary_catalog.data_descriptors[&identity.data_descriptor_id];
-        // 10s panes for a 10s evaluation cadence; the 1m readout range is
-        // carried by `readout_lookback_ms`, not by the stored pane width.
+        // Both operands now hold a summary. The filtered denominator is no
+        // longer a typed residual: its 5m range has a candidate, so it gets
+        // its own summary over the filtered population rather than exact
+        // execution. Nothing about the filter forced the residual -- the
+        // missing 5m window candidate did, and this test previously pinned
+        // that artifact as intended behavior.
+        let bound = bindings
+            .iter()
+            .map(|binding| {
+                let identity = &plan.summary_catalog.materializations[&binding.materialization];
+                let data = &plan.summary_catalog.data_descriptors[&identity.data_descriptor_id];
+                (
+                    data.time_series_metric().unwrap(),
+                    data.population_filter_canonical.clone(),
+                    binding.readout_lookback_ms,
+                )
+            })
+            .collect::<BTreeSet<_>>();
         assert_eq!(
-            (data.time_series_metric().unwrap(), bindings[0].window_ms),
-            ("a", 10_000)
+            bound,
+            BTreeSet::from([
+                ("a", String::new(), Some(60_000)),
+                ("b", "{job!=\"x\"}".to_string(), Some(300_000)),
+            ])
         );
         assert!(!query
             .nodes
             .values()
             .any(|node| matches!(node, QueryPlanNode::ExactFallback { .. })));
-        assert!(query.nodes.values().any(|node| matches!(node,
-            QueryPlanNode::Logical { operator: LogicalOperator::ExactSubquery { query }, .. }
-            if query == "sum_over_time(b{job!=\"x\"}[5m])" || query == "sum(sum_over_time(b{job!=\"x\"}[5m]))")));
         assert!(!query.nodes.values().any(|node| matches!(
             node,
             QueryPlanNode::Logical {
@@ -5161,7 +5255,7 @@ mod tests {
                 ..
             }
         )));
-        assert_eq!(plan.precompute_plan.materializations.len(), 1);
+        assert_eq!(plan.precompute_plan.materializations.len(), 2);
     }
 
     #[test]
diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs
index ecf98038..0a82532a 100644
--- a/control_plane/src/physical/workload_cost.rs
+++ b/control_plane/src/physical/workload_cost.rs
@@ -991,7 +991,10 @@ mod tests {
                 .count(),
             1
         );
-        assert!(manifest
+        // Both ranges are summarized now, so nothing in this workload is
+        // priced as exact backend execution. The 5m operand used to land there
+        // only because its window had no implementation candidate.
+        assert!(!manifest
             .components
             .values()
             .any(|demand| demand.implementation.get("location")

From bf2896ea37b61bfb9850ddae9d208f156696ca8d Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Sat, 12 Sep 2026 10:34:36 -0600
Subject: [PATCH 3/6] feat(planning): price the derived window layouts so they
 can be ranked

A sliding window has two legal physical layouts, and they trade opposite
resources. `Pane { P }` takes each sample once and composes `W / P`
states per read; `FullWindow` takes each sample into all `ceil(W / S)`
overlapping windows and is read whole. Which is cheaper depends on how
hard the source is pushing, and nothing was comparing them: the
derivation emitted only the pane form, and a lone candidate is selected
by a `min_by` over one element.

Emit both for a sliding shape and price each from the snapshot's own
lifecycle unit costs.

`ImplementationCostEvidence` is normally measured evidence, so becoming
its producer here is only sound if nothing is invented. Every unit cost
is supplied (`implementation.lifecycle_costs`), and every multiplier is
a structural count that follows from the layout's definition: states
sealed per horizon, update fanout, finalizations per read, and
`retained_state_count` -- the same function that fills
`num_aggregates_to_retain`, so quote and plan cannot disagree.
Concurrent in-flight full windows are not charged twice; the fanout term
already prices that write amplification.

The byte fields stay exactly as the snapshot supplied them. State size
needs sketch parameters that do not exist yet at this point, and
guessing them would be the fabrication this avoids. Only `cpu_cost` and
`weighted_cost` are derived, and only those two are read -- by
`validate_window_implementations` and by the ranking `min_by`.
`model_version` records that the quote is derived.

Against the planning snapshot's own evidence the crossover lands where
it should: idle, the full window wins because its fanout is free and the
pane's ten merges per read are not; at the declared 100 updates/s the
tenfold fanout dominates and panes win. Tumbling shapes are unchanged --
the validator pairs them only with `Pane`, so there is no alternative to
rank and their identity keeps the snapshot's label.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 control_plane/src/physical/compiler.rs | 285 +++++++++++++++++++++++--
 1 file changed, 263 insertions(+), 22 deletions(-)

diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index 9123c9ff..0783418c 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -608,6 +608,7 @@ impl BackendLocalPlanningSnapshot {
                 horizon_seconds: self.implementation.horizon_seconds,
                 costs: self.implementation.lifecycle_costs.clone(),
             };
+            let derived_lifecycle = lifecycle.clone();
             let post_asap = crate::planner_selection::keep_pre_asap(&parsed)
                 .map_err(|error| CompileError::Snapshot(format!("query {index}: {error}")))?;
             canonical_roots.push(Rc::new(parsed));
@@ -642,6 +643,8 @@ impl BackendLocalPlanningSnapshot {
                             lookback_ms,
                             evaluation_interval_ms,
                             cost,
+                            &derived_lifecycle,
+                            self.implementation.query_staleness_margin_ms,
                         )
                     }),
                 runtime_policy: RuntimeRulePolicy::default(),
@@ -2185,6 +2188,97 @@ fn validate_lifecycle_input(
     Ok(())
 }
 
+/// Price one derived layout from the snapshot's own lifecycle unit costs.
+///
+/// `ImplementationCostEvidence` is normally measured evidence, and its
+/// `weighted_cost` doc puts pricing update CPU, query-time merges, retained
+/// memory, storage, scans and network on the evidence producer. When a
+/// snapshot prices no candidate, the control plane becomes that producer for
+/// the derived shapes — and it does so without inventing a single magnitude.
+/// Every unit cost below is supplied evidence (`LifecycleCostEvidence`, from
+/// `implementation.lifecycle_costs`); every multiplier is a structural count
+/// that follows from the layout's definition. Nothing here is a measurement.
+///
+/// Over `horizon_seconds`, for window `W`, slide `S` and the layout's own
+/// shape:
+///
+/// - **states created**: a `Pane { P }` seals a state every `P`, a
+///   `FullWindow` every `S`. Each one is built once and retired once, so
+///   `build` and `retirement` are charged per created state.
+/// - **update fanout**: this is the layout's whole point (`worker.rs`'s
+///   `stores_full_windows` branch). A pane takes each sample exactly once; a
+///   full window takes it into every overlapping window that contains it,
+///   `ceil(W / S)` of them. Charged at the supplied ingestion rate.
+/// - **finalizations per read**: the mirror image. A full window is read
+///   whole; `W / P` panes are composed into one answer. Charged at the
+///   query's own evaluation cadence.
+/// - **retention**: `retained_state_count`, the same function that fills
+///   `num_aggregates_to_retain`, so the quote and the plan cannot disagree.
+///   Concurrent in-flight full windows are not double-charged here — the
+///   fanout term already prices that write amplification.
+///
+/// The byte fields are left exactly as the snapshot supplied them: state size
+/// needs sketch parameters that do not exist yet at this point, and guessing
+/// them would be the fabrication this function otherwise avoids. Only
+/// `cpu_cost` and `weighted_cost` are derived, and only those two are read —
+/// by `validate_window_implementations` and by the `min_by` that ranks
+/// candidates. `model_version` records that the quote is derived.
+fn derived_window_cost(
+    template: &ImplementationCostEvidence,
+    lifecycle: &LifecyclePlanningInput,
+    window_secs: u64,
+    slide_secs: u64,
+    layout: &asap_types::WindowMaterializationLayout,
+    staleness_margin_ms: u64,
+) -> ImplementationCostEvidence {
+    let costs = &lifecycle.costs;
+    let horizon = lifecycle.horizon_seconds.max(0.0);
+    let window = window_secs.max(1) as f64;
+    let slide = slide_secs.max(1) as f64;
+    let (seal_interval, update_fanout, finalizations_per_read) = match layout {
+        asap_types::WindowMaterializationLayout::FullWindow => {
+            (slide, (window / slide).ceil(), 1.0)
+        }
+        asap_types::WindowMaterializationLayout::Pane { pane_secs } => {
+            let pane = (*pane_secs).max(1) as f64;
+            (pane, 1.0, (window / pane).ceil())
+        }
+        asap_types::WindowMaterializationLayout::HierarchicalRollup { base_pane_secs, .. } => {
+            let pane = (*base_pane_secs).max(1) as f64;
+            (pane, 1.0, (window / pane).ceil())
+        }
+    };
+    let states_created = horizon / seal_interval;
+    let updates = lifecycle.ingestion_rate_per_second.max(0.0) * horizon * update_fanout;
+    let evaluation_secs = (f64::from(lifecycle.evaluation_interval_ms) / 1_000.0).max(1.0);
+    let reads = horizon / evaluation_secs;
+    let retained = retained_state_count(
+        window_secs.saturating_mul(1_000),
+        staleness_margin_ms,
+        slide_secs.saturating_mul(1_000),
+        layout,
+    ) as f64;
+
+    let build = costs.build * states_created;
+    let maintenance = costs.maintenance_per_update * updates;
+    let read = costs.read * reads * finalizations_per_read;
+    let retention = costs.retention_per_second * horizon * retained;
+    let retirement = costs.retirement * states_created;
+    let cpu_cost = build + maintenance;
+    let weighted_cost = cpu_cost + read + retention + retirement;
+
+    ImplementationCostEvidence {
+        model_version: format!("{}+derived-window-layout-v1", template.model_version),
+        cpu_cost: if cpu_cost.is_finite() { cpu_cost } else { 0.0 },
+        weighted_cost: if weighted_cost.is_finite() {
+            weighted_cost
+        } else {
+            0.0
+        },
+        ..template.clone()
+    }
+}
+
 /// Every distinct range-selector window in `expr`, as seconds.
 ///
 /// A single query can carry several. `sum(sum_over_time(a[1m])) / sum(sum_over_time(b[5m]))`
@@ -2274,6 +2368,8 @@ fn derived_window_candidates(
     lookback_ms: u64,
     evaluation_interval_ms: u32,
     cost: ImplementationCostEvidence,
+    lifecycle: &LifecyclePlanningInput,
+    staleness_margin_ms: u64,
 ) -> Vec<WindowImplementationCandidate> {
     let mut windows = range_selector_windows_secs(expr);
     if windows.is_empty() {
@@ -2285,7 +2381,7 @@ fn derived_window_candidates(
     let distinct = windows.len() > 1;
     windows
         .into_iter()
-        .map(|window_secs| {
+        .flat_map(|window_secs| {
             let evaluation_secs = u64::from(evaluation_interval_ms) / 1_000;
             let advances_within_window = evaluation_secs != 0
                 && evaluation_secs < window_secs
@@ -2295,24 +2391,58 @@ fn derived_window_candidates(
             } else {
                 window_secs
             };
-            WindowImplementationCandidate {
-                implementation_id: if distinct {
-                    format!("{implementation_id}-{window_secs}s")
-                } else {
-                    implementation_id.to_string()
-                },
-                framework: if advances_within_window {
-                    SummaryWindowFramework::Sliding
+            let window_label = if distinct {
+                format!("{implementation_id}-{window_secs}s")
+            } else {
+                implementation_id.to_string()
+            };
+            // `Tumbling` pairs only with `Pane` in the validator's
+            // framework/layout table, so a non-sliding shape has no
+            // alternative to rank against and keeps its label unchanged.
+            let layouts: Vec<(String, asap_types::WindowMaterializationLayout)> =
+                if advances_within_window {
+                    vec![
+                        (
+                            format!("{window_label}-pane-{slide_secs}s"),
+                            asap_types::WindowMaterializationLayout::Pane {
+                                pane_secs: slide_secs,
+                            },
+                        ),
+                        (
+                            format!("{window_label}-full-window"),
+                            asap_types::WindowMaterializationLayout::FullWindow,
+                        ),
+                    ]
                 } else {
-                    SummaryWindowFramework::Tumbling
-                },
-                window_secs,
-                slide_secs,
-                layout: asap_types::WindowMaterializationLayout::Pane {
-                    pane_secs: slide_secs,
-                },
-                cost: cost.clone(),
-            }
+                    vec![(
+                        window_label,
+                        asap_types::WindowMaterializationLayout::Pane {
+                            pane_secs: slide_secs,
+                        },
+                    )]
+                };
+            layouts
+                .into_iter()
+                .map(|(id, layout)| WindowImplementationCandidate {
+                    implementation_id: id,
+                    framework: if advances_within_window {
+                        SummaryWindowFramework::Sliding
+                    } else {
+                        SummaryWindowFramework::Tumbling
+                    },
+                    window_secs,
+                    slide_secs,
+                    cost: derived_window_cost(
+                        &cost,
+                        lifecycle,
+                        window_secs,
+                        slide_secs,
+                        &layout,
+                        staleness_margin_ms,
+                    ),
+                    layout,
+                })
+                .collect::<Vec<_>>()
         })
         .collect()
 }
@@ -5024,6 +5154,12 @@ mod tests {
         assert_eq!(plan.precompute_plan.materializations.len(), 2);
     }
 
+    fn planning_lifecycle() -> LifecyclePlanningInput {
+        planning_snapshot().planning_request().unwrap().0.queries[0]
+            .lifecycle
+            .clone()
+    }
+
     fn planning_snapshot() -> BackendLocalPlanningSnapshot {
         serde_json::from_str(include_str!(
             "../../../docs/examples/asapquery-planning-snapshot.json"
@@ -5042,8 +5178,26 @@ mod tests {
             AccuracyTarget::Exact,
         )
         .unwrap();
-        let derived = derived_window_candidates("id", &expr, 300_000, 30_000, cost);
-        assert_eq!(derived.len(), 1);
+        let derived =
+            derived_window_candidates("id", &expr, 300_000, 30_000, cost, &planning_lifecycle(), 0);
+        // A sliding shape has two legal layouts, so both are offered and the
+        // cost model picks between them.
+        assert_eq!(
+            derived
+                .iter()
+                .map(|c| (c.implementation_id.as_str(), c.layout.clone()))
+                .collect::<Vec<_>>(),
+            vec![
+                (
+                    "id-pane-30s",
+                    asap_types::WindowMaterializationLayout::Pane { pane_secs: 30 }
+                ),
+                (
+                    "id-full-window",
+                    asap_types::WindowMaterializationLayout::FullWindow
+                ),
+            ]
+        );
         let candidate = &derived[0];
         assert_eq!(candidate.framework, SummaryWindowFramework::Sliding);
         assert_eq!((candidate.window_secs, candidate.slide_secs), (300, 30));
@@ -5053,6 +5207,82 @@ mod tests {
         );
     }
 
+    // The layout choice is a write-amplification-versus-read-amplification
+    // trade, and the derived quote has to price it in the right direction: a
+    // full window takes every sample into all ten overlapping windows but is
+    // read whole, panes take each sample once but compose ten per read. Which
+    // wins depends on how hard the source is pushing, so pin the crossover,
+    // not the magnitudes.
+    #[test]
+    fn derived_window_layout_prices_write_against_read_amplification() {
+        let cost = planning_snapshot().implementation.implementation_cost;
+        let expr = crate::query_parser::parse_query_expr_canonical(
+            "quantile_over_time(0.5, data[5m])",
+            AccuracyTarget::Exact,
+        )
+        .unwrap();
+        let quote = |rate: f64| {
+            let mut lifecycle = planning_lifecycle();
+            lifecycle.ingestion_rate_per_second = rate;
+            let derived = derived_window_candidates(
+                "id",
+                &expr,
+                300_000,
+                30_000,
+                cost.clone(),
+                &lifecycle,
+                0,
+            );
+            let weighted = |layout: &asap_types::WindowMaterializationLayout| {
+                derived
+                    .iter()
+                    .find(|c| c.layout == *layout)
+                    .expect("both layouts offered")
+                    .cost
+                    .weighted_cost
+            };
+            (
+                weighted(&asap_types::WindowMaterializationLayout::Pane { pane_secs: 30 }),
+                weighted(&asap_types::WindowMaterializationLayout::FullWindow),
+            )
+        };
+        let (idle_pane, idle_full) = quote(0.0);
+        assert!(
+            idle_full < idle_pane,
+            "with no arriving data the fanout is free and the merges are not: \
+             pane {idle_pane} full {idle_full}"
+        );
+        let (busy_pane, busy_full) = quote(100.0);
+        assert!(
+            busy_pane < busy_full,
+            "under load the tenfold update fanout dominates: pane {busy_pane} full {busy_full}"
+        );
+    }
+
+    // A tumbling shape pairs only with `Pane` in the validator's
+    // framework/layout table, so there is no alternative to price against it.
+    #[test]
+    fn tumbling_shapes_have_no_layout_alternative_to_rank() {
+        let cost = planning_snapshot().implementation.implementation_cost;
+        let expr = crate::query_parser::parse_query_expr_canonical(
+            "quantile_over_time(0.5, data[5m])",
+            AccuracyTarget::Exact,
+        )
+        .unwrap();
+        let derived = derived_window_candidates(
+            "id",
+            &expr,
+            300_000,
+            300_000,
+            cost,
+            &planning_lifecycle(),
+            0,
+        );
+        assert_eq!(derived.len(), 1);
+        assert_eq!(derived[0].implementation_id, "id");
+        assert_eq!(derived[0].framework, SummaryWindowFramework::Tumbling);
+    }
+
     // Every shape this function can emit must survive the validator, or a bad
     // derivation would reach a plan instead of a compile error.
     #[test]
@@ -5079,6 +5309,8 @@ mod tests {
                 lookback_ms,
                 evaluation_ms,
                 cost.clone(),
+                &planning_lifecycle(),
+                0,
             );
             validate_window_implementations(&query, &environment).unwrap_or_else(|error| {
                 panic!("lookback {lookback_ms} cadence {evaluation_ms}: {error:?}")
@@ -5098,8 +5330,15 @@ mod tests {
         )
         .unwrap();
         for evaluation_ms in [300_000, 450_000, 45_000, 0] {
-            let derived =
-                derived_window_candidates("id", &expr, 300_000, evaluation_ms, cost.clone());
+            let derived = derived_window_candidates(
+                "id",
+                &expr,
+                300_000,
+                evaluation_ms,
+                cost.clone(),
+                &planning_lifecycle(),
+                0,
+            );
             let candidate = &derived[0];
             assert_eq!(
                 (
@@ -5137,6 +5376,8 @@ mod tests {
             60_000,
             60_000,
             snapshot.implementation.implementation_cost.clone(),
+            &planning_lifecycle(),
+            0,
         )
         .remove(0);
         supplied.framework = SummaryWindowFramework::Sliding;

From 577bd83e3f4fb1ac91d5a964e904a78371148df3 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Sat, 12 Sep 2026 11:22:11 -0600
Subject: [PATCH 4/6] Require complete cost evidence for snapshot deployment

---
 control_plane/src/backend_client.rs           |   6 +-
 control_plane/src/physical/compiler.rs        | 190 +++++++++++++-----
 control_plane/src/physical/erp.rs             |   4 +-
 control_plane/src/physical/workload_cost.rs   |   3 +-
 .../asapquery_compatibility_process_e2e.rs    |  69 ++++++-
 .../support/distinct_planning_process.rs      |  21 +-
 .../tests/support/durable_summary_process.rs  |   2 +-
 .../tests/support/erp_planning_process.rs     |   5 +-
 .../support/immutable_maintenance_process.rs  |  17 +-
 .../tests/support/univmon_erp_process.rs      |  16 +-
 ...asapquery-compatibility-demo-snapshot.json |   2 +-
 .../examples/asapquery-planning-snapshot.json |   2 +-
 12 files changed, 248 insertions(+), 89 deletions(-)

diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs
index fb60d3cc..75841913 100644
--- a/control_plane/src/backend_client.rs
+++ b/control_plane/src/backend_client.rs
@@ -601,7 +601,11 @@ mod tests {
                 "../../docs/examples/asapquery-planning-snapshot.json"
             ))
             .unwrap();
-        let publication = snapshot.compile().unwrap().publication().unwrap();
+        let publication = crate::physical::compiler::tests::quoted_snapshot(snapshot, false)
+            .compile()
+            .unwrap()
+            .publication()
+            .unwrap();
         let hits: StdArc<Mutex<Vec<serde_json::Value>>> = StdArc::new(Mutex::new(Vec::new()));
         let route_hits = hits.clone();
         let app = Router::new().route(
diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index b57f00e8..3528a6a1 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -198,7 +198,8 @@ pub enum PhysicalDeploymentTarget {
     BackendLocalRemoteWrite,
 }
 
-/// Versioned startup input for the Collector-free compatibility profile.
+/// Startup and candidate-discovery input for backend-local planning.
+/// Version 2 is the sole supported schema; deployment always requires quotes.
 /// Query/data semantics use ASAPPlanner's canonical workload types directly;
 /// this wrapper adds only backend-owned implementation evidence and lifecycle
 /// identity required to choose a concrete physical realization.
@@ -206,6 +207,7 @@ pub enum PhysicalDeploymentTarget {
 #[serde(deny_unknown_fields)]
 pub struct BackendLocalPlanningSnapshot {
     pub snapshot_version: u32,
+    /// May be absent during candidate discovery, never during deployment.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub workload_cost_evidence: Option<super::workload_cost::WorkloadCostEvidence>,
     pub query_workload: QueryWorkload,
@@ -480,34 +482,17 @@ impl BackendLocalPlanningSnapshot {
     }
 
     fn compile_frontend(self, metricsql: bool) -> Result<PhysicalPlan, CompileError> {
-        let evidence = self.workload_cost_evidence.clone();
-        if self.snapshot_version == 2 && evidence.is_none() {
-            return Err(CompileError::Snapshot(
-                "version 2 requires complete workload cost evidence".into(),
-            ));
-        }
+        let evidence = self.workload_cost_evidence.clone().ok_or_else(|| {
+            CompileError::Snapshot(
+                "deployment requires complete workload cost evidence; export candidates and price them before compiling".into(),
+            )
+        })?;
         let (request, environment) = self.planning_request()?;
-        match evidence {
-            Some(evidence) => {
-                let candidates = super::workload_cost::with_exact_alternative(request)?;
-                if metricsql {
-                    super::workload_cost::select_metricsql(candidates, environment, &evidence)
-                } else {
-                    super::workload_cost::select(candidates, environment, &evidence)
-                }
-            }
-            None => {
-                // Unquoted v1 startup snapshots keep the established summary/native
-                // compatibility policy. Local residual candidates are enumerated by
-                // planning_request and admitted through measured workload selection.
-                let mut request = request;
-                request.hybrid_execution = false;
-                if metricsql {
-                    PhysicalCompiler.compile_metricsql(request, environment)
-                } else {
-                    PhysicalCompiler.compile(request, environment)
-                }
-            }
+        let candidates = super::workload_cost::with_exact_alternative(request)?;
+        if metricsql {
+            super::workload_cost::select_metricsql(candidates, environment, &evidence)
+        } else {
+            super::workload_cost::select(candidates, environment, &evidence)
         }
     }
 
@@ -515,9 +500,9 @@ impl BackendLocalPlanningSnapshot {
     pub fn planning_request(
         self,
     ) -> Result<(PlanningRequest, DeploymentEnvironment), CompileError> {
-        if self.snapshot_version != 1 && self.snapshot_version != 2 {
+        if self.snapshot_version != 2 {
             return Err(CompileError::Snapshot(format!(
-                "unsupported workload snapshot version {}",
+                "unsupported workload snapshot version {}; only version 2 is supported",
                 self.snapshot_version
             )));
         }
@@ -1058,7 +1043,9 @@ impl PhysicalCompiler {
                                 .ok()
                                 .flatten()
                             });
-                            key.is_some_and(|key| policy.contains(&key))
+                            // Masks enumerate counter/max choices only. Other selected
+                            // summaries remain required by this physical alternative.
+                            key.is_none_or(|key| policy.contains(&key))
                         })
                 })
                 .collect::<Vec<_>>();
@@ -1553,7 +1540,31 @@ impl PhysicalCompiler {
                 full_history: false,
                 cumulative_readout: true,
             };
-            let mut entry = if request.hybrid_execution {
+            // A whole-query native fallback need not be expressible in the local
+            // residual algebra (for example an ERP-rejected entropy readout).
+            // Retain its native boundary without discarding other workload roots.
+            let native_root = request.hybrid_execution
+                && if let SummaryExpr::KeepPreAsap(expr) = &query.post_asap.expr {
+                    let original = crate::query_parser::parse_query_expr_canonical(
+                        &query.query_string,
+                        query.accuracy.clone(),
+                    )
+                    .map_err(|error| CompileError::Query {
+                        query_id: query.query_id.clone(),
+                        reason: error.to_string(),
+                    })?;
+                    expr.as_ref() == &original
+                        && crate::query_plan::logical::compile_logical(
+                            query.query_id.clone(),
+                            canonical.clone(),
+                            instant.clone(),
+                            FallbackPolicy::ExactBackend,
+                        )
+                        .is_err()
+                } else {
+                    false
+                };
+            let mut entry = if request.hybrid_execution && !native_root {
                 crate::query_plan::compile_bound_composable_mapped(
                     query.query_id.clone(),
                     canonical.clone(),
@@ -3237,9 +3248,88 @@ fn stable_workload_plan_id(
 }
 
 #[cfg(test)]
-mod tests {
+pub(crate) mod tests {
     use super::*;
 
+    // Synthetic quotes exercise deployment selection in tests, never production defaults.
+    pub(crate) fn quoted_snapshot(
+        mut snapshot: BackendLocalPlanningSnapshot,
+        metricsql: bool,
+    ) -> BackendLocalPlanningSnapshot {
+        use super::super::workload_cost::{
+            manifest, with_exact_alternative, WorkloadCostEvidence, WorkloadQuote,
+        };
+        let (request, environment) = snapshot.clone().planning_request().unwrap();
+        let quotes = with_exact_alternative(request)
+            .unwrap()
+            .into_iter()
+            .enumerate()
+            .filter_map(|(index, candidate)| {
+                let plan = if metricsql {
+                    PhysicalCompiler.compile_metricsql(candidate.clone(), environment.clone())
+                } else {
+                    PhysicalCompiler.compile(candidate.clone(), environment.clone())
+                }
+                .ok()?;
+                let manifest = manifest(&plan, &candidate.queries).unwrap();
+                Some(WorkloadQuote {
+                    unit_costs: manifest
+                        .components
+                        .keys()
+                        .map(|key| (key.clone(), if index == 0 { 1.0 } else { 1e12 }))
+                        .collect(),
+                    manifest,
+                    executable: true,
+                })
+            })
+            .collect();
+        snapshot.workload_cost_evidence = Some(WorkloadCostEvidence {
+            backend_revision: BACKEND_REVISION.into(),
+            planner_revision: PLANNER_REVISION.into(),
+            data_snapshot_id: "compiler-unit-fixture".into(),
+            model_version: "test-only-unit-costs".into(),
+            observed_at_unix_ms: environment.observed_at_unix_ms,
+            valid_for_ms: environment.max_evidence_age_ms,
+            quotes,
+        });
+        snapshot
+    }
+
+    /// Optional counter masks must retain the workload's mandatory sketch bindings.
+    #[test]
+    fn costed_mixed_workload_retains_sketches_and_counter_readouts() {
+        let snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!(
+            "../../../docs/examples/asapquery-compatibility-demo-snapshot.json"
+        ))
+        .unwrap();
+        let plan = quoted_snapshot(snapshot, false).compile().unwrap();
+        assert!(!plan.precompute_plan.materializations.is_empty());
+        for entry in plan.query_plan.entries.values() {
+            assert!(!entry.materialization_bindings().is_empty(), "{entry:#?}");
+        }
+    }
+
+    /// A schema marker cannot opt into a legacy deployment policy.
+    #[test]
+    fn only_current_snapshot_schema_is_accepted() {
+        let snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!(
+            "../../../docs/examples/asapquery-planning-snapshot.json"
+        ))
+        .unwrap();
+        for version in [0, 1, 3] {
+            let mut old = snapshot.clone();
+            old.snapshot_version = version;
+            assert!(old
+                .clone()
+                .planning_request()
+                .unwrap_err()
+                .to_string()
+                .contains("only version 2"));
+            assert!(old.compile().is_err());
+        }
+        assert!(snapshot.planning_request().is_ok());
+    }
+
     #[test]
     fn installed_partition_must_match_the_bound_dag_reduction() {
         let mut env = environment(10_000);
@@ -3957,7 +4047,7 @@ mod tests {
             "../../../docs/examples/asapquery-compatibility-demo-snapshot.json"
         ))
         .unwrap();
-        let plan = snapshot.compile_metricsql().unwrap();
+        let plan = quoted_snapshot(snapshot, true).compile_metricsql().unwrap();
         assert!(!plan.query_plan.entries.is_empty());
         assert!(plan
             .query_plan
@@ -4993,7 +5083,7 @@ mod tests {
             "../../../docs/examples/asapquery-planning-snapshot.json"
         ))
         .unwrap();
-        let bundle = snapshot.compile().unwrap();
+        let bundle = quoted_snapshot(snapshot, false).compile().unwrap();
         let catalog = &bundle.summary_catalog;
         let mut transmission = bundle.transmission_plan.clone();
         transmission.validate_against_catalog(catalog).unwrap();
@@ -5061,7 +5151,7 @@ mod tests {
         let mut second = entries[0].clone();
         second.query = Query("sum(sum_over_time(m[1m])) * 2".into());
         entries.push(second);
-        let bundle = snapshot.compile().unwrap();
+        let bundle = quoted_snapshot(snapshot, false).compile().unwrap();
         assert_eq!(bundle.query_plan.entries.len(), 2);
         assert_eq!(bundle.precompute_plan.materializations.len(), 1);
         let query_plan: QueryPlan =
@@ -5403,7 +5493,7 @@ mod tests {
             .queries
             .remove(0);
         let snapshot = BackendLocalPlanningSnapshot {
-            snapshot_version: 1,
+            snapshot_version: 2,
             workload_cost_evidence: None,
             query_workload,
             data_workload,
@@ -5434,12 +5524,10 @@ mod tests {
                 .as_ref(),
             Some(&snapshot.query_workload)
         );
-        let first = snapshot
-            .clone()
+        let first = quoted_snapshot(snapshot.clone(), false)
             .compile()
             .expect("first deterministic plan");
-        let second = snapshot
-            .clone()
+        let second = quoted_snapshot(snapshot.clone(), false)
             .compile()
             .expect("second deterministic plan");
         assert_eq!(first.envelope, second.envelope);
@@ -5454,7 +5542,9 @@ mod tests {
         let encoded = serde_json::to_vec(&snapshot).expect("serialize startup snapshot");
         let decoded: BackendLocalPlanningSnapshot =
             serde_json::from_slice(&encoded).expect("deserialize startup snapshot");
-        let bundle = decoded.compile().expect("canonical startup planning");
+        let bundle = quoted_snapshot(decoded, false)
+            .compile()
+            .expect("canonical startup planning");
 
         assert!(bundle.collector_plans.is_empty());
         assert!(bundle.transmission_plan.rules.is_empty());
@@ -5542,10 +5632,10 @@ mod tests {
         let fixture: serde_json::Value = serde_json::from_str(source).expect("fixture JSON");
         assert_eq!(encoded, fixture);
 
-        snapshot
-            .clone()
-            .compile()
-            .expect("unquoted v1 compatibility startup remains available");
+        assert!(
+            snapshot.clone().compile().is_err(),
+            "discovery fixtures must be priced before deployment"
+        );
         let (local, env) = snapshot.clone().planning_request().unwrap();
         let isolated = PhysicalCompiler.compile(local, env).unwrap();
         assert!(!isolated.precompute_plan.materializations.is_empty());
@@ -5578,10 +5668,10 @@ mod tests {
             include_str!("../../../docs/examples/asapquery-compatibility-demo-snapshot.json");
         let snapshot: BackendLocalPlanningSnapshot =
             serde_json::from_str(source).expect("strict compatibility demo fixture");
-        snapshot
-            .clone()
-            .compile()
-            .expect("unquoted v1 compatibility startup remains available");
+        assert!(
+            snapshot.clone().compile().is_err(),
+            "discovery fixtures must be priced before deployment"
+        );
         let (local, env) = snapshot.clone().planning_request().unwrap();
         let isolated = PhysicalCompiler.compile(local, env).unwrap();
         assert!(!isolated.precompute_plan.materializations.is_empty());
diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs
index c41fa2fe..a9e139f1 100644
--- a/control_plane/src/physical/erp.rs
+++ b/control_plane/src/physical/erp.rs
@@ -1459,7 +1459,9 @@ mod tests {
         fixture["query_workload"]["repeating_queries"] = serde_json::json!([query]);
         let snapshot: crate::physical::compiler::BackendLocalPlanningSnapshot =
             serde_json::from_value(fixture).unwrap();
-        let plan = snapshot.compile().unwrap();
+        let plan = crate::physical::compiler::tests::quoted_snapshot(snapshot, false)
+            .compile()
+            .unwrap();
         let (mut policy, mut observed) = online_population_fixture();
         observed.catalog_generation = plan.summary_catalog.reference().unwrap();
         observed.summary_definition_id =
diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs
index ecf98038..b3dde46c 100644
--- a/control_plane/src/physical/workload_cost.rs
+++ b/control_plane/src/physical/workload_cost.rs
@@ -1144,9 +1144,8 @@ mod tests {
     }
 
     #[test]
-    fn v2_snapshot_requires_quotes_and_roundtrips_selection() {
+    fn snapshot_requires_quotes_and_roundtrips_selection() {
         let mut snapshot = fixture();
-        snapshot.snapshot_version = 2;
         assert!(snapshot.clone().compile().is_err());
         let (_, _, evidence) = quoted();
         snapshot.workload_cost_evidence = Some(evidence);
diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs
index 54cad0d8..e6953605 100644
--- a/data_plane/tests/asapquery_compatibility_process_e2e.rs
+++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs
@@ -28,6 +28,60 @@ mod durable_summary_process;
 #[path = "support/immutable_maintenance_process.rs"]
 mod immutable_maintenance_process;
 
+// Test-only quotes preserve the fixture's local candidate without a production bypass.
+fn quote_snapshot_for_test(
+    snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot,
+) -> control_plane::physical::compiler::BackendLocalPlanningSnapshot {
+    quote_snapshot_for_frontend_test(snapshot, false)
+}
+
+fn quote_snapshot_for_frontend_test(
+    mut snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot,
+    metricsql: bool,
+) -> control_plane::physical::compiler::BackendLocalPlanningSnapshot {
+    use control_plane::physical::{
+        compiler::{PhysicalCompiler, BACKEND_REVISION, PLANNER_REVISION},
+        workload_cost::{self, WorkloadCostEvidence, WorkloadQuote},
+    };
+    let (request, environment) = snapshot.clone().planning_request().unwrap();
+    let mut preferred = true;
+    let quotes = workload_cost::with_exact_alternative(request)
+        .unwrap()
+        .into_iter()
+        .enumerate()
+        .filter_map(|(_index, candidate)| {
+            let plan = if metricsql {
+                PhysicalCompiler.compile_metricsql(candidate.clone(), environment.clone())
+            } else {
+                PhysicalCompiler.compile(candidate.clone(), environment.clone())
+            }
+            .ok()?;
+            let unit_cost = if preferred { 1.0 } else { 1e12 };
+            preferred = false;
+            let manifest = workload_cost::manifest(&plan, &candidate.queries).unwrap();
+            Some(WorkloadQuote {
+                unit_costs: manifest
+                    .components
+                    .keys()
+                    .map(|key| (key.clone(), unit_cost))
+                    .collect(),
+                manifest,
+                executable: true,
+            })
+        })
+        .collect();
+    snapshot.workload_cost_evidence = Some(WorkloadCostEvidence {
+        backend_revision: BACKEND_REVISION.into(),
+        planner_revision: PLANNER_REVISION.into(),
+        data_snapshot_id: "process-fixture".into(),
+        model_version: "test-only-unit-costs".into(),
+        observed_at_unix_ms: environment.observed_at_unix_ms,
+        valid_for_ms: environment.max_evidence_age_ms,
+        quotes,
+    });
+    snapshot
+}
+
 struct ChildGuard(Child);
 
 impl Drop for ChildGuard {
@@ -1143,15 +1197,18 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
 
     let backend_port = unused_port();
     let output_dir = tempfile::tempdir().expect("backend output directory");
-    let snapshot = concat!(
-        env!("CARGO_MANIFEST_DIR"),
-        "/../docs/examples/asapquery-compatibility-demo-snapshot.json"
-    );
+    let fixture = serde_json::from_str(include_str!(
+        "../../docs/examples/asapquery-compatibility-demo-snapshot.json"
+    ))
+    .unwrap();
+    let priced = quote_snapshot_for_test(fixture);
+    let snapshot = output_dir.path().join("snapshot.json");
+    std::fs::write(&snapshot, serde_json::to_vec(&priced).unwrap()).unwrap();
     let child = Command::new(env!("CARGO_BIN_EXE_data_plane"))
         .arg("--profile")
         .arg("asapquery")
         .arg("--planning-snapshot")
-        .arg(snapshot)
+        .arg(&snapshot)
         .arg("--prometheus-server")
         .arg(format!("http://{fallback_address}"))
         .arg("--forward-unsupported-queries")
@@ -1588,7 +1645,7 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
         .as_array()
         .expect("materialization statuses");
     let planned_snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot =
-        serde_json::from_str(&std::fs::read_to_string(snapshot).unwrap()).unwrap();
+        serde_json::from_str(&std::fs::read_to_string(&snapshot).unwrap()).unwrap();
     let planned = planned_snapshot.compile().unwrap();
     // Every selected state must be serving; the Planner may share or separate
     // physical populations, so compare identities rather than a frozen count.
diff --git a/data_plane/tests/support/distinct_planning_process.rs b/data_plane/tests/support/distinct_planning_process.rs
index 123624db..a67a197a 100644
--- a/data_plane/tests/support/distinct_planning_process.rs
+++ b/data_plane/tests/support/distinct_planning_process.rs
@@ -23,10 +23,11 @@ async fn distinct_range_uses_planner_selected_hll_and_source_labels() {
     entry["query"] = QUERY.into();
     entry["requirements"]["accuracy"] = serde_json::json!({"explicit": {"Epsilon": 0.05}});
     fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]);
-    let plan = serde_json::from_value::<BackendLocalPlanningSnapshot>(fixture.clone())
-        .unwrap()
-        .compile()
-        .unwrap();
+    let plan = quote_snapshot_for_test(
+        serde_json::from_value::<BackendLocalPlanningSnapshot>(fixture.clone()).unwrap(),
+    )
+    .compile()
+    .unwrap();
     assert_eq!(plan.precompute_plan.materializations.len(), 1);
     assert_eq!(
         plan.precompute_plan.materializations[0].aggregation_type,
@@ -38,7 +39,8 @@ async fn distinct_range_uses_planner_selected_hll_and_source_labels() {
     );
     let output = tempfile::tempdir().unwrap();
     let path = output.path().join("planning.json");
-    std::fs::write(&path, serde_json::to_vec(&fixture).unwrap()).unwrap();
+    let priced = quote_snapshot_for_test(serde_json::from_value(fixture.clone()).unwrap());
+    std::fs::write(&path, serde_json::to_vec(&priced).unwrap()).unwrap();
     let port = unused_port();
     let mut vm_port = unused_port();
     while vm_port == port {
@@ -80,11 +82,10 @@ async fn distinct_range_uses_planner_selected_hll_and_source_labels() {
     // Source syntax uses the shared parser fork; serving semantics and exact
     // routing belong to the MetricsQL adapter and its installed query entries.
     let snapshot = serde_json::from_value::<BackendLocalPlanningSnapshot>(fixture).unwrap();
-    let (mut request, mut environment) = snapshot.planning_request().unwrap();
-    request.hybrid_execution = false;
-    environment.plan_version = 2;
-    let compiled = control_plane::physical::compiler::PhysicalCompiler
-        .compile_metricsql(request, environment)
+    let mut snapshot = snapshot;
+    snapshot.environment.plan_version = 2;
+    let compiled = quote_snapshot_for_frontend_test(snapshot, true)
+        .compile_metricsql()
         .unwrap();
     let identity = serde_json::json!({"plan_id": compiled.envelope.plan_id, "plan_version": compiled.envelope.plan_version});
     let install = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest {
diff --git a/data_plane/tests/support/durable_summary_process.rs b/data_plane/tests/support/durable_summary_process.rs
index b37d39c8..626392ae 100644
--- a/data_plane/tests/support/durable_summary_process.rs
+++ b/data_plane/tests/support/durable_summary_process.rs
@@ -15,7 +15,7 @@ async fn persisted_summary_restarts_without_live_reregistration() {
         serde_json::json!([fixture["query_workload"]["repeating_queries"][2].clone()]);
     let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot =
         serde_json::from_value(fixture).unwrap();
-    let plan = snapshot.compile().unwrap();
+    let plan = quote_snapshot_for_test(snapshot).compile().unwrap();
     let install = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest {
         summary_catalog: plan.summary_catalog,
         collector_plans: plan.collector_plans,
diff --git a/data_plane/tests/support/erp_planning_process.rs b/data_plane/tests/support/erp_planning_process.rs
index aa0cf508..6e88beac 100644
--- a/data_plane/tests/support/erp_planning_process.rs
+++ b/data_plane/tests/support/erp_planning_process.rs
@@ -104,7 +104,7 @@ async fn observed_shape_selects_installed_parameters_and_executes_remote_write()
         ));
         let snapshot: BackendLocalPlanningSnapshot =
             serde_json::from_value(fixture.clone()).unwrap();
-        let plan = snapshot.compile().unwrap();
+        let plan = quote_snapshot_for_test(snapshot).compile().unwrap();
         assert_eq!(
             plan.precompute_plan.materializations.len(),
             1,
@@ -131,7 +131,8 @@ async fn observed_shape_selects_installed_parameters_and_executes_remote_write()
         );
         let output = tempfile::tempdir().unwrap();
         let path = output.path().join("planning.json");
-        std::fs::write(&path, serde_json::to_vec(&fixture).unwrap()).unwrap();
+        let priced = quote_snapshot_for_test(serde_json::from_value(fixture.clone()).unwrap());
+        std::fs::write(&path, serde_json::to_vec(&priced).unwrap()).unwrap();
         let port = unused_port();
         let mut child = ChildGuard(
             Command::new(env!("CARGO_BIN_EXE_data_plane"))
diff --git a/data_plane/tests/support/immutable_maintenance_process.rs b/data_plane/tests/support/immutable_maintenance_process.rs
index eb191c0a..f105cdfa 100644
--- a/data_plane/tests/support/immutable_maintenance_process.rs
+++ b/data_plane/tests/support/immutable_maintenance_process.rs
@@ -35,7 +35,7 @@ async fn run_maintenance_process(multi_source: bool, distinct_groups: bool) {
     fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]);
     let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot =
         serde_json::from_value(fixture.clone()).unwrap();
-    let plan = snapshot.compile().unwrap();
+    let plan = quote_snapshot_for_test(snapshot).compile().unwrap();
     assert_eq!(
         plan.precompute_plan.materializations.len(),
         if multi_source { 3 } else { 2 }
@@ -303,12 +303,15 @@ async fn run_maintenance_process(multi_source: bool, distinct_groups: bool) {
         // singleton-derived output while new source populations can arrive.
         let mut next_fixture = fixture.clone();
         next_fixture["environment"]["plan_version"] = 2.into();
-        let next = serde_json::from_value::<
-            control_plane::physical::compiler::BackendLocalPlanningSnapshot,
-        >(next_fixture)
-        .unwrap()
-        .compile()
-        .unwrap();
+        let next =
+            quote_snapshot_for_test(
+                serde_json::from_value::<
+                    control_plane::physical::compiler::BackendLocalPlanningSnapshot,
+                >(next_fixture)
+                .unwrap(),
+            )
+            .compile()
+            .unwrap();
         let next_install = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest {
             summary_catalog: next.summary_catalog,
             collector_plans: next.collector_plans,
diff --git a/data_plane/tests/support/univmon_erp_process.rs b/data_plane/tests/support/univmon_erp_process.rs
index 9c933b8c..7c496d4a 100644
--- a/data_plane/tests/support/univmon_erp_process.rs
+++ b/data_plane/tests/support/univmon_erp_process.rs
@@ -125,7 +125,7 @@ async fn measured_readout_evidence_selects_and_executes_univmon() {
         "runtime": {"allowed_algorithms": ["Hll", "Kll", "UnivMon"], "max_memory_bytes": null}
     });
     let snapshot: BackendLocalPlanningSnapshot = serde_json::from_value(fixture.clone()).unwrap();
-    let plan = snapshot.compile().unwrap();
+    let plan = quote_snapshot_for_test(snapshot).compile().unwrap();
     eprintln!(
         "UNIVMON_PLANNED {}",
         serde_json::json!({"query_plan": plan.query_plan, "materializations": plan.precompute_plan.materializations, "lifecycle_estimates": plan.lifecycle_estimates, "executable_dags": plan.precompute_plan.executable_dags, "observation": observation})
@@ -148,10 +148,11 @@ async fn measured_readout_evidence_selects_and_executes_univmon() {
             .unwrap()
             .remove("max_frequency_entropy_absolute_bits_error");
     }
-    let missing = serde_json::from_value::<BackendLocalPlanningSnapshot>(missing_entropy)
-        .unwrap()
-        .compile()
-        .unwrap();
+    let missing = quote_snapshot_for_test(
+        serde_json::from_value::<BackendLocalPlanningSnapshot>(missing_entropy).unwrap(),
+    )
+    .compile()
+    .unwrap();
     use control_plane::query_plan::{QueryPlanNode, QueryReadout};
     assert!(missing
         .query_plan
@@ -210,7 +211,8 @@ async fn measured_readout_evidence_selects_and_executes_univmon() {
     });
     let output = tempfile::tempdir().unwrap();
     let path = output.path().join("planning.json");
-    std::fs::write(&path, serde_json::to_vec(&fixture).unwrap()).unwrap();
+    let priced = quote_snapshot_for_test(serde_json::from_value(fixture.clone()).unwrap());
+    std::fs::write(&path, serde_json::to_vec(&priced).unwrap()).unwrap();
     let port = unused_port();
     let mut child = ChildGuard(
         Command::new(env!("CARGO_BIN_EXE_data_plane"))
@@ -326,7 +328,7 @@ async fn measured_readout_evidence_selects_and_executes_univmon() {
                 .unwrap()
                 .invalid_reason
                 .is_none());
-            let replanned = live_snapshot.compile().unwrap();
+            let replanned = quote_snapshot_for_test(live_snapshot).compile().unwrap();
             assert!(
                 replanned
                     .precompute_plan
diff --git a/docs/examples/asapquery-compatibility-demo-snapshot.json b/docs/examples/asapquery-compatibility-demo-snapshot.json
index e345114c..8641f93b 100644
--- a/docs/examples/asapquery-compatibility-demo-snapshot.json
+++ b/docs/examples/asapquery-compatibility-demo-snapshot.json
@@ -1,5 +1,5 @@
 {
-  "snapshot_version": 1,
+  "snapshot_version": 2,
   "query_workload": {
     "language": "promql",
     "query_batch": null,
diff --git a/docs/examples/asapquery-planning-snapshot.json b/docs/examples/asapquery-planning-snapshot.json
index 99c2a284..cd63e8bc 100644
--- a/docs/examples/asapquery-planning-snapshot.json
+++ b/docs/examples/asapquery-planning-snapshot.json
@@ -1,5 +1,5 @@
 {
-  "snapshot_version": 1,
+  "snapshot_version": 2,
   "query_workload": {
     "language": "promql",
     "query_batch": null,

From cff91375d330f78f4d11d79e3354e1d7fb3f4102 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Sat, 12 Sep 2026 11:22:18 -0600
Subject: [PATCH 5/6] Unify snapshot deployment CLI and migrate discovery
 examples

---
 README.md                                     | 31 ++++++------
 .../docs/candidate-physical-explain.md        |  2 +-
 .../examples/compile_workload_artifact.rs     | 23 ++++++---
 .../examples/inspect_physical_dag.rs          | 49 -------------------
 demos/asapquery/run.sh                        |  3 +-
 docs/evaluation/e2e-physical-dag.md           | 26 +++++-----
 docs/examples/workload-cost-evidence.md       |  8 +--
 7 files changed, 53 insertions(+), 89 deletions(-)
 delete mode 100644 control_plane/examples/inspect_physical_dag.rs

diff --git a/README.md b/README.md
index 57181620..102e1783 100644
--- a/README.md
+++ b/README.md
@@ -264,7 +264,7 @@ docker version
 cargo +1.98.0 fetch --locked
 cargo +1.98.0 build --locked -p control_plane -p data_plane
 cargo +1.98.0 build --locked -p control_plane \
-  --example calibration_candidates --example inspect_physical_dag \
+  --example calibration_candidates --example compile_workload_artifact \
   --example compile_clickhouse_workload
 target/debug/data_plane --help
 mkdir -p target/readme-evidence
@@ -278,7 +278,7 @@ mkdir -p target/readme-evidence
 The executable is `target/debug/data_plane`, or `target/release/data_plane` if
 you build with `--release`. Access to the pinned Git dependencies is required.
 
-### 3. Export candidates and inspect the ordinary selected plan
+### 3. Export candidates and select a deployable plan by complete cost
 
 ```bash
 target/debug/examples/calibration_candidates \
@@ -287,10 +287,10 @@ target/debug/examples/calibration_candidates \
 jq '.candidates[] | {candidate_index, unavailable_reason}' \
   target/readme-evidence/candidates.json
 
-target/debug/examples/inspect_physical_dag \
-  docs/examples/asapquery-compatibility-demo-snapshot.json \
+target/debug/examples/compile_workload_artifact \
+  "$ASAPQUERY_PLANNING_SNAPSHOT" \
   > target/readme-evidence/selected.json
-jq '.purpose' target/readme-evidence/selected.json
+jq '.cost_comparison' target/readme-evidence/selected.json
 jq '.install_request.summary_catalog' target/readme-evidence/selected.json
 jq '.install_request.precompute_plan | {materializations, executable_dags}' \
   target/readme-evidence/selected.json
@@ -299,13 +299,13 @@ jq '.install_request.precompute_plan.schemas[] | {materialization, schema_id}' \
   target/readme-evidence/selected.json
 ```
 
-Expect `inspection_only`, catalog/plan objects and explicit candidate rejection
-reasons where unsupported. One verified demo export contained five candidate
-entries (one installable), five selected materializations and six query entries;
-these are inspection evidence, not a permanent optimizer-count contract.
-Materialization IDs are definitions, not physical SIDs. Demo costs are not
-measurements. The [E2E walkthrough](docs/evaluation/e2e-physical-dag.md) explains
-ERP evidence and the version-2 measured-cost workflow.
+Candidate discovery accepts the checked-in unquoted templates. Deployment and
+selected-plan inspection require `ASAPQUERY_PLANNING_SNAPSHOT` to point to a
+snapshot with complete, valid workload cost evidence. Prepare that input using
+the [cost evidence workflow](docs/examples/workload-cost-evidence.md).
+There is one snapshot compiler: it compares complete executable alternatives,
+including exact fallback. Materialization IDs are definitions, not physical SIDs.
+For `--metricsql`, collect quotes for the MetricsQL frontend.
 
 ## Prometheus runbook
 
@@ -342,7 +342,7 @@ done
 curl -fsS http://127.0.0.1:19090/-/healthy
 
 target/debug/data_plane --profile asapquery \
-  --planning-snapshot docs/examples/asapquery-compatibility-demo-snapshot.json \
+  --planning-snapshot "$ASAPQUERY_PLANNING_SNAPSHOT" \
   --prometheus-server http://127.0.0.1:19090 \
   --forward-unsupported-queries --http-port 19091 \
   --output-dir target/readme-evidence/prometheus/runtime \
@@ -409,6 +409,7 @@ kill "$(cat target/readme-evidence/prometheus/backend.pid)"
 docker rm -f asap-readme-prometheus asap-readme-pushgateway
 cargo +1.98.0 test --locked -p data_plane --test asapquery_compatibility_process_e2e \
   collector_free_profile_serves_complete_matrix_and_falls_back_exactly -- --exact
+export ASAPQUERY_PLANNING_SNAPSHOT=/absolute/path/priced-snapshot.json
 ./scripts/e2e.sh asapquery-demo
 ```
 
@@ -467,8 +468,8 @@ kill "$(cat target/readme-evidence/victoriametrics/backend.pid)"
 To inspect supported MetricsQL planning independently:
 
 ```bash
-target/debug/examples/inspect_physical_dag \
-  docs/examples/asapquery-compatibility-demo-snapshot.json --metricsql \
+target/debug/examples/compile_workload_artifact \
+  "$ASAPQUERY_PLANNING_SNAPSHOT" --metricsql \
   > target/readme-evidence/victoriametrics/selected.json
 jq '.install_request.query_plan.entries' \
   target/readme-evidence/victoriametrics/selected.json
diff --git a/control_plane/docs/candidate-physical-explain.md b/control_plane/docs/candidate-physical-explain.md
index 993ce39f..fded1a0a 100644
--- a/control_plane/docs/candidate-physical-explain.md
+++ b/control_plane/docs/candidate-physical-explain.md
@@ -18,7 +18,7 @@ Physical identity combines the logical/mask alternative with existing materializ
 
 The read-only `/api/v1/physical-plan/cost-manifests` and MetricsQL equivalent retain their default manifest-array response. Add `"explain": true` to the existing request to receive `{ "manifests": [...], "alternatives": [...], "logical_selection": [...] }`. Failed alternatives remain alongside usable manifests. When none can bind or be completely priced, the error retains an `all_infeasible` report and every accumulated alternative rather than only a generic message.
 
-Snapshot compilation exposes the same logical trace on `PhysicalPlan`; `inspect_physical_dag` prints it outside the install request. Compile-and-publish returns the trace without adding it to executable wire DTOs. SQL's existing selection trace gains the same semantic candidate/root identities.
+Snapshot compilation exposes the same logical trace on `PhysicalPlan`; `compile_workload_artifact` prints it outside the install request. Compile-and-publish returns the trace without adding it to executable wire DTOs. SQL's existing selection trace gains the same semantic candidate/root identities.
 
 These are bounded explanations: they cover the actual Planner search and the existing physical materialization/exact inventory, not every possible placement or resource-constrained cluster assignment. Missing numeric measurements remain missing. The next provider integration must occur before logical commitment and reuse Planner's provider/resource contracts.
 
diff --git a/control_plane/examples/compile_workload_artifact.rs b/control_plane/examples/compile_workload_artifact.rs
index b9c3c348..fa6ceed5 100644
--- a/control_plane/examples/compile_workload_artifact.rs
+++ b/control_plane/examples/compile_workload_artifact.rs
@@ -3,17 +3,21 @@ use control_plane::physical::compiler::BackendLocalPlanningSnapshot;
 use serde_json::json;
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let path = std::env::args()
-        .nth(1)
+    let mut args = std::env::args().skip(1);
+    let path = args
+        .next()
         .ok_or("usage: compile_workload_artifact SNAPSHOT.json [--metricsql]")?;
-    let snapshot: BackendLocalPlanningSnapshot = serde_json::from_slice(&std::fs::read(path)?)?;
-    if snapshot.snapshot_version != 2 {
-        return Err(
-            "execution evaluation requires version 2 complete workload cost evidence".into(),
-        );
+    let metricsql = match args.next().as_deref() {
+        None => false,
+        Some("--metricsql") => true,
+        Some(_) => return Err("expected optional --metricsql".into()),
+    };
+    if args.next().is_some() {
+        return Err("unexpected arguments".into());
     }
+    let snapshot: BackendLocalPlanningSnapshot = serde_json::from_slice(&std::fs::read(path)?)?;
     let start = std::time::Instant::now();
-    let plan = if std::env::args().skip(2).any(|arg| arg == "--metricsql") {
+    let plan = if metricsql {
         snapshot.compile_metricsql()?
     } else {
         snapshot.compile()?
@@ -29,6 +33,9 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
             "planning_elapsed_ns": elapsed,
             "envelope": plan.envelope,
             "cost_comparison": comparison,
+            "logical_selection": plan.logical_selection,
+            "backend_revision": control_plane::physical::compiler::BACKEND_REVISION,
+            "planner_revision": control_plane::physical::compiler::PLANNER_REVISION,
             "lifecycle_estimates": plan.lifecycle_estimates,
             "install_request": {
                 "summary_catalog": plan.summary_catalog,
diff --git a/control_plane/examples/inspect_physical_dag.rs b/control_plane/examples/inspect_physical_dag.rs
deleted file mode 100644
index b9d812ff..00000000
--- a/control_plane/examples/inspect_physical_dag.rs
+++ /dev/null
@@ -1,49 +0,0 @@
-//! Inspect an ordinary planning snapshot without treating demo costs as measurements.
-use control_plane::physical::compiler::BackendLocalPlanningSnapshot;
-use serde_json::json;
-
-fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let mut args = std::env::args().skip(1);
-    let path = args
-        .next()
-        .ok_or("usage: inspect_physical_dag SNAPSHOT.json [--metricsql]")?;
-    let metricsql = match args.next().as_deref() {
-        None => false,
-        Some("--metricsql") => true,
-        Some(_) => return Err("expected optional --metricsql".into()),
-    };
-    if args.next().is_some() {
-        return Err("unexpected arguments".into());
-    }
-    let snapshot: BackendLocalPlanningSnapshot = serde_json::from_slice(&std::fs::read(path)?)?;
-    let snapshot_version = snapshot.snapshot_version;
-    let erp_supplied = snapshot.implementation.erp.is_some();
-    let plan = if metricsql {
-        snapshot.compile_metricsql()?
-    } else {
-        snapshot.compile()?
-    };
-    println!(
-        "{}",
-        serde_json::to_string_pretty(&json!({
-            "purpose": "inspection_only",
-            "snapshot_version": snapshot_version,
-            "erp_input_supplied": erp_supplied,
-            "backend_revision": control_plane::physical::compiler::BACKEND_REVISION,
-            "planner_revision": control_plane::physical::compiler::PLANNER_REVISION,
-            "cost_comparison": plan.cost_comparison,
-            "logical_selection": plan.logical_selection,
-            "lifecycle_estimates": plan.lifecycle_estimates,
-            "install_request": {
-                "summary_catalog": plan.summary_catalog,
-                "collector_plans": plan.collector_plans,
-                "precompute_plan": plan.precompute_plan,
-                "transmission_plan": plan.transmission_plan,
-                "query_plan": plan.query_plan,
-                "storage_routing": null,
-                "adaptation_evidence": []
-            }
-        }))?
-    );
-    Ok(())
-}
diff --git a/demos/asapquery/run.sh b/demos/asapquery/run.sh
index addad6b5..95546995 100755
--- a/demos/asapquery/run.sh
+++ b/demos/asapquery/run.sh
@@ -11,6 +11,7 @@ PROM_CONTAINER="${RUN_ID}-prometheus"
 PUSH_CONTAINER="${RUN_ID}-pushgateway"
 EVIDENCE_DIR="${ASAPQUERY_DEMO_EVIDENCE_DIR:-${REPO_DIR}/target/asapquery-demo-evidence}"
 BACKEND_PID=""
+: "${ASAPQUERY_PLANNING_SNAPSHOT:?Set ASAPQUERY_PLANNING_SNAPSHOT to a snapshot with complete workload cost evidence}"
 
 cleanup() {
     if [[ -n "${BACKEND_PID}" ]]; then
@@ -52,7 +53,7 @@ curl -fsS http://127.0.0.1:19090/-/healthy >/dev/null
 
 "${REPO_DIR}/target/debug/data_plane" \
     --profile asapquery \
-    --planning-snapshot "${REPO_DIR}/docs/examples/asapquery-compatibility-demo-snapshot.json" \
+    --planning-snapshot "${ASAPQUERY_PLANNING_SNAPSHOT}" \
     --prometheus-server http://127.0.0.1:19090 \
     --forward-unsupported-queries \
     --http-port 19091 \
diff --git a/docs/evaluation/e2e-physical-dag.md b/docs/evaluation/e2e-physical-dag.md
index c17d609f..1a3c2807 100644
--- a/docs/evaluation/e2e-physical-dag.md
+++ b/docs/evaluation/e2e-physical-dag.md
@@ -39,10 +39,9 @@ Planner owns query semantics, summary families, parameters and candidate
 selection. ERP can affect supported evidence-based choices, but supplying an
 artifact does not prove that it was eligible or used. Freshness, source/update
 semantics, parameters and accuracy constraints still apply. The checked-in demo
-is not an empirical-ERP benchmark. For measured cost selection use the version-2
-snapshot workflow in [execution calibration](../../tools/o11y-execution/CALIBRATION.md).
-`compile_workload_artifact` deliberately requires that version; do not bypass it
-by relabeling demo costs as measured evidence.
+is not an empirical-ERP benchmark. For deployment, use the priced snapshot workflow in [execution calibration](../../tools/o11y-execution/CALIBRATION.md).
+`compile_workload_artifact` requires complete workload quotes; do not relabel
+demo costs as measured evidence.
 
 For the existing observation → ERP-selected KLL → installed HTTP correctness
 fixture, see [ERP process validation](../developer_docs/erp-process-validation.md):
@@ -62,8 +61,8 @@ Run the normal snapshot compiler, without selecting a candidate index or
 substituting a preferred sketch family:
 
 ```bash
-cargo run --locked -p control_plane --example inspect_physical_dag -- \
-  docs/examples/asapquery-compatibility-demo-snapshot.json \
+cargo run --locked -p control_plane --example compile_workload_artifact -- \
+  "$ASAPQUERY_PLANNING_SNAPSHOT" \
   > target/physical-dag-inspection/selected.json
 jq '.install_request' target/physical-dag-inspection/selected.json \
   > target/physical-dag-inspection/physical-plan.json
@@ -71,10 +70,12 @@ jq '.install_request | {summary_catalog, precompute_plan, query_plan}' \
   target/physical-dag-inspection/selected.json
 ```
 
-`inspect_physical_dag` calls the same `BackendLocalPlanningSnapshot::compile`
-entry point used by startup. It accepts the checked-in demonstration snapshot
-and labels its output `inspection_only`. `erp_input_supplied` reports only input
-presence. A null `cost_comparison` must not be interpreted as a measured win.
+`compile_workload_artifact` calls the same evidence-required snapshot compiler
+used by startup. Set `ASAPQUERY_PLANNING_SNAPSHOT` to a priced snapshot prepared
+using the [cost evidence workflow](../examples/workload-cost-evidence.md).
+The checked-in unquoted templates support candidate discovery only. The output
+includes the selected plan, logical selection trace and complete cost comparison.
+MetricsQL compilation requires quotes collected for that frontend.
 
 The compiler derives sibling plans from the selected post-ASAP DAG:
 
@@ -95,6 +96,7 @@ participate in identity, cost and coverage checks.
 For the simplest live run:
 
 ```bash
+export ASAPQUERY_PLANNING_SNAPSHOT=/absolute/path/priced-snapshot.json
 ./scripts/e2e.sh asapquery-demo
 ```
 
@@ -116,7 +118,7 @@ Then keep the backend running for inspection:
 ```bash
 cargo run --locked -p data_plane -- \
   --profile asapquery \
-  --planning-snapshot docs/examples/asapquery-compatibility-demo-snapshot.json \
+  --planning-snapshot "$ASAPQUERY_PLANNING_SNAPSHOT" \
   --prometheus-server http://127.0.0.1:9090 \
   --forward-unsupported-queries --http-port 9091 \
   --output-dir target/physical-dag-inspection/runtime
@@ -242,7 +244,7 @@ when comparing, `comparison.json`. A finished run is not proof of benefit.
 | Backend | Drop-in surface and current boundary |
 | --- | --- |
 | Prometheus | Remote Write v1 samples plus PromQL instant/range HTTP; keep Prometheus as the exact fallback and raw-data authority. The strict `asapquery` profile has explicit startup exclusions. |
-| VictoriaMetrics | The broader backend has a MetricsQL compile/adapter path. `inspect_physical_dag SNAPSHOT.json --metricsql` uses it. This does not turn the strict Prometheus profile into a general VM replacement; counter boundary semantics and unsupported expressions must retain exact routing. |
+| VictoriaMetrics | The broader backend has a MetricsQL compile/adapter path. `compile_workload_artifact SNAPSHOT.json --metricsql` uses it. This does not turn the strict Prometheus profile into a general VM replacement; counter boundary semantics and unsupported expressions must retain exact routing. |
 | ClickHouse | The broader backend has a SQL workload compiler and typed exact/relational execution. `compile_clickhouse_workload` reads its own `ClickHouseSqlAutomaticWorkload` JSON from stdin; it does not consume the Prometheus snapshot. List/Map/Tuple support does not imply arbitrary lambdas/counter SQL or full protocol compatibility. External-only DAG coverage is not summary acceleration. |
 
 Use the existing process suites to validate a specific supported protocol shape;
diff --git a/docs/examples/workload-cost-evidence.md b/docs/examples/workload-cost-evidence.md
index 612f58e7..64e60498 100644
--- a/docs/examples/workload-cost-evidence.md
+++ b/docs/examples/workload-cost-evidence.md
@@ -1,8 +1,10 @@
 # Complete workload cost evidence
 
-The migrated, evidence-required startup profile uses `snapshot_version: 2`.
-Version 1 and live requests without `workload_cost_evidence` remain compatibility
-paths: their lifecycle estimates are not complete workload costs.
+Planning snapshots use one schema, `snapshot_version: 2`. Version 1 is rejected.
+Candidate discovery may omit `workload_cost_evidence`; compiling a deployable
+snapshot requires complete, valid quotes and selects by complete workload cost.
+There is no unquoted snapshot deployment path. The checked-in JSON examples are
+discovery templates, not ready-to-deploy plans.
 
 ## Workflow
 

From be1aa1d45ce3fd1eb4a8ab2bea185fb3c4e74fb7 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Sat, 12 Sep 2026 11:24:55 -0600
Subject: [PATCH 6/6] fix(planning): preserve derived cohorts and share
 compatible sum panes

---
 Cargo.lock                                    |  10 +-
 control_plane/Cargo.toml                      |   8 +-
 control_plane/src/main.rs                     |   1 +
 control_plane/src/physical/compiler.rs        | 314 +++++++++++++++---
 control_plane/src/physical/mod.rs             |   1 +
 control_plane/src/physical/pane_reuse.rs      | 187 +++++++++++
 crates/asap_types/Cargo.toml                  |   2 +-
 data_plane/Cargo.toml                         |   6 +-
 .../asap_query_engine/post_asap_readout.rs    |  89 +++++
 .../planning/shared-window-panes.md           |  47 +++
 tools/test_shared_panes.py                    |  54 +++
 11 files changed, 658 insertions(+), 61 deletions(-)
 create mode 100644 control_plane/src/physical/pane_reuse.rs
 create mode 100644 docs/developer_docs/planning/shared-window-panes.md
 create mode 100644 tools/test_shared_panes.py

diff --git a/Cargo.lock b/Cargo.lock
index 96ae6df2..7912b94a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -373,7 +373,7 @@ dependencies = [
 [[package]]
 name = "asap-aware-mapping"
 version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=3be523fa0f06a905188e42cbe482d06aa843ba5d#3be523fa0f06a905188e42cbe482d06aa843ba5d"
+source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ca7546de792d74aee8231e9a1100ca893d9e86d3#ca7546de792d74aee8231e9a1100ca893d9e86d3"
 dependencies = [
  "asap-types",
  "serde",
@@ -384,7 +384,7 @@ dependencies = [
 [[package]]
 name = "asap-frontend-promql"
 version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=3be523fa0f06a905188e42cbe482d06aa843ba5d#3be523fa0f06a905188e42cbe482d06aa843ba5d"
+source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ca7546de792d74aee8231e9a1100ca893d9e86d3#ca7546de792d74aee8231e9a1100ca893d9e86d3"
 dependencies = [
  "asap-types",
  "promql-parser",
@@ -393,7 +393,7 @@ dependencies = [
 [[package]]
 name = "asap-frontend-sql"
 version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=3be523fa0f06a905188e42cbe482d06aa843ba5d#3be523fa0f06a905188e42cbe482d06aa843ba5d"
+source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ca7546de792d74aee8231e9a1100ca893d9e86d3#ca7546de792d74aee8231e9a1100ca893d9e86d3"
 dependencies = [
  "asap-sql-function-catalog",
  "asap-types",
@@ -415,12 +415,12 @@ dependencies = [
 [[package]]
 name = "asap-sql-function-catalog"
 version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=3be523fa0f06a905188e42cbe482d06aa843ba5d#3be523fa0f06a905188e42cbe482d06aa843ba5d"
+source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ca7546de792d74aee8231e9a1100ca893d9e86d3#ca7546de792d74aee8231e9a1100ca893d9e86d3"
 
 [[package]]
 name = "asap-types"
 version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=3be523fa0f06a905188e42cbe482d06aa843ba5d#3be523fa0f06a905188e42cbe482d06aa843ba5d"
+source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ca7546de792d74aee8231e9a1100ca893d9e86d3#ca7546de792d74aee8231e9a1100ca893d9e86d3"
 dependencies = [
  "serde",
  "serde_json",
diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml
index 85d7d33d..6773b0b3 100644
--- a/control_plane/Cargo.toml
+++ b/control_plane/Cargo.toml
@@ -77,8 +77,8 @@ asap_types.workspace = true
 # scaffolding, unaware that `data_plane`'s `summary_executor.rs` in *this*
 # repo is a real one. Vendored locally instead of chased upstream -- see
 # `data_plane/src/query_engines/asap_query_engine/summary_exec.rs`.
-planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "3be523fa0f06a905188e42cbe482d06aa843ba5d" }
-asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "3be523fa0f06a905188e42cbe482d06aa843ba5d" }
+planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ca7546de792d74aee8231e9a1100ca893d9e86d3" }
+asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ca7546de792d74aee8231e9a1100ca893d9e86d3" }
 
 # L1 adoption (design-target-architecture.md Part B): the PromQL front
 # end itself, replacing control_plane's own query_parser/promql.rs.
@@ -86,8 +86,8 @@ asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev =
 # `planner-types`/`asap-aware-mapping` above -- these three MUST move
 # together (two revs of the same upstream repo's types in one workspace
 # resolve to distinct Rust types that won't unify).
-asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "3be523fa0f06a905188e42cbe482d06aa843ba5d" }
-asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "3be523fa0f06a905188e42cbe482d06aa843ba5d" }
+asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ca7546de792d74aee8231e9a1100ca893d9e86d3" }
+asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ca7546de792d74aee8231e9a1100ca893d9e86d3" }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }
diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs
index 0beeb2ed..1cf1580c 100644
--- a/control_plane/src/main.rs
+++ b/control_plane/src/main.rs
@@ -1006,6 +1006,7 @@ fn compile_physical_plan_request(
     };
 
     let planning_request = physical::compiler::PlanningRequest {
+        synthesized_window_queries: Default::default(),
         logical_selection,
         query_workload: None,
         queries,
diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index 0783418c..537063d5 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -148,6 +148,8 @@ 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>,
+    /// Compiler-owned quotes eligible for joint pane repricing; never inferred from model labels.
+    pub synthesized_window_queries: BTreeSet<String>,
     pub evidence: HashMap<String, TopKMembershipEvidence>,
     /// Fresh measured costs for Planner exact/summary composition sites,
     /// scoped to query IDs just like accuracy evidence.
@@ -678,6 +680,67 @@ impl BackendLocalPlanningSnapshot {
             &exact_costs_by_id,
             self.implementation.erp.as_ref(),
         )?;
+        // Derived maintenance currently consumes full, non-overlapping source cohorts.
+        // Restrict only synthesized candidates; deployment-supplied evidence is authoritative.
+        for query in &mut queries {
+            if self
+                .implementation
+                .window_candidates
+                .contains_key(&query.query_string)
+            {
+                continue;
+            }
+            let states =
+                collect_selected_materializations(&query.post_asap, true).map_err(|reason| {
+                    CompileError::Query {
+                        query_id: query.query_id.clone(),
+                        reason,
+                    }
+                })?;
+            let mut full_windows = BTreeSet::new();
+            for state in &states {
+                if let Some(sources) = immutable_materialization_sources(&state.node) {
+                    full_windows.insert(state.window_secs.unwrap_or(query.window_secs));
+                    for source in sources {
+                        let (_, window, _) =
+                            selected_input_contract(&source).map_err(|reason| {
+                                CompileError::Query {
+                                    query_id: query.query_id.clone(),
+                                    reason,
+                                }
+                            })?;
+                        full_windows.insert(window.unwrap_or(query.window_secs));
+                    }
+                }
+            }
+            let mut seen = BTreeSet::new();
+            query.window_implementations.retain_mut(|candidate| {
+                if !full_windows.contains(&candidate.window_secs) {
+                    return true;
+                }
+                if !seen.insert(candidate.window_secs) {
+                    return false;
+                }
+                candidate.slide_secs = candidate.window_secs;
+                candidate.framework = SummaryWindowFramework::Tumbling;
+                candidate.layout = asap_types::WindowMaterializationLayout::Pane {
+                    pane_secs: candidate.window_secs,
+                };
+                candidate.implementation_id = format!(
+                    "{}-{}s-derived-cohort",
+                    self.implementation.window_implementation_id, candidate.window_secs
+                );
+                candidate.cost = derived_window_cost(
+                    &candidate.cost,
+                    &query.lifecycle,
+                    candidate.window_secs,
+                    candidate.slide_secs,
+                    &candidate.layout,
+                    self.implementation.query_staleness_margin_ms,
+                );
+                true
+            });
+        }
         // Composable lowering residualizes unsafe leaves individually; retain Planner siblings.
         Ok((
             PlanningRequest {
@@ -685,6 +748,16 @@ impl BackendLocalPlanningSnapshot {
                 hybrid_execution: true,
                 materialization_policy: None,
                 query_workload: Some(workload),
+                synthesized_window_queries: queries
+                    .iter()
+                    .filter(|q| {
+                        !self
+                            .implementation
+                            .window_candidates
+                            .contains_key(&q.query_string)
+                    })
+                    .map(|q| q.query_id.clone())
+                    .collect(),
                 queries,
                 evidence: topk_evidence_by_id,
                 exact_composition_costs: exact_costs_by_id,
@@ -1421,6 +1494,18 @@ impl PhysicalCompiler {
             }
         }
 
+        if environment.target == PhysicalDeploymentTarget::BackendLocalRemoteWrite {
+            super::pane_reuse::share_additive_panes(
+                &request,
+                &mut compiled_materializations,
+                &mut collector_materializations,
+                &mut plan_materializations,
+                &mut node_bindings,
+                &mut runtime_policies,
+                &mut lifecycle_estimates,
+            );
+        }
+
         let plan_id = if request.hybrid_execution {
             use std::hash::{Hash, Hasher};
             let mut hash = std::collections::hash_map::DefaultHasher::new();
@@ -2214,8 +2299,8 @@ fn validate_lifecycle_input(
 ///   query's own evaluation cadence.
 /// - **retention**: `retained_state_count`, the same function that fills
 ///   `num_aggregates_to_retain`, so the quote and the plan cannot disagree.
-///   Concurrent in-flight full windows are not double-charged here — the
-///   fanout term already prices that write amplification.
+///   Open worker accumulators are charged separately from published states;
+///   updating a state and keeping it resident are different resources.
 ///
 /// The byte fields are left exactly as the snapshot supplied them: state size
 /// needs sketch parameters that do not exist yet at this point, and guessing
@@ -2223,7 +2308,7 @@ fn validate_lifecycle_input(
 /// `cpu_cost` and `weighted_cost` are derived, and only those two are read —
 /// by `validate_window_implementations` and by the `min_by` that ranks
 /// candidates. `model_version` records that the quote is derived.
-fn derived_window_cost(
+pub(super) fn derived_window_cost(
     template: &ImplementationCostEvidence,
     lifecycle: &LifecyclePlanningInput,
     window_secs: u64,
@@ -2258,23 +2343,29 @@ fn derived_window_cost(
         slide_secs.saturating_mul(1_000),
         layout,
     ) as f64;
+    // Store retention does not include worker accumulators that are still open.
+    let active = match layout {
+        asap_types::WindowMaterializationLayout::FullWindow => (window / slide).ceil(),
+        _ => 1.0,
+    };
 
     let build = costs.build * states_created;
     let maintenance = costs.maintenance_per_update * updates;
     let read = costs.read * reads * finalizations_per_read;
-    let retention = costs.retention_per_second * horizon * retained;
+    let retention = costs.retention_per_second * horizon * (retained + active);
     let retirement = costs.retirement * states_created;
     let cpu_cost = build + maintenance;
     let weighted_cost = cpu_cost + read + retention + retirement;
 
     ImplementationCostEvidence {
-        model_version: format!("{}+derived-window-layout-v1", template.model_version),
-        cpu_cost: if cpu_cost.is_finite() { cpu_cost } else { 0.0 },
-        weighted_cost: if weighted_cost.is_finite() {
-            weighted_cost
-        } else {
-            0.0
-        },
+        model_version: format!(
+            "{}+derived-window-layout-v1",
+            template
+                .model_version
+                .trim_end_matches("+derived-window-layout-v1")
+        ),
+        cpu_cost,
+        weighted_cost,
         ..template.clone()
     }
 }
@@ -2326,42 +2417,10 @@ fn range_selector_windows_secs(expr: &QueryExpr) -> BTreeSet<u64> {
     windows
 }
 
-/// The window implementations to plan with when the snapshot priced none for
-/// this query.
-///
-/// These are *shapes*, not cost quotes. `ImplementationCostEvidence` is
-/// measured evidence: its `weighted_cost` doc puts pricing update CPU,
-/// query-time merges, retained memory, storage, scans and network on the
-/// evidence producer. So this never synthesizes competing candidates for one
-/// window to rank against each other — one shape per window, selected by a
-/// `min_by` over a one-element list where the cost cannot change the outcome.
-/// Ranking `Pane` against `FullWindow` requires a snapshot supplying both with
-/// their own priced evidence in `window_candidates`.
-///
-/// **One candidate per range-selector window, not one per query.** A state
-/// whose window has no candidate is dropped from selection outright
-/// (`hybrid_execution`'s filter), and the surviving candidates are narrowed to
-/// the selected window before validation. Deriving a single candidate from
-/// `time_selection.lookback` therefore silently costs every operand whose own
-/// range differs from it its summary: `sum(sum_over_time(a[1m])) /
-/// sum(sum_over_time(b[5m]))` under a 1m lookback kept `a` and dropped `b` to
-/// exact execution, with nothing reported.
-///
-/// Within one window, the shape follows the query's evaluation cadence. A
-/// workload evaluated every 30s over a 5m window needs its state to advance
-/// every 30s; one 5m tumbling window answers with results that only change
-/// once every five minutes. `evaluation_interval_ms` already reaches this
-/// function — it was read for lifecycle costing and then dropped here.
-///
-/// The guards are the validator's own rules, so a bad shape is a compile error
-/// rather than a silent plan: `WindowMaterializationLayout::validate` requires
-/// the pane to divide both window and slide (45s into 300s has no such pane),
-/// and the framework/layout table admits `Tumbling + Pane` and `Sliding + Pane`.
-/// `pane_secs == slide_secs` is the coarsest legal pane for a cadence, so it
-/// is the one with the fewest query-time merges. `FullWindow` is the other
-/// legal `Sliding` layout and is deliberately not emitted alongside it:
-/// preferring it is a write-amplification-versus-read-amplification tradeoff,
-/// which is a cost comparison, and there is no second quote to compare.
+/// Derive and price one implementation per supported layout for each range.
+/// Explicit snapshot candidates bypass this path. After logical selection,
+/// derived maintenance cohorts are restricted to their supported full windows;
+/// raw additive pane producers may subsequently be shared by Planner.
 fn derived_window_candidates(
     implementation_id: &str,
     expr: &QueryExpr,
@@ -2524,7 +2583,7 @@ pub(super) fn validate_window_implementations(
     Ok(candidates)
 }
 
-fn retained_state_count(
+pub(super) fn retained_state_count(
     lookback_ms: u64,
     staleness_margin_ms: u64,
     slide_ms: u64,
@@ -3987,6 +4046,7 @@ mod tests {
         }
         Ok(PlanningRequest {
             logical_selection: Vec::new(),
+            synthesized_window_queries: BTreeSet::new(),
             hybrid_execution: false,
             materialization_policy: None,
             query_workload: None,
@@ -5154,6 +5214,164 @@ mod tests {
         assert_eq!(plan.precompute_plan.materializations.len(), 2);
     }
 
+    // Derived programs and their raw inputs must keep a runtime-supported cohort.
+    #[test]
+    fn derived_window_regression_nested_snapshot() {
+        for rate in [0.0, 100.0] {
+            let mut value: Value = serde_json::from_str(include_str!(
+                "../../../docs/examples/asapquery-planning-snapshot.json"
+            ))
+            .unwrap();
+            value["query_workload"]["repeating_queries"][0]["query"] =
+                json!("quantile(0.9, sum_over_time(m[1m]))");
+            value["data_workload"]["ingestion_rate"]["value"] = json!(rate);
+            value["query_workload"]["data_workload"]["ingestion_rate"]["value"] = json!(rate);
+            let snapshot: BackendLocalPlanningSnapshot = serde_json::from_value(value).unwrap();
+            let (request, env) = snapshot.planning_request().unwrap();
+            let plan = PhysicalCompiler.compile(request, env).unwrap();
+            assert!(plan
+                .precompute_plan
+                .materializations
+                .iter()
+                .any(|m| m.derived_input.is_some()));
+            assert!(plan
+                .precompute_plan
+                .materializations
+                .iter()
+                .all(|m| m.window_size == m.slide_interval));
+        }
+    }
+
+    // A full-window producer keeps overlapping accumulators alive even before publication.
+    #[test]
+    fn derived_window_regression_resident_cost() {
+        let template = planning_snapshot().implementation.implementation_cost;
+        let mut lifecycle = planning_lifecycle();
+        lifecycle.costs = LifecycleCostEvidence {
+            build: 0.0,
+            maintenance_per_update: 0.0,
+            read: 0.0,
+            retention_per_second: 1.0,
+            retirement: 0.0,
+        };
+        let full = derived_window_cost(
+            &template,
+            &lifecycle,
+            300,
+            30,
+            &asap_types::WindowMaterializationLayout::FullWindow,
+            0,
+        );
+        assert!(full.weighted_cost >= lifecycle.horizon_seconds * 11.0);
+    }
+
+    // Temporal SUM readouts share raw state only for identical source populations.
+    #[test]
+    fn derived_window_regression_shared_sum_panes() {
+        for interval in [10_000, 60_000] {
+            for (rhs, expected_states) in [("a", 1), ("b", 2), ("a{job=\"x\"}", 2)] {
+                let mut snapshot = planning_snapshot();
+                let query = format!("sum_over_time(a[1m]) / sum_over_time({rhs}[10m])");
+                let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0];
+                entry.query = Query(query);
+                entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact);
+                entry.demand = RepeatedDemand::FixedIntervalAt {
+                    interval: RepetitionInterval(interval),
+                    evaluation_phase: planner_types::workload::TimestampMs(0),
+                };
+                let (request, env) = snapshot.planning_request().unwrap();
+                let plan = PhysicalCompiler.compile(request, env).unwrap();
+                assert_eq!(plan.precompute_plan.materializations.len(), expected_states);
+                let entry = plan.query_plan.entries.values().next().unwrap();
+                let bindings = entry.materialization_bindings();
+                assert_eq!(
+                    bindings
+                        .iter()
+                        .filter_map(|b| b.readout_lookback_ms)
+                        .collect::<BTreeSet<_>>(),
+                    BTreeSet::from([60_000, 600_000])
+                );
+                if rhs == "a" {
+                    assert_eq!(bindings[0].materialization, bindings[1].materialization);
+                    assert_eq!(
+                        plan.precompute_plan.materializations[0].num_aggregates_to_retain,
+                        Some(600_000 / u64::from(interval) + 1)
+                    );
+                }
+            }
+        }
+    }
+
+    // Distinct workload entries share one producer and retain both lifecycle consumers.
+    #[test]
+    fn shared_panes_preserve_workload_consumers_and_phase() {
+        for phase in [0, 5_000] {
+            let mut snapshot = planning_snapshot();
+            let entries = snapshot.query_workload.repeating_queries.as_mut().unwrap();
+            entries[0].query = Query("sum_over_time(a[1m])".into());
+            entries[0].requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact);
+            let mut second = entries[0].clone();
+            second.query = Query("sum_over_time(a[10m])".into());
+            second.demand = RepeatedDemand::FixedIntervalAt {
+                interval: RepetitionInterval(10_000),
+                evaluation_phase: planner_types::workload::TimestampMs(phase),
+            };
+            entries.push(second);
+            let (request, env) = snapshot.planning_request().unwrap();
+            let plan = PhysicalCompiler.compile(request, env).unwrap();
+            assert_eq!(
+                plan.precompute_plan.materializations.len(),
+                if phase == 0 { 1 } else { 2 }
+            );
+            if phase == 0 {
+                assert_eq!(plan.lifecycle_estimates.len(), 1);
+                let estimate = &plan.lifecycle_estimates[0];
+                assert_eq!(estimate.consumer_query_ids.len(), 2);
+                assert_eq!(estimate.expected_reads, 60.0);
+                assert_eq!(estimate.expected_updates, 30_000.0);
+            }
+        }
+    }
+
+    // Overflow must fail candidate validation, never turn an expensive layout into a free one.
+    #[test]
+    fn derived_cost_overflow_is_rejected() {
+        let snapshot = planning_snapshot();
+        let (mut request, env) = snapshot.planning_request().unwrap();
+        let query = &mut request.queries[0];
+        let mut lifecycle = query.lifecycle.clone();
+        lifecycle.costs.build = f64::MAX;
+        let candidate = &mut query.window_implementations[0];
+        candidate.cost = derived_window_cost(
+            &candidate.cost,
+            &lifecycle,
+            candidate.window_secs,
+            candidate.slide_secs,
+            &candidate.layout,
+            0,
+        );
+        assert!(validate_window_implementations(query, &env).is_err());
+    }
+
+    // Serialized derived quotes become authoritative when a deployment supplies them explicitly.
+    #[test]
+    fn explicit_window_quotes_are_not_repriced_for_sharing() {
+        let mut snapshot = planning_snapshot();
+        let query = "sum_over_time(a[1m]) / sum_over_time(a[10m])";
+        let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0];
+        entry.query = Query(query.into());
+        entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact);
+        let (derived, _) = snapshot.clone().planning_request().unwrap();
+        snapshot.implementation.window_candidates.insert(
+            query.into(),
+            derived.queries[0].window_implementations.clone(),
+        );
+        let (request, env) = snapshot.planning_request().unwrap();
+        assert!(request.synthesized_window_queries.is_empty());
+        let plan = PhysicalCompiler.compile(request, env).unwrap();
+        assert_eq!(plan.precompute_plan.materializations.len(), 2);
+    }
+
     fn planning_lifecycle() -> LifecyclePlanningInput {
         planning_snapshot().planning_request().unwrap().0.queries[0]
             .lifecycle
diff --git a/control_plane/src/physical/mod.rs b/control_plane/src/physical/mod.rs
index 896594da..f26665fb 100644
--- a/control_plane/src/physical/mod.rs
+++ b/control_plane/src/physical/mod.rs
@@ -27,6 +27,7 @@ pub mod deployment;
 pub mod deployment_cost;
 pub mod erp;
 pub mod executable_binding;
+mod pane_reuse;
 pub mod plan;
 pub mod plan_cache;
 pub mod planner;
diff --git a/control_plane/src/physical/pane_reuse.rs b/control_plane/src/physical/pane_reuse.rs
new file mode 100644
index 00000000..ae34cb15
--- /dev/null
+++ b/control_plane/src/physical/pane_reuse.rs
@@ -0,0 +1,187 @@
+use super::compiler::{
+    derived_window_cost, retained_state_count, CollectorMaterialization,
+    MaterializationLifecycleEstimate, PlanningRequest, RuntimeRulePolicy,
+};
+use planner_types::post_asap::{PostAsapNodeId, SummaryWindowFramework};
+use std::collections::{BTreeMap, BTreeSet};
+
+/// Lower Planner's costed pane-reuse groups without changing logical readouts.
+/// Only raw additive states are eligible; derived cohorts retain their full-window identity.
+#[allow(clippy::too_many_arguments)]
+pub(super) fn share_additive_panes(
+    request: &PlanningRequest,
+    materializations: &mut [asap_types::PrecomputeMaterialization],
+    producers: &mut Vec<CollectorMaterialization>,
+    plan_producers: &mut [CollectorMaterialization],
+    bindings: &mut BTreeMap<(usize, PostAsapNodeId), asap_types::PolicyFingerprint>,
+    policies: &mut BTreeMap<asap_types::PolicyFingerprint, RuntimeRulePolicy>,
+    estimates: &mut BTreeMap<asap_types::PolicyFingerprint, MaterializationLifecycleEstimate>,
+) {
+    use asap_aware_mapping::pane_sharing::{select_shared_panes, PaneReuseCandidate};
+    use asap_types::{AggregationType, WindowKind, WindowMaterializationLayout};
+    let derived_sources = materializations
+        .iter()
+        .filter_map(|m| m.derived_input.as_ref())
+        .flat_map(|d| d.inputs.iter().map(|id| id.fingerprint()))
+        .collect::<BTreeSet<_>>();
+    let mut seen = BTreeSet::new();
+    let mut physical = Vec::new();
+    let mut offers = Vec::new();
+    for m in materializations.iter() {
+        let old = m.policy_fingerprint();
+        if !seen.insert(old)
+            || m.derived_input.is_some()
+            || derived_sources.contains(&old)
+            || !matches!(
+                m.aggregation_type,
+                AggregationType::Sum | AggregationType::MultipleSum
+            )
+        {
+            continue;
+        }
+        let WindowMaterializationLayout::Pane { pane_secs } = m.window_layout else {
+            continue;
+        };
+        if pane_secs == 0 || m.pane_origin_ms.is_none() {
+            continue;
+        }
+        let consumers = bindings
+            .iter()
+            .filter(|(_, id)| **id == old)
+            .map(|((query, _), _)| *query)
+            .collect::<BTreeSet<_>>();
+        let Some(&first) = consumers.first() else {
+            continue;
+        };
+        let query = &request.queries[first];
+        // Explicitly priced implementations are not repriced or replaced.
+        if consumers.iter().any(|index| {
+            let q = &request.queries[*index];
+            q.lifecycle != query.lifecycle
+                || q.accuracy != query.accuracy
+                || !request.synthesized_window_queries.contains(&q.query_id)
+        }) {
+            continue;
+        }
+        let mut canonical = m.clone();
+        canonical.window_size = pane_secs;
+        canonical.slide_interval = pane_secs;
+        canonical.window_type = WindowKind::Tumbling;
+        let Some(policy) = policies.get(&old) else {
+            continue;
+        };
+        let key = (
+            canonical.policy_fingerprint(),
+            serde_json::to_string(&query.lifecycle).unwrap(),
+            serde_json::to_string(policy).unwrap(),
+        );
+        let mut maintenance = query.lifecycle.clone();
+        maintenance.costs.read = 0.0;
+        let Some(template) = query.window_implementations.first() else {
+            continue;
+        };
+        let producer_cost = derived_window_cost(
+            &template.cost,
+            &maintenance,
+            m.window_size,
+            m.slide_interval,
+            &m.window_layout,
+            request.query_staleness_margin_ms,
+        )
+        .weighted_cost;
+        let read_cost = query.lifecycle.costs.read * query.lifecycle.horizon_seconds
+            / (f64::from(query.lifecycle.evaluation_interval_ms) / 1000.0)
+            * (m.window_size / pane_secs) as f64
+            * consumers.len() as f64;
+        offers.push(PaneReuseCandidate {
+            compatibility: key,
+            lookback_ms: m.window_size.saturating_mul(1000),
+            producer_cost,
+            read_cost,
+        });
+        physical.push((old, canonical));
+    }
+    let groups = select_shared_panes(&offers);
+    let mut target_counts = BTreeMap::new();
+    for group in &groups {
+        *target_counts
+            .entry(physical[group.members[0]].1.policy_fingerprint())
+            .or_insert(0) += 1;
+    }
+    let mut replacements = BTreeMap::new();
+    for group in groups {
+        let mut canonical = physical[group.members[0]].1.clone();
+        canonical.num_aggregates_to_retain = Some(retained_state_count(
+            group.lookback_ms,
+            request.query_staleness_margin_ms,
+            canonical.slide_interval * 1000,
+            &canonical.window_layout,
+        ));
+        let new = canonical.policy_fingerprint();
+        let members = group
+            .members
+            .iter()
+            .map(|index| physical[*index].0)
+            .collect::<BTreeSet<_>>();
+        // A physical ID cannot hide a second policy/evidence cohort or an
+        // independently installed producer that was not offered for sharing.
+        if target_counts[&new] != 1
+            || materializations.iter().any(|m| {
+                let id = m.policy_fingerprint();
+                id == new && !members.contains(&id)
+            })
+        {
+            continue;
+        }
+        let mut combined = estimates[&physical[group.members[0]].0].clone();
+        combined.materialization = new.into();
+        combined.consumer_query_ids.clear();
+        combined.expected_reads = 0.0;
+        combined.expected_updates = 0.0;
+        combined.lifecycle_cost = group.cost;
+        combined.window_implementation_id = format!("shared-pane-{}", new.0);
+        for index in group.members {
+            let old = physical[index].0;
+            if let Some(estimate) = estimates.remove(&old) {
+                combined
+                    .consumer_query_ids
+                    .extend(estimate.consumer_query_ids);
+                combined.expected_reads += estimate.expected_reads;
+                combined.expected_updates =
+                    combined.expected_updates.max(estimate.expected_updates);
+            }
+            replacements.insert(old, canonical.clone());
+        }
+        combined.consumer_query_ids.sort();
+        combined.consumer_query_ids.dedup();
+        estimates.insert(new, combined);
+    }
+    for m in materializations {
+        if let Some(canonical) = replacements.get(&m.policy_fingerprint()) {
+            *m = canonical.clone();
+        }
+    }
+    for id in bindings.values_mut() {
+        if let Some(canonical) = replacements.get(id) {
+            *id = canonical.policy_fingerprint();
+        }
+    }
+    for (old, canonical) in &replacements {
+        if let Some(policy) = policies.remove(old) {
+            policies.insert(canonical.policy_fingerprint(), policy);
+        }
+    }
+    for producer in producers.iter_mut().chain(plan_producers.iter_mut()) {
+        if let Some(canonical) = replacements.get(&producer.materialization.fingerprint()) {
+            let new = canonical.policy_fingerprint();
+            producer.materialization = new.into();
+            producer.query_id = format!("state-{}", new.0);
+            producer.window_secs = canonical.window_size;
+            producer.slide_secs = canonical.slide_interval;
+            producer.abstract_window_framework = SummaryWindowFramework::Tumbling;
+            producer.window_implementation_id = estimates[&new].window_implementation_id.clone();
+        }
+    }
+    let mut seen = BTreeSet::new();
+    producers.retain(|producer| seen.insert(producer.materialization));
+}
diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml
index f4ccaec9..c4bdb708 100644
--- a/crates/asap_types/Cargo.toml
+++ b/crates/asap_types/Cargo.toml
@@ -34,4 +34,4 @@ sha2 = "0.10"
 # exactly (`control_plane/Cargo.toml`) -- two different revs of the same
 # git dependency in one workspace resolve to two distinct Rust types that
 # won't unify.
-planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "3be523fa0f06a905188e42cbe482d06aa843ba5d" }
+planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ca7546de792d74aee8231e9a1100ca893d9e86d3" }
diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml
index 51d712b7..50527e01 100644
--- a/data_plane/Cargo.toml
+++ b/data_plane/Cargo.toml
@@ -39,8 +39,8 @@ sha2 = "0.10"
 # reduction: Reduction, .. }`) are `pre_asap` types, in the same crate now
 # (not a separate `asap-ir` import). Query serving consumes the compiled
 # QueryPlan; these types are used at physical-plan compilation boundaries.
-planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "3be523fa0f06a905188e42cbe482d06aa843ba5d" }
-asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "3be523fa0f06a905188e42cbe482d06aa843ba5d" }
+planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ca7546de792d74aee8231e9a1100ca893d9e86d3" }
+asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ca7546de792d74aee8231e9a1100ca893d9e86d3" }
 
 # Shared external (workspace)
 serde.workspace = true
@@ -133,7 +133,7 @@ fs2 = "0.4"
 # none of them.
 
 [dev-dependencies]
-asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "3be523fa0f06a905188e42cbe482d06aa843ba5d" }
+asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ca7546de792d74aee8231e9a1100ca893d9e86d3" }
 tempfile = "3.20.0"
 criterion = { version = "0.5", features = ["html_reports"] }
 tokio-tungstenite = "0.21"
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 6a4c6268..b91738c3 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
@@ -1160,6 +1160,95 @@ mod tests {
         assert_eq!(outcome.coverage, Some((2_000, 2_000)));
     }
 
+    // Compile the two readouts, store one pane series, and execute the actual ratio.
+    #[test]
+    fn compiled_shared_sum_panes_preserve_each_lookback() {
+        use control_plane::physical::compiler::{BackendLocalPlanningSnapshot, PhysicalCompiler};
+        let mut snapshot: serde_json::Value = serde_json::from_str(include_str!(
+            "../../../../docs/examples/asapquery-planning-snapshot.json"
+        ))
+        .unwrap();
+        let entry = &mut snapshot["query_workload"]["repeating_queries"][0];
+        entry["query"] = serde_json::json!("sum_over_time(a[1m]) / sum_over_time(a[10m])");
+        entry["requirements"]["accuracy"]["explicit"] = serde_json::json!("Exact");
+        entry["demand"]["fixed_interval_at"]["interval"] = serde_json::json!(60_000);
+        let snapshot: BackendLocalPlanningSnapshot = serde_json::from_value(snapshot).unwrap();
+        let (request, env) = snapshot.planning_request().unwrap();
+        let plan = PhysicalCompiler.compile(request, env).unwrap();
+        assert_eq!(plan.precompute_plan.materializations.len(), 1);
+        let config = &plan.precompute_plan.materializations[0];
+        let policy = config.policy_fingerprint();
+        let idx = SketchStore::new();
+        idx.register(SketchInstanceMetadata {
+            sid: 7,
+            metric_name: "a".into(),
+            group_by_keys: Default::default(),
+            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,
+        });
+        for pane in 0..11 {
+            idx.append_precompute(
+                7,
+                BTreeMap::new(),
+                (pane * 60_000, (pane + 1) * 60_000),
+                Box::new(
+                    crate::precompute_engine::operators::SumAccumulator::with_sum(
+                        (pane + 1) as f64,
+                    ),
+                ),
+            );
+        }
+        let entry = plan.query_plan.entries.values().next().unwrap();
+        for (now, expected) in [(600_000, 10.0 / 55.0), (660_000, 11.0 / 65.0)] {
+            let (outcome, stats) = super::super::logical_dag::execute_installed(
+                entry,
+                &BTreeMap::new(),
+                now,
+                |root, at| {
+                    let mut subtree = entry.clone();
+                    subtree.root = root;
+                    let reachable = subtree.topological_order().unwrap();
+                    subtree.nodes.retain(|id, _| reachable.contains(id));
+                    subtree.instant.lookback_ms = subtree.materialization_bindings()[0]
+                        .readout_lookback_ms
+                        .unwrap();
+                    super::super::live_serve::serve_instant_from_query_plan(&idx, &subtree, at)
+                        .map(|(result, _)| {
+                            use crate::query_engines::query_result::{InstantVectorElement, QueryResult};
+                            let values = result.series.into_iter().map(|(labels, samples)| {
+                                let (keys, values) = labels.into_iter().unzip();
+                                InstantVectorElement::new(crate::storage_engines::types::KeyByLabelValues::new_with_labels(values), samples.last().unwrap().1)
+                                    .with_label_keys_override(keys)
+                            }).collect();
+                            QueryResult::vector(values, at)
+                        })
+                        .map_err(|error| {
+                            crate::query_engines::EngineError::capability_miss(
+                                "test",
+                                format!("{error:?}"),
+                            )
+                        })
+                },
+            )
+            .unwrap();
+            let crate::query_engines::query_result::QueryResult::Vector(outcome) = outcome else {
+                panic!("expected vector");
+            };
+            assert_eq!(stats.summary_readout_evaluations, 2);
+            assert_eq!(outcome.values.len(), 1);
+            assert!((outcome.values[0].value - expected).abs() < 1e-12);
+        }
+    }
+
     #[test]
     fn repeated_multi_pane_reads_exclude_expired_state_and_reject_gaps() {
         let idx = SketchStore::new();
diff --git a/docs/developer_docs/planning/shared-window-panes.md b/docs/developer_docs/planning/shared-window-panes.md
new file mode 100644
index 00000000..15eea28b
--- /dev/null
+++ b/docs/developer_docs/planning/shared-window-panes.md
@@ -0,0 +1,47 @@
+# Shared additive window panes
+
+The fallback window provider prices each query range. After logical selection,
+raw inputs of derived maintenance programs use full, non-overlapping windows,
+matching the current maintenance executor. Explicit `window_candidates` remain
+authoritative and are not rewritten.
+
+For backend-local raw SUM states, the backend offers compatible selected pane
+producers to ASAPPlanner's `pane_sharing::select_shared_panes`. The compatibility
+key preserves source, population predicate, projection, accumulator parameters,
+grouping, partitioning, pane width/origin, runtime policy and lifecycle evidence.
+Planner compares independent producers with one producer retained for the longest
+lookback, charging every readout. Only beneficial groups are installed.
+
+The shared state's physical window is one pane. Logical readout windows stay on
+their original QueryPlan bindings. Retention uses the largest bound lookback.
+For `sum_over_time(a[1m]) / sum_over_time(a[10m])` evaluated every minute, one
+60-second producer retains 11 published states; readouts merge one and ten panes.
+Different metrics, predicates or phases remain separate. Derived inputs, explicit
+window quotes, non-additive state and independently selected FullWindow layouts
+are not rewritten. This pass reuses compatible selected panes; it does not search
+all possible pane widths or re-optimize FullWindow choices jointly.
+
+Layout pricing charges both published retained states and open worker
+accumulators. A full window of width W and step S keeps ceil(W/S) open states;
+a pane producer keeps one. These residency costs are separate from update CPU.
+Nonfinite arithmetic remains invalid evidence rather than becoming a zero quote.
+
+## Cross-repository validation
+
+The backend pins Planner commit `ca7546de792d74aee8231e9a1100ca893d9e86d3`, which provides
+`pane_sharing::select_shared_panes`. Normal builds use the Git dependency.
+For coordinated local development, the optional validation script exports only
+the optimizer crate while retaining the pinned IR/frontend revision and restores
+Cargo.lock after the run.
+
+```sh
+python3 tools/test_shared_panes.py --planner /path/to/ASAPPlanner -- \
+  test -p control_plane
+python3 tools/test_shared_panes.py --planner /path/to/ASAPPlanner \
+  --sketchlib /path/to/compatible/asap_sketchlib -- \
+  test -p data_plane --lib compiled_shared_sum_panes_preserve_each_lookback
+```
+
+Use `--toolchain 1.98.0` if needed in the development environment. The data-plane
+checkout requires sketchlib's standard-update guard and interpolated quantile
+interfaces; validation used revision `8c03d7c` for those existing dependencies.
diff --git a/tools/test_shared_panes.py b/tools/test_shared_panes.py
new file mode 100644
index 00000000..353c9e38
--- /dev/null
+++ b/tools/test_shared_panes.py
@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+"""Validate the unmerged Planner/backend pair without changing immutable IR pins."""
+import argparse
+import json
+from pathlib import Path
+import re
+import shutil
+import subprocess
+import tempfile
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--planner", required=True, type=Path)
+    parser.add_argument("--sketchlib", type=Path)
+    parser.add_argument("--toolchain")
+    parser.add_argument("cargo_args", nargs=argparse.REMAINDER)
+    args = parser.parse_args()
+    backend = Path(__file__).resolve().parents[1]
+    manifest = (backend / "control_plane/Cargo.toml").read_text()
+    revision = re.search(r'planner-types = .*rev = "([0-9a-f]+)"', manifest).group(1)
+    lock = backend / "Cargo.lock"
+    original_lock = lock.read_bytes()
+    with tempfile.TemporaryDirectory(prefix="asap-pane-reuse-") as temporary:
+        root = Path(temporary)
+        mapping = root / "asap-aware-mapping"
+        shutil.copytree(args.planner.resolve() / "crates/asap-aware-mapping", mapping)
+        cargo_toml = mapping / "Cargo.toml"
+        source = cargo_toml.read_text()
+        old = 'asap-types = { path = "../types" }'
+        if old not in source:
+            raise RuntimeError("unexpected Planner dependency declaration")
+        cargo_toml.write_text(source.replace(old, 'asap-types = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "' + revision + '" }'))
+        config = root / "validation.toml"
+        text = ""
+        if args.sketchlib:
+            text += "paths = [" + json.dumps(str(args.sketchlib.resolve())) + "]\n"
+        text += '[patch."https://github.com/ProjectASAP/ASAPPlanner"]\nasap-aware-mapping = { path = ' + json.dumps(str(mapping)) + ' }\n'
+        config.write_text(text)
+        command = ["cargo"]
+        if args.toolchain:
+            command.append("+" + args.toolchain)
+        cargo_args = args.cargo_args
+        if cargo_args[:1] == ["--"]:
+            cargo_args = cargo_args[1:]
+        command += ["--config", str(config)] + (cargo_args or ["test", "-p", "control_plane"])
+        try:
+            return subprocess.call(command, cwd=backend)
+        finally:
+            lock.write_bytes(original_lock)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())