feat(control_plane): adopt asap_sketch::L4Node for the L4 IR (Step B)

control_plane/Cargo.toml

@@ -61,10 +61,10 @@ asap_types.workspace = true
 # this doesn't pull in datafusion or any front-end weight. asap-sketch is
 # asap-plan's own dependency (SummaryKind/SummaryParams), needed here only
 # to translate Implementation into this repo's own Capability vocabulary.
-asap-ir = { git = "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/ProjectASAP/ASAPController", rev = "7fcaf914d87e71407c3a6d7ccac613b867f9c11b" }
-asap-l2 = { git = "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/ProjectASAP/ASAPController", rev = "7fcaf914d87e71407c3a6d7ccac613b867f9c11b" }
-asap-sketch = { git = "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/ProjectASAP/ASAPController", rev = "7fcaf914d87e71407c3a6d7ccac613b867f9c11b" }
-asap-plan = { git = "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/ProjectASAP/ASAPController", rev = "7fcaf914d87e71407c3a6d7ccac613b867f9c11b" }
+asap-ir = { git = "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/ProjectASAP/ASAPController", rev = "01745cceac857be21fd1d80584c045e6f932ebfc" }
+asap-l2 = { git = "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/ProjectASAP/ASAPController", rev = "01745cceac857be21fd1d80584c045e6f932ebfc" }
+asap-sketch = { git = "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/ProjectASAP/ASAPController", rev = "01745cceac857be21fd1d80584c045e6f932ebfc" }
+asap-plan = { git = "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/ProjectASAP/ASAPController", rev = "01745cceac857be21fd1d80584c045e6f932ebfc" }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }

control_plane/src/emit/backend_push.rs

