diff --git a/Cargo.lock b/Cargo.lock
index 1e736da88..d4ecdd643 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -343,17 +343,7 @@ dependencies = [
 [[package]]
 name = "asap-ir"
 version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6#64df20d90c3ddd519c726dea45c05e1fe5225ce6"
-dependencies = [
- "serde",
- "serde_json",
- "thiserror 2.0.18",
-]
-
-[[package]]
-name = "asap-ir"
-version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/ASAPController?rev=fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4#fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4"
+source = "git+https://github.com/ProjectASAP/ASAPController?rev=cc18c9872bbaf0cadf43566892c3974ec9948eba#cc18c9872bbaf0cadf43566892c3974ec9948eba"
 dependencies = [
  "serde",
  "serde_json",
@@ -363,19 +353,19 @@ dependencies = [
 [[package]]
 name = "asap-l2"
 version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6#64df20d90c3ddd519c726dea45c05e1fe5225ce6"
+source = "git+https://github.com/ProjectASAP/ASAPController?rev=cc18c9872bbaf0cadf43566892c3974ec9948eba#cc18c9872bbaf0cadf43566892c3974ec9948eba"
 dependencies = [
- "asap-ir 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6)",
+ "asap-ir",
  "thiserror 2.0.18",
 ]
 
 [[package]]
 name = "asap-plan"
 version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6#64df20d90c3ddd519c726dea45c05e1fe5225ce6"
+source = "git+https://github.com/ProjectASAP/ASAPController?rev=cc18c9872bbaf0cadf43566892c3974ec9948eba#cc18c9872bbaf0cadf43566892c3974ec9948eba"
 dependencies = [
- "asap-ir 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6)",
- "asap-sketch 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6)",
+ "asap-ir",
+ "asap-sketch",
  "serde_json",
  "thiserror 2.0.18",
 ]
@@ -393,17 +383,9 @@ dependencies = [
 [[package]]
 name = "asap-sketch"
 version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6#64df20d90c3ddd519c726dea45c05e1fe5225ce6"
-dependencies = [
- "asap-ir 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6)",
-]
-
-[[package]]
-name = "asap-sketch"
-version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/ASAPController?rev=fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4#fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4"
+source = "git+https://github.com/ProjectASAP/ASAPController?rev=cc18c9872bbaf0cadf43566892c3974ec9948eba#cc18c9872bbaf0cadf43566892c3974ec9948eba"
 dependencies = [
- "asap-ir 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4)",
+ "asap-ir",
 ]
 
 [[package]]
@@ -438,8 +420,8 @@ name = "asap_types"
 version = "0.1.0"
 dependencies = [
  "anyhow",
- "asap-ir 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4)",
- "asap-sketch 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4)",
+ "asap-ir",
+ "asap-sketch",
  "clap 4.6.1",
  "serde",
  "serde_json",
@@ -809,10 +791,10 @@ name = "control_plane"
 version = "0.1.0"
 dependencies = [
  "anyhow",
- "asap-ir 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6)",
+ "asap-ir",
  "asap-l2",
  "asap-plan",
- "asap-sketch 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6)",
+ "asap-sketch",
  "asap_types",
  "axum",
  "bytes",
@@ -1007,9 +989,9 @@ dependencies = [
  "anyhow",
  "arc-swap",
  "arrow",
- "asap-ir 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6)",
+ "asap-ir",
  "asap-precompute-rs",
- "asap-sketch 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6)",
+ "asap-sketch",
  "asap_otel_proto",
  "asap_sketchlib",
  "asap_types",
diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml
index 764594ee6..62e9dbc98 100644
--- a/control_plane/Cargo.toml
+++ b/control_plane/Cargo.toml
@@ -68,10 +68,10 @@ asap_types.workspace = true
 # `AggIntent::Extension` hook this repo's `Extension{"frequency"}` intent
 # needs (ASAPController#150). 64df20d is a strict descendant of d4c1756
 # (the previous pin), so nothing this repo already consumes moves.
-asap-ir = { git = "https://github.com/ProjectASAP/ASAPController", rev = "64df20d90c3ddd519c726dea45c05e1fe5225ce6" }
-asap-l2 = { git = "https://github.com/ProjectASAP/ASAPController", rev = "64df20d90c3ddd519c726dea45c05e1fe5225ce6" }
-asap-sketch = { git = "https://github.com/ProjectASAP/ASAPController", rev = "64df20d90c3ddd519c726dea45c05e1fe5225ce6" }
-asap-plan = { git = "https://github.com/ProjectASAP/ASAPController", rev = "64df20d90c3ddd519c726dea45c05e1fe5225ce6" }
+asap-ir = { git = "https://github.com/ProjectASAP/ASAPController", rev = "cc18c9872bbaf0cadf43566892c3974ec9948eba" }
+asap-l2 = { git = "https://github.com/ProjectASAP/ASAPController", rev = "cc18c9872bbaf0cadf43566892c3974ec9948eba" }
+asap-sketch = { git = "https://github.com/ProjectASAP/ASAPController", rev = "cc18c9872bbaf0cadf43566892c3974ec9948eba" }
+asap-plan = { git = "https://github.com/ProjectASAP/ASAPController", rev = "cc18c9872bbaf0cadf43566892c3974ec9948eba" }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }
diff --git a/control_plane/src/emit/mod.rs b/control_plane/src/emit/mod.rs
index 4197b97d8..0b86bab93 100644
--- a/control_plane/src/emit/mod.rs
+++ b/control_plane/src/emit/mod.rs
@@ -328,14 +328,14 @@ fn is_exact_accumulator(kind: &SummaryKind) -> bool {
 
 fn extract_from_node(node: &Rc<L4Node>) -> Option<SummaryKind> {
     match &node.expr {
-        SummaryExpr::SummaryAgg { sketch, .. } if !is_exact_accumulator(sketch) => {
-            Some(sketch.clone())
+        SummaryExpr::SummaryAgg { summary, .. } if !is_exact_accumulator(summary) => {
+            Some(summary.clone())
         }
         // 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::SummaryEstimate { summary_input, .. } => extract_from_node(summary_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).
diff --git a/control_plane/src/intent_algebra/lower.rs b/control_plane/src/intent_algebra/lower.rs
index ef5f86c55..b3a5a3bf6 100644
--- a/control_plane/src/intent_algebra/lower.rs
+++ b/control_plane/src/intent_algebra/lower.rs
@@ -121,7 +121,7 @@ use crate::intent_algebra::column_resolution::{
 };
 use crate::intent_algebra::query_expr::{
     GroupKeys, L3Scalar, Predicate, ProjectItem as CProjectItem, QueryExpr as CQueryExpr,
-    QueryExprError, SortKey as CSortKey, Source, WindowKind as CWindowKind,
+    QueryExprError, Reduction, SortKey as CSortKey, Source, WindowKind as CWindowKind,
 };
 use crate::intent_algebra::relational::{AggFunc, QueryExpr as LQueryExpr};
 use crate::intent_algebra::schema::Schema;
@@ -270,6 +270,21 @@ pub fn convert(legacy: &LQueryExpr, schema: &Schema) -> Result<CQueryExpr, Conve
                 let item = &aggs[0];
                 let intents = agg_func_to_intents(&item.func, frequency_trigger);
                 if !intents.is_empty() {
+                    // Per-entity (issue ASAPController#163/#165): a single
+                    // intent, no `by`/`without` at all, whose intent is
+                    // inherently per-series or whose child is a range
+                    // window (`windowed`, computed above -- this repo's L3
+                    // shape keeps `Window` *above* the `Aggregate` rather
+                    // than a `TimeRange` child inside it, unlike
+                    // ASAPController's current canonical shape, but it's
+                    // the same structural marker for "this is a bare range
+                    // reduction with no grouping syntax to begin with").
+                    // `by.is_without()` is never true here -- `without` PromQL
+                    // grouping always resolves via the `without` bool above,
+                    // which only ever produces `GroupKeys::without` with a
+                    // concrete (possibly empty) exclusion list, and that
+                    // path already implies a genuine reduction regardless.
+                    let per_entity_base = by.is_empty() && !by.is_without();
                     let nodes: Vec<CQueryExpr> = match input.as_ref() {
                         LQueryExpr::Window {
                             duration,
@@ -284,17 +299,26 @@ pub fn convert(legacy: &LQueryExpr, schema: &Schema) -> Result<CQueryExpr, Conve
                             let win_child = convert(win_input, schema)?;
                             intents
                                 .into_iter()
-                                .map(|intent| CQueryExpr::Window {
-                                    kind: kind.clone(),
-                                    size: *duration,
-                                    slide: *slide,
-                                    child: Box::new(CQueryExpr::Aggregate {
-                                        by: by.clone(),
-                                        aggs: vec![intent],
-                                        output_names: Vec::new(),
-                                        having: None,
-                                        child: Box::new(win_child.clone()),
-                                    }),
+                                .map(|intent| {
+                                    let reduction = if per_entity_base
+                                        && (intent.is_per_series() || windowed)
+                                    {
+                                        Reduction::PerEntity
+                                    } else {
+                                        Reduction::Reduce(by.clone())
+                                    };
+                                    CQueryExpr::Window {
+                                        kind: kind.clone(),
+                                        size: *duration,
+                                        slide: *slide,
+                                        child: Box::new(CQueryExpr::Aggregate {
+                                            reduction,
+                                            aggs: vec![intent],
+                                            output_names: Vec::new(),
+                                            having: None,
+                                            child: Box::new(win_child.clone()),
+                                        }),
+                                    }
                                 })
                                 .collect()
                         }
@@ -302,12 +326,21 @@ pub fn convert(legacy: &LQueryExpr, schema: &Schema) -> Result<CQueryExpr, Conve
                             let child = convert(other, schema)?;
                             intents
                                 .into_iter()
-                                .map(|intent| CQueryExpr::Aggregate {
-                                    by: by.clone(),
-                                    aggs: vec![intent],
-                                    output_names: Vec::new(),
-                                    having: None,
-                                    child: Box::new(child.clone()),
+                                .map(|intent| {
+                                    let reduction = if per_entity_base
+                                        && (intent.is_per_series() || windowed)
+                                    {
+                                        Reduction::PerEntity
+                                    } else {
+                                        Reduction::Reduce(by.clone())
+                                    };
+                                    CQueryExpr::Aggregate {
+                                        reduction,
+                                        aggs: vec![intent],
+                                        output_names: Vec::new(),
+                                        having: None,
+                                        child: Box::new(child.clone()),
+                                    }
                                 })
                                 .collect()
                         }
@@ -338,7 +371,11 @@ pub fn convert(legacy: &LQueryExpr, schema: &Schema) -> Result<CQueryExpr, Conve
                 .map(|h| resolve_expr(h, schema).map(Predicate))
                 .transpose()?;
             CQueryExpr::Aggregate {
-                by,
+                // Always a genuine reduction: this branch is multi-intent,
+                // HAVING-bearing, or a (structurally unreachable) unmapped
+                // AggFunc -- per-entity reductions are always the single,
+                // HAVING-less intent handled above.
+                reduction: Reduction::Reduce(by),
                 aggs: intents,
                 output_names: Vec::new(),
                 having,
@@ -372,7 +409,9 @@ pub fn convert(legacy: &LQueryExpr, schema: &Schema) -> Result<CQueryExpr, Conve
             // see `topk_bridge` module doc.
             let by: GroupKeys = resolve_group_keys_promql(by, schema)?.into();
             CQueryExpr::Aggregate {
-                by,
+                // A ranking always reduces (empty by ranks the whole
+                // input, never per-entity).
+                reduction: Reduction::Reduce(by),
                 aggs: vec![AggIntent::TopK {
                     k: *k as usize,
                     accuracy: AccuracyTarget::Epsilon(0.05),
@@ -745,8 +784,13 @@ mod tests {
     fn single_aggregate_folds_to_canonical_aggregate() {
         let legacy = agg(vec![], false, vec![agg_item("s", AggFunc::Sum)], src("m"));
         match convert_root(&legacy).unwrap() {
-            CQueryExpr::Aggregate { by, aggs, .. } => {
-                assert!(by.is_empty(), "no GROUP BY → empty `by`: {by:?}");
+            CQueryExpr::Aggregate {
+                reduction, aggs, ..
+            } => {
+                assert!(
+                    matches!(&reduction, Reduction::Reduce(k) if k.is_empty()),
+                    "no GROUP BY, unwindowed, non-per-series intent → global reduce: {reduction:?}"
+                );
                 assert!(matches!(aggs.as_slice(), [AggIntent::Sum { col: None }]));
             }
             other => panic!("expected Aggregate, got {other:?}"),
@@ -822,7 +866,9 @@ mod tests {
             src("trades"),
         );
         match convert_root(&legacy).unwrap() {
-            CQueryExpr::Aggregate { by, .. } => assert!(by.is_empty()),
+            CQueryExpr::Aggregate { reduction, .. } => {
+                assert!(matches!(&reduction, Reduction::Reduce(k) if k.is_empty()))
+            }
             other => panic!("expected Aggregate, got {other:?}"),
         }
     }
@@ -844,8 +890,8 @@ mod tests {
                 assert_eq!(kind, CWindowKind::Tumbling);
                 assert!(matches!(
                     *child,
-                    CQueryExpr::Aggregate { ref by, ref aggs, .. }
-                        if by.is_empty()
+                    CQueryExpr::Aggregate { ref reduction, ref aggs, .. }
+                        if matches!(reduction, Reduction::PerEntity)
                             && matches!(aggs.as_slice(), [AggIntent::Quantile { .. }])
                 ));
             }
@@ -939,8 +985,10 @@ mod tests {
             input: Box::new(src("m")),
         };
         match convert_root(&legacy).unwrap() {
-            CQueryExpr::Aggregate { by, aggs, .. } => {
-                assert!(by.is_empty());
+            CQueryExpr::Aggregate {
+                reduction, aggs, ..
+            } => {
+                assert!(matches!(&reduction, Reduction::Reduce(k) if k.is_empty()));
                 assert!(matches!(aggs.as_slice(), [AggIntent::TopK { k: 5, .. }]));
             }
             other => panic!("expected Aggregate, got {other:?}"),
diff --git a/control_plane/src/intent_algebra/mod.rs b/control_plane/src/intent_algebra/mod.rs
index 55267affb..cc08b7007 100644
--- a/control_plane/src/intent_algebra/mod.rs
+++ b/control_plane/src/intent_algebra/mod.rs
@@ -113,8 +113,8 @@ pub use expr_ir::{ArithOp, ColumnRef, CompareOp, Expr, L2Expr, L3Expr, L3Scalar}
 pub use query_expr::{
     aggregate_output_schema, between, conjoin, label_filter_to_predicate, AtModifier, BinaryOpKind,
     BindingScope, DataModel, GroupKeys, GroupSide, InfoMatcher, JoinKind, LabelFilter, Predicate,
-    ProjectItem, QueryExpr, QueryExprError, SampleKind, SetOpKind, SortKey, Source, TimeShift,
-    VectorGrouping, VectorMatch, VectorMatchKind, WindowFuncKind, WindowKind,
+    ProjectItem, QueryExpr, QueryExprError, Reduction, SampleKind, SetOpKind, SortKey, Source,
+    TimeShift, VectorGrouping, VectorMatch, VectorMatchKind, WindowFuncKind, WindowKind,
 };
 pub use schema::{cse_reuse_is_legal, Column, ColumnId, CseError, DataType, Schema};
 
diff --git a/control_plane/src/intent_algebra/query_expr.rs b/control_plane/src/intent_algebra/query_expr.rs
index ed787243f..0ba5e459d 100644
--- a/control_plane/src/intent_algebra/query_expr.rs
+++ b/control_plane/src/intent_algebra/query_expr.rs
@@ -61,7 +61,7 @@ use asap_ir::intent_algebra::schema::ColumnId;
 pub use asap_ir::intent_algebra::{
     aggregate_output_schema, AtModifier, BinaryOpKind, BindingScope, DataModel, GroupKeys,
     GroupSide, InfoMatcher, JoinKind, Predicate, ProjectItem, QueryExpr, QueryExprError,
-    SampleKind, SetOpKind, SortKey, Source, TimeShift, VectorGrouping, VectorMatch,
+    Reduction, SampleKind, SetOpKind, SortKey, Source, TimeShift, VectorGrouping, VectorMatch,
     VectorMatchKind, WindowFuncKind, WindowKind,
 };
 pub use asap_ir::intent_algebra::{ArithOp, ColumnRef, CompareOp, Expr, L3Scalar};
diff --git a/control_plane/src/optimizer/cost/mod.rs b/control_plane/src/optimizer/cost/mod.rs
index 6c10eb113..628e680c9 100644
--- a/control_plane/src/optimizer/cost/mod.rs
+++ b/control_plane/src/optimizer/cost/mod.rs
@@ -355,7 +355,7 @@ fn apply_delta_decision_with(
 use asap_ir::intent_algebra::{BindingName, QueryId};
 
 #[allow(unused_imports)]
-use crate::intent_algebra::{AggIntent, BindingScope, QueryExpr, QueryExprError, Schema};
+use crate::intent_algebra::{AggIntent, BindingScope, QueryExpr, QueryExprError, Reduction, Schema};
 
 /// Bundled cost of a multi-query workload, with per-root contributions
 /// and the savings unlocked by shared-producer credit. Returned by
@@ -503,10 +503,14 @@ fn subtree_cost_bundled(
             Ok(node_cost_window(&in_schema) + cs)
         }
         QueryExpr::Aggregate {
-            by, aggs, child, ..
+            reduction,
+            aggs,
+            child,
+            ..
         } => {
             let cs = subtree_cost_bundled(child, binding_costs, schema_scope)?;
             let in_schema = child.output_schema_in(schema_scope)?;
+            let by = reduction.group_keys().map(|k| k.keys()).unwrap_or(&[]);
             Ok(node_cost_aggregate(by, aggs, &in_schema) + cs)
         }
         QueryExpr::LetBinding { name, expr, child } => {
@@ -589,10 +593,14 @@ fn subtree_cost_standalone(
             Ok(node_cost_window(&in_schema) + cs)
         }
         QueryExpr::Aggregate {
-            by, aggs, child, ..
+            reduction,
+            aggs,
+            child,
+            ..
         } => {
             let cs = subtree_cost_standalone(child, binding_costs, schema_scope)?;
             let in_schema = child.output_schema_in(schema_scope)?;
+            let by = reduction.group_keys().map(|k| k.keys()).unwrap_or(&[]);
             Ok(node_cost_aggregate(by, aggs, &in_schema) + cs)
         }
         QueryExpr::LetBinding { name, expr, child } => {
@@ -783,10 +791,18 @@ mod workload_cost_tests {
         }
     }
 
-    /// Wrap `child` in `Aggregate { by: [], aggs: [Quantile{q}] }`.
+    /// Wrap `child` in `Aggregate { reduction: [], aggs: [Quantile{q}] }`
+    /// — no `by()`, so `reduction` follows `lower.rs`'s own rule:
+    /// `PerEntity` when `child` is itself a `Window` (matches a bare
+    /// `quantile_over_time(...)`), `Reduce([])` otherwise.
     fn quantile_root(q: f64, child: QueryExpr) -> QueryExpr {
+        let reduction = if matches!(child, QueryExpr::Window { .. }) {
+            Reduction::PerEntity
+        } else {
+            Reduction::by(vec![])
+        };
         QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction,
             aggs: vec![AggIntent::Quantile {
                 col: None,
                 q,
@@ -798,10 +814,16 @@ mod workload_cost_tests {
         }
     }
 
-    /// Wrap `child` in `Aggregate { by: [], aggs: [Max] }`.
+    /// Wrap `child` in `Aggregate { reduction: [], aggs: [Max] }` — same
+    /// `PerEntity`-when-windowed rule as `quantile_root`.
     fn max_root(child: QueryExpr) -> QueryExpr {
+        let reduction = if matches!(child, QueryExpr::Window { .. }) {
+            Reduction::PerEntity
+        } else {
+            Reduction::by(vec![])
+        };
         QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction,
             aggs: vec![AggIntent::Max { col: None }],
             output_names: Vec::new(),
             having: None,
diff --git a/control_plane/src/optimizer/cse.rs b/control_plane/src/optimizer/cse.rs
index 1014fce90..99c88ae63 100644
--- a/control_plane/src/optimizer/cse.rs
+++ b/control_plane/src/optimizer/cse.rs
@@ -163,13 +163,13 @@ pub fn dedupe_subtrees(roots: Vec<(QueryId, QueryExpr)>) -> CseWorkloadPlan {
     for (qid, root) in roots {
         let new_root = match root {
             QueryExpr::Aggregate {
-                by,
+                reduction,
                 aggs,
                 output_names,
                 having,
                 child,
             } if *child == shared_expr => QueryExpr::Aggregate {
-                by,
+                reduction,
                 aggs,
                 output_names,
                 having,
@@ -194,7 +194,7 @@ pub fn dedupe_subtrees(roots: Vec<(QueryId, QueryExpr)>) -> CseWorkloadPlan {
 mod tests {
     use super::*;
     use crate::intent_algebra::agg_intent::AggIntent;
-    use crate::intent_algebra::query_expr::{LabelFilter, Source, WindowKind};
+    use crate::intent_algebra::query_expr::{LabelFilter, Reduction, Source, WindowKind};
     use crate::intent_algebra::schema::{Column, DataType, Schema};
     use crate::types_v2::AccuracyTarget;
     use std::time::Duration;
@@ -254,7 +254,7 @@ mod tests {
     #[test]
     fn dedupe_subtrees_single_root_passthrough() {
         let q = QueryExpr::Aggregate {
-            by: vec![1].into(),
+            reduction: Reduction::by(vec![1]),
             aggs: vec![AggIntent::Quantile {
                 col: None,
                 q: 0.99,
@@ -280,7 +280,7 @@ mod tests {
     #[test]
     fn quantiles_over_different_columns_do_not_dedupe() {
         let mk = |col: usize| QueryExpr::Aggregate {
-            by: vec![1].into(),
+            reduction: Reduction::by(vec![1]),
             aggs: vec![AggIntent::Quantile {
                 col: Some(col),
                 q: 0.5,
@@ -306,7 +306,7 @@ mod tests {
     #[test]
     fn dedupe_subtrees_basic() {
         let q1 = QueryExpr::Aggregate {
-            by: vec![1].into(),
+            reduction: Reduction::by(vec![1]),
             aggs: vec![AggIntent::Quantile {
                 col: None,
                 q: 0.99,
@@ -317,7 +317,7 @@ mod tests {
             child: Box::new(windowed_scan()),
         };
         let q2 = QueryExpr::Aggregate {
-            by: vec![1].into(),
+            reduction: Reduction::by(vec![1]),
             aggs: vec![AggIntent::Quantile {
                 col: None,
                 q: 0.95,
@@ -354,7 +354,7 @@ mod tests {
     #[test]
     fn dedupe_subtrees_no_shared_subexpr() {
         let q1 = QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction: Reduction::PerEntity,
             aggs: vec![AggIntent::Sum { col: None }],
             output_names: Vec::new(),
             having: None,
@@ -378,7 +378,7 @@ mod tests {
             ),
         };
         let q2 = QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction: Reduction::PerEntity,
             aggs: vec![AggIntent::Max { col: None }],
             output_names: Vec::new(),
             having: None,
@@ -417,7 +417,7 @@ mod tests {
             ),
         };
         let mk = || QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction: Reduction::by(vec![]),
             aggs: vec![AggIntent::Sum { col: None }],
             output_names: Vec::new(),
             having: None,
diff --git a/control_plane/src/optimizer/engine.rs b/control_plane/src/optimizer/engine.rs
index 3ebbb9cc8..6de5826a1 100644
--- a/control_plane/src/optimizer/engine.rs
+++ b/control_plane/src/optimizer/engine.rs
@@ -33,7 +33,7 @@ use std::collections::HashMap;
 use asap_ir::intent_algebra::BindingName;
 
 use crate::intent_algebra::agg_intent::AggIntent;
-use crate::intent_algebra::query_expr::{GroupKeys, QueryExpr, SetOpKind, Source};
+use crate::intent_algebra::query_expr::{QueryExpr, Reduction, SetOpKind, Source};
 use crate::intent_algebra::relational::{agg_is_exact, agg_is_mergeable};
 use crate::optimizer::cost::sketch_capability::{
     default_capability_table, load_capability_overrides, SketchCapability,
@@ -396,7 +396,7 @@ impl RewriteRule for MergeLifting {
     fn try_rewrite(&self, expr: QueryExpr, _model: &dyn CostModel) -> Option<QueryExpr> {
         match expr {
             QueryExpr::Aggregate {
-                ref by,
+                ref reduction,
                 ref aggs,
                 ref having,
                 ref child,
@@ -406,7 +406,7 @@ impl RewriteRule for MergeLifting {
                     let new_children: Vec<QueryExpr> = children
                         .iter()
                         .map(|branch| QueryExpr::Aggregate {
-                            by: by.clone(),
+                            reduction: reduction.clone(),
                             aggs: aggs.clone(),
                             output_names: Vec::new(),
                             having: None,
@@ -440,7 +440,7 @@ impl RewriteRule for HLLDedupElim {
     fn try_rewrite(&self, expr: QueryExpr, _model: &dyn CostModel) -> Option<QueryExpr> {
         match expr {
             QueryExpr::Aggregate {
-                by,
+                reduction,
                 aggs,
                 output_names,
                 having: None,
@@ -448,7 +448,7 @@ impl RewriteRule for HLLDedupElim {
             } if aggs.len() == 1 && matches!(&aggs[0], AggIntent::Cardinality { .. }) => {
                 if let QueryExpr::Distinct { child: inner, .. } = *child {
                     return Some(QueryExpr::Aggregate {
-                        by,
+                        reduction,
                         aggs,
                         output_names,
                         having: None,
@@ -523,7 +523,7 @@ impl RewriteRule for TopKFusion {
                     // Only fuse when all keys are DESC (top-k semantics).
                     if !keys.is_empty() && keys.iter().all(|k| !k.ascending) {
                         return Some(QueryExpr::Aggregate {
-                            by: GroupKeys::none(),
+                            reduction: Reduction::by(vec![]),
                             aggs: vec![AggIntent::TopK {
                                 k: n,
                                 accuracy: AccuracyTarget::Epsilon(0.05),
@@ -861,7 +861,7 @@ impl QueryOptimizer {
                 )
             }
             QueryExpr::Aggregate {
-                by,
+                reduction,
                 aggs,
                 output_names,
                 having,
@@ -870,7 +870,7 @@ impl QueryOptimizer {
                 let (new_child, c) = recurse!(child);
                 (
                     QueryExpr::Aggregate {
-                        by,
+                        reduction,
                         aggs,
                         output_names,
                         having,
@@ -1165,7 +1165,7 @@ impl OptimizerRule for CommonSubexprElim {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::intent_algebra::query_expr::Predicate;
+    use crate::intent_algebra::query_expr::{GroupKeys, Predicate};
     use crate::intent_algebra::relational::{default_cardinality, default_quantile};
     use crate::intent_algebra::{
         BinaryOpKind, L3Expr, L3Scalar, Schema, SortKey, Source, WindowKind,
@@ -1187,7 +1187,7 @@ mod tests {
     /// the legacy `SketchAgg`.
     fn sketch_agg(intent: AggIntent, child: QueryExpr) -> QueryExpr {
         QueryExpr::Aggregate {
-            by: GroupKeys::none(),
+            reduction: Reduction::by(vec![]),
             aggs: vec![intent],
             output_names: Vec::new(),
             having: None,
diff --git a/control_plane/src/optimizer/rules/mod.rs b/control_plane/src/optimizer/rules/mod.rs
index 816bd2867..3d78e5775 100644
--- a/control_plane/src/optimizer/rules/mod.rs
+++ b/control_plane/src/optimizer/rules/mod.rs
@@ -231,7 +231,12 @@ pub fn bind_workload_typed_with_item_filter(
         child: Box::new(scan),
     };
     let aggregate = QueryExpr::Aggregate {
-        by: crate::intent_algebra::GroupKeys::none(),
+        // Synthetic probe only -- `boundary::implementation_for` (what
+        // this shape actually drives) keys off `AggIntent`/accuracy alone,
+        // never `Reduction`, so this value doesn't affect the family pick.
+        // `PerEntity` is the representative choice for a windowed shape
+        // with no `by` (ASAPController#163/#165).
+        reduction: crate::intent_algebra::Reduction::PerEntity,
         aggs: vec![intent],
         output_names: Vec::new(),
         having: None,
diff --git a/control_plane/src/physical/allocator.rs b/control_plane/src/physical/allocator.rs
index d49ee2c11..85a092363 100644
--- a/control_plane/src/physical/allocator.rs
+++ b/control_plane/src/physical/allocator.rs
@@ -40,9 +40,8 @@
 
 use super::plan::{CostEstimate, ExecutionMode, NodeAnnotation, PipelineStage, PlanNode};
 use crate::intent_algebra::agg_intent::AggIntent;
-use crate::intent_algebra::query_expr::GroupKeys;
 use crate::intent_algebra::relational::agg_is_exact;
-use crate::intent_algebra::QueryExpr;
+use crate::intent_algebra::{QueryExpr, Reduction};
 use crate::types::{SketchType, StageResourceBudgets};
 
 // ── Resource budget tracker ───────────────────────────────────────────────────
@@ -202,7 +201,7 @@ impl SketchAllocator {
             //   * single other intent, no HAVING → budget-driven sketch
             //   * multi-intent or HAVING → general exact Aggregate at Db
             QueryExpr::Aggregate {
-                by,
+                reduction,
                 aggs,
                 output_names,
                 having,
@@ -214,7 +213,7 @@ impl SketchAllocator {
                         let child = self.alloc_node(*child, budget);
                         return PlanNode {
                             expr: QueryExpr::Aggregate {
-                                by,
+                                reduction,
                                 aggs,
                                 output_names,
                                 having,
@@ -239,14 +238,14 @@ impl SketchAllocator {
                     }
                     // Single non-TopK intent → budget-driven sketch agg.
                     let child = self.alloc_node(*child, budget);
-                    return self.alloc_sketch_agg(by, aggs, output_names, child, budget);
+                    return self.alloc_sketch_agg(reduction, aggs, output_names, child, budget);
                 }
                 // General multi-intent / HAVING aggregate → Db (exact).
                 let child = self.alloc_node(*child, budget);
                 let kinds: Vec<&'static str> = aggs.iter().map(canonical_intent_kind_str).collect();
                 PlanNode {
                     expr: QueryExpr::Aggregate {
-                        by,
+                        reduction,
                         aggs,
                         output_names,
                         having,
@@ -609,7 +608,7 @@ impl SketchAllocator {
     /// `aggs.len() == 1` and that the single intent is not `TopK`.
     fn alloc_sketch_agg(
         &self,
-        by: GroupKeys,
+        reduction: Reduction,
         aggs: Vec<AggIntent>,
         output_names: Vec<String>,
         child: PlanNode,
@@ -621,7 +620,7 @@ impl SketchAllocator {
         if matches!(intent, AggIntent::Avg { .. }) {
             return PlanNode {
                 expr: QueryExpr::Aggregate {
-                    by,
+                    reduction,
                     aggs,
                     output_names,
                     having: None,
@@ -645,7 +644,7 @@ impl SketchAllocator {
         if agg_is_exact(&intent) {
             return PlanNode {
                 expr: QueryExpr::Aggregate {
-                    by,
+                    reduction,
                     aggs,
                     output_names,
                     having: None,
@@ -675,7 +674,7 @@ impl SketchAllocator {
             budget.consume_agent(mem);
             return PlanNode {
                 expr: QueryExpr::Aggregate {
-                    by,
+                    reduction,
                     aggs,
                     output_names,
                     having: None,
@@ -703,7 +702,7 @@ impl SketchAllocator {
             budget.consume_backend(mem);
             return PlanNode {
                 expr: QueryExpr::Aggregate {
-                    by,
+                    reduction,
                     aggs,
                     output_names,
                     having: None,
@@ -731,7 +730,7 @@ impl SketchAllocator {
         // Both Agent and Backend budgets exceeded → Precompute.
         PlanNode {
             expr: QueryExpr::Aggregate {
-                by,
+                reduction,
                 aggs,
                 output_names,
                 having: None,
@@ -845,7 +844,7 @@ mod tests {
     /// `Scan` — the canonical shape the legacy `SketchAgg` folded into.
     fn agg(intent: AggIntent) -> QueryExpr {
         QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction: Reduction::by(vec![]),
             aggs: vec![intent],
             output_names: Vec::new(),
             having: None,
@@ -1006,7 +1005,7 @@ mod tests {
     #[test]
     fn multi_intent_aggregate_goes_to_db() {
         let expr = QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction: Reduction::by(vec![]),
             aggs: vec![AggIntent::Sum { col: None }, AggIntent::Min { col: None }],
             output_names: Vec::new(),
             having: None,
diff --git a/control_plane/src/physical/colored_dag/allocator.rs b/control_plane/src/physical/colored_dag/allocator.rs
index 7dcd41dc1..3eb6c1db1 100644
--- a/control_plane/src/physical/colored_dag/allocator.rs
+++ b/control_plane/src/physical/colored_dag/allocator.rs
@@ -193,8 +193,8 @@ impl ThreeStageWalker {
             // The "SketchEstimate MUST be on the same stage as its
             // consumers (typically Backend)" invariant is satisfied
             // because consumers above SummaryEstimate are also backend.
-            SummaryExpr::SummaryEstimate { sketch_input, .. } => {
-                let (cid, child_stage) = self.visit_l4node(sketch_input)?;
+            SummaryExpr::SummaryEstimate { summary_input, .. } => {
+                let (cid, child_stage) = self.visit_l4node(summary_input)?;
                 self.dag.edges.push((id, cid));
                 // If child is on edge or gateway, this is a cross-stage
                 // edge — that's expected (the wire-format hop).
@@ -232,8 +232,8 @@ impl ThreeStageWalker {
                 self.dag.edges.push((id, rid));
                 StageId::Gateway
             }
-            SummaryExpr::SummaryDelete { sketch_input, .. } => {
-                let (cid, _) = self.visit_l4node(sketch_input)?;
+            SummaryExpr::SummaryDelete { summary_input, .. } => {
+                let (cid, _) = self.visit_l4node(summary_input)?;
                 self.dag.edges.push((id, cid));
                 StageId::Gateway
             }
@@ -399,7 +399,7 @@ mod tests {
     #[test]
     fn three_stage_quantile_dag_basic() {
         let q = QueryExpr::Aggregate {
-            by: crate::intent_algebra::GroupKeys::none(),
+            reduction: crate::intent_algebra::Reduction::PerEntity,
             aggs: vec![crate::intent_algebra::AggIntent::Quantile {
                 col: None,
                 q: 0.99,
diff --git a/control_plane/src/physical/colored_dag/dag.rs b/control_plane/src/physical/colored_dag/dag.rs
index fe08ca8bd..ae268f9c3 100644
--- a/control_plane/src/physical/colored_dag/dag.rs
+++ b/control_plane/src/physical/colored_dag/dag.rs
@@ -186,7 +186,7 @@ mod tests {
 
     fn dummy_agg() -> PhysicalExpr {
         let q = QueryExpr::Aggregate {
-            by: crate::intent_algebra::GroupKeys::none(),
+            reduction: crate::intent_algebra::Reduction::by(vec![]),
             aggs: vec![crate::intent_algebra::AggIntent::Sum { col: None }],
             output_names: Vec::new(),
             having: None,
@@ -197,7 +197,7 @@ mod tests {
 
     fn dummy_estimate() -> PhysicalExpr {
         let q = QueryExpr::Aggregate {
-            by: crate::intent_algebra::GroupKeys::none(),
+            reduction: crate::intent_algebra::Reduction::by(vec![]),
             aggs: vec![crate::intent_algebra::AggIntent::Quantile {
                 col: None,
                 q: 0.99,
diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs
index dcd71a4ab..60eb02d8d 100644
--- a/control_plane/src/physical/colored_dag/emitter.rs
+++ b/control_plane/src/physical/colored_dag/emitter.rs
@@ -92,12 +92,12 @@ fn classify(expr: &PhysicalExpr) -> NodeKind<'_> {
     match expr {
         PhysicalExpr::Committed(L4Plan::Summary(node)) => match &node.expr {
             SummaryExpr::Logical(qe) => NodeKind::Logical(qe),
-            SummaryExpr::SummaryAgg { sketch, params, .. } if is_exact_accumulator(sketch) => {
+            SummaryExpr::SummaryAgg { summary, params, .. } if is_exact_accumulator(summary) => {
                 let _ = params;
                 NodeKind::ExactAgg
             }
-            SummaryExpr::SummaryAgg { sketch, params, .. } => NodeKind::SketchAgg {
-                sketch_type: sketch,
+            SummaryExpr::SummaryAgg { summary, params, .. } => NodeKind::SketchAgg {
+                sketch_type: summary,
                 params,
             },
             SummaryExpr::SummaryEstimate { query, .. } => NodeKind::SketchEstimate { query },
diff --git a/control_plane/src/physical/colored_dag/tests.rs b/control_plane/src/physical/colored_dag/tests.rs
index 642411fb9..a73b0c693 100644
--- a/control_plane/src/physical/colored_dag/tests.rs
+++ b/control_plane/src/physical/colored_dag/tests.rs
@@ -13,7 +13,7 @@ use std::time::Duration;
 use asap_sketch::{L4Node, L4Schema, SketchQuery, SummaryExpr, SummaryKind, SummaryParams};
 
 use crate::intent_algebra::{
-    BindingScope, ColumnRef, GroupKeys, LabelFilter, QueryExpr, Schema, Source, WindowKind,
+    BindingScope, ColumnRef, LabelFilter, QueryExpr, Reduction, Schema, Source, WindowKind,
 };
 use crate::intent_algebra::schema::{Column, DataType};
 use crate::physical::colored_dag::allocator::StageAllocator;
@@ -104,14 +104,14 @@ fn logical_l4(qe: QueryExpr) -> Rc<L4Node> {
 /// `PhysicalExpr::SketchAgg { sketch_type, params, child }` construction,
 /// for fixtures that need a specific family without going through
 /// `implement_tree`'s cost-model selection.
-fn sketch_agg_l4(sketch: SummaryKind, params: SummaryParams, child: Rc<L4Node>) -> Rc<L4Node> {
+fn sketch_agg_l4(summary: SummaryKind, params: SummaryParams, child: Rc<L4Node>) -> Rc<L4Node> {
     Rc::new(L4Node {
         expr: SummaryExpr::SummaryAgg {
             child,
-            sketch,
+            summary,
             params,
             col: ColumnRef::SampleValue,
-            by: vec![],
+            reduction: Reduction::by(vec![]),
         },
         schema: dummy_l4_schema(),
     })
@@ -119,9 +119,9 @@ fn sketch_agg_l4(sketch: SummaryKind, params: SummaryParams, child: Rc<L4Node>)
 
 /// Hand-build a `SummaryEstimate` node — mirrors the old
 /// `PhysicalExpr::SketchEstimate { op, child }`.
-fn estimate_l4(query: SketchQuery, sketch_input: Rc<L4Node>) -> Rc<L4Node> {
+fn estimate_l4(query: SketchQuery, summary_input: Rc<L4Node>) -> Rc<L4Node> {
     Rc::new(L4Node {
-        expr: SummaryExpr::SummaryEstimate { sketch_input, query },
+        expr: SummaryExpr::SummaryEstimate { summary_input, query },
         schema: dummy_l4_schema(),
     })
 }
@@ -164,7 +164,7 @@ fn is_ref(expr: &PhysicalExpr) -> bool {
 /// `boundary::summary_candidates`).
 fn quantile_kll_dag() -> PhysicalExpr {
     let q = QueryExpr::Aggregate {
-        by: GroupKeys::none(),
+        reduction: Reduction::by(vec![]),
         aggs: vec![crate::intent_algebra::AggIntent::Quantile {
             col: None,
             q: 0.99,
diff --git a/control_plane/src/physical/planner.rs b/control_plane/src/physical/planner.rs
index 65fec2226..bff56df67 100644
--- a/control_plane/src/physical/planner.rs
+++ b/control_plane/src/physical/planner.rs
@@ -290,7 +290,7 @@ fn plan_node(expr: &QueryExpr, config: &PhysicalPlannerConfig) -> PhysicalNode {
         //   * single other intent, no HAVING → sketch build, budget-placed
         //   * multi-intent or HAVING → exact HashAggregate at QueryEngine
         QueryExpr::Aggregate {
-            by,
+            reduction,
             aggs,
             having,
             child,
@@ -334,6 +334,9 @@ fn plan_node(expr: &QueryExpr, config: &PhysicalPlannerConfig) -> PhysicalNode {
             }
             // Multi-intent / HAVING aggregate → no single sketch can serve
             // it; fall back to an exact hash aggregation at the query engine.
+            // Always a genuine reduction (never per-entity -- see
+            // `intent_algebra::lower`'s single-intent-only per-entity rule).
+            let by = reduction.expect_reduce();
             let child = plan_node(child, config);
             let mut node = PhysicalNode {
                 op: PhysicalOp::HashAggregate {
@@ -588,7 +591,7 @@ mod tests {
     use crate::intent_algebra::relational::{
         default_cardinality, default_frequency, default_quantile,
     };
-    use crate::intent_algebra::{Schema, Source, WindowKind};
+    use crate::intent_algebra::{Reduction, Schema, Source, WindowKind};
     use crate::types_v2::AccuracyTarget;
 
     fn default_config() -> PhysicalPlannerConfig {
@@ -613,7 +616,7 @@ mod tests {
     /// canonical fold of the legacy `SketchAgg`.
     fn sketch_agg(intent: AggIntent, metric: &str) -> QueryExpr {
         QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction: Reduction::by(vec![]),
             aggs: vec![intent],
             output_names: Vec::new(),
             having: None,
@@ -695,7 +698,7 @@ mod tests {
     #[test]
     fn plan_topk_at_query_engine() {
         let expr = QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction: Reduction::by(vec![]),
             aggs: vec![AggIntent::TopK {
                 k: 10,
                 accuracy: AccuracyTarget::Epsilon(0.05),
@@ -714,7 +717,7 @@ mod tests {
         // TopK(QueryEngine) wrapping a sketch Aggregate(Agent) → Exchange
         // between them.
         let expr = QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction: Reduction::by(vec![]),
             aggs: vec![AggIntent::TopK {
                 k: 5,
                 accuracy: AccuracyTarget::Epsilon(0.05),
@@ -744,7 +747,7 @@ mod tests {
         // Multi-intent Aggregate → exact HashAggregate at QueryEngine
         // (no single sketch serves multiple intents).
         let expr = QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction: Reduction::by(vec![]),
             aggs: vec![AggIntent::Sum { col: None }, AggIntent::Min { col: None }],
             output_names: Vec::new(),
             having: None,
@@ -762,7 +765,7 @@ mod tests {
         // Backend, and the outer `TopK` intent runs at QueryEngine.
         // Should span: Agent → Backend → QueryEngine
         let expr = QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction: Reduction::by(vec![]),
             aggs: vec![AggIntent::TopK {
                 k: 10,
                 accuracy: AccuracyTarget::Epsilon(0.05),
diff --git a/control_plane/src/physical/window_fusion.rs b/control_plane/src/physical/window_fusion.rs
index bf42aab60..99845e1bf 100644
--- a/control_plane/src/physical/window_fusion.rs
+++ b/control_plane/src/physical/window_fusion.rs
@@ -46,7 +46,10 @@ pub struct FusedWindowSketch<'a> {
     pub window_size: Duration,
     /// Outer `Window`'s `slide` (`Some` only for `Sliding`).
     pub window_slide: Option<Duration>,
-    /// The inner `Aggregate`'s `by` columns.
+    /// The inner `Aggregate`'s grouping columns -- empty both when it's a
+    /// genuine empty-`by` reduction and when it's per-entity (no grouping
+    /// concept at all, ASAPController#163/#165); this field doesn't
+    /// distinguish the two, since nothing downstream currently needs to.
     pub by: &'a [ColumnId],
     /// The subtree below the inner `Aggregate` — the sketch's input.
     pub inner_child: &'a QueryExpr,
@@ -74,7 +77,7 @@ pub fn recognize_windowed_sketch(expr: &QueryExpr) -> Option<FusedWindowSketch<'
         return None;
     };
     let QueryExpr::Aggregate {
-        by,
+        reduction,
         aggs,
         having,
         child: inner_child,
@@ -88,6 +91,10 @@ pub fn recognize_windowed_sketch(expr: &QueryExpr) -> Option<FusedWindowSketch<'
     if aggs.len() != 1 || having.is_some() {
         return None;
     }
+    let by: &[ColumnId] = reduction
+        .group_keys()
+        .map(|keys| keys.keys())
+        .unwrap_or(&[]);
     Some(FusedWindowSketch {
         agg: &aggs[0],
         window_kind: kind.clone(),
@@ -208,7 +215,7 @@ mod tests {
             size,
             slide,
             child: Box::new(QueryExpr::Aggregate {
-                by: crate::intent_algebra::GroupKeys::none(),
+                reduction: crate::intent_algebra::Reduction::by(vec![]),
                 aggs: vec![agg],
                 output_names: Vec::new(),
                 having: None,
@@ -332,7 +339,7 @@ mod tests {
         // A canonical `Aggregate` with no enclosing `Window` is the unfused
         // sketch case — not a windowed sketch.
         let canonical = QueryExpr::Aggregate {
-            by: crate::intent_algebra::GroupKeys::none(),
+            reduction: crate::intent_algebra::Reduction::by(vec![]),
             aggs: vec![AggIntent::Sum { col: None }],
             output_names: Vec::new(),
             having: None,
diff --git a/control_plane/src/query_parser/mod.rs b/control_plane/src/query_parser/mod.rs
index 7e29e9ded..d58c4cda9 100644
--- a/control_plane/src/query_parser/mod.rs
+++ b/control_plane/src/query_parser/mod.rs
@@ -24,7 +24,7 @@ use std::time::Duration;
 
 use crate::intent_algebra::agg_intent::AggIntent;
 use crate::intent_algebra::query_expr::{Predicate, QueryExpr, Source};
-use crate::intent_algebra::{CompareOp, L3Expr, L3Scalar};
+use crate::intent_algebra::{ColumnId, CompareOp, L3Expr, L3Scalar};
 use crate::types::AggType;
 
 // ── Output types (legacy — consumed by analyzer and planner) ──────────────────
@@ -242,14 +242,20 @@ impl QeCollector {
                 self.visit(child, schema);
             }
             QueryExpr::Aggregate {
-                by, aggs, child, ..
+                reduction,
+                aggs,
+                child,
+                ..
             } => {
                 // The canonical IR folds legacy SketchAgg / WindowedAgg-inner
                 // / TopK / Aggregate into one variant carrying `AggIntent`s.
                 // `by` is positional — recover the group-by label *names*
                 // from the Binder schema (the legacy `Aggregate.keys` were
                 // names; multi-agg aggregates reach here with a non-empty
-                // `by` after the Binder resolves them).
+                // `by` after the Binder resolves them). A per-entity
+                // reduction (ASAPController#163/#165) has no `by` at all —
+                // same as an empty one here, no label names to recover.
+                let by: &[ColumnId] = reduction.group_keys().map(|k| k.keys()).unwrap_or(&[]);
                 if let Some(s) = schema {
                     for &id in by {
                         if let Some(col) = s.columns.get(id) {
@@ -562,7 +568,7 @@ mod doc_verify_all {
     // `Aggregate { by: [], .. }` stacked forms.
     use super::parse_query_expr_canonical;
     use crate::intent_algebra::query_expr::QueryExpr;
-    use crate::intent_algebra::AggIntent;
+    use crate::intent_algebra::{AggIntent, Reduction};
 
     #[test]
     fn example4_promql_quantile() {
@@ -571,11 +577,15 @@ mod doc_verify_all {
         )
         .unwrap();
         // Canonical fold of the legacy `WindowedAgg { Quantile }`:
-        // `Window { Aggregate { by: [], [Quantile] } }`.
+        // `Window { Aggregate { reduction: PerEntity, [Quantile] } }` — no
+        // explicit `by()`, and windowed with a single non-per-series intent,
+        // so there's no grouping concept at all (see #165's `Reduction`).
         match &expr {
             QueryExpr::Window { child, .. } => match child.as_ref() {
-                QueryExpr::Aggregate { by, aggs, .. } => {
-                    assert!(by.is_empty());
+                QueryExpr::Aggregate {
+                    reduction, aggs, ..
+                } => {
+                    assert!(matches!(reduction, Reduction::PerEntity));
                     assert!(matches!(aggs.as_slice(), [AggIntent::Quantile { .. }]));
                 }
                 other => panic!("expected Aggregate under Window, got {other:?}"),
diff --git a/control_plane/src/sketch_algebra/capability.rs b/control_plane/src/sketch_algebra/capability.rs
index 9fe06fd4f..912e8631b 100644
--- a/control_plane/src/sketch_algebra/capability.rs
+++ b/control_plane/src/sketch_algebra/capability.rs
@@ -614,8 +614,9 @@ pub fn capability_for(intent: &AggIntent) -> Option<Capability> {
 /// Translate `asap-plan`'s per-intent implementation decision into this
 /// repo's own [`Capability`] vocabulary.
 ///
-/// `Implementation::Sketch`/`ExactAccumulator` both carry an
-/// `asap_sketch::SummaryKind` — this repo's `Capability` groups those
+/// `Implementation::Summary` carries an `asap_sketch::SummaryKind` for
+/// both the approximate-sketch and exact-accumulator cases (told apart
+/// via `kind.is_exact()`) — this repo's `Capability` groups those
 /// into coarser families (`QuantileApprox`/`CardinalityApprox`/
 /// `FrequencyEstimate`/`FrequencyTopk` for sketches; `ExactAgg(AggregationType)`
 /// for accumulators) because that's the granularity the sketch index
@@ -630,31 +631,11 @@ fn implementation_to_capability(implementation: asap_plan::Implementation) -> Op
 
     match implementation {
         Implementation::PassThrough => None,
-        Implementation::Sketch { kind, .. } => match kind {
-            SummaryKind::Kll | SummaryKind::DDSketch => {
-                Some(Capability::QuantileApprox(SketchKindHandle::Any))
-            }
-            SummaryKind::Hll | SummaryKind::Theta | SummaryKind::Kmv => {
-                Some(Capability::CardinalityApprox)
-            }
-            SummaryKind::Cms | SummaryKind::CountSketch => {
-                Some(Capability::FrequencyEstimate(SketchKindHandle::Any))
-            }
-            SummaryKind::CmsWithHeap | SummaryKind::CountSketchWithHeap => {
-                Some(Capability::FrequencyTopk(SketchKindHandle::Any))
-            }
-            SummaryKind::Sum
-            | SummaryKind::Count
-            | SummaryKind::MinMax
-            | SummaryKind::Increase
-            | SummaryKind::Rate => {
-                unreachable!(
-                    "{kind:?} is an exact-accumulator SummaryKind, never returned inside \
-                     Implementation::Sketch by asap_plan::boundary::implementation_for"
-                )
-            }
-        },
-        Implementation::ExactAccumulator { kind, .. } => match kind {
+        // Exact accumulator half of the merged `Summary` variant
+        // (ASAPController#170 collapsed `Sketch`/`ExactAccumulator` into
+        // one `Summary { kind, params }`, recoverable via `kind.is_exact()`
+        // — see `asap_sketch::SummaryKind::is_exact`'s doc).
+        Implementation::Summary { kind, .. } if kind.is_exact() => match kind {
             SummaryKind::Sum => Some(Capability::ExactAgg(AggregationType::Sum)),
             SummaryKind::MinMax => Some(Capability::ExactAgg(AggregationType::MinMax)),
             SummaryKind::Increase | SummaryKind::Rate => {
@@ -667,11 +648,10 @@ fn implementation_to_capability(implementation: asap_plan::Implementation) -> Op
             // `count_over_time` query matched against a `Sum` policy
             // would silently return sum-of-values, not sample-count).
             // `asap_plan::boundary::implementation_for` still reports
-            // `Count{Exact}` as an `ExactAccumulator` (it assumes a real
-            // count accumulator exists, which is true in ASAPController's
-            // own reference implementation) — deliberately overridden
-            // here to `None` (archive) until a real
-            // `SumCountAccumulator` lands.
+            // `Count{Exact}` as exact (it assumes a real count
+            // accumulator exists, which is true in ASAPController's own
+            // reference implementation) — deliberately overridden here to
+            // `None` (archive) until a real `SumCountAccumulator` lands.
             SummaryKind::Count => None,
             SummaryKind::Kll
             | SummaryKind::DDSketch
@@ -683,8 +663,33 @@ fn implementation_to_capability(implementation: asap_plan::Implementation) -> Op
             | SummaryKind::CountSketch
             | SummaryKind::CountSketchWithHeap => {
                 unreachable!(
-                    "{kind:?} is a sketch-family SummaryKind, never returned inside \
-                     Implementation::ExactAccumulator by asap_plan::boundary::implementation_for"
+                    "{kind:?} is a sketch-family SummaryKind, so kind.is_exact() is false — \
+                     never reached inside the is_exact() guard"
+                )
+            }
+        },
+        // Approximate-sketch half.
+        Implementation::Summary { kind, .. } => match kind {
+            SummaryKind::Kll | SummaryKind::DDSketch => {
+                Some(Capability::QuantileApprox(SketchKindHandle::Any))
+            }
+            SummaryKind::Hll | SummaryKind::Theta | SummaryKind::Kmv => {
+                Some(Capability::CardinalityApprox)
+            }
+            SummaryKind::Cms | SummaryKind::CountSketch => {
+                Some(Capability::FrequencyEstimate(SketchKindHandle::Any))
+            }
+            SummaryKind::CmsWithHeap | SummaryKind::CountSketchWithHeap => {
+                Some(Capability::FrequencyTopk(SketchKindHandle::Any))
+            }
+            SummaryKind::Sum
+            | SummaryKind::Count
+            | SummaryKind::MinMax
+            | SummaryKind::Increase
+            | SummaryKind::Rate => {
+                unreachable!(
+                    "{kind:?} is an exact-accumulator SummaryKind, so kind.is_exact() is \
+                     true — the arm above already handles it"
                 )
             }
         },
diff --git a/control_plane/src/sketch_algebra/cost_model.rs b/control_plane/src/sketch_algebra/cost_model.rs
index 80e584efe..0bf18863b 100644
--- a/control_plane/src/sketch_algebra/cost_model.rs
+++ b/control_plane/src/sketch_algebra/cost_model.rs
@@ -277,7 +277,7 @@ impl CostModel for ControlPlaneCostModel {
             return Implementation::PassThrough;
         };
         let (width, depth) = Self::cms_width_depth(eps, delta);
-        Implementation::Sketch {
+        Implementation::Summary {
             kind: SummaryKind::Cms,
             params: SummaryParams::Cms {
                 width: width.next_power_of_two(),
diff --git a/control_plane/src/sketch_algebra/lower.rs b/control_plane/src/sketch_algebra/lower.rs
index ec8cb3103..6067b2e3a 100644
--- a/control_plane/src/sketch_algebra/lower.rs
+++ b/control_plane/src/sketch_algebra/lower.rs
@@ -87,7 +87,7 @@ fn bind_recursive(expr: &QueryExpr, accuracy: &AccuracyTarget) -> Result<L4Plan,
             child,
         } if matches!(child.as_ref(), QueryExpr::Aggregate { .. }) => {
             let QueryExpr::Aggregate {
-                by,
+                reduction,
                 aggs,
                 output_names,
                 having,
@@ -97,7 +97,7 @@ fn bind_recursive(expr: &QueryExpr, accuracy: &AccuracyTarget) -> Result<L4Plan,
                 unreachable!("guarded by the `matches!` above")
             };
             let pushed = QueryExpr::Aggregate {
-                by: by.clone(),
+                reduction: reduction.clone(),
                 aggs: aggs.clone(),
                 output_names: output_names.clone(),
                 having: having.clone(),
@@ -162,13 +162,13 @@ fn bind_recursive(expr: &QueryExpr, accuracy: &AccuracyTarget) -> Result<L4Plan,
 fn rewrite_rate_to_increase(expr: &QueryExpr) -> QueryExpr {
     match expr {
         QueryExpr::Aggregate {
-            by,
+            reduction,
             aggs,
             output_names,
             having,
             child,
         } => QueryExpr::Aggregate {
-            by: by.clone(),
+            reduction: reduction.clone(),
             aggs: aggs
                 .iter()
                 .map(|intent| {
diff --git a/control_plane/src/sketch_algebra/matcher.rs b/control_plane/src/sketch_algebra/matcher.rs
index 9463e624a..4f9d4c2e6 100644
--- a/control_plane/src/sketch_algebra/matcher.rs
+++ b/control_plane/src/sketch_algebra/matcher.rs
@@ -54,16 +54,22 @@ impl Matcher for SummaryFamilyMatcher {
     ///   but not the reverse (a heap-less sketch cannot enumerate top-k
     ///   items it never tracked).
     fn is_satisfied_by(&self, required: &Implementation, available: &Implementation) -> bool {
+        // ASAPController#170 merged `Sketch`/`ExactAccumulator` into one
+        // `Summary { kind, params }` variant, recoverable via
+        // `kind.is_exact()`. The variant-tag mismatch that used to fall
+        // through to `_ => false` (comparing a `Sketch` against an
+        // `ExactAccumulator`) is now an explicit `is_exact()` mismatch
+        // between the two sides, still falling through the same way.
         match (required, available) {
             (Implementation::PassThrough, _) => true,
             (
-                Implementation::ExactAccumulator { kind: required, .. },
-                Implementation::ExactAccumulator { kind: have, .. },
-            ) => required == have,
+                Implementation::Summary { kind: required, .. },
+                Implementation::Summary { kind: have, .. },
+            ) if required.is_exact() && have.is_exact() => required == have,
             (
-                Implementation::Sketch { kind: required, .. },
-                Implementation::Sketch { kind: have, .. },
-            ) => sketch_family_satisfied(required, have),
+                Implementation::Summary { kind: required, .. },
+                Implementation::Summary { kind: have, .. },
+            ) if !required.is_exact() && !have.is_exact() => sketch_family_satisfied(required, have),
             _ => false,
         }
     }
@@ -174,12 +180,12 @@ mod tests {
 
     fn sketch(kind: SummaryKind) -> Implementation {
         let params = params_for(&kind);
-        Implementation::Sketch { kind, params }
+        Implementation::Summary { kind, params }
     }
 
     fn accumulator(kind: SummaryKind) -> Implementation {
         let params = params_for(&kind);
-        Implementation::ExactAccumulator { kind, params }
+        Implementation::Summary { kind, params }
     }
 
     #[test]
diff --git a/control_plane/src/sketch_algebra/physical_expr.rs b/control_plane/src/sketch_algebra/physical_expr.rs
index ad0e6059d..237493512 100644
--- a/control_plane/src/sketch_algebra/physical_expr.rs
+++ b/control_plane/src/sketch_algebra/physical_expr.rs
@@ -200,7 +200,7 @@ mod tests {
     #[test]
     fn committed_wraps_an_implement_tree_result() {
         let q = QueryExpr::Aggregate {
-            by: crate::intent_algebra::GroupKeys::none(),
+            reduction: crate::intent_algebra::Reduction::by(vec![]),
             aggs: vec![crate::intent_algebra::AggIntent::Quantile {
                 col: None,
                 q: 0.99,
@@ -214,13 +214,13 @@ mod tests {
         let e = PhysicalExpr::committed(node);
         match e {
             PhysicalExpr::Committed(L4Plan::Summary(node)) => match &node.expr {
-                asap_sketch::SummaryExpr::SummaryEstimate { query, sketch_input } => {
+                asap_sketch::SummaryExpr::SummaryEstimate { query, summary_input } => {
                     assert!(matches!(query, asap_sketch::SketchQuery::Quantile { q } if *q == 0.99));
-                    match &sketch_input.expr {
+                    match &summary_input.expr {
                         asap_sketch::SummaryExpr::SummaryAgg {
-                            sketch, params, child, ..
+                            summary, params, child, ..
                         } => {
-                            assert_eq!(sketch, &SummaryKind::Kll);
+                            assert_eq!(summary, &SummaryKind::Kll);
                             assert_eq!(params, &SummaryParams::Kll { k: 200 });
                             assert!(matches!(child.expr, asap_sketch::SummaryExpr::Logical(_)));
                         }
diff --git a/control_plane/src/sketch_algebra/tests.rs b/control_plane/src/sketch_algebra/tests.rs
index 1fd9d55f8..4b060b309 100644
--- a/control_plane/src/sketch_algebra/tests.rs
+++ b/control_plane/src/sketch_algebra/tests.rs
@@ -10,7 +10,7 @@ use asap_sketch::{L4Node, SketchQuery, SummaryExpr, SummaryKind, SummaryParams};
 
 use crate::intent_algebra::schema::{Column, DataType};
 use crate::intent_algebra::{
-    AggIntent, BindingScope, LabelFilter, QueryExpr, Schema, Source, WindowKind,
+    AggIntent, BindingScope, LabelFilter, QueryExpr, Reduction, Schema, Source, WindowKind,
 };
 use crate::sketch_algebra::cost_model::ForcedFamilyCostModel;
 use crate::sketch_algebra::lower::bind_query_expr;
@@ -64,7 +64,10 @@ fn windowed_scan() -> QueryExpr {
 
 fn agg_quantile(q: f64, accuracy: AccuracyTarget) -> QueryExpr {
     QueryExpr::Aggregate {
-        by: crate::intent_algebra::GroupKeys::none(),
+        // No `by()` and windowed (child is `Window`) — the shape
+        // `quantile_over_time(...)` lowers to: per-series, not a
+        // cross-series reduction (see #165's `Reduction`).
+        reduction: Reduction::PerEntity,
         aggs: vec![AggIntent::Quantile {
             col: None,
             q,
@@ -106,7 +109,7 @@ fn node_is_archive(node: &Rc<L4Node>) -> bool {
             _ => false,
         },
         SummaryExpr::SummaryAgg { child, .. } => node_is_archive(child),
-        SummaryExpr::SummaryEstimate { sketch_input, .. } => node_is_archive(sketch_input),
+        SummaryExpr::SummaryEstimate { summary_input, .. } => node_is_archive(summary_input),
         SummaryExpr::SummaryMerge { children } => children.iter().any(node_is_archive),
         SummaryExpr::SummaryJoin { outer, inner, .. } => {
             node_is_archive(outer) || node_is_archive(inner)
@@ -114,7 +117,7 @@ fn node_is_archive(node: &Rc<L4Node>) -> bool {
         SummaryExpr::SummarySubtract { left, right } => {
             node_is_archive(left) || node_is_archive(right)
         }
-        SummaryExpr::SummaryDelete { sketch_input, .. } => node_is_archive(sketch_input),
+        SummaryExpr::SummaryDelete { summary_input, .. } => node_is_archive(summary_input),
     }
 }
 
@@ -135,17 +138,17 @@ fn bind_kll_quantile_basic() {
     match &node.expr {
         SummaryExpr::SummaryEstimate {
             query,
-            sketch_input,
+            summary_input,
         } => {
             assert!(matches!(query, SketchQuery::Quantile { q } if *q == 0.99));
-            match &sketch_input.expr {
+            match &summary_input.expr {
                 SummaryExpr::SummaryAgg {
-                    sketch,
+                    summary,
                     params,
                     child,
                     ..
                 } => {
-                    assert_eq!(sketch, &SummaryKind::Kll);
+                    assert_eq!(summary, &SummaryKind::Kll);
                     assert_eq!(params, &SummaryParams::Kll { k: 200 });
                     assert!(matches!(child.expr, SummaryExpr::Logical(_)));
                 }
@@ -169,12 +172,12 @@ fn bind_ddsketch_quantile_basic() {
     match &node.expr {
         SummaryExpr::SummaryEstimate {
             query,
-            sketch_input,
+            summary_input,
         } => {
             assert!(matches!(query, SketchQuery::Quantile { q } if *q == 0.99));
-            match &sketch_input.expr {
-                SummaryExpr::SummaryAgg { sketch, params, .. } => {
-                    assert_eq!(sketch, &SummaryKind::DDSketch);
+            match &summary_input.expr {
+                SummaryExpr::SummaryAgg { summary, params, .. } => {
+                    assert_eq!(summary, &SummaryKind::DDSketch);
                     match params {
                         SummaryParams::DDSketch { alpha } => {
                             assert!((alpha - 0.01).abs() < 1e-12)
@@ -200,10 +203,10 @@ fn bind_picks_ddsketch_over_kll_when_eps_explicit() {
         .expect("bind_query_expr should not error");
     match bound {
         PhysicalExpr::Committed(L4Plan::Summary(node)) => match &node.expr {
-            SummaryExpr::SummaryEstimate { sketch_input, .. } => match &sketch_input.expr {
-                SummaryExpr::SummaryAgg { sketch, .. } => {
+            SummaryExpr::SummaryEstimate { summary_input, .. } => match &summary_input.expr {
+                SummaryExpr::SummaryAgg { summary, .. } => {
                     assert_eq!(
-                        sketch,
+                        summary,
                         &SummaryKind::DDSketch,
                         "dispatcher should pick DDSketch over KLL on ε-driven Quantile"
                     );
@@ -219,7 +222,9 @@ fn bind_picks_ddsketch_over_kll_when_eps_explicit() {
 /// Build an `Aggregate{TopK{k, accuracy}}` over the windowed scan.
 fn agg_topk(k: usize, accuracy: AccuracyTarget) -> QueryExpr {
     QueryExpr::Aggregate {
-        by: vec![].into(),
+        // A ranking always reduces — empty `by` ranks the whole input,
+        // never per-entity (see `lower.rs`'s `LQueryExpr::TopK` handling).
+        reduction: Reduction::by(vec![]),
         aggs: vec![AggIntent::TopK { k, accuracy }],
         output_names: Vec::new(),
         having: None,
@@ -237,16 +242,16 @@ fn topk_binding_family(bound: &PhysicalExpr) -> (SummaryKind, u32, u32) {
         PhysicalExpr::Committed(L4Plan::Summary(node)) => match &node.expr {
             SummaryExpr::SummaryEstimate {
                 query,
-                sketch_input,
+                summary_input,
             } => {
                 assert!(matches!(query, SketchQuery::TopK { k } if *k == 10));
-                match &sketch_input.expr {
-                    SummaryExpr::SummaryAgg { sketch, params, .. } => match params {
+                match &summary_input.expr {
+                    SummaryExpr::SummaryAgg { summary, params, .. } => match params {
                         SummaryParams::CmsWithHeap { width, depth, .. } => {
-                            (sketch.clone(), *width, *depth)
+                            (summary.clone(), *width, *depth)
                         }
                         SummaryParams::CountSketchWithHeap { width, depth, .. } => {
-                            (sketch.clone(), *width, *depth)
+                            (summary.clone(), *width, *depth)
                         }
                         other => {
                             panic!("expected CmsWithHeap/CountSketchWithHeap params, got {other:?}")
@@ -290,7 +295,7 @@ fn bind_cms_topk_loose_recall_picks_cms_heap() {
 /// the old fixture used `AggIntent::TopK{accuracy: Exact}` (the intent's
 /// OWN accuracy) to signal "tight/exact-recall". Under
 /// `asap_plan::boundary::implementation_for_with`, the per-intent
-/// sketch-vs-exact boundary decision checks the intent's own `accuracy`
+/// summary-vs-exact boundary decision checks the intent's own `accuracy`
 /// field FIRST: `TopK{accuracy: Exact}` now declines to bind at all
 /// (`SummaryExpr::Logical`) rather than reaching the cost model's
 /// family-selection logic at all — see `topk_exact_accuracy_declines_to_bind`
@@ -355,7 +360,7 @@ fn bind_cms_topk_picks_cost_min_meeting_sla() {
 #[test]
 fn bind_hll_cardinality_basic() {
     let expr = QueryExpr::Aggregate {
-        by: vec![].into(),
+        reduction: Reduction::PerEntity,
         aggs: vec![AggIntent::Cardinality {
             col: None,
             accuracy: AccuracyTarget::Epsilon(0.01),
@@ -369,12 +374,12 @@ fn bind_hll_cardinality_basic() {
         PhysicalExpr::Committed(L4Plan::Summary(node)) => match &node.expr {
             SummaryExpr::SummaryEstimate {
                 query,
-                sketch_input,
+                summary_input,
             } => {
                 assert!(matches!(query, SketchQuery::Cardinality));
-                match &sketch_input.expr {
-                    SummaryExpr::SummaryAgg { sketch, params, .. } => {
-                        assert_eq!(sketch, &SummaryKind::Hll);
+                match &summary_input.expr {
+                    SummaryExpr::SummaryAgg { summary, params, .. } => {
+                        assert_eq!(summary, &SummaryKind::Hll);
                         match params {
                             SummaryParams::Hll { precision } => {
                                 assert!(
@@ -396,7 +401,7 @@ fn bind_hll_cardinality_basic() {
 
 #[test]
 fn sum_now_binds_to_exact_agg_after_pr_6_followup() {
-    // `AggIntent::Sum` binds to a bare `SummaryAgg` with `sketch:
+    // `AggIntent::Sum` binds to a bare `SummaryAgg` with `summary:
     // SummaryKind::Sum` and no `SummaryEstimate` wrapper (the partial
     // state *is* the value — see `asap_plan::bind`'s module docs). The
     // old locally-defined `PhysicalExpr::ExactAgg { agg_type, .. }`
@@ -406,7 +411,7 @@ fn sum_now_binds_to_exact_agg_after_pr_6_followup() {
     // node shape, keyed by `SummaryKind` (see `physical_expr.rs`'s
     // module docs).
     let expr = QueryExpr::Aggregate {
-        by: vec![].into(),
+        reduction: Reduction::PerEntity,
         aggs: vec![AggIntent::Sum { col: None }],
         output_names: Vec::new(),
         having: None,
@@ -415,9 +420,9 @@ fn sum_now_binds_to_exact_agg_after_pr_6_followup() {
     let bound = bind_query_expr(&expr, AccuracyTarget::Exact).expect("no error");
     match bound {
         PhysicalExpr::Committed(L4Plan::Summary(node)) => match &node.expr {
-            SummaryExpr::SummaryAgg { sketch, params, .. } => {
+            SummaryExpr::SummaryAgg { summary, params, .. } => {
                 assert_eq!(
-                    sketch,
+                    summary,
                     &SummaryKind::Sum,
                     "Sum should bind to SummaryAgg(Sum)"
                 );
@@ -433,14 +438,14 @@ fn sum_now_binds_to_exact_agg_after_pr_6_followup() {
 fn bind_exact_accuracy_disables_quantile_binding() {
     // Quantile under `AccuracyTarget::Exact` should NOT bind — the
     // optimizer falls back to an exact path. (Per design.md §6 line
-    // ~1254 — "the sketch path is selected, not mandated".)
+    // ~1254 — "the summary path is selected, not mandated".)
     let expr = agg_quantile(0.99, AccuracyTarget::Exact);
     let bound = bind_query_expr(&expr, AccuracyTarget::Exact).expect("no error");
     match bound {
         PhysicalExpr::Committed(L4Plan::Summary(node)) => {
             assert!(
                 matches!(&node.expr, SummaryExpr::Logical(qe) if matches!(**qe, QueryExpr::Aggregate { .. })),
-                "Exact accuracy should disable sketch binding and pass through as Logical, got {:?}",
+                "Exact accuracy should disable summary binding and pass through as Logical, got {:?}",
                 node.expr
             );
         }
@@ -456,13 +461,13 @@ fn bind_exact_accuracy_disables_quantile_binding() {
 // matching binding without going back through asap-planner-rs.
 
 /// `ONLY_TEMPORAL` — `quantile_over_time(0.99, m[5m])`.
-/// asap-planner-rs path: ONLY_TEMPORAL pattern 1 → KLL/DDSketch sketch.
+/// asap-planner-rs path: ONLY_TEMPORAL pattern 1 → KLL/DDSketch summary.
 /// Control plane path: `Aggregate{Quantile{0.99}}` over `Window` →
 /// binds a quantile-capable family → `SummaryAgg{KLL/DDSketch}`.
 #[test]
 fn phase_b_pattern_only_temporal_quantile_binds_to_sketch() {
     let expr = QueryExpr::Aggregate {
-        by: vec![].into(),
+        reduction: Reduction::PerEntity,
         aggs: vec![AggIntent::Quantile {
             col: None,
             q: 0.99,
@@ -477,12 +482,12 @@ fn phase_b_pattern_only_temporal_quantile_binds_to_sketch() {
         PhysicalExpr::Committed(L4Plan::Summary(node)) => match &node.expr {
             SummaryExpr::SummaryEstimate {
                 query,
-                sketch_input,
+                summary_input,
             } => {
                 assert!(matches!(query, SketchQuery::Quantile { .. }));
-                match &sketch_input.expr {
-                    SummaryExpr::SummaryAgg { sketch, .. } => {
-                        assert!(matches!(sketch, SummaryKind::Kll | SummaryKind::DDSketch));
+                match &summary_input.expr {
+                    SummaryExpr::SummaryAgg { summary, .. } => {
+                        assert!(matches!(summary, SummaryKind::Kll | SummaryKind::DDSketch));
                     }
                     other => panic!("expected SummaryAgg under SummaryEstimate, got {other:?}"),
                 }
@@ -497,13 +502,13 @@ fn phase_b_pattern_only_temporal_quantile_binds_to_sketch() {
 /// variants that legacy `single_query.rs` accepts).
 ///
 /// Control plane path: `Aggregate{Sum}` over `Window` → binds to a bare
-/// `SummaryAgg{sketch: SummaryKind::Sum}` (an exact mergeable
+/// `SummaryAgg{summary: SummaryKind::Sum}` (an exact mergeable
 /// accumulator — see `sum_now_binds_to_exact_agg_after_pr_6_followup`'s
 /// doc comment for the `ExactAgg` → `SummaryAgg` unification).
 #[test]
 fn phase_b_pattern_only_temporal_sum_binds_to_exact_agg() {
     let expr = QueryExpr::Aggregate {
-        by: vec![].into(),
+        reduction: Reduction::PerEntity,
         aggs: vec![AggIntent::Sum { col: None }],
         output_names: Vec::new(),
         having: None,
@@ -512,8 +517,8 @@ fn phase_b_pattern_only_temporal_sum_binds_to_exact_agg() {
     let bound = bind_query_expr(&expr, AccuracyTarget::Epsilon(0.01)).unwrap();
     match bound {
         PhysicalExpr::Committed(L4Plan::Summary(node)) => match &node.expr {
-            SummaryExpr::SummaryAgg { sketch, .. } => {
-                assert_eq!(sketch, &SummaryKind::Sum);
+            SummaryExpr::SummaryAgg { summary, .. } => {
+                assert_eq!(summary, &SummaryKind::Sum);
             }
             other => panic!("expected SummaryAgg(Sum), got {other:?}"),
         },
@@ -532,7 +537,7 @@ fn phase_b_pattern_only_temporal_sum_binds_to_exact_agg() {
 #[test]
 fn phase_b_pattern_only_spatial_aggregate_binds_to_multiple_sum() {
     let expr = QueryExpr::Aggregate {
-        by: vec![1].into(), // service column
+        reduction: Reduction::by(vec![1]), // service column
         aggs: vec![AggIntent::Sum { col: None }],
         output_names: Vec::new(),
         having: None,
@@ -541,11 +546,13 @@ fn phase_b_pattern_only_spatial_aggregate_binds_to_multiple_sum() {
     let bound = bind_query_expr(&expr, AccuracyTarget::Epsilon(0.01)).unwrap();
     match bound {
         PhysicalExpr::Committed(L4Plan::Summary(node)) => match &node.expr {
-            SummaryExpr::SummaryAgg { sketch, by, .. } => {
-                assert_eq!(sketch, &SummaryKind::Sum);
+            SummaryExpr::SummaryAgg {
+                summary, reduction, ..
+            } => {
+                assert_eq!(summary, &SummaryKind::Sum);
                 assert_eq!(
-                    by,
-                    &vec![1],
+                    reduction.group_keys().map(|k| k.keys()),
+                    Some(&[1][..]),
                     "keyed sum must carry the group-by column (the MultipleSum-equivalent signal)"
                 );
             }
@@ -565,7 +572,7 @@ fn phase_b_pattern_only_spatial_aggregate_binds_to_multiple_sum() {
 #[test]
 fn phase_b_pattern_temporal_and_spatial_combined_binds_to_multiple_increase() {
     let expr = QueryExpr::Aggregate {
-        by: vec![1].into(),
+        reduction: Reduction::by(vec![1]),
         aggs: vec![AggIntent::Rate],
         output_names: Vec::new(),
         having: None,
@@ -574,9 +581,11 @@ fn phase_b_pattern_temporal_and_spatial_combined_binds_to_multiple_increase() {
     let bound = bind_query_expr(&expr, AccuracyTarget::Epsilon(0.01)).unwrap();
     match bound {
         PhysicalExpr::Committed(L4Plan::Summary(node)) => match &node.expr {
-            SummaryExpr::SummaryAgg { sketch, by, .. } => {
-                assert_eq!(sketch, &SummaryKind::Increase);
-                assert_eq!(by, &vec![1]);
+            SummaryExpr::SummaryAgg {
+                summary, reduction, ..
+            } => {
+                assert_eq!(summary, &SummaryKind::Increase);
+                assert_eq!(reduction.group_keys().map(|k| k.keys()), Some(&[1][..]));
             }
             other => panic!("expected SummaryAgg(Increase, by=[1]), got {other:?}"),
         },
@@ -603,7 +612,7 @@ fn phase_b_pattern_archive_only_routes_to_archive() {
         "Phase β intent must flag archive"
     );
     let expr = QueryExpr::Aggregate {
-        by: vec![].into(),
+        reduction: Reduction::PerEntity,
         aggs: vec![intent.clone()],
         output_names: Vec::new(),
         having: None,
@@ -665,7 +674,7 @@ fn phase_b_e2e_quantile_over_time_binds_to_quantile_sketch() {
     let kind = crate::emit::extract_root_sketch_kind(&bound);
     assert!(
         matches!(kind, Some(SummaryKind::Kll) | Some(SummaryKind::DDSketch)),
-        "expected quantile sketch family, got {kind:?}"
+        "expected quantile summary family, got {kind:?}"
     );
     assert!(
         !binding_is_archive(&bound),
@@ -674,9 +683,9 @@ fn phase_b_e2e_quantile_over_time_binds_to_quantile_sketch() {
 }
 
 /// `sum_over_time.yaml` — the legacy planner produces an exact-sum
-/// aggregation row (no sketch). Control plane path: `Aggregate{Sum}` over
+/// aggregation row (no summary). Control plane path: `Aggregate{Sum}` over
 /// `Window` → binds to an exact accumulator (`SummaryAgg{Sum}`), which is
-/// neither an approximate sketch (so `extract_root_sketch_kind`, which
+/// neither an approximate summary (so `extract_root_sketch_kind`, which
 /// excludes exact accumulators — see its doc comment — returns `None`)
 /// nor archive-routed.
 #[test]
@@ -687,7 +696,7 @@ fn phase_b_e2e_sum_over_time_falls_through_to_logical() {
     );
     assert!(
         crate::emit::extract_root_sketch_kind(&bound).is_none(),
-        "sum_over_time should not produce an approximate sketch"
+        "sum_over_time should not produce an approximate summary"
     );
     assert!(
         !binding_is_archive(&bound),
@@ -699,7 +708,7 @@ fn phase_b_e2e_sum_over_time_falls_through_to_logical() {
 /// temporal aggregation; the legacy planner emits an exact-sum row keyed
 /// on the by-label. Control plane path: `Aggregate{Sum, by=[…]}` over
 /// `Window` → binds to an exact accumulator (`SummaryAgg{Sum, by=[…]}`) —
-/// no approximate sketch family. The by-label is preserved on the L3
+/// no approximate summary family. The by-label is preserved on the L3
 /// group-by-id list, which Phase α's routing emit reads to build the
 /// per-label rollup partition.
 #[test]
@@ -708,7 +717,7 @@ fn phase_b_e2e_sum_by_preserves_grouping_label() {
         "sum by (instance) (sum_over_time(http_requests_total[5m]))",
         AccuracyTarget::Epsilon(0.01),
     );
-    // No approximate sketch family for plain Sum.
+    // No approximate summary family for plain Sum.
     assert!(crate::emit::extract_root_sketch_kind(&bound).is_none());
     // The end shape may carry `Logical(Aggregate{by, ...})` beneath a
     // `SummaryAgg{Sum}` wrapper, or `Logical(Window{...})` when the
@@ -733,8 +742,8 @@ fn phase_b_e2e_sum_by_preserves_grouping_label() {
 /// `rate_increase.yaml` — the legacy planner emits a MultipleIncrease
 /// (counter-reset adjusted) row. Control plane path: `Aggregate{Rate}` over
 /// `Window` → `bind_query_expr` rewrites `Rate` to `Increase` and binds an
-/// exact accumulator (`SummaryAgg{Increase}`) — no approximate sketch
-/// family. Both paths produce a single non-sketch streaming row; the L5
+/// exact accumulator (`SummaryAgg{Increase}`) — no approximate summary
+/// family. Both paths produce a single non-summary streaming row; the L5
 /// emitter is the one that picks the actual MultipleIncrease processor.
 #[test]
 fn phase_b_e2e_rate_falls_through_to_logical() {
@@ -755,7 +764,7 @@ fn phase_b_e2e_rate_falls_through_to_logical() {
 /// At the time of writing, the control plane's `parse_query` may flatten
 /// `topk` differently (no `inside_topk` propagation through TopK +
 /// nested aggregate). The test asserts the END-STATE: either a
-/// CountSketch sketch fired, OR a Logical pass-through (which Phase γ
+/// CountSketch summary fired, OR a Logical pass-through (which Phase γ
 /// can decide whether to refine). The contract Phase β cares about is
 /// that the bound expression is well-formed.
 #[test]
@@ -787,7 +796,7 @@ fn phase_b_e2e_topk_well_formed() {
 fn phase_b_e2e_archive_only_e2e_binding() {
     let intent = AggIntent::Absent;
     let expr = QueryExpr::Aggregate {
-        by: vec![].into(),
+        reduction: Reduction::PerEntity,
         aggs: vec![intent.clone()],
         output_names: Vec::new(),
         having: None,
@@ -798,7 +807,7 @@ fn phase_b_e2e_archive_only_e2e_binding() {
         binding_is_archive(&bound),
         "archive-only intent must surface archive flag through L4 binding"
     );
-    // No approximate sketch fires for archive-only intents.
+    // No approximate summary fires for archive-only intents.
     assert!(crate::emit::extract_root_sketch_kind(&bound).is_none());
 }
 
@@ -830,7 +839,7 @@ fn phase_b_archive_only_intents_round_trip_through_binder() {
     ];
     for intent in intents {
         let expr = QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction: Reduction::PerEntity,
             aggs: vec![intent.clone()],
             output_names: Vec::new(),
             having: None,
@@ -861,11 +870,11 @@ fn phase_b_archive_only_intents_round_trip_through_binder() {
 //
 // `AggIntent::Extension` (this deployment's `Frequency` point-query,
 // built via `crate::intent_algebra::frequency(accuracy, item)`) now binds to a
-// real `Cms` sketch via `ControlPlaneCostModel::realize_extension`/
+// real `Cms` summary 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`.
 // `AggIntent::TopK { accuracy: Exact }` still declines to bind
-// (`SummaryExpr::Logical`) rather than sketch — a REAL, accepted
+// (`SummaryExpr::Logical`) rather than summary — a REAL, accepted
 // behavior change from this migration that remains open
 // (`TopK{Exact}`'s `exact_realization` has no accumulator form for it —
 // see `lower.rs`'s module docs and `cost_model.rs`'s module docs,
@@ -875,10 +884,10 @@ fn phase_b_archive_only_intents_round_trip_through_binder() {
 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`.
+    // as a real `Cms` summary instead of declining to `Logical`.
     let intent = crate::intent_algebra::frequency(AccuracyTarget::Epsilon(0.01), None);
     let expr = QueryExpr::Aggregate {
-        by: vec![].into(),
+        reduction: Reduction::PerEntity,
         aggs: vec![intent],
         output_names: Vec::new(),
         having: None,
@@ -888,7 +897,7 @@ fn frequency_extension_binds_cms() {
     match bound {
         PhysicalExpr::Committed(L4Plan::Summary(node)) => {
             let SummaryExpr::SummaryEstimate {
-                sketch_input,
+                summary_input,
                 query,
             } = &node.expr
             else {
@@ -896,14 +905,14 @@ fn frequency_extension_binds_cms() {
             };
             assert!(
                 matches!(
-                    &sketch_input.expr,
+                    &summary_input.expr,
                     SummaryExpr::SummaryAgg {
-                        sketch: SummaryKind::Cms,
+                        summary: SummaryKind::Cms,
                         ..
                     }
                 ),
                 "expected a Cms SummaryAgg, got {:?}",
-                sketch_input.expr
+                summary_input.expr
             );
             assert!(
                 matches!(
diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml
index 11c54ca7a..e343d8a0d 100644
--- a/crates/asap_types/Cargo.toml
+++ b/crates/asap_types/Cargo.toml
@@ -16,7 +16,7 @@ xxhash-rust = { version = "0.8", features = ["xxh64"] }
 # via this crate): WindowType -> asap_ir::intent_algebra::query_expr::WindowKind
 # unification (scratchpad/artifacts/enum-unification-plan.md). Pin matches
 # control_plane's -- see control_plane/Cargo.toml's comment for the rationale.
-asap-ir = { git = "https://github.com/ProjectASAP/ASAPController", rev = "fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4" }
+asap-ir = { git = "https://github.com/ProjectASAP/ASAPController", rev = "cc18c9872bbaf0cadf43566892c3974ec9948eba" }
 # Step 5 of the sketch-identity unification (see
 # scratchpad/artifacts/enum-unification-plan.md): `AccumulatorSpec`
 # (accumulator_spec.rs) converges data_plane's identity representation
@@ -25,4 +25,4 @@ asap-ir = { git = "https://github.com/ProjectASAP/ASAPController", rev = "fc09c3
 # control_plane's pin exactly (`control_plane/Cargo.toml`) -- two
 # different revs of the same git dependency in one workspace resolve
 # to two distinct Rust types that won't unify.
-asap-sketch = { git = "https://github.com/ProjectASAP/ASAPController", rev = "fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4" }
+asap-sketch = { git = "https://github.com/ProjectASAP/ASAPController", rev = "cc18c9872bbaf0cadf43566892c3974ec9948eba" }
diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml
index 02ff4e353..87a1882db 100644
--- a/data_plane/Cargo.toml
+++ b/data_plane/Cargo.toml
@@ -29,12 +29,12 @@ control_plane = { path = "../control_plane" }
 # control_plane's own pin) to pick up
 # `SketchQuery::PointCount.value: Option<String>`, needed for the
 # named-key PointCount readout.
-asap-sketch = { git = "https://github.com/ProjectASAP/ASAPController", rev = "64df20d90c3ddd519c726dea45c05e1fe5225ce6" }
+asap-sketch = { git = "https://github.com/ProjectASAP/ASAPController", rev = "cc18c9872bbaf0cadf43566892c3974ec9948eba" }
 # `asap_sketch::L4Node`'s own fields (`SummaryExpr::Logical(Box<QueryExpr>)`,
-# `SummaryAgg { col: ColumnRef, by: Vec<ColumnId>, .. }`) are `asap-ir`
+# `SummaryAgg { col: ColumnRef, reduction: Reduction, .. }`) are `asap-ir`
 # types, not re-exported by `asap-sketch` -- `find_candidates` needs to
 # walk/match them directly. Same pin-must-match rule as `asap-sketch` above.
-asap-ir = { git = "https://github.com/ProjectASAP/ASAPController", rev = "64df20d90c3ddd519c726dea45c05e1fe5225ce6" }
+asap-ir = { git = "https://github.com/ProjectASAP/ASAPController", rev = "cc18c9872bbaf0cadf43566892c3974ec9948eba" }
 
 # Shared external (workspace)
 serde.workspace = true
diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs
index 547c6a5bb..570d51b15 100644
--- a/data_plane/src/query_engines/asap_query_engine/engine.rs
+++ b/data_plane/src/query_engines/asap_query_engine/engine.rs
@@ -896,6 +896,14 @@ impl ASAPQueryEngine {
             )
         })?;
 
+        // Shadow-mode comparison against the new SummaryExecutor path —
+        // see `data_plane/docs/l4node-plan-executor-design.md`'s
+        // "Rollout" section. No-op unless `ASAP_SHADOW_SUMMARY_EXECUTOR`
+        // is set; never affects `result`/the response below.
+        crate::query_engines::asap_query_engine::shadow_compare::maybe_shadow_compare(
+            idx, query, start_ms, end_ms, false, &result,
+        );
+
         // Matrix shape — the range_query wire format requires it.
         let warm_qr = asap_tier_result_to_query_result(result.clone(), end_ms, true);
 
@@ -1883,15 +1891,24 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu
                 // adapter's `format_success_response` rejects with
                 // a 500 ”shape mismatch” / empty-body response.
                 let _ = any_range_candidate;
+                let stitch_t0 = if combined_t0 == u64::MAX {
+                    now_ms.saturating_sub(DEFAULT_LOOKBACK_MS)
+                } else {
+                    combined_t0
+                };
+                // Shadow-mode comparison against the new SummaryExecutor
+                // path — see
+                // `data_plane/docs/l4node-plan-executor-design.md`'s
+                // "Rollout" section. No-op unless
+                // `ASAP_SHADOW_SUMMARY_EXECUTOR` is set; never affects
+                // `result`/the response below.
+                crate::query_engines::asap_query_engine::shadow_compare::maybe_shadow_compare(
+                    idx, query, stitch_t0, now_ms, true, &result,
+                );
                 let warm_qr = asap_tier_result_to_query_result(result.clone(), now_ms, false);
                 if let (Some((cov_lo, cov_hi)), Some(archive)) =
                     (result.coverage, self.archive_engine.as_ref())
                 {
-                    let stitch_t0 = if combined_t0 == u64::MAX {
-                        now_ms.saturating_sub(DEFAULT_LOOKBACK_MS)
-                    } else {
-                        combined_t0
-                    };
                     if cov_lo > stitch_t0 || cov_hi < now_ms {
                         let archive_qr = archive.execute(query).await;
                         if let Ok(archive_qr) = archive_qr {
diff --git a/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs b/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs
new file mode 100644
index 000000000..56f0364de
--- /dev/null
+++ b/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs
@@ -0,0 +1,191 @@
+//! PromQL string → `asap_sketch::L4Node` bridge for shadow-mode
+//! `SummaryExecutor` comparison. See
+//! `data_plane/docs/l4node-plan-executor-design.md`'s "Rollout" section
+//! for the full design and why this calls
+//! `control_plane::sketch_algebra::lower::bind_query_expr`
+//! (`ControlPlaneCostModel`) rather than
+//! `control_plane::asap_tier_implement::implement_promql_for_asap_tier`
+//! (`DefaultCostModel`, which can't realize the Frequency intent at all).
+//!
+//! `control_plane` runs in-process with `data_plane` in this deployment
+//! (see `data_plane/Cargo.toml`'s "Phase 9" comment), so this is a
+//! same-binary library call, not a new planning implementation living
+//! here.
+
+use std::rc::Rc;
+
+use asap_sketch::{L4Node, SummaryExpr};
+
+use control_plane::sketch_algebra::capability::OuterFn;
+use control_plane::sketch_algebra::{BindingError, L4Plan, PhysicalExpr};
+use control_plane::types_v2::AccuracyTarget;
+
+/// Why `lower_promql_to_l4node` didn't produce a comparable `L4Node`.
+/// None of these are errors in the alarming sense — every variant is an
+/// expected, frequent outcome for *some* fraction of live traffic; the
+/// caller's only obligation is "don't attempt a shadow comparison," never
+/// "log this as a problem."
+#[derive(Debug)]
+pub enum LoweringSkip {
+    /// `parse_query_expr_canonical` failed — same failure mode the legacy
+    /// `analyze_promql_for_asap_tier` path already tolerates.
+    ParseFailed(String),
+    /// The query contains `rate(...)`/`irate(...)` (`OuterFn::Rate`, per
+    /// `control_plane::asap_tier_analysis`'s own candidate analysis).
+    /// `lower.rs`'s `bind_recursive` rewrites `AggIntent::Rate` →
+    /// `Increase` before binding, so this WOULD otherwise bind
+    /// successfully to a valid `SummaryAgg{Increase}` tree — but
+    /// `summary_executor.rs` has no rate-division logic (dividing by a
+    /// coverage-clamped range), so comparing against it would produce a
+    /// spurious mismatch, not a real one. Must be excluded before ever
+    /// calling into `control_plane`'s binder, not just deprioritized.
+    RateShape,
+    /// `bind_query_expr` itself failed (a genuine `BindingError`, e.g.
+    /// L3→L4 schema-derivation failure).
+    Implement(String),
+    /// `bind_query_expr` returned a `PhysicalExpr` variant other than
+    /// `Committed(L4Plan::Summary(_))`. Per `bind_query_expr`'s own doc
+    /// this shouldn't happen in practice (it never picks a Phase ε.1
+    /// placement), but the match is kept exhaustive and defensive rather
+    /// than assuming.
+    UnsupportedPhysicalShape,
+    /// The root node is `SummaryExpr::Logical(_)` — the query didn't
+    /// realize to any sketch/exact-agg binding at all (e.g.
+    /// `topk(K, sum by(...)(rate(m[r])))`: `implement_tree_in_with` only
+    /// recurses through `Aggregate` nodes, so hitting the outer
+    /// `Sort`/`Limit` wraps the WHOLE tree as one opaque `Logical` blob
+    /// even though the inner aggregate would bind fine on its own — see
+    /// this crate's design doc). Not an error: this is exactly the
+    /// existing `SummaryExecutorError::Logical`/"no candidate bound"
+    /// outcome, just detected one step earlier so the caller can skip
+    /// without even constructing a `QueryExecutionContext`.
+    NotRealized,
+}
+
+/// Lower a raw PromQL query string to the `L4Node` tree
+/// `asap_sketch::exec::execute`/`SummaryExecutor` needs, for shadow-mode
+/// comparison against the legacy `SketchReducer` path. Returns `Err` for
+/// any shape shadow-mode shouldn't attempt (parse failure, `rate()`,
+/// or anything that doesn't realize to a concrete sketch/exact-agg
+/// binding) — see `LoweringSkip`'s variants.
+pub fn lower_promql_to_l4node(
+    query: &str,
+    accuracy: AccuracyTarget,
+) -> Result<Rc<L4Node>, LoweringSkip> {
+    // Reuse the SAME candidate analysis `engine.rs` already runs for the
+    // legacy dispatch, rather than re-deriving rate detection via a
+    // second raw-AST walk. `OuterFn::Rate` is the one shape that binds
+    // SUCCESSFULLY today (via the Rate->Increase rewrite) but would
+    // produce a semantically wrong comparison -- see `LoweringSkip::RateShape`.
+    let analysis = control_plane::asap_tier_analysis::analyze_promql_for_asap_tier(query);
+    if analysis
+        .candidates
+        .iter()
+        .any(|c| c.outer_fn == OuterFn::Rate)
+    {
+        return Err(LoweringSkip::RateShape);
+    }
+
+    let qe = control_plane::query_parser::parse_query_expr_canonical(query)
+        .map_err(|e| LoweringSkip::ParseFailed(e.to_string()))?;
+
+    let physical = control_plane::sketch_algebra::bind_query_expr(&qe, accuracy)
+        .map_err(|e: BindingError| LoweringSkip::Implement(e.to_string()))?;
+
+    match physical {
+        PhysicalExpr::Committed(L4Plan::Summary(node)) => {
+            if matches!(node.expr, SummaryExpr::Logical(_)) {
+                Err(LoweringSkip::NotRealized)
+            } else {
+                Ok(node)
+            }
+        }
+        _ => Err(LoweringSkip::UnsupportedPhysicalShape),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn accuracy() -> AccuracyTarget {
+        AccuracyTarget::Epsilon(0.01)
+    }
+
+    #[test]
+    fn rate_query_is_skipped_before_binding() {
+        let result = lower_promql_to_l4node("rate(http_requests_total[5m])", accuracy());
+        assert!(
+            matches!(result, Err(LoweringSkip::RateShape)),
+            "expected RateShape, got {result:?}"
+        );
+    }
+
+    #[test]
+    fn irate_query_is_skipped_before_binding() {
+        let result = lower_promql_to_l4node("irate(http_requests_total[5m])", accuracy());
+        assert!(
+            matches!(result, Err(LoweringSkip::RateShape)),
+            "expected RateShape, got {result:?}"
+        );
+    }
+
+    #[test]
+    fn unparseable_query_is_skipped() {
+        let result = lower_promql_to_l4node("this is not promql (((", accuracy());
+        assert!(
+            matches!(result, Err(LoweringSkip::ParseFailed(_))),
+            "expected ParseFailed, got {result:?}"
+        );
+    }
+
+    #[test]
+    fn bare_selector_realizes_to_a_summary_agg() {
+        // Mirrors `implement_promql_for_asap_tier`'s own
+        // `bare_selector_implements_to_an_exact_sum_agg` test -- a bare
+        // selector is `Aggregate { Sum }` over the sample value.
+        let node = lower_promql_to_l4node("http_requests_total", accuracy())
+            .expect("bare selector should realize");
+        assert!(
+            matches!(node.expr, SummaryExpr::SummaryAgg { .. }),
+            "expected SummaryAgg, got {:?}",
+            node.expr
+        );
+    }
+
+    #[test]
+    fn frequency_intent_realizes_via_bind_query_expr() {
+        // The exact shape `asap_tier_implement.rs`'s own
+        // `implement_frequency_as_agg_test` pins as a KNOWN, documented gap
+        // for `implement_promql_for_asap_tier`/`DefaultCostModel` (asserts
+        // it stays `Logical` "for now"). `bind_query_expr`/
+        // `ControlPlaneCostModel` is exactly the fix -- via
+        // `realize_extension`/`readout_extension` (ASAPController#150) --
+        // so this must realize to a real binding here, confirming this
+        // module picked the seam that actually handles Frequency.
+        let node = lower_promql_to_l4node("count_over_time(http_requests_total[5m])", accuracy())
+            .expect("Frequency intent must realize via bind_query_expr/ControlPlaneCostModel");
+        assert!(
+            !matches!(node.expr, SummaryExpr::Logical(_)),
+            "expected a real SummaryAgg/SummaryEstimate binding, got Logical (the gap \
+             this module exists to avoid): {:?}",
+            node.expr
+        );
+    }
+
+    #[test]
+    fn topk_over_rate_is_not_realized() {
+        // The outer `Sort{Limit{Aggregate}}` shape: `implement_tree_in_with`
+        // only recurses through `Aggregate`, so the whole tree wraps as
+        // one opaque `Logical` blob -- self-excludes via `NotRealized`,
+        // no special-case detection needed for this shape specifically.
+        let result = lower_promql_to_l4node(
+            "topk(5, sum by (host) (rate(http_requests_total[5m])))",
+            accuracy(),
+        );
+        assert!(
+            matches!(result, Err(LoweringSkip::NotRealized) | Err(LoweringSkip::RateShape)),
+            "expected NotRealized or RateShape (both are valid skips for this shape), got {result:?}"
+        );
+    }
+}
diff --git a/data_plane/src/query_engines/asap_query_engine/mod.rs b/data_plane/src/query_engines/asap_query_engine/mod.rs
index f4b202d86..5ff8b3aea 100644
--- a/data_plane/src/query_engines/asap_query_engine/mod.rs
+++ b/data_plane/src/query_engines/asap_query_engine/mod.rs
@@ -10,6 +10,8 @@
 //! the JSONL leg has been deleted).
 
 pub mod engine;
+pub mod l4_lowering;
+pub mod shadow_compare;
 pub mod summary_executor;
 
 // Phase-5 reorg: ASAP-tier reducer moved to `sketch_db::query`. The
diff --git a/data_plane/src/query_engines/asap_query_engine/shadow_compare.rs b/data_plane/src/query_engines/asap_query_engine/shadow_compare.rs
new file mode 100644
index 000000000..30dfc9ea2
--- /dev/null
+++ b/data_plane/src/query_engines/asap_query_engine/shadow_compare.rs
@@ -0,0 +1,477 @@
+//! Shadow-mode comparison of the new `SummaryExecutor` path against the
+//! live `SketchReducer` path — see
+//! `data_plane/docs/l4node-plan-executor-design.md`'s "Rollout" section
+//! for the design. Computes the new answer alongside the old, diffs the
+//! two, logs discrepancies via `tracing`, and **always returns nothing to
+//! the caller** — this module can never change what a query serves.
+//! Mirrors `docs/design-sketch-db-roadmap.md` § 13.2 "Shadow mode".
+
+use std::collections::BTreeMap;
+
+use asap_sketch::exec::{execute, ExecOutcome};
+
+use crate::query_engines::asap_query_engine::l4_lowering::lower_promql_to_l4node;
+use crate::query_engines::asap_query_engine::summary_executor::{
+    QueryExecutionContext, SummaryValue,
+};
+use crate::storage_engines::sketch_db::index::SketchStore;
+use crate::storage_engines::sketch_db::query::ASAPTierResult;
+
+/// Fixed accuracy target for this phase — `data_plane` doesn't carry a
+/// per-workload `AccuracyTarget` today (see the design doc's "Rollout"
+/// section); threading a real one through is a possible fast-follow, not
+/// blocking. `0.01` matches this deployment's typical default accuracy
+/// bound.
+const SHADOW_ACCURACY: control_plane::types_v2::AccuracyTarget =
+    control_plane::types_v2::AccuracyTarget::Epsilon(0.01);
+
+/// Relative tolerance for comparing an approximate sketch readout against
+/// itself across two independent code paths -- both paths decode the SAME
+/// underlying sketch state, so any difference here is a REAL divergence
+/// (a bug in one path or the other), not sketch estimation error. A small
+/// tolerance absorbs floating-point summation-order differences only.
+const RELATIVE_TOLERANCE: f64 = 1e-6;
+
+/// Mirrors `ASAPTierResult.series`'s row shape -- `(label_values, samples)`
+/// where `samples` is `(window_end_unix_ms, value)`.
+type SeriesRows = Vec<(BTreeMap<String, String>, Vec<(i64, f64)>)>;
+
+/// Whether shadow-mode comparison is enabled for this process. Mirrors
+/// `ASAP_LEGACY_DUAL_WRITE`'s exact mechanics (`drivers/ingest/otel.rs`) --
+/// trimmed, case-insensitive `1`/`true`/`on`, default off.
+pub fn shadow_summary_executor_enabled() -> bool {
+    std::env::var("ASAP_SHADOW_SUMMARY_EXECUTOR")
+        .map(|v| {
+            let v = v.trim();
+            v == "1" || v.eq_ignore_ascii_case("true") || v.eq_ignore_ascii_case("on")
+        })
+        .unwrap_or(false)
+}
+
+/// Compute the new (`SummaryExecutor`) answer for `query` alongside the
+/// already-computed legacy `old` answer, diff the two, and log via
+/// `tracing`. Never returns anything, never panics, never affects what
+/// the caller serves -- every fallible step is `Result`/`Option`-handled
+/// and logged rather than `.unwrap()`ed, so a bug in this module's own
+/// conversion/diff logic degrades to "no useful log line," not a crash.
+pub fn maybe_shadow_compare(
+    index: &SketchStore,
+    query: &str,
+    t0_ms: u64,
+    t1_ms: u64,
+    is_cumulative: bool,
+    old: &ASAPTierResult,
+) {
+    if !shadow_summary_executor_enabled() {
+        return;
+    }
+
+    let node = match lower_promql_to_l4node(query, SHADOW_ACCURACY) {
+        Ok(node) => node,
+        Err(skip) => {
+            tracing::debug!(query, ?skip, "shadow: query not comparable, skipping");
+            return;
+        }
+    };
+
+    let ctx = QueryExecutionContext {
+        index,
+        t0_ms,
+        t1_ms,
+        is_cumulative,
+    };
+
+    let new_series = match execute(&node, &ctx) {
+        Ok(ExecOutcome::Value(values)) => {
+            let mut coverage: Option<(u64, u64)> = None;
+            let mut series = Vec::new();
+            for (group_key, value) in &values {
+                fold_coverage(&mut coverage, value.coverage());
+                series.extend(summary_value_to_series(group_key, value));
+            }
+            (series, coverage)
+        }
+        Ok(ExecOutcome::State(groups)) => {
+            let mut series = Vec::new();
+            for (group_key, state, _kind, _params) in &groups {
+                let Some(value) = state.exact_value(&None) else {
+                    tracing::debug!(
+                        query,
+                        ?group_key,
+                        "shadow: ExactAgg group had no comparable value, skipping group"
+                    );
+                    continue;
+                };
+                series.push((group_key.clone(), vec![(t1_ms as i64, value)]));
+            }
+            (series, None)
+        }
+        Err(e) => {
+            tracing::debug!(query, error = ?e, "shadow: execute() failed, skipping");
+            return;
+        }
+    };
+
+    diff_and_log(query, old, &new_series.0, new_series.1);
+}
+
+/// `SummaryValue::Points`/`TopK` -> `ASAPTierResult.series`'s row shape.
+/// `TopK`'s ranked-list-per-timestamp shape is pivoted into one row per
+/// item (each row = the group's label map plus an `item` label, one point
+/// per timestamp that item appeared in the ranked list) -- the SAME
+/// convention `sketch_reducer.rs`'s own topk arm already uses, not a new
+/// one invented here.
+fn summary_value_to_series(
+    group_key: &BTreeMap<String, String>,
+    value: &SummaryValue,
+) -> SeriesRows {
+    match value {
+        SummaryValue::Points(points, _coverage) => {
+            vec![(group_key.clone(), points.clone())]
+        }
+        SummaryValue::TopK(ranked_per_ts, _coverage) => {
+            let mut by_item: BTreeMap<String, Vec<(i64, f64)>> = BTreeMap::new();
+            for (ts, items) in ranked_per_ts {
+                for (item, val) in items {
+                    by_item.entry(item.clone()).or_default().push((*ts, *val));
+                }
+            }
+            by_item
+                .into_iter()
+                .map(|(item, points)| {
+                    let mut lv = group_key.clone();
+                    lv.insert("item".to_string(), item);
+                    (lv, points)
+                })
+                .collect()
+        }
+    }
+}
+
+fn fold_coverage(coverage: &mut Option<(u64, u64)>, next: Option<(u64, u64)>) {
+    let Some((lo, hi)) = next else { return };
+    *coverage = Some(match *coverage {
+        Some((clo, chi)) => (clo.min(lo), chi.max(hi)),
+        None => (lo, hi),
+    });
+}
+
+/// Diff the new path's series/coverage against the old `ASAPTierResult`
+/// and log via `tracing` -- `warn!` on a real discrepancy, `debug!` on a
+/// clean match. Never returns anything the caller could act on.
+///
+/// `single_ungrouped_series` below used to be the primary explanation for
+/// a real, confirmed gap: a bare per-series range function with no PromQL
+/// `by(...)` (e.g. `quantile_over_time(m[r])`) got an empty `{}` group key
+/// from `find_candidates`, losing (and for multiple matching series,
+/// silently MERGING) the underlying sid's own labels.
+///
+/// Both halves of that ambiguity are now fixed at the root, in
+/// `find_candidates`/`resolve_group_key` (see its doc), which reads L3/L4's
+/// own `Reduction` (ASAPController#163/#164/#165) instead of inferring
+/// intent from an empty `by`: `PerEntity` keeps each sid's own full label
+/// map (the per-series-range-function case above), while `Reduce([])`
+/// deliberately shares one group key across every candidate -- so a true
+/// global-merge aggregate like `count(hll_metric)`, previously called out
+/// here as "not modeled at all yet," is now handled correctly too.
+///
+/// This function's group-set check therefore stays only as a defensive
+/// classifier for whatever OTHER shape might still produce a genuine
+/// one-row-both-sides mismatch -- if it fires now, treat it as a real,
+/// unclassified discrepancy worth investigating, not either known gap.
+fn diff_and_log(
+    query: &str,
+    old: &ASAPTierResult,
+    new_series: &SeriesRows,
+    new_coverage: Option<(u64, u64)>,
+) {
+    let old_by_group: BTreeMap<&BTreeMap<String, String>, &Vec<(i64, f64)>> =
+        old.series.iter().map(|(k, v)| (k, v)).collect();
+    let new_by_group: BTreeMap<&BTreeMap<String, String>, &Vec<(i64, f64)>> =
+        new_series.iter().map(|(k, v)| (k, v)).collect();
+
+    if old_by_group.keys().collect::<Vec<_>>() != new_by_group.keys().collect::<Vec<_>>() {
+        if single_ungrouped_series(old, new_series) {
+            tracing::warn!(
+                query,
+                old_group = ?old.series[0].0,
+                "shadow mismatch: one row on each side but new path's group key is empty -- \
+                 both the per-series-range-function gap AND the global-merge-aggregate gap \
+                 are now fixed in find_candidates/resolve_group_key (driven by L3/L4's \
+                 Reduction), so this shape firing now means an UNCLASSIFIED gap, \
+                 not either known one"
+            );
+            return;
+        }
+        tracing::warn!(
+            query,
+            old_groups = ?old_by_group.keys().collect::<Vec<_>>(),
+            new_groups = ?new_by_group.keys().collect::<Vec<_>>(),
+            "shadow mismatch: group sets differ"
+        );
+        return;
+    }
+
+    let mut any_mismatch = false;
+    for (group, old_points) in &old_by_group {
+        // `expect`-free: the key-set equality check above guarantees this
+        // lookup succeeds; still handled defensively rather than indexed.
+        let Some(new_points) = new_by_group.get(group) else {
+            any_mismatch = true;
+            continue;
+        };
+        if !points_match(old_points, new_points) {
+            any_mismatch = true;
+            tracing::warn!(
+                query,
+                ?group,
+                old = ?old_points,
+                new = ?new_points,
+                "shadow mismatch: values differ"
+            );
+        }
+    }
+
+    if !any_mismatch && old.coverage != new_coverage {
+        tracing::debug!(
+            query,
+            old_coverage = ?old.coverage,
+            new_coverage = ?new_coverage,
+            "shadow: coverage differs (informational, not scored as a value mismatch)"
+        );
+    }
+
+    if !any_mismatch {
+        tracing::debug!(query, "shadow: match");
+    }
+}
+
+/// Detects the specific "one row on each side, new path's key is `{}`"
+/// shape -- see `diff_and_log`'s doc: this used to classify a known,
+/// now-fixed gap; it's kept as a distinct classifier for whatever else
+/// might still produce this shape, not because it's expected to fire.
+/// Deliberately does NOT compare values here: if the group keys differ,
+/// comparing the vectors would be comparing two potentially-unrelated
+/// series by coincidence of list position, not by any real correspondence.
+fn single_ungrouped_series(old: &ASAPTierResult, new_series: &SeriesRows) -> bool {
+    old.series.len() == 1 && new_series.len() == 1 && new_series[0].0.is_empty()
+}
+
+fn points_match(a: &[(i64, f64)], b: &[(i64, f64)]) -> bool {
+    if a.len() != b.len() {
+        return false;
+    }
+    let mut a_sorted = a.to_vec();
+    let mut b_sorted = b.to_vec();
+    a_sorted.sort_by_key(|(ts, _)| *ts);
+    b_sorted.sort_by_key(|(ts, _)| *ts);
+    a_sorted
+        .iter()
+        .zip(b_sorted.iter())
+        .all(|((ta, va), (tb, vb))| ta == tb && relative_eq(*va, *vb))
+}
+
+fn relative_eq(a: f64, b: f64) -> bool {
+    if a == b {
+        return true;
+    }
+    let scale = a.abs().max(b.abs()).max(1.0);
+    (a - b).abs() / scale <= RELATIVE_TOLERANCE
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::storage_engines::sketch_db::data::{AggKind, SketchConfig};
+    use crate::storage_engines::sketch_db::index::{
+        AccuracyBound, Capability, SketchInstanceMetadata, SketchKindHandle, SketchSampleState,
+    };
+
+    /// `std::env::set_var`/`remove_var` mutate process-global state, and
+    /// `cargo test` runs tests in the same process across multiple
+    /// threads by default -- every test touching `ASAP_SHADOW_SUMMARY_EXECUTOR`
+    /// must hold this for its duration to avoid racing the others (mirrors
+    /// `control_plane/src/main.rs`'s `EnvVarGuard` pattern for the same
+    /// reason). Resets the var on drop so tests don't leak global state
+    /// into whatever runs next in this process.
+    static ENV_VAR_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
+    #[allow(dead_code)] // held for its lock-lifetime/Drop side effect, never read
+    struct ShadowEnvGuard(std::sync::MutexGuard<'static, ()>);
+
+    impl Drop for ShadowEnvGuard {
+        fn drop(&mut self) {
+            std::env::remove_var("ASAP_SHADOW_SUMMARY_EXECUTOR");
+        }
+    }
+
+    fn set_shadow_env(value: &str) -> ShadowEnvGuard {
+        let guard = ENV_VAR_LOCK
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        std::env::set_var("ASAP_SHADOW_SUMMARY_EXECUTOR", value);
+        ShadowEnvGuard(guard)
+    }
+
+    fn kll_fixture() -> SketchStore {
+        let idx = SketchStore::new();
+        let cfg = SketchConfig::Kll { k: 200 };
+        idx.register(SketchInstanceMetadata {
+            sid: 1,
+            metric_name: "latency_ms".to_string(),
+            group_by_keys: std::collections::BTreeSet::new(),
+            capability: Some(Capability::QuantileApprox(SketchKindHandle::Kll)),
+            agg_kind: AggKind::Sketch {
+                kind: SketchKindHandle::Kll,
+                config: cfg.clone(),
+                spatial_filter_canonical: String::new(),
+            },
+            accuracy: Some(AccuracyBound::from_config(&cfg)),
+            first_seen_unix_ms: 0,
+            retired_at_ms: None,
+            expires_at_ms: None,
+            policy_fp: asap_types::PolicyFingerprint::UNSET,
+        });
+
+        use asap_sketchlib::proto::sketchlib::{sketch_envelope, KllState, SketchEnvelope};
+        use prost::Message;
+        let items: Vec<f64> = (1..=100).map(|i| i as f64).collect();
+        let state = KllState {
+            k: 200,
+            items,
+            levels: vec![],
+            num_levels: 0,
+            ..Default::default()
+        };
+        let env = SketchEnvelope {
+            sketch_state: Some(sketch_envelope::SketchState::Kll(state)),
+            ..Default::default()
+        };
+        idx.append_sample(
+            1,
+            BTreeMap::new(),
+            (1_000, 2_000),
+            SketchSampleState {
+                bytes: env.encode_to_vec(),
+                encoding: crate::storage_engines::sketch_db::index::SketchEncoding::ProtoFull,
+            },
+        );
+        idx
+    }
+
+    #[test]
+    fn maybe_shadow_compare_matching_fixture_does_not_panic() {
+        let _guard = set_shadow_env("1");
+        let idx = kll_fixture();
+        // Median of 1..=100 is ~50 -- matches what the new path should
+        // independently compute from the SAME underlying sketch state.
+        let old = ASAPTierResult {
+            series: vec![(BTreeMap::new(), vec![(2_000, 50.0)])],
+            coverage: Some((2_000, 2_000)),
+        };
+        maybe_shadow_compare(
+            &idx,
+            "quantile_over_time(latency_ms[1m])",
+            1_000,
+            2_000,
+            true,
+            &old,
+        );
+    }
+
+    #[test]
+    fn maybe_shadow_compare_mismatched_fixture_does_not_panic() {
+        let _guard = set_shadow_env("1");
+        let idx = kll_fixture();
+        // Deliberately wrong value -- proves the mismatch path (not just
+        // the match path) runs cleanly too.
+        let old = ASAPTierResult {
+            series: vec![(BTreeMap::new(), vec![(2_000, 999.0)])],
+            coverage: Some((2_000, 2_000)),
+        };
+        maybe_shadow_compare(
+            &idx,
+            "quantile_over_time(latency_ms[1m])",
+            1_000,
+            2_000,
+            true,
+            &old,
+        );
+    }
+
+    // One test, not three: `std::env::set_var`/`remove_var` mutate
+    // process-global state, and `cargo test` runs tests in the same
+    // process across multiple threads by default -- separate test fns
+    // touching the same env var can race. Merging into one sequential
+    // test avoids adding a synchronization primitive just for this.
+    #[test]
+    fn shadow_env_var_gate() {
+        // Acquire the same lock `set_shadow_env` uses (without its value,
+        // since this test sweeps through several values itself) so it
+        // can't race the other env-var tests in this module.
+        let _guard = ENV_VAR_LOCK
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        std::env::remove_var("ASAP_SHADOW_SUMMARY_EXECUTOR");
+        assert!(
+            !shadow_summary_executor_enabled(),
+            "expected disabled by default"
+        );
+
+        for v in ["1", "true", "TRUE", "on", " 1 "] {
+            std::env::set_var("ASAP_SHADOW_SUMMARY_EXECUTOR", v);
+            assert!(
+                shadow_summary_executor_enabled(),
+                "expected {v:?} to enable shadow mode"
+            );
+        }
+
+        for v in ["0", "false", "no", ""] {
+            std::env::set_var("ASAP_SHADOW_SUMMARY_EXECUTOR", v);
+            assert!(
+                !shadow_summary_executor_enabled(),
+                "expected {v:?} to NOT enable shadow mode"
+            );
+        }
+
+        // Same test, same env-var state (disabled): `maybe_shadow_compare`
+        // must not even attempt to lower/execute -- pass a query that
+        // would otherwise fail loudly to prove the early return is
+        // genuinely taken, not just "happened to not crash."
+        std::env::remove_var("ASAP_SHADOW_SUMMARY_EXECUTOR");
+        let idx = SketchStore::new();
+        let old = ASAPTierResult {
+            series: vec![],
+            coverage: None,
+        };
+        maybe_shadow_compare(&idx, "this is not promql (((", 0, 1000, true, &old);
+    }
+
+    #[test]
+    fn points_match_ignores_order() {
+        let a = vec![(1, 1.0), (2, 2.0)];
+        let b = vec![(2, 2.0), (1, 1.0)];
+        assert!(points_match(&a, &b));
+    }
+
+    #[test]
+    fn points_match_within_relative_tolerance() {
+        let a = vec![(1, 100.0)];
+        let b = vec![(1, 100.0000001)];
+        assert!(points_match(&a, &b));
+    }
+
+    #[test]
+    fn points_mismatch_beyond_tolerance() {
+        let a = vec![(1, 100.0)];
+        let b = vec![(1, 105.0)];
+        assert!(!points_match(&a, &b));
+    }
+
+    #[test]
+    fn points_mismatch_different_lengths() {
+        let a = vec![(1, 1.0)];
+        let b = vec![(1, 1.0), (2, 2.0)];
+        assert!(!points_match(&a, &b));
+    }
+}
diff --git a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs
index 3df7b8f9c..c5667fbde 100644
--- a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs
+++ b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs
@@ -63,7 +63,7 @@ use std::collections::{BTreeMap, BTreeSet};
 use std::rc::Rc;
 use std::sync::Arc;
 
-use asap_ir::intent_algebra::{ColumnId, ColumnRef, QueryExpr, Source};
+use asap_ir::intent_algebra::{ColumnId, ColumnRef, QueryExpr, Reduction, Source};
 use asap_sketch::exec::SummaryExecutor;
 use asap_sketch::{L4Node, SketchQuery, SummaryExpr, SummaryKind, SummaryParams};
 
@@ -303,11 +303,12 @@ impl<'a> SummaryExecutor for QueryExecutionContext<'a> {
         sketch: &SummaryKind,
         params: &SummaryParams,
         _col: &ColumnRef,
-        by: &[ColumnId],
+        reduction: &Reduction,
         child: &L4Node,
     ) -> Result<Vec<(Self::GroupKey, Self::Handle)>, Self::Error> {
         let metric = find_metric(child).ok_or(SummaryExecutorError::NoMetricFound)?;
 
+        let by: &[ColumnId] = reduction.group_keys().map(|k| k.keys()).unwrap_or(&[]);
         let mut by_names: Vec<String> = Vec::with_capacity(by.len());
         for &col_id in by {
             let name = child
@@ -363,7 +364,7 @@ impl<'a> SummaryExecutor for QueryExecutionContext<'a> {
                     let Some(series) = series.into_iter().next() else {
                         continue;
                     };
-                    let group_key = project_group_key(&by_names, &series.series_label_values);
+                    let group_key = resolve_group_key(reduction, &by_names, &series.series_label_values);
                     out.push((
                         group_key,
                         SidHandle::Sketch {
@@ -383,7 +384,7 @@ impl<'a> SummaryExecutor for QueryExecutionContext<'a> {
                     let Some((label_values, windows)) = series_list.into_iter().next() else {
                         continue;
                     };
-                    let group_key = project_group_key(&by_names, &label_values);
+                    let group_key = resolve_group_key(reduction, &by_names, &label_values);
                     out.push((
                         group_key,
                         SidHandle::ExactAgg {
@@ -761,12 +762,15 @@ fn exact_agg_kind_match(
 }
 
 /// Project a full label-values map down to the requested `by` columns --
-/// shared by the `Sketch`/`ExactAgg` branches of `find_candidates` (both
-/// build a group key the same way from whatever label map their own
-/// range-query returns). Missing keys become empty strings so a sid
-/// registered with a subset of the requested keys still groups
-/// deterministically (mirrors `sketch_reducer.rs::evaluate_exact_agg`'s
-/// same projection).
+/// used by the `ExactAgg` branch of `find_candidates`. Missing keys
+/// become empty strings so a sid registered with a subset of the
+/// requested keys still groups deterministically (mirrors
+/// `sketch_reducer.rs::evaluate_exact_agg`'s identical projection,
+/// including its `by=[]` behavior: Sum/Increase are additive PromQL
+/// aggregation operators, so an empty `by` legitimately means "reduce
+/// fully" -- every matching sid collapses to ONE group and gets summed
+/// together, which is the correct `sum(metric)`/`increase(metric[r])`
+/// answer, not a bug to route around).
 fn project_group_key(
     by_names: &[String],
     label_values: &BTreeMap<String, String>,
@@ -777,6 +781,53 @@ fn project_group_key(
         .collect()
 }
 
+/// Group-key construction for `find_candidates`, shared by both the
+/// `Sketch` and `ExactAgg` branches -- driven directly by L3/L4's own
+/// `Reduction` (ASAPController#163/#164/#165), not inferred from whether
+/// `by` happens to be empty.
+///
+/// This replaces the old family-specific split (`sketch_group_key` vs.
+/// `project_group_key` used bare): before `Reduction` existed on
+/// `SummaryAgg`, an empty `by: Vec<ColumnId>` was genuinely ambiguous --
+/// it could mean either "no explicit grouping was even resolvable" (a
+/// bare per-series range function like `quantile_over_time(0.99,
+/// http_latency_ms[10s])`, where L3/L4 planning has no reference to any
+/// label column at all) or "a real cross-series reduction with zero
+/// grouping columns" (`count(hll_metric)`, `sum(...)`-shaped). Those two
+/// cases need OPPOSITE group-key behavior and the old `by: &[ColumnId]`
+/// signature could not tell them apart -- `sketch_group_key`'s heuristic
+/// (treat empty `by` as "keep every series distinct" for the Sketch
+/// family only) fixed the first case but could not fix the second, since
+/// by the time `find_candidates` saw a bare `[]`, the distinction was
+/// already lost.
+///
+/// `Reduction` restores it directly:
+/// - `PerEntity`: no grouping concept at all -- use the sid's own FULL
+///   label map, matching the legacy `sketch_reducer.rs::evaluate_core`
+///   path's behavior exactly (it passes `series_label_values` straight
+///   through, unconditionally), so distinct series always stay distinct
+///   rows. Applies uniformly to both families now (previously
+///   `ExactAgg`'s `project_group_key` had no equivalent, since `Sum`/
+///   `Increase`-shaped exact aggregations only ever reach an unqualified
+///   PromQL aggregation operator, which is never `PerEntity`).
+/// - `Reduce(by)`: a genuine reduction. Project onto `by_names` as
+///   before -- when `by_names` is empty this naturally returns the SAME
+///   `{}` key for every matching candidate, correctly merging them into
+///   one group (the fix for the `count(hll_metric)`-style case the old
+///   `by: &[ColumnId]` signature couldn't resolve). When non-empty, an
+///   explicit grouping was resolvable from the query (e.g. `quantile by
+///   (zone) (...)`), so project onto it as requested.
+fn resolve_group_key(
+    reduction: &Reduction,
+    by_names: &[String],
+    label_values: &BTreeMap<String, String>,
+) -> BTreeMap<String, String> {
+    match reduction {
+        Reduction::PerEntity => label_values.clone(),
+        Reduction::Reduce(_) => project_group_key(by_names, label_values),
+    }
+}
+
 /// `SketchConfig` (data_plane's per-sid stored params) -> `DeltaSketchKind`
 /// (`delta_apply`'s decode/merge parameter carrier).
 fn to_delta_kind(kind: SketchKindHandle, config: &SketchConfig) -> Option<DeltaSketchKind> {
@@ -840,7 +891,7 @@ fn find_metric(node: &L4Node) -> Option<String> {
     match &node.expr {
         SummaryExpr::Logical(qe) => find_metric_in_query_expr(qe),
         SummaryExpr::SummaryAgg { child, .. } => find_metric(child),
-        SummaryExpr::SummaryEstimate { sketch_input, .. } => find_metric(sketch_input),
+        SummaryExpr::SummaryEstimate { summary_input, .. } => find_metric(summary_input),
         SummaryExpr::SummaryMerge { children } => children.first().and_then(|c| find_metric(c)),
         _ => None,
     }
@@ -926,14 +977,14 @@ mod tests {
         })
     }
 
-    fn kll_agg_node(child: Rc<L4Node>, by: Vec<ColumnId>) -> Rc<L4Node> {
+    fn kll_agg_node(child: Rc<L4Node>, reduction: Reduction) -> Rc<L4Node> {
         Rc::new(L4Node {
             expr: SummaryExpr::SummaryAgg {
                 child,
-                sketch: SummaryKind::Kll,
+                summary: SummaryKind::Kll,
                 params: SummaryParams::Kll { k: 200 },
                 col: ColumnRef::SampleValue,
-                by,
+                reduction,
             },
             schema: L4Schema {
                 fields: vec![],
@@ -943,13 +994,17 @@ mod tests {
     }
 
     fn hll_agg_node(child: Rc<L4Node>) -> Rc<L4Node> {
+        hll_agg_node_with(child, Reduction::by(vec![]))
+    }
+
+    fn hll_agg_node_with(child: Rc<L4Node>, reduction: Reduction) -> Rc<L4Node> {
         Rc::new(L4Node {
             expr: SummaryExpr::SummaryAgg {
                 child,
-                sketch: SummaryKind::Hll,
+                summary: SummaryKind::Hll,
                 params: SummaryParams::Hll { precision: 10 },
                 col: ColumnRef::SampleValue,
-                by: vec![],
+                reduction,
             },
             schema: L4Schema {
                 fields: vec![],
@@ -958,10 +1013,10 @@ mod tests {
         })
     }
 
-    fn estimate_node(sketch_input: Rc<L4Node>, query: SketchQuery) -> Rc<L4Node> {
+    fn estimate_node(summary_input: Rc<L4Node>, query: SketchQuery) -> Rc<L4Node> {
         Rc::new(L4Node {
             expr: SummaryExpr::SummaryEstimate {
-                sketch_input,
+                summary_input,
                 query,
             },
             schema: L4Schema {
@@ -1088,13 +1143,13 @@ mod tests {
         Rc::new(L4Node {
             expr: SummaryExpr::SummaryAgg {
                 child,
-                sketch: SummaryKind::Cms,
+                summary: SummaryKind::Cms,
                 params: SummaryParams::Cms {
                     width: 256,
                     depth: 4,
                 },
                 col: ColumnRef::SampleValue,
-                by: vec![],
+                reduction: Reduction::by(vec![]),
             },
             schema: L4Schema {
                 fields: vec![],
@@ -1144,14 +1199,14 @@ mod tests {
         Rc::new(L4Node {
             expr: SummaryExpr::SummaryAgg {
                 child,
-                sketch: SummaryKind::CmsWithHeap,
+                summary: SummaryKind::CmsWithHeap,
                 params: SummaryParams::CmsWithHeap {
                     width: 256,
                     depth: 4,
                     heap_size: 10,
                 },
                 col: ColumnRef::SampleValue,
-                by: vec![],
+                reduction: Reduction::by(vec![]),
             },
             schema: L4Schema {
                 fields: vec![],
@@ -1188,10 +1243,13 @@ mod tests {
         Rc::new(L4Node {
             expr: SummaryExpr::SummaryAgg {
                 child,
-                sketch: SummaryKind::Sum,
+                summary: SummaryKind::Sum,
                 params: SummaryParams::Sum,
                 col: ColumnRef::SampleValue,
-                by,
+                // Sum is a genuine PromQL aggregation operator -- an empty
+                // `by` always means "reduce fully," never `PerEntity` (see
+                // `resolve_group_key`'s doc).
+                reduction: Reduction::by(by),
             },
             schema: L4Schema {
                 fields: vec![],
@@ -1239,7 +1297,7 @@ mod tests {
 
         let child = scan_node("latency_ms", None);
         let tree = estimate_node(
-            kll_agg_node(child, vec![]),
+            kll_agg_node(child, Reduction::by(vec![])),
             SketchQuery::Quantile { q: 0.5 },
         );
 
@@ -1298,7 +1356,7 @@ mod tests {
 
         let child = scan_node("latency_ms", None);
         let tree = estimate_node(
-            kll_agg_node(child, vec![]),
+            kll_agg_node(child, Reduction::by(vec![])),
             SketchQuery::Quantile { q: 0.5 },
         );
 
@@ -1354,7 +1412,7 @@ mod tests {
         let child = scan_node("latency_ms", Some("zone"));
         // "zone" is field index 1 in `scan_node`'s schema (0 = value).
         let tree = estimate_node(
-            kll_agg_node(child, vec![1]),
+            kll_agg_node(child, Reduction::by(vec![1])),
             SketchQuery::Quantile { q: 0.5 },
         );
 
@@ -1391,6 +1449,90 @@ mod tests {
         );
     }
 
+    #[test]
+    fn bare_per_series_query_keeps_distinct_series_separate_even_with_no_by() {
+        // The `Reduction::PerEntity` half of the empty-`by` ambiguity
+        // (ASAPController#163/#164/#165), confirmed against a real e2e
+        // shadow-mode run for `quantile_over_time(m[r])` (a bare per-series
+        // range function -- no PromQL `by(...)`, no label selector, so
+        // there's no grouping concept for the query to express at all,
+        // NOT a request to merge everything). Two sids with DIFFERENT real
+        // labels ("zone") under a `PerEntity` query must still produce TWO
+        // separate output series -- naively projecting onto an empty `by`
+        // would collapse both sids' group keys to the SAME `{}` and
+        // silently merge two unrelated distributions into one wrong
+        // answer. See `genuine_full_reduction_merges_distinct_series_unlike_per_entity`
+        // for the opposite (`Reduce([])`) case, which MUST merge.
+        let idx = SketchStore::new();
+        idx.register(kll_meta(1, "latency_ms", &["zone"]));
+        idx.register(kll_meta(2, "latency_ms", &["zone"]));
+        let items1: Vec<f64> = (1..=50).map(|i| i as f64).collect();
+        let items2: Vec<f64> = (51..=100).map(|i| i as f64).collect();
+        let mut labels_east = BTreeMap::new();
+        labels_east.insert("zone".to_string(), "us-east".to_string());
+        let mut labels_west = BTreeMap::new();
+        labels_west.insert("zone".to_string(), "us-west".to_string());
+        idx.append_sample(
+            1,
+            labels_east,
+            (T0, T0 + 1000),
+            SketchSampleState {
+                bytes: encode_kll_items_proto(200, &items1),
+                encoding: crate::storage_engines::sketch_db::index::SketchEncoding::ProtoFull,
+            },
+        );
+        idx.append_sample(
+            2,
+            labels_west,
+            (T0, T0 + 1000),
+            SketchSampleState {
+                bytes: encode_kll_items_proto(200, &items2),
+                encoding: crate::storage_engines::sketch_db::index::SketchEncoding::ProtoFull,
+            },
+        );
+
+        // No `by` requested at all -- mirrors `quantile_over_time(0.99,
+        // latency_ms[r])` with no `by(...)`/label selector.
+        let child = scan_node("latency_ms", Some("zone"));
+        let tree = estimate_node(
+            kll_agg_node(child, Reduction::PerEntity),
+            SketchQuery::Quantile { q: 0.5 },
+        );
+
+        let exec = ctx(&idx);
+        let ExecOutcome::Value(mut v) = execute(&tree, &exec).expect("execute should succeed")
+        else {
+            panic!("expected a value");
+        };
+        assert_eq!(
+            v.len(),
+            2,
+            "two distinct series must stay separate even with no explicit `by` -- \
+             got {v:?}"
+        );
+        v.sort_by(|a, b| a.0.get("zone").cmp(&b.0.get("zone")));
+        let (east_group, east_value) = &v[0];
+        let (west_group, west_value) = &v[1];
+        assert_eq!(east_group.get("zone").map(String::as_str), Some("us-east"));
+        assert_eq!(west_group.get("zone").map(String::as_str), Some("us-west"));
+        let SummaryValue::Points(east_samples, _coverage) = east_value else {
+            panic!("expected Points, got {east_value:?}");
+        };
+        let SummaryValue::Points(west_samples, _coverage) = west_value else {
+            panic!("expected Points, got {west_value:?}");
+        };
+        assert!(
+            (20.0..=30.0).contains(&east_samples[0].1),
+            "us-east median {} should reflect only sid 1's data (~25), not a merge with sid 2",
+            east_samples[0].1
+        );
+        assert!(
+            (70.0..=80.0).contains(&west_samples[0].1),
+            "us-west median {} should reflect only sid 2's data (~75), not a merge with sid 1",
+            west_samples[0].1
+        );
+    }
+
     #[test]
     fn hll_cardinality_readout() {
         let idx = SketchStore::new();
@@ -1425,6 +1567,81 @@ mod tests {
         );
     }
 
+    #[test]
+    fn genuine_full_reduction_merges_distinct_series_unlike_per_entity() {
+        // The other half of the empty-`by` ambiguity `Reduction` resolves
+        // (ASAPController#163/#164/#165): a bare per-series range function
+        // like `quantile_over_time(m[r])` (Reduction::PerEntity, see
+        // `bare_per_series_query_keeps_distinct_series_separate_even_with_no_by`
+        // above) must NOT merge distinct series, but a genuine
+        // cross-series reduction with zero grouping columns --
+        // `count(hll_metric)`-shaped, Reduction::Reduce(GroupKeys::by([]))
+        // -- legitimately MUST merge them into one combined answer. Before
+        // `find_candidates` took `&Reduction` instead of `&[ColumnId]`,
+        // these two cases were indistinguishable from an empty `by` alone
+        // (the old `sketch_group_key` heuristic could only ever pick ONE
+        // of the two behaviors for every empty-`by` query).
+        //
+        // Two HLL sids with DIFFERENT real "zone" labels (same shape as
+        // the PerEntity test above) and DISJOINT item sets: under a
+        // genuine `Reduce([])`, they must merge into ONE group whose
+        // cardinality reflects BOTH sids' items combined (~10), not two
+        // separate ~5-item answers.
+        let idx = SketchStore::new();
+        idx.register(hll_meta(1, "unique_users"));
+        idx.register(hll_meta(2, "unique_users"));
+        let items1: Vec<&str> = vec!["a", "b", "c", "d", "e"];
+        let items2: Vec<&str> = vec!["f", "g", "h", "i", "j"];
+        let mut labels_east = BTreeMap::new();
+        labels_east.insert("zone".to_string(), "us-east".to_string());
+        let mut labels_west = BTreeMap::new();
+        labels_west.insert("zone".to_string(), "us-west".to_string());
+        idx.append_sample(
+            1,
+            labels_east,
+            (T0, T0 + 1000),
+            SketchSampleState {
+                bytes: encode_hll_from_items(10, &items1),
+                encoding: crate::storage_engines::sketch_db::index::SketchEncoding::MsgpackFull,
+            },
+        );
+        idx.append_sample(
+            2,
+            labels_west,
+            (T0, T0 + 1000),
+            SketchSampleState {
+                bytes: encode_hll_from_items(10, &items2),
+                encoding: crate::storage_engines::sketch_db::index::SketchEncoding::MsgpackFull,
+            },
+        );
+
+        let child = scan_node("unique_users", Some("zone"));
+        let tree = estimate_node(
+            hll_agg_node_with(child, Reduction::by(vec![])),
+            SketchQuery::Cardinality,
+        );
+
+        let exec = ctx(&idx);
+        let ExecOutcome::Value(v) = execute(&tree, &exec).expect("execute should succeed") else {
+            panic!("expected a value");
+        };
+        assert_eq!(
+            v.len(),
+            1,
+            "a genuine full reduction must merge both sids into one group, got {v:?}"
+        );
+        let (_group, value) = &v[0];
+        let SummaryValue::Points(samples, _coverage) = value else {
+            panic!("expected Points, got {value:?}");
+        };
+        let (_ts, card) = samples[0];
+        assert!(
+            (8.0..=12.0).contains(&card),
+            "merged cardinality {card} should be ~10 (both sids' disjoint item sets combined), \
+             not ~5 (one sid dropped or kept separate)"
+        );
+    }
+
     #[test]
     fn single_cms_sid_total_readout() {
         let idx = SketchStore::new();
@@ -1783,7 +2000,7 @@ mod tests {
         let idx = SketchStore::new();
         let child = scan_node("nonexistent_metric", None);
         let tree = estimate_node(
-            kll_agg_node(child, vec![]),
+            kll_agg_node(child, Reduction::by(vec![])),
             SketchQuery::Quantile { q: 0.5 },
         );
         let exec = ctx(&idx);
@@ -1812,10 +2029,10 @@ mod tests {
         let mismatched = Rc::new(L4Node {
             expr: SummaryExpr::SummaryAgg {
                 child,
-                sketch: SummaryKind::Kll,
+                summary: SummaryKind::Kll,
                 params: SummaryParams::Kll { k: 500 },
                 col: ColumnRef::SampleValue,
-                by: vec![],
+                reduction: Reduction::by(vec![]),
             },
             schema: L4Schema {
                 fields: vec![],
@@ -1863,7 +2080,7 @@ mod tests {
 
         let child = scan_node("latency_ms", None);
         let tree = estimate_node(
-            kll_agg_node(child, vec![]),
+            kll_agg_node(child, Reduction::by(vec![])),
             SketchQuery::Quantile { q: 0.5 },
         );
 
@@ -1935,7 +2152,7 @@ mod tests {
 
         let child = scan_node("latency_ms", None);
         let tree = estimate_node(
-            kll_agg_node(child, vec![]),
+            kll_agg_node(child, Reduction::by(vec![])),
             SketchQuery::Quantile { q: 0.5 },
         );
 
@@ -1991,7 +2208,7 @@ mod tests {
 
         let child = scan_node("latency_ms", None);
         let tree = estimate_node(
-            kll_agg_node(child, vec![]),
+            kll_agg_node(child, Reduction::by(vec![])),
             SketchQuery::Quantile { q: 0.5 },
         );
 
@@ -2206,10 +2423,10 @@ mod tests {
         let tree = Rc::new(L4Node {
             expr: SummaryExpr::SummaryAgg {
                 child,
-                sketch: SummaryKind::MinMax,
+                summary: SummaryKind::MinMax,
                 params: SummaryParams::MinMax,
                 col: ColumnRef::SampleValue,
-                by: vec![],
+                reduction: Reduction::by(vec![]),
             },
             schema: L4Schema {
                 fields: vec![],
diff --git a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs
index e30a1d749..b9b225696 100644
--- a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs
+++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs
@@ -2213,11 +2213,7 @@ async fn controller_plan_to_query_ddsketch_delta_subwindow_roundtrip() {
 
         // Split this window's distribution into three sub-window increments.
         let third = dist.len() / 3;
-        let sub: [&[f64]; 3] = [
-            &dist[..third],
-            &dist[third..2 * third],
-            &dist[2 * third..],
-        ];
+        let sub: [&[f64]; 3] = [&dist[..third], &dist[third..2 * third], &dist[2 * third..]];
 
         for (f_idx, frame_vals) in sub.iter().enumerate() {
             let sk = dd_over_values(alpha, frame_vals);
@@ -2283,8 +2279,8 @@ async fn controller_plan_to_query_ddsketch_delta_subwindow_roundtrip() {
     // windows (1..=30 ∪ 100..=130 ∪ 1000..=1030 = 1..=1030, 93 samples).
     // Its p99 ≈ 1020. Pull the scalar out of the (instant or range) shape
     // and assert it lands near that within a generous DDSketch-α envelope.
-    let value = extract_first_scalar(result)
-        .expect("could not extract a scalar from the query result");
+    let value =
+        extract_first_scalar(result).expect("could not extract a scalar from the query result");
     // Truth: p99 of the unioned distribution.
     let mut all: Vec<f64> = Vec::new();
     for d in &window_dists {
@@ -2308,12 +2304,140 @@ fn extract_first_scalar(result: &JsonValue) -> Option<f64> {
     let arr = result.as_array()?;
     let first = arr.first()?;
     if let Some(v) = first.get("value").and_then(|v| v.as_array()) {
-        return v.get(1).and_then(|s| s.as_str()).and_then(|s| s.parse().ok());
+        return v
+            .get(1)
+            .and_then(|s| s.as_str())
+            .and_then(|s| s.parse().ok());
     }
     if let Some(vals) = first.get("values").and_then(|v| v.as_array()) {
         let last = vals.last()?.as_array()?;
-        return last.get(1).and_then(|s| s.as_str()).and_then(|s| s.parse().ok());
+        return last
+            .get(1)
+            .and_then(|s| s.as_str())
+            .and_then(|s| s.parse().ok());
     }
     None
 }
 
+// ── Test — shadow-mode `SummaryExecutor` comparison is inert ────────────────
+//
+// `data_plane/docs/l4node-plan-executor-design.md`'s "Rollout" section:
+// enabling `ASAP_SHADOW_SUMMARY_EXECUTOR` computes the new path alongside
+// the old and logs a diff, but must NEVER change what's served. This test
+// is the regression safety net for that claim — same shape as Test 3
+// (`controller_plan_to_query_full_roundtrip_ddsketch`), but with shadow
+// mode on for the duration, asserting the served response still succeeds
+// with the SAME quantile value the flag-off test expects (~p99 of
+// `[5,10,15,20]` bucket counts).
+
+/// RAII guard for `ASAP_SHADOW_SUMMARY_EXECUTOR`. `std::env::set_var`/
+/// `remove_var` mutate process-global state and `cargo test` runs tests
+/// in the same process across threads by default, so every test touching
+/// this var must serialize against the others (mirrors
+/// `control_plane/src/main.rs`'s `EnvVarGuard` pattern for
+/// `USE_TYPED_STAGE_SPLIT`, same reason).
+#[allow(dead_code)] // held for its lock-lifetime/Drop side effect, never read
+struct ShadowEnvGuard(std::sync::MutexGuard<'static, ()>);
+
+impl ShadowEnvGuard {
+    fn enable() -> Self {
+        static LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
+        let guard = LOCK.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+        std::env::set_var("ASAP_SHADOW_SUMMARY_EXECUTOR", "1");
+        Self(guard)
+    }
+}
+
+impl Drop for ShadowEnvGuard {
+    fn drop(&mut self) {
+        std::env::remove_var("ASAP_SHADOW_SUMMARY_EXECUTOR");
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn shadow_mode_does_not_change_served_ddsketch_quantile() {
+    let _shadow = ShadowEnvGuard::enable();
+
+    let stack = start_full_stack(19_591, 19_592).await;
+    let client = reqwest::Client::new();
+
+    let workload = build_workload(
+        "http_latency_ms",
+        vec![AggType::Quantile],
+        0.01,
+        Duration::from_secs(1),
+        vec!["service".to_string()],
+        vec![0.99],
+    );
+    let streaming_config_json = plan_streaming_config_json(&workload);
+    post_streaming_config(&client, stack.backend_port, &streaming_config_json).await;
+
+    // Same fixture data as `controller_plan_to_query_full_roundtrip_ddsketch`
+    // — this test isn't checking quantile accuracy (that's Test 3's job),
+    // it's checking that turning shadow mode on doesn't change whether/what
+    // this query serves.
+    let alpha = 0.01;
+    let store_counts = vec![5u64, 10, 15, 20];
+    let dd_state = build_dd_sketch_state(alpha, store_counts, -1);
+    let sketch_bytes = dd_state.encode_to_vec();
+
+    let now_ns = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .expect("system time before UNIX epoch")
+        .as_nanos() as u64;
+    let sketch_t_ns = now_ns.saturating_sub(3_000_000_000);
+    let watermark_t_ns = now_ns.saturating_sub(1_000_000_000);
+
+    let req = build_dd_sketch_export(
+        "http_latency_ms",
+        &[("service", "e2e-test")],
+        sketch_t_ns,
+        sketch_bytes,
+        alpha,
+    );
+    post_otlp_http(&client, stack.otlp_http_port, req).await;
+
+    let watermark_state = build_dd_sketch_state(alpha, Vec::new(), 0);
+    let watermark_req = build_dd_sketch_export(
+        "http_latency_ms",
+        &[("service", "e2e-test")],
+        watermark_t_ns,
+        watermark_state.encode_to_vec(),
+        alpha,
+    );
+    post_otlp_http(&client, stack.otlp_http_port, watermark_req).await;
+
+    tokio::time::sleep(Duration::from_millis(800)).await;
+
+    let query_url = format!("http://127.0.0.1:{}/api/v1/query", stack.backend_port);
+    let response: JsonValue = client
+        .get(&query_url)
+        .query(&[("query", "quantile_over_time(0.99, http_latency_ms[10s])")])
+        .send()
+        .await
+        .expect("PromQL query failed to send")
+        .json()
+        .await
+        .expect("PromQL response was not JSON");
+
+    assert_eq!(
+        response["status"].as_str().unwrap_or("(missing)"),
+        "success",
+        "shadow mode must not change whether this query succeeds. Response:\n{}",
+        serde_json::to_string_pretty(&response).unwrap_or_default()
+    );
+
+    // Same accuracy contract as the flag-off test: p99 of [5,10,15,20]
+    // DDSketch bucket counts should land in a plausible range (not
+    // asserting exact equality with Test 3's own run -- different process,
+    // different wall-clock timestamps -- but the same fixture must
+    // produce the same class of answer regardless of the shadow flag).
+    let value = response["data"]["result"]
+        .as_array()
+        .and_then(|r| extract_first_scalar(&JsonValue::Array(r.clone())))
+        .expect("expected a scalar quantile result");
+    assert!(
+        value.is_finite() && value > 0.0,
+        "shadow mode must not corrupt the served quantile value, got {value}"
+    );
+}