feat(erp): publish bounded materialization observations after finite completion

control_plane/Cargo.toml

@@ -76,17 +76,17 @@ 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 = "0402384e589df6e087d6d2d22b463ddc2eea0774" }
-asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "0402384e589df6e087d6d2d22b463ddc2eea0774" }
+planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "5e9396033a5a5d5f350bfa675d12770c972ee991" }
+asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "5e9396033a5a5d5f350bfa675d12770c972ee991" }
 
 # 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 = "0402384e589df6e087d6d2d22b463ddc2eea0774" }
-asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "0402384e589df6e087d6d2d22b463ddc2eea0774" }
+asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "5e9396033a5a5d5f350bfa675d12770c972ee991" }
+asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "5e9396033a5a5d5f350bfa675d12770c972ee991" }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }

control_plane/examples/audit_clickhouse_corpus.rs

@@ -4,7 +4,7 @@ use asap_frontend_sql::SqlCatalog;
 use asap_types::{AggregationType, KeyByLabelNames, PrecomputeMaterialization, WindowKind};
 use control_plane::clickhouse::{ClickHouseSqlWorkload, ClickHouseSqlWorkloadEntry};
 use control_plane::physical::compiler::{
-    PlanEnvelope, PrecomputePlan, TransmissionPlan, BACKEND_COMPAT, PLANNER_REVISION,
+    PlanEnvelope, PrecomputePlan, BACKEND_COMPAT, PLANNER_REVISION,
 };
 use planner_types::pre_asap::{Column, DataType, Schema};
 use planner_types::types::AccuracyTarget;
