feat(query): execute authoritative QueryPlan

control_plane/Cargo.toml

@@ -93,16 +93,16 @@ 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 = "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/ProjectASAP/ASAPPlanner", rev = "5d0b6f6edcac65edc89a72051f37977ab0c83031" }
-asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "5d0b6f6edcac65edc89a72051f37977ab0c83031" }
+planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "739753e33e096c01faccca8e7a1e3da5ad3aab9c" }
+asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "739753e33e096c01faccca8e7a1e3da5ad3aab9c" }
 
 # L1 adoption (design-target-architecture.md Part B): the PromQL front
 # end itself, replacing control_plane's own query_parser/promql.rs.
 # Pinned via `rev`, not a floating branch reference. Same rev as
 # `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 = "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/ProjectASAP/ASAPPlanner", rev = "5d0b6f6edcac65edc89a72051f37977ab0c83031" }
+asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "739753e33e096c01faccca8e7a1e3da5ad3aab9c" }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }

control_plane/src/backend_client.rs

@@ -367,6 +367,7 @@ impl BackendClient {
         &self,
         precompute_plan: &crate::physical::compiler::PrecomputePlan,
         backend_plan: Vec<u8>,
+        query_plan: &crate::query_plan::QueryPlan,
         storage_routing: Option<serde_json::Value>,
     ) -> std::result::Result<(), BackendPostError> {
         let url = derive_physical_plan_url(&self.endpoint);
@@ -376,6 +377,7 @@ impl BackendClient {
             .json(&serde_json::json!({
             "precompute_plan": precompute_plan,
             "backend_plan": backend_plan,
+            "query_plan": query_plan,
             "storage_routing": storage_routing,
             }))
             .send()

control_plane/src/emit/backend_push.rs

@@ -240,10 +240,24 @@ async fn push_documents_coupled(
             return (false, false, 0);
         }
     };
+    // The compatibility emitter has no Planner-selected query catalog.  It
+    // may still install producer/storage state, but publishes an empty
+    // QueryPlan so every serving request fails closed to the exact tier.
+    let query_plan = crate::query_plan::QueryPlan {
+        plan_id: crate::backend_plan::BackendPlan::decode(&plan_bytes)
+            .map(|plan| plan.plan_id)
+            .unwrap_or_default(),
+        entries: Default::default(),
+    };
 
     for attempt in 1..=RETRY_MAX_ATTEMPTS {
         match client
-            .post_physical_plan_typed(precompute_plan, plan_bytes.clone(), Some(routing.clone()))
+            .post_physical_plan_typed(
+                precompute_plan,
+                plan_bytes.clone(),
+                &query_plan,
+                Some(routing.clone()),
+            )
             .await
         {
             Ok(()) => return (true, true, attempt),

control_plane/src/lib.rs

@@ -68,6 +68,7 @@ pub mod physical;
 pub mod pipeline;
 pub mod planner_selection;
 pub mod query_parser;
+pub mod query_plan;
 pub mod query_planning;
 pub mod replan;
 pub mod runtime_samples;

control_plane/src/main.rs

@@ -582,6 +582,7 @@ struct PhysicalPlanQueryRequest {
     group_by: Vec<String>,
     accuracy: types_v2::AccuracyTarget,
     lifecycle: physical::compiler::LifecyclePlanningInput,
+    window_implementations: Vec<physical::compiler::WindowImplementationCandidate>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -644,6 +645,7 @@ async fn handle_compile_and_publish_physical_plan(
         .post_physical_plan_typed(
             &bundle.precompute_plan,
             bundle.backend_plan.encode_to_vec(),
+            &bundle.query_plan,
             None,
         )
         .await
@@ -722,6 +724,7 @@ fn compile_physical_plan_request(
         };
         queries.push(physical::compiler::PlanningQuery {
             query_id: query.query_id,
+            query_string: query.query_string,
             post_asap,
             source: planner_types::pre_asap::Source::TimeSeries {
                 metric: query.metric,
@@ -730,6 +733,7 @@ fn compile_physical_plan_request(
             group_by: query.group_by,
             accuracy: query.accuracy,
             lifecycle: query.lifecycle,
+            window_implementations: query.window_implementations,
         });
     }
 

control_plane/src/opamp/mod.rs

@@ -1000,6 +1000,11 @@ mod tests {
                 parameters: serde_json::json!({"precision": 14}),
                 group_by: vec!["service".into()],
                 window_secs: 60,
+                abstract_window_framework:
+                    planner_types::post_asap::SummaryWindowFramework::Tumbling,
+                window_implementation_id: "collector-tumbling-v1".into(),
+                pane_secs: 60,
+                state_layout: "anchored-pane-v1".into(),
                 evidence_source: None,
                 lifecycle: crate::physical::compiler::CollectorLifecycle {
                     kind: "continuously_maintained".into(),