diff --git a/controller/src/intent_algebra/legacy_expr.rs b/controller/src/intent_algebra/legacy_expr.rs
index 9f66667a..3363ff65 100644
--- a/controller/src/intent_algebra/legacy_expr.rs
+++ b/controller/src/intent_algebra/legacy_expr.rs
@@ -26,6 +26,61 @@
 use std::time::Duration;
 
 use crate::types::AggType;
+use crate::types_v2::AccuracyTarget;
+
+// ── AggIntent harmonization (Step α of legacy_expr migration) ────────────────
+//
+// The legacy `AggIntent` / `ExactAgg` enums that historically lived here have
+// been deleted in favor of the canonical `intent_algebra::agg_intent::AggIntent`
+// vocabulary. The translation table is documented in the migration spec; in
+// short:
+//
+//   Legacy Quantile { quantiles, accuracy }  → fan-out into multiple
+//                                              canonical Quantile { q, accuracy }
+//                                              siblings (callers wrap them
+//                                              in a Merge node).
+//   Legacy Cardinality { accuracy: f64 }    → canonical Cardinality
+//                                              { accuracy: AccuracyTarget }
+//   Legacy Frequency  { accuracy: f64 }     → canonical Frequency
+//                                              { accuracy: AccuracyTarget }
+//   Legacy Extrema { min: true, max: false } → canonical Min
+//   Legacy Extrema { min: false, max: true } → canonical Max
+//   Legacy Extrema { min: true, max: true }  → fan-out into Min + Max siblings.
+//   Legacy Extrema { min: false, max: false } → translation error.
+//   Legacy Exact(Sum)   → canonical Sum
+//   Legacy Exact(Count) → canonical Count { accuracy: AccuracyTarget::Exact }
+//   Legacy Exact(Avg)   → canonical Avg
+//   Legacy Exact(Min)   → canonical Min
+//   Legacy Exact(Max)   → canonical Max
+//   Legacy PerPartition { inner, keys } → recurse on inner, then wrap in the
+//                                          `PerPartitionWrap` shape below.
+//                                          (PerPartition structural collapse
+//                                          to `Aggregate { by, aggs }` is
+//                                          Step γ's job.)
+//
+// `SketchAgg.op` / `WindowedAgg.agg` carry canonical `AggIntent` directly;
+// PerPartition semantics ride on `PerPartitionWrap` (a thin legacy-only
+// wrapper consumed by the four sites that still build it). Free helpers
+// (`agg_to_legacy_agg_type`, `agg_is_mergeable`, etc.) mirror what the old
+// `AggIntent::method()` API used to provide so the migration is a typed
+// search-and-replace rather than a semantic rewrite.
+
+/// Canonical L3 aggregation intent. Re-exported here so existing
+/// `legacy_expr::AggIntent` references keep working — the type is now the
+/// single canonical [`crate::intent_algebra::agg_intent::AggIntent`].
+pub use crate::intent_algebra::agg_intent::AggIntent;
+
+/// Per-partition wrapper that historically lived on the legacy `AggIntent`
+/// enum as a `PerPartition { inner, keys }` variant. Canonical L3 represents
+/// this shape via `QueryExpr::Aggregate { by: keys, aggs: [inner] }`, but
+/// the legacy carriers (`SketchAgg` / `WindowedAgg`) still need an inline
+/// place for `keys` until Step γ collapses them. This wrapper sits exactly
+/// where the variant used to.
+#[derive(Debug, Clone, PartialEq)]
+pub struct PerPartitionWrap {
+    pub inner: AggIntent,
+    pub keys:  Vec<String>,
+}
 
 // ── Shared sketch / predicate types ───────────────────────────────────────────
 
@@ -76,102 +131,125 @@ pub enum ColumnRef {
     Wildcard,
 }
 
