refactor: align backend summary types with ASAPPlanner

control_plane/src/emit/backend_push.rs

@@ -685,19 +685,25 @@ mod tests {
             AggregationInput, BackendAggregation, BackendReadout,
         };
         use planner_types::post_asap::SketchQuery;
-        use planner_types::post_asap::{SketchAlgorithm, SketchParams};
+        use planner_types::post_asap::{
+            GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams, SummaryFamilyType,
+        };
         BackendStageConfig {
             aggregations: vec![BackendAggregation {
                 aggregation_id: agg_id.to_string(),
                 metric_name: metric.to_string(),
-                sketch_kind: SketchAlgorithm::DDSketch.into(),
-                sketch_params: SketchParams::DDSketch { alpha: 0.01 }.into(),
+                family: SummaryFamilyType::Sketch(
+                    SketchKind::new(
+                        SketchAlgorithm::DDSketch,
+                        SketchParams::DDSketch { alpha: 0.01 },
+                    ),
+                    GroupingStrategy::PerSubpopulationInstance,
+                ),
                 grouping: vec![],
                 item_label: None,
                 spatial_filter: String::new(),
                 window_secs: 60,
                 aggregation_input: AggregationInput::SketchEnvelope,
-                agg_type_override: None,
             }],
             readouts: vec![BackendReadout {
                 aggregation_id: agg_id.to_string(),

control_plane/src/emit/mod.rs

@@ -258,7 +258,7 @@ fn apply_cold_format_from_env(edge_cfg: &mut EdgeStageConfig) {
 // HashMap — the typed bootstrap / replan paths emitted single-pipeline
 // YAML and the routing-connector path stayed dormant.
 //
-// `extract_root_sketch_kind` walks a `PhysicalExpr` tree and returns the
+// `extract_root_sketch_algorithm` walks a `PhysicalExpr` tree and returns the
 // committed sketch family — looking through `SketchEstimate`,
 // `SketchAgg`, `SketchMerge`, `LetBinding`, and `RawAtEdgeSketchAtBackend`.
 // `SketchAgg::sketch_type` is the canonical source of truth (the typed
@@ -284,7 +284,7 @@ fn apply_cold_format_from_env(edge_cfg: &mut EdgeStageConfig) {
 /// (`Logical`-only, unresolved `Ref`, raw Mode-3 archive). These map
 /// onto the raw-passthrough default pipeline in the routing emitter,
 /// which is correct.
-pub fn extract_root_sketch_kind(expr: &PhysicalExpr) -> Option<SketchAlgorithm> {
+pub fn extract_root_sketch_algorithm(expr: &PhysicalExpr) -> Option<SketchAlgorithm> {
     match expr {
         PhysicalExpr::Committed(plan) => extract_from_plan(plan),
         PhysicalExpr::RawAtEdgeSketchAtBackend { family, .. } => Some(family.clone()),
@@ -394,7 +394,7 @@ pub fn collect_metric_to_family(
             else {
                 continue;
             };
-            if let Some(kind) = extract_root_sketch_kind(&deployment_expr) {
+            if let Some(kind) = extract_root_sketch_algorithm(&deployment_expr) {
                 out.entry(entry.metric_name.clone())
                     .or_default()
                     .insert(kind);
@@ -772,7 +772,7 @@ mod runtime_tests {
             window_secs: Some(60),
             sketch_processors: vec![EdgeSketchProcessor {
                 processor_name: "ddsketch".to_string(),
-                sketch_kind: SketchAlgorithm::DDSketch,
+                sketch_algorithm: SketchAlgorithm::DDSketch,
                 sketch_params: SketchParams::DDSketch { alpha: 0.01 },
                 aggregation_id: "agg0".to_string(),
             }],

control_plane/src/emit/monitor.rs

@@ -16,43 +16,9 @@
 //! planner stage gains a monitor-intent slot. See
 //! `ASAPCollector/docs/continuous-monitoring-tumbling-cost-analysis.md`.
 
+pub use asap_types::MonitorFunctional as Functional;
 use serde_yaml::{Mapping, Value};
 
-/// Which additive readout to threshold. Mirrors the Go `monitor.Functional`
-/// and the edge `ThresholdConfig.functional` string values.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum Functional {
-    Sum,
-    CmsPoint,
-    LinearBuckets,
-    /// Whole-sketch second frequency moment F2 = ‖f‖₂². Monitors the entire
-    /// sketch's L2 mass (no per-point key) so any future point query stays within
-    /// ε — see `data_plane::monitor` module docs for the whole-sketch-vs-point
-    /// decision rule. The edge reports its local F2 = Σ_x f_i(x)² as the value.
-    F2,
-}
-
-impl Functional {
-    pub fn as_str(self) -> &'static str {
-        match self {
-            Functional::Sum => "sum",
-            Functional::CmsPoint => "cms_point",
-            Functional::LinearBuckets => "linear_buckets",
-            Functional::F2 => "f2",
-        }
-    }
-
-    /// Parse a functional name (workload-spec value); unknown/empty → Sum.
-    pub fn from_name(s: &str) -> Functional {
-        match s {
-            "cms_point" => Functional::CmsPoint,
-            "linear_buckets" => Functional::LinearBuckets,
-            "f2" | "l2" => Functional::F2,
-            _ => Functional::Sum,
-        }
-    }
-}
-
 /// One monitored standing-query intent: "alert when the global Σ of `metric`'s
 /// `functional` crosses `tau`". τ/ε/window are authoritative at the coordinator;
 /// the edge copies are advisory.

control_plane/src/emit/otap.rs

@@ -295,7 +295,7 @@ fn build_asap_sketches_config(sp: &EdgeSketchProcessor, window_secs: Option<u64>
     // the DAG walk; it just doesn't reach the wire here.
     m.insert(
         "sketch_kind".into(),
-        Value::String(sketch_kind_tag(&sp.sketch_kind).into()),
+        Value::String(sketch_algorithm_tag(&sp.sketch_algorithm).into()),
     );
     match &sp.sketch_params {
         SketchParams::Kll { k } => {
@@ -335,7 +335,7 @@ fn build_asap_sketches_config(sp: &EdgeSketchProcessor, window_secs: Option<u64>
     Value::Mapping(m)
 }
 
-fn sketch_kind_tag(kind: &SketchAlgorithm) -> &'static str {
+fn sketch_algorithm_tag(kind: &SketchAlgorithm) -> &'static str {
     match kind {
         SketchAlgorithm::Kll => "kll",
         SketchAlgorithm::DDSketch => "ddsketch",
@@ -404,7 +404,7 @@ mod tests {
             window_secs: Some(60),
             sketch_processors: vec![EdgeSketchProcessor {
                 processor_name: "ddsketch".to_string(),
-                sketch_kind: SketchAlgorithm::DDSketch,
+                sketch_algorithm: SketchAlgorithm::DDSketch,
                 sketch_params: SketchParams::DDSketch { alpha: 0.01 },
                 aggregation_id: "agg0".to_string(),
             }],