diff --git a/crates/e2e/tests/aggregate.rs b/crates/e2e/tests/aggregate.rs
index e263b20c..58bf404b 100644
--- a/crates/e2e/tests/aggregate.rs
+++ b/crates/e2e/tests/aggregate.rs
@@ -10,7 +10,7 @@
 
 use asap_e2e::fixtures::metric_schema;
 use asap_frontend_promql::lower_promql;
-use asap_ir::intent_algebra::{AggIntent, QueryExpr, Source};
+use asap_ir::intent_algebra::{AggIntent, QueryExpr, Reduction, Source};
 use asap_ir::types::AccuracyTarget;
 
 fn lower(q: &str) -> QueryExpr {
@@ -29,7 +29,7 @@ fn scan(metric: &str, labels: &[&str]) -> QueryExpr {
 
 fn agg(by: Vec<usize>, intent: AggIntent, child: QueryExpr) -> QueryExpr {
     QueryExpr::Aggregate {
-        by: by.into(),
+        reduction: Reduction::by(by),
         aggs: vec![intent],
         output_names: vec!["".into()],
         having: None,
diff --git a/crates/e2e/tests/binary_op.rs b/crates/e2e/tests/binary_op.rs
index 2771bf30..1a2ba88f 100644
--- a/crates/e2e/tests/binary_op.rs
+++ b/crates/e2e/tests/binary_op.rs
@@ -10,8 +10,8 @@ use std::time::Duration;
 use asap_e2e::fixtures::metric_schema;
 use asap_frontend_promql::lower_promql;
 use asap_ir::intent_algebra::{
-    AggIntent, ArithOp, BinaryOpKind, CompareOp, GroupSide, QueryExpr, Source, VectorGrouping,
-    VectorMatch, VectorMatchKind,
+    AggIntent, ArithOp, BinaryOpKind, CompareOp, GroupSide, QueryExpr, Reduction, Source,
+    VectorGrouping, VectorMatch, VectorMatchKind,
 };
 use asap_ir::types::AccuracyTarget;
 
@@ -31,7 +31,7 @@ fn scan(metric: &str, labels: &[&str]) -> QueryExpr {
 
 fn rate_agg(metric: &str) -> QueryExpr {
     QueryExpr::Aggregate {
-        by: vec![].into(),
+        reduction: Reduction::PerEntity,
         aggs: vec![AggIntent::Rate],
         output_names: vec!["".into()],
         having: None,
@@ -44,7 +44,7 @@ fn rate_agg(metric: &str) -> QueryExpr {
 
 fn sum_by_job(metric: &str) -> QueryExpr {
     QueryExpr::Aggregate {
-        by: vec![2].into(),
+        reduction: Reduction::by(vec![2]),
         aggs: vec![AggIntent::Sum { col: None }],
         output_names: vec!["".into()],
         having: None,
@@ -245,7 +245,7 @@ fn q36_unary_negation_is_multiply_by_minus_one() {
 #[test]
 fn q36_sum_of_negation_nests() {
     let expected = QueryExpr::Aggregate {
-        by: vec![].into(),
+        reduction: Reduction::by(vec![]),
         aggs: vec![AggIntent::Sum { col: None }],
         output_names: vec!["".into()],
         having: None,
diff --git a/crates/e2e/tests/nested.rs b/crates/e2e/tests/nested.rs
index 287c6e71..54be79c8 100644
--- a/crates/e2e/tests/nested.rs
+++ b/crates/e2e/tests/nested.rs
@@ -14,7 +14,7 @@ use asap_e2e::fixtures::metric_schema;
 use asap_frontend_promql::lower_promql;
 use asap_ir::intent_algebra::{
     AggIntent, ArithOp, AtModifier, BinaryOpKind, CompareOp, GroupKeys, L3Expr, L3Scalar,
-    Predicate, QueryExpr, Source, TimeShift, VectorMatch, VectorMatchKind,
+    Predicate, QueryExpr, Reduction, Source, TimeShift, VectorMatch, VectorMatchKind,
 };
 use asap_ir::types::AccuracyTarget;
 
@@ -24,7 +24,17 @@ fn lower(q: &str) -> QueryExpr {
 
 fn agg(by: Vec<usize>, intent: AggIntent, child: QueryExpr) -> QueryExpr {
     QueryExpr::Aggregate {
-        by: by.into(),
+        reduction: Reduction::by(by),
+        aggs: vec![intent],
+        output_names: vec!["".into()],
+        having: None,
+        child: Box::new(child),
+    }
+}
+
+fn agg_per_entity(intent: AggIntent, child: QueryExpr) -> QueryExpr {
+    QueryExpr::Aggregate {
+        reduction: Reduction::PerEntity,
         aggs: vec![intent],
         output_names: vec!["".into()],
         having: None,
@@ -43,8 +53,7 @@ fn q22_sum_by_job_over_rate() {
         predicates: vec![],
         schema: metric_schema(&["job"]),
     };
-    let inner_rate = agg(
-        vec![],
+    let inner_rate = agg_per_entity(
         AggIntent::Rate,
         QueryExpr::TimeRange {
             range: Duration::from_secs(300),
@@ -102,8 +111,7 @@ fn q25_div_over_complex_subtrees() {
     let lhs = agg(
         vec![2],
         AggIntent::Sum { col: None },
-        agg(
-            vec![],
+        agg_per_entity(
             AggIntent::Rate,
             QueryExpr::TimeRange {
                 range: Duration::from_secs(300),
@@ -122,8 +130,7 @@ fn q25_div_over_complex_subtrees() {
     let rhs = agg(
         vec![2],
         AggIntent::Sum { col: None },
-        agg(
-            vec![],
+        agg_per_entity(
             AggIntent::Rate,
             QueryExpr::TimeRange {
                 range: Duration::from_secs(300),
@@ -160,8 +167,7 @@ fn q27_max_over_sum_by_job_over_rate() {
         predicates: vec![],
         schema: metric_schema(&["job"]),
     };
-    let inner_rate = agg(
-        vec![],
+    let inner_rate = agg_per_entity(
         AggIntent::Rate,
         QueryExpr::TimeRange {
             range: Duration::from_secs(300),
@@ -256,8 +262,7 @@ fn q39_sum_without_instance_over_rate() {
         predicates: vec![],
         schema: metric_schema(&["instance"]),
     };
-    let inner_rate = agg(
-        vec![],
+    let inner_rate = agg_per_entity(
         AggIntent::Rate,
         QueryExpr::TimeRange {
             range: Duration::from_secs(300),
@@ -265,7 +270,7 @@ fn q39_sum_without_instance_over_rate() {
         },
     );
     let expected = QueryExpr::Aggregate {
-        by: GroupKeys::without(vec![2]), // exclude `instance`
+        reduction: Reduction::Reduce(GroupKeys::without(vec![2])), // exclude `instance`
         aggs: vec![AggIntent::Sum { col: None }],
         output_names: vec!["".into()],
         having: None,
@@ -298,8 +303,7 @@ fn q40_week_over_week_offset() {
             },
             None => scan,
         };
-        agg(
-            vec![],
+        agg_per_entity(
             AggIntent::Rate,
             QueryExpr::TimeRange {
                 range: Duration::from_secs(300),
@@ -352,8 +356,7 @@ fn q24_sum_by_job_over_rate_over_filtered_scan() {
         })],
         schema: metric_schema(&["job", "status"]),
     };
-    let inner_rate = agg(
-        vec![],
+    let inner_rate = agg_per_entity(
         AggIntent::Rate,
         QueryExpr::TimeRange {
             range: Duration::from_secs(300),
@@ -382,16 +385,14 @@ fn q27_nested_subquery_prometheus_docs_example() {
         predicates: vec![],
         schema: metric_schema(&[]),
     };
-    let rate = agg(
-        vec![],
+    let rate = agg_per_entity(
         AggIntent::Rate,
         QueryExpr::TimeRange {
             range: Duration::from_secs(5),
             child: Box::new(scan),
         },
     );
-    let deriv = agg(
-        vec![],
+    let deriv = agg_per_entity(
         AggIntent::Deriv,
         QueryExpr::Subquery {
             range: Duration::from_secs(30),
@@ -399,8 +400,7 @@ fn q27_nested_subquery_prometheus_docs_example() {
             child: Box::new(rate),
         },
     );
-    let expected = agg(
-        vec![],
+    let expected = agg_per_entity(
         AggIntent::Max { col: None },
         QueryExpr::Subquery {
             range: Duration::from_secs(600),
diff --git a/crates/e2e/tests/time_range.rs b/crates/e2e/tests/time_range.rs
index 22828f74..b4a64395 100644
--- a/crates/e2e/tests/time_range.rs
+++ b/crates/e2e/tests/time_range.rs
@@ -11,7 +11,7 @@ use std::time::Duration;
 
 use asap_e2e::fixtures::metric_schema;
 use asap_frontend_promql::lower_promql;
-use asap_ir::intent_algebra::{AggIntent, QueryExpr, Source};
+use asap_ir::intent_algebra::{AggIntent, QueryExpr, Reduction, Source};
 use asap_ir::types::AccuracyTarget;
 
 fn lower(q: &str) -> QueryExpr {
@@ -30,7 +30,7 @@ fn scan(metric: &str) -> QueryExpr {
 
 fn range_agg(range_secs: u64, intent: AggIntent, metric: &str) -> QueryExpr {
     QueryExpr::Aggregate {
-        by: vec![].into(),
+        reduction: Reduction::PerEntity,
         aggs: vec![intent],
         output_names: vec!["".into()],
         having: None,
diff --git a/crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs b/crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs
index 16069746..649263ca 100644
--- a/crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs
+++ b/crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs
@@ -27,7 +27,7 @@
 #![allow(non_snake_case)]
 
 use asap_frontend_promql::{lower_promql, PromqlError as LoweringError};
-use asap_ir::intent_algebra::{AggIntent, BinaryOpKind, CompareOp, QueryExpr};
+use asap_ir::intent_algebra::{AggIntent, BinaryOpKind, CompareOp, QueryExpr, Reduction};
 use asap_ir::types::AccuracyTarget;
 
 const CORPUS: &str = include_str!("data/awesome_prometheus_alerts.txt");
@@ -230,10 +230,17 @@ fn all_targets_missing_core_lowers() {
     // Prometheus self-monitoring `sum by (job) (up)` (the corpus query is
     // `… == 0`). Cross-series sum grouped positionally on `job`.
     let qe = ok("sum by (job) (up)");
-    let QueryExpr::Aggregate { by, aggs, .. } = &qe else {
+    let QueryExpr::Aggregate {
+        reduction, aggs, ..
+    } = &qe
+    else {
         panic!("expected Aggregate, got {qe:?}");
     };
-    assert_eq!(by, &vec![2], "job grouping → col 2 in [ts, value, job]");
+    assert_eq!(
+        reduction,
+        &Reduction::by(vec![2]),
+        "job grouping → col 2 in [ts, value, job]"
+    );
     assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }]));
 }
 
@@ -325,9 +332,15 @@ fn without_grouping_lowers_to_the_exclusion_form() {
     let QueryExpr::BinaryOp { lhs, .. } = &qe else {
         panic!("expected a comparison BinaryOp, got {qe:?}");
     };
-    let QueryExpr::Aggregate { by, aggs, .. } = lhs.as_ref() else {
+    let QueryExpr::Aggregate {
+        reduction, aggs, ..
+    } = lhs.as_ref()
+    else {
         panic!("expected a `min without` Aggregate on the LHS, got {lhs:?}");
     };
-    assert!(by.is_without(), "grouping is the exclusion form");
+    assert!(
+        reduction.expect_reduce().is_without(),
+        "grouping is the exclusion form"
+    );
     assert!(matches!(aggs.as_slice(), [AggIntent::Min { .. }]));
 }
diff --git a/crates/frontend-promql/tests/promql_conformance.rs b/crates/frontend-promql/tests/promql_conformance.rs
index bd05c0da..664a68d8 100644
--- a/crates/frontend-promql/tests/promql_conformance.rs
+++ b/crates/frontend-promql/tests/promql_conformance.rs
@@ -37,7 +37,7 @@ use asap_frontend_promql::{lower_promql, PromqlError as LoweringError};
 use asap_ir::intent_algebra::schema::DataType;
 use asap_ir::intent_algebra::{
     AggIntent, ArithOp, AtModifier, BinaryOpKind, CompareOp, L3Expr, MathFunc, QueryExpr,
-    SampleKind, Source, TimeFunc,
+    Reduction, SampleKind, Source, TimeFunc,
 };
 use asap_ir::types::AccuracyTarget;
 
@@ -290,14 +290,17 @@ fn sum_by_groups_via_positional_aggregate() {
     // keys appended sorted), so the keys resolve to columns [2, 3].
     let qe = ok("sum by(job, instance) (node_filesystem_size_bytes)");
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = &qe
     else {
         panic!("expected positional Aggregate for `by(...)`, got {qe:?}");
     };
     assert_eq!(
-        by,
-        &vec![2, 3],
+        reduction,
+        &Reduction::by(vec![2, 3]),
         "group keys resolve to positional ColumnIds"
     );
     assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }]));
@@ -339,9 +342,13 @@ fn sum_without_groups_by_the_complement() {
     // the runtime: the grouping is the exclusion form and the output schema
     // stays OPEN (unlike `by`, which freezes to closed).
     let qe = ok("sum without(instance) (node_filesystem_size_bytes)");
-    let QueryExpr::Aggregate { by, aggs, .. } = &qe else {
+    let QueryExpr::Aggregate {
+        reduction, aggs, ..
+    } = &qe
+    else {
         panic!("expected an Aggregate, got {qe:?}");
     };
+    let by = reduction.expect_reduce();
     assert!(
         by.is_without(),
         "the grouping is the `without` exclusion form"
@@ -398,12 +405,15 @@ fn sum_by_of_rate_groups_outer_level() {
     // label-preserving inner Rate. Leaf = [ts, value, instance] → by = [2].
     let qe = ok("sum by(instance) (rate(node_network_receive_bytes_total[5m]))");
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = &qe
     else {
         panic!("expected outer Aggregate grouped by instance, got {qe:?}");
     };
-    assert_eq!(by, &vec![2]);
+    assert_eq!(reduction, &Reduction::by(vec![2]));
     assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }]));
     // child is the inner per-series Rate aggregate.
     assert!(matches!(
@@ -420,12 +430,15 @@ fn sum_by_of_over_time_groups_outer_level() {
     // name-based Partition). Leaf = [ts, value, instance] → by = [2].
     let qe = ok("sum by(instance) (avg_over_time(node_cpu_seconds_total[5m]))");
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = &qe
     else {
         panic!("expected outer Aggregate grouped by instance, got {qe:?}");
     };
-    assert_eq!(by, &vec![2]);
+    assert_eq!(reduction, &Reduction::by(vec![2]));
     assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }]));
     // child is the inner per-series reduction: Aggregate{Avg} over TimeRange.
     let QueryExpr::Aggregate { aggs, child, .. } = child.as_ref() else {
@@ -514,10 +527,13 @@ fn histogram_quantile_over_sum_by_le_preserves_le_grouping() {
     ));
     // `sum by(le)` now survives as a positional Aggregate (by = [2], `le`), over
     // the inner Rate — no name-based Partition.
-    let QueryExpr::Aggregate { by, aggs, .. } = child.as_ref() else {
+    let QueryExpr::Aggregate {
+        reduction, aggs, ..
+    } = child.as_ref()
+    else {
         panic!("expected `sum by(le)` as a positional Aggregate, got {child:?}");
     };
-    assert_eq!(by, &vec![2]);
+    assert_eq!(reduction, &Reduction::by(vec![2]));
     assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }]));
 }
 
@@ -805,10 +821,17 @@ fn outer_aggregate_over_nested_aggregate_nests() {
         panic!("expected outer Aggregate, got {qe:?}");
     };
     assert!(matches!(aggs.as_slice(), [AggIntent::Max { .. }]));
-    let QueryExpr::Aggregate { by, aggs, .. } = child.as_ref() else {
+    let QueryExpr::Aggregate {
+        reduction, aggs, ..
+    } = child.as_ref()
+    else {
         panic!("expected inner `sum by (job)` Aggregate, got {child:?}");
     };
-    assert_eq!(by, &vec![2], "job grouping survives on the inner aggregate");
+    assert_eq!(
+        reduction,
+        &Reduction::by(vec![2]),
+        "job grouping survives on the inner aggregate"
+    );
     assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }]));
     assert!(has(&qe, |i| matches!(i, AggIntent::Rate)), "rate preserved");
 }
@@ -824,17 +847,28 @@ fn outer_group_key_absent_from_nested_aggregate_is_dropped() {
     // `sum(sum by (group)(…))`.
     let qe = ok(r#"sum(sum by (group)(http_requests{job="api-server"})) by (job)"#);
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = &qe
     else {
         panic!("expected outer Aggregate, got {qe:?}");
     };
-    assert!(by.is_empty(), "absent `job` key dropped → global aggregate");
+    assert_eq!(
+        reduction,
+        &Reduction::by(vec![]),
+        "absent `job` key dropped → global aggregate"
+    );
     assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }]));
-    let QueryExpr::Aggregate { by, .. } = child.as_ref() else {
+    let QueryExpr::Aggregate { reduction, .. } = child.as_ref() else {
         panic!("expected inner `sum by (group)` Aggregate, got {child:?}");
     };
-    assert_eq!(by, &vec![2], "inner grouping on `group` survives");
+    assert_eq!(
+        reduction,
+        &Reduction::by(vec![2]),
+        "inner grouping on `group` survives"
+    );
 }
 
 #[test]
@@ -844,20 +878,27 @@ fn outer_group_key_present_after_inner_aggregate_still_resolves() {
     // resolving positionally — the absent-key drop only fires on provable
     // absence, never on a resolvable key.
     let qe = ok("sum(sum by (job, group)(http_requests)) by (job)");
-    let QueryExpr::Aggregate { by, child, .. } = &qe else {
+    let QueryExpr::Aggregate {
+        reduction, child, ..
+    } = &qe
+    else {
         panic!("expected outer Aggregate, got {qe:?}");
     };
-    let QueryExpr::Aggregate { by: inner_by, .. } = child.as_ref() else {
+    let QueryExpr::Aggregate {
+        reduction: inner_reduction,
+        ..
+    } = child.as_ref()
+    else {
         panic!("expected inner Aggregate, got {child:?}");
     };
     // Inner output schema is [group, job, sum] (keys in label-column order,
     // labels alphabetical on the scan) → job = col 1.
     assert_eq!(
-        by,
-        &vec![1],
+        reduction,
+        &Reduction::by(vec![1]),
         "outer `job` resolves against the inner output"
     );
-    assert_eq!(inner_by.len(), 2);
+    assert_eq!(inner_reduction.expect_reduce().len(), 2);
 }
 
 #[test]
@@ -867,11 +908,18 @@ fn outer_group_key_over_binary_op_resolves_on_both_sides() {
     // still resolve. Each `or` side is bound independently against its own
     // sub-tree, so the key is seeded as an inherited column on both sides.
     let qe = ok(r#"sum by (__name__)(metric_a{env="1"} or metric_b{env="2"})"#);
-    let QueryExpr::Aggregate { by, child, .. } = &qe else {
+    let QueryExpr::Aggregate {
+        reduction, child, ..
+    } = &qe
+    else {
         panic!("expected outer Aggregate, got {qe:?}");
     };
     // `__name__` resolves to a single positional id against the binary op output.
-    assert_eq!(by.len(), 1, "grouped by the one `__name__` key");
+    assert_eq!(
+        reduction.expect_reduce().len(),
+        1,
+        "grouped by the one `__name__` key"
+    );
     let QueryExpr::BinaryOp { lhs, rhs, .. } = child.as_ref() else {
         panic!("expected a BinaryOp child, got {child:?}");
     };
@@ -879,7 +927,10 @@ fn outer_group_key_over_binary_op_resolves_on_both_sides() {
     // the outer group key is consistent across the union.
     let (ls, rs) = (lhs.output_schema().unwrap(), rhs.output_schema().unwrap());
     assert_eq!(ls.column_id("__name__"), rs.column_id("__name__"));
-    assert_eq!(ls.column_id("__name__"), Some(by[0]));
+    assert_eq!(
+        ls.column_id("__name__"),
+        Some(reduction.expect_reduce().keys()[0])
+    );
 
     // The general case (a plain label, not just `__name__`) also lowers.
     assert!(matches!(
@@ -924,13 +975,17 @@ fn over_time_of_subquery_reduces_per_series() {
     // to a per-series `Max` reduction over a `Subquery` (issue #27).
     let qe = ok("max_over_time(rate(demo_api_request_duration_seconds_count[5m])[1h:])");
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = &qe
     else {
         panic!("expected an Aggregate at the root, got {qe:?}");
     };
-    assert!(
-        by.is_empty(),
+    assert_eq!(
+        reduction,
+        &Reduction::PerEntity,
         "`*_over_time` has no grouping — reduces per series"
     );
     assert!(matches!(aggs.as_slice(), [AggIntent::Max { .. }]));
@@ -963,16 +1018,22 @@ fn aggregation_over_over_time_of_subquery_keeps_labels() {
     // inner `Max` collapsed labels, `job` would not resolve here.
     let qe = ok("sum by (job) (max_over_time(rate(demo{job=\"api\"}[5m])[1h:]))");
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = &qe
     else {
         panic!("expected outer Aggregate, got {qe:?}");
     };
-    assert!(!by.is_empty(), "outer `sum by (job)` groups on a label");
+    assert!(
+        matches!(reduction, Reduction::Reduce(by) if !by.is_empty()),
+        "outer `sum by (job)` groups on a label"
+    );
     assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }]));
     // Inner node is the per-series `max_over_time` reduction over the subquery.
     let QueryExpr::Aggregate {
-        by: inner_by,
+        reduction: inner_reduction,
         aggs: inner_aggs,
         child: inner_child,
         ..
@@ -980,7 +1041,7 @@ fn aggregation_over_over_time_of_subquery_keeps_labels() {
     else {
         panic!("expected inner Aggregate, got {child:?}");
     };
-    assert!(inner_by.is_empty());
+    assert_eq!(inner_reduction, &Reduction::PerEntity);
     assert!(matches!(inner_aggs.as_slice(), [AggIntent::Max { .. }]));
     assert!(matches!(inner_child.as_ref(), QueryExpr::Subquery { .. }));
 }
@@ -1004,12 +1065,15 @@ fn nested_subquery_from_prometheus_docs() {
     let qe = ok("max_over_time(deriv(rate(distance_covered_total[5s])[30s:5s])[10m:])");
 
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = &qe
     else {
         panic!("expected `max_over_time` Aggregate at the root, got {qe:?}");
     };
-    assert!(by.is_empty());
+    assert_eq!(reduction, &Reduction::PerEntity);
     assert!(matches!(aggs.as_slice(), [AggIntent::Max { .. }]));
 
     let QueryExpr::Subquery {
@@ -1024,12 +1088,15 @@ fn nested_subquery_from_prometheus_docs() {
     assert_eq!(*resolution, None, "`[10m:]` keeps the default resolution");
 
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = child.as_ref()
     else {
         panic!("expected the `deriv` Aggregate, got {child:?}");
     };
-    assert!(by.is_empty());
+    assert_eq!(reduction, &Reduction::PerEntity);
     assert!(matches!(aggs.as_slice(), [AggIntent::Deriv]));
 
     let QueryExpr::Subquery {
@@ -1188,12 +1255,19 @@ fn counter_derivative_functions_lower_to_distinct_intents() {
     ] {
         let qe = ok(q);
         let QueryExpr::Aggregate {
-            by, aggs, child, ..
+            reduction,
+            aggs,
+            child,
+            ..
         } = &qe
         else {
             panic!("expected an Aggregate for {q:?}, got {qe:?}");
         };
-        assert!(by.is_empty(), "{q}: per-series, no grouping");
+        assert_eq!(
+            reduction,
+            &Reduction::PerEntity,
+            "{q}: per-series, no grouping"
+        );
         assert_eq!(
             aggs.as_slice(),
             std::slice::from_ref(&want),
@@ -1243,12 +1317,18 @@ fn aggregation_over_counter_derivative_keeps_labels() {
     // `sum by (job)` can group on a label the inner `changes` preserved.
     let qe = ok(r#"sum by (job) (changes(m{job="api"}[15m]))"#);
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = &qe
     else {
         panic!("expected outer Aggregate, got {qe:?}");
     };
-    assert!(!by.is_empty(), "outer sum groups on job");
+    assert!(
+        matches!(reduction, Reduction::Reduce(by) if !by.is_empty()),
+        "outer sum groups on job"
+    );
     assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }]));
     assert!(intents(&qe).iter().any(|i| matches!(i, AggIntent::Changes)));
     let _ = child;
@@ -1263,22 +1343,32 @@ fn outer_stat_over_counter_derivative_nests_two_levels() {
     // inner reduction preserved, threading any scalar param (predict horizon).
     let qe = ok("avg by (dc) (predict_linear(m[3h], 3600))");
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = &qe
     else {
         panic!("expected outer Aggregate, got {qe:?}");
     };
-    assert!(!by.is_empty(), "outer `avg by (dc)` groups on a label");
+    assert!(
+        matches!(reduction, Reduction::Reduce(by) if !by.is_empty()),
+        "outer `avg by (dc)` groups on a label"
+    );
     assert!(matches!(aggs.as_slice(), [AggIntent::Avg { .. }]));
     let QueryExpr::Aggregate {
-        by: inner_by,
+        reduction: inner_reduction,
         aggs: inner_aggs,
         ..
     } = child.as_ref()
     else {
         panic!("expected inner per-series Aggregate, got {child:?}");
     };
-    assert!(inner_by.is_empty(), "inner derivative stays per-series");
+    assert_eq!(
+        inner_reduction,
+        &Reduction::PerEntity,
+        "inner derivative stays per-series"
+    );
     assert_eq!(
         inner_aggs.as_slice(),
         std::slice::from_ref(&AggIntent::PredictLinear { seconds: 3600.0 })
@@ -1352,12 +1442,19 @@ fn range_functions_over_a_subquery_reduce_per_series() {
     ] {
         let qe = ok(q);
         let QueryExpr::Aggregate {
-            by, aggs, child, ..
+            reduction,
+            aggs,
+            child,
+            ..
         } = &qe
         else {
             panic!("{q}: expected an Aggregate, got {qe:?}");
         };
-        assert!(by.is_empty(), "{q}: per-series, no grouping");
+        assert_eq!(
+            reduction,
+            &Reduction::PerEntity,
+            "{q}: per-series, no grouping"
+        );
         assert_eq!(
             aggs.as_slice(),
             std::slice::from_ref(&want),
@@ -1446,10 +1543,17 @@ fn histogram_accessors_lower_to_per_series_intents() {
         ("histogram_stdvar(v)", AggIntent::HistogramStdVar),
     ] {
         let qe = ok(q);
-        let QueryExpr::Aggregate { by, aggs, .. } = &qe else {
+        let QueryExpr::Aggregate {
+            reduction, aggs, ..
+        } = &qe
+        else {
             panic!("{q}: expected an Aggregate, got {qe:?}");
         };
-        assert!(by.is_empty(), "{q}: per-series, no grouping");
+        assert_eq!(
+            reduction,
+            &Reduction::PerEntity,
+            "{q}: per-series, no grouping"
+        );
         assert_eq!(
             aggs.as_slice(),
             std::slice::from_ref(&want),
@@ -1491,10 +1595,17 @@ fn math_functions_lower_to_per_series_math_intents() {
         ("rad(v)", MathFunc::Rad),
     ] {
         let qe = ok(q);
-        let QueryExpr::Aggregate { by, aggs, .. } = &qe else {
+        let QueryExpr::Aggregate {
+            reduction, aggs, ..
+        } = &qe
+        else {
             panic!("{q}: expected an Aggregate, got {qe:?}");
         };
-        assert!(by.is_empty(), "{q}: per-series, no grouping");
+        assert_eq!(
+            reduction,
+            &Reduction::PerEntity,
+            "{q}: per-series, no grouping"
+        );
         assert!(
             matches!(aggs.as_slice(), [AggIntent::Math(m)] if *m == want),
             "{q}: wrong intent, got {aggs:?}"
diff --git a/crates/frontend-promql/tests/promql_lowering.rs b/crates/frontend-promql/tests/promql_lowering.rs
index 85d9abad..51df670a 100644
--- a/crates/frontend-promql/tests/promql_lowering.rs
+++ b/crates/frontend-promql/tests/promql_lowering.rs
@@ -3,7 +3,7 @@
 use std::time::Duration;
 
 use asap_ir::intent_algebra::{
-    AggIntent, ArithOp, BinaryOpKind, CompareOp, L3Expr, L3Scalar, QueryExpr, Source,
+    AggIntent, ArithOp, BinaryOpKind, CompareOp, L3Expr, L3Scalar, QueryExpr, Reduction, Source,
 };
 use asap_ir::types::AccuracyTarget;
 use asap_ir::workload::{BatchEntry, Query, QueryLanguage, QueryRequirements, QueryWorkload};
@@ -59,12 +59,15 @@ fn regex_matcher_lowers_to_regex_compareop() {
 fn quantile_over_time_is_time_range_aggregate() {
     let qe = lower(r#"quantile_over_time(0.99, http_request_duration{env="prod"}[5m])"#);
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = &qe
     else {
         panic!("expected Aggregate, got {qe:?}");
     };
-    assert!(by.is_empty());
+    assert_eq!(reduction, &Reduction::PerEntity);
     assert!(matches!(aggs.as_slice(), [AggIntent::Quantile { q, .. }] if (*q - 0.99).abs() < 1e-9));
     let QueryExpr::TimeRange { range, child } = child.as_ref() else {
         panic!("expected TimeRange child, got {child:?}");
@@ -83,12 +86,15 @@ fn outer_sum_by_over_quantile_over_time_groups_positionally() {
     // names appended sorted) → host = col 2.
     let qe = lower(r#"sum by (host) (quantile_over_time(0.99, latency{service="web"}[5m]))"#);
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = &qe
     else {
         panic!("expected outer Aggregate grouped by host, got {qe:?}");
     };
-    assert_eq!(by, &vec![2]);
+    assert_eq!(reduction, &Reduction::by(vec![2]));
     assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }]));
     // Inner: Aggregate{Quantile} over TimeRange (per-series over_time reduction).
     let QueryExpr::Aggregate { aggs, child, .. } = child.as_ref() else {
@@ -184,10 +190,13 @@ fn histogram_quantile_over_sum_by_le_preserves_grouping() {
     );
     // `sum by (le)` survives as a positional Aggregate (by = [2], `le`) over the
     // inner Rate — no name-based Partition.
-    let QueryExpr::Aggregate { by, aggs, .. } = child.as_ref() else {
+    let QueryExpr::Aggregate {
+        reduction, aggs, ..
+    } = child.as_ref()
+    else {
         panic!("expected `sum by (le)` as a positional Aggregate, got {child:?}");
     };
-    assert_eq!(by, &vec![2]);
+    assert_eq!(reduction, &Reduction::by(vec![2]));
     assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }]));
 }
 
@@ -244,12 +253,15 @@ fn sum_by_over_rate_groups_the_outer_sum() {
     // label-preserving inner Rate. Leaf = [ts, value, job] → by = [2].
     let qe = lower("sum by (job) (rate(http_requests_total[5m]))");
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = &qe
     else {
         panic!("expected outer Aggregate grouped by job, got {qe:?}");
     };
-    assert_eq!(by, &vec![2]);
+    assert_eq!(reduction, &Reduction::by(vec![2]));
     assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }]));
     assert!(matches!(
         child.as_ref(),
@@ -290,12 +302,15 @@ fn outer_count_is_cardinality() {
     // on a positional `Aggregate.by`. Leaf = [ts, value, symbol] → symbol = col 2.
     let qe = lower("count by (symbol) (count_over_time(financial_last_trade_price[5m]))");
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = &qe
     else {
         panic!("expected outer Aggregate grouped by symbol, got {qe:?}");
     };
-    assert_eq!(by, &vec![2]);
+    assert_eq!(reduction, &Reduction::by(vec![2]));
     assert!(matches!(aggs.as_slice(), [AggIntent::Cardinality { .. }]));
     // Inner: Aggregate{Count} over TimeRange (per-series count_over_time).
     let QueryExpr::Aggregate { aggs, child, .. } = child.as_ref() else {
@@ -312,13 +327,16 @@ fn topk_over_count_is_heavy_hitter_topk() {
     let qe = lower(r#"topk by (service) (10, count_over_time(requests{env="prod"}[1m]))"#);
     // Heavy-hitter: Aggregate{TopK} with grouping resolved to positional ids.
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = &qe
     else {
         panic!("expected Aggregate with TopK, got {qe:?}");
     };
     // `service` is the only group key → resolved to a positional ColumnId.
-    assert_eq!(by.len(), 1);
+    assert_eq!(reduction.expect_reduce().len(), 1);
     assert!(matches!(aggs.as_slice(), [AggIntent::TopK { k: 10, .. }]));
     // The count_over_time under the TopK is a TimeRange-backed aggregate.
     let QueryExpr::Aggregate { aggs, child, .. } = child.as_ref() else {
@@ -357,8 +375,8 @@ fn topk_over_avg_is_generic_sort_limit() {
     // Underneath: the label-preserving windowed avg aggregate (by: []), no
     // intervening Partition.
     assert!(
-        matches!(child.as_ref(), QueryExpr::Aggregate { by, aggs, .. }
-            if by.is_empty() && matches!(aggs.as_slice(), [AggIntent::Avg { .. }])),
+        matches!(child.as_ref(), QueryExpr::Aggregate { reduction, aggs, .. }
+            if reduction == &Reduction::PerEntity && matches!(aggs.as_slice(), [AggIntent::Avg { .. }])),
         "expected bare per-series Avg aggregate under Sort, got {child:?}"
     );
 }
@@ -413,12 +431,19 @@ fn topk_count_output_schema_carries_group_key() {
     // [ts, value, service] → TopK groups on service (col 2).
     let qe = lower("topk by (service) (5, count_over_time(m[1m]))");
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = &qe
     else {
         panic!("expected Aggregate{{TopK}}, got {qe:?}");
     };
-    assert_eq!(by, &vec![2], "service is col 2 in [ts, value, service]");
+    assert_eq!(
+        reduction,
+        &Reduction::by(vec![2]),
+        "service is col 2 in [ts, value, service]"
+    );
     assert!(matches!(aggs.as_slice(), [AggIntent::TopK { k: 5, .. }]));
     // Inner Count aggregate is visible with its TimeRange child.
     let QueryExpr::Aggregate { aggs, child, .. } = child.as_ref() else {
@@ -533,11 +558,15 @@ fn without_grouping_lowers_to_the_exclusion_form() {
     // `without` form, and the output schema stays open.
     let qe = lower("sum without (instance) (rate(m[5m]))");
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = &qe
     else {
         panic!("expected an Aggregate, got {qe:?}");
     };
+    let by = reduction.expect_reduce();
     assert!(by.is_without());
     assert_eq!(by.keys().len(), 1, "excluded `instance`");
     assert!(matches!(aggs.as_slice(), [AggIntent::Sum { .. }]));
@@ -707,18 +736,22 @@ fn batch_rejects_non_promql_language() {
 
 #[test]
 fn reducing_group_by_lowers_to_aggregate_by() {
-    // Cross-series reduce, no keys → bare `Aggregate { by: [] }`.
+    // Cross-series reduce, no keys → bare `Aggregate { reduction: Reduce([]) }`.
     let q = lower("sum(http_requests_total)");
-    assert!(matches!(q, QueryExpr::Aggregate { ref by, .. } if by.is_empty()));
+    assert!(
+        matches!(q, QueryExpr::Aggregate { ref reduction, .. } if reduction == &Reduction::by(vec![]))
+    );
 
-    // Cross-series reduce grouped by a label → `Aggregate.by`.
+    // Cross-series reduce grouped by a label → `Aggregate.reduction`.
     let q = lower("sum by (job) (http_requests_total)");
-    assert!(matches!(q, QueryExpr::Aggregate { ref by, .. } if by.len() == 1));
+    assert!(matches!(q, QueryExpr::Aggregate { ref reduction, .. }
+        if reduction.expect_reduce().len() == 1));
 
     // Reduce over a label-preserving `rate` grouped by a label → still
-    // `Aggregate.by` (the keys resolve against rate's preserved schema).
+    // `Aggregate.reduction` (the keys resolve against rate's preserved schema).
     let q = lower("sum by (job) (rate(http_requests_total[5m]))");
-    assert!(matches!(q, QueryExpr::Aggregate { ref by, .. } if by.len() == 1));
+    assert!(matches!(q, QueryExpr::Aggregate { ref reduction, .. }
+        if reduction.expect_reduce().len() == 1));
 }
 
 #[test]
@@ -739,7 +772,9 @@ fn generic_topk_grouping_lowers_to_sort_partition_by() {
         panic!("expected Sort, got {child:?}");
     };
     assert_eq!(partition_by, &vec![2], "host is col 2 in [ts, value, host]");
-    assert!(matches!(child.as_ref(), QueryExpr::Aggregate { by, .. } if by.is_empty()));
+    assert!(
+        matches!(child.as_ref(), QueryExpr::Aggregate { reduction, .. } if reduction == &Reduction::PerEntity)
+    );
 }
 
 #[test]
diff --git a/crates/frontend-sql/tests/bgp_analytics/bgp_analytics.rs b/crates/frontend-sql/tests/bgp_analytics/bgp_analytics.rs
index 88adc14e..4f607c24 100644
--- a/crates/frontend-sql/tests/bgp_analytics/bgp_analytics.rs
+++ b/crates/frontend-sql/tests/bgp_analytics/bgp_analytics.rs
@@ -205,7 +205,9 @@ async fn corpus_lowering_matches_the_pinned_per_query_outcome() {
 
 fn first_aggregate(qe: &QueryExpr) -> Option<(&GroupKeys, &Vec<AggIntent>)> {
     match qe {
-        QueryExpr::Aggregate { by, aggs, .. } => Some((by, aggs)),
+        QueryExpr::Aggregate {
+            reduction, aggs, ..
+        } => Some((reduction.expect_reduce(), aggs)),
         QueryExpr::Project { child, .. }
         | QueryExpr::Filter { child, .. }
         | QueryExpr::Sort { child, .. }
diff --git a/crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs b/crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs
index 54a6f853..1aab4f41 100644
--- a/crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs
+++ b/crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs
@@ -122,10 +122,14 @@ fn intents(e: &QueryExpr) -> Vec<AggIntent> {
     out
 }
 
-/// The first `Aggregate`'s `(by, aggs)` along the single-child spine.
+/// The first `Aggregate`'s `(by, aggs)` along the single-child spine. SQL
+/// never lowers to `Reduction::PerEntity` (it has no per-series concept), so
+/// `expect_reduce()` here is a safe, load-bearing assumption for these tests.
 fn first_aggregate(qe: &QueryExpr) -> Option<(&GroupKeys, &Vec<AggIntent>)> {
     match qe {
-        QueryExpr::Aggregate { by, aggs, .. } => Some((by, aggs)),
+        QueryExpr::Aggregate {
+            reduction, aggs, ..
+        } => Some((reduction.expect_reduce(), aggs)),
         QueryExpr::Project { child, .. }
         | QueryExpr::Filter { child, .. }
         | QueryExpr::Distinct { child, .. }
diff --git a/crates/frontend-sql/tests/netflow/netflow.rs b/crates/frontend-sql/tests/netflow/netflow.rs
index 05207e26..187d138f 100644
--- a/crates/frontend-sql/tests/netflow/netflow.rs
+++ b/crates/frontend-sql/tests/netflow/netflow.rs
@@ -192,7 +192,9 @@ impl AggKind {
 
 fn first_aggregate(qe: &QueryExpr) -> Option<(&GroupKeys, &Vec<AggIntent>)> {
     match qe {
-        QueryExpr::Aggregate { by, aggs, .. } => Some((by, aggs)),
+        QueryExpr::Aggregate {
+            reduction, aggs, ..
+        } => Some((reduction.expect_reduce(), aggs)),
         QueryExpr::Project { child, .. }
         | QueryExpr::Filter { child, .. }
         | QueryExpr::Window { child, .. }
@@ -229,8 +231,11 @@ fn aggregate_by_with(
     let expected_by = GroupKeys::by(by.to_vec());
     let mut found = false;
     visit(qe, &mut |node| {
-        if let QueryExpr::Aggregate { by, aggs, .. } = node {
-            found |= *by == expected_by && aggs.iter().any(&pred);
+        if let QueryExpr::Aggregate {
+            reduction, aggs, ..
+        } = node
+        {
+            found |= *reduction.expect_reduce() == expected_by && aggs.iter().any(&pred);
         }
     });
     found
diff --git a/crates/frontend-sql/tests/sql_lowering.rs b/crates/frontend-sql/tests/sql_lowering.rs
index fb22711c..d8e80279 100644
--- a/crates/frontend-sql/tests/sql_lowering.rs
+++ b/crates/frontend-sql/tests/sql_lowering.rs
@@ -7,7 +7,8 @@
 use asap_frontend_sql::{lower_sql, SqlCatalog};
 use asap_ir::intent_algebra::schema::{Column, DataType, Schema};
 use asap_ir::intent_algebra::{
-    AggIntent, CompareOp, GroupKeys, JoinKind, L3Expr, L3Scalar, QueryExpr, Source, WindowFuncKind,
+    AggIntent, CompareOp, GroupKeys, JoinKind, L3Expr, L3Scalar, QueryExpr, Reduction, Source,
+    WindowFuncKind,
 };
 use asap_ir::types::AccuracyTarget;
 
@@ -49,7 +50,9 @@ async fn lower(sql: &str) -> QueryExpr {
 /// Find the first `Aggregate` node along the single-child spine.
 fn find_aggregate(qe: &QueryExpr) -> Option<(&GroupKeys, &Vec<AggIntent>)> {
     match qe {
-        QueryExpr::Aggregate { by, aggs, .. } => Some((by, aggs)),
+        QueryExpr::Aggregate {
+            reduction, aggs, ..
+        } => Some((reduction.expect_reduce(), aggs)),
         QueryExpr::Project { child, .. }
         | QueryExpr::Filter { child, .. }
         | QueryExpr::Window { child, .. }
@@ -1049,7 +1052,10 @@ async fn time_bucketing_group_by_lowers_to_a_derived_key() {
         lower("SELECT date_trunc('minute', ts) AS m, SUM(bytes) FROM metrics GROUP BY m").await;
     let node = find_aggregate_node(&qe).expect("expected an Aggregate");
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = node
     else {
         unreachable!()
@@ -1059,7 +1065,7 @@ async fn time_bucketing_group_by_lowers_to_a_derived_key() {
         "expected a materializing Project beneath the Aggregate"
     );
     let schema = child.output_schema().expect("child schema");
-    assert_eq!(by, &GroupKeys::by(vec![0]));
+    assert_eq!(reduction, &Reduction::by(vec![0]));
     assert!(
         schema.columns[0].name.contains("date_trunc"),
         "group key should be the projected bucket, got {:?}",
@@ -1155,7 +1161,7 @@ fn grouping_levels(qe: &QueryExpr) -> Vec<(GroupKeys, Vec<String>)> {
             let QueryExpr::Project { child, .. } = b else {
                 panic!("expected a Project per level, got {b:?}");
             };
-            let QueryExpr::Aggregate { by, .. } = child.as_ref() else {
+            let QueryExpr::Aggregate { reduction, .. } = child.as_ref() else {
                 panic!("expected an Aggregate under the Project, got {child:?}");
             };
             let names = b
@@ -1165,7 +1171,7 @@ fn grouping_levels(qe: &QueryExpr) -> Vec<(GroupKeys, Vec<String>)> {
                 .iter()
                 .map(|c| c.name.clone())
                 .collect();
-            (by.clone(), names)
+            (reduction.expect_reduce().clone(), names)
         })
         .collect()
 }
diff --git a/crates/ir/src/intent_algebra/mod.rs b/crates/ir/src/intent_algebra/mod.rs
index 841a9862..b693414a 100644
--- a/crates/ir/src/intent_algebra/mod.rs
+++ b/crates/ir/src/intent_algebra/mod.rs
@@ -28,7 +28,7 @@ pub use expr_ir::{ArithOp, ColumnRef, CompareOp, Expr, L2Expr, L3Expr, L3Scalar}
 pub use names::{BindingName, QueryId};
 pub use query_expr::{
     aggregate_output_schema, AtModifier, BinaryOpKind, BindingScope, DataModel, GroupKeys,
-    GroupSide, InfoMatcher, JoinKind, Predicate, ProjectItem, QueryExpr, QueryExprError,
+    GroupSide, InfoMatcher, JoinKind, Predicate, ProjectItem, QueryExpr, QueryExprError, Reduction,
     SampleKind, SetOpKind, SortKey, Source, TimeShift, VectorGrouping, VectorMatch,
     VectorMatchKind, WindowFuncKind, WindowKind,
 };
diff --git a/crates/ir/src/intent_algebra/query_expr.rs b/crates/ir/src/intent_algebra/query_expr.rs
index e1934c04..f4669f0e 100644
--- a/crates/ir/src/intent_algebra/query_expr.rs
+++ b/crates/ir/src/intent_algebra/query_expr.rs
@@ -2,8 +2,9 @@
 //!
 //! Language- and deployment-independent. Box-owned tree (DAG fan-in is
 //! expressed via `LetBinding` / `Ref`); column identity is **positional**
-//! (`Aggregate.by: GroupKeys`), resolved by the [`Binder`](super::binder)
-//! against the self-contained [`Schema`] carried on each `Scan`.
+//! (`Aggregate.reduction: Reduction`, wrapping `GroupKeys` for the
+//! grouped case), resolved by the [`Binder`](super::binder) against the
+//! self-contained [`Schema`] carried on each `Scan`.
 
 use std::collections::HashMap;
 use std::time::Duration;
@@ -423,6 +424,57 @@ pub struct ProjectItem {
 
 // ── L3 intent algebra IR ──────────────────────────────────────────────────────
 
+/// What kind of computation an `Aggregate` node performs — orthogonal to
+/// *which* columns it groups by (that's still [`GroupKeys`], inside
+/// `Reduce`). Explicit, decided once by whichever pass constructs the node
+/// (structural, at L2→L3 lowering), rather than inferred downstream from
+/// whether a grouping-key list happens to be empty or from a neighboring
+/// node's shape. See design proposal #165.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub enum Reduction {
+    /// Collapses input rows via `by` — `by`/`without` semantics are exactly
+    /// [`GroupKeys`]'s. May still collapse every row into one (an empty,
+    /// non-`without` `by`) — that's a genuine reduction with zero grouping
+    /// columns, not "no grouping concept."
+    Reduce(GroupKeys),
+    /// No grouping concept at all: preserves one output row per input
+    /// entity (e.g. a per-series windowed computation with no `by(...)`
+    /// clause to begin with, because there's no aggregation operator here
+    /// for such a clause to attach to). Never merges across entities, and
+    /// never collapses an entity's own row structure (e.g. a time axis) —
+    /// unlike `Reduce(GroupKeys::without(vec![]))` ("group by every
+    /// label"), which is still a genuine reduction and does collapse it.
+    PerEntity,
+}
+
+impl Reduction {
+    /// Shorthand for the common case — group by these (possibly empty)
+    /// keys, kept rather than excluded.
+    pub fn by(keys: Vec<ColumnId>) -> Self {
+        Self::Reduce(GroupKeys::by(keys))
+    }
+
+    /// The grouping keys, if this is a genuine reduction — `None` for
+    /// `PerEntity`, which has no grouping-keys concept to report.
+    pub fn group_keys(&self) -> Option<&GroupKeys> {
+        match self {
+            Self::Reduce(by) => Some(by),
+            Self::PerEntity => None,
+        }
+    }
+
+    /// The grouping keys, panicking if this is `PerEntity` — for call sites
+    /// (tests, mostly) that already know, from the shape they built or are
+    /// asserting on, that this must be a genuine reduction. Prefer
+    /// [`group_keys`](Self::group_keys) wherever the caller can't assume that.
+    pub fn expect_reduce(&self) -> &GroupKeys {
+        match self {
+            Self::Reduce(by) => by,
+            Self::PerEntity => panic!("expected Reduction::Reduce, got PerEntity"),
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub enum QueryExpr {
     /// Outermost leaf. `schema` is the **binding schema** — the resolved column
@@ -523,7 +575,7 @@ pub enum QueryExpr {
 
     /// γ + α — GROUP BY (positional) + aggregate intents.
     Aggregate {
-        by: GroupKeys,
+        reduction: Reduction,
         aggs: Vec<AggIntent>,
         /// Output column names parallel to `aggs`. A non-empty entry overrides
         /// the synthetic intent-keyed name — SQL threads DataFusion's generated
@@ -696,36 +748,14 @@ impl QueryExpr {
             QueryExpr::Window { child, .. } => child.output_schema_in(scope),
 
             QueryExpr::Aggregate {
-                by,
+                reduction,
                 aggs,
                 output_names,
                 child,
                 ..
             } => {
                 let in_schema = child.output_schema_in(scope)?;
-
-                // Per-series range reduction: `rate`/`increase` (is_per_series)
-                // OR any single aggregate whose direct child is a `TimeRange`
-                // (`*_over_time` functions) or a `Subquery` (`*_over_time` over a
-                // sub-query, e.g. `max_over_time(rate(m[5m])[1h:])`). All produce
-                // one value per series and are label-preserving — the range child
-                // is the structural marker that confers per-series semantics on
-                // otherwise cross-series intents like `Avg`/`Sum`/`Count`. A
-                // cross-series aggregation operator over a range vector is a
-                // PromQL type error the parser rejects, so an `Aggregate` over a
-                // `Subquery` is only ever this per-series `*_over_time` shape.
-                let is_range_child = matches!(
-                    child.as_ref(),
-                    QueryExpr::TimeRange { .. } | QueryExpr::Subquery { .. }
-                );
-                // A `without(...)` grouping is cross-series even when its
-                // exclusion list is empty (`without ()` = group by all labels),
-                // so it never takes the per-series-global path.
-                let per_series = by.is_empty()
-                    && !by.is_without()
-                    && aggs.len() == 1
-                    && (aggs[0].is_per_series() || is_range_child);
-                aggregate_output_schema(&in_schema, by, aggs, output_names, per_series)
+                aggregate_output_schema(&in_schema, reduction, aggs, output_names)
             }
 
             QueryExpr::LetBinding { name, expr, child } => {
@@ -995,29 +1025,30 @@ fn per_series_reduction_schema(input: &Schema, agg: &AggIntent) -> Schema {
     }
 }
 
-/// The output schema of an `Aggregate { by, aggs }` over `in_schema` — the
-/// **single** canonical derivation shared by [`QueryExpr::output_schema_in`]'s
-/// `Aggregate` arm and the converter's HAVING-resolution path
-/// (`column_resolution::output_schema_for_aggregate`), so the two can never
-/// drift (issue #41).
+/// The output schema of an `Aggregate { reduction, aggs }` over `in_schema` —
+/// the **single** canonical derivation shared by
+/// [`QueryExpr::output_schema_in`]'s `Aggregate` arm and the converter's
+/// HAVING-resolution path (`column_resolution::output_schema_for_aggregate`),
+/// so the two can never drift (issue #41).
 ///
-/// `per_series` selects the label-preserving [`per_series_reduction_schema`]
-/// (`rate`/`increase`/`*_over_time`) instead of the cross-series `by ++ aggs`
-/// shape. The caller supplies it because the decision depends on the *child
-/// node* (a `TimeRange`/`Subquery` marker), which this function does not see;
-/// the child-independent part is `by.is_empty() && aggs.len() == 1 &&
-/// aggs[0].is_per_series()`.
+/// `Reduction::PerEntity` selects the label-preserving
+/// [`per_series_reduction_schema`] (`rate`/`increase`/`*_over_time`) instead
+/// of the cross-series `by ++ aggs` shape. Which one applies is read directly
+/// off `reduction` — decided once, at construction, by whoever built the
+/// `Aggregate` node (issue #165) — not re-derived here from `by`/child shape.
 pub fn aggregate_output_schema(
     in_schema: &Schema,
-    by: &GroupKeys,
+    reduction: &Reduction,
     aggs: &[AggIntent],
     output_names: &[String],
-    per_series: bool,
 ) -> Result<Schema, QueryExprError> {
-    if per_series {
-        debug_assert_eq!(aggs.len(), 1, "a per-series reduction is single-aggregate");
-        return Ok(per_series_reduction_schema(in_schema, &aggs[0]));
-    }
+    let by = match reduction {
+        Reduction::PerEntity => {
+            debug_assert_eq!(aggs.len(), 1, "a per-entity reduction is single-aggregate");
+            return Ok(per_series_reduction_schema(in_schema, &aggs[0]));
+        }
+        Reduction::Reduce(by) => by,
+    };
 
     // `without(excluded)` groups by every label *except* those listed: the kept
     // labels are the input's label columns minus the excluded positions (and the
@@ -1433,7 +1464,7 @@ mod tests {
             ),
         };
         let agg = QueryExpr::Aggregate {
-            by: GroupKeys::without(vec![2]), // exclude `instance`
+            reduction: Reduction::Reduce(GroupKeys::without(vec![2])), // exclude `instance`
             aggs: vec![AggIntent::Sum { col: None }],
             output_names: vec![],
             having: None,
@@ -1511,7 +1542,7 @@ mod tests {
             vec![],
         );
         let rate = QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction: Reduction::PerEntity,
             aggs: vec![AggIntent::Rate],
             output_names: vec![],
             having: None,
@@ -1549,7 +1580,7 @@ mod tests {
             vec![],
         );
         let avg_over_time = QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction: Reduction::PerEntity,
             aggs: vec![AggIntent::Avg { col: None }],
             output_names: vec![],
             having: None,
@@ -1598,7 +1629,7 @@ mod tests {
         );
 
         let rate = QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction: Reduction::PerEntity,
             aggs: vec![AggIntent::Rate],
             output_names: vec![],
             having: None,
@@ -1610,7 +1641,7 @@ mod tests {
         );
 
         let sum_by_job = QueryExpr::Aggregate {
-            by: vec![2].into(), // `job`
+            reduction: Reduction::by(vec![2]), // `job`
             aggs: vec![AggIntent::Sum { col: None }],
             output_names: vec![],
             having: None,
diff --git a/crates/l2/src/canonicalize.rs b/crates/l2/src/canonicalize.rs
index 2fa91601..cc8b9532 100644
--- a/crates/l2/src/canonicalize.rs
+++ b/crates/l2/src/canonicalize.rs
@@ -14,7 +14,7 @@
 //! generic `topk` path); this pass promotes that shape to the canonical
 //!
 //! ```text
-//! Aggregate { by: <partition>, aggs: [TopK{k}],
+//! Aggregate { reduction: Reduce(<partition>), aggs: [TopK{k}],
 //!             child: Aggregate { aggs: [Count], … } }
 //! ```
 //!
@@ -27,7 +27,7 @@
 use asap_ir::intent_algebra::agg_intent::{is_frequency_heavy_hitter, ranking_measure, AggIntent};
 use asap_ir::intent_algebra::expr_ir::{CompareOp, L3Scalar};
 use asap_ir::intent_algebra::query_expr::{
-    GroupKeys, Predicate, QueryExpr, SortKey, WindowFuncKind,
+    Predicate, QueryExpr, Reduction, SortKey, WindowFuncKind,
 };
 use asap_ir::intent_algebra::L3Expr;
 
@@ -129,8 +129,16 @@ fn try_promote_heavy_hitter(expr: &QueryExpr) -> Option<QueryExpr> {
     };
 
     // Exactly one aggregate, ranked by *its* output column — the measure sits at
-    // index `by.len()` (after the group keys).
-    let QueryExpr::Aggregate { by, aggs, .. } = agg_expr else {
+    // index `by.len()` (after the group keys). A `PerEntity` reduction has no
+    // `by` to rank a measure against — this shape can't be heavy-hitter
+    // promoted, so it's a non-match rather than an error.
+    let QueryExpr::Aggregate {
+        reduction, aggs, ..
+    } = agg_expr
+    else {
+        return None;
+    };
+    let Reduction::Reduce(by) = reduction else {
         return None;
     };
     let [ranked_agg] = aggs.as_slice() else {
@@ -160,7 +168,7 @@ fn try_promote_heavy_hitter(expr: &QueryExpr) -> Option<QueryExpr> {
     // global `ORDER BY … LIMIT k`; the `by` labels for a partitioned `topk by`),
     // over the *unchanged* inner `Count` aggregate.
     Some(QueryExpr::Aggregate {
-        by: GroupKeys::by(partition_by.to_vec()),
+        reduction: Reduction::by(partition_by.to_vec()),
         aggs: vec![AggIntent::TopK {
             k: *k,
             accuracy: accuracy.clone(),
@@ -246,7 +254,7 @@ fn try_rewrite_rownumber_topk(expr: &QueryExpr) -> Option<QueryExpr> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use asap_ir::intent_algebra::query_expr::{ProjectItem, Source};
+    use asap_ir::intent_algebra::query_expr::{GroupKeys, ProjectItem, Source};
     use asap_ir::intent_algebra::schema::{Column, DataType, Schema};
     use asap_ir::types::AccuracyTarget;
 
@@ -269,7 +277,7 @@ mod tests {
     /// `Aggregate{ by: [1], [Count] }` over the scan — output cols `[service, count]`.
     fn count_by_service() -> QueryExpr {
         QueryExpr::Aggregate {
-            by: GroupKeys::by(vec![1]),
+            reduction: Reduction::by(vec![1]),
             aggs: vec![AggIntent::Count {
                 accuracy: AccuracyTarget::Exact,
             }],
@@ -383,7 +391,7 @@ mod tests {
         // Sort+Limit (issue #38). This pins the reserved-but-not-promoted
         // contract: flipping it on is a future weighted-sketch change.
         let sum = QueryExpr::Aggregate {
-            by: GroupKeys::by(vec![1]),
+            reduction: Reduction::by(vec![1]),
             aggs: vec![AggIntent::Sum { col: None }],
             output_names: vec![],
             having: None,
@@ -417,7 +425,7 @@ mod tests {
     /// region, <agg>]` (3 cols), so a ROW_NUMBER over it appends `rn` at index 3.
     fn grouped(agg: AggIntent) -> QueryExpr {
         QueryExpr::Aggregate {
-            by: GroupKeys::by(vec![1, 2]),
+            reduction: Reduction::by(vec![1, 2]),
             aggs: vec![agg],
             output_names: vec![],
             having: None,
@@ -459,11 +467,17 @@ mod tests {
         }));
         let out = canonicalize(q);
         let QueryExpr::Aggregate {
-            by, aggs, child, ..
+            reduction,
+            aggs,
+            child,
+            ..
         } = &out
         else {
             panic!("expected outer Aggregate([TopK]), got {out:?}");
         };
+        let Reduction::Reduce(by) = reduction else {
+            panic!("expected a Reduce grouping, got {reduction:?}");
+        };
         assert!(matches!(aggs.as_slice(), [AggIntent::TopK { k: 5, .. }]));
         assert_eq!(**by, vec![2], "outer TopK partitioned by region");
         assert!(matches!(child.as_ref(), QueryExpr::Aggregate { aggs, .. }
diff --git a/crates/l2/src/column_resolution.rs b/crates/l2/src/column_resolution.rs
index b256650f..f173f0a5 100644
--- a/crates/l2/src/column_resolution.rs
+++ b/crates/l2/src/column_resolution.rs
@@ -12,7 +12,9 @@ use crate::relational::QueryExpr;
 use asap_ir::intent_algebra::agg_intent::AggIntent;
 use asap_ir::intent_algebra::expr_ir::ColumnRef;
 use asap_ir::intent_algebra::expr_ir::{L2Expr, L3Expr};
-use asap_ir::intent_algebra::query_expr::{aggregate_output_schema, GroupKeys, QueryExprError};
+use asap_ir::intent_algebra::query_expr::{
+    aggregate_output_schema, GroupKeys, QueryExprError, Reduction,
+};
 use asap_ir::intent_algebra::schema::{Column, ColumnId, DataType, Schema};
 
 /// Errors returned by the resolution helpers.
@@ -201,13 +203,18 @@ pub fn output_schema_for_aggregate(
     // Delegate to the single canonical derivation so HAVING resolution can never
     // drift from `QueryExpr::output_schema_in` (issue #41). HAVING is SQL-only
     // and cross-series (SQL has no `without`), but detect the child-independent
-    // per-series case anyway (a lone `rate`/`increase`/`*_over_time` intent) so
-    // the two agree on every shared input — the `TimeRange`/`Subquery` marker the
+    // per-entity case anyway (a lone `rate`/`increase`/`*_over_time` intent) so
+    // the two agree on every shared input — the range-window child marker the
     // canonical arm also keys off is not visible here, and never co-occurs with
     // HAVING.
-    let per_series =
+    let per_entity =
         by.is_empty() && !by.is_without() && aggs.len() == 1 && aggs[0].is_per_series();
-    aggregate_output_schema(input, by, aggs, output_names, per_series)
+    let reduction = if per_entity {
+        Reduction::PerEntity
+    } else {
+        Reduction::Reduce(by.clone())
+    };
+    aggregate_output_schema(input, &reduction, aggs, output_names)
 }
 
 #[cfg(test)]
@@ -364,10 +371,10 @@ mod tests {
             predicates: vec![],
             schema: leaf_schema.clone(),
         };
-        // Aggregate{ by: [], [Rate], child: TimeRange{ Scan } } — a per-series
-        // reduction (label-preserving).
+        // Aggregate{ reduction: PerEntity, [Rate], child: TimeRange{ Scan } } —
+        // a per-series reduction (label-preserving).
         let agg = L3::Aggregate {
-            by: Default::default(),
+            reduction: Reduction::PerEntity,
             aggs: vec![AggIntent::Rate],
             output_names: vec![],
             having: None,
diff --git a/crates/l2/src/lower.rs b/crates/l2/src/lower.rs
index 97aae3ce..ed812990 100644
--- a/crates/l2/src/lower.rs
+++ b/crates/l2/src/lower.rs
@@ -24,7 +24,7 @@ use asap_ir::intent_algebra::agg_intent::AggIntent;
 use asap_ir::intent_algebra::expr_ir::{ColumnRef, L2Expr, L3Expr, L3Scalar};
 use asap_ir::intent_algebra::names::BindingName;
 use asap_ir::intent_algebra::query_expr::{
-    GroupKeys, Predicate, ProjectItem, QueryExpr as CQueryExpr, SortKey, Source,
+    GroupKeys, Predicate, ProjectItem, QueryExpr as CQueryExpr, Reduction, SortKey, Source,
 };
 use asap_ir::intent_algebra::schema::{ColumnId, Schema};
 use asap_ir::types::AccuracyTarget;
@@ -263,15 +263,13 @@ pub fn convert(
                     None => agg_child_raw,
                 };
                 // Resolve the group keys positionally against the aggregate's
-                // input so the grouping lives in `Aggregate.by` — the *same*
-                // shape SQL produces. Only for instant (non-range) aggregates:
-                // an instant aggregate (`sum by (job) (m)`) or a cross-series
-                // reduction over a label-preserving `rate`/`increase`.
+                // input so the grouping lives in `Aggregate.reduction` — the
+                // *same* shape SQL produces. Only for instant (non-range)
+                // aggregates: an instant aggregate (`sum by (job) (m)`) or a
+                // cross-series reduction over a label-preserving `rate`/`increase`.
                 //
                 // A range reduction's keys can't go into `by`: this node must stay
-                // a per-series (label-preserving) reduction, and per-series schema
-                // detection in `output_schema_in` keys off `by.is_empty()` — adding
-                // keys here would flip it to a cross-series reduce. So a windowed
+                // a per-entity (label-preserving) reduction. So a windowed
                 // reduction must carry no group keys: the only PromQL shape that
                 // would put keys here is a generic `topk by (…)`, and that routes
                 // its grouping to `Sort.partition_by` instead (issue #12). Any keys
@@ -286,8 +284,27 @@ pub fn convert(
                 // groups every series into one partition and is omitted from
                 // the output — drop it instead of rejecting the query.
                 let by = group_keys(resolve_group_keys_promql(keys, &agg_in_schema)?, *without);
+                // Decide the reduction kind once, right here, rather than
+                // leaving a downstream consumer to re-derive it (issue #165):
+                // a per-entity reduction is a single intent with no grouping
+                // keys at all, whose intent is inherently per-series
+                // (`rate`/`increase`/…) or whose child is a range-window
+                // wrapper (`TimeRange`/`Subquery` — the `*_over_time` family).
+                // `without ()` is still a genuine reduction (groups by every
+                // label), never per-entity, even with an empty exclusion list.
+                let is_range_child = matches!(
+                    agg_child,
+                    CQueryExpr::TimeRange { .. } | CQueryExpr::Subquery { .. }
+                );
+                let per_entity =
+                    by.is_empty() && !by.is_without() && (intent.is_per_series() || is_range_child);
+                let reduction = if per_entity {
+                    Reduction::PerEntity
+                } else {
+                    Reduction::Reduce(by)
+                };
                 return Ok(CQueryExpr::Aggregate {
-                    by,
+                    reduction,
                     aggs: vec![intent],
                     output_names: vec![aggs[0].alias.clone().unwrap_or_default()],
                     having: None,
@@ -322,7 +339,10 @@ pub fn convert(
                 })
                 .transpose()?;
             CQueryExpr::Aggregate {
-                by,
+                // Always a genuine reduction: this branch is multi-intent or
+                // HAVING-bearing, and per-entity reductions are always a
+                // single, HAVING-less intent (handled above).
+                reduction: Reduction::Reduce(by),
                 aggs: intents,
                 output_names,
                 having,
@@ -380,7 +400,9 @@ pub fn convert(
             let child_schema = child.output_schema()?;
             let by: GroupKeys = resolve_column_refs(by, &child_schema)?.into();
             CQueryExpr::Aggregate {
-                by,
+                // A ranking always reduces (a `by`-empty TopK ranks the whole
+                // input into one ordering, never per-entity).
+                reduction: Reduction::Reduce(by),
                 aggs: vec![AggIntent::TopK {
                     k: *k as usize,
                     accuracy: acc.clone(),
@@ -812,9 +834,15 @@ mod tests {
         };
 
         let l3 = convert(&tree, &schema, &AccuracyTarget::Exact).unwrap();
-        let CQueryExpr::Aggregate { by, aggs, .. } = &l3 else {
+        let CQueryExpr::Aggregate {
+            reduction, aggs, ..
+        } = &l3
+        else {
             panic!("expected Aggregate, got {l3:?}");
         };
+        let Reduction::Reduce(by) = reduction else {
+            panic!("expected a Reduce grouping, got {reduction:?}");
+        };
         assert!(by.is_empty());
         // bytes is column 1, latency is column 2 in the input schema.
         assert_eq!(
@@ -908,11 +936,17 @@ mod tests {
         };
         let l3 = convert_root(&tree, &AccuracyTarget::Exact).unwrap();
         let CQueryExpr::Aggregate {
-            by, aggs, child, ..
+            reduction,
+            aggs,
+            child,
+            ..
         } = &l3
         else {
             panic!("expected multi-agg Aggregate, got {l3:?}");
         };
+        let Reduction::Reduce(by) = reduction else {
+            panic!("expected a Reduce grouping, got {reduction:?}");
+        };
         assert_eq!(by, &vec![3], "region is column 3 of the joined schema");
         assert_eq!(
             aggs[0],
diff --git a/crates/lower/tests/cross_language.rs b/crates/lower/tests/cross_language.rs
index 8b2be5a6..a03c47b0 100644
--- a/crates/lower/tests/cross_language.rs
+++ b/crates/lower/tests/cross_language.rs
@@ -53,7 +53,10 @@ fn promql(q: &str) -> QueryExpr {
 /// `by`) over an inner `Aggregate([Count])`. Returns `(k, outer_by)`.
 fn heavy_hitter(qe: &QueryExpr) -> Option<(usize, GroupKeys)> {
     let QueryExpr::Aggregate {
-        by, aggs, child, ..
+        reduction,
+        aggs,
+        child,
+        ..
     } = qe
     else {
         return None;
@@ -66,7 +69,8 @@ fn heavy_hitter(qe: &QueryExpr) -> Option<(usize, GroupKeys)> {
     let QueryExpr::Aggregate { aggs: inner, .. } = child.as_ref() else {
         return None;
     };
-    matches!(inner.as_slice(), [AggIntent::Count { .. }]).then(|| (*k, by.clone()))
+    matches!(inner.as_slice(), [AggIntent::Count { .. }])
+        .then(|| (*k, reduction.expect_reduce().clone()))
 }
 
 #[tokio::test]
diff --git a/crates/plan/src/bind.rs b/crates/plan/src/bind.rs
index dd9215ce..ab9b1ec2 100644
--- a/crates/plan/src/bind.rs
+++ b/crates/plan/src/bind.rs
@@ -33,7 +33,7 @@ use std::rc::Rc;
 
 use asap_ir::intent_algebra::agg_intent::AggIntent;
 use asap_ir::intent_algebra::expr_ir::ColumnRef;
-use asap_ir::intent_algebra::query_expr::{BindingScope, QueryExpr, QueryExprError};
+use asap_ir::intent_algebra::query_expr::{BindingScope, QueryExpr, QueryExprError, Reduction};
 use asap_ir::intent_algebra::schema::Schema;
 use asap_sketch::{
     L4DataType, L4Field, L4Node, L4Schema, SketchQuery, SummaryExpr, SummaryKind, SummaryParams,
@@ -84,7 +84,7 @@ pub fn implement_tree_in_with(
     cost_model: &dyn CostModel,
 ) -> Result<Rc<L4Node>, ImplementError> {
     if let QueryExpr::Aggregate {
-        by,
+        reduction,
         aggs,
         having,
         child,
@@ -97,12 +97,12 @@ pub fn implement_tree_in_with(
             match implementation_for_with(intent, cost_model) {
                 Implementation::Sketch { kind, params } => {
                     return bind_summary_agg(
-                        expr, by, intent, child, kind, params, scope, true, cost_model,
+                        expr, reduction, intent, child, kind, params, scope, true, cost_model,
                     )
                 }
                 Implementation::ExactAccumulator { kind, params } => {
                     return bind_summary_agg(
-                        expr, by, intent, child, kind, params, scope, false, cost_model,
+                        expr, reduction, intent, child, kind, params, scope, false, cost_model,
                     )
                 }
                 Implementation::PassThrough => {}
@@ -117,7 +117,7 @@ pub fn implement_tree_in_with(
 #[allow(clippy::too_many_arguments)]
 fn bind_summary_agg(
     node: &QueryExpr,
-    by: &[usize],
+    reduction: &Reduction,
     intent: &AggIntent,
     child: &QueryExpr,
     kind: SummaryKind,
@@ -129,9 +129,16 @@ fn bind_summary_agg(
     let child_schema = child.output_schema_in(scope)?;
     // The single canonical L3 derivation (per-series vs cross-series, name
     // overrides) already computes the row shape; L4 only retypes the summary
-    // state column.
+    // state column. `reduction` (read directly, not re-derived — issue #165)
+    // settles both the schema shape and, downstream of this PR, the
+    // `by`-emptiness ambiguity a bare `Vec<ColumnId>` alone can't (issue #163).
+    let per_series = matches!(reduction, Reduction::PerEntity);
+    let by: Vec<usize> = reduction
+        .group_keys()
+        .map(|g| g.to_vec())
+        .unwrap_or_default();
     let out_schema = node.output_schema_in(scope)?;
-    let state_idx = summary_col_index(node, &out_schema, by);
+    let state_idx = summary_col_index(&out_schema, &by, per_series);
 
     let col = summarised_column(intent, &child_schema);
     let query = estimate.then(|| readout(intent, &col, cost_model));
@@ -147,7 +154,7 @@ fn bind_summary_agg(
             sketch: kind,
             params,
             col,
-            by: by.to_vec(),
+            by,
         },
         schema: state_schema,
     });
@@ -169,19 +176,9 @@ fn bind_summary_agg(
 /// cross-series output is `by ++ [agg]` (the column after the keys);
 /// a per-series reduction keeps every label and replaces the sample value
 /// (named `value` — mirror `per_series_reduction_schema`'s fallback).
-fn summary_col_index(node: &QueryExpr, out_schema: &Schema, by: &[usize]) -> usize {
-    let per_series = match node {
-        QueryExpr::Aggregate {
-            by, aggs, child, ..
-        } => {
-            let is_range_child = matches!(
-                child.as_ref(),
-                QueryExpr::TimeRange { .. } | QueryExpr::Subquery { .. }
-            );
-            by.is_empty() && aggs.len() == 1 && (aggs[0].is_per_series() || is_range_child)
-        }
-        _ => false,
-    };
+/// `per_series` is the caller's already-read `Reduction` (issue #165) —
+/// this never re-derives it, so it can't disagree with the caller.
+fn summary_col_index(out_schema: &Schema, by: &[usize], per_series: bool) -> usize {
     if per_series {
         out_schema
             .column_id("value")
@@ -285,9 +282,22 @@ mod tests {
         }
     }
 
+    /// A cross-series reduction, grouped by `by` (possibly empty — a
+    /// genuine full reduction, never "no grouping concept").
     fn agg(by: Vec<usize>, intent: AggIntent, child: QueryExpr) -> QueryExpr {
         QueryExpr::Aggregate {
-            by: by.into(),
+            reduction: Reduction::by(by),
+            aggs: vec![intent],
+            output_names: vec![],
+            having: None,
+            child: Box::new(child),
+        }
+    }
+
+    /// A per-entity reduction: no grouping concept at all (issue #165).
+    fn agg_per_entity(intent: AggIntent, child: QueryExpr) -> QueryExpr {
+        QueryExpr::Aggregate {
+            reduction: Reduction::PerEntity,
             aggs: vec![intent],
             output_names: vec![],
             having: None,
@@ -517,8 +527,7 @@ mod tests {
     fn per_series_rate_keeps_labels_and_retypes_value() {
         // rate(m[5m]) — per-series: every label survives; the sample value
         // column becomes the Rate accumulator state.
-        let q = agg(
-            vec![],
+        let q = agg_per_entity(
             AggIntent::Rate,
             QueryExpr::TimeRange {
                 range: Duration::from_secs(300),
@@ -657,7 +666,7 @@ mod tests {
         ));
 
         let multi = QueryExpr::Aggregate {
-            by: vec![2].into(),
+            reduction: Reduction::by(vec![2]),
             aggs: vec![AggIntent::Sum { col: None }, AggIntent::Avg { col: None }],
             output_names: vec![],
             having: None,
diff --git a/crates/plan/src/cse.rs b/crates/plan/src/cse.rs
index 73bf1c38..7f71a7a8 100644
--- a/crates/plan/src/cse.rs
+++ b/crates/plan/src/cse.rs
@@ -87,13 +87,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,
@@ -116,7 +116,7 @@ pub fn dedupe_subtrees(roots: Vec<(QueryId, QueryExpr)>) -> CseWorkloadPlan {
 mod tests {
     use super::*;
     use asap_ir::intent_algebra::agg_intent::AggIntent;
-    use asap_ir::intent_algebra::query_expr::{Source, WindowKind};
+    use asap_ir::intent_algebra::query_expr::{Reduction, Source, WindowKind};
     use asap_ir::intent_algebra::schema::{Column, DataType, Schema};
     use asap_ir::types::AccuracyTarget;
     use std::time::Duration;
@@ -155,7 +155,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,
@@ -177,7 +177,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,
@@ -201,7 +201,7 @@ mod tests {
     #[test]
     fn dedupe_subtrees_basic() {
         let mk = |q: f64| QueryExpr::Aggregate {
-            by: vec![1].into(),
+            reduction: Reduction::by(vec![1]),
             aggs: vec![AggIntent::Quantile {
                 col: None,
                 q,
@@ -247,7 +247,7 @@ mod tests {
             ),
         };
         let mk = || QueryExpr::Aggregate {
-            by: vec![].into(),
+            reduction: Reduction::by(vec![]),
             aggs: vec![AggIntent::Sum { col: None }],
             output_names: vec![],
             having: None,
diff --git a/docs/design.md b/docs/design.md
index 09ee34c1..8c00f1c5 100644
--- a/docs/design.md
+++ b/docs/design.md
@@ -403,7 +403,7 @@ pub enum QueryExpr {
     /// γ + α — GROUP BY + aggregate intents, with optional HAVING.
     /// `aggs` carry `AggIntent`; concrete sketch / non-sketch operator
     /// is chosen by L4 and lives in the L4-extended IR (`SketchExpr`).
-    Aggregate { child: Box<QueryExpr>, by: GroupKeys,
+    Aggregate { child: Box<QueryExpr>, reduction: Reduction,
                 aggs: Vec<AggIntent>, having: Option<Predicate> },
 
     // ── Time / streaming windows ──────────────────────────────────────────
@@ -416,7 +416,7 @@ pub enum QueryExpr {
 
     // ── Distributed-execution structure ───────────────────────────────────
     // There is no `Partition` node (issue #12). Grouping has one home per
-    // concept: a reducing GROUP BY → `Aggregate.by`; per-group *ranking*
+    // concept: a reducing GROUP BY → `Aggregate.reduction`; per-group *ranking*
     // (split without reduce) → `Sort.partition_by` (below); a parallel/sharding
     // split is physical and lives in L5's stage allocator, not the symbolic IR.
     // All three carry the same `GroupKeys` type — a newtype over the positional