@@ -658,7 +658,7 @@ mod tests {
         use crate::physical::colored_dag::emitter::{
             AggregationInput, BackendAggregation, BackendReadout,
         };
-        use crate::sketch_algebra::physical_expr::EstimateOp;
+        use asap_sketch::SketchQuery;
         use asap_sketch::{SummaryKind, SummaryParams};
         BackendStageConfig {
             aggregations: vec![BackendAggregation {
@@ -675,7 +675,7 @@ mod tests {
             }],
             readouts: vec![BackendReadout {
                 aggregation_id: agg_id.to_string(),
-                op: EstimateOp::Quantile { q: 0.99 },
+                op: SketchQuery::Quantile { q: 0.99 },
             }],
         }
     }

control_plane/src/emit/mod.rs

@@ -51,10 +51,12 @@ pub use trait_def::{
 pub use crate::workload::WorkloadRegistry;
 
 use crate::physical::colored_dag::emitter::EdgeStageConfig;
+use crate::sketch_algebra::physical_expr::L4Plan;
 use crate::sketch_algebra::PhysicalExpr;
 use crate::store::WorkloadStore;
 use anyhow::Result;
-use asap_sketch::SummaryKind;
+use asap_sketch::{L4Node, SummaryExpr, SummaryKind};
+use std::rc::Rc;
 
 /// Phase ε.1.5 — which edge runtime an agent identifies as.
 ///
@@ -292,24 +294,55 @@ fn apply_cold_format_from_env(edge_cfg: &mut EdgeStageConfig) {
 /// which is correct.
 pub fn extract_root_sketch_kind(expr: &PhysicalExpr) -> Option<SummaryKind> {
     match expr {
-        PhysicalExpr::SketchAgg { sketch_type, .. } => Some(sketch_type.clone()),
+        PhysicalExpr::Committed(plan) => extract_from_plan(plan),
         PhysicalExpr::RawAtEdgeSketchAtBackend { family, .. } => Some(family.clone()),
-        PhysicalExpr::SketchEstimate { child, .. } => extract_root_sketch_kind(child),
-        PhysicalExpr::SketchMerge { children, .. } => {
-            children.iter().find_map(extract_root_sketch_kind)
+        PhysicalExpr::RawAtEdgePrometheusArchive { .. } => None,
+    }
+}
+
+fn extract_from_plan(plan: &L4Plan) -> Option<SummaryKind> {
+    match plan {
+        L4Plan::Summary(node) => extract_from_node(node),
+        L4Plan::LetBinding { expr, child, .. } => {
+            extract_from_plan(expr).or_else(|| extract_from_plan(child))
         }
-        PhysicalExpr::LetBinding { expr, child, .. } => {
-            extract_root_sketch_kind(expr).or_else(|| extract_root_sketch_kind(child))
+        L4Plan::Ref { .. } => None,
+    }
+}
+
+/// Is `kind` an exact accumulator (Sum/Count/MinMax/Increase/Rate) rather
+/// than an approximate sketch? Exact accumulators have no sketch family
+/// for the 5-sketch routing connector to route on — same as the old,
+/// now-retired `PhysicalExpr::ExactAgg` variant, which this function
+/// treated as `None`.
+fn is_exact_accumulator(kind: &SummaryKind) -> bool {
+    matches!(
+        kind,
+        SummaryKind::Sum
+            | SummaryKind::Count
+            | SummaryKind::MinMax
+            | SummaryKind::Increase
+            | SummaryKind::Rate
+    )
+}
+
+fn extract_from_node(node: &Rc<L4Node>) -> Option<SummaryKind> {
+    match &node.expr {
+        SummaryExpr::SummaryAgg { sketch, .. } if !is_exact_accumulator(sketch) => {
+            Some(sketch.clone())
         }
-        PhysicalExpr::Logical(_)
-        | PhysicalExpr::Ref { .. }
-        | PhysicalExpr::RawAtEdgePrometheusArchive { .. }
-        // ExactAgg has no sketch family — it produces an exact
-        // aggregation accumulator, not a sketch state. The
-        // routing emitter routes these to the
-        // `metrics/raw_passthrough` / exact-precompute pipeline
-        // alongside Logical pass-throughs.
-        | PhysicalExpr::ExactAgg { .. } => None,
+        // An exact accumulator has no sketch family beneath it (its own
+        // child is always a plain `Logical` leaf) — same as the old
+        // `ExactAgg` case.
+        SummaryExpr::SummaryAgg { .. } => None,
+        SummaryExpr::SummaryEstimate { sketch_input, .. } => extract_from_node(sketch_input),
+        SummaryExpr::SummaryMerge { children } => children.iter().find_map(extract_from_node),
+        // Not surfaced by any `Bind*` path yet (gated on rules that
+        // haven't landed — see `physical_expr.rs`'s module docs).
+        SummaryExpr::SummaryJoin { .. }
+        | SummaryExpr::SummarySubtract { .. }
+        | SummaryExpr::SummaryDelete { .. }
+        | SummaryExpr::Logical(_) => None,
     }
 }
 
@@ -990,10 +1023,11 @@ mod runtime_tests {
                 "top_endpoint_qps",
                 Some(BTreeSet::from([SummaryKind::CountSketchWithHeap])),
             ),
-            (
-                "endpoint_request_freq",
-                Some(BTreeSet::from([SummaryKind::Cms])),
-            ),
+            // `CountMinSketch` override re-derives statistic to
+            // `Frequency`, `AggIntent::Extension`-shaped — declines to
+            // bind pending ASAPController#150 (see
+            // `optimizer::rules::tests::typed_binding_endpoint_request_freq_declines_pending_upstream_extension_support`).
+            ("endpoint_request_freq", None),
         ];
         for (metric, want) in &expected {
             let got = map.get(*metric).cloned();
@@ -1003,11 +1037,12 @@ mod runtime_tests {
                  full map: {map:?}",
             );
         }
-        // Routing table covers all 5 sketched metrics.
+        // Routing table covers the 4 sketched metrics (endpoint_request_freq
+        // and http_requests_total both decline — see above).
         assert_eq!(
             map.len(),
-            5,
-            "routing table should have 5 entries (5 sketches; raw declines), got: {map:?}"
+            4,
+            "routing table should have 4 entries (4 sketches; raw + Extension both decline), got: {map:?}"
         );
     }
 
@@ -1177,11 +1212,20 @@ mod runtime_tests {
             mk(AggType::Cardinality, Some(SketchType::HLL), Vec::new()),
             WorkloadCharacteristics::default(),
         );
-        // Frequency → CMS (capability-matched default for Frequency).
+        // TopK → CountSketch-with-heap. Not plain `Frequency, None`
+        // (capability-matched CMS default) any more — `Frequency`'s
+        // capability-matched default is `AggIntent::Extension`-shaped,
+        // which `asap_plan::boundary::implementation_for` maps to
+        // `PassThrough` unconditionally (ASAPController#150), so it no
+        // longer contributes a family to the union at all. Use a
+        // `CountSketch` override instead — it re-derives the statistic to
+        // `TopK` (not `Extension`-shaped), so it still binds, and still
+        // exercises "3 distinct capabilities on one metric → union of 3
+        // distinct families".
         store.set(
             METRIC,
             AggRole::Other,
-            mk(AggType::Frequency, None, Vec::new()),
+            mk(AggType::Frequency, Some(SketchType::CountSketch), Vec::new()),
             WorkloadCharacteristics::default(),
         );
 
@@ -1192,7 +1236,11 @@ mod runtime_tests {
             .unwrap_or_else(|| panic!("http_requests must be in the map\nmap: {map:?}"));
         assert_eq!(
             got,
-            BTreeSet::from([SummaryKind::DDSketch, SummaryKind::Hll, SummaryKind::Cms]),
+            BTreeSet::from([
+                SummaryKind::DDSketch,
+                SummaryKind::Hll,
+                SummaryKind::CountSketchWithHeap
+            ]),
             "a metric queried by 3 capabilities must accumulate 3 families (UNION, not first-wins)\nmap: {map:?}"
         );
     }

control_plane/src/emit/stage_config.rs

@@ -62,9 +62,9 @@ use crate::physical::colored_dag::emitter::{
 // archive-tier metric lists). Importing them at module scope produced an
 // unused-import warning on every non-test build, so they're scoped into the
 // test module's `use super::*` instead (P2-5).
+use crate::intent_algebra::ColumnRef;
 use crate::physical::colored_dag::stage_id::StageId;
-use crate::sketch_algebra::physical_expr::EstimateOp;
-use asap_sketch::{SummaryKind, SummaryParams};
+use asap_sketch::{SketchQuery, SummaryKind, SummaryParams};
 
 // ── YAML structural types ─────────────────────────────────────────────────────
 //
@@ -3031,24 +3031,40 @@ fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonValue {
 /// `aggregations` list by the same `PolicyFingerprint` recipe.
 fn build_backend_readout_json(r: &BackendReadout) -> JsonValue {
     match &r.op {
-        EstimateOp::Quantile { q } => json!({
+        SketchQuery::Quantile { q } => json!({
             "op": "quantile",
             "q": q,
         }),
-        EstimateOp::Cardinality => json!({
+        SketchQuery::Cardinality => json!({
             "op": "cardinality",
         }),
-        EstimateOp::PointCount { key } => json!({
+        SketchQuery::PointCount { key } => json!({
             "op": "point_count",
-            "key": key,
+            "key": column_ref_to_wire_key(key),
         }),
-        EstimateOp::TopK { k } => json!({
+        SketchQuery::TopK { k } => json!({
             "op": "topk",
             "k": k,
         }),
     }
 }
 
+/// The wire-string key for a `SketchQuery::PointCount` readout.
+///
+/// `SampleValue` and `Wildcard` both wire to the legacy `"*"` sentinel
+/// (`sketch_algebra::rules::bind_cms_count`, retired by Step B, used the
+/// literal string `"*"` to mean "all rows / no specific key"; the L5
+/// emitter's per-group resolution already special-cases that string) —
+/// there's no real queryable column for a plain `Count`/`Frequency`
+/// readout in either case, so both collapse to the same sentinel.
+fn column_ref_to_wire_key(col: &ColumnRef) -> String {
+    match col {
+        ColumnRef::Named(name) => name.clone(),
+        ColumnRef::Qualified { table, name } => format!("{table}.{name}"),
+        ColumnRef::SampleValue | ColumnRef::Wildcard => "*".to_string(),
+    }
+}
+
 /// Collapse a heap-bearing `SummaryKind` to its bare counterpart.
 /// Identity for every other kind.
 ///
@@ -3518,11 +3534,11 @@ mod tests {
             readouts: vec![
                 BackendReadout {
                     aggregation_id: "agg0".into(),
-                    op: EstimateOp::Quantile { q: 0.99 },
+                    op: SketchQuery::Quantile { q: 0.99 },
                 },
                 BackendReadout {
                     aggregation_id: "agg1".into(),
-                    op: EstimateOp::Cardinality,
+                    op: SketchQuery::Cardinality,
                 },
             ],
         };
@@ -3591,12 +3607,12 @@ mod tests {
             readouts: vec![
                 BackendReadout {
                     aggregation_id: "agg0".into(),
-                    op: EstimateOp::TopK { k: 10 },
+                    op: SketchQuery::TopK { k: 10 },
                 },
                 BackendReadout {
                     aggregation_id: "agg1".into(),
-                    op: EstimateOp::PointCount {
-                        key: "user_42".into(),
+                    op: SketchQuery::PointCount {
+                        key: ColumnRef::Named("user_42".into()),
                     },
                 },
             ],
@@ -3679,11 +3695,11 @@ mod tests {
             readouts: vec![BackendReadout {
                 aggregation_id: "agg0".into(),
                 op: match kind {
-                    SummaryKind::DDSketch | SummaryKind::Kll => EstimateOp::Quantile { q: 0.99 },
-                    SummaryKind::Hll => EstimateOp::Cardinality,
-                    SummaryKind::CountSketch => EstimateOp::TopK { k: 10 },
-                    SummaryKind::Cms => EstimateOp::PointCount {
-                        key: "user_42".into(),
+                    SummaryKind::DDSketch | SummaryKind::Kll => SketchQuery::Quantile { q: 0.99 },
+                    SummaryKind::Hll => SketchQuery::Cardinality,
+                    SummaryKind::CountSketch => SketchQuery::TopK { k: 10 },
+                    SummaryKind::Cms => SketchQuery::PointCount {
+                        key: ColumnRef::Named("user_42".into()),
                     },
                     other => unreachable!(
                         "backend_cfg_with_kind: unsupported test fixture kind {other:?}"
@@ -4134,7 +4150,7 @@ mod tests {
             }],
             readouts: vec![BackendReadout {
                 aggregation_id: "phase_b_agg0".into(),
-                op: EstimateOp::Quantile { q: 0.99 },
+                op: SketchQuery::Quantile { q: 0.99 },
             }],
         };
         let v = emit_backend_streaming_config_json(&cfg, &[]).expect("emit ok");
@@ -5901,7 +5917,7 @@ mod tests {
         use crate::physical::colored_dag::emitter::{
             AggregationInput, BackendAggregation, BackendReadout, BackendStageConfig,
         };
-        use crate::sketch_algebra::physical_expr::EstimateOp;
+        use asap_sketch::SketchQuery;
 
         let cfg = BackendStageConfig {
             aggregations: vec![BackendAggregation {
@@ -5918,7 +5934,7 @@ mod tests {
             }],
             readouts: vec![BackendReadout {
                 aggregation_id: "agg0".to_string(),
-                op: EstimateOp::Quantile { q: 0.99 },
+                op: SketchQuery::Quantile { q: 0.99 },
             }],
         };
         let v = emit_backend_streaming_config_json(&cfg, &[]).expect("emit ok");

control_plane/src/emit/trait_def.rs

@@ -200,7 +200,7 @@ mod tests {
         BackendAggregation, BackendReadout, ExportTarget, GatewayMergeProcessor,
     };
     use crate::physical::colored_dag::stage_id::StageId;
-    use crate::sketch_algebra::physical_expr::EstimateOp;
+    use asap_sketch::SketchQuery;
     use asap_sketch::{SummaryKind, SummaryParams};
     use std::collections::HashMap;
 
@@ -256,7 +256,7 @@ mod tests {
             }],
             readouts: vec![BackendReadout {
                 aggregation_id: "agg0".to_string(),
-                op: EstimateOp::Quantile { q: 0.99 },
+                op: SketchQuery::Quantile { q: 0.99 },
             }],
         }
     }