feat(transport): add TransmissionPlan and summary frame identity

control_plane/src/backend_client.rs

@@ -366,6 +366,7 @@ impl BackendClient {
     pub async fn post_physical_plan_typed(
         &self,
         precompute_plan: &crate::physical::compiler::PrecomputePlan,
+        transmission_plan: &crate::physical::compiler::TransmissionPlan,
         backend_plan: Vec<u8>,
         query_plan: &crate::query_plan::QueryPlan,
         storage_routing: Option<serde_json::Value>,
@@ -376,6 +377,7 @@ impl BackendClient {
             .post(&url)
             .json(&serde_json::json!({
             "precompute_plan": precompute_plan,
+            "transmission_plan": transmission_plan,
             "backend_plan": backend_plan,
             "query_plan": query_plan,
             "storage_routing": storage_routing,

control_plane/src/emit/backend_push.rs

@@ -252,11 +252,22 @@ async fn push_documents_coupled(
             .unwrap_or_default(),
         entries: Default::default(),
     };
+    let transmission_plan = match crate::physical::compiler::TransmissionPlan::build(
+        precompute_plan.envelope.clone(),
+        precompute_plan,
+    ) {
+        Ok(plan) => plan,
+        Err(error) => {
+            warn!(%error, "failed to build compatibility TransmissionPlan");
+            return (false, false, 0);
+        }
+    };
 
     for attempt in 1..=RETRY_MAX_ATTEMPTS {
         match client
             .post_physical_plan_typed(
                 precompute_plan,
+                &transmission_plan,
                 plan_bytes.clone(),
                 &query_plan,
                 Some(routing.clone()),

control_plane/src/main.rs

@@ -650,6 +650,7 @@ async fn handle_compile_and_publish_physical_plan(
     if let Err(error) = backend
         .post_physical_plan_typed(
             &bundle.precompute_plan,
+            &bundle.transmission_plan,
             bundle.backend_plan.encode_to_vec(),
             &bundle.query_plan,
             None,

control_plane/src/opamp/mod.rs

@@ -1035,6 +1035,16 @@ mod tests {
                     output_representation: "summary_state".into(),
                 },
             }],
+            transmission_rules: vec![crate::physical::compiler::TransmissionRule {
+                materialization: asap_types::PolicyFingerprint(1),
+                producer_id: collector_id.into(),
+                schema_id: "asap-query-backend.v1:summary-state:v1:1".into(),
+                mode: crate::physical::compiler::TransmissionMode::Full,
+                encoding: crate::physical::compiler::StateEncoding::SketchlibProtobufV1,
+                emit_every_ms: 60_000,
+                full_checkpoint_every_ms: None,
+                destination_ref: "asapquery-backend".into(),
+            }],
         }
     }