From fe21bcedc15f2154a0a5f59357d25c5ada80a9ed Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Sun, 13 Sep 2026 19:33:13 -0400
Subject: [PATCH 1/3] derive snapshot windows from PromQL

---
 control_plane/src/main.rs                     |   1 -
 control_plane/src/physical/compiler.rs        | 153 +++++++++++++++---
 control_plane/src/physical/workload_cost.rs   |   1 -
 control_plane/src/query_plan/logical.rs       |  18 ---
 .../multisource_coordinator.rs                |   1 -
 .../sketch_db/index/maintenance.rs            |   1 -
 .../asapquery_compatibility_process_e2e.rs    |   3 -
 .../support/immutable_maintenance_process.rs  |   1 -
 .../planning/repeated-dashboard-panes.md      |   8 +
 ...asapquery-compatibility-demo-snapshot.json |  13 +-
 .../examples/asapquery-planning-snapshot.json |   4 +-
 docs/user_guide/asapquery-profile.md          |   9 +-
 12 files changed, 151 insertions(+), 62 deletions(-)

diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs
index 7e8833e6..3618b3e2 100644
--- a/control_plane/src/main.rs
+++ b/control_plane/src/main.rs
@@ -960,7 +960,6 @@ fn compile_physical_plan_request(
         exact_composition_costs: request.exact_composition_costs,
         erp: request.erp,
         planner_revision: request.planner_revision,
-        source_sample_interval_ms: None,
         query_staleness_margin_ms: 0,
         retained_summary_memory_budget_bytes: None,
     };
diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index 660ec1c0..313aeabc 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -26,7 +26,7 @@ use planner_types::workload::{
     Predictability, Query, QueryLanguage, QueryRecurrence, QueryRequirements, QueryTimeScope,
     QueryWorkload, Rate, RepeatedDemand, RepeatingEntry, RepetitionInterval, TimeSelection,
 };
-use serde::{Deserialize, Serialize};
+use serde::{de::Error as _, Deserialize, Deserializer, Serialize};
 use serde_json::{json, Value};
 use thiserror::Error;
 
@@ -170,9 +170,6 @@ pub struct PlanningRequest {
     /// mode falls back to theoretical sizing and then exact execution.
     pub erp: Option<super::erp::ErpPlanningInput>,
     pub planner_revision: String,
-    /// Observed cadence of source samples. Exact temporal panes must divide
-    /// both this cadence and the repeated-query evaluation interval.
-    pub source_sample_interval_ms: Option<u64>,
     /// How far behind the newest ingested sample an admitted query may be
     /// evaluated. This extends physical retention only; it never changes the
     /// PromQL range selector used for readout.
@@ -224,6 +221,7 @@ pub struct BackendLocalPlanningSnapshot {
     /// 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>,
+    #[serde(deserialize_with = "deserialize_snapshot_query_workload")]
     pub query_workload: QueryWorkload,
     pub data_workload: DataWorkload,
     pub implementation: BackendLocalImplementation,
@@ -238,8 +236,9 @@ pub struct BackendLocalImplementation {
     pub evidence_valid_for_ms: u64,
     pub horizon_seconds: f64,
     pub window_cost_model: WindowCostModel,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub source_sample_interval_ms: Option<u64>,
+    /// Prometheus source scrape cadence. Rangeless instant-vector expressions
+    /// use this as their derived readout window.
+    pub scrape_interval_ms: u64,
     #[serde(default, skip_serializing_if = "u64_is_zero")]
     pub query_staleness_margin_ms: u64,
     /// Admission budget for all retained panes and estimated partitions.
@@ -262,6 +261,33 @@ pub struct BackendLocalImplementation {
     pub erp: Option<super::erp::ErpPlanningInput>,
 }
 
+/// Snapshot lookback is derived from PromQL and the declared scrape interval;
+/// accepting it here would silently restore the competing legacy input.
+fn deserialize_snapshot_query_workload<'de, D>(deserializer: D) -> Result<QueryWorkload, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let value = Value::deserialize(deserializer)?;
+    let has_legacy_lookback = value
+        .get("repeating_queries")
+        .and_then(Value::as_array)
+        .is_some_and(|entries| {
+            entries.iter().any(|entry| {
+                entry
+                    .get("time_selection")
+                    .and_then(Value::as_object)
+                    .and_then(|selection| selection.get("lookback"))
+                    .is_some_and(|lookback| !lookback.is_null())
+            })
+        });
+    if has_legacy_lookback {
+        return Err(D::Error::custom(
+            "query_workload.repeating_queries[].time_selection.lookback is not supported; it is derived from PromQL",
+        ));
+    }
+    serde_json::from_value(value).map_err(D::Error::custom)
+}
+
 fn u64_is_zero(value: &u64) -> bool {
     *value == 0
 }
@@ -573,16 +599,11 @@ impl BackendLocalPlanningSnapshot {
                     )))
                 }
             };