@@ -56,8 +56,12 @@ fn publication_inputs(schema: &Schema, sql: String) -> ClickHouseSqlWorkload {
     let mut precompute_plan =
         PrecomputePlan::build_backend_local(envelope.clone(), vec![materialization.clone()])
             .unwrap();
-    let mut transmission_plan =
-        TransmissionPlan::build(envelope, &precompute_plan, &Default::default()).unwrap();
+    let mut transmission_plan = control_plane::physical::compiler::compile_transmission_plan(
+        envelope,
+        &precompute_plan,
+        &Default::default(),
+    )
+    .unwrap();
     let sds = asap_types::summary_catalog::SummaryCatalog::from_materializations(
         27,
         1,

control_plane/src/clickhouse.rs

@@ -183,7 +183,7 @@ pub async fn compile_automatic_clickhouse_workload(
             .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?,
     );
     precompute.executable_dags = installed_dags;
-    let mut transmission = TransmissionPlan::build(
+    let mut transmission = crate::physical::compiler::compile_transmission_plan(
         request.envelope.clone(),
         &precompute,
         &std::collections::BTreeMap::new(),
@@ -360,7 +360,7 @@ where
         .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?;
     let mut materialization_nodes = std::collections::BTreeMap::new();
     let mut query_nodes = std::collections::BTreeMap::new();
-    let executable = QueryPlanEntry::compile_bound_relational_mapped(
+    let executable = crate::query_plan::compile_bound_relational_mapped(
         query.sql.clone(),
         planned.canonical_sql.clone(),
         &root,
@@ -1010,9 +1010,12 @@ mod tests {
         let mut precompute =
             PrecomputePlan::build_backend_local(envelope.clone(), vec![config]).unwrap();
         precompute.summary_catalog = Some(sds.reference().unwrap());
-        let mut transmission =
-            TransmissionPlan::build(envelope, &precompute, &std::collections::BTreeMap::new())
-                .unwrap();
+        let mut transmission = crate::physical::compiler::compile_transmission_plan(
+            envelope,
+            &precompute,
+            &std::collections::BTreeMap::new(),
+        )
+        .unwrap();
         transmission.summary_catalog = Some(sds.reference().unwrap());
         let timestamped = |time_name: &str, value_name: &str| {
             Schema::with_time_index(
@@ -1204,7 +1207,7 @@ mod tests {
         request.precompute_plan =
             PrecomputePlan::build_backend_local(envelope.clone(), vec![config]).unwrap();
         request.precompute_plan.summary_catalog = Some(request.sds.reference().unwrap());
-        request.transmission_plan = TransmissionPlan::build(
+        request.transmission_plan = crate::physical::compiler::compile_transmission_plan(
             envelope,
             &request.precompute_plan,
             &std::collections::BTreeMap::new(),

control_plane/src/emit/backend_push.rs

@@ -260,7 +260,7 @@ async fn push_documents_coupled(
         warn!(%error, "failed to bind compatibility PrecomputePlan to SummaryCatalog");
         return (false, false, 0);
     }
-    let transmission_plan = match crate::physical::compiler::TransmissionPlan::build(
+    let transmission_plan = match crate::physical::compiler::compile_transmission_plan(
         precompute_plan.envelope.clone(),
         &precompute_plan,
         &Default::default(),

control_plane/src/main.rs

@@ -69,6 +69,10 @@ struct AppState {
     /// agents' `sketch-runtime::PushExporter`. Read by decision
     /// loops in the replanner.
     runtime_samples: Arc<runtime_samples::RuntimeSamplesStore>,
+    /// The successfully activated typed catalog is authoritative for live ERP
+    /// input identity; incoming telemetry cannot supply its own descriptors.
+    active_summary_catalog:
+        Arc<tokio::sync::Mutex<Option<Arc<asap_types::summary_catalog::SummaryCatalog>>>>,
     /// Phase C (MVP v6): shared `BackendClient` for posting
     /// `StreamingConfig` JSON / YAML to the ASAPQuery-backend's
     /// `POST /api/v1/streaming-config` endpoint. Phase B had this
@@ -487,6 +491,7 @@ async fn main() {
         opamp_endpoint: opamp_ep,
         workload_registry: Arc::clone(&workload_registry),
         runtime_samples: Arc::clone(&runtime_samples_store),
+        active_summary_catalog: Arc::new(tokio::sync::Mutex::new(None)),
         backend_client: backend_client_shared,
         backend_routing_cache: Arc::clone(&backend_routing_cache),
     };
@@ -611,6 +616,8 @@ struct PhysicalPlanQueryRequest {
 #[derive(Debug, Deserialize)]
 #[serde(deny_unknown_fields)]
 struct CompileAndPublishPhysicalPlanRequest {
+    #[serde(default = "default_physical_deployment_target")]
+    target: physical::compiler::PhysicalDeploymentTarget,
     #[serde(default)]
     workload_cost_evidence: Option<physical::workload_cost::WorkloadCostEvidence>,
     queries: Vec<PhysicalPlanQueryRequest>,
@@ -635,6 +642,10 @@ struct CompileAndPublishPhysicalPlanRequest {
     apply_timeout_ms: u64,
 }
 
+fn default_physical_deployment_target() -> physical::compiler::PhysicalDeploymentTarget {
+    physical::compiler::PhysicalDeploymentTarget::DistributedCollectors
+}
+
 fn default_physical_plan_timeout_ms() -> u64 {
     10_000
 }
@@ -704,10 +715,15 @@ async fn compile_and_publish_physical_plan(
     mut request: CompileAndPublishPhysicalPlanRequest,
     frontend: PhysicalQueryFrontend,
 ) -> Response {
+    // Serialize typed activations so an older response cannot overwrite the
+    // catalog recorded after a newer backend activation.
+    let mut active_catalog = st.active_summary_catalog.lock().await;
     if let Some(erp) = &mut request.erp {
         if let Err(error) = erp.hydrate_observed_shape(&st.runtime_samples) {
             return (StatusCode::UNPROCESSABLE_ENTITY, error).into_response();
         }
+        let catalog = active_catalog.clone();
+        erp.resolve_population_data_descriptor(catalog.as_deref());
     }
     let (bundle, collector_ids, apply_timeout, adaptation_evidence, _) =
         match compile_physical_plan_request(request, false, frontend) {
@@ -802,6 +818,8 @@ async fn compile_and_publish_physical_plan(
             .into_response();
     }
 
+    *active_catalog = Some(Arc::new(bundle.summary_catalog));
+
     Json(CompileAndPublishPhysicalPlanResponse {
         cost_comparison: bundle.cost_comparison,
         plan_id: bundle.envelope.plan_id,
@@ -842,6 +860,7 @@ async fn publish_clickhouse_plan(
     publication: physical::publication::PhysicalPlanPublication,
     selection_trace: Option<serde_json::Value>,
 ) -> axum::response::Response {
+    let mut active_catalog = state.active_summary_catalog.lock().await;
     let plan_id = publication.summary_catalog.plan_id;
     let plan_version = publication.summary_catalog.plan_version;
     let Some(client) = state.backend_client.as_ref() else {
@@ -863,6 +882,7 @@ async fn publish_clickhouse_plan(
             .await;
         return (StatusCode::BAD_GATEWAY, error.to_string()).into_response();
     }
+    *active_catalog = Some(Arc::new(publication.summary_catalog));
     Json(serde_json::json!({
         "plan_id": plan_id,
         "plan_version": plan_version,
@@ -888,10 +908,15 @@ fn compile_physical_plan_request(
     ),
     (StatusCode, String),
 > {
-    if request.queries.is_empty() || request.collector_ids.is_empty() {
+    if request.queries.is_empty()
+        || (request.target == physical::compiler::PhysicalDeploymentTarget::DistributedCollectors
+            && request.collector_ids.is_empty())
+        || (request.target == physical::compiler::PhysicalDeploymentTarget::BackendLocalRemoteWrite
+            && !request.collector_ids.is_empty())
+    {
         return Err((
             StatusCode::UNPROCESSABLE_ENTITY,
-            "queries and collector_ids must both be non-empty".to_string(),
+            "queries must be non-empty; distributed deployment requires collectors and backend-local deployment requires none".to_string(),
         ));
     }
     if request.max_evidence_age_ms == 0 || request.apply_timeout_ms == 0 {
@@ -967,7 +992,8 @@ fn compile_physical_plan_request(
     let planning_request = physical::compiler::PlanningRequest {
         query_workload: None,
         queries,
-        hybrid_execution: false,
+        hybrid_execution: request.target
+            == physical::compiler::PhysicalDeploymentTarget::BackendLocalRemoteWrite,
         materialization_policy: None,
         evidence: request.evidence,
         exact_composition_costs: request.exact_composition_costs,
@@ -978,7 +1004,7 @@ fn compile_physical_plan_request(
         retained_summary_memory_budget_bytes: None,
     };
     let environment = physical::compiler::DeploymentEnvironment {
-        target: physical::compiler::PhysicalDeploymentTarget::DistributedCollectors,
+        target: request.target,
         collector_ids: request.collector_ids.clone(),
         capability_snapshot_id: request.capability_snapshot_id,
         observed_at_unix_ms: now,
@@ -2242,6 +2268,7 @@ fn test_app_with_backend(backend_url: Option<String>) -> (AppState, axum::Router
         opamp_endpoint: "ws://ctrl:4320/v1/opamp".into(),
         workload_registry: Arc::new(WorkloadRegistry::empty()),
         runtime_samples: runtime_samples::RuntimeSamplesStore::new(64),
+        active_summary_catalog: Arc::new(tokio::sync::Mutex::new(None)),
         backend_client,
         backend_routing_cache: Arc::new(Mutex::new(HashMap::new())),
     };
@@ -2309,6 +2336,58 @@ mod api_tests {
         assert_eq!(manifests.as_array().unwrap().len(), 1);
     }
 
+    #[test]
+    fn backend_local_typed_request_compiles_without_collectors() {
+        let snapshot: physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_str(
+            include_str!("../../docs/examples/asapquery-compatibility-demo-snapshot.json"),
+        )
+        .unwrap();
+        let (planning, _) = snapshot.planning_request().unwrap();
+        let mut query = planning.queries[0].clone();
+        let now = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_millis() as u64;
+        query.lifecycle.evidence_observed_at_unix_ms = now;
+        for implementation in &mut query.window_implementations {
+            implementation.cost.observed_at_unix_ms = now;
+        }
+        let planner_types::pre_asap::Source::TimeSeries { metric } = &query.source else {
+            panic!("expected time series fixture");
+        };
+        let value = serde_json::json!({
+            "target": "backend_local_remote_write",
+            "queries": [{
+                "query_id": query.query_id, "query_string": query.query_string,
+                "metric": metric, "window_secs": query.window_secs, "accuracy": query.accuracy,
+                "lifecycle": query.lifecycle, "window_implementations": query.window_implementations
+            }],
+            "collector_ids": [], "capability_snapshot_id": "test",
+            "planner_revision": physical::compiler::PLANNER_REVISION,
+            "max_evidence_age_ms": 60000, "plan_version": 1,
+            "activation_unix_ms": now, "backend_compat": physical::compiler::BACKEND_COMPAT
+        });
+        let request = serde_json::from_value(value.clone()).unwrap();
+        let (plan, collectors, _, _, _) =
+            compile_physical_plan_request(request, false, PhysicalQueryFrontend::PromQl).unwrap();
+        let plan = plan.unwrap();
+        assert!(collectors.is_empty());
+        assert!(plan.collector_plans.is_empty());
+        assert_eq!(
+            plan.precompute_plan.ingest.protocol,
+            physical::compiler::IngestProtocol::PrometheusRemoteWriteV1
+        );
+        assert!(!plan.precompute_plan.materializations.is_empty());
+        let mut distributed = value;
+        distributed["target"] = serde_json::json!("distributed_collectors");
+        assert!(compile_physical_plan_request(
+            serde_json::from_value(distributed).unwrap(),
+            false,
+            PhysicalQueryFrontend::PromQl
+        )
+        .is_err());
+    }
+
     async fn body_json(resp: axum::response::Response) -> serde_json::Value {
         let bytes = resp.into_body().collect().await.unwrap().to_bytes();
         serde_json::from_slice(&bytes).unwrap()