feat(intent_algebra): thread the actual item filter value into Frequency (Phase 3a)

control_plane/src/emit/mod.rs

@@ -392,8 +392,22 @@ pub fn collect_metric_to_family(
         // samples fan into each per-family pipeline at the agent and the
         // backend serves every (metric, capability) the workload needs.
         for (_, workload, _wc) in workload_store.get_all_for_metric(&entry.metric_name) {
-            let Some(physical_expr) = crate::optimizer::rules::bind_workload_typed(&workload)
-            else {
+            // If this metric declares an `item_label` (its inner
+            // high-cardinality dimension, e.g. "endpoint") and the
+            // parsed query's own label filters name a value for it (e.g.
+            // `{endpoint="checkout"}`), thread that through as the
+            // `Frequency` intent's actual per-item filter -- see
+            // `bind_workload_typed_with_item_filter`'s doc.
+            let item_filter = entry.item_label.as_deref().and_then(|label| {
+                workload
+                    .label_filters
+                    .get(label)
+                    .map(|v| (label, v.as_str()))
+            });
+            let Some(physical_expr) = crate::optimizer::rules::bind_workload_typed_with_item_filter(
+                &workload,
+                item_filter,
+            ) else {
                 continue;
             };
             if let Some(kind) = extract_root_sketch_kind(&physical_expr) {

control_plane/src/intent_algebra/agg_intent.rs

@@ -67,17 +67,30 @@ pub(crate) const FREQUENCY_EXT_KIND: &str = "frequency";
 
 /// Construct control_plane's point-frequency-via-CMS intent. See module
 /// docs for why this is an `Extension`, not a shared first-class variant.
-pub fn frequency(accuracy: AccuracyTarget) -> AggIntent {
+///
+/// `item` is `Some((label, value))` for a per-item point lookup (e.g.
+/// `count(cms_metric{item="checkout"})` -> `("item", "checkout")`) --
+/// `sketch_algebra::cost_model::ControlPlaneCostModel::readout_extension`
+/// reads these same `item_label`/`item_value` payload keys back out to
+/// build `SketchQuery::PointCount{key: Named(label), value: Some(value)}`.
+/// `None` for a bare frequency total (no specific item), which reads out
+/// as `PointCount{key: SampleValue, value: None}`.
+pub fn frequency(accuracy: AccuracyTarget, item: Option<(String, String)>) -> AggIntent {
+    let mut payload = serde_json::json!({ "accuracy": accuracy });
+    if let Some((label, value)) = item {
+        payload["item_label"] = serde_json::Value::String(label);
+        payload["item_value"] = serde_json::Value::String(value);
+    }
     AggIntent::Extension {
         ext_kind: FREQUENCY_EXT_KIND.to_string(),
-        payload: serde_json::json!({ "accuracy": accuracy }),
+        payload,
     }
 }
 
-/// Default `Frequency` intent — `accuracy = e / 2000`. Unchanged default
-/// from before the merge.
+/// Default `Frequency` intent — `accuracy = e / 2000`, no item filter.
+/// Unchanged default from before the merge.
 pub fn default_frequency() -> AggIntent {
-    frequency(AccuracyTarget::Epsilon(std::f64::consts::E / 2000.0))
+    frequency(AccuracyTarget::Epsilon(std::f64::consts::E / 2000.0), None)
 }
 
 /// If `intent` is control_plane's `Frequency` extension, extract its

control_plane/src/optimizer/rules/mod.rs

@@ -63,6 +63,24 @@ pub fn typed_sketch_algebra_enabled() -> bool {
 /// `PhysicalExpr` is then fed into `planner::stage_split::split_typed_three_stage`
 /// + the per-stage emitters in `config::stage_config`.
 pub fn bind_workload_typed(w: &QueryWorkload) -> Option<crate::sketch_algebra::PhysicalExpr> {
+    bind_workload_typed_with_item_filter(w, None)
+}
+
+/// Like [`bind_workload_typed`], but for a `Frequency` statistic, `item_filter`
+/// (a `(label, value)` pair, e.g. `("item", "checkout")`) threads the
+/// query's actual per-item filter value through to the bound
+/// `SketchQuery::PointCount` -- `None` (what `bind_workload_typed` itself
+/// passes) gives the bare bucket total, same as before this parameter
+/// existed. `QueryWorkload` itself carries no `item_label` field (adding
+/// one would break its 30+ struct-literal construction sites across the
+/// crate), so callers that know a metric's item_label -- e.g.
+/// `emit::collect_metric_to_family`'s loop, which already has `entry:
+/// &WorkloadEntry` and `workload.label_filters` in scope -- pass it in
+/// directly instead.
+pub fn bind_workload_typed_with_item_filter(
+    w: &QueryWorkload,
+    item_filter: Option<(&str, &str)>,
+) -> Option<crate::sketch_algebra::PhysicalExpr> {
     use crate::intent_algebra::schema::{Column, DataType};
     use crate::intent_algebra::{AggIntent as L3AggIntent, QueryExpr, Schema, Source, WindowKind};
     use crate::sketch_algebra::capability_matching::{
@@ -187,7 +205,10 @@ pub fn bind_workload_typed(w: &QueryWorkload) -> Option<crate::sketch_algebra::P
             col: None,
             accuracy: intent_accuracy,
         },
-        StatisticClass::Frequency => crate::intent_algebra::frequency(intent_accuracy),
+        StatisticClass::Frequency => crate::intent_algebra::frequency(
+            intent_accuracy,
+            item_filter.map(|(label, value)| (label.to_string(), value.to_string())),
+        ),
         StatisticClass::TopK => L3AggIntent::TopK {
             k: 10,
             accuracy: intent_accuracy,
@@ -643,6 +664,7 @@ mod tests {
     // | `endpoint_request_freq` | CMS          |
 
     use crate::sketch_algebra::physical_expr::PhysicalExpr;
+    use asap_ir::intent_algebra::expr_ir::ColumnRef;
     use asap_sketch::SummaryKind;
 
     /// Walk the L4 binding output and pull out the approximate sketch
@@ -653,6 +675,22 @@ mod tests {
         crate::emit::extract_root_sketch_kind(expr)
     }
 
+    /// Pull the `SketchQuery` out of a bound `PhysicalExpr`'s top-level
+    /// `SummaryEstimate` -- unlike `extract_family`, this needs the
+    /// readout itself (to check `PointCount`'s `key`/`value`), not just
+    /// the sketch family underneath it.
+    fn extract_query(expr: &PhysicalExpr) -> Option<asap_sketch::SketchQuery> {
+        let PhysicalExpr::Committed(crate::sketch_algebra::physical_expr::L4Plan::Summary(node)) =
+            expr
+        else {
+            return None;
+        };
+        match &node.expr {
+            asap_sketch::SummaryExpr::SummaryEstimate { query, .. } => Some(query.clone()),
+            _ => None,
+        }
+    }
+
     /// Build a workload with the given metric name + reasonable
     /// AggType-driven default for the contract row. The metric-name match
     /// in `classify_demo_metric` overrides the AggType for the
@@ -761,6 +799,41 @@ mod tests {
         );
     }
 
+    #[test]
+    fn bind_workload_typed_with_item_filter_threads_the_actual_value() {
+        // `bind_workload_typed` itself (no item filter) must still read
+        // out as the bare bucket total -- unchanged behavior.
+        let w = workload_for("endpoint_request_freq", AggType::Frequency);
+        let bound = bind_workload_typed(&w).expect("must bind");
+        assert!(
+            matches!(
+                extract_query(&bound),
+                Some(asap_sketch::SketchQuery::PointCount {
+                    key: ColumnRef::SampleValue,
+                    value: None
+                })
+            ),
+            "no item filter given -> bare bucket total, got {:?}",
+            extract_query(&bound)
+        );
+
+        // With an item filter, the SAME workload must read out as a
+        // per-item point lookup carrying the actual value.
+        let bound_filtered =
+            bind_workload_typed_with_item_filter(&w, Some(("endpoint", "checkout")))
+                .expect("must bind");
+        match extract_query(&bound_filtered) {
+            Some(asap_sketch::SketchQuery::PointCount {
+                key: ColumnRef::Named(label),
+                value: Some(value),
+            }) => {
+                assert_eq!(label, "endpoint");
+                assert_eq!(value, "checkout");
+            }
+            other => panic!("expected PointCount{{key: Named(\"endpoint\"), value: Some(\"checkout\")}}, got {other:?}"),
+        }
+    }
+
     // ── sketch_type_override (= sketch_family_override) wins ──────────────────
 
     #[test]

control_plane/src/sketch_algebra/capability.rs

@@ -897,7 +897,7 @@ mod tests {
 
     #[test]
     fn frequency_estimate_with_epsilon_returns_frequency_estimate_approx() {
-        let intent = crate::intent_algebra::frequency(AccuracyTarget::Epsilon(0.01));
+        let intent = crate::intent_algebra::frequency(AccuracyTarget::Epsilon(0.01), None);
         assert_eq!(
             capability_for(&intent),
             Some(Capability::FrequencyEstimate(SketchKindHandle::Any))
@@ -906,10 +906,13 @@ mod tests {
 
     #[test]
     fn frequency_estimate_with_epsilon_delta_returns_frequency_estimate_approx() {
-        let intent = crate::intent_algebra::frequency(AccuracyTarget::EpsilonDelta {
-            epsilon: 0.01,
-            delta: 0.001,
-        });
+        let intent = crate::intent_algebra::frequency(
+            AccuracyTarget::EpsilonDelta {
+                epsilon: 0.01,
+                delta: 0.001,
+            },
+            None,
+        );
         assert_eq!(
             capability_for(&intent),
             Some(Capability::FrequencyEstimate(SketchKindHandle::Any))
@@ -920,7 +923,7 @@ mod tests {
     fn frequency_estimate_with_exact_returns_none() {
         // Exact aggregation routes to archive (sketch fallback only
         // meaningful when raw counters aren't kept).
-        let intent = crate::intent_algebra::frequency(AccuracyTarget::Exact);
+        let intent = crate::intent_algebra::frequency(AccuracyTarget::Exact, None);
         assert_eq!(capability_for(&intent), None);
     }
 

control_plane/src/sketch_algebra/tests.rs

@@ -860,7 +860,7 @@ fn phase_b_archive_only_intents_round_trip_through_binder() {
 // ── Deliberate behavior changes (ASAPController#150 / #151) ──────────────────
 //
 // `AggIntent::Extension` (this deployment's `Frequency` point-query,
-// built via `crate::intent_algebra::frequency(accuracy)`) now binds to a
+// built via `crate::intent_algebra::frequency(accuracy, item)`) now binds to a
 // real `Cms` sketch via `ControlPlaneCostModel::realize_extension`/
 // `readout_extension` (ASAPController#150) — see `frequency_extension_binds_cms`
 // below and `optimizer::rules::mod::tests::typed_binding_endpoint_request_freq_binds_cms`.
@@ -876,7 +876,7 @@ fn frequency_extension_binds_cms() {
     // `ControlPlaneCostModel::realize_extension`/`readout_extension`
     // (ASAPController#150) now realize `AggIntent::Extension{"frequency"}`
     // as a real `Cms` sketch instead of declining to `Logical`.
-    let intent = crate::intent_algebra::frequency(AccuracyTarget::Epsilon(0.01));
+    let intent = crate::intent_algebra::frequency(AccuracyTarget::Epsilon(0.01), None);
     let expr = QueryExpr::Aggregate {
         by: vec![].into(),
         aggs: vec![intent],