fix(control-plane): derive lookback from PromQL

control_plane/src/main.rs

@@ -600,7 +600,7 @@ 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,
+        scrape_interval_ms: None,
         query_retention_margin_ms: 0,
         retained_summary_memory_budget_bytes: None,
     };

control_plane/src/physical/maintained_population.rs

@@ -89,7 +89,7 @@ pub(super) fn operator(
         max_bytes,
         max_series: 100_000.min((max_bytes / 1024) as usize),
         max_input_lag_ms: request
-            .source_sample_interval_ms
+            .scrape_interval_ms
             .unwrap_or(60_000)
             .saturating_add(request.query_retention_margin_ms)
             .clamp(1, 300_000),

control_plane/src/physical/workload_cost.rs

@@ -1101,7 +1101,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(),

control_plane/src/query_plan/residual.rs

@@ -575,19 +575,6 @@ mod planner_workload_tests {
     use super::*;
     use crate::physical::compiler::{BackendLocalPlanningInput, PhysicalPlanCompiler};
 
-    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::CompiledPhysicalPlan {
         let mut fixture: serde_json::Value = serde_json::from_str(include_str!(
             "../../../docs/examples/asapquery-planning-snapshot.json"
@@ -596,8 +583,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: BackendLocalPlanningInput = serde_json::from_value(fixture).unwrap();
         let (request, environment) = snapshot
@@ -683,9 +668,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();

control_plane/tests/discovery_snapshot.rs

@@ -0,0 +1,58 @@
+//! Discovery output must be accepted by the production snapshot planner.
+use control_plane::physical::compiler::BackendLocalPlanningInput;
+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: BackendLocalPlanningInput =
+        serde_json::from_str(&fs::read_to_string(&output).unwrap()).unwrap();
+    assert_eq!(snapshot.physical_inputs.scrape_interval_ms, 60_000);
+    let (request, _) = snapshot
+        .clone()
+        .into_physical_compilation_request()
+        .unwrap();
+    assert_eq!(request.scrape_interval_ms, Some(60_000));
+    assert_eq!(request.queries[0].query_lookback_seconds, 3660);
+    let roundtrip: BackendLocalPlanningInput =
+        serde_json::from_value(serde_json::to_value(&snapshot).unwrap()).unwrap();
+    assert_eq!(snapshot, roundtrip);
+}

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::BackendLocalPlanningInput =
             serde_json::from_value(wire).unwrap();

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::BackendLocalPlanningInput =
             serde_json::from_value(fixture).unwrap();

data_plane/tests/asapquery_compatibility_process_e2e.rs

@@ -801,9 +801,6 @@ async fn run_shared_dashboard(multi_pane: bool) {
     let mut typed: control_plane::physical::compiler::BackendLocalPlanningInput =
         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),

data_plane/tests/support/current_series_process.rs

@@ -26,7 +26,7 @@ async fn current_series_quantiles_topk_share_and_replace_values() {
     ))
     .unwrap();
     snapshot.schema_version = 2;
-    snapshot.physical_inputs.source_sample_interval_ms = Some(60_000);
+    snapshot.physical_inputs.scrape_interval_ms = 60_000;
     let template = snapshot.query_workload.repeating_queries.as_ref().unwrap()[0].clone();
     let mut queries = vec![];
     for q in [0.5, 0.9, 0.95, 0.99] {

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::BackendLocalPlanningInput =
         serde_json::from_value(fixture.clone()).unwrap();

data_plane/tests/support/issue_701_702_process.rs

@@ -121,15 +121,14 @@ async fn issue_workloads_execute_warm_at_successive_evaluations() {
         "../../../docs/examples/asapquery-planning-snapshot.json"
     ))
     .unwrap();
-    fixture["implementation"]["source_sample_interval_ms"] = 1000.into();
+    fixture["implementation"]["scrape_interval_ms"] = 1000.into();
     fixture["implementation"]["horizon_seconds"] = 3600.into();
     let template = fixture["query_workload"]["repeating_queries"][0].clone();
     fixture["query_workload"]["repeating_queries"] = queries
         .iter()
-        .map(|(query, lookback, cadence)| {
+        .map(|(query, _lookback, cadence)| {
             let mut entry = template.clone();
             entry["query"] = query.clone().into();
-            entry["time_selection"]["lookback"] = (lookback * 1000).into();
             entry["demand"]["fixed_interval_at"]["interval"] = (cadence * 1000).into();
             if !query.contains("quantile") {
                 entry["requirements"]["accuracy"] = serde_json::json!({"explicit":"Exact"});
@@ -359,13 +358,12 @@ async fn temporal_average_overflow_falls_back_after_state_is_warm() {
         "../../../docs/examples/asapquery-planning-snapshot.json"
     ))
     .unwrap();
-    fixture["implementation"]["source_sample_interval_ms"] = 1000.into();
+    fixture["implementation"]["scrape_interval_ms"] = 1000.into();
     let template = fixture["query_workload"]["repeating_queries"][0].clone();
     fixture["query_workload"]["repeating_queries"] = ["avg", "sum", "count"]
         .map(|op| {
             let mut entry = template.clone();
             entry["query"] = format!("{op}_over_time(average_overflow[5s])").into();
-            entry["time_selection"]["lookback"] = 5000.into();
             entry["demand"]["fixed_interval_at"]["interval"] = 1000.into();
             entry["requirements"]["accuracy"] = serde_json::json!({"explicit":"Exact"});
             entry

docs/developer_docs/planning/repeated-dashboard-panes.md

@@ -25,6 +25,28 @@ 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; 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.
+
 ## Cases
 
 All ranges below use PromQL's `(start,end]` convention. W is the lookback and E

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", "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", "lookback": 5000, "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", "lookback": 5000, "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", "lookback": 5000, "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", "lookback": 5000, "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", "lookback": 5000, "as_of": null }
+        "time_selection": { "scope": "real_time", "lookback": null, "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,

docs/examples/asapquery-planning-snapshot.json

@@ -18,7 +18,7 @@
         "predictability": { "predictable": { "known_at": null } },
         "time_selection": {
           "scope": "real_time",
-          "lookback": 60000,
+          "lookback": null,
           "as_of": null
         }
       }
@@ -64,7 +64,8 @@
         "source_scan_bytes": 0,
         "weighted_cost": 1.0
       }
-    }
+    },
+    "scrape_interval_ms": 5000
   },
   "environment": {
     "target": "backend_local_remote_write",

docs/user_guide/asapquery-profile.md

@@ -137,7 +137,12 @@ 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; 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.

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.