fix(planner): make BackendPlan authoritative for warm execution

control_plane/proto/backend_plan.proto

@@ -168,6 +168,8 @@ message Materialization {
   ColumnRef col = 7;
   optional RetentionPolicy retention = 8;
   optional SummaryMaintenanceLifecycle lifecycle = 9;
+  // Canonical label matcher set (sorted `key="value"` entries).
+  string spatial_filter = 10;
 }
 
 message SummaryMaintenanceLifecycle {

control_plane/src/backend_client.rs

@@ -359,6 +359,47 @@ impl BackendClient {
             Err(classify_http_status(status, body, "BackendPlan POST"))
         }
     }
+
+    /// Publish the backend-facing portions of one PhysicalPlan in a single
+    /// request, preventing independently retried documents from mixing
+    /// generations at the backend.
+    pub async fn post_physical_plan_typed(
+        &self,
+        precompute_plan: &crate::physical::compiler::PrecomputePlan,
+        backend_plan: Vec<u8>,
+        storage_routing: Option<serde_json::Value>,
+    ) -> std::result::Result<(), BackendPostError> {
+        let url = derive_physical_plan_url(&self.endpoint);
+        let response = self
+            .http
+            .post(&url)
+            .json(&serde_json::json!({
+            "precompute_plan": precompute_plan,
+            "backend_plan": backend_plan,
+            "storage_routing": storage_routing,
+            }))
+            .send()
+            .await
+            .map_err(classify_reqwest_error)?;
+        let status = response.status();
+        if status.is_success() {
+            Ok(())
+        } else {
+            let body = response.text().await.unwrap_or_default();
+            Err(classify_http_status(status, body, "PhysicalPlan POST"))
+        }
+    }
+}
+
+fn derive_physical_plan_url(endpoint: &str) -> String {
+    const DASH: &str = "/api/v1/streaming-config";
+    const UNDERSCORE: &str = "/api/v1/streaming_config";
+    const PHYSICAL: &str = "/api/v1/physical-plan";
+    endpoint
+        .strip_suffix(DASH)
+        .or_else(|| endpoint.strip_suffix(UNDERSCORE))
+        .map(|base| format!("{base}{PHYSICAL}"))
+        .unwrap_or_else(|| endpoint.to_string())
 }
 
 /// Map a streaming-config endpoint URL to the sibling storage-routing

control_plane/src/backend_plan/from_stage_config.rs

