feat(emit): thread workload.group_by_labels into streaming-config labels.grouping

control_plane/src/emit/stage_config.rs

@@ -1395,7 +1395,7 @@ fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonValue {
         "aggregationSubType": "",
         "metric": agg.metric_name,
         "labels": {
-            "grouping": Vec::<String>::new(),
+            "grouping": agg.grouping,
             "rollup": Vec::<String>::new(),
             "aggregated": Vec::<String>::new(),
         },
@@ -1744,6 +1744,7 @@ mod tests {
                     sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }),
                     window_secs: 60,
                     spatial_filter: String::new(),
+                    grouping: Vec::new(),
                     aggregation_input: AggregationInput::SketchEnvelope,
                 },
                 BackendAggregation {
@@ -1753,6 +1754,7 @@ mod tests {
                     sketch_params: SketchParams::Hll(HllParams { precision: 14 }),
                     window_secs: 60,
                     spatial_filter: String::new(),
+                    grouping: Vec::new(),
                     aggregation_input: AggregationInput::SketchEnvelope,
                 },
             ],
@@ -1808,6 +1810,7 @@ mod tests {
                     }),
                     window_secs: 60,
                     spatial_filter: String::new(),
+                    grouping: Vec::new(),
                     aggregation_input: AggregationInput::SketchEnvelope,
                 },
                 BackendAggregation {
@@ -1817,6 +1820,7 @@ mod tests {
                     sketch_params: SketchParams::Cms(CmsParams { w: 4096, d: 4 }),
                     window_secs: 60,
                     spatial_filter: String::new(),
+                    grouping: Vec::new(),
                     aggregation_input: AggregationInput::SketchEnvelope,
                 },
             ],
@@ -1880,6 +1884,7 @@ mod tests {
                 sketch_params: params,
                 window_secs: 60,
                 spatial_filter: String::new(),
+                grouping: Vec::new(),
                 aggregation_input: AggregationInput::SketchEnvelope,
             }],
             readouts: vec![BackendReadout {
@@ -2225,6 +2230,48 @@ mod tests {
         }
     }
 
+    /// Grouping labels surface under `labels.grouping` in the emitted
+    /// JSON. The L5 emitter itself leaves the list empty; `handle_plan`
+    /// patches it from `workload.group_by_labels` before posting, so
+    /// here we simulate that by setting `grouping` on the
+    /// `BackendAggregation` directly and assert the JSON round-trips.
+    #[test]
+    fn backend_json_emits_grouping_under_labels() {
+        let cfg = BackendStageConfig {
+            aggregations: vec![BackendAggregation {
+                aggregation_id: "agg0".into(),
+                metric_name: "http_latency_ms".into(),
+                sketch_kind: SketchKind::DDSketch,
+                sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }),
+                window_secs: 30,
+                spatial_filter: String::new(),
+                grouping: vec!["zone".into(), "service".into()],
+                aggregation_input: AggregationInput::SketchEnvelope,
+            }],
+            readouts: vec![],
+        };
+        let v = emit_backend_streaming_config_json(&cfg).expect("emit ok");
+        let grouping = v["aggregations"][0]["labels"]["grouping"]
+            .as_array()
+            .expect("grouping array");
+        let names: Vec<&str> = grouping.iter().filter_map(|s| s.as_str()).collect();
+        assert_eq!(
+            names,
+            vec!["zone", "service"],
+            "labels.grouping must surface BackendAggregation.grouping verbatim\n{v}"
+        );
+        // The other label lists stay empty — the L5 controller doesn't
+        // yet emit rollup / aggregated.
+        assert!(v["aggregations"][0]["labels"]["rollup"]
+            .as_array()
+            .unwrap()
+            .is_empty());
+        assert!(v["aggregations"][0]["labels"]["aggregated"]
+            .as_array()
+            .unwrap()
+            .is_empty());
+    }
+
     /// Snapshot: aggregations + readouts together exhibit the
     /// id-aliasing the backend uses to wire readouts back to their
     /// producing aggregation. Pins the sort order + key names. Phase β
@@ -2242,6 +2289,7 @@ mod tests {
                 sketch_params: SketchParams::Kll(KllParams { k: 200 }),
                 window_secs: 60,
                 spatial_filter: String::new(),
+                grouping: Vec::new(),
                 aggregation_input: AggregationInput::SketchEnvelope,
             }],
             readouts: vec![BackendReadout {
@@ -2284,6 +2332,7 @@ mod tests {
                 sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }),
                 window_secs: 60,
                 spatial_filter: String::new(),
+                grouping: Vec::new(),
                 aggregation_input: AggregationInput::SketchEnvelope,
             }],
             readouts: vec![],