-            let lookback_ms = entry
-                .time_selection
-                .lookback
-                .ok_or_else(|| {
-                    CompileError::Snapshot(format!("query {index} requires an explicit lookback"))
-                })?
-                .0;
-            if lookback_ms == 0 || lookback_ms % 1_000 != 0 {
+            if self.implementation.scrape_interval_ms == 0
+                || self.implementation.scrape_interval_ms % 1_000 != 0
+            {
                 return Err(CompileError::Snapshot(format!(
-                    "query {index} lookback must be a positive whole number of seconds"
+                    "scrape_interval_ms must be a positive whole number of seconds"
                 )));
             }
             let accuracy = entry.requirements.accuracy.target();
@@ -590,6 +611,8 @@ impl BackendLocalPlanningSnapshot {
             let parsed =
                 crate::query_parser::parse_query_expr_canonical(&query_string, accuracy.clone())
                     .map_err(|error| CompileError::Snapshot(format!("query {index}: {error}")))?;
+            let lookback_ms =
+                query_history_window_ms(&parsed).unwrap_or(self.implementation.scrape_interval_ms);
             let metadata = crate::query_parser::qe_to_parsed_query(&parsed);
             let source_metrics = super::workload_cost::exact_source_metrics(&parsed)?;
             let source_hint = source_metrics.iter().next().cloned().ok_or_else(|| {
@@ -673,7 +696,6 @@ impl BackendLocalPlanningSnapshot {
                 exact_composition_costs: exact_costs_by_id,
                 erp: self.implementation.erp,
                 planner_revision: PLANNER_REVISION.into(),
-                source_sample_interval_ms: self.implementation.source_sample_interval_ms,
                 query_staleness_margin_ms: self.implementation.query_staleness_margin_ms,
                 retained_summary_memory_budget_bytes: Some(
                     self.implementation.max_retained_summary_bytes,
@@ -2374,12 +2396,62 @@ pub(super) fn derived_window_cost(
     }
 }
 
+/// The furthest point in the past that evaluating `expr` can read, in ms.
+///
+/// A range selector and a subquery each evaluate their child over prior
+/// history; a positive offset moves that history further back. The result is a
+/// lower-bound horizon, not a claim that the entire interval is read. For
+/// example, `a[1m] offset 1h` selects `(t - 61m, t - 60m]`.
+fn query_history_window_ms(expr: &QueryExpr) -> Option<u64> {
+    fn duration_ms(duration: std::time::Duration) -> u64 {
+        u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
+    }
+
+    fn visit(expr: &QueryExpr, preceding_history_ms: u64) -> u64 {
+        match expr {
+            QueryExpr::PromqlSubquery { range, child, .. }
+            | QueryExpr::TimeRange { range, child } => visit(
+                child,
+                preceding_history_ms.saturating_add(duration_ms(*range)),
+            ),
+            QueryExpr::TimeShift { shift, child } => visit(
+                child,
+                preceding_history_ms.saturating_add(shift.offset_ms.max(0) as u64),
+            ),
+            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, .. } => visit(child, preceding_history_ms),
+            QueryExpr::BinaryOp {
+                lhs: left,
+                rhs: right,
+                ..
+            }
+            | QueryExpr::Join { left, right, .. }
+            | QueryExpr::SetOp { left, right, .. } => {
+                visit(left, preceding_history_ms).max(visit(right, preceding_history_ms))
+            }
+            _ => preceding_history_ms,
+        }
+    }
+
+    let history_ms = visit(expr, 0);
+    (history_ms != 0).then_some(history_ms)
+}
+
 /// 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.
+/// These are materialization contracts. Enclosing subquery ranges and offsets
+/// affect a query's furthest lookback but do not change a leaf contract.
 #[cfg(test)]
 fn range_selector_windows_secs(expr: &QueryExpr) -> BTreeSet<u64> {
     fn visit(expr: &QueryExpr, windows: &mut BTreeSet<u64>) {
@@ -2423,8 +2495,7 @@ fn range_selector_windows_secs(expr: &QueryExpr) -> BTreeSet<u64> {
 }
 
 /// 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;
+/// Derived maintenance cohorts are restricted to their supported full windows;
 /// raw additive pane producers may subsequently be shared by Planner.
 #[cfg(test)]
 fn derived_window_candidates(
@@ -4171,7 +4242,6 @@ pub(crate) mod tests {
             erp: None,
             exact_composition_costs: HashMap::new(),
             planner_revision: PLANNER_REVISION.into(),
-            source_sample_interval_ms: None,
             query_staleness_margin_ms: 0,
             retained_summary_memory_budget_bytes: None,
         })
@@ -5549,6 +5619,42 @@ pub(crate) mod tests {
         .unwrap()
     }
 
+    fn derived_query_window_secs(query: &str) -> u64 {
+        let mut snapshot = planning_snapshot();
+        let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0];
+        entry.query = Query(query.into());
+        entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact);
+        snapshot.planning_request().unwrap().0.queries[0].window_secs
+    }
+
+    #[test]
+    fn snapshot_rejects_manual_lookback() {
+        let mut wire = serde_json::to_value(planning_snapshot()).unwrap();
+        wire["query_workload"]["repeating_queries"][0]["time_selection"]["lookback"] =
+            60_000.into();
+        let error = serde_json::from_value::<BackendLocalPlanningSnapshot>(wire)
+            .expect_err("manual lookback must not be accepted")
+            .to_string();
+        assert!(error.contains("lookback is not supported"), "{error}");
+    }
+
+    #[test]
+    fn subquery_range_derives_the_query_window() {
+        assert_eq!(
+            derived_query_window_secs("avg_over_time((sum(a))[6h:])"),
+            6 * 60 * 60
+        );
+    }
+
+    #[test]
+    fn offset_extends_the_query_furthest_lookback() {
+        // Evaluated at t, `a[1m] offset 1h` selects `(t - 61m, t - 60m]`.
+        assert_eq!(
+            derived_query_window_secs("sum_over_time(a[1m] offset 1h)"),
+            61 * 60
+        );
+    }
+
     // 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.
@@ -5944,7 +6050,6 @@ pub(crate) mod tests {
         {
             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),
@@ -6481,7 +6586,7 @@ pub(crate) mod tests {
                 predictability: Predictability::Predictable { known_at: None },
                 time_selection: TimeSelection {
                     scope: QueryTimeScope::RealTime,
-                    lookback: Some(DurationMs(60_000)),
+                    lookback: None,
                     as_of: None,
                 },
             }]),
@@ -6508,7 +6613,7 @@ pub(crate) mod tests {
                     cost: template.window_implementations[0].cost.clone(),
                     quotes: Vec::new(),
                 },
-                source_sample_interval_ms: None,
+                scrape_interval_ms: 1_000,
                 query_staleness_margin_ms: 0,
                 max_retained_summary_bytes: DEFAULT_RETAINED_SUMMARY_MEMORY_BUDGET_BYTES,
                 topk_evidence: HashMap::new(),
diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs
index 73566018..a99ab827 100644
--- a/control_plane/src/physical/workload_cost.rs
+++ b/control_plane/src/physical/workload_cost.rs
@@ -822,7 +822,6 @@ mod tests {
         entries[0].requirements.accuracy = planner_types::workload::AccuracyRequirement::Explicit(
             planner_types::types::AccuracyTarget::Exact,
         );
-        entries[0].time_selection.lookback = Some(planner_types::workload::DurationMs(21_600_000));
         let mut second = entries[0].clone();
         second.query = planner_types::workload::Query(
             "max_over_time(service_retry_queue_depth{job=\"order-service\"}[6h])".into(),
diff --git a/control_plane/src/query_plan/logical.rs b/control_plane/src/query_plan/logical.rs
index a2cbbb17..d431b9e1 100644
--- a/control_plane/src/query_plan/logical.rs
+++ b/control_plane/src/query_plan/logical.rs
@@ -716,19 +716,6 @@ mod planner_workload_tests {
     use super::*;
     use crate::physical::compiler::{BackendLocalPlanningSnapshot, PhysicalCompiler};
 
-    fn lookback(expr: &Expr) -> u64 {
-        match expr {
-            Expr::MatrixSelector(e) => e.range.as_millis() as u64,
-            Expr::Subquery(e) => (e.range.as_millis() as u64).max(lookback(&e.expr)),
-            Expr::Aggregate(e) => lookback(&e.expr),
-            Expr::Paren(e) => lookback(&e.expr),
-            Expr::Unary(e) => lookback(&e.expr),
-            Expr::Binary(e) => lookback(&e.lhs).max(lookback(&e.rhs)),
-            Expr::Call(e) => e.args.args.iter().map(|e| lookback(e)).max().unwrap_or(0),
-            _ => 0,
-        }
-    }
-
     fn compile_one(query: &str) -> crate::physical::compiler::PhysicalPlan {
         let mut fixture: serde_json::Value = serde_json::from_str(include_str!(
             "../../../docs/examples/asapquery-planning-snapshot.json"
@@ -737,8 +724,6 @@ mod planner_workload_tests {
         let mut entry = fixture["query_workload"]["repeating_queries"][0].clone();
         entry["query"] = query.into();
         entry["requirements"]["accuracy"] = serde_json::json!({"explicit":"Exact"});
-        let window = lookback(&parser::parse(query).unwrap());
-        entry["time_selection"]["lookback"] = (if window == 0 { 300_000 } else { window }).into();
         fixture["query_workload"]["repeating_queries"] = vec![entry].into();
         let snapshot: BackendLocalPlanningSnapshot = serde_json::from_value(fixture).unwrap();
         let (request, environment) = snapshot
@@ -824,9 +809,6 @@ mod planner_workload_tests {
             let mut entry = template.clone();
             entry["query"] = query.into();
             entry["requirements"]["accuracy"] = serde_json::json!({"explicit":"Exact"});
-            let window = lookback(&parser::parse(query).unwrap());
-            entry["time_selection"]["lookback"] =
-                (if window == 0 { 300_000 } else { window }).into();
             entries.push(entry);
         }
         fixture["query_workload"]["repeating_queries"] = entries.into();
diff --git a/data_plane/src/precompute_engine/multisource_coordinator.rs b/data_plane/src/precompute_engine/multisource_coordinator.rs
index 0b305fa8..e0fa3dba 100644
--- a/data_plane/src/precompute_engine/multisource_coordinator.rs
+++ b/data_plane/src/precompute_engine/multisource_coordinator.rs
@@ -615,7 +615,6 @@ mod tests {
         query["query"] = "quantile(0.9, sum_over_time(m[1m]) + sum_over_time(n[1m]))".into();
         query["demand"]["fixed_interval_at"]["interval"] = 60000.into();
         query["demand"]["fixed_interval_at"]["evaluation_phase"] = 0.into();
-        query["time_selection"]["lookback"] = 60000.into();
         wire["query_workload"]["repeating_queries"] = serde_json::json!([query]);
         let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot =
             serde_json::from_value(wire).unwrap();
diff --git a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs
index 20cedea5..d3091369 100644
--- a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs
+++ b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs
@@ -1082,7 +1082,6 @@ mod tests {
         entry["query"] = "quantile(0.9, sum_over_time(immutable_value[1m]))".into();
         entry["demand"]["fixed_interval_at"]["interval"] = 60_000.into();
         entry["demand"]["fixed_interval_at"]["evaluation_phase"] = 0.into();
-        entry["time_selection"]["lookback"] = 60_000.into();
         fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]);
         let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot =
             serde_json::from_value(fixture).unwrap();
diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs
index a50a3452..8f93448a 100644
--- a/data_plane/tests/asapquery_compatibility_process_e2e.rs
+++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs
@@ -786,9 +786,6 @@ async fn run_shared_dashboard(multi_pane: bool) {
     let mut typed: control_plane::physical::compiler::BackendLocalPlanningSnapshot =
         serde_json::from_value(snapshot.clone()).unwrap();
     if multi_pane {
-        for entry in typed.query_workload.repeating_queries.as_mut().unwrap() {
-            entry.time_selection.lookback = Some(planner_types::workload::DurationMs(10_000));
-        }
         for entry in typed.query_workload.repeating_queries.as_mut().unwrap() {
             entry.demand = planner_types::workload::RepeatedDemand::FixedIntervalAt {
                 interval: planner_types::workload::RepetitionInterval(5_000),
diff --git a/data_plane/tests/support/immutable_maintenance_process.rs b/data_plane/tests/support/immutable_maintenance_process.rs
index f105cdfa..f2b6dd54 100644
--- a/data_plane/tests/support/immutable_maintenance_process.rs
+++ b/data_plane/tests/support/immutable_maintenance_process.rs
@@ -31,7 +31,6 @@ async fn run_maintenance_process(multi_source: bool, distinct_groups: bool) {
     entry["query"] = query.into();
     entry["demand"]["fixed_interval_at"]["interval"] = 60_000.into();
     entry["demand"]["fixed_interval_at"]["evaluation_phase"] = 0.into();
-    entry["time_selection"]["lookback"] = 60_000.into();
     fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]);
     let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot =
         serde_json::from_value(fixture.clone()).unwrap();
diff --git a/docs/developer_docs/planning/repeated-dashboard-panes.md b/docs/developer_docs/planning/repeated-dashboard-panes.md
index 69bef94c..a9603d9c 100644
--- a/docs/developer_docs/planning/repeated-dashboard-panes.md
+++ b/docs/developer_docs/planning/repeated-dashboard-panes.md
@@ -25,6 +25,14 @@ Unquoted layouts use supplied lifecycle unit costs multiplied by structural
 counts. Layout changes never inherit another layout's measured scalar cost.
 Workload-versus-exact deployment evidence is still required by snapshot version 2.
 
+Temporal requirements are derived from PromQL. A range selector supplies its
+own readout window; a rangeless instant-vector expression uses required
+`implementation.scrape_interval_ms`. Snapshot `time_selection.lookback` is
+rejected because it duplicates and can conflict with query semantics. A
+subquery or positive offset extends only the furthest lookback: evaluated at
+`t`, `a[1m] offset 1h` selects `(t - 61m, t - 60m]`, not a continuous
+61-minute interval.
+
 ## Cases
 
 All ranges below use PromQL's `(start,end]` convention. W is the lookback and E
diff --git a/docs/examples/asapquery-compatibility-demo-snapshot.json b/docs/examples/asapquery-compatibility-demo-snapshot.json
index 87ecd811..905981ed 100644
--- a/docs/examples/asapquery-compatibility-demo-snapshot.json
+++ b/docs/examples/asapquery-compatibility-demo-snapshot.json
@@ -9,21 +9,21 @@
         "demand": { "fixed_interval_at": { "interval": 1000, "evaluation_phase": 0 } },
         "requirements": { "accuracy": "implicit_exact", "response_latency": "unspecified" },
         "predictability": { "predictable": { "known_at": null } },
-        "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null }
+        "time_selection": { "scope": "real_time", "as_of": null }
       },
       {
         "query": "increase(asap_demo_counter_total[5s])",
         "demand": { "fixed_interval_at": { "interval": 1000, "evaluation_phase": 0 } },
         "requirements": { "accuracy": "implicit_exact", "response_latency": "unspecified" },
         "predictability": { "predictable": { "known_at": null } },
-        "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null }
+        "time_selection": { "scope": "real_time", "as_of": null }
       },
       {
         "query": "sum(sum_over_time(asap_demo_gauge[5s]))",
         "demand": { "fixed_interval_at": { "interval": 1000, "evaluation_phase": 0 } },
         "requirements": { "accuracy": "implicit_exact", "response_latency": "unspecified" },
         "predictability": { "predictable": { "known_at": null } },
-        "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null }
+        "time_selection": { "scope": "real_time", "as_of": null }
       },
       {
         "query": "quantile_over_time(0.5, asap_demo_latency_ms[5s])",
@@ -33,7 +33,7 @@
           "response_latency": "unspecified"
         },
         "predictability": { "predictable": { "known_at": null } },
-        "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null }
+        "time_selection": { "scope": "real_time", "as_of": null }
       },
       {
         "query": "topk(1, sum_over_time(asap_demo_gauge[5s]))",
@@ -43,7 +43,7 @@
           "response_latency": "unspecified"
         },
         "predictability": { "predictable": { "known_at": null } },
-        "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null }
+        "time_selection": { "scope": "real_time", "as_of": null }
       },
       {
         "query": "topk(1, count_over_time(asap_demo_gauge[5s]))",
@@ -53,7 +53,7 @@
           "response_latency": "unspecified"
         },
         "predictability": { "predictable": { "known_at": null } },
-        "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null }
+        "time_selection": { "scope": "real_time", "as_of": null }
       }
     ],
     "data_workload": {
@@ -98,6 +98,7 @@
         "weighted_cost": 1.0
       }
     },
+    "scrape_interval_ms": 5000,
     "topk_evidence": {
       "topk(1, sum_over_time(asap_demo_gauge[5s]))": {
         "selected_lower_bound": 101.0,
diff --git a/docs/examples/asapquery-planning-snapshot.json b/docs/examples/asapquery-planning-snapshot.json
index b4147ae6..46d4d164 100644
--- a/docs/examples/asapquery-planning-snapshot.json
+++ b/docs/examples/asapquery-planning-snapshot.json
@@ -18,7 +18,6 @@
         "predictability": { "predictable": { "known_at": null } },
         "time_selection": {
           "scope": "real_time",
-          "lookback": 60000,
           "as_of": null
         }
       }
@@ -64,7 +63,8 @@
         "source_scan_bytes": 0,
         "weighted_cost": 1.0
       }
-    }
+    },
+    "scrape_interval_ms": 5000
   },
   "environment": {
     "target": "backend_local_remote_write",
diff --git a/docs/user_guide/asapquery-profile.md b/docs/user_guide/asapquery-profile.md
index 1cf44766..3f0b4366 100644
--- a/docs/user_guide/asapquery-profile.md
+++ b/docs/user_guide/asapquery-profile.md
@@ -137,7 +137,8 @@ coverage is a capability miss, never a partial warm success. Inspect the
 per-materialization state and observed coverage through
 `GET /api/v1/physical-plan/status`.
 
-Snapshot schema version `1` currently accepts fixed-interval repeating PromQL
-queries with explicit whole-second lookbacks and fresh ingestion-rate
-evidence. Unsupported snapshot semantics fail startup rather than silently
-inventing cost or placement evidence.
+Snapshot schema version `2` accepts fixed-interval repeating PromQL queries
+with `implementation.scrape_interval_ms` and fresh ingestion-rate evidence.
+Range selectors derive their own lookback; rangeless instant-vector queries use
+the scrape interval. Unsupported snapshot semantics fail startup rather than
+silently inventing cost or placement evidence.

From 821fc07e73187a9da176e19caa662df5fc2a1ae5 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Sun, 13 Sep 2026 21:34:29 -0600
Subject: [PATCH 2/3] fix(planning): preserve history across every query branch

---
 control_plane/src/physical/compiler.rs        | 128 ++++++++++++++----
 .../planning/repeated-dashboard-panes.md      |  22 ++-
 ...asapquery-compatibility-demo-snapshot.json |  12 +-
 .../examples/asapquery-planning-snapshot.json |   1 +
 docs/user_guide/asapquery-profile.md          |  10 +-
 5 files changed, 135 insertions(+), 38 deletions(-)

diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index 313aeabc..4d8d49ec 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -237,7 +237,8 @@ pub struct BackendLocalImplementation {
     pub horizon_seconds: f64,
     pub window_cost_model: WindowCostModel,
     /// Prometheus source scrape cadence. Rangeless instant-vector expressions
-    /// use this as their derived readout window.
+    /// use this as their derived readout window. This backend-local planning
+    /// default is distinct from Prometheus instant-selector lookback delta.
     pub scrape_interval_ms: u64,
     #[serde(default, skip_serializing_if = "u64_is_zero")]
     pub query_staleness_margin_ms: u64,
@@ -600,11 +601,11 @@ impl BackendLocalPlanningSnapshot {
                 }
             };
             if self.implementation.scrape_interval_ms == 0
-                || self.implementation.scrape_interval_ms % 1_000 != 0
+                || !self.implementation.scrape_interval_ms.is_multiple_of(1_000)
             {
-                return Err(CompileError::Snapshot(format!(
-                    "scrape_interval_ms must be a positive whole number of seconds"
-                )));
+                return Err(CompileError::Snapshot(
+                    "scrape_interval_ms must be a positive whole number of seconds".into(),
+                ));
             }
             let accuracy = entry.requirements.accuracy.target();
             let query_string = entry.query.0;
@@ -612,7 +613,8 @@ impl BackendLocalPlanningSnapshot {
                 crate::query_parser::parse_query_expr_canonical(&query_string, accuracy.clone())
                     .map_err(|error| CompileError::Snapshot(format!("query {index}: {error}")))?;
             let lookback_ms =
-                query_history_window_ms(&parsed).unwrap_or(self.implementation.scrape_interval_ms);
+                query_history_window_ms(&parsed, self.implementation.scrape_interval_ms)
+                    .map_err(|error| CompileError::Snapshot(format!("query {index}: {error}")))?;
             let metadata = crate::query_parser::qe_to_parsed_query(&parsed);
             let source_metrics = super::workload_cost::exact_source_metrics(&parsed)?;
             let source_hint = source_metrics.iter().next().cloned().ok_or_else(|| {
@@ -2402,22 +2404,51 @@ pub(super) fn derived_window_cost(
 /// history; a positive offset moves that history further back. The result is a
 /// lower-bound horizon, not a claim that the entire interval is read. For
 /// example, `a[1m] offset 1h` selects `(t - 61m, t - 60m]`.
-fn query_history_window_ms(expr: &QueryExpr) -> Option<u64> {
-    fn duration_ms(duration: std::time::Duration) -> u64 {
-        u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
+fn query_history_window_ms(expr: &QueryExpr, scrape_interval_ms: u64) -> Result<u64, CompileError> {
+    fn duration_ms(duration: std::time::Duration) -> Result<u64, CompileError> {
+        if duration.subsec_nanos() != 0 {
+            return Err(CompileError::Snapshot(
+                "PromQL ranges must be a whole number of seconds in the backend-local profile"
+                    .into(),
+            ));
+        }
+        u64::try_from(duration.as_millis())
+            .map_err(|_| CompileError::Snapshot("PromQL range exceeds supported history".into()))
+    }
+
+    fn add(left: u64, right: u64) -> Result<u64, CompileError> {
+        left.checked_add(right).ok_or_else(|| {
+            CompileError::Snapshot("PromQL history exceeds supported duration".into())
+        })
     }
 
-    fn visit(expr: &QueryExpr, preceding_history_ms: u64) -> u64 {
+    fn visit(
+        expr: &QueryExpr,
+        scrape_interval_ms: u64,
+        in_range: bool,
+    ) -> Result<u64, CompileError> {
         match expr {
-            QueryExpr::PromqlSubquery { range, child, .. }
-            | QueryExpr::TimeRange { range, child } => visit(
-                child,
-                preceding_history_ms.saturating_add(duration_ms(*range)),
+            // A subquery evaluates instant expressions at earlier timestamps;
+            // those leaves still need their own default readout window.
+            QueryExpr::PromqlSubquery { range, child, .. } => add(
+                duration_ms(*range)?,
+                visit(child, scrape_interval_ms, false)?,
             ),
-            QueryExpr::TimeShift { shift, child } => visit(
-                child,
-                preceding_history_ms.saturating_add(shift.offset_ms.max(0) as u64),
+            QueryExpr::TimeRange { range, child } => add(
+                duration_ms(*range)?,
+                visit(child, scrape_interval_ms, true)?,
             ),
+            QueryExpr::TimeShift { shift, child } => {
+                if !shift.offset_ms.unsigned_abs().is_multiple_of(1_000) {
+                    return Err(CompileError::Snapshot(
+                        "PromQL offsets must be a whole number of seconds in the backend-local profile".into(),
+                    ));
+                }
+                add(
+                    shift.offset_ms.max(0) as u64,
+                    visit(child, scrape_interval_ms, in_range)?,
+                )
+            }
             QueryExpr::PromqlScalarBridge(child)
             | QueryExpr::PromqlVectorFromScalar(child)
             | QueryExpr::PromqlScalarFromVector(child)
@@ -2428,22 +2459,28 @@ fn query_history_window_ms(expr: &QueryExpr) -> Option<u64> {
             | QueryExpr::Aggregate { child, .. }
             | QueryExpr::Dedup { child, .. }
             | QueryExpr::Sort { child, .. }
-            | QueryExpr::Limit { child, .. } => visit(child, preceding_history_ms),
+            | QueryExpr::Limit { child, .. } => visit(child, scrape_interval_ms, in_range),
             QueryExpr::BinaryOp {
                 lhs: left,
                 rhs: right,
                 ..
             }
             | QueryExpr::Join { left, right, .. }
-            | QueryExpr::SetOp { left, right, .. } => {
-                visit(left, preceding_history_ms).max(visit(right, preceding_history_ms))
-            }
-            _ => preceding_history_ms,
+            | QueryExpr::SetOp { left, right, .. } => Ok(visit(
+                left,
+                scrape_interval_ms,
+                in_range,
+            )?
+            .max(visit(right, scrape_interval_ms, in_range)?)),
+            QueryExpr::Concat { children, .. } => children.iter().try_fold(0, |history, child| {
+                Ok(history.max(visit(child, scrape_interval_ms, in_range)?))
+            }),
+            QueryExpr::Scan { .. } if !in_range => Ok(scrape_interval_ms),
+            _ => Ok(0),
         }
     }
 
-    let history_ms = visit(expr, 0);
-    (history_ms != 0).then_some(history_ms)
+    visit(expr, scrape_interval_ms, false)
 }
 
 /// Every distinct range-selector window in `expr`, as seconds.
@@ -5627,6 +5664,47 @@ pub(crate) mod tests {
         snapshot.planning_request().unwrap().0.queries[0].window_secs
     }
 
+    // Every concatenated histogram result contributes its source history.
+    #[test]
+    fn derived_history_visits_concat_branches() {
+        assert_eq!(derived_query_window_secs(
+            "histogram_quantiles(rate(request_duration_seconds_bucket[5m]), \"quantile\", 0.5, 0.9)"
+        ), 300);
+    }
+
+    // A short ranged sibling must not suppress a raw selector's default window.
+    #[test]
+    fn derived_history_applies_cadence_at_each_rangeless_source() {
+        assert_eq!(
+            derived_query_window_secs("sum(a) + sum(sum_over_time(b[1s]))"),
+            5
+        );
+        assert_eq!(derived_query_window_secs("sum(a offset 1h)"), 3605);
+        assert_eq!(derived_query_window_secs("sum_over_time(a[1s])"), 1);
+    }
+
+    // The seconds-based backend must fail explicitly instead of shrinking history.
+    #[test]
+    fn derived_history_rejects_fractional_seconds() {
+        for query in [
+            "sum_over_time(a[1500ms])",
+            "sum_over_time(a[500ms])",
+            "sum_over_time(a[1500ms] offset 500ms)",
+            "sum_over_time(a[1s]) + sum(sum_over_time(b[500ms]))",
+            "avg_over_time((sum(a))[1500ms:])",
+            "sum(a offset 500ms)",
+        ] {
+            let mut snapshot = planning_snapshot();
+            snapshot.query_workload.repeating_queries.as_mut().unwrap()[0].query =
+                Query(query.into());
+            let error = snapshot.planning_request().expect_err(query).to_string();
+            assert!(
+                error.contains("whole number of seconds"),
+                "{query}: {error}"
+            );
+        }
+    }
+
     #[test]
     fn snapshot_rejects_manual_lookback() {
         let mut wire = serde_json::to_value(planning_snapshot()).unwrap();
@@ -5642,7 +5720,7 @@ pub(crate) mod tests {
     fn subquery_range_derives_the_query_window() {
         assert_eq!(
             derived_query_window_secs("avg_over_time((sum(a))[6h:])"),
-            6 * 60 * 60
+            6 * 60 * 60 + 5
         );
     }
 
diff --git a/docs/developer_docs/planning/repeated-dashboard-panes.md b/docs/developer_docs/planning/repeated-dashboard-panes.md
index a9603d9c..119c2014 100644
--- a/docs/developer_docs/planning/repeated-dashboard-panes.md
+++ b/docs/developer_docs/planning/repeated-dashboard-panes.md
@@ -26,10 +26,24 @@ counts. Layout changes never inherit another layout's measured scalar cost.
 Workload-versus-exact deployment evidence is still required by snapshot version 2.
 
 Temporal requirements are derived from PromQL. A range selector supplies its
-own readout window; a rangeless instant-vector expression uses required
-`implementation.scrape_interval_ms`. Snapshot `time_selection.lookback` is
-rejected because it duplicates and can conflict with query semantics. A
-subquery or positive offset extends only the furthest lookback: evaluated at
+own readout window; each rangeless source uses required
+`implementation.scrape_interval_ms` before enclosing offsets and subqueries
+are added. Branches, including concatenations, contribute their maximum history.
+For example, with a 5-second cadence, `sum(a offset 1h)` needs 3605 seconds,
+and `avg_over_time((sum(a))[6h:])` needs 21605 seconds. A ranged source keeps
+its explicit window even when it is shorter than the scrape cadence.
+
+This is the backend-local cadence-based planning contract, not Prometheus's
+[instant-selector lookback delta](https://prometheus.io/docs/prometheus/latest/querying/basics/#staleness)
+(which defaults to five minutes and governs selection of the latest non-stale sample). Scrape cadence does not configure
+that Prometheus setting or provide its staleness semantics.
+
+Non-null snapshot `time_selection.lookback` is rejected because it duplicates
+and can conflict with query semantics; omission and `null` are accepted.
+The backend stores window widths in seconds, so fractional-second ranges and
+offsets are rejected explicitly rather than rounded down.
+
+A subquery or positive offset extends only the furthest lookback: evaluated at
 `t`, `a[1m] offset 1h` selects `(t - 61m, t - 60m]`, not a continuous
 61-minute interval.
 
diff --git a/docs/examples/asapquery-compatibility-demo-snapshot.json b/docs/examples/asapquery-compatibility-demo-snapshot.json
index 905981ed..0c466eff 100644
--- a/docs/examples/asapquery-compatibility-demo-snapshot.json
+++ b/docs/examples/asapquery-compatibility-demo-snapshot.json
@@ -9,21 +9,21 @@
         "demand": { "fixed_interval_at": { "interval": 1000, "evaluation_phase": 0 } },
         "requirements": { "accuracy": "implicit_exact", "response_latency": "unspecified" },
         "predictability": { "predictable": { "known_at": null } },
-        "time_selection": { "scope": "real_time", "as_of": null }
+        "time_selection": { "scope": "real_time", "lookback": null, "as_of": null }
       },
       {
         "query": "increase(asap_demo_counter_total[5s])",
         "demand": { "fixed_interval_at": { "interval": 1000, "evaluation_phase": 0 } },
         "requirements": { "accuracy": "implicit_exact", "response_latency": "unspecified" },
         "predictability": { "predictable": { "known_at": null } },
-        "time_selection": { "scope": "real_time", "as_of": null }
+        "time_selection": { "scope": "real_time", "lookback": null, "as_of": null }
       },
       {
         "query": "sum(sum_over_time(asap_demo_gauge[5s]))",
         "demand": { "fixed_interval_at": { "interval": 1000, "evaluation_phase": 0 } },
         "requirements": { "accuracy": "implicit_exact", "response_latency": "unspecified" },
         "predictability": { "predictable": { "known_at": null } },
-        "time_selection": { "scope": "real_time", "as_of": null }
+        "time_selection": { "scope": "real_time", "lookback": null, "as_of": null }
       },
       {
         "query": "quantile_over_time(0.5, asap_demo_latency_ms[5s])",
@@ -33,7 +33,7 @@
           "response_latency": "unspecified"
         },
         "predictability": { "predictable": { "known_at": null } },
-        "time_selection": { "scope": "real_time", "as_of": null }
+        "time_selection": { "scope": "real_time", "lookback": null, "as_of": null }
       },
       {
         "query": "topk(1, sum_over_time(asap_demo_gauge[5s]))",
@@ -43,7 +43,7 @@
           "response_latency": "unspecified"
         },
         "predictability": { "predictable": { "known_at": null } },
-        "time_selection": { "scope": "real_time", "as_of": null }
+        "time_selection": { "scope": "real_time", "lookback": null, "as_of": null }
       },
       {
         "query": "topk(1, count_over_time(asap_demo_gauge[5s]))",
@@ -53,7 +53,7 @@
           "response_latency": "unspecified"
         },
         "predictability": { "predictable": { "known_at": null } },
-        "time_selection": { "scope": "real_time", "as_of": null }
+        "time_selection": { "scope": "real_time", "lookback": null, "as_of": null }
       }
     ],
     "data_workload": {
diff --git a/docs/examples/asapquery-planning-snapshot.json b/docs/examples/asapquery-planning-snapshot.json
index 46d4d164..4f0ed030 100644
--- a/docs/examples/asapquery-planning-snapshot.json
+++ b/docs/examples/asapquery-planning-snapshot.json
@@ -18,6 +18,7 @@
         "predictability": { "predictable": { "known_at": null } },
         "time_selection": {
           "scope": "real_time",
+          "lookback": null,
           "as_of": null
         }
       }
diff --git a/docs/user_guide/asapquery-profile.md b/docs/user_guide/asapquery-profile.md
index 3f0b4366..fba4e022 100644
--- a/docs/user_guide/asapquery-profile.md
+++ b/docs/user_guide/asapquery-profile.md
@@ -139,6 +139,10 @@ per-materialization state and observed coverage through
 
 Snapshot schema version `2` accepts fixed-interval repeating PromQL queries
 with `implementation.scrape_interval_ms` and fresh ingestion-rate evidence.
-Range selectors derive their own lookback; rangeless instant-vector queries use
-the scrape interval. Unsupported snapshot semantics fail startup rather than
-silently inventing cost or placement evidence.
+Range selectors derive their own lookback; each rangeless source uses the
+scrape interval for backend-local planning. This default window is distinct
+from Prometheus's instant-selector lookback delta and does not configure it.
+Omit `time_selection.lookback` or set it to `null`. Ranges and offsets must be
+whole seconds; unsupported fractional durations fail startup instead of
+silently shortening the window. Unsupported snapshot semantics fail startup
+rather than silently inventing cost or placement evidence.

From 2d176fa900cda1d60252026a1cd9efe09e85a699 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Sun, 13 Sep 2026 21:34:29 -0600
Subject: [PATCH 3/3] fix(discovery): emit derived-window planning snapshots

---
 control_plane/tests/discovery_snapshot.rs     | 54 +++++++++++++++++++
 tools/o11y-execution/discover_snapshot.py     | 23 ++++----
 .../o11y-execution/test_discover_snapshot.py  |  4 ++
 3 files changed, 68 insertions(+), 13 deletions(-)
 create mode 100644 control_plane/tests/discovery_snapshot.rs

diff --git a/control_plane/tests/discovery_snapshot.rs b/control_plane/tests/discovery_snapshot.rs
new file mode 100644
index 00000000..9e55a9cd
--- /dev/null
+++ b/control_plane/tests/discovery_snapshot.rs
@@ -0,0 +1,54 @@
+//! Discovery output must be accepted by the production snapshot planner.
+use control_plane::physical::compiler::BackendLocalPlanningSnapshot;
+use std::{fs, path::PathBuf, process::Command};
+
+struct TempDirectory(PathBuf);
+impl Drop for TempDirectory {
+    fn drop(&mut self) {
+        let _ = fs::remove_dir_all(&self.0);
+    }
+}
+
+#[test]
+fn discovered_snapshot_plans_with_observed_cadence_and_promql_history() {
+    let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        .parent()
+        .unwrap()
+        .to_path_buf();
+    let temporary = TempDirectory(
+        std::env::temp_dir().join(format!("asap-discovery-snapshot-{}", std::process::id())),
+    );
+    fs::create_dir(&temporary.0).unwrap();
+    let corpus = temporary.0.join("corpus.json");
+    let metrics = temporary.0.join("metrics.prom");
+    let output = temporary.0.join("snapshot.json");
+    fs::write(&corpus, r#"{"upstream_revision":"test","queries":[{"id":"q","query":"max_over_time(m[1m] offset 1h)","eval_timestamp_ms":120000}]}"#).unwrap();
+    fs::write(&metrics, "m{job=\"test\"} 1 60\nm{job=\"test\"} 2 120\n").unwrap();
+    let result = Command::new("python3")
+        .env("PYTHONDONTWRITEBYTECODE", "1")
+        .arg(root.join("tools/o11y-execution/discover_snapshot.py"))
+        .arg("--corpus")
+        .arg(&corpus)
+        .arg("--metrics")
+        .arg(&metrics)
+        .arg("--template")
+        .arg(root.join("docs/examples/asapquery-planning-snapshot.json"))
+        .arg("--output")
+        .arg(&output)
+        .args(["--repetitions", "1"])
+        .output()
+        .unwrap();
+    assert!(
+        result.status.success(),
+        "{}",
+        String::from_utf8_lossy(&result.stderr)
+    );
+    let snapshot: BackendLocalPlanningSnapshot =
+        serde_json::from_str(&fs::read_to_string(&output).unwrap()).unwrap();
+    assert_eq!(snapshot.implementation.scrape_interval_ms, 60_000);
+    let (request, _) = snapshot.clone().planning_request().unwrap();
+    assert_eq!(request.queries[0].window_secs, 3660);
+    let roundtrip: BackendLocalPlanningSnapshot =
+        serde_json::from_value(serde_json::to_value(&snapshot).unwrap()).unwrap();
+    assert_eq!(snapshot, roundtrip);
+}
diff --git a/tools/o11y-execution/discover_snapshot.py b/tools/o11y-execution/discover_snapshot.py
index ad72510c..7308b925 100644
--- a/tools/o11y-execution/discover_snapshot.py
+++ b/tools/o11y-execution/discover_snapshot.py
@@ -10,17 +10,11 @@
 import hashlib
 import json
 from pathlib import Path
-import re
 import time
 import math
 from replay import iter_samples
 
 
-def duration_ms(text):
-    units = {"ms": 1, "s": 1000, "m": 60000, "h": 3600000, "d": 86400000, "w": 604800000, "y": 31536000000}
-    return sum(int(n) * units[u] for n, u in re.findall(r"(\d+)(ms|[smhdwy])", text))
-
-
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--corpus", type=Path, required=True)
@@ -67,8 +61,6 @@ def evidence(value):
     registrations, query_audit = [], []
     frequencies = Counter(row["query"] for row in corpus["queries"])
     for query in dict.fromkeys(row["query"] for row in corpus["queries"]):
-        windows = [duration_ms(x) for x in re.findall(r"\[([0-9a-z]+)(?::[^\]]*)?\]", query)]
-        lookback = max(windows, default=300000)
         if args.interval_ms % frequencies[query]:
             raise ValueError("base interval must divide exactly by query occurrence frequency")
         interval = args.interval_ms // frequencies[query]
@@ -79,17 +71,22 @@ def evidence(value):
         registrations.append({"query": query, "demand": {"fixed_interval_at": {"interval": interval, "evaluation_phase": phase}},
                               "requirements": {"accuracy": {"explicit": "Exact"}, "response_latency": "unspecified"},
                               "predictability": {"predictable": {"known_at": None}},
-                              "time_selection": {"scope": "real_time", "lookback": lookback, "as_of": None}})
-        query_audit.append({"query": query, "window_lookback_ms": lookback,
+                              "time_selection": {"scope": "real_time", "lookback": None, "as_of": None}})
+        query_audit.append({"query": query,
                             "occurrence_count": frequencies[query], "expected_evaluations": frequencies[query] * args.repetitions,
                             "declared_interval_ms": interval, "evaluation_phase_ms": phase,
-                            "lookback_method": "largest explicit range; instant selector defaults to Prometheus 5m; original offsets/subqueries preserved in query"})
+                            "lookback_method": "derived by the backend compiler from PromQL and the declared scrape cadence"})
     snapshot["query_workload"].update(repeating_queries=registrations, data_workload=data, query_batch=None)
     snapshot["snapshot_version"] = 2
     snapshot.pop("workload_cost_evidence", None)
     implementation = snapshot["implementation"]
-    if source_sample_interval_ms:
-        implementation["source_sample_interval_ms"] = source_sample_interval_ms
+    implementation.pop("source_sample_interval_ms", None)
+    # Prefer the observed cadence; retain the declared template cadence when
+    # the input has no repeated series from which to infer it.
+    scrape_interval_ms = source_sample_interval_ms or implementation.get("scrape_interval_ms")
+    if not isinstance(scrape_interval_ms, int) or scrape_interval_ms <= 0 or scrape_interval_ms % 1000:
+        raise ValueError("scrape_interval_ms must be a positive whole number of seconds")
+    implementation["scrape_interval_ms"] = scrape_interval_ms
     # Finite replay evaluates the oldest repetition first after loading the
     # complete input. Preserve that admitted historical-query span separately
     # from each query's PromQL range selector.
diff --git a/tools/o11y-execution/test_discover_snapshot.py b/tools/o11y-execution/test_discover_snapshot.py
index 94497ec8..f195eb77 100644
--- a/tools/o11y-execution/test_discover_snapshot.py
+++ b/tools/o11y-execution/test_discover_snapshot.py
@@ -25,6 +25,10 @@ def test_historical_input_does_not_backdate_plan_activation(self):
             ], check=True)
             snapshot = json.loads(output.read_text())
             self.assertEqual(snapshot["query_workload"]["repeating_queries"][0]["demand"], {"fixed_interval_at": {"interval": 60000, "evaluation_phase": 0}})
+            # Discovery output must obey the same schema as the compiler input.
+            self.assertIsNone(snapshot["query_workload"]["repeating_queries"][0]["time_selection"].get("lookback"))
+            self.assertEqual(snapshot["implementation"]["scrape_interval_ms"], 60_000)
+            self.assertNotIn("source_sample_interval_ms", snapshot["implementation"])
             environment = snapshot["environment"]
             self.assertEqual(environment["activation_unix_ms"], environment["observed_at_unix_ms"])
             provenance = json.loads(output.with_suffix(".provenance.json").read_text())