@@ -48,8 +48,9 @@ pub fn from_stage_config(
         HashMap::with_capacity(cfg.aggregations.len());
 
     for agg in &cfg.aggregations {
-        let fingerprint = policy_fingerprint_for_aggregation(agg)
+        let fingerprint = aggregation_config_for_materialization(agg)
             .with_context(|| format!("aggregation_id {:?}", agg.aggregation_id))?;
+        let fingerprint = fingerprint.policy_fingerprint();
         fingerprint_by_agg_id.insert(agg.aggregation_id.as_str(), fingerprint);
 
         let (kind, params) = match &agg.agg_type_override {
@@ -71,6 +72,7 @@ pub fn from_stage_config(
                 },
                 group_by: agg.grouping.clone(),
                 rollup: Vec::new(),
+                spatial_filter: asap_types::utils::normalize_spatial_filter(&agg.spatial_filter),
                 kind,
                 params,
                 col: ColumnRef::SampleValue,
@@ -134,14 +136,15 @@ pub fn from_stage_config(
 /// parser the real `POST /api/v1/streaming-config` handler uses (which
 /// parses its body as YAML regardless of declared content-type, since
 /// JSON is valid YAML) — rather than re-deriving the field mapping here.
-fn policy_fingerprint_for_aggregation(agg: &BackendAggregation) -> Result<PolicyFingerprint> {
+pub fn aggregation_config_for_materialization(
+    agg: &BackendAggregation,
+) -> Result<AggregationConfig> {
     let json = build_backend_aggregation_json(agg);
     let text = serde_json::to_string(&json).context("serialize synthesized aggregation JSON")?;
     let yaml_value: serde_yaml::Value =
         serde_yaml::from_str(&text).context("parse synthesized aggregation JSON as YAML")?;
-    let cfg = AggregationConfig::from_yaml_data(&yaml_value, None, QueryLanguage::promql)
-        .context("build AggregationConfig from synthesized aggregation JSON")?;
-    Ok(cfg.policy_fingerprint())
+    AggregationConfig::from_yaml_data(&yaml_value, None, QueryLanguage::promql)
+        .context("build AggregationConfig from synthesized aggregation JSON")
 }
 
 /// Option B (post-#287) exact-agg override: `s` is already the wire

control_plane/src/backend_plan/mod.rs

@@ -28,6 +28,7 @@ pub mod proto {
 }
 
 mod from_stage_config;
+pub use from_stage_config::aggregation_config_for_materialization;
 pub use from_stage_config::from_stage_config;
 
 use std::collections::HashMap;
@@ -70,6 +71,12 @@ pub enum ValidationError {
     ZeroSlide { fingerprint: u64 },
     #[error("route references unknown materialization {fingerprint}")]
     UnknownMaterialization { fingerprint: u64 },
+    #[error("materialization {fingerprint} has mismatched kind/parameters")]
+    KindParamsMismatch { fingerprint: u64 },
+    #[error("route capability is incompatible with materialization {fingerprint}")]
+    IncompatibleRoute { fingerprint: u64 },
+    #[error("stale plan generation: incoming={incoming}, active={active}")]
+    StaleGeneration { incoming: u64, active: u64 },
 }
 
 // ── WindowSpec ───────────────────────────────────────────────────────────────
@@ -586,6 +593,7 @@ pub struct Materialization {
     pub window: WindowSpec,
     pub group_by: Vec<String>,
     pub rollup: Vec<String>,
+    pub spatial_filter: String,
     pub kind: SummaryKind,
     pub params: SummaryParams,
     pub col: ColumnRef,
@@ -631,6 +639,7 @@ impl From<&Materialization> for proto::Materialization {
             window: Some((&m.window).into()),
             group_by: m.group_by.clone(),
             rollup: m.rollup.clone(),
+            spatial_filter: m.spatial_filter.clone(),
             params: Some((&m.params).into()),
             col: Some((&m.col).into()),
             retention: m.retention.as_ref().map(Into::into),
@@ -674,6 +683,7 @@ impl TryFrom<proto::Materialization> for Materialization {
                 .try_into()?,
             group_by: m.group_by,
             rollup: m.rollup,
+            spatial_filter: m.spatial_filter,
             kind,
             params,
             col: m
@@ -811,18 +821,70 @@ impl BackendPlan {
             if materialization.window.slide_ms == Some(0) {
                 return Err(ValidationError::ZeroSlide { fingerprint: key.0 });
             }
+            if !kind_params_match(&materialization.kind, &materialization.params) {
+                return Err(ValidationError::KindParamsMismatch { fingerprint: key.0 });
+            }
         }
         for route in &self.routing {
-            if !self.materializations.contains_key(&route.materialization) {
+            let Some(materialization) = self.materializations.get(&route.materialization) else {
                 return Err(ValidationError::UnknownMaterialization {
                     fingerprint: route.materialization.0,
                 });
+            };
+            if !materialization_satisfies(&route.satisfies, materialization) {
+                return Err(ValidationError::IncompatibleRoute {
+                    fingerprint: route.materialization.0,
+                });
             }
         }
         Ok(())
     }
 }
 
+fn kind_params_match(kind: &SummaryKind, params: &SummaryParams) -> bool {
+    matches!(
+        (kind, params),
+        (SummaryKind::Sum, SummaryParams::Sum)
+            | (SummaryKind::Count, SummaryParams::Count)
+            | (SummaryKind::MinMax, SummaryParams::MinMax)
+            | (SummaryKind::Increase, SummaryParams::Increase)
+            | (SummaryKind::Rate, SummaryParams::Rate)
+            | (SummaryKind::Kll, SummaryParams::Kll { .. })
+            | (SummaryKind::Cms, SummaryParams::Cms { .. })
+            | (SummaryKind::Hll, SummaryParams::Hll { .. })
+            | (SummaryKind::DDSketch, SummaryParams::DDSketch { .. })
+            | (SummaryKind::CmsWithHeap, SummaryParams::CmsWithHeap { .. })
+            | (SummaryKind::Kmv, SummaryParams::Kmv { .. })
+            | (SummaryKind::Theta, SummaryParams::Theta { .. })
+            | (SummaryKind::CountSketch, SummaryParams::CountSketch { .. })
+            | (
+                SummaryKind::CountSketchWithHeap,
+                SummaryParams::CountSketchWithHeap { .. }
+            )
+    )
+}
+
+fn materialization_satisfies(required: &Capability, m: &Materialization) -> bool {
+    let available = match m.kind {
+        SummaryKind::DDSketch => Capability::QuantileApprox(SketchKindHandle::DDSketch),
+        SummaryKind::Kll => Capability::QuantileApprox(SketchKindHandle::Kll),
+        SummaryKind::Hll => Capability::CardinalityApprox,
+        SummaryKind::Cms => Capability::FrequencyEstimate(SketchKindHandle::CountMin),
+        SummaryKind::CountSketch => Capability::FrequencyEstimate(SketchKindHandle::CountSketch),
+        SummaryKind::CmsWithHeap => Capability::FrequencyTopk(SketchKindHandle::CmsWithHeap),
+        SummaryKind::CountSketchWithHeap => {
+            Capability::FrequencyTopk(SketchKindHandle::CountSketchWithHeap)
+        }
+        SummaryKind::Sum => Capability::ExactAgg(AggregationType::Sum),
+        SummaryKind::MinMax => Capability::ExactAgg(AggregationType::MinMax),
+        SummaryKind::Increase => Capability::ExactAgg(AggregationType::Increase),
+        SummaryKind::Count | SummaryKind::Rate | SummaryKind::Kmv | SummaryKind::Theta => {
+            return false
+        }
+    };
+    required.is_satisfied_by(&available)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -845,6 +907,7 @@ mod tests {
             },
             group_by: vec!["zone".to_string()],
             rollup: vec![],
+            spatial_filter: String::new(),
             kind,
             params,
             col: ColumnRef::SampleValue,