@@ -2306,6 +2355,7 @@ mod tests {
                 sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }),
                 window_secs: 60,
                 spatial_filter: String::new(),
+                grouping: Vec::new(),
                 aggregation_input: AggregationInput::Raw,
             }],
             readouts: vec![],

control_plane/src/emit/trait_def.rs

@@ -223,6 +223,7 @@ mod tests {
                 sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }),
                 window_secs: 60,
                 spatial_filter: String::new(),
+                grouping: Vec::new(),
                 aggregation_input:
                     crate::physical::colored_dag::emitter::AggregationInput::SketchEnvelope,
             }],

control_plane/src/main.rs

@@ -558,7 +558,25 @@ async fn handle_plan(
                                 Err(e) => warn!(error = %e, "emit_gateway_yaml failed"),
                             }
                         }
-                        crate::physical::colored_dag::StageConfig::Backend(be) => {
+                        crate::physical::colored_dag::StageConfig::Backend(mut be) => {
+                            // Patch grouping label names from the
+                            // workload spec. The typed L5 emitter
+                            // produces `BackendAggregation` with
+                            // `grouping: vec![]` because the canonical
+                            // L3 `QueryExpr::Aggregate.by` is
+                            // positional `ColumnId`s against a
+                            // synthesized schema that has no label
+                            // columns (open-set label naming is a
+                            // Step γ TODO in
+                            // `intent_algebra::column_resolution`).
+                            // `QueryWorkload.group_by_labels` carries
+                            // the names unambiguously, so we patch
+                            // them in here — every aggregation under
+                            // the same workload shares the same
+                            // grouping today.
+                            for agg in &mut be.aggregations {
+                                agg.grouping = workload.group_by_labels.clone();
+                            }
                             // Phase C: post the typed L5 streaming-config
                             // JSON to ASAPQuery-backend via the shared
                             // BackendClient when configured. Without a

control_plane/src/physical/colored_dag/emitter.rs

@@ -311,6 +311,20 @@ pub struct BackendAggregation {
     /// `label_filters`). Empty string when no filter applies.
     #[serde(default)]
     pub spatial_filter: String,
+    /// Group-by label names — keys in `labels.grouping` on the backend
+    /// side, where the precompute engine's accumulator pipeline keys
+    /// its per-aggregation state by the projected attribute set.
+    ///
+    /// The L5 emitter populates this empty (`vec![]`); the caller
+    /// (`handle_plan`) patches it from `workload.group_by_labels`
+    /// before posting the streaming-config JSON. The canonical L3
+    /// `QueryExpr::Aggregate.by` carries the keys as positional
+    /// `ColumnId`s against a synthesized schema that has no label
+    /// columns (open-set label naming is a Step γ TODO in
+    /// `intent_algebra::column_resolution`), so the workload-spec
+    /// strings are the only reliable source of the names today.
+    #[serde(default)]
+    pub grouping: Vec<String>,
     /// Phase ε.1 — what shape the backend ingests for this
     /// aggregation. Mode 1 (sketch at edge) / sketch_envelope is the
     /// default (the wire payload is a sketch state already). Mode 2
@@ -467,6 +481,10 @@ impl Emitter for ThreeStageEmitter {
                         sketch_params: params.clone(),
                         window_secs: edge.window_secs.unwrap_or(0),
                         spatial_filter: spatial_filter_from_label_filters(&edge.label_filters),
+                        // Populated post-emit by the caller (handle_plan)
+                        // from workload.group_by_labels — see the
+                        // struct doc-comment for the rationale.
+                        grouping: Vec::new(),
                         // Mode 1 — sketch built at edge, ships envelope.
                         aggregation_input: AggregationInput::SketchEnvelope,
                     });
@@ -544,6 +562,7 @@ impl Emitter for ThreeStageEmitter {
                         sketch_params: params.clone(),
                         window_secs: edge.window_secs.unwrap_or(0),
                         spatial_filter: spatial_filter_from_label_filters(&edge.label_filters),
+                        grouping: Vec::new(),
                         // Mode 2 — backend builds sketch from raw OTLP.
                         aggregation_input: AggregationInput::Raw,
                     });