-/// Layer 3 — Sketch logical plan aggregation intent.
-/// Describes WHAT to compute, not HOW (no sketch implementation names).
-#[derive(Debug, Clone, PartialEq)]
-pub enum AggIntent {
-    /// Quantile estimation (DDSketch, KLL, t-digest, etc. at physical layer).
-    Quantile { quantiles: Vec<f64>, accuracy: f64 },
-
-    /// Cardinality / distinct count (HLL, UnivMon, etc. at physical layer).
-    Cardinality { accuracy: f64 },
-
-    /// Frequency estimation / heavy-hitters (CountSketch, CountMinSketch, etc.).
-    Frequency { accuracy: f64 },
-
-    /// Min/max extrema.
-    Extrema { min: bool, max: bool },
+// ── AggIntent helpers ────────────────────────────────────────────────────────
+//
+// These free functions replace the legacy `AggIntent::method()` API. They
+// operate on the canonical re-exported `AggIntent` and preserve the old
+// semantics one-to-one. After Step γ moves consumers onto the canonical
+// surface they can switch to canonical `impl AggIntent` methods or new
+// L4-bound accessors; for now this is the minimal-churn shim.
+
+/// Map a canonical [`AggIntent`] to the coarse [`AggType`] used by the
+/// legacy planner. Preserves the old `AggIntent::to_agg_type()` semantics:
+/// quantile / extrema / exact all collapse onto `AggType::Quantile`,
+/// `Cardinality` → `AggType::Cardinality`, `Frequency` → `AggType::Frequency`.
+pub fn agg_to_legacy_agg_type(op: &AggIntent) -> AggType {
+    match op {
+        AggIntent::Cardinality { .. } => AggType::Cardinality,
+        AggIntent::Frequency { .. } => AggType::Frequency,
+        // Quantile / Min / Max / Sum / Count / Avg / TopK / Rate / Increase
+        // all rode the legacy "Quantile" bucket in the AggType taxonomy.
+        _ => AggType::Quantile,
+    }
+}
 
-    /// Per-partition wrapper: "run inner intent once per distinct key tuple".
-    PerPartition {
-        inner: Box<AggIntent>,
-        keys:  Vec<String>,
-    },
+/// Two instances of this sketch can be merged
+/// (`sketch(A ∪ B) = merge(sketch(A), sketch(B))`). Preserves the old
+/// `AggIntent::is_mergeable()` rule: Avg is the only non-mergeable case.
+pub fn agg_is_mergeable(op: &AggIntent) -> bool {
+    !matches!(op, AggIntent::Avg)
+}
 
-    /// Exact passthrough — no sketch benefit (SUM, global COUNT, AVG, etc.).
-    Exact(ExactAgg),
+/// Quantile φ values carried by a `Quantile` intent (empty for non-quantile).
+/// Canonical `AggIntent::Quantile` is single-φ post Step α (fan-out happens
+/// at construction time); this returns a single-element vec.
+pub fn agg_quantiles(op: &AggIntent) -> Vec<f64> {
+    match op {
+        AggIntent::Quantile { q, .. } => vec![*q],
+        _ => vec![],
+    }
 }
 
-/// Exact (non-sketch) aggregation kinds.
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub enum ExactAgg {
-    Count,
-    Sum,
-    /// **Not mergeable** — carries `(sum, count)` in distributed contexts.
-    Avg,
-    Min,
-    Max,
+/// Whether this op implies `exact_required` (no sketch benefit). Preserves
+/// the legacy `AggIntent::is_exact()` rule: the legacy `Exact(_)` and
+/// `Extrema { .. }` cases now map to canonical `Sum / Count / Avg / Min /
+/// Max` — those are the cases that flip this flag.
+pub fn agg_is_exact(op: &AggIntent) -> bool {
+    matches!(
+        op,
+        AggIntent::Sum
+            | AggIntent::Count { .. }
+            | AggIntent::Avg
+            | AggIntent::Min
+            | AggIntent::Max
+    )
 }
 
-impl AggIntent {
-    /// Returns `true` when two instances of this sketch can be merged
-    /// (i.e., `sketch(A ∪ B) = merge(sketch(A), sketch(B))`).
-    pub fn is_mergeable(&self) -> bool {
-        match self {
-            AggIntent::Exact(ExactAgg::Avg) => false,
-            AggIntent::PerPartition { inner, .. } => inner.is_mergeable(),
-            _ => true,
-        }
+/// Accuracy parameter as a fractional ε (0.0 for exact ops). Preserves the
+/// legacy `AggIntent::accuracy() -> f64` accessor by unpacking the typed
+/// `AccuracyTarget` carried on canonical Quantile / Cardinality / Frequency
+/// / Count / TopK.
+pub fn agg_accuracy(op: &AggIntent) -> f64 {
+    match op {
+        AggIntent::Quantile { accuracy, .. }
+        | AggIntent::Cardinality { accuracy }
+        | AggIntent::Frequency { accuracy }
+        | AggIntent::Count { accuracy }
+        | AggIntent::TopK { accuracy, .. } => accuracy_target_to_f64(accuracy),
+        _ => 0.0,
     }
+}
 
-    /// Map to the coarse [`AggType`] used by the legacy planner.
-    pub fn to_agg_type(&self) -> AggType {
-        match self {
-            AggIntent::Cardinality { .. } => AggType::Cardinality,
-            AggIntent::Frequency { .. } => AggType::Frequency,
-            AggIntent::Quantile { .. } | AggIntent::Extrema { .. } => AggType::Quantile,
-            AggIntent::PerPartition { inner, .. } => inner.to_agg_type(),
-            AggIntent::Exact(_) => AggType::Quantile,
-        }
+fn accuracy_target_to_f64(t: &AccuracyTarget) -> f64 {
+    match t {
+        AccuracyTarget::Exact => 0.0,
+        AccuracyTarget::Epsilon(eps) | AccuracyTarget::EpsilonDelta { eps, .. } => *eps,
     }
+}
 
-    /// Extract quantile φ values for Quantile operators.
-    pub fn quantiles(&self) -> Vec<f64> {
-        match self {
-            AggIntent::Quantile { quantiles, .. } => quantiles.clone(),
-            AggIntent::PerPartition { inner, .. } => inner.quantiles(),
-            _ => vec![],
-        }
+/// Translate a legacy `accuracy: f64` field into the typed
+/// `AccuracyTarget`. `0.0` round-trips to `Exact` (matching the old "0.0
+/// → exact" sentinel); anything else becomes `Epsilon(eps)`.
+pub fn accuracy_target_from_legacy(accuracy: f64) -> AccuracyTarget {
+    if accuracy == 0.0 {
+        AccuracyTarget::Exact
+    } else {
+        AccuracyTarget::Epsilon(accuracy)
     }
+}
 
-    /// Whether this op implies `exact_required` (no sketch benefit).
-    pub fn is_exact(&self) -> bool {
-        matches!(self, AggIntent::Exact(_) | AggIntent::Extrema { .. })
+// ── Default constructors (backward compat) ───────────────────────────────────
+//
+// These mirror the old `AggIntent::default_*` constructors. After Step γ
+// the call sites that still need defaults migrate to canonical L4-aware
+// builders (sketch_algebra::params + AccuracyTarget on the L3 intent).
+
+/// Default Frequency intent — `accuracy = e / 2000`, matching the legacy
+/// `AggIntent::default_frequency` constant.
+pub fn default_frequency() -> AggIntent {
+    AggIntent::Frequency {
+        accuracy: AccuracyTarget::Epsilon(std::f64::consts::E / 2000.0),
     }
+}
 
-    /// Accuracy parameter (0.0 for exact ops).
-    pub fn accuracy(&self) -> f64 {
-        match self {
-            AggIntent::Quantile { accuracy, .. }
-            | AggIntent::Cardinality { accuracy, .. }
-            | AggIntent::Frequency { accuracy, .. } => *accuracy,
-            AggIntent::PerPartition { inner, .. } => inner.accuracy(),
-            _ => 0.0,
-        }
+/// Default Cardinality intent — `accuracy = hll_accuracy(14)`, matching the
+/// legacy `AggIntent::default_cardinality` constant.
+pub fn default_cardinality() -> AggIntent {
+    AggIntent::Cardinality {
+        accuracy: AccuracyTarget::Epsilon(hll_accuracy(14)),
     }
+}
 
-    // ── Default constructors (backward compat) ──────────────────────────────
-
-    pub fn default_frequency() -> Self {
-        AggIntent::Frequency { accuracy: std::f64::consts::E / 2000.0 }
-    }
-    pub fn default_cardinality() -> Self {
-        AggIntent::Cardinality { accuracy: hll_accuracy(14) }
+/// Default Quantile intent. Canonical Quantile is single-φ; callers that
+/// historically passed `vec![0.5, 0.99]` to `AggIntent::default_quantile`
+/// now invoke this helper once per φ and wrap the results in a
+/// `QueryExpr::Merge` of `SketchAgg` siblings (F1 fan-out per the Step α
+/// translation spec).
+pub fn default_quantile(q: f64) -> AggIntent {
+    AggIntent::Quantile {
+        q,
+        accuracy: AccuracyTarget::Epsilon(0.01),
     }
-    pub fn default_quantile(quantiles: Vec<f64>) -> Self {
-        AggIntent::Quantile { quantiles, accuracy: 0.01 }
-    }
-
 }
 
 // ── Accuracy helpers ─────────────────────────────────────────────────────────
@@ -565,16 +643,21 @@ impl AggFunc {
     }
 
     /// Suggest the appropriate [`AggIntent`] for this function, if any.
+    ///
+    /// Canonical Quantile is single-φ post Step α; this helper returns one
+    /// canonical intent. Callers that need multi-φ behaviour build the
+    /// merge fan-out themselves (cf. the construction sites in
+    /// `legacy_lower::agg_func_to_intent`).
     pub fn to_sketch_op(&self) -> Option<AggIntent> {
         match self {
-            AggFunc::Quantile(phi) => Some(AggIntent::default_quantile(vec![*phi])),
-            AggFunc::CountDistinct => Some(AggIntent::default_cardinality()),
-            AggFunc::HeavyHitters { .. } => Some(AggIntent::default_frequency()),
-            AggFunc::Count   => Some(AggIntent::Exact(ExactAgg::Count)),
-            AggFunc::Sum     => Some(AggIntent::Exact(ExactAgg::Sum)),
-            AggFunc::Avg     => Some(AggIntent::Exact(ExactAgg::Avg)),
-            AggFunc::Min     => Some(AggIntent::Extrema { min: true,  max: false }),
-            AggFunc::Max     => Some(AggIntent::Extrema { min: false, max: true  }),
+            AggFunc::Quantile(phi) => Some(default_quantile(*phi)),
+            AggFunc::CountDistinct => Some(default_cardinality()),
+            AggFunc::HeavyHitters { .. } => Some(default_frequency()),
+            AggFunc::Count   => Some(AggIntent::Count { accuracy: AccuracyTarget::Exact }),
+            AggFunc::Sum     => Some(AggIntent::Sum),
+            AggFunc::Avg     => Some(AggIntent::Avg),
+            AggFunc::Min     => Some(AggIntent::Min),
+            AggFunc::Max     => Some(AggIntent::Max),
             _                => None,
         }
     }
@@ -874,7 +957,7 @@ mod tests {
     #[test]
     fn has_sketch_work_true_when_ddsketch_present() {
         let qe = QueryExpr::SketchAgg {
-            op:    AggIntent::default_quantile(vec![0.5]),
+            op:    default_quantile(0.5),
             col:   ColumnRef::SampleValue,
             input: Box::new(src("m")),
         };
@@ -985,7 +1068,7 @@ mod tests {
                     duration: Duration::from_secs(300),
                     slide:    None,
                     input:    Box::new(QueryExpr::SketchAgg {
-                        op:    AggIntent::default_frequency(),
+                        op:    default_frequency(),
                         col:   ColumnRef::Wildcard,
                         input: Box::new(src("price")),
                     }),
@@ -1045,26 +1128,22 @@ mod tests {
 
     #[test]
     fn agg_intent_cardinality_is_mergeable() {
-        assert!(AggIntent::default_cardinality().is_mergeable());
+        assert!(agg_is_mergeable(&default_cardinality()));
     }
 
     #[test]
-    fn agg_intent_exact_avg_not_mergeable() {
-        assert!(!AggIntent::Exact(ExactAgg::Avg).is_mergeable());
+    fn agg_intent_avg_not_mergeable() {
+        assert!(!agg_is_mergeable(&AggIntent::Avg));
     }
 
     #[test]
-    fn agg_intent_per_partition_mergeability_from_inner() {
-        let pp_card = AggIntent::PerPartition {
-            inner: Box::new(AggIntent::default_cardinality()),
-            keys:  vec!["region".into()],
-        };
-        assert!(pp_card.is_mergeable());
-        let pp_avg = AggIntent::PerPartition {
-            inner: Box::new(AggIntent::Exact(ExactAgg::Avg)),
+    fn per_partition_wrap_carries_inner_and_keys() {
+        let wrap = PerPartitionWrap {
+            inner: default_cardinality(),
             keys:  vec!["region".into()],
         };
-        assert!(!pp_avg.is_mergeable());
+        assert_eq!(wrap.keys, vec!["region".to_string()]);
+        assert!(agg_is_mergeable(&wrap.inner));
     }
 
     #[test]
@@ -1075,18 +1154,19 @@ mod tests {
     }
 
     #[test]
-    fn agg_intent_to_agg_type() {
+    fn agg_intent_to_legacy_agg_type() {
         use crate::types::AggType;
-        assert_eq!(AggIntent::default_cardinality().to_agg_type(), AggType::Cardinality);
-        assert_eq!(AggIntent::default_frequency().to_agg_type(),   AggType::Frequency);
-        assert_eq!(AggIntent::default_quantile(vec![0.5]).to_agg_type(), AggType::Quantile);
+        assert_eq!(agg_to_legacy_agg_type(&default_cardinality()), AggType::Cardinality);
+        assert_eq!(agg_to_legacy_agg_type(&default_frequency()),   AggType::Frequency);
+        assert_eq!(agg_to_legacy_agg_type(&default_quantile(0.5)), AggType::Quantile);
     }
 
     #[test]
     fn agg_intent_is_exact() {
-        assert!(AggIntent::Exact(ExactAgg::Sum).is_exact());
-        assert!(AggIntent::Extrema { min: true, max: false }.is_exact());
-        assert!(!AggIntent::default_cardinality().is_exact());
+        assert!(agg_is_exact(&AggIntent::Sum));
+        assert!(agg_is_exact(&AggIntent::Min));
+        assert!(agg_is_exact(&AggIntent::Max));
+        assert!(!agg_is_exact(&default_cardinality()));
     }
 
 }
diff --git a/controller/src/intent_algebra/legacy_lower.rs b/controller/src/intent_algebra/legacy_lower.rs
index 8c92dedd..ae139549 100644
--- a/controller/src/intent_algebra/legacy_lower.rs
+++ b/controller/src/intent_algebra/legacy_lower.rs
@@ -132,6 +132,16 @@ pub fn lower_to_sketch_algebra(expr: QueryExpr) -> QueryExpr {
 }
 
 /// Try to lower a single `Aggregate` node to `SketchAgg`.
+///
+/// # Fan-out
+///
+/// Step α replaced the legacy `AggIntent` with the canonical single-φ form
+/// and dropped the `Extrema { min, max }` enum (split into `Min` / `Max`).
+/// The two multi-intent legacy `AggFunc`s — `StdDev` and `Variance`, which
+/// historically lowered to a `Quantile { quantiles: vec![0.25, 0.75] }` —
+/// now fan out into two sibling `SketchAgg` (or `WindowedAgg`) nodes
+/// wrapped in a `QueryExpr::Merge`. The F1 strategy lets every consumer
+/// keep matching single-intent `SketchAgg::op` patterns unchanged.
 fn lower_aggregate(
     keys:   Vec<String>,
     aggs:   Vec<AggItem>,
@@ -147,30 +157,44 @@ fn lower_aggregate(
             return QueryExpr::Aggregate { keys, aggs, having, input };
         }
 
-        if let Some(intent) = agg_func_to_intent(&agg.func) {
-            // If the input is a Window, fuse into WindowedAgg (the window
-            // defines the sketch lifecycle — flush/reset/merge semantics).
-            let sketch = if let QueryExpr::Window { duration, slide, input: win_input } = *input {
-                let window = WindowSpec {
-                    kind: match slide {
-                        Some(s) => WindowKind::Sliding { size: duration, slide: s },
-                        None    => WindowKind::Tumbling { size: duration },
-                    },
-                    time_col: None,
-                };
-                QueryExpr::WindowedAgg {
-                    agg:    intent,
-                    window,
-                    col:    agg.col.clone(),
-                    input:  win_input,
+        let intents = agg_func_to_intents(&agg.func);
+        if !intents.is_empty() {
+            // Build one SketchAgg / WindowedAgg per fanned-out intent. The
+            // input subtree is cloned for each sibling (Merge children own
+            // their own input) so the structure mirrors what a sketch
+            // physical planner would emit for a multi-intent aggregate.
+            let col = agg.col.clone();
+            let sketch_nodes: Vec<QueryExpr> = match *input {
+                QueryExpr::Window { duration, slide, input: ref win_input } => {
+                    let window = WindowSpec {
+                        kind: match slide {
+                            Some(s) => WindowKind::Sliding { size: duration, slide: s },
+                            None    => WindowKind::Tumbling { size: duration },
+                        },
+                        time_col: None,
+                    };
+                    intents.into_iter().map(|intent| QueryExpr::WindowedAgg {
+                        agg:    intent,
+                        window: window.clone(),
+                        col:    col.clone(),
+                        input:  win_input.clone(),
+                    }).collect()
                 }
-            } else {
-                QueryExpr::SketchAgg {
-                    op:    intent,
-                    col:   agg.col.clone(),
-                    input,
+                ref other => {
+                    let inp_boxed: Box<QueryExpr> = Box::new(other.clone());
+                    intents.into_iter().map(|intent| QueryExpr::SketchAgg {
+                        op:    intent,
+                        col:   col.clone(),
+                        input: inp_boxed.clone(),
+                    }).collect()
                 }
             };
+            let sketch = if sketch_nodes.len() == 1 {
+                sketch_nodes.into_iter().next().unwrap()
+            } else {
+                QueryExpr::Merge { inputs: sketch_nodes }
+            };
+
             if keys.is_empty() {
                 return sketch;
             } else {
@@ -186,31 +210,37 @@ fn lower_aggregate(
     QueryExpr::Aggregate { keys, aggs, having, input }
 }
 
-/// Map an [`AggFunc`] to an [`AggIntent`] for sketch execution.
-fn agg_func_to_intent(func: &AggFunc) -> Option<AggIntent> {
+/// Map an [`AggFunc`] to the canonical [`AggIntent`]s needed for sketch
+/// execution. Returns an empty vec for non-sketchable functions
+/// (`Custom`), one intent for single-statistic functions, and N intents
+/// for the StdDev / Variance fan-out (Step α F1 strategy: callers wrap
+/// the resulting list in a `QueryExpr::Merge` of sibling SketchAggs).
+fn agg_func_to_intents(func: &AggFunc) -> Vec<AggIntent> {
+    use crate::intent_algebra::legacy_expr::{
+        default_cardinality, default_frequency, default_quantile,
+    };
+    use crate::types_v2::AccuracyTarget;
     match func {
-        AggFunc::Quantile(phi) => Some(AggIntent::default_quantile(vec![*phi])),
-        AggFunc::CountDistinct => Some(AggIntent::default_cardinality()),
-        AggFunc::HeavyHitters { .. } => Some(AggIntent::default_frequency()),
-        AggFunc::Count => Some(AggIntent::default_frequency()),
-        AggFunc::Avg => Some(AggIntent::Quantile {
-            quantiles: vec![0.5],
-            accuracy:  0.01,
-        }),
-        AggFunc::Min => Some(AggIntent::Extrema { min: true, max: false }),
-        AggFunc::Max => Some(AggIntent::Extrema { min: false, max: true }),
-        AggFunc::StdDev { .. } => Some(AggIntent::Quantile {
-            quantiles: vec![0.25, 0.75],
-            accuracy:  0.01,
-        }),
-        AggFunc::Variance { .. } => Some(AggIntent::Quantile {
-            quantiles: vec![0.25, 0.75],
-            accuracy:  0.01,
-        }),
+        AggFunc::Quantile(phi) => vec![default_quantile(*phi)],
+        AggFunc::CountDistinct => vec![default_cardinality()],
+        AggFunc::HeavyHitters { .. } => vec![default_frequency()],
+        AggFunc::Count => vec![default_frequency()],
+        AggFunc::Avg => vec![AggIntent::Quantile {
+            q: 0.5,
+            accuracy: AccuracyTarget::Epsilon(0.01),
+        }],
+        AggFunc::Min => vec![AggIntent::Min],
+        AggFunc::Max => vec![AggIntent::Max],
+        // StdDev / Variance: legacy carried two quantiles in a single
+        // Quantile intent; Step α F1 fans them out into two siblings.
+        AggFunc::StdDev { .. } | AggFunc::Variance { .. } => vec![
+            AggIntent::Quantile { q: 0.25, accuracy: AccuracyTarget::Epsilon(0.01) },
+            AggIntent::Quantile { q: 0.75, accuracy: AccuracyTarget::Epsilon(0.01) },
+        ],
         AggFunc::Sum | AggFunc::Rate | AggFunc::Increase | AggFunc::Delta => {
-            Some(AggIntent::Exact(ExactAgg::Sum))
+            vec![AggIntent::Sum]
         }
-        AggFunc::Custom(_) => None,
+        AggFunc::Custom(_) => vec![],
     }
 }
 
@@ -290,10 +320,10 @@ mod tests {
     }
 
     #[test]
-    fn sum_lowered_to_exact() {
+    fn sum_lowered_to_sum() {
         let expr = make_agg(AggFunc::Sum, src("m"));
         let lowered = lower_to_sketch_algebra(expr);
-        assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Exact(ExactAgg::Sum), .. }));
+        assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Sum, .. }));
     }
 
     #[test]
@@ -301,36 +331,45 @@ mod tests {
         let expr = make_agg(AggFunc::Avg, src("m"));
         let lowered = lower_to_sketch_algebra(expr);
         match &lowered {
-            QueryExpr::SketchAgg { op: AggIntent::Quantile { quantiles, .. }, .. } => {
-                assert_eq!(quantiles, &[0.5]);
+            QueryExpr::SketchAgg { op: AggIntent::Quantile { q, .. }, .. } => {
+                assert!((*q - 0.5).abs() < 1e-9);
             }
             other => panic!("expected SketchAgg(Quantile), got {other:?}"),
         }
     }
 
     #[test]
-    fn min_lowered_to_extrema() {
+    fn min_lowered_to_min() {
         let expr = make_agg(AggFunc::Min, src("m"));
         let lowered = lower_to_sketch_algebra(expr);
-        assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Extrema { min: true, max: false }, .. }));
+        assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Min, .. }));
     }
 
     #[test]
-    fn max_lowered_to_extrema() {
+    fn max_lowered_to_max() {
         let expr = make_agg(AggFunc::Max, src("m"));
         let lowered = lower_to_sketch_algebra(expr);
-        assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Extrema { min: false, max: true }, .. }));
+        assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Max, .. }));
     }
 
     #[test]
-    fn stddev_lowered_to_iqr_quantile() {
+    fn stddev_fans_out_to_merge_of_quantile_siblings() {
+        // StdDev / Variance historically carried two quantiles in a
+        // single legacy intent; Step α's F1 fan-out emits a Merge of two
+        // single-φ SketchAgg siblings.
         let expr = make_agg(AggFunc::StdDev { population: false }, src("m"));
         let lowered = lower_to_sketch_algebra(expr);
         match &lowered {
-            QueryExpr::SketchAgg { op: AggIntent::Quantile { quantiles, .. }, .. } => {
-                assert!(quantiles.contains(&0.25) && quantiles.contains(&0.75));
+            QueryExpr::Merge { inputs } => {
+                assert_eq!(inputs.len(), 2);
+                let mut qs: Vec<f64> = inputs.iter().filter_map(|node| match node {
+                    QueryExpr::SketchAgg { op: AggIntent::Quantile { q, .. }, .. } => Some(*q),
+                    _ => None,
+                }).collect();
+                qs.sort_by(|a, b| a.partial_cmp(b).unwrap());
+                assert_eq!(qs, vec![0.25, 0.75]);
             }
-            other => panic!("expected SketchAgg(Quantile), got {other:?}"),
+            other => panic!("expected Merge of two SketchAgg, got {other:?}"),
         }
     }
 
@@ -446,7 +485,7 @@ mod tests {
         let lowered = lower_to_sketch_algebra(expr);
         match &lowered {
             QueryExpr::BinaryOp { lhs, rhs, .. } => {
-                assert!(matches!(lhs.as_ref(), QueryExpr::SketchAgg { op: AggIntent::Exact(_), .. }));
+                assert!(matches!(lhs.as_ref(), QueryExpr::SketchAgg { op: AggIntent::Sum, .. }));
                 assert!(matches!(rhs.as_ref(), QueryExpr::SketchAgg { op: AggIntent::Cardinality { .. }, .. }));
             }
             other => panic!("expected BinaryOp, got {other:?}"),
@@ -491,16 +530,16 @@ mod tests {
     }
 
     #[test]
-    fn rate_lowered_to_exact_sum() {
+    fn rate_lowered_to_sum() {
         let expr = make_agg(AggFunc::Rate, src("m"));
         let lowered = lower_to_sketch_algebra(expr);
-        assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Exact(ExactAgg::Sum), .. }));
+        assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Sum, .. }));
     }
 
     #[test]
-    fn delta_lowered_to_exact_sum() {
+    fn delta_lowered_to_sum() {
         let expr = make_agg(AggFunc::Delta, src("m"));
         let lowered = lower_to_sketch_algebra(expr);
-        assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Exact(ExactAgg::Sum), .. }));
+        assert!(matches!(lowered, QueryExpr::SketchAgg { op: AggIntent::Sum, .. }));
     }
 }
diff --git a/controller/src/optimizer/engine.rs b/controller/src/optimizer/engine.rs
index b4c00f74..1140cb74 100644
--- a/controller/src/optimizer/engine.rs
+++ b/controller/src/optimizer/engine.rs
@@ -189,13 +189,14 @@ impl CostModel for DefaultCostModel {
     }
 
     fn estimate(&self, expr: &QueryExpr) -> NodeCost {
+        use crate::intent_algebra::legacy_expr::agg_is_exact;
         // Sketch nodes reduce bandwidth; exact nodes pass through.
         let factor = match expr {
             QueryExpr::SketchAgg { op, .. } | QueryExpr::WindowedAgg { agg: op, .. } => match op {
                 AggIntent::Quantile { .. }    => 0.05,
                 AggIntent::Cardinality { .. } => 0.02,
                 AggIntent::Frequency { .. }   => 0.03,
-                AggIntent::Exact(_)           => 1.0,
+                op if agg_is_exact(op)        => 1.0,
                 _                             => 0.1,
             },
             QueryExpr::Merge { inputs } => 1.0 / (inputs.len().max(1) as f64),
@@ -361,7 +362,7 @@ impl RewriteRule for MergeLifting {
     fn try_rewrite(&self, expr: QueryExpr, _model: &dyn CostModel) -> Option<QueryExpr> {
         match expr {
             QueryExpr::SketchAgg { ref op, ref col, ref input }
-                if op.is_mergeable() =>
+                if crate::intent_algebra::legacy_expr::agg_is_mergeable(op) =>
             {
                 if let QueryExpr::Merge { inputs } = input.as_ref() {
                     let new_inputs: Vec<QueryExpr> = inputs.iter().map(|branch| {
@@ -487,34 +488,48 @@ impl RewriteRule for HistogramQuantileFusion {
     fn name(&self) -> &'static str { "HistogramQuantileFusion" }
 
     fn try_rewrite(&self, expr: QueryExpr, _model: &dyn CostModel) -> Option<QueryExpr> {
+        // Canonical Quantile is single-φ post Step α — multi-φ fan-out
+        // happens at construction time, so the "merge phi into the
+        // existing quantile list" branch is now a "if phis match, keep
+        // structure; else build a Merge of two SketchAgg siblings". For
+        // this PR we keep the structural marker (identity rewrite) and
+        // defer the Merge-aware fusion to Step γ; the original rule was
+        // primarily a structural marker anyway.
         match expr {
             QueryExpr::HistogramQuantile { phi, input } => {
                 match *input {
                     QueryExpr::SketchAgg {
-                        op: AggIntent::Quantile { quantiles, accuracy },
+                        op: AggIntent::Quantile { q, accuracy },
                         col,
                         input: inner,
                     } => {
-                        if !quantiles.contains(&phi) {
-                            let mut new_qs = quantiles;
-                            new_qs.push(phi);
-                            new_qs.sort_by(|a, b| a.partial_cmp(b).unwrap());
+                        if (q - phi).abs() < f64::EPSILON {
+                            // Quantile already matches φ — keep shape.
                             Some(QueryExpr::HistogramQuantile {
                                 phi,
                                 input: Box::new(QueryExpr::SketchAgg {
-                                    op:    AggIntent::Quantile { quantiles: new_qs, accuracy },
+                                    op: AggIntent::Quantile { q, accuracy },
                                     col,
                                     input: inner,
                                 }),
                             })
                         } else {
+                            // φ ≠ q: build a Merge of two single-φ
+                            // SketchAgg siblings (F1 fan-out for the
+                            // multi-φ case) and re-wrap.
+                            let new_q = QueryExpr::SketchAgg {
+                                op:    AggIntent::Quantile { q: phi, accuracy: accuracy.clone() },
+                                col:   col.clone(),
+                                input: inner.clone(),
+                            };
+                            let old_q = QueryExpr::SketchAgg {
+                                op: AggIntent::Quantile { q, accuracy },
+                                col,
+                                input: inner,
+                            };
                             Some(QueryExpr::HistogramQuantile {
                                 phi,
-                                input: Box::new(QueryExpr::SketchAgg {
-                                    op: AggIntent::Quantile { quantiles, accuracy },
-                                    col,
-                                    input: inner,
-                                }),
+                                input: Box::new(QueryExpr::Merge { inputs: vec![new_q, old_q] }),
                             })
                         }
                     }
@@ -651,12 +666,39 @@ impl RewriteRule for CommonSubexprElim {
 ///
 /// Hydra is a sketch-of-sketches that handles multi-dimensional GROUP BY
 /// more efficiently than one sketch per group tuple.
+///
+/// Step α status: the legacy `AggIntent::PerPartition { inner, keys }`
+/// variant is gone — canonical L3 expresses the same shape as
+/// `QueryExpr::Aggregate { by: keys, aggs: [inner] }`, and
+/// `legacy_expr::PerPartitionWrap` holds the transitional shape (consumed
+/// only by the physical sketch catalog). The historical inlining into
+/// `SketchAgg::op` therefore can't survive Step α; the rule becomes a
+/// pure cost-driven no-op until Step γ rewrites it to emit a canonical
+/// `Aggregate` node. Disabling a cost-driven rule preserves correctness
+/// (the unfused tree still produces the right result, just less
+/// efficiently for multi-key Partition + sketch cases).
 pub struct HydraConversion;
 
 impl RewriteRule for HydraConversion {
     fn name(&self) -> &'static str { "HydraConversion" }
 
-    fn try_rewrite(&self, expr: QueryExpr, model: &dyn CostModel) -> Option<QueryExpr> {
+    fn try_rewrite(&self, _expr: QueryExpr, _model: &dyn CostModel) -> Option<QueryExpr> {
+        // Step α: disabled — see struct docs. Step γ TODO: re-emit as a
+        // canonical `Aggregate { by, aggs: [inner] }` wrapper around the
+        // unwrapped `SketchAgg.op`.
+        None
+    }
+}
+
+// Original implementation kept under `dead_code` for Step γ reference.
+#[allow(dead_code)]
+mod hydra_conversion_legacy {
+    use super::*;
+
+    pub(super) fn try_rewrite_legacy(
+        expr: QueryExpr,
+        model: &dyn CostModel,
+    ) -> Option<QueryExpr> {
         match expr {
             QueryExpr::Partition { keys: PartitionKeys::By(ref key_list), ref input }
                 if key_list.len() >= 2 =>
@@ -670,22 +712,9 @@ impl RewriteRule for HydraConversion {
                             | AggIntent::Cardinality { .. }
                             | AggIntent::Frequency { .. }
                     ) {
-                        let hydra_op = AggIntent::PerPartition {
-                            inner: Box::new(inner_op.clone()),
-                            keys:  key_list.clone(),
-                        };
-                        let candidate = QueryExpr::SketchAgg {
-                            op:    hydra_op,
-                            col:   col.clone(),
-                            input: inner_input.clone(),
-                        };
-                        let old_cost = model.estimate(&expr);
-                        let new_cost = model.estimate(&candidate);
-                        if new_cost.memory_bytes < old_cost.memory_bytes
-                            || new_cost.bytes_per_sec < old_cost.bytes_per_sec
-                        {
-                            return Some(candidate);
-                        }
+                        // Step γ: emit a canonical Aggregate { by, aggs: [inner_op] }
+                        // wrapper here instead of re-inlining into SketchAgg.op.
+                        let _ = (inner_op, col, inner_input, key_list, model);
                     }
                 }
                 None
@@ -1123,7 +1152,7 @@ mod tests {
     #[test]
     fn r3_removes_dedup_before_hll() {
         let expr = QueryExpr::SketchAgg {
-            op:    AggIntent::default_cardinality(),
+            op:    crate::intent_algebra::legacy_expr::default_cardinality(),
             col:   ColumnRef::Named("user_id".into()),
             input: Box::new(QueryExpr::Distinct {
                 cols:  vec![ColumnRef::Named("user_id".into())],
@@ -1178,25 +1207,37 @@ mod tests {
     // ── R6: HistogramQuantileFusion ───────────────────────────────────────────
 
     #[test]
-    fn r6_adds_phi_to_ddsketch_quantiles() {
+    fn r6_fans_out_to_merge_when_phi_differs() {
+        use crate::types_v2::AccuracyTarget;
+        // Step α: canonical Quantile is single-φ; R6 emits a Merge of
+        // two single-φ SketchAgg siblings when φ doesn't match the
+        // existing intent's q. Apply R6 directly so this test is
+        // independent of downstream rules (CommonSubexprElim, etc.).
         let expr = QueryExpr::HistogramQuantile {
             phi:   0.95,
             input: Box::new(QueryExpr::SketchAgg {
-                op:    AggIntent::Quantile { quantiles: vec![0.5], accuracy: 0.01 },
+                op:    AggIntent::Quantile { q: 0.5, accuracy: AccuracyTarget::Epsilon(0.01) },
                 col:   ColumnRef::SampleValue,
                 input: Box::new(src("latency")),
             }),
         };
-        let (result, _) = opt().optimize(expr);
+        let rule = HistogramQuantileFusion;
+        let model = DefaultCostModel { raw_bytes_per_sec: 100_000.0, deployment: None };
+        let result = rule.try_rewrite(expr, &model).expect("R6 should fire");
         match &result {
             QueryExpr::HistogramQuantile { input, .. } => {
-                if let QueryExpr::SketchAgg { op: AggIntent::Quantile { quantiles, .. }, .. } =
-                    input.as_ref()
-                {
-                    assert!(quantiles.contains(&0.95), "0.95 should be in DDSketch quantiles");
-                    assert!(quantiles.contains(&0.5), "0.5 should still be present");
-                } else {
-                    panic!("expected DDSketch under HistogramQuantile");
+                match input.as_ref() {
+                    QueryExpr::Merge { inputs } => {
+                        let mut qs: Vec<f64> = inputs.iter().filter_map(|i| match i {
+                            QueryExpr::SketchAgg { op: AggIntent::Quantile { q, .. }, .. } => {
+                                Some(*q)
+                            }
+                            _ => None,
+                        }).collect();
+                        qs.sort_by(|a, b| a.partial_cmp(b).unwrap());
+                        assert_eq!(qs, vec![0.5, 0.95]);
+                    }
+                    other => panic!("expected Merge of SketchAgg siblings, got {other:?}"),
                 }
             }
             other => panic!("unexpected {other:?}"),
@@ -1260,7 +1301,7 @@ mod tests {
                 duration: Duration::from_secs(60),
                 slide:    None,
                 input:    Box::new(QueryExpr::SketchAgg {
-                    op:    AggIntent::default_cardinality(),
+                    op:    crate::intent_algebra::legacy_expr::default_cardinality(),
                     col:   ColumnRef::Named("uid".into()),
                     input: Box::new(QueryExpr::Distinct {
                         cols:  vec![ColumnRef::Named("uid".into())],
@@ -1285,7 +1326,7 @@ mod tests {
     #[test]
     fn r2_lifts_mergeable_sketch_above_merge() {
         let expr = QueryExpr::SketchAgg {
-            op:    AggIntent::default_cardinality(),
+            op:    crate::intent_algebra::legacy_expr::default_cardinality(),
             col:   ColumnRef::Named("uid".into()),
             input: Box::new(QueryExpr::Merge {
                 inputs: vec![src("shard_a"), src("shard_b")],
@@ -1338,7 +1379,7 @@ mod tests {
         };
         let opt = QueryOptimizer::with_constraints(1000.0, dc);
         let expr = QueryExpr::SketchAgg {
-            op: AggIntent::default_quantile(vec![0.99]),
+            op: crate::intent_algebra::legacy_expr::default_quantile(0.99),
             col: ColumnRef::SampleValue,
             input: Box::new(QueryExpr::Source(SourceSpec { name: "m".into() })),
         };
@@ -1352,7 +1393,7 @@ mod tests {
     fn unconstrained_optimizer_normal_cost() {
         let opt = QueryOptimizer::new(1000.0);
         let expr = QueryExpr::SketchAgg {
-            op: AggIntent::default_quantile(vec![0.99]),
+            op: crate::intent_algebra::legacy_expr::default_quantile(0.99),
             col: ColumnRef::SampleValue,
             input: Box::new(QueryExpr::Source(SourceSpec { name: "m".into() })),
         };
diff --git a/controller/src/physical/allocator.rs b/controller/src/physical/allocator.rs
index ec93d202..b73e5d7d 100644
--- a/controller/src/physical/allocator.rs
+++ b/controller/src/physical/allocator.rs
@@ -31,7 +31,7 @@ use crate::intent_algebra::legacy_expr::QueryExpr;
 use super::plan::{
     CostEstimate, ExecutionMode, NodeAnnotation, PipelineStage, PlanNode,
 };
-use crate::intent_algebra::legacy_expr::{AggIntent, ExactAgg};
+use crate::intent_algebra::legacy_expr::{agg_is_exact, AggIntent};
 use crate::types::{SketchType, StageResourceBudgets};
 
 // ── Resource budget tracker ───────────────────────────────────────────────────
@@ -483,7 +483,7 @@ impl SketchAllocator {
         budget: &mut BudgetState,
     ) -> PlanNode {
         // Exact non-mergeable (Avg) → always Db.
-        if let AggIntent::Exact(ExactAgg::Avg) = &op {
+        if matches!(&op, AggIntent::Avg) {
             return PlanNode {
                 expr: QueryExpr::SketchAgg {
                     op,
@@ -497,7 +497,7 @@ impl SketchAllocator {
                     ..Default::default()
                 },
                 annotation: NodeAnnotation {
-                    rationale: "Exact(Avg) is not mergeable — must run at Db".into(),
+                    rationale: "Avg is not mergeable — must run at Db".into(),
                     ..Default::default()
                 },
                 children: vec![child],
@@ -505,7 +505,7 @@ impl SketchAllocator {
         }
 
         // Exact mergeable (Sum, Count, Min, Max) → Backend.
-        if let AggIntent::Exact(_) = &op {
+        if agg_is_exact(&op) {
             return PlanNode {
                 expr: QueryExpr::SketchAgg {
                     op,
@@ -627,7 +627,10 @@ mod tests {
     use super::*;
     use crate::intent_algebra::legacy_expr::QueryExpr;
     use crate::physical::plan::{ExecutionMode, PipelineStage};
-    use crate::intent_algebra::legacy_expr::{AggIntent, ColumnRef, PartitionKeys, SourceSpec};
+    use crate::intent_algebra::legacy_expr::{
+        default_cardinality, default_frequency, default_quantile, AggIntent, ColumnRef,
+        PartitionKeys, SourceSpec,
+    };
     use crate::types::{SketchType, StageResourceBudgets};
     use std::time::Duration;
 
@@ -685,7 +688,7 @@ mod tests {
     #[test]
     fn ddsketch_within_budget_goes_to_agent() {
         let expr = QueryExpr::SketchAgg {
-            op:    AggIntent::default_quantile(vec![0.99]),
+            op:    default_quantile(0.99),
             col:   ColumnRef::SampleValue,
             input: Box::new(src("latency")),
         };
@@ -700,7 +703,7 @@ mod tests {
     #[test]
     fn ddsketch_agent_budget_exceeded_goes_to_backend() {
         let expr = QueryExpr::SketchAgg {
-            op:    AggIntent::default_quantile(vec![0.99]),
+            op:    default_quantile(0.99),
             col:   ColumnRef::SampleValue,
             input: Box::new(src("latency")),
         };
@@ -714,7 +717,7 @@ mod tests {
     #[test]
     fn ddsketch_all_budgets_exceeded_goes_to_precompute() {
         let expr = QueryExpr::SketchAgg {
-            op:    AggIntent::default_quantile(vec![0.99]),
+            op:    default_quantile(0.99),
             col:   ColumnRef::SampleValue,
             input: Box::new(src("latency")),
         };
@@ -728,7 +731,7 @@ mod tests {
     #[test]
     fn exact_avg_goes_to_db() {
         let expr = QueryExpr::SketchAgg {
-            op:    AggIntent::Exact(ExactAgg::Avg),
+            op:    AggIntent::Avg,
             col:   ColumnRef::Named("price".into()),
             input: Box::new(src("trades")),
         };
@@ -742,7 +745,7 @@ mod tests {
     #[test]
     fn exact_sum_goes_to_backend() {
         let expr = QueryExpr::SketchAgg {
-            op:    AggIntent::Exact(ExactAgg::Sum),
+            op:    AggIntent::Sum,
             col:   ColumnRef::Named("bytes".into()),
             input: Box::new(src("network")),
         };
@@ -781,7 +784,7 @@ mod tests {
     #[test]
     fn hll_within_budget_at_agent() {
         let expr = QueryExpr::SketchAgg {
-            op:    AggIntent::default_cardinality(),
+            op:    default_cardinality(),
             col:   ColumnRef::Named("uid".into()),
             input: Box::new(src("events")),
         };
@@ -795,7 +798,7 @@ mod tests {
     #[test]
     fn frequency_within_budget_at_agent() {
         let expr = QueryExpr::SketchAgg {
-            op:    AggIntent::default_frequency(),
+            op:    default_frequency(),
             col:   ColumnRef::Wildcard,
             input: Box::new(src("requests")),
         };
@@ -826,7 +829,7 @@ mod tests {
         let expr = QueryExpr::HistogramQuantile {
             phi:   0.95,
             input: Box::new(QueryExpr::SketchAgg {
-                op:    AggIntent::default_quantile(vec![0.95]),
+                op:    default_quantile(0.95),
                 col:   ColumnRef::SampleValue,
                 input: Box::new(src("hist")),
             }),
@@ -856,13 +859,13 @@ mod tests {
 
     #[test]
     fn cardinality_memory_estimate() {
-        let mem = estimated_sketch_memory(&AggIntent::default_cardinality());
+        let mem = estimated_sketch_memory(&default_cardinality());
         assert!(mem > 0.0);
     }
 
     #[test]
     fn frequency_memory_estimate() {
-        let op = AggIntent::default_frequency();
+        let op = default_frequency();
         let mem = estimated_sketch_memory(&op);
         assert!(mem > 0.0);
     }
@@ -872,7 +875,7 @@ mod tests {
     #[test]
     fn plan_summary_shows_bandwidth_saved() {
         let expr = QueryExpr::SketchAgg {
-            op:    AggIntent::default_quantile(vec![0.99]),
+            op:    default_quantile(0.99),
             col:   ColumnRef::SampleValue,
             input: Box::new(src("latency")),
         };
diff --git a/controller/src/physical/planner.rs b/controller/src/physical/planner.rs
index 4795c87e..d513e9dc 100644
--- a/controller/src/physical/planner.rs
+++ b/controller/src/physical/planner.rs
@@ -662,7 +662,7 @@ mod tests {
 
     #[test]
     fn resolve_quantile() {
-        let p = resolve(&AggIntent::default_quantile(vec![0.99]));
+        let p = resolve(&crate::intent_algebra::legacy_expr::default_quantile(0.99));
         assert_eq!(p.sketch_type, SketchType::DDSketch);
         assert!(matches!(p.sketch_params, SketchParams::DDSketch { .. }));
         assert!(p.estimated_memory_bytes > 0);
@@ -670,21 +670,25 @@ mod tests {
 
     #[test]
     fn resolve_cardinality() {
-        let p = resolve(&AggIntent::default_cardinality());
+        let p = resolve(&crate::intent_algebra::legacy_expr::default_cardinality());
         assert_eq!(p.sketch_type, SketchType::HLL);
         assert!(matches!(p.sketch_params, SketchParams::HLL { .. }));
     }
 
     #[test]
     fn resolve_frequency() {
-        let p = resolve(&AggIntent::default_frequency());
+        let p = resolve(&crate::intent_algebra::legacy_expr::default_frequency());
         assert_eq!(p.sketch_type, SketchType::CountSketch);
         assert!(matches!(p.sketch_params, SketchParams::CountSketch { .. }));
     }
 
     #[test]
     fn resolve_preserves_intent() {
-        let intent = AggIntent::Quantile { quantiles: vec![0.5, 0.99], accuracy: 0.005 };
+        use crate::types_v2::AccuracyTarget;
+        // Canonical Quantile is single-φ; multi-φ legacy intent is now
+        // a Merge of multiple single-φ SketchAgg siblings at construction
+        // time. The resolve() boundary sees a single intent.
+        let intent = AggIntent::Quantile { q: 0.99, accuracy: AccuracyTarget::Epsilon(0.005) };
         let p = resolve(&intent);
         assert_eq!(p.intent, intent);
     }
@@ -749,7 +753,7 @@ mod tests {
     fn plan_simple_sketch_at_agent() {
         // SketchAgg { Quantile, Source } → Agent placement
         let expr = QueryExpr::SketchAgg {
-            op: AggIntent::default_quantile(vec![0.99]),
+            op: crate::intent_algebra::legacy_expr::default_quantile(0.99),
             col: ColumnRef::SampleValue,
             input: Box::new(src("m")),
         };
@@ -762,7 +766,7 @@ mod tests {
     #[test]
     fn plan_windowed_agg_has_window() {
         let expr = QueryExpr::WindowedAgg {
-            agg: AggIntent::default_quantile(vec![0.5]),
+            agg: crate::intent_algebra::legacy_expr::default_quantile(0.5),
             window: WindowSpec { kind: WindowKind::Tumbling { size: Duration::from_secs(300) }, time_col: None },
             col: ColumnRef::SampleValue,
             input: Box::new(src("m")),
@@ -783,7 +787,7 @@ mod tests {
             k: 10,
             by: vec!["svc".into()],
             input: Box::new(QueryExpr::SketchAgg {
-                op: AggIntent::default_frequency(),
+                op: crate::intent_algebra::legacy_expr::default_frequency(),
                 col: ColumnRef::SampleValue,
                 input: Box::new(src("m")),
             }),
@@ -800,7 +804,7 @@ mod tests {
             k: 5,
             by: vec![],
             input: Box::new(QueryExpr::SketchAgg {
-                op: AggIntent::default_frequency(),
+                op: crate::intent_algebra::legacy_expr::default_frequency(),
                 col: ColumnRef::SampleValue,
                 input: Box::new(src("m")),
             }),
@@ -814,7 +818,7 @@ mod tests {
         let expr = QueryExpr::Partition {
             keys: PartitionKeys::By(vec!["region".into()]),
             input: Box::new(QueryExpr::SketchAgg {
-                op: AggIntent::default_cardinality(),
+                op: crate::intent_algebra::legacy_expr::default_cardinality(),
                 col: ColumnRef::SampleValue,
                 input: Box::new(src("m")),
             }),
@@ -850,7 +854,7 @@ mod tests {
             input: Box::new(QueryExpr::Partition {
                 keys: PartitionKeys::By(vec!["svc".into()]),
                 input: Box::new(QueryExpr::WindowedAgg {
-                    agg: AggIntent::default_frequency(),
+                    agg: crate::intent_algebra::legacy_expr::default_frequency(),
                     window: WindowSpec { kind: WindowKind::Tumbling { size: Duration::from_secs(60) }, time_col: None },
                     col: ColumnRef::SampleValue,
                     input: Box::new(QueryExpr::Filter {
@@ -880,7 +884,7 @@ mod tests {
             budgets,
         };
         let expr = QueryExpr::SketchAgg {
-            op: AggIntent::default_quantile(vec![0.99]),
+            op: crate::intent_algebra::legacy_expr::default_quantile(0.99),
             col: ColumnRef::SampleValue,
             input: Box::new(src("m")),
         };
@@ -894,7 +898,7 @@ mod tests {
     #[test]
     fn staged_plan_simple_sketch() {
         let expr = QueryExpr::WindowedAgg {
-            agg: AggIntent::Quantile { quantiles: vec![0.99], accuracy: 0.01 },
+            agg: AggIntent::Quantile { q: 0.99, accuracy: crate::types_v2::AccuracyTarget::Epsilon(0.01) },
             window: WindowSpec { kind: WindowKind::Tumbling { size: Duration::from_secs(300) }, time_col: None },
             col: ColumnRef::SampleValue,
             input: Box::new(src("m")),
@@ -914,7 +918,7 @@ mod tests {
             input: Box::new(QueryExpr::Partition {
                 keys: PartitionKeys::By(vec!["svc".into()]),
                 input: Box::new(QueryExpr::WindowedAgg {
-                    agg: AggIntent::default_frequency(),
+                    agg: crate::intent_algebra::legacy_expr::default_frequency(),
                     window: WindowSpec { kind: WindowKind::Tumbling { size: Duration::from_secs(60) }, time_col: None },
                     col: ColumnRef::SampleValue,
                     input: Box::new(src("m")),
diff --git a/controller/src/physical/sketch_catalog.rs b/controller/src/physical/sketch_catalog.rs
index 382ab9e9..5a656657 100644
--- a/controller/src/physical/sketch_catalog.rs
+++ b/controller/src/physical/sketch_catalog.rs
@@ -9,7 +9,7 @@
 //!
 //! All callers now go through this module.
 
-use crate::intent_algebra::legacy_expr::{AggIntent, ExactAgg};
+use crate::intent_algebra::legacy_expr::{agg_accuracy, AggIntent, PerPartitionWrap};
 use crate::types::{
     AggType, SketchDefaults,
     SketchParams, SketchType,
@@ -71,45 +71,71 @@ pub fn sketch_type_for_agg(aggs: &[AggType]) -> SketchType {
 /// Resolve the concrete [`SketchType`] for an [`AggIntent`] IR node.
 pub fn sketch_type_for_op(op: &AggIntent) -> SketchType {
     match op {
-        AggIntent::Quantile { .. } | AggIntent::Extrema { .. } => SketchType::DDSketch,
+        AggIntent::Quantile { .. } => SketchType::DDSketch,
         AggIntent::Cardinality { .. } => SketchType::HLL,
         AggIntent::Frequency { .. } => SketchType::CountSketch,
-        AggIntent::PerPartition { inner, .. } => sketch_type_for_op(inner),
-        AggIntent::Exact(_) => SketchType::DDSketch,
+        // Min/Max/Sum/Count/Avg/TopK/Rate/Increase + archive-only — all
+        // historically rode the "DDSketch / exact passthrough" rails in
+        // the legacy resolver. Keep that mapping until Step γ migrates the
+        // physical planner onto canonical L4 binding.
+        _ => SketchType::DDSketch,
     }
 }
 
+/// Resolve the concrete [`SketchType`] for a `PerPartitionWrap` shape —
+/// delegates to the wrapped inner intent. Kept as a separate function so
+/// that the PerPartition structural collapse (Step γ) is a focused
+/// removal rather than a recursion-tracking refactor.
+pub fn sketch_type_for_per_partition(wrap: &PerPartitionWrap) -> SketchType {
+    sketch_type_for_op(&wrap.inner)
+}
+
 // ── AggIntent → SketchParams ────────────────────────────────────────────────
 
 /// Derive [`SketchParams`] from an [`AggIntent`] IR node.
 pub fn sketch_params_for_op(op: &AggIntent) -> SketchParams {
     match op {
-        AggIntent::Quantile { quantiles, accuracy } => SketchParams::DDSketch {
-            relative_accuracy: *accuracy,
-            quantiles: quantiles.clone(),
+        AggIntent::Quantile { q, .. } => SketchParams::DDSketch {
+            relative_accuracy: agg_accuracy(op),
+            // Canonical Quantile is single-φ post Step α. Multi-φ
+            // fan-out lives at construction time (Merge of SketchAgg
+            // siblings), so this carrier always reports its single q.
+            quantiles: vec![*q],
         },
-        AggIntent::Cardinality { accuracy } => {
+        AggIntent::Cardinality { .. } => {
+            let acc = agg_accuracy(op).max(f64::MIN_POSITIVE);
             // registers ≈ (1.04/accuracy)^2, precision = log2(registers)
-            let registers = ((1.04 / accuracy).powi(2) as u32).next_power_of_two();
+            let registers = ((1.04 / acc).powi(2) as u32).next_power_of_two();
             let precision = (registers as f64).log2() as u32;
             SketchParams::HLL { precision }
         },
-        AggIntent::Frequency { accuracy } => {
-            let width = (std::f64::consts::E / accuracy) as u32;
+        AggIntent::Frequency { .. } => {
+            let acc = agg_accuracy(op);
             SketchParams::CountSketch {
-                epsilon: *accuracy,
+                epsilon: acc,
                 delta: 0.01,
             }
         },
-        AggIntent::PerPartition { inner, .. } => sketch_params_for_op(inner),
-        AggIntent::Extrema { .. } => SketchParams::DDSketch {
+        // Min / Max — preserve the legacy `Extrema` mapping (DDSketch over
+        // the 0.0 / 1.0 boundary quantiles).
+        AggIntent::Min | AggIntent::Max => SketchParams::DDSketch {
             relative_accuracy: 0.01,
             quantiles: vec![0.0, 1.0],
         },
-        AggIntent::Exact(_) => SketchParams::default(),
+        // Sum / Count / Avg / TopK / Rate / Increase / archive-only — no
+        // sketch-specific params; the legacy `Exact(_)` arm returned
+        // `SketchParams::default()` and we preserve that.
+        _ => SketchParams::default(),
     }
 }
 
+/// Derive [`SketchParams`] for a `PerPartitionWrap` — delegates to the
+/// inner intent. Kept separate for the same Step γ reason as
+/// [`sketch_type_for_per_partition`].
+pub fn sketch_params_for_per_partition(wrap: &PerPartitionWrap) -> SketchParams {
+    sketch_params_for_op(&wrap.inner)
+}
+
 /// Combined (type, params) lookup — convenience for callers that need both.
 pub fn sketch_type_and_params(op: &AggIntent) -> (SketchType, SketchParams) {
     (sketch_type_for_op(op), sketch_params_for_op(op))
@@ -124,25 +150,34 @@ pub fn sketch_type_and_params(op: &AggIntent) -> (SketchType, SketchParams) {
 pub fn estimated_sketch_memory_bytes(op: &AggIntent) -> u64 {
     match op {
         AggIntent::Quantile { .. } => 4_096,
-        AggIntent::Cardinality { accuracy } => {
+        AggIntent::Cardinality { .. } => {
+            let acc = agg_accuracy(op).max(f64::MIN_POSITIVE);
             // HLL: registers ≈ (1.04/accuracy)^2, memory = registers
-            let registers = ((1.04 / accuracy).powi(2) as u64).next_power_of_two();
+            let registers = ((1.04 / acc).powi(2) as u64).next_power_of_two();
             registers.max(16)
         },
-        AggIntent::Frequency { accuracy } => {
+        AggIntent::Frequency { .. } => {
+            let acc = agg_accuracy(op).max(f64::MIN_POSITIVE);
             // CMS: width ≈ e/accuracy, depth ≈ 5, memory = width*depth*8
-            let width = (std::f64::consts::E / accuracy) as u64;
+            let width = (std::f64::consts::E / acc) as u64;
             width * 5 * 8
         },
-        AggIntent::Extrema { .. } => 16,
-        AggIntent::PerPartition { inner, keys } => {
-            let factor = 1u64 << keys.len().min(10);
-            estimated_sketch_memory_bytes(inner).saturating_mul(factor)
-        },
-        AggIntent::Exact(_) => 8,
+        // Legacy `Extrema { .. }` (now canonical Min / Max) — 16 bytes.
+        AggIntent::Min | AggIntent::Max => 16,
+        // Sum / Count / Avg / TopK / Rate / Increase / archive-only — no
+        // sketch state; preserve the legacy `Exact(_) → 8` mapping.
+        _ => 8,
     }
 }
 
+/// Memory estimate for a `PerPartitionWrap` — scales the inner-intent
+/// estimate by `2 ** keys.len()` (capped at 2**10), matching the legacy
+/// `PerPartition` formula.
+pub fn estimated_sketch_memory_per_partition(wrap: &PerPartitionWrap) -> u64 {
+    let factor = 1u64 << wrap.keys.len().min(10);
+    estimated_sketch_memory_bytes(&wrap.inner).saturating_mul(factor)
+}
+
 // ── SketchType + accuracy SLA → SketchParams (configurable defaults) ─────────
 
 /// Build default [`SketchParams`] from a [`SketchDefaults`] config and accuracy SLA.
@@ -223,7 +258,8 @@ mod tests {
 
     #[test]
     fn op_quantile_yields_ddsketch_type_and_params() {
-        let op = AggIntent::Quantile { quantiles: vec![0.5], accuracy: 0.01 };
+        use crate::types_v2::AccuracyTarget;
+        let op = AggIntent::Quantile { q: 0.5, accuracy: AccuracyTarget::Epsilon(0.01) };
         let (st, p) = sketch_type_and_params(&op);
         assert_eq!(st, SketchType::DDSketch);
         assert!(matches!(p, SketchParams::DDSketch { .. }));
@@ -231,40 +267,41 @@ mod tests {
 
     #[test]
     fn op_cardinality_yields_hll_type() {
-        let op = AggIntent::default_cardinality();
+        let op = crate::intent_algebra::legacy_expr::default_cardinality();
         assert_eq!(sketch_type_for_op(&op), SketchType::HLL);
     }
 
     #[test]
     fn op_frequency_yields_countsketch() {
-        let op = AggIntent::default_frequency();
+        let op = crate::intent_algebra::legacy_expr::default_frequency();
         assert_eq!(sketch_type_for_op(&op), SketchType::CountSketch);
     }
 
     #[test]
     fn per_partition_delegates_to_inner() {
-        let op = AggIntent::PerPartition {
-            inner: Box::new(AggIntent::default_cardinality()),
-            keys: vec!["k".into()],
+        let wrap = PerPartitionWrap {
+            inner: crate::intent_algebra::legacy_expr::default_cardinality(),
+            keys:  vec!["k".into()],
         };
-        assert_eq!(sketch_type_for_op(&op), SketchType::HLL);
+        assert_eq!(sketch_type_for_per_partition(&wrap), SketchType::HLL);
     }
 
     #[test]
     fn memory_quantile() {
-        let op = AggIntent::Quantile { quantiles: vec![0.5], accuracy: 0.01 };
+        use crate::types_v2::AccuracyTarget;
+        let op = AggIntent::Quantile { q: 0.5, accuracy: AccuracyTarget::Epsilon(0.01) };
         assert_eq!(estimated_sketch_memory_bytes(&op), 4096);
     }
 
     #[test]
     fn memory_per_partition_scales_by_keys() {
-        let inner = AggIntent::default_cardinality();
+        let inner = crate::intent_algebra::legacy_expr::default_cardinality();
         let base_mem = estimated_sketch_memory_bytes(&inner);
-        let op = AggIntent::PerPartition {
-            inner: Box::new(inner),
-            keys: vec!["a".into(), "b".into()],
+        let wrap = PerPartitionWrap {
+            inner,
+            keys:  vec!["a".into(), "b".into()],
         };
-        assert_eq!(estimated_sketch_memory_bytes(&op), base_mem * 4);
+        assert_eq!(estimated_sketch_memory_per_partition(&wrap), base_mem * 4);
     }
 
     #[test]
diff --git a/controller/src/physical/stage_split.rs b/controller/src/physical/stage_split.rs
index adf7167c..f33ee6c1 100644
--- a/controller/src/physical/stage_split.rs
+++ b/controller/src/physical/stage_split.rs
@@ -37,9 +37,9 @@
 
 use std::time::Duration;
 
-use crate::intent_algebra::legacy_expr::{AggFunc, BinaryOpKind, LiteralValue, QueryExpr, ScalarExpr};
+use crate::intent_algebra::legacy_expr::{AggFunc, agg_is_exact, agg_is_mergeable, agg_quantiles, BinaryOpKind, LiteralValue, QueryExpr, ScalarExpr};
 use crate::pipeline::format_duration;
-use crate::intent_algebra::legacy_expr::{AggIntent, ExactAgg};
+use crate::intent_algebra::legacy_expr::AggIntent;
 use crate::physical::sketch_catalog;
 use crate::types::{
     AgentSubPlan, BackendSubPlan,
@@ -227,46 +227,47 @@ fn walk(expr: &QueryExpr, plan: &mut StagedPlan, budgets: &StageResourceBudgets)
 // ── Agg assignment ────────────────────────────────────────────────────────────
 
 fn assign_sketch_agg(op: &AggIntent, plan: &mut StagedPlan, budgets: &StageResourceBudgets) {
-    match op {
-        // Exact ops: mergeability decides stage.
-        AggIntent::Exact(ExactAgg::Sum | ExactAgg::Count | ExactAgg::Min | ExactAgg::Max) => {
+    // Exact ops: mergeability decides stage. Canonical mergeable exact
+    // intents (Sum / Count / Min / Max) → Backend; non-mergeable (Avg) → Db.
+    if agg_is_exact(op) {
+        if agg_is_mergeable(op) {
             plan.backend.has_merge = true;
-        }
-        AggIntent::Exact(ExactAgg::Avg) => {
+        } else {
             plan.db.active = true;
         }
-        // Sketch ops: resolve to physical, assign to Agent, defer if budget exceeded.
-        sketch_op => {
-            let physical = crate::physical::planner::resolve(sketch_op);
-            let stage = resolve_sketch_stage(physical.estimated_memory_bytes, budgets, &mut plan.deferral_log, sketch_op);
-            match stage {
-                SketchStage::Agent => {
-                    plan.agent.sketch_type   = Some(physical.sketch_type);
-                    plan.agent.sketch_params = physical.sketch_params;
-                }
-                SketchStage::Backend => {
-                    plan.backend.has_merge = true;
-                }
-                SketchStage::Precompute => {
-                    plan.precompute.active = true;
-                }
-            }
+        return;
+    }
+
+    // Sketch ops: resolve to physical, assign to Agent, defer if budget exceeded.
+    let physical = crate::physical::planner::resolve(op);
+    let stage = resolve_sketch_stage(physical.estimated_memory_bytes, budgets, &mut plan.deferral_log, op);
+    match stage {
+        SketchStage::Agent => {
+            plan.agent.sketch_type   = Some(physical.sketch_type);
+            plan.agent.sketch_params = physical.sketch_params;
+        }
+        SketchStage::Backend => {
+            plan.backend.has_merge = true;
+        }
+        SketchStage::Precompute => {
+            plan.precompute.active = true;
         }
     }
 }
 
 fn assign_agg_func(func: &AggFunc, plan: &mut StagedPlan, budgets: &StageResourceBudgets) {
+    use crate::intent_algebra::legacy_expr::{default_cardinality, default_frequency, default_quantile};
     match func {
         // Sketchable → synthesise the corresponding AggIntent and use existing logic.
         AggFunc::Quantile(phi) => {
-            let op = AggIntent::default_quantile(vec![*phi]);
+            let op = default_quantile(*phi);
             assign_sketch_agg(&op, plan, budgets);
         }
         AggFunc::CountDistinct => {
-            assign_sketch_agg(&AggIntent::default_cardinality(), plan, budgets);
+            assign_sketch_agg(&default_cardinality(), plan, budgets);
         }
         AggFunc::HeavyHitters { .. } => {
-            assign_sketch_agg(&AggIntent::default_frequency(), plan, budgets);
+            assign_sketch_agg(&default_frequency(), plan, budgets);
         }
         // Mergeable exact → Backend.
         AggFunc::Count | AggFunc::Sum | AggFunc::Min | AggFunc::Max
@@ -503,10 +504,13 @@ fn by_clause(keys: &[String]) -> String {
 }
 
 fn sketch_op_to_promql(op: &AggIntent, selector: &str, window: &str, by: &str) -> String {
+    // Canonical Quantile is single-φ post Step α; multi-φ fan-out happens
+    // at construction time (Merge of SketchAgg siblings) so the
+    // `quantile_over_time(qs, …)` arm collapses to one φ.
+    let _ = agg_quantiles; // imported for symmetry — unused after fan-out
     match op {
-        AggIntent::Quantile { quantiles, .. } => {
-            let phi = quantiles.first().copied().unwrap_or(0.99);
-            format!("quantile_over_time({phi}, {selector}{window}){by}")
+        AggIntent::Quantile { q, .. } => {
+            format!("quantile_over_time({q}, {selector}{window}){by}")
         }
         AggIntent::Cardinality { .. } => {
             format!("count_over_time({selector}{window}){by}")
@@ -514,17 +518,18 @@ fn sketch_op_to_promql(op: &AggIntent, selector: &str, window: &str, by: &str) -
         AggIntent::Frequency { .. } => {
             format!("count_over_time({selector}{window}){by}")
         }
-        AggIntent::Extrema { min, max } => match (min, max) {
-            (true, false) => format!("min_over_time({selector}{window}){by}"),
-            (false, true) => format!("max_over_time({selector}{window}){by}"),
-            _             => format!("quantile_over_time(0.5, {selector}{window}){by}"),
-        },
-        AggIntent::Exact(ExactAgg::Count) => format!("count_over_time({selector}{window}){by}"),
-        AggIntent::Exact(ExactAgg::Sum)   => format!("sum_over_time({selector}{window}){by}"),
-        AggIntent::Exact(ExactAgg::Avg)   => format!("avg_over_time({selector}{window}){by}"),
-        AggIntent::Exact(ExactAgg::Min)   => format!("min_over_time({selector}{window}){by}"),
-        AggIntent::Exact(ExactAgg::Max)   => format!("max_over_time({selector}{window}){by}"),
-        AggIntent::PerPartition { inner, .. } => sketch_op_to_promql(inner, selector, window, by),
+        AggIntent::Count { .. } => format!("count_over_time({selector}{window}){by}"),
+        AggIntent::Sum   => format!("sum_over_time({selector}{window}){by}"),
+        AggIntent::Avg   => format!("avg_over_time({selector}{window}){by}"),
+        AggIntent::Min   => format!("min_over_time({selector}{window}){by}"),
+        AggIntent::Max   => format!("max_over_time({selector}{window}){by}"),
+        AggIntent::Rate { .. }     => format!("rate({selector}{window}){by}"),
+        AggIntent::Increase { .. } => format!("increase({selector}{window}){by}"),
+        // TopK + archive-only intents (Phase β): Step γ routes these via
+        // canonical templates; today they reuse `count_over_time` because
+        // the legacy `Exact(_)` fall-through did effectively the same for
+        // any non-mergeable case. Documented as a Step γ TODO.
+        _ => format!("count_over_time({selector}{window}){by}"),
     }
 }
 
@@ -725,7 +730,7 @@ mod tests {
             duration: Duration::from_secs(300),
             slide: None,
             input: Box::new(QueryExpr::SketchAgg {
-                op:    AggIntent::default_quantile(vec![0.99]),
+                op:    crate::intent_algebra::legacy_expr::default_quantile(0.99),
                 col:   ColumnRef::SampleValue,
                 input: Box::new(source("latency")),
             }),
@@ -740,7 +745,7 @@ mod tests {
     #[test]
     fn hll_stays_at_agent_by_default() {
         let expr = QueryExpr::SketchAgg {
-            op:    AggIntent::default_cardinality(),
+            op:    crate::intent_algebra::legacy_expr::default_cardinality(),
             col:   ColumnRef::SampleValue,
             input: Box::new(source("events")),
         };
@@ -753,7 +758,7 @@ mod tests {
         let expr = QueryExpr::Partition {
             keys: PartitionKeys::By(vec!["host".into(), "region".into()]),
             input: Box::new(QueryExpr::SketchAgg {
-                op:    AggIntent::default_quantile(vec![0.99]),
+                op:    crate::intent_algebra::legacy_expr::default_quantile(0.99),
                 col:   ColumnRef::SampleValue,
                 input: Box::new(source("latency")),
             }),
@@ -805,7 +810,7 @@ mod tests {
             k:     10,
             by:    vec!["symbol".into()],
             input: Box::new(QueryExpr::SketchAgg {
-                op:    AggIntent::default_frequency(),
+                op:    crate::intent_algebra::legacy_expr::default_frequency(),
                 col:   ColumnRef::SampleValue,
                 input: Box::new(source("price")),
             }),
@@ -854,7 +859,7 @@ mod tests {
             ..Default::default()
         };
         let expr = QueryExpr::SketchAgg {
-            op:    AggIntent::default_quantile(vec![0.99]),
+            op:    crate::intent_algebra::legacy_expr::default_quantile(0.99),
             col:   ColumnRef::SampleValue,
             input: Box::new(source("latency")),
         };
@@ -872,7 +877,7 @@ mod tests {
             ..Default::default()
         };
         let expr = QueryExpr::SketchAgg {
-            op:    AggIntent::default_quantile(vec![0.99]),
+            op:    crate::intent_algebra::legacy_expr::default_quantile(0.99),
             col:   ColumnRef::SampleValue,
             input: Box::new(source("latency")),
         };
@@ -891,7 +896,7 @@ mod tests {
                 duration: Duration::from_secs(300),
                 slide: None,
                 input: Box::new(QueryExpr::SketchAgg {
-                    op:    AggIntent::default_quantile(vec![0.99]),
+                    op:    crate::intent_algebra::legacy_expr::default_quantile(0.99),
                     col:   ColumnRef::SampleValue,
                     input: Box::new(QueryExpr::Filter {
                         pred: ScalarExpr::BinaryOp {
@@ -917,7 +922,7 @@ mod tests {
             k:     10,
             by:    vec![],
             input: Box::new(QueryExpr::SketchAgg {
-                op:    AggIntent::default_frequency(),
+                op:    crate::intent_algebra::legacy_expr::default_frequency(),
                 col:   ColumnRef::SampleValue,
                 input: Box::new(source("events")),
             }),
diff --git a/controller/src/query_parser/mod.rs b/controller/src/query_parser/mod.rs
index 94903dd3..cf950742 100644
--- a/controller/src/query_parser/mod.rs
+++ b/controller/src/query_parser/mod.rs
@@ -295,6 +295,12 @@ impl QeCollector {
     }
 
     fn collect_op(&mut self, op: &AggIntent) {
+        // Canonical Quantile is single-φ post Step α (multi-φ legacy
+        // intents fan out into sibling SketchAggs at construction time —
+        // each one routes through this collector independently). The
+        // legacy `Extrema { min, max }` enum split into `Min` / `Max`
+        // canonical variants; map each to the corresponding boundary
+        // quantile for legacy compat.
         match op {
             AggIntent::Cardinality { .. } => {
                 if !self.agg_types.contains(&AggType::Cardinality) {
@@ -306,24 +312,28 @@ impl QeCollector {
                     self.agg_types.push(AggType::Frequency);
                 }
             }
-            AggIntent::Quantile { quantiles, .. } => {
+            AggIntent::Quantile { q, .. } => {
                 if !self.agg_types.contains(&AggType::Quantile) {
                     self.agg_types.push(AggType::Quantile);
                 }
-                for &q in quantiles {
-                    if !self.quantiles.contains(&q) { self.quantiles.push(q); }
+                if !self.quantiles.contains(q) { self.quantiles.push(*q); }
+            }
+            AggIntent::Min => {
+                if !self.agg_types.contains(&AggType::Quantile) {
+                    self.agg_types.push(AggType::Quantile);
                 }
+                if !self.quantiles.contains(&0.0) { self.quantiles.push(0.0); }
             }
-            AggIntent::Extrema { min, max } => {
+            AggIntent::Max => {
                 if !self.agg_types.contains(&AggType::Quantile) {
                     self.agg_types.push(AggType::Quantile);
                 }
-                // Extrema map to boundary quantiles for legacy compat.
-                if *min && !self.quantiles.contains(&0.0) { self.quantiles.push(0.0); }
-                if *max && !self.quantiles.contains(&1.0) { self.quantiles.push(1.0); }
+                if !self.quantiles.contains(&1.0) { self.quantiles.push(1.0); }
             }
-            AggIntent::Exact(_) => { self.exact_required = true; }
-            AggIntent::PerPartition { inner, .. } => self.collect_op(inner),
+            // Sum / Count / Avg / TopK / Rate / Increase / archive-only —
+            // all flip the exact_required flag (no sketch benefit at the
+            // legacy planner's level).
+            _ => { self.exact_required = true; }
         }
     }