diff --git a/.cargo/config.toml b/.cargo/config.toml
new file mode 100644
index 00000000..c91c3f38
--- /dev/null
+++ b/.cargo/config.toml
@@ -0,0 +1,2 @@
+[net]
+git-fetch-with-cli = true
diff --git a/Cargo.lock b/Cargo.lock
index c7104816..39683b5f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -340,6 +340,16 @@ dependencies = [
  "regex-syntax",
 ]
 
+[[package]]
+name = "asap-ir"
+version = "0.1.0"
+source = "git+https://github.com/ProjectASAP/ASAPController?rev=c73c2a5099e3e0d0439e86d895f5a61a7e1231b1#c73c2a5099e3e0d0439e86d895f5a61a7e1231b1"
+dependencies = [
+ "serde",
+ "serde_json",
+ "thiserror 2.0.18",
+]
+
 [[package]]
 name = "asap-precompute-rs"
 version = "0.1.0"
@@ -751,6 +761,7 @@ name = "control_plane"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "asap-ir",
  "asap_types",
  "axum",
  "bytes",
diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml
index 24916bd5..32eae76d 100644
--- a/control_plane/Cargo.toml
+++ b/control_plane/Cargo.toml
@@ -35,6 +35,12 @@ tokio-stream   = { version = "0.1", features = ["net"] }
 promql_utilities.workspace = true
 asap_types.workspace = true
 
+# Transitional pin (docs/migration-plan-backend-plan.md Phase 1b): locked to a
+# specific commit on ASAPController's main, not a tag -- ASAPController has no
+# tagged releases yet. Re-pin as ASAPController's IR evolves; move to a tag
+# once one exists.
+asap-ir = { git = "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/ProjectASAP/ASAPController", rev = "c73c2a5099e3e0d0439e86d895f5a61a7e1231b1" }
+
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }
 tower = { version = "0.4", features = ["util"] }
diff --git a/control_plane/src/asap_tier_analysis.rs b/control_plane/src/asap_tier_analysis.rs
index df2e861e..3bc2a064 100644
--- a/control_plane/src/asap_tier_analysis.rs
+++ b/control_plane/src/asap_tier_analysis.rs
@@ -353,28 +353,52 @@ fn collect_agg_intents(expr: &QueryExpr, out: &mut Vec<AggIntent>) {
 /// the AST walker can recover them; this fallback runs when the AST
 /// walk fails.
 fn intent_kind_label(intent: &AggIntent) -> &'static str {
+    if crate::intent_algebra::as_frequency(intent).is_some() {
+        return "frequency";
+    }
     match intent {
         AggIntent::Count { .. } => "count",
-        AggIntent::Sum => "sum",
-        AggIntent::Min => "min",
-        AggIntent::Max => "max",
-        AggIntent::Avg => "avg",
+        AggIntent::Sum { .. } => "sum",
+        AggIntent::Min { .. } => "min",
+        AggIntent::Max { .. } => "max",
+        AggIntent::Avg { .. } => "avg",
+        AggIntent::StdDev { .. } => "stddev",
+        AggIntent::Variance { .. } => "variance",
         AggIntent::Quantile { .. } => "quantile",
         AggIntent::TopK { .. } => "topk",
         AggIntent::Cardinality { .. } => "cardinality",
-        AggIntent::Frequency { .. } => "frequency",
-        AggIntent::Rate { .. } => "rate",
-        AggIntent::Increase { .. } => "increase",
+        AggIntent::Rate => "rate",
+        AggIntent::Increase => "increase",
         AggIntent::Absent => "absent",
-        AggIntent::Present => "present",
-        AggIntent::Delta { .. } => "delta",
-        AggIntent::Deriv { .. } => "deriv",
+        AggIntent::AbsentOverTime => "absent_over_time",
+        AggIntent::PresentOverTime => "present_over_time",
+        AggIntent::Delta => "delta",
+        AggIntent::Deriv => "deriv",
         AggIntent::PredictLinear { .. } => "predict_linear",
-        AggIntent::HoltWinters { .. } => "holt_winters",
-        AggIntent::Idelta { .. } => "idelta",
-        AggIntent::Irate { .. } => "irate",
-        AggIntent::Resets { .. } => "resets",
-        AggIntent::Changes { .. } => "changes",
+        AggIntent::DoubleExpSmoothing { .. } => "double_exponential_smoothing",
+        AggIntent::IDelta => "idelta",
+        AggIntent::Resets => "resets",
+        AggIntent::Changes => "changes",
+        AggIntent::HistogramCount => "histogram_count",
+        AggIntent::HistogramSum => "histogram_sum",
+        AggIntent::HistogramAvg => "histogram_avg",
+        AggIntent::HistogramStdDev => "histogram_stddev",
+        AggIntent::HistogramStdVar => "histogram_stdvar",
+        AggIntent::HistogramFraction { .. } => "histogram_fraction",
+        AggIntent::HistogramQuantile { .. } => "histogram_quantile",
+        AggIntent::Math(_) => "math",
+        AggIntent::TimeFn(_) => "time_fn",
+        AggIntent::Group => "group",
+        AggIntent::CountValues { .. } => "count_values",
+        AggIntent::LastOverTime => "last_over_time",
+        AggIntent::FirstOverTime => "first_over_time",
+        AggIntent::MadOverTime => "mad_over_time",
+        AggIntent::TsOfMinOverTime => "ts_of_min_over_time",
+        AggIntent::TsOfMaxOverTime => "ts_of_max_over_time",
+        AggIntent::TsOfFirstOverTime => "ts_of_first_over_time",
+        AggIntent::TsOfLastOverTime => "ts_of_last_over_time",
+        // Unrecognized Extension (not the Frequency one, guarded above).
+        AggIntent::Extension { .. } => "extension",
     }
 }
 
diff --git a/control_plane/src/intent_algebra/agg_intent.rs b/control_plane/src/intent_algebra/agg_intent.rs
index 81c55fef..d84c145e 100644
--- a/control_plane/src/intent_algebra/agg_intent.rs
+++ b/control_plane/src/intent_algebra/agg_intent.rs
@@ -1,638 +1,200 @@
 //! Layer 3 aggregation-intent vocabulary.
 //!
-//! Per `control_plane/docs/design.md` §6 "`AggIntent` — what to compute, not
-//! how" (around line ~468). L3 carries intent ("compute a quantile to
-//! ε=0.01 accuracy"). The choice between `HashAgg` / `SortAgg` /
-//! `SketchAgg(KLL{k=200})` is made by L4 cost-aware rules, not encoded
-//! here.
+//! ## Phase 1b (docs/migration-plan-backend-plan.md)
 //!
-//! Intent vs operator distinction. `AggIntent::TopK` is an *intent* (a
-//! dedicated heavy-hitter sketch primitive — SpaceSaving, CMS-with-heap
-//! — computes it in a single pass). The generic `Sort + Limit` operator
-//! pair survives in `QueryExpr` for non-heavy-hitter cases (`ORDER BY
-//! name LIMIT 10`). L1→L2→L3 lowering picks one or the other
-//! deterministically.
+//! `AggIntent` is no longer defined in this repo. This file re-exports the
+//! canonical type from ASAPController's `asap-ir` crate (a git dependency
+//! in `control_plane/Cargo.toml`, currently pinned to a commit SHA —
+//! ASAPController has no tagged releases yet) and holds only what's
+//! genuinely control_plane-specific:
 //!
-//! No `QuantileOverTime` intent. The window is fully captured by the
-//! surrounding `QueryExpr::Window` node; the quantile *operation* is the
-//! same regardless. PromQL `quantile_over_time(0.99, m[5m])` lowers to
-//! `Window{size=5m} → Aggregate{aggs:[Quantile{q=0.99}]}`.
+//! - **`frequency()` / `as_frequency()`** — control_plane's standalone
+//!   point-frequency-via-CMS query (`count(*) WHERE key = k`), carried
+//!   through the shared `AggIntent` as an `Extension` rather than a
+//!   first-class shared variant. This is **not** the same capability as
+//!   `RankingMeasure::Frequency` (which classifies what a `TopK` ranks
+//!   by) — see `ASAPController#137` for why `Extension` exists and why
+//!   this isn't folded into that. `Extension` is a generic escape hatch
+//!   (issue `ASAPController#131`): a deployment-model-specific intent
+//!   core doesn't know the shape of, tagged by an `ext_kind` string, with
+//!   `payload` opaque to core.
+//! - **`archive_only()` / `output_column()`** — free functions, not
+//!   methods. Rust's orphan rules don't allow an inherent `impl AggIntent`
+//!   block from a crate that doesn't own the type, so these can't be
+//!   `.archive_only()`/`.output_column()` call syntax anymore (that
+//!   syntax survives for `asap_ir`'s own inherent methods, e.g.
+//!   `intent.input_col()`, `intent.is_per_series()` — those aren't
+//!   control_plane-specific and need no wrapper).
 //!
-//! `Rate` and `Increase` survive that argument because they include
-//! PromQL's counter-reset adjustment, a non-trivial transformation that
-//! exact `Sum` does not perform.
-
-#![allow(dead_code)]
-
-use std::time::Duration;
-
-use serde::{Deserialize, Serialize};
+//! ## What moved, what didn't
+//!
+//! Per Phase 0's tie-break rule, `asap_ir::AggIntent`'s shape wins
+//! wherever it differs from this repo's pre-merge version. One
+//! consequence: `Rate` / `Increase` / `Changes` / `Delta` / `IDelta` /
+//! `Deriv` / `Resets` no longer carry `window: Duration` — the window
+//! lives on the enclosing `QueryExpr::Window` node instead (ASAPController's
+//! design). Phase 1 (the copy-based first attempt, `#391`) deferred this;
+//! depending on the real external type instead of a local copy makes it
+//! unavoidable — you can't add a field to a variant you don't own. Grep
+//! for `.window` reads (not constructions) before assuming a call site
+//! needs updating: as of this migration there was exactly one real
+//! consumer (`sketch_algebra/rules/bind_exact_agg.rs`), which now reads
+//! the window off the enclosing `QueryExpr::Window` node it already has
+//! in scope, not off the intent.
+
+use asap_ir::intent_algebra::agg_accuracy as asap_agg_accuracy;
+use asap_ir::intent_algebra::schema::{Column as AsapColumn, DataType as AsapDataType};
+pub use asap_ir::intent_algebra::{
+    agg_is_exact, agg_is_mergeable, default_cardinality, default_quantile,
+    is_frequency_heavy_hitter, ranking_measure, AggIntent, MathFunc, RankingMeasure, TimeFunc,
+};
 
 use crate::intent_algebra::schema::{Column, DataType};
 use crate::types_v2::AccuracyTarget;
 
-/// "What to compute" at L3 — vocabulary the planner pivots on. See module
-/// doc for the intent vs operator distinction.
-///
-/// Variants intentionally mirror `design.md` §6 line ~468; data-model-
-/// agnostic intents come first, time-series-streaming derivatives
-/// (`Rate` / `Increase`) come last.
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-#[serde(tag = "kind", rename_all = "snake_case")]
-pub enum AggIntent {
-    // ── Data-model-agnostic ──────────────────────────────────────────────
-    /// COUNT(*) / `count` — number of rows / samples per group.
-    /// `accuracy: Exact` selects an exact counter; `Epsilon` / `EpsilonDelta`
-    /// unlock CMS / linear-counting sketch families.
-    Count { accuracy: AccuracyTarget },
-    /// SUM(col). Always exact at L3 — no approximation intent for `Sum`
-    /// in the catalog (`design.md` §6 line ~485).
-    Sum,
-    /// Per-group minimum — exact at L3.
-    Min,
-    /// Per-group maximum — exact at L3.
-    Max,
-    /// Arithmetic mean. Exact at L3; sketch backends fold this onto a
-    /// `Quantile{q=0.5}` only when the cost model allows the relaxation.
-    Avg,
-    /// Compute the φ-th quantile (0 ≤ q ≤ 1) to the given accuracy.
-    /// Sketch families: KLL, DDSketch, t-digest.
-    Quantile { q: f64, accuracy: AccuracyTarget },
-    /// Heavy-hitter top-k. Distinct from generic `Sort + Limit` because
-    /// a dedicated sketch primitive (SpaceSaving, CMS-with-heap,
-    /// Misra-Gries) computes it as a single operation. L1→L2→L3 lowering
-    /// produces this when it recognises `topk(k, …)` (PromQL) or
-    /// `ORDER BY count DESC LIMIT k` (SQL).
-    TopK { k: usize, accuracy: AccuracyTarget },
-    /// COUNT DISTINCT — number of distinct values in the input column,
-    /// to the given accuracy. Sketch families: HLL, theta-sketch.
-    Cardinality { accuracy: AccuracyTarget },
-    /// Frequency of a key in the input — `count(*) WHERE key = k` modeled
-    /// as a sketch query. Sketch families: CMS, count-min-log.
-    Frequency { accuracy: AccuracyTarget },
-
-    // ── Time-series streaming derivatives ────────────────────────────────
-    /// Per-second average derivative with PromQL's counter-reset
-    /// adjustment. Specialized — exact `Sum / Count over Window` does
-    /// NOT serve this intent.
-    Rate { window: Duration },
-    /// Cumulative increase over the given window with counter-reset
-    /// adjustment. Specialized — see `Rate` above.
-    Increase { window: Duration },
-
-    // ── Archive-only intents (Phase β migration) ─────────────────────────
-    // Intents below have no ASAP-tier sketch family today; the L4 binder
-    // emits a `PhysicalExpr::Logical` pass-through and the L5 emitter routes
-    // them to the cold archive tier (Gorilla / Thanos). Adding a streaming
-    // sketch family for any of these is a follow-up — the L3 vocabulary
-    // captures the intent so the routing decision is layered above intent.
-    //
-    // Note: `histogram_quantile(φ, …)` is NOT an L3 intent — it's a PromQL
-    // /MetricsQL language-level operator. Per Step γ5 of the relational
-    // migration, the PromQL parser substitutes it directly into a plain
-    // `Aggregate { Quantile(φ) }` (which lowers to
-    // `AggIntent::Quantile { q, accuracy }`); bucket-aware handling is a
-    // physical-planner concern, not an L3 intent.
-    //
-    /// `absent(vector_selector)` — 1 iff the selector matched no series in
-    /// the evaluation window, no value otherwise. Routed to archive: the
-    /// engine answers it directly off the index.
-    Absent,
-    /// `present_over_time(m[range])` — 1 iff the selector had at least one
-    /// sample in the window. Inverse of `Absent`. Archive-routed.
-    Present,
-    /// `delta(m[range])` — last − first sample within the window, NO
-    /// counter-reset adjustment. Distinct from [`AggIntent::Increase`].
-    Delta { window: Duration },
-    /// `deriv(m[range])` — per-second derivative via simple linear
-    /// regression. Archive-routed (no streaming sketch).
-    Deriv { window: Duration },
-    /// `predict_linear(m[range], t)` — linear-regression prediction `t`
-    /// seconds into the future. Archive-routed.
-    PredictLinear { window: Duration, ahead: Duration },
-    /// `holt_winters(m[range], sf, tf)` — exponential-smoothing forecast.
-    /// Archive-routed.
-    HoltWinters {
-        window: Duration,
-        smoothing_factor: f64,
-        trend_factor: f64,
-    },
-    /// `idelta(m[range])` — `last − second_to_last`, instant delta. No
-    /// streaming sketch.
-    Idelta { window: Duration },
-    /// `irate(m[range])` — instant per-second rate computed from the last
-    /// two samples. Counter-reset adjusted but evaluated point-wise; not
-    /// the same as the streaming [`AggIntent::Rate`].
-    Irate { window: Duration },
-    /// `resets(m[range])` — count of counter resets over the window.
-    /// Archive-routed.
-    Resets { window: Duration },
-    /// `changes(m[range])` — count of value changes over the window.
-    /// Archive-routed.
-    Changes { window: Duration },
-}
-
-impl AggIntent {
-    /// True iff this intent has no ASAP-tier (streaming sketch) binding
-    /// today. `false` means a `Bind*` rule may match. `true` means the
-    /// L5 emitter routes the intent to the cold-store / archive tier.
-    ///
-    /// Per Phase β orchestrator spec, the PromQL functions that were
-    /// previously refused outright by `asap-planner-rs::single_query::
-    /// is_supported()` (everything outside the 5 patterns) all return
-    /// `true` here. Adding a sketch family for any of them is a future
-    /// PR — flipping the flag to `false` is the single point of change.
-    ///
-    /// Note: `histogram_quantile(...)` was previously listed as
-    /// archive-only here but is no longer an `AggIntent` variant —
-    /// it's a PromQL/MetricsQL language-level operator. Per Step γ5,
-    /// the PromQL parser substitutes it into a plain
-    /// `Aggregate { Quantile(φ) }`, which lowers to
-    /// `AggIntent::Quantile { q, .. }`.
-    pub fn archive_only(&self) -> bool {
-        matches!(
-            self,
-            AggIntent::Absent
-                | AggIntent::Present
-                | AggIntent::Delta { .. }
-                | AggIntent::Deriv { .. }
-                | AggIntent::PredictLinear { .. }
-                | AggIntent::HoltWinters { .. }
-                | AggIntent::Idelta { .. }
-                | AggIntent::Irate { .. }
-                | AggIntent::Resets { .. }
-                | AggIntent::Changes { .. }
-        )
+const FREQUENCY_EXT_KIND: &str = "frequency";
+
+/// `control_plane::Column`/`DataType` and `asap_ir::Column`/`DataType`
+/// are structurally identical but not the same type — merging `schema.rs`
+/// itself is Phase 2 scope (it cascades into `Schema`/`QueryExpr`, used
+/// pervasively; ~38 `Column{}` literals across this repo). Convert at
+/// this boundary instead of widening this change.
+fn to_asap_dtype(dt: &DataType) -> AsapDataType {
+    match dt {
+        DataType::Int64 => AsapDataType::Int64,
+        DataType::Float64 => AsapDataType::Float64,
+        DataType::Utf8 => AsapDataType::Utf8,
+        DataType::Bool => AsapDataType::Bool,
+        DataType::Timestamp => AsapDataType::Timestamp,
     }
 }
 
-impl AggIntent {
-    /// Output column name + type produced by this intent when applied to
-    /// `input`. Used by `QueryExpr::Aggregate`'s schema-derivation rule
-    /// (`design.md` §6 schema-flow table: "one new column per entry in
-    /// `aggs`, each named and typed by `AggIntent::output_type(input_field)`").
-    ///
-    /// PromQL convention: aggregate column name = intent kind (`count`,
-    /// `quantile_0_99`, …) so consumers can locate it without an alias
-    /// lookup.
-    pub fn output_column(&self, input: &Column) -> Column {
-        match self {
-            AggIntent::Count { .. } => Column {
-                name: "count".into(),
-                dtype: DataType::Int64,
-                nullable: false,
-            },
-            AggIntent::Sum => Column {
-                name: "sum".into(),
-                dtype: input.dtype.clone(),
-                nullable: false,
-            },
-            AggIntent::Min => Column {
-                name: "min".into(),
-                dtype: input.dtype.clone(),
-                nullable: input.nullable,
-            },
-            AggIntent::Max => Column {
-                name: "max".into(),
-                dtype: input.dtype.clone(),
-                nullable: input.nullable,
-            },
-            AggIntent::Avg => Column {
-                name: "avg".into(),
-                dtype: DataType::Float64,
-                nullable: false,
-            },
-            AggIntent::Quantile { q, .. } => Column {
-                name: format!("quantile_{}", quantile_suffix(*q)),
-                dtype: DataType::Float64,
-                nullable: false,
-            },
-            AggIntent::TopK { k, .. } => Column {
-                name: format!("topk_{k}"),
-                // TopK output is a struct/list per row; modeled as Utf8
-                // for L3 (the L4 sketch-bound IR upgrades the dtype).
-                dtype: DataType::Utf8,
-                nullable: false,
-            },
-            AggIntent::Cardinality { .. } => Column {
-                name: "cardinality".into(),
-                dtype: DataType::Int64,
-                nullable: false,
-            },
-            AggIntent::Frequency { .. } => Column {
-                name: "frequency".into(),
-                dtype: DataType::Int64,
-                nullable: false,
-            },
-            AggIntent::Rate { .. } => Column {
-                name: "rate".into(),
-                dtype: DataType::Float64,
-                nullable: false,
-            },
-            AggIntent::Increase { .. } => Column {
-                name: "increase".into(),
-                dtype: DataType::Float64,
-                nullable: false,
-            },
-            // ── Archive-only intents (Phase β) ────────────────────────────
-            // Each carries a stable column name keyed on the intent kind so
-            // the StreamingConfig emitter and Phase α routing entry can
-            // locate them. All are Float64 except the boolean Absent /
-            // Present, which surface as Int64 (1 / 0) per PromQL convention.
-            AggIntent::Absent => Column {
-                name: "absent".into(),
-                dtype: DataType::Int64,
-                nullable: false,
-            },
-            AggIntent::Present => Column {
-                name: "present".into(),
-                dtype: DataType::Int64,
-                nullable: false,
-            },
-            AggIntent::Delta { .. } => Column {
-                name: "delta".into(),
-                dtype: DataType::Float64,
-                nullable: false,
-            },
-            AggIntent::Deriv { .. } => Column {
-                name: "deriv".into(),
-                dtype: DataType::Float64,
-                nullable: false,
-            },
-            AggIntent::PredictLinear { .. } => Column {
-                name: "predict_linear".into(),
-                dtype: DataType::Float64,
-                nullable: false,
-            },
-            AggIntent::HoltWinters { .. } => Column {
-                name: "holt_winters".into(),
-                dtype: DataType::Float64,
-                nullable: false,
-            },
-            AggIntent::Idelta { .. } => Column {
-                name: "idelta".into(),
-                dtype: DataType::Float64,
-                nullable: false,
-            },
-            AggIntent::Irate { .. } => Column {
-                name: "irate".into(),
-                dtype: DataType::Float64,
-                nullable: false,
-            },
-            AggIntent::Resets { .. } => Column {
-                name: "resets".into(),
-                dtype: DataType::Int64,
-                nullable: false,
-            },
-            AggIntent::Changes { .. } => Column {
-                name: "changes".into(),
-                dtype: DataType::Int64,
-                nullable: false,
-            },
-        }
+fn from_asap_dtype(dt: &AsapDataType) -> DataType {
+    match dt {
+        AsapDataType::Int64 => DataType::Int64,
+        AsapDataType::Float64 => DataType::Float64,
+        AsapDataType::Utf8 => DataType::Utf8,
+        AsapDataType::Bool => DataType::Bool,
+        AsapDataType::Timestamp => DataType::Timestamp,
     }
 }
 
-/// `0.99` → `"0_99"`, `0.5` → `"0_5"`. Used by `Quantile` output naming
-/// so `quantile_0_99` is a valid identifier downstream.
-fn quantile_suffix(q: f64) -> String {
-    let mut s = format!("{q}");
-    if let Some(stripped) = s.strip_prefix('-') {
-        s = format!("neg_{stripped}");
-    }
-    s.replace('.', "_")
+fn to_asap_column(c: &Column) -> AsapColumn {
+    AsapColumn::new(c.name.clone(), to_asap_dtype(&c.dtype), c.nullable)
 }
 
-// ── AggIntent helpers ────────────────────────────────────────────────────────
-//
-// Step γ7: relocated from `relational.rs` (where they were free fns
-// operating on the canonical re-exported `AggIntent`). `relational`
-// re-exports them during the legacy-IR retirement; consumers migrate to
-// `intent_algebra::*` paths and the re-exports drop with `relational`.
-
-/// Two instances of this aggregation can be merged
-/// (`agg(A ∪ B) = combine(agg(A), agg(B))`). `Avg` is the only
-/// non-mergeable case (needs `(sum, count)`, not a single value).
-pub fn agg_is_mergeable(op: &AggIntent) -> bool {
-    !matches!(op, AggIntent::Avg)
-}
-
-/// Whether this op implies `exact_required` — no sketch benefit. The
-/// exact intents are `Sum / Count / Avg / Min / Max`.
-pub fn agg_is_exact(op: &AggIntent) -> bool {
-    matches!(
-        op,
-        AggIntent::Sum | AggIntent::Count { .. } | AggIntent::Avg | AggIntent::Min | AggIntent::Max
-    )
-}
-
-/// Accuracy parameter as a fractional ε (`0.0` for exact ops), unpacked
-/// from the typed `AccuracyTarget` on 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,
+fn from_asap_column(c: AsapColumn) -> Column {
+    Column {
+        name: c.name,
+        dtype: from_asap_dtype(&c.dtype),
+        nullable: c.nullable,
     }
 }
 
-fn accuracy_target_to_f64(t: &AccuracyTarget) -> f64 {
-    match t {
-        AccuracyTarget::Exact => 0.0,
-        AccuracyTarget::Epsilon(eps) | AccuracyTarget::EpsilonDelta { eps, .. } => *eps,
+/// Construct control_plane's point-frequency-via-CMS intent. See module
+/// docs for why this is an `Extension`, not a shared first-class variant.
+pub fn frequency(accuracy: AccuracyTarget) -> AggIntent {
+    AggIntent::Extension {
+        ext_kind: FREQUENCY_EXT_KIND.to_string(),
+        payload: serde_json::json!({ "accuracy": accuracy }),
     }
 }
 
-/// Default `Frequency` intent — `accuracy = e / 2000`.
+/// Default `Frequency` intent — `accuracy = e / 2000`. Unchanged default
+/// from before the merge.
 pub fn default_frequency() -> AggIntent {
-    AggIntent::Frequency {
-        accuracy: AccuracyTarget::Epsilon(std::f64::consts::E / 2000.0),
-    }
+    frequency(AccuracyTarget::Epsilon(std::f64::consts::E / 2000.0))
 }
 
-/// Default `Cardinality` intent — `accuracy = hll_accuracy(14)`.
-pub fn default_cardinality() -> AggIntent {
-    AggIntent::Cardinality {
-        accuracy: AccuracyTarget::Epsilon(crate::sketch_algebra::capability::hll_accuracy(14)),
+/// If `intent` is control_plane's `Frequency` extension, extract its
+/// accuracy target. `None` for every other intent, including other
+/// (currently hypothetical) `Extension` kinds.
+pub fn as_frequency(intent: &AggIntent) -> Option<AccuracyTarget> {
+    match intent {
+        AggIntent::Extension { ext_kind, payload } if ext_kind == FREQUENCY_EXT_KIND => {
+            serde_json::from_value(payload.get("accuracy")?.clone()).ok()
+        }
+        _ => None,
     }
 }
 
-/// Default `Quantile` intent at φ = `q`, `accuracy = ε 0.01`. Canonical
-/// `Quantile` is single-φ; multi-φ callers invoke this once per φ.
-pub fn default_quantile(q: f64) -> AggIntent {
-    AggIntent::Quantile {
-        q,
-        accuracy: AccuracyTarget::Epsilon(0.01),
-    }
+/// Accuracy parameter as a fractional ε (`0.0` for exact ops). Wraps
+/// `asap_ir::agg_accuracy`, which returns `0.0` for `Frequency` (it's
+/// opaque `Extension` payload to core) — special-cased here so callers
+/// don't need to know `Frequency` isn't a first-class shared variant.
+pub fn agg_accuracy(intent: &AggIntent) -> f64 {
+    if let Some(acc) = as_frequency(intent) {
+        return match acc {
+            AccuracyTarget::Exact => 0.0,
+            AccuracyTarget::Epsilon(eps) | AccuracyTarget::EpsilonDelta { epsilon: eps, .. } => eps,
+        };
+    }
+    asap_agg_accuracy(intent)
 }
 
-// ── Tests ─────────────────────────────────────────────────────────────────────
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::intent_algebra::schema::{Column, DataType};
-
-    fn col(name: &str, dtype: DataType) -> Column {
-        Column {
-            name: name.into(),
-            dtype,
-            nullable: false,
-        }
-    }
-
-    #[test]
-    fn agg_intent_serde_roundtrip() {
-        let cases = vec![
-            AggIntent::Count {
-                accuracy: AccuracyTarget::Exact,
-            },
-            AggIntent::Sum,
-            AggIntent::Min,
-            AggIntent::Max,
-            AggIntent::Avg,
-            AggIntent::Quantile {
-                q: 0.99,
-                accuracy: AccuracyTarget::Epsilon(0.01),
-            },
-            AggIntent::TopK {
-                k: 10,
-                accuracy: AccuracyTarget::Epsilon(0.05),
-            },
-            AggIntent::Cardinality {
-                accuracy: AccuracyTarget::EpsilonDelta {
-                    eps: 0.01,
-                    delta: 0.001,
-                },
-            },
-            AggIntent::Frequency {
-                accuracy: AccuracyTarget::Epsilon(0.01),
-            },
-            AggIntent::Rate {
-                window: Duration::from_secs(60),
-            },
-            AggIntent::Increase {
-                window: Duration::from_secs(300),
-            },
-        ];
-        for variant in cases {
-            let json = serde_json::to_string(&variant).unwrap();
-            let back: AggIntent = serde_json::from_str(&json).unwrap();
-            assert_eq!(variant, back, "round-trip failed for {variant:?}");
-        }
-    }
-
-    #[test]
-    fn output_column_names_are_intent_keyed() {
-        let v = col("value", DataType::Float64);
-        assert_eq!(
-            AggIntent::Count {
-                accuracy: AccuracyTarget::Exact,
-            }
-            .output_column(&v)
-            .name,
-            "count"
-        );
-        assert_eq!(AggIntent::Sum.output_column(&v).name, "sum");
-        assert_eq!(
-            AggIntent::Quantile {
-                q: 0.99,
-                accuracy: AccuracyTarget::Epsilon(0.01),
-            }
-            .output_column(&v)
-            .name,
-            "quantile_0_99"
-        );
-        assert_eq!(
-            AggIntent::TopK {
-                k: 5,
-                accuracy: AccuracyTarget::Exact,
-            }
-            .output_column(&v)
-            .name,
-            "topk_5"
-        );
-    }
-
-    #[test]
-    fn quantile_output_is_float64() {
-        let v = col("value", DataType::Int64);
-        let out = AggIntent::Quantile {
-            q: 0.5,
-            accuracy: AccuracyTarget::Epsilon(0.01),
-        }
-        .output_column(&v);
-        assert!(matches!(out.dtype, DataType::Float64));
-    }
-
-    #[test]
-    fn sum_preserves_input_dtype() {
-        let int_col = col("c", DataType::Int64);
-        let float_col = col("c", DataType::Float64);
-        assert!(matches!(
-            AggIntent::Sum.output_column(&int_col).dtype,
-            DataType::Int64
-        ));
-        assert!(matches!(
-            AggIntent::Sum.output_column(&float_col).dtype,
-            DataType::Float64
-        ));
-    }
-
-    // ── Phase β archive-only intent tests ─────────────────────────────────
-
-    /// Every intent the legacy `asap-planner-rs::single_query::is_supported`
-    /// previously refused now lifts to L3 with `archive_only() == true`.
-    /// The negative cases are the ASAP-tier-bound intents — they must
-    /// continue to return false, otherwise the L4 binder would short-circuit
-    /// them to the cold tier.
-    #[test]
-    fn archive_only_flag_partitions_intents() {
-        // Archive-only — every Phase β migration target.
-        let archive: Vec<AggIntent> = vec![
-            AggIntent::Absent,
-            AggIntent::Present,
-            AggIntent::Delta {
-                window: Duration::from_secs(60),
-            },
-            AggIntent::Deriv {
-                window: Duration::from_secs(60),
-            },
-            AggIntent::PredictLinear {
-                window: Duration::from_secs(300),
-                ahead: Duration::from_secs(60),
-            },
-            AggIntent::HoltWinters {
-                window: Duration::from_secs(300),
-                smoothing_factor: 0.3,
-                trend_factor: 0.3,
-            },
-            AggIntent::Idelta {
-                window: Duration::from_secs(60),
-            },
-            AggIntent::Irate {
-                window: Duration::from_secs(60),
-            },
-            AggIntent::Resets {
-                window: Duration::from_secs(300),
-            },
-            AggIntent::Changes {
-                window: Duration::from_secs(300),
-            },
-        ];
-        for v in archive {
-            assert!(
-                v.archive_only(),
-                "{v:?} should be archive-only after Phase β migration"
-            );
-        }
-
-        // Warm-tier — must NOT be flagged archive-only or the L4 binder
-        // breaks.
-        let warm: Vec<AggIntent> = vec![
-            AggIntent::Count {
-                accuracy: AccuracyTarget::Exact,
-            },
-            AggIntent::Sum,
-            AggIntent::Min,
-            AggIntent::Max,
-            AggIntent::Avg,
-            AggIntent::Quantile {
-                q: 0.99,
-                accuracy: AccuracyTarget::Epsilon(0.01),
-            },
-            AggIntent::TopK {
-                k: 10,
-                accuracy: AccuracyTarget::Epsilon(0.05),
-            },
-            AggIntent::Cardinality {
-                accuracy: AccuracyTarget::EpsilonDelta {
-                    eps: 0.01,
-                    delta: 0.001,
-                },
-            },
-            AggIntent::Frequency {
-                accuracy: AccuracyTarget::Epsilon(0.01),
-            },
-            AggIntent::Rate {
-                window: Duration::from_secs(60),
-            },
-            AggIntent::Increase {
-                window: Duration::from_secs(300),
-            },
-        ];
-        for v in warm {
-            assert!(
-                !v.archive_only(),
-                "{v:?} is ASAP-tier and must not be archive-only"
-            );
-        }
-    }
-
-    #[test]
-    fn archive_only_intent_serde_roundtrip() {
-        let cases = vec![
-            AggIntent::Absent,
-            AggIntent::Present,
-            AggIntent::Delta {
-                window: Duration::from_secs(60),
-            },
-            AggIntent::Deriv {
-                window: Duration::from_secs(60),
-            },
-            AggIntent::PredictLinear {
-                window: Duration::from_secs(300),
-                ahead: Duration::from_secs(60),
-            },
-            AggIntent::HoltWinters {
-                window: Duration::from_secs(300),
-                smoothing_factor: 0.3,
-                trend_factor: 0.3,
-            },
-            AggIntent::Idelta {
-                window: Duration::from_secs(60),
-            },
-            AggIntent::Irate {
-                window: Duration::from_secs(60),
-            },
-            AggIntent::Resets {
-                window: Duration::from_secs(300),
-            },
-            AggIntent::Changes {
-                window: Duration::from_secs(300),
-            },
-        ];
-        for v in cases {
-            let json = serde_json::to_string(&v).unwrap();
-            let back: AggIntent = serde_json::from_str(&json).unwrap();
-            assert_eq!(v, back, "round-trip failed for {v:?}");
-        }
+/// True iff this intent has no ASAP-tier (streaming sketch) binding
+/// today. `false` means a `Bind*` rule may match. `true` means the L5
+/// emitter routes the intent to the cold-store / archive tier.
+///
+/// Every intent `asap_ir::AggIntent` carries that this repo didn't have
+/// before the merge (histogram accessors, math/trig, time/calendar
+/// accessors, presence functions, `Group`/`CountValues`, the extended
+/// range-vector reducers) is archive-only — none has a `Bind*` rule yet.
+/// An unrecognized `Extension` (not control_plane's `Frequency`) is also
+/// archive-only by default — no binding exists for a shape core can't
+/// even see into.
+pub fn archive_only(intent: &AggIntent) -> bool {
+    if as_frequency(intent).is_some() {
+        return false; // real CMS binding — sketch_algebra/rules/bind_cms_count.rs
     }
+    matches!(
+        intent,
+        AggIntent::Absent
+            | AggIntent::AbsentOverTime
+            | AggIntent::PresentOverTime
+            | AggIntent::Delta
+            | AggIntent::Deriv
+            | AggIntent::PredictLinear { .. }
+            | AggIntent::DoubleExpSmoothing { .. }
+            | AggIntent::IDelta
+            | AggIntent::Resets
+            | AggIntent::Changes
+            | AggIntent::HistogramCount
+            | AggIntent::HistogramSum
+            | AggIntent::HistogramAvg
+            | AggIntent::HistogramStdDev
+            | AggIntent::HistogramStdVar
+            | AggIntent::HistogramFraction { .. }
+            | AggIntent::HistogramQuantile { .. }
+            | AggIntent::Math(_)
+            | AggIntent::TimeFn(_)
+            | AggIntent::Group
+            | AggIntent::CountValues { .. }
+            | AggIntent::LastOverTime
+            | AggIntent::FirstOverTime
+            | AggIntent::MadOverTime
+            | AggIntent::TsOfMinOverTime
+            | AggIntent::TsOfMaxOverTime
+            | AggIntent::TsOfFirstOverTime
+            | AggIntent::TsOfLastOverTime
+            | AggIntent::Extension { .. } // unrecognized extensions only reach here
+    )
+}
 
-    #[test]
-    fn archive_only_output_column_names() {
-        let v = col("value", DataType::Float64);
-        assert_eq!(AggIntent::Absent.output_column(&v).name, "absent");
-        assert_eq!(AggIntent::Present.output_column(&v).name, "present");
-        assert_eq!(
-            AggIntent::Delta {
-                window: Duration::from_secs(60)
-            }
-            .output_column(&v)
-            .name,
-            "delta"
-        );
-        assert_eq!(
-            AggIntent::Resets {
-                window: Duration::from_secs(60)
-            }
-            .output_column(&v)
-            .name,
-            "resets"
-        );
+/// Output column name + type produced by this intent when applied to
+/// `input`. Delegates to `asap_ir::AggIntent::output_column` (the
+/// inherent method on the shared type) for everything except
+/// control_plane's `Frequency` extension, which core's generic
+/// `Extension` handling can't name/type correctly (core has no idea
+/// `ext_kind == "frequency"` means Int64, non-nullable, named
+/// `"frequency"` — that's control_plane-only knowledge).
+pub fn output_column(intent: &AggIntent, input: &Column) -> Column {
+    if as_frequency(intent).is_some() {
+        return Column {
+            name: "frequency".into(),
+            dtype: DataType::Int64,
+            nullable: false,
+        };
     }
+    from_asap_column(intent.output_column(&to_asap_column(input)))
 }
diff --git a/control_plane/src/intent_algebra/column_resolution.rs b/control_plane/src/intent_algebra/column_resolution.rs
index 6ece5c1b..89357c7c 100644
--- a/control_plane/src/intent_algebra/column_resolution.rs
+++ b/control_plane/src/intent_algebra/column_resolution.rs
@@ -236,7 +236,7 @@ pub fn resolve_named_keys(keys: &[String], schema: &Schema) -> Result<Vec<Column
 /// let by: Vec<usize> = vec![];
 /// let aggs = vec![
 ///     AggIntent::Count { accuracy: types_v2::AccuracyTarget::Exact },
-///     AggIntent::Sum,
+///     AggIntent::Sum { col: None },
 /// ];
 /// let output = output_schema_for_aggregate(&input, &by, &aggs);
 /// assert_eq!(output.columns.len(), 2);          // count + sum
@@ -267,7 +267,7 @@ pub fn output_schema_for_aggregate(input: &Schema, by: &[ColumnId], aggs: &[AggI
             nullable: false,
         });
     for intent in aggs {
-        out_cols.push(intent.output_column(&probe));
+        out_cols.push(crate::intent_algebra::output_column(intent, &probe));
     }
     // Output unique_keys = [by] when by is non-empty; empty (global) → no UK.
     let unique_keys = if by.is_empty() {
@@ -410,7 +410,7 @@ mod tests {
         });
         // Group by host, region (positions 2 and 3).
         let by = vec![2usize, 3usize];
-        let aggs = vec![AggIntent::Sum];
+        let aggs = vec![AggIntent::Sum { col: None }];
         let out = output_schema_for_aggregate(&input, &by, &aggs);
         // Output columns: host, region, sum.
         assert_eq!(out.columns.len(), 3);
@@ -425,7 +425,7 @@ mod tests {
     fn output_schema_for_aggregate_drops_out_of_range_by_ids() {
         let input = infer_source_schema("m");
         // schema only has columns 0..=1; ask for by=[5] which is out of range.
-        let aggs = vec![AggIntent::Sum];
+        let aggs = vec![AggIntent::Sum { col: None }];
         let out = output_schema_for_aggregate(&input, &[5usize], &aggs);
         // The out-of-range by id is silently dropped; output has only the agg.
         assert_eq!(out.columns.len(), 1);
diff --git a/control_plane/src/intent_algebra/cse.rs b/control_plane/src/intent_algebra/cse.rs
index 008157bc..0863d6ac 100644
--- a/control_plane/src/intent_algebra/cse.rs
+++ b/control_plane/src/intent_algebra/cse.rs
@@ -217,6 +217,7 @@ mod tests {
         let q = QueryExpr::Aggregate {
             by: vec![1],
             aggs: vec![AggIntent::Quantile {
+                col: None,
                 q: 0.99,
                 accuracy: AccuracyTarget::Epsilon(0.01),
             }],
@@ -237,6 +238,7 @@ mod tests {
         let q1 = QueryExpr::Aggregate {
             by: vec![1],
             aggs: vec![AggIntent::Quantile {
+                col: None,
                 q: 0.99,
                 accuracy: AccuracyTarget::Epsilon(0.01),
             }],
@@ -246,6 +248,7 @@ mod tests {
         let q2 = QueryExpr::Aggregate {
             by: vec![1],
             aggs: vec![AggIntent::Quantile {
+                col: None,
                 q: 0.95,
                 accuracy: AccuracyTarget::Epsilon(0.01),
             }],
@@ -280,7 +283,7 @@ mod tests {
     fn dedupe_subtrees_no_shared_subexpr() {
         let q1 = QueryExpr::Aggregate {
             by: vec![],
-            aggs: vec![AggIntent::Sum],
+            aggs: vec![AggIntent::Sum { col: None }],
             having: None,
             child: Box::new(windowed_scan()),
         };
@@ -303,7 +306,7 @@ mod tests {
         };
         let q2 = QueryExpr::Aggregate {
             by: vec![],
-            aggs: vec![AggIntent::Max],
+            aggs: vec![AggIntent::Max { col: None }],
             having: None,
             child: Box::new(QueryExpr::Window {
                 kind: WindowKind::Sliding,
diff --git a/control_plane/src/intent_algebra/lower.rs b/control_plane/src/intent_algebra/lower.rs
index c0eee13a..76333c85 100644
--- a/control_plane/src/intent_algebra/lower.rs
+++ b/control_plane/src/intent_algebra/lower.rs
@@ -488,25 +488,28 @@ fn agg_func_to_intents(func: &AggFunc) -> Vec<AggIntent> {
         AggFunc::Frequency => vec![default_frequency()],
         AggFunc::Count => vec![default_frequency()],
         AggFunc::Avg => vec![AggIntent::Quantile {
+            col: None,
             q: 0.5,
             accuracy: AccuracyTarget::Epsilon(0.01),
         }],
-        AggFunc::Min => vec![AggIntent::Min],
-        AggFunc::Max => vec![AggIntent::Max],
+        AggFunc::Min => vec![AggIntent::Min { col: None }],
+        AggFunc::Max => vec![AggIntent::Max { col: None }],
         // 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 {
+                col: None,
                 q: 0.25,
                 accuracy: AccuracyTarget::Epsilon(0.01),
             },
             AggIntent::Quantile {
+                col: None,
                 q: 0.75,
                 accuracy: AccuracyTarget::Epsilon(0.01),
             },
         ],
         AggFunc::Sum | AggFunc::Rate | AggFunc::Increase | AggFunc::Delta => {
-            vec![AggIntent::Sum]
+            vec![AggIntent::Sum { col: None }]
         }
         AggFunc::Custom(_) => vec![],
     }
@@ -590,7 +593,7 @@ mod tests {
                 assert_eq!(size, Duration::from_secs(300));
                 match *child {
                     CQueryExpr::Aggregate { aggs, child, .. } => {
-                        assert!(matches!(aggs.as_slice(), [AggIntent::Sum]));
+                        assert!(matches!(aggs.as_slice(), [AggIntent::Sum { col: None }]));
                         assert!(matches!(*child, CQueryExpr::Scan { .. }));
                     }
                     other => panic!("expected Aggregate, got {other:?}"),
@@ -629,7 +632,7 @@ mod tests {
         match convert_root(&legacy).unwrap() {
             CQueryExpr::Aggregate { by, aggs, .. } => {
                 assert!(by.is_empty(), "no GROUP BY → empty `by`: {by:?}");
-                assert!(matches!(aggs.as_slice(), [AggIntent::Sum]));
+                assert!(matches!(aggs.as_slice(), [AggIntent::Sum { col: None }]));
             }
             other => panic!("expected Aggregate, got {other:?}"),
         }
diff --git a/control_plane/src/intent_algebra/mod.rs b/control_plane/src/intent_algebra/mod.rs
index b186f47a..533c2df5 100644
--- a/control_plane/src/intent_algebra/mod.rs
+++ b/control_plane/src/intent_algebra/mod.rs
@@ -102,8 +102,9 @@ pub use binder::{Binder, SchemaCatalog, UsageDerivedCatalog};
 // Re-exports for the canonical surface — `crate::intent_algebra::*` for
 // downstream callers that don't want to chase sub-module paths.
 pub use agg_intent::{
-    agg_accuracy, agg_is_exact, agg_is_mergeable, default_cardinality, default_frequency,
-    default_quantile, AggIntent,
+    agg_accuracy, agg_is_exact, agg_is_mergeable, archive_only, as_frequency, default_cardinality,
+    default_frequency, default_quantile, frequency, is_frequency_heavy_hitter, output_column,
+    ranking_measure, AggIntent, MathFunc, RankingMeasure, TimeFunc,
 };
 pub use cse::{dedupe_subtrees, CseWorkloadPlan};
 pub use query_expr::{
diff --git a/control_plane/src/intent_algebra/query_expr.rs b/control_plane/src/intent_algebra/query_expr.rs
index 30e1bd84..cb3aed3d 100644
--- a/control_plane/src/intent_algebra/query_expr.rs
+++ b/control_plane/src/intent_algebra/query_expr.rs
@@ -562,7 +562,7 @@ impl QueryExpr {
                         nullable: false,
                     });
                 for intent in aggs {
-                    out_cols.push(intent.output_column(&probe));
+                    out_cols.push(crate::intent_algebra::output_column(intent, &probe));
                 }
                 // Output unique_keys = [by]. The group-by column tuple
                 // is unique in the output by construction (design.md §6
@@ -823,6 +823,7 @@ mod tests {
         let expr = QueryExpr::Aggregate {
             by: vec![1], // service
             aggs: vec![AggIntent::Quantile {
+                col: None,
                 q: 0.99,
                 accuracy: AccuracyTarget::Epsilon(0.01),
             }],
@@ -860,7 +861,7 @@ mod tests {
             }),
             child: Box::new(QueryExpr::Aggregate {
                 by: vec![1],
-                aggs: vec![AggIntent::Max],
+                aggs: vec![AggIntent::Max { col: None }],
                 having: None,
                 child: Box::new(QueryExpr::Ref {
                     name: BindingName::new("w"),
@@ -906,7 +907,7 @@ mod tests {
     fn query_expr_aggregate_invalid_by_column() {
         let expr = QueryExpr::Aggregate {
             by: vec![99],
-            aggs: vec![AggIntent::Sum],
+            aggs: vec![AggIntent::Sum { col: None }],
             having: None,
             child: Box::new(ts_scan()),
         };
@@ -919,6 +920,7 @@ mod tests {
         let expr = QueryExpr::Aggregate {
             by: vec![1],
             aggs: vec![AggIntent::Quantile {
+                col: None,
                 q: 0.99,
                 accuracy: AccuracyTarget::Epsilon(0.01),
             }],
@@ -942,7 +944,7 @@ mod tests {
         // consumers can legally share this.
         let producer_with_uk = QueryExpr::Aggregate {
             by: vec![1],
-            aggs: vec![AggIntent::Sum],
+            aggs: vec![AggIntent::Sum { col: None }],
             having: None,
             child: Box::new(QueryExpr::Window {
                 kind: WindowKind::Sliding,
diff --git a/control_plane/src/intent_algebra/relational.rs b/control_plane/src/intent_algebra/relational.rs
index b418fef3..bc776ba8 100644
--- a/control_plane/src/intent_algebra/relational.rs
+++ b/control_plane/src/intent_algebra/relational.rs
@@ -515,10 +515,10 @@ impl AggFunc {
             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),
+            AggFunc::Sum => Some(AggIntent::Sum { col: None }),
+            AggFunc::Avg => Some(AggIntent::Avg { col: None }),
+            AggFunc::Min => Some(AggIntent::Min { col: None }),
+            AggFunc::Max => Some(AggIntent::Max { col: None }),
             _ => None,
         }
     }
@@ -795,7 +795,7 @@ mod tests {
     #[test]
     fn agg_func_to_sketch_op_heavy_hitters() {
         let op = AggFunc::HeavyHitters { k: 50 }.to_sketch_op();
-        assert!(matches!(op, Some(AggIntent::Frequency { .. })));
+        assert!(op.is_some_and(|i| crate::intent_algebra::as_frequency(&i).is_some()));
     }
 
     // ── ScalarExpr predicate list conversion ──────────────────────────────────
@@ -914,7 +914,7 @@ mod tests {
 
     #[test]
     fn agg_intent_avg_not_mergeable() {
-        assert!(!agg_is_mergeable(&AggIntent::Avg));
+        assert!(!agg_is_mergeable(&AggIntent::Avg { col: None }));
     }
 
     #[test]
@@ -936,9 +936,9 @@ mod tests {
 
     #[test]
     fn agg_intent_is_exact() {
-        assert!(agg_is_exact(&AggIntent::Sum));
-        assert!(agg_is_exact(&AggIntent::Min));
-        assert!(agg_is_exact(&AggIntent::Max));
+        assert!(agg_is_exact(&AggIntent::Sum { col: None }));
+        assert!(agg_is_exact(&AggIntent::Min { col: None }));
+        assert!(agg_is_exact(&AggIntent::Max { col: None }));
         assert!(!agg_is_exact(&default_cardinality()));
     }
 }
diff --git a/control_plane/src/optimizer/cost/mod.rs b/control_plane/src/optimizer/cost/mod.rs
index 8368966a..c60a5be1 100644
--- a/control_plane/src/optimizer/cost/mod.rs
+++ b/control_plane/src/optimizer/cost/mod.rs
@@ -703,20 +703,25 @@ fn node_cost_aggregate(by: &[usize], aggs: &[AggIntent], _input: &Schema) -> f64
 /// agrees with L4's "sketch is cheaper than exact for these" intuition.
 #[allow(dead_code)]
 fn intent_cost(intent: &AggIntent) -> f64 {
+    if crate::intent_algebra::as_frequency(intent).is_some() {
+        return 15.0;
+    }
     match intent {
-        AggIntent::Sum | AggIntent::Min | AggIntent::Max | AggIntent::Avg => 5.0,
+        AggIntent::Sum { .. }
+        | AggIntent::Min { .. }
+        | AggIntent::Max { .. }
+        | AggIntent::Avg { .. } => 5.0,
         AggIntent::Count { .. } => 5.0,
         AggIntent::Quantile { .. } => 20.0,
         AggIntent::Cardinality { .. } => 15.0,
         AggIntent::TopK { .. } => 25.0,
-        AggIntent::Frequency { .. } => 15.0,
-        AggIntent::Rate { .. } | AggIntent::Increase { .. } => 8.0,
+        AggIntent::Rate | AggIntent::Increase => 8.0,
         // Phase β archive-only intents — priced as a cold-tier scan
         // rather than a streaming aggregate. Higher than `Sum` (the engine
         // must read the raw archive) but lower than the sketch intents
         // (no per-sample sketch update on the hot path). Tightening this
         // is a follow-up once real measurements land.
-        intent if intent.archive_only() => 12.0,
+        intent if crate::intent_algebra::archive_only(intent) => 12.0,
         // Defensive fallback — any future intent that isn't archive-only
         // and doesn't match an explicit arm prices as a generic aggregate.
         _ => 5.0,
@@ -777,6 +782,7 @@ mod workload_cost_tests {
         QueryExpr::Aggregate {
             by: vec![],
             aggs: vec![AggIntent::Quantile {
+                col: None,
                 q,
                 accuracy: AccuracyTarget::Epsilon(0.01),
             }],
@@ -789,7 +795,7 @@ mod workload_cost_tests {
     fn max_root(child: QueryExpr) -> QueryExpr {
         QueryExpr::Aggregate {
             by: vec![],
-            aggs: vec![AggIntent::Max],
+            aggs: vec![AggIntent::Max { col: None }],
             having: None,
             child: Box::new(child),
         }
diff --git a/control_plane/src/optimizer/engine.rs b/control_plane/src/optimizer/engine.rs
index e22b7318..4db922d5 100644
--- a/control_plane/src/optimizer/engine.rs
+++ b/control_plane/src/optimizer/engine.rs
@@ -220,7 +220,7 @@ impl CostModel for DefaultCostModel {
                 match &aggs[0] {
                     AggIntent::Quantile { .. } => 0.05,
                     AggIntent::Cardinality { .. } => 0.02,
-                    AggIntent::Frequency { .. } => 0.03,
+                    op if crate::intent_algebra::as_frequency(op).is_some() => 0.03,
                     AggIntent::TopK { k, .. } => (*k as f64).recip().min(0.1),
                     op if agg_is_exact(op) => 1.0,
                     _ => 0.1,
diff --git a/control_plane/src/optimizer/rules/mod.rs b/control_plane/src/optimizer/rules/mod.rs
index ad277398..08fe6dba 100644
--- a/control_plane/src/optimizer/rules/mod.rs
+++ b/control_plane/src/optimizer/rules/mod.rs
@@ -177,15 +177,15 @@ pub fn bind_workload_typed(w: &QueryWorkload) -> Option<crate::sketch_algebra::P
     // PromQL `topk(10, …)` lowering uses.
     let intent = match statistic {
         StatisticClass::Quantile => L3AggIntent::Quantile {
+            col: None,
             q: w.quantiles.first().copied().unwrap_or(0.99),
             accuracy: intent_accuracy,
         },
         StatisticClass::Cardinality => L3AggIntent::Cardinality {
+            col: None,
             accuracy: intent_accuracy,
         },
-        StatisticClass::Frequency => L3AggIntent::Frequency {
-            accuracy: intent_accuracy,
-        },
+        StatisticClass::Frequency => crate::intent_algebra::frequency(intent_accuracy),
         StatisticClass::TopK => L3AggIntent::TopK {
             k: 10,
             accuracy: intent_accuracy,
diff --git a/control_plane/src/physical/allocator.rs b/control_plane/src/physical/allocator.rs
index d99e5865..f5fda46f 100644
--- a/control_plane/src/physical/allocator.rs
+++ b/control_plane/src/physical/allocator.rs
@@ -527,7 +527,7 @@ impl SketchAllocator {
         let intent = aggs[0].clone();
 
         // Exact non-mergeable (Avg) → always Db.
-        if matches!(intent, AggIntent::Avg) {
+        if matches!(intent, AggIntent::Avg { .. }) {
             return PlanNode {
                 expr: QueryExpr::Aggregate {
                     by,
@@ -671,28 +671,52 @@ fn estimated_sketch_memory(op: &AggIntent) -> f64 {
 /// Map a canonical [`AggIntent`] to a short stable kind string for
 /// annotation rationale text.
 fn canonical_intent_kind_str(intent: &AggIntent) -> &'static str {
+    if crate::intent_algebra::as_frequency(intent).is_some() {
+        return "frequency";
+    }
     match intent {
         AggIntent::Count { .. } => "count",
-        AggIntent::Sum => "sum",
-        AggIntent::Min => "min",
-        AggIntent::Max => "max",
-        AggIntent::Avg => "avg",
+        AggIntent::Sum { .. } => "sum",
+        AggIntent::Min { .. } => "min",
+        AggIntent::Max { .. } => "max",
+        AggIntent::Avg { .. } => "avg",
+        AggIntent::StdDev { .. } => "stddev",
+        AggIntent::Variance { .. } => "variance",
         AggIntent::Quantile { .. } => "quantile",
         AggIntent::TopK { .. } => "topk",
         AggIntent::Cardinality { .. } => "cardinality",
-        AggIntent::Frequency { .. } => "frequency",
-        AggIntent::Rate { .. } => "rate",
-        AggIntent::Increase { .. } => "increase",
+        AggIntent::Rate => "rate",
+        AggIntent::Increase => "increase",
         AggIntent::Absent => "absent",
-        AggIntent::Present => "present",
-        AggIntent::Delta { .. } => "delta",
-        AggIntent::Deriv { .. } => "deriv",
+        AggIntent::AbsentOverTime => "absent_over_time",
+        AggIntent::PresentOverTime => "present_over_time",
+        AggIntent::Delta => "delta",
+        AggIntent::Deriv => "deriv",
         AggIntent::PredictLinear { .. } => "predict_linear",
-        AggIntent::HoltWinters { .. } => "holt_winters",
-        AggIntent::Idelta { .. } => "idelta",
-        AggIntent::Irate { .. } => "irate",
-        AggIntent::Resets { .. } => "resets",
-        AggIntent::Changes { .. } => "changes",
+        AggIntent::DoubleExpSmoothing { .. } => "double_exponential_smoothing",
+        AggIntent::IDelta => "idelta",
+        AggIntent::Resets => "resets",
+        AggIntent::Changes => "changes",
+        AggIntent::HistogramCount => "histogram_count",
+        AggIntent::HistogramSum => "histogram_sum",
+        AggIntent::HistogramAvg => "histogram_avg",
+        AggIntent::HistogramStdDev => "histogram_stddev",
+        AggIntent::HistogramStdVar => "histogram_stdvar",
+        AggIntent::HistogramFraction { .. } => "histogram_fraction",
+        AggIntent::HistogramQuantile { .. } => "histogram_quantile",
+        AggIntent::Math(_) => "math",
+        AggIntent::TimeFn(_) => "time_fn",
+        AggIntent::Group => "group",
+        AggIntent::CountValues { .. } => "count_values",
+        AggIntent::LastOverTime => "last_over_time",
+        AggIntent::FirstOverTime => "first_over_time",
+        AggIntent::MadOverTime => "mad_over_time",
+        AggIntent::TsOfMinOverTime => "ts_of_min_over_time",
+        AggIntent::TsOfMaxOverTime => "ts_of_max_over_time",
+        AggIntent::TsOfFirstOverTime => "ts_of_first_over_time",
+        AggIntent::TsOfLastOverTime => "ts_of_last_over_time",
+        // Unrecognized Extension (not the Frequency one, guarded above).
+        AggIntent::Extension { .. } => "extension",
     }
 }
 
@@ -808,7 +832,7 @@ mod tests {
 
     #[test]
     fn exact_avg_goes_to_db() {
-        let node = alloc(unlimited(), agg(AggIntent::Avg));
+        let node = alloc(unlimited(), agg(AggIntent::Avg { col: None }));
         assert_eq!(node.stage, PipelineStage::Db);
         assert_eq!(node.mode, ExecutionMode::Exact);
     }
@@ -817,7 +841,7 @@ mod tests {
 
     #[test]
     fn exact_sum_goes_to_backend() {
-        let node = alloc(unlimited(), agg(AggIntent::Sum));
+        let node = alloc(unlimited(), agg(AggIntent::Sum { col: None }));
         assert_eq!(node.stage, PipelineStage::Backend);
         assert_eq!(node.mode, ExecutionMode::Exact);
     }
@@ -886,7 +910,7 @@ mod tests {
     fn multi_intent_aggregate_goes_to_db() {
         let expr = QueryExpr::Aggregate {
             by: vec![],
-            aggs: vec![AggIntent::Sum, AggIntent::Min],
+            aggs: vec![AggIntent::Sum { col: None }, AggIntent::Min { col: None }],
             having: None,
             child: Box::new(scan("m")),
         };
diff --git a/control_plane/src/physical/planner.rs b/control_plane/src/physical/planner.rs
index 91802d0a..8c757752 100644
--- a/control_plane/src/physical/planner.rs
+++ b/control_plane/src/physical/planner.rs
@@ -635,6 +635,7 @@ mod tests {
     #[test]
     fn resolve_preserves_intent() {
         let intent = AggIntent::Quantile {
+            col: None,
             q: 0.99,
             accuracy: AccuracyTarget::Epsilon(0.005),
         };
@@ -721,7 +722,7 @@ mod tests {
         // (no single sketch serves multiple intents).
         let expr = QueryExpr::Aggregate {
             by: vec![],
-            aggs: vec![AggIntent::Sum, AggIntent::Min],
+            aggs: vec![AggIntent::Sum { col: None }, AggIntent::Min { col: None }],
             having: None,
             child: Box::new(scan("trades")),
         };
diff --git a/control_plane/src/physical/sketch_catalog.rs b/control_plane/src/physical/sketch_catalog.rs
index fd464cdf..186e925d 100644
--- a/control_plane/src/physical/sketch_catalog.rs
+++ b/control_plane/src/physical/sketch_catalog.rs
@@ -67,10 +67,12 @@ 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 {
+    if crate::intent_algebra::as_frequency(op).is_some() {
+        return SketchType::CountSketch;
+    }
     match op {
         AggIntent::Quantile { .. } => SketchType::DDSketch,
         AggIntent::Cardinality { .. } => SketchType::HLL,
-        AggIntent::Frequency { .. } => SketchType::CountSketch,
         // 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
@@ -91,6 +93,13 @@ pub fn sketch_type_for_per_partition(wrap: &PerPartitionWrap) -> SketchType {
 
 /// Derive [`SketchParams`] from an [`AggIntent`] IR node.
 pub fn sketch_params_for_op(op: &AggIntent) -> SketchParams {
+    if crate::intent_algebra::as_frequency(op).is_some() {
+        let acc = agg_accuracy(op);
+        return SketchParams::CountSketch {
+            epsilon: acc,
+            delta: 0.01,
+        };
+    }
     match op {
         AggIntent::Quantile { q, .. } => SketchParams::DDSketch {
             relative_accuracy: agg_accuracy(op),
@@ -106,16 +115,9 @@ pub fn sketch_params_for_op(op: &AggIntent) -> SketchParams {
             let precision = (registers as f64).log2() as u32;
             SketchParams::HLL { precision }
         }
-        AggIntent::Frequency { .. } => {
-            let acc = agg_accuracy(op);
-            SketchParams::CountSketch {
-                epsilon: acc,
-                delta: 0.01,
-            }
-        }
         // Min / Max — preserve the legacy `Extrema` mapping (DDSketch over
         // the 0.0 / 1.0 boundary quantiles).
-        AggIntent::Min | AggIntent::Max => SketchParams::DDSketch {
+        AggIntent::Min { .. } | AggIntent::Max { .. } => SketchParams::DDSketch {
             relative_accuracy: 0.01,
             quantiles: vec![0.0, 1.0],
         },
@@ -145,6 +147,12 @@ pub fn sketch_type_and_params(op: &AggIntent) -> (SketchType, SketchParams) {
 /// Used by the physical planner's placement decision to defer a sketch
 /// build off a stage when its budget would be exceeded.
 pub fn estimated_sketch_memory_bytes(op: &AggIntent) -> u64 {
+    if crate::intent_algebra::as_frequency(op).is_some() {
+        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 / acc) as u64;
+        return width * 5 * 8;
+    }
     match op {
         AggIntent::Quantile { .. } => 4_096,
         AggIntent::Cardinality { .. } => {
@@ -153,14 +161,8 @@ pub fn estimated_sketch_memory_bytes(op: &AggIntent) -> u64 {
             let registers = ((1.04 / acc).powi(2) as u64).next_power_of_two();
             registers.max(16)
         }
-        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 / acc) as u64;
-            width * 5 * 8
-        }
         // Legacy `Extrema { .. }` (now canonical Min / Max) — 16 bytes.
-        AggIntent::Min | AggIntent::Max => 16,
+        AggIntent::Min { .. } | AggIntent::Max { .. } => 16,
         // Sum / Count / Avg / TopK / Rate / Increase / archive-only — no
         // sketch state; preserve the legacy `Exact(_) → 8` mapping.
         _ => 8,
@@ -270,6 +272,7 @@ mod tests {
     fn op_quantile_yields_ddsketch_type_and_params() {
         use crate::types_v2::AccuracyTarget;
         let op = AggIntent::Quantile {
+            col: None,
             q: 0.5,
             accuracy: AccuracyTarget::Epsilon(0.01),
         };
@@ -303,6 +306,7 @@ mod tests {
     fn memory_quantile() {
         use crate::types_v2::AccuracyTarget;
         let op = AggIntent::Quantile {
+            col: None,
             q: 0.5,
             accuracy: AccuracyTarget::Epsilon(0.01),
         };
diff --git a/control_plane/src/physical/window_fusion.rs b/control_plane/src/physical/window_fusion.rs
index 7d6b9b04..46758857 100644
--- a/control_plane/src/physical/window_fusion.rs
+++ b/control_plane/src/physical/window_fusion.rs
@@ -321,7 +321,7 @@ mod tests {
     #[test]
     fn equivalence_sum_intent() {
         assert_equivalent(
-            AggIntent::Sum,
+            AggIntent::Sum { col: None },
             WindowKind::Tumbling,
             Duration::from_secs(300),
             None,
@@ -334,7 +334,7 @@ mod tests {
         // sketch case — not a windowed sketch.
         let canonical = QueryExpr::Aggregate {
             by: Vec::new(),
-            aggs: vec![AggIntent::Sum],
+            aggs: vec![AggIntent::Sum { col: None }],
             having: None,
             child: Box::new(canonical_scan("m")),
         };
diff --git a/control_plane/src/pipeline.rs b/control_plane/src/pipeline.rs
index 257b4589..20b102c5 100644
--- a/control_plane/src/pipeline.rs
+++ b/control_plane/src/pipeline.rs
@@ -163,7 +163,7 @@ impl Analyzer {
         let accuracy_sla = match &spec.accuracy {
             Some(AccuracyTarget::Exact) => 1.0,
             Some(AccuracyTarget::Epsilon(eps)) => (1.0 - eps).clamp(0.0, 1.0),
-            Some(AccuracyTarget::EpsilonDelta { eps, .. }) => (1.0 - eps).clamp(0.0, 1.0),
+            Some(AccuracyTarget::EpsilonDelta { epsilon, .. }) => (1.0 - epsilon).clamp(0.0, 1.0),
             None => spec.accuracy_sla,
         };
 
@@ -734,7 +734,7 @@ mod tests {
             "accuracy_sla":     0.5,
             "id":               "q-001",
             "language":         "prom_ql",
-            "accuracy":         { "kind": "epsilon", "value": 0.02 },
+            "accuracy":         { "Epsilon": 0.02 },
             "dollars":          0.001,
             "deployment_model": "asaplifecycle",
             "shape":            { "kind": "periodic", "every": { "secs": 60, "nanos": 0 } },
diff --git a/control_plane/src/query_parser/mod.rs b/control_plane/src/query_parser/mod.rs
index bcd425ce..3c0521e2 100644
--- a/control_plane/src/query_parser/mod.rs
+++ b/control_plane/src/query_parser/mod.rs
@@ -311,7 +311,7 @@ impl QeCollector {
                     self.agg_types.push(AggType::Cardinality);
                 }
             }
-            AggIntent::Frequency { .. } => {
+            op if crate::intent_algebra::as_frequency(op).is_some() => {
                 if !self.agg_types.contains(&AggType::Frequency) {
                     self.agg_types.push(AggType::Frequency);
                 }
@@ -324,7 +324,7 @@ impl QeCollector {
                     self.quantiles.push(*q);
                 }
             }
-            AggIntent::Min => {
+            AggIntent::Min { .. } => {
                 if !self.agg_types.contains(&AggType::Quantile) {
                     self.agg_types.push(AggType::Quantile);
                 }
@@ -332,7 +332,7 @@ impl QeCollector {
                     self.quantiles.push(0.0);
                 }
             }
-            AggIntent::Max => {
+            AggIntent::Max { .. } => {
                 if !self.agg_types.contains(&AggType::Quantile) {
                     self.agg_types.push(AggType::Quantile);
                 }
diff --git a/control_plane/src/query_parser/promql.rs b/control_plane/src/query_parser/promql.rs
index 49e39e21..8aa1eb69 100644
--- a/control_plane/src/query_parser/promql.rs
+++ b/control_plane/src/query_parser/promql.rs
@@ -33,9 +33,7 @@
 use std::time::Duration;
 
 use anyhow::anyhow;
-use promql_parser::parser::{
-    self, AggregateExpr, Call, Expr, LabelModifier, VectorSelector,
-};
+use promql_parser::parser::{self, AggregateExpr, Call, Expr, LabelModifier, VectorSelector};
 
 use crate::intent_algebra::relational::{FilterOp, FilterVal, PartitionKeys, Predicate};
 
diff --git a/control_plane/src/sketch_algebra/capability.rs b/control_plane/src/sketch_algebra/capability.rs
index bbb27560..2477bf5a 100644
--- a/control_plane/src/sketch_algebra/capability.rs
+++ b/control_plane/src/sketch_algebra/capability.rs
@@ -475,6 +475,22 @@ fn multi_pop_satisfies_single(required: AggregationType, available: AggregationT
 /// | `Avg` | `None` — needs cross-policy join (Sum + Count); follow-up |
 /// | Every archive-only intent | `None` |
 pub fn capability_for(intent: &AggIntent) -> Option<Capability> {
+    if let Some(accuracy) = crate::intent_algebra::as_frequency(intent) {
+        return if is_exact(&accuracy) {
+            // Exact aggregation — sketch fallback is only meaningful
+            // when raw counters aren't kept at the ingest tier; with
+            // accuracy=Exact the caller wants exact `sum by (label)
+            // (rate(...))`, which routes to archive.
+            None
+        } else {
+            // Bare frequency point-query uses a frequency-family
+            // sketch — any of CMS / CountSketch / CmsWithHeap /
+            // CountSketchWithHeap works (the heap is additional
+            // info that the FrequencyTopk path uses). `Any` here
+            // means the optimizer picks the cheapest indexed sid.
+            Some(Capability::FrequencyEstimate(SketchKindHandle::Any))
+        };
+    }
     match intent {
         AggIntent::Quantile { accuracy, .. } => {
             if is_exact(accuracy) {
@@ -483,7 +499,7 @@ pub fn capability_for(intent: &AggIntent) -> Option<Capability> {
                 Some(Capability::QuantileApprox(SketchKindHandle::Any))
             }
         }
-        AggIntent::Cardinality { accuracy } => {
+        AggIntent::Cardinality { accuracy, .. } => {
             if is_exact(accuracy) {
                 None
             } else {
@@ -532,27 +548,13 @@ pub fn capability_for(intent: &AggIntent) -> Option<Capability> {
                 Some(Capability::FrequencyTopk(SketchKindHandle::Any))
             }
         }
-        AggIntent::Frequency { accuracy } => {
-            if is_exact(accuracy) {
-                // Exact aggregation — sketch fallback is only meaningful
-                // when raw counters aren't kept at the ingest tier; with
-                // accuracy=Exact the caller wants exact `sum by (label)
-                // (rate(...))`, which routes to archive.
-                None
-            } else {
-                // Bare frequency point-query uses a frequency-family
-                // sketch — any of CMS / CountSketch / CmsWithHeap /
-                // CountSketchWithHeap works (the heap is additional
-                // info that the FrequencyTopk path uses). `Any` here
-                // means the optimizer picks the cheapest indexed sid.
-                Some(Capability::FrequencyEstimate(SketchKindHandle::Any))
-            }
-        }
         // ── Min / Max via quantile sketches ──────────────────────────
         // DDSketch / KLL answer min = quantile(0) and max = quantile(1)
         // out of the box. No dedicated extrema sketch is needed; route
         // these through the quantile-family handler.
-        AggIntent::Min | AggIntent::Max => Some(Capability::QuantileApprox(SketchKindHandle::Any)),
+        AggIntent::Min { .. } | AggIntent::Max { .. } => {
+            Some(Capability::QuantileApprox(SketchKindHandle::Any))
+        }
         // ── ExactAgg (PR-6 follow-up) ────────────────────────────────
         // These intents previously returned `None` and routed to the
         // archive engine. Now that the data plane carries
@@ -561,28 +563,53 @@ pub fn capability_for(intent: &AggIntent) -> Option<Capability> {
         // state instead. `is_satisfied_by` checks `agg_type` equality
         // structurally — a sid registered as `ExactAgg(Sum)` only
         // satisfies a required `ExactAgg(Sum)`.
-        AggIntent::Sum => Some(Capability::ExactAgg(AggregationType::Sum)),
-        AggIntent::Rate { .. } | AggIntent::Increase { .. } => {
+        AggIntent::Sum { .. } => Some(Capability::ExactAgg(AggregationType::Sum)),
+        AggIntent::Rate | AggIntent::Increase => {
             Some(Capability::ExactAgg(AggregationType::Increase))
         }
-        // ── Avg: still no ASAP-tier substitute ───────────────────────
+        // ── Avg / StdDev / Variance: still no ASAP-tier substitute ────
         // Avg = Sum / Count, which needs two separate ExactAgg policies
         // (one for Sum, one for Count) joined at query time. The L4
         // binder doesn't yet emit that pattern, so capability_for keeps
-        // Avg on the archive path for now. Follow-up.
-        AggIntent::Avg => None,
+        // these on the archive path for now. Follow-up.
+        AggIntent::Avg { .. } | AggIntent::StdDev { .. } | AggIntent::Variance { .. } => None,
         // Archive-only intents — never bind to a ASAP-tier capability;
-        // routed to the cold tier (Gorilla / Thanos).
+        // routed to the cold tier (Gorilla / Thanos). Includes every
+        // intent added by the Phase 1 IR merge (none has a `Bind*` rule
+        // yet) plus the pre-existing archive-only set. `Irate` is
+        // intentionally absent — folded into `Rate` above (see
+        // `agg_intent.rs` module docs).
         AggIntent::Absent
-        | AggIntent::Present
-        | AggIntent::Delta { .. }
-        | AggIntent::Deriv { .. }
+        | AggIntent::AbsentOverTime
+        | AggIntent::PresentOverTime
+        | AggIntent::Delta
+        | AggIntent::Deriv
         | AggIntent::PredictLinear { .. }
-        | AggIntent::HoltWinters { .. }
-        | AggIntent::Idelta { .. }
-        | AggIntent::Irate { .. }
-        | AggIntent::Resets { .. }
-        | AggIntent::Changes { .. } => None,
+        | AggIntent::DoubleExpSmoothing { .. }
+        | AggIntent::IDelta
+        | AggIntent::Resets
+        | AggIntent::Changes
+        | AggIntent::HistogramCount
+        | AggIntent::HistogramSum
+        | AggIntent::HistogramAvg
+        | AggIntent::HistogramStdDev
+        | AggIntent::HistogramStdVar
+        | AggIntent::HistogramFraction { .. }
+        | AggIntent::HistogramQuantile { .. }
+        | AggIntent::Math(_)
+        | AggIntent::TimeFn(_)
+        | AggIntent::Group
+        | AggIntent::CountValues { .. }
+        | AggIntent::LastOverTime
+        | AggIntent::FirstOverTime
+        | AggIntent::MadOverTime
+        | AggIntent::TsOfMinOverTime
+        | AggIntent::TsOfMaxOverTime
+        | AggIntent::TsOfFirstOverTime
+        | AggIntent::TsOfLastOverTime => None,
+        // Unrecognized Extension (not the Frequency one, guarded above) -- no
+        // binding exists for a shape core cannot even see into.
+        AggIntent::Extension { .. } => None,
     }
 }
 
@@ -838,6 +865,7 @@ mod tests {
     #[test]
     fn capability_for_quantile_returns_quantile_approx() {
         let intent = AggIntent::Quantile {
+            col: None,
             q: 0.99,
             accuracy: AccuracyTarget::Epsilon(0.01),
         };
@@ -850,6 +878,7 @@ mod tests {
     #[test]
     fn capability_for_quantile_exact_returns_none() {
         let intent = AggIntent::Quantile {
+            col: None,
             q: 0.99,
             accuracy: AccuracyTarget::Exact,
         };
@@ -861,6 +890,7 @@ mod tests {
     #[test]
     fn capability_for_cardinality_with_epsilon_returns_cardinality_approx() {
         let intent = AggIntent::Cardinality {
+            col: None,
             accuracy: AccuracyTarget::Epsilon(0.01),
         };
         assert_eq!(capability_for(&intent), Some(Capability::CardinalityApprox));
@@ -869,8 +899,9 @@ mod tests {
     #[test]
     fn capability_for_cardinality_with_epsilon_delta_returns_cardinality_approx() {
         let intent = AggIntent::Cardinality {
+            col: None,
             accuracy: AccuracyTarget::EpsilonDelta {
-                eps: 0.01,
+                epsilon: 0.01,
                 delta: 0.001,
             },
         };
@@ -880,6 +911,7 @@ mod tests {
     #[test]
     fn capability_for_cardinality_with_exact_returns_none() {
         let intent = AggIntent::Cardinality {
+            col: None,
             accuracy: AccuracyTarget::Exact,
         };
         assert_eq!(capability_for(&intent), None);
@@ -913,7 +945,7 @@ mod tests {
         // PR-6 follow-up: Sum routes to ASAP-tier ExactAgg(Sum) state.
         // Pre-follow-up this returned `None`.
         assert_eq!(
-            capability_for(&AggIntent::Sum),
+            capability_for(&AggIntent::Sum { col: None }),
             Some(Capability::ExactAgg(AggregationType::Sum))
         );
     }
@@ -924,14 +956,14 @@ mod tests {
         // today (a sketch-bound `Avg` would fold onto `Quantile{q=0.5}`
         // only when the cost model allows the relaxation, which is a
         // follow-up).
-        assert_eq!(capability_for(&AggIntent::Avg), None);
+        assert_eq!(capability_for(&AggIntent::Avg { col: None }), None);
     }
 
     #[test]
     fn capability_for_min_returns_quantile_approx() {
         // Min = quantile(0); DDSketch / KLL answer it directly.
         assert_eq!(
-            capability_for(&AggIntent::Min),
+            capability_for(&AggIntent::Min { col: None }),
             Some(Capability::QuantileApprox(SketchKindHandle::Any))
         );
     }
@@ -940,7 +972,7 @@ mod tests {
     fn capability_for_max_returns_quantile_approx() {
         // Max = quantile(1); DDSketch / KLL answer it directly.
         assert_eq!(
-            capability_for(&AggIntent::Max),
+            capability_for(&AggIntent::Max { col: None }),
             Some(Capability::QuantileApprox(SketchKindHandle::Any))
         );
     }
@@ -951,18 +983,8 @@ mod tests {
         // ExactAgg(Increase) — the counter-reset-aware exact precompute.
         // Pre-follow-up this returned `None`.
         let exact_inc = Some(Capability::ExactAgg(AggregationType::Increase));
-        assert_eq!(
-            capability_for(&AggIntent::Rate {
-                window: Duration::from_secs(60)
-            }),
-            exact_inc
-        );
-        assert_eq!(
-            capability_for(&AggIntent::Increase {
-                window: Duration::from_secs(60)
-            }),
-            exact_inc
-        );
+        assert_eq!(capability_for(&AggIntent::Rate), exact_inc);
+        assert_eq!(capability_for(&AggIntent::Increase), exact_inc);
     }
 
     #[test]
@@ -994,9 +1016,7 @@ mod tests {
 
     #[test]
     fn frequency_estimate_with_epsilon_returns_frequency_estimate_approx() {
-        let intent = AggIntent::Frequency {
-            accuracy: AccuracyTarget::Epsilon(0.01),
-        };
+        let intent = crate::intent_algebra::frequency(AccuracyTarget::Epsilon(0.01));
         assert_eq!(
             capability_for(&intent),
             Some(Capability::FrequencyEstimate(SketchKindHandle::Any))
@@ -1005,12 +1025,10 @@ mod tests {
 
     #[test]
     fn frequency_estimate_with_epsilon_delta_returns_frequency_estimate_approx() {
-        let intent = AggIntent::Frequency {
-            accuracy: AccuracyTarget::EpsilonDelta {
-                eps: 0.01,
-                delta: 0.001,
-            },
-        };
+        let intent = crate::intent_algebra::frequency(AccuracyTarget::EpsilonDelta {
+            epsilon: 0.01,
+            delta: 0.001,
+        });
         assert_eq!(
             capability_for(&intent),
             Some(Capability::FrequencyEstimate(SketchKindHandle::Any))
@@ -1021,35 +1039,23 @@ mod tests {
     fn frequency_estimate_with_exact_returns_none() {
         // Exact aggregation routes to archive (sketch fallback only
         // meaningful when raw counters aren't kept).
-        let intent = AggIntent::Frequency {
-            accuracy: AccuracyTarget::Exact,
-        };
+        let intent = crate::intent_algebra::frequency(AccuracyTarget::Exact);
         assert_eq!(capability_for(&intent), None);
     }
 
     #[test]
     fn capability_for_archive_only_intents_return_none() {
-        // Spot-check each archive-only variant.
+        // Spot-check each archive-only variant, including a few added by
+        // the Phase 1 IR merge. `Irate` is intentionally absent — folded
+        // into `Rate` (see `agg_intent.rs` module docs); `rate`/`irate`
+        // now share `capability_for`'s `Rate` arm.
         assert_eq!(capability_for(&AggIntent::Absent), None);
-        assert_eq!(capability_for(&AggIntent::Present), None);
-        assert_eq!(
-            capability_for(&AggIntent::Delta {
-                window: Duration::from_secs(60)
-            }),
-            None
-        );
-        assert_eq!(
-            capability_for(&AggIntent::Idelta {
-                window: Duration::from_secs(60)
-            }),
-            None
-        );
-        assert_eq!(
-            capability_for(&AggIntent::Irate {
-                window: Duration::from_secs(60)
-            }),
-            None
-        );
+        assert_eq!(capability_for(&AggIntent::AbsentOverTime), None);
+        assert_eq!(capability_for(&AggIntent::PresentOverTime), None);
+        assert_eq!(capability_for(&AggIntent::Delta), None);
+        assert_eq!(capability_for(&AggIntent::IDelta), None);
+        assert_eq!(capability_for(&AggIntent::HistogramCount), None);
+        assert_eq!(capability_for(&AggIntent::Group), None);
     }
 
     // ── Capability::is_satisfied_by ──────────────────────────────────────
@@ -1238,19 +1244,15 @@ mod tests {
         // data plane has a real accumulator for: `Sum` (SumAccumulator)
         // and `Rate` / `Increase` (IncreaseAccumulator).
         assert_eq!(
-            capability_for(&AggIntent::Sum),
+            capability_for(&AggIntent::Sum { col: None }),
             Some(Capability::ExactAgg(AggregationType::Sum))
         );
         assert_eq!(
-            capability_for(&AggIntent::Rate {
-                window: Duration::from_secs(60)
-            }),
+            capability_for(&AggIntent::Rate),
             Some(Capability::ExactAgg(AggregationType::Increase))
         );
         assert_eq!(
-            capability_for(&AggIntent::Increase {
-                window: Duration::from_secs(60)
-            }),
+            capability_for(&AggIntent::Increase),
             Some(Capability::ExactAgg(AggregationType::Increase))
         );
         // `Count{Exact}` (count_over_time) and `Avg` both need a real
@@ -1262,7 +1264,7 @@ mod tests {
             }),
             None
         );
-        assert_eq!(capability_for(&AggIntent::Avg), None);
+        assert_eq!(capability_for(&AggIntent::Avg { col: None }), None);
     }
 
     #[test]
diff --git a/control_plane/src/sketch_algebra/physical_expr.rs b/control_plane/src/sketch_algebra/physical_expr.rs
index 593d868b..90a24b03 100644
--- a/control_plane/src/sketch_algebra/physical_expr.rs
+++ b/control_plane/src/sketch_algebra/physical_expr.rs
@@ -325,6 +325,7 @@ mod tests {
         QueryExpr::Aggregate {
             by: vec![],
             aggs: vec![AggIntent::Quantile {
+                col: None,
                 q: 0.99,
                 accuracy: AccuracyTarget::Epsilon(0.01),
             }],
diff --git a/control_plane/src/sketch_algebra/rules/bind_archive_only.rs b/control_plane/src/sketch_algebra/rules/bind_archive_only.rs
index b6dbfd57..1747d7dc 100644
--- a/control_plane/src/sketch_algebra/rules/bind_archive_only.rs
+++ b/control_plane/src/sketch_algebra/rules/bind_archive_only.rs
@@ -53,7 +53,7 @@ impl Rule for BindArchiveOnly {
             QueryExpr::Aggregate { aggs, .. } => {
                 // Single-intent Aggregate is the canonical Phase β shape;
                 // multi-intent fans out to per-intent rules elsewhere.
-                if aggs.len() == 1 && aggs[0].archive_only() {
+                if aggs.len() == 1 && crate::intent_algebra::archive_only(&aggs[0]) {
                     Some(PhysicalExpr::Logical(expr.clone()))
                 } else {
                     None
@@ -143,34 +143,20 @@ mod tests {
     fn binds_each_archive_only_intent() {
         let intents = vec![
             AggIntent::Absent,
-            AggIntent::Present,
-            AggIntent::Delta {
-                window: Duration::from_secs(60),
-            },
-            AggIntent::Deriv {
-                window: Duration::from_secs(60),
-            },
-            AggIntent::PredictLinear {
-                window: Duration::from_secs(300),
-                ahead: Duration::from_secs(60),
-            },
-            AggIntent::HoltWinters {
-                window: Duration::from_secs(300),
-                smoothing_factor: 0.3,
-                trend_factor: 0.3,
-            },
-            AggIntent::Idelta {
-                window: Duration::from_secs(60),
-            },
-            AggIntent::Irate {
-                window: Duration::from_secs(60),
-            },
-            AggIntent::Resets {
-                window: Duration::from_secs(300),
-            },
-            AggIntent::Changes {
-                window: Duration::from_secs(300),
+            AggIntent::AbsentOverTime,
+            AggIntent::PresentOverTime,
+            AggIntent::Delta,
+            AggIntent::Deriv,
+            AggIntent::PredictLinear { seconds: 60.0 },
+            AggIntent::DoubleExpSmoothing {
+                smoothing: 0.3,
+                trend: 0.3,
             },
+            AggIntent::IDelta,
+            AggIntent::Resets,
+            AggIntent::Changes,
+            AggIntent::HistogramCount,
+            AggIntent::Group,
         ];
         for intent in intents {
             let expr = agg_with(intent.clone());
@@ -187,27 +173,23 @@ mod tests {
         // Sum / Quantile / Cardinality / TopK are NOT archive-only — they
         // must NOT trigger BindArchiveOnly (the ASAP-tier rules own them).
         for intent in [
-            AggIntent::Sum,
+            AggIntent::Sum { col: None },
             AggIntent::Quantile {
+                col: None,
                 q: 0.99,
                 accuracy: AccuracyTarget::Epsilon(0.01),
             },
             AggIntent::Cardinality {
+                col: None,
                 accuracy: AccuracyTarget::Epsilon(0.01),
             },
             AggIntent::TopK {
                 k: 10,
                 accuracy: AccuracyTarget::Epsilon(0.05),
             },
-            AggIntent::Frequency {
-                accuracy: AccuracyTarget::Epsilon(0.01),
-            },
-            AggIntent::Rate {
-                window: Duration::from_secs(60),
-            },
-            AggIntent::Increase {
-                window: Duration::from_secs(60),
-            },
+            crate::intent_algebra::frequency(AccuracyTarget::Epsilon(0.01)),
+            AggIntent::Rate,
+            AggIntent::Increase,
         ] {
             let expr = agg_with(intent.clone());
             assert!(
diff --git a/control_plane/src/sketch_algebra/rules/bind_cms_count.rs b/control_plane/src/sketch_algebra/rules/bind_cms_count.rs
index 76f02d5f..9cac62e7 100644
--- a/control_plane/src/sketch_algebra/rules/bind_cms_count.rs
+++ b/control_plane/src/sketch_algebra/rules/bind_cms_count.rs
@@ -9,7 +9,7 @@
 //! - `AggIntent::Frequency{accuracy}` — `count(*) WHERE key = k`. CMS is
 //!   the textbook fit (Cormode-Muthukrishnan).
 //!
-//! Accuracy mapping: `AccuracyTarget::EpsilonDelta { eps, delta }` →
+//! Accuracy mapping: `AccuracyTarget::EpsilonDelta { epsilon: eps, delta }` →
 //! `(w, d) = (⌈e/eps⌉, ⌈ln(1/delta)⌉)`. CMS guarantees additive error
 //! `≤ eps · ‖f‖₁` with probability `≥ 1 − delta` (see
 //! `accuracy_profile.rs` in ASAPQuery-backend for the formal bound).
@@ -56,8 +56,8 @@ impl Rule for BindCmsOnCount {
                         child,
                     )
                 }
-                AggIntent::Frequency { accuracy } => (
-                    accuracy.clone(),
+                intent if crate::intent_algebra::as_frequency(intent).is_some() => (
+                    crate::intent_algebra::as_frequency(intent).unwrap(),
                     EstimateOp::PointCount { key: "*".into() },
                     child,
                 ),
@@ -71,13 +71,29 @@ impl Rule for BindCmsOnCount {
         let (eps, delta) = match (accuracy, &intent_accuracy) {
             (AccuracyTarget::Exact, _) | (_, AccuracyTarget::Exact) => return None,
             (AccuracyTarget::Epsilon(a), AccuracyTarget::Epsilon(b)) => (a.min(*b), 0.01),
-            (AccuracyTarget::Epsilon(a), AccuracyTarget::EpsilonDelta { eps, delta })
-            | (AccuracyTarget::EpsilonDelta { eps, delta }, AccuracyTarget::Epsilon(a)) => {
-                (a.min(*eps), *delta)
-            }
             (
-                AccuracyTarget::EpsilonDelta { eps: a, delta: da },
-                AccuracyTarget::EpsilonDelta { eps: b, delta: db },
+                AccuracyTarget::Epsilon(a),
+                AccuracyTarget::EpsilonDelta {
+                    epsilon: eps,
+                    delta,
+                },
+            )
+            | (
+                AccuracyTarget::EpsilonDelta {
+                    epsilon: eps,
+                    delta,
+                },
+                AccuracyTarget::Epsilon(a),
+            ) => (a.min(*eps), *delta),
+            (
+                AccuracyTarget::EpsilonDelta {
+                    epsilon: a,
+                    delta: da,
+                },
+                AccuracyTarget::EpsilonDelta {
+                    epsilon: b,
+                    delta: db,
+                },
             ) => (a.min(*b), da.min(*db)),
         };
 
diff --git a/control_plane/src/sketch_algebra/rules/bind_cms_topk.rs b/control_plane/src/sketch_algebra/rules/bind_cms_topk.rs
index 6b492a0a..bdef52c6 100644
--- a/control_plane/src/sketch_algebra/rules/bind_cms_topk.rs
+++ b/control_plane/src/sketch_algebra/rules/bind_cms_topk.rs
@@ -185,8 +185,8 @@ impl BindCountSketchOnTopK {
         // in play. Note we no longer bail to `None` on `Exact`: an exact
         // top-k intent picks the unbiased CountSketch family (the closest
         // sketch-tier approximation) rather than declining the binding.
-        let tier =
-            forced_tier.unwrap_or_else(|| TopkRecallTier::from_accuracy(accuracy, &intent_accuracy));
+        let tier = forced_tier
+            .unwrap_or_else(|| TopkRecallTier::from_accuracy(accuracy, &intent_accuracy));
 
         // Derive (w, d) from the frequency-error budget. `Exact` on either
         // side leaves the ε/δ unspecified (it's a *recall* tier signal,
@@ -196,17 +196,36 @@ impl BindCountSketchOnTopK {
             (AccuracyTarget::Exact, AccuracyTarget::Exact) => (0.01, 0.01),
             (AccuracyTarget::Exact, other) | (other, AccuracyTarget::Exact) => match other {
                 AccuracyTarget::Epsilon(a) => (*a, 0.01),
-                AccuracyTarget::EpsilonDelta { eps, delta } => (*eps, *delta),
+                AccuracyTarget::EpsilonDelta {
+                    epsilon: eps,
+                    delta,
+                } => (*eps, *delta),
                 AccuracyTarget::Exact => (0.01, 0.01),
             },
             (AccuracyTarget::Epsilon(a), AccuracyTarget::Epsilon(b)) => (a.min(*b), 0.01),
-            (AccuracyTarget::Epsilon(a), AccuracyTarget::EpsilonDelta { eps, delta })
-            | (AccuracyTarget::EpsilonDelta { eps, delta }, AccuracyTarget::Epsilon(a)) => {
-                (a.min(*eps), *delta)
-            }
             (
-                AccuracyTarget::EpsilonDelta { eps: a, delta: da },
-                AccuracyTarget::EpsilonDelta { eps: b, delta: db },
+                AccuracyTarget::Epsilon(a),
+                AccuracyTarget::EpsilonDelta {
+                    epsilon: eps,
+                    delta,
+                },
+            )
+            | (
+                AccuracyTarget::EpsilonDelta {
+                    epsilon: eps,
+                    delta,
+                },
+                AccuracyTarget::Epsilon(a),
+            ) => (a.min(*eps), *delta),
+            (
+                AccuracyTarget::EpsilonDelta {
+                    epsilon: a,
+                    delta: da,
+                },
+                AccuracyTarget::EpsilonDelta {
+                    epsilon: b,
+                    delta: db,
+                },
             ) => (a.min(*b), da.min(*db)),
         };
 
diff --git a/control_plane/src/sketch_algebra/rules/bind_ddsketch_quantile.rs b/control_plane/src/sketch_algebra/rules/bind_ddsketch_quantile.rs
index 9ec73ba7..8773de9d 100644
--- a/control_plane/src/sketch_algebra/rules/bind_ddsketch_quantile.rs
+++ b/control_plane/src/sketch_algebra/rules/bind_ddsketch_quantile.rs
@@ -46,7 +46,7 @@ impl Rule for BindDDSketchOnQuantile {
             QueryExpr::Aggregate {
                 aggs, child, by, ..
             } if aggs.len() == 1 && by.is_empty() => match &aggs[0] {
-                AggIntent::Quantile { q, accuracy } => (*q, accuracy.clone(), child),
+                AggIntent::Quantile { q, accuracy, .. } => (*q, accuracy.clone(), child),
                 _ => return None,
             },
             _ => return None,
@@ -62,11 +62,13 @@ impl Rule for BindDDSketchOnQuantile {
         let alpha = match (accuracy, &intent_accuracy) {
             (AccuracyTarget::Exact, _) | (_, AccuracyTarget::Exact) => return None,
             (AccuracyTarget::Epsilon(a), AccuracyTarget::Epsilon(b)) => a.min(*b),
-            (AccuracyTarget::Epsilon(a), AccuracyTarget::EpsilonDelta { eps, .. })
-            | (AccuracyTarget::EpsilonDelta { eps, .. }, AccuracyTarget::Epsilon(a)) => a.min(*eps),
+            (AccuracyTarget::Epsilon(a), AccuracyTarget::EpsilonDelta { epsilon: eps, .. })
+            | (AccuracyTarget::EpsilonDelta { epsilon: eps, .. }, AccuracyTarget::Epsilon(a)) => {
+                a.min(*eps)
+            }
             (
-                AccuracyTarget::EpsilonDelta { eps: a, .. },
-                AccuracyTarget::EpsilonDelta { eps: b, .. },
+                AccuracyTarget::EpsilonDelta { epsilon: a, .. },
+                AccuracyTarget::EpsilonDelta { epsilon: b, .. },
             ) => a.min(*b),
         };
 
diff --git a/control_plane/src/sketch_algebra/rules/bind_exact_agg.rs b/control_plane/src/sketch_algebra/rules/bind_exact_agg.rs
index 3d5819cd..09f215cf 100644
--- a/control_plane/src/sketch_algebra/rules/bind_exact_agg.rs
+++ b/control_plane/src/sketch_algebra/rules/bind_exact_agg.rs
@@ -90,20 +90,27 @@ impl Rule for BindExactAgg {
         // the query can fan results out over the surviving labels.
         // Unkeyed aggregations stay on the single-pop variant.
         let agg_type = match intent {
-            AggIntent::Sum => {
+            AggIntent::Sum { .. } => {
                 if keyed {
                     AggregationType::MultipleSum
                 } else {
                     AggregationType::Sum
                 }
             }
-            AggIntent::Rate { window } | AggIntent::Increase { window } => {
-                // The window is informational here — the data plane keys
-                // the policy on (metric, attrs, agg_kind, filter) plus
-                // the per-policy `window_size`. Empty windows are
-                // semantically meaningless; reject them so the rule
-                // doesn't fire on a malformed L3 input.
-                if *window == Duration::ZERO {
+            AggIntent::Rate | AggIntent::Increase => {
+                // The window used to live on the intent (`Rate { window
+                // }`); after the ASAPController IR merge it lives on the
+                // enclosing `QueryExpr::Window` node instead. The data
+                // plane keys the policy on (metric, attrs, agg_kind,
+                // filter) plus the per-policy `window_size` -- the value
+                // is informational here, not consumed further. Empty
+                // windows are semantically meaningless; reject them so
+                // the rule doesn't fire on a malformed L3 input.
+                let window = match &**child {
+                    QueryExpr::Window { size, .. } => *size,
+                    _ => Duration::ZERO,
+                };
+                if window == Duration::ZERO {
                     return None;
                 }
                 if keyed {
@@ -153,6 +160,24 @@ mod tests {
         }
     }
 
+    /// Like `agg_over`, but wraps the scan in a `QueryExpr::Window` --
+    /// `Rate`/`Increase` read their window off this enclosing node, not
+    /// off the intent (post ASAPController IR merge; the intent used to
+    /// carry `window: Duration` itself).
+    fn windowed_agg_over(intent: AggIntent, metric: &str, window: Duration) -> QueryExpr {
+        QueryExpr::Aggregate {
+            aggs: vec![intent],
+            child: Box::new(QueryExpr::Window {
+                kind: crate::intent_algebra::WindowKind::Sliding,
+                size: window,
+                slide: None,
+                child: Box::new(scan(metric)),
+            }),
+            by: vec![],
+            having: None,
+        }
+    }
+
     fn check_binds(intent: AggIntent, expected: AggregationType) {
         let expr = agg_over(intent, "test_metric");
         let bound = BindExactAgg
@@ -164,27 +189,36 @@ mod tests {
         }
     }
 
+    fn check_binds_windowed(intent: AggIntent, window: Duration, expected: AggregationType) {
+        let expr = windowed_agg_over(intent, "test_metric", window);
+        let bound = BindExactAgg
+            .apply(&expr, &AccuracyTarget::Exact)
+            .unwrap_or_else(|| panic!("rule didn't fire on {expected:?}"));
+        match bound {
+            PhysicalExpr::ExactAgg { agg_type, .. } => assert_eq!(agg_type, expected),
+            other => panic!("expected ExactAgg, got {other:?}"),
+        }
+    }
+
     #[test]
     fn binds_sum_to_exact_agg_sum() {
-        check_binds(AggIntent::Sum, AggregationType::Sum);
+        check_binds(AggIntent::Sum { col: None }, AggregationType::Sum);
     }
 
     #[test]
     fn binds_rate_to_exact_agg_increase() {
-        check_binds(
-            AggIntent::Rate {
-                window: Duration::from_secs(60),
-            },
+        check_binds_windowed(
+            AggIntent::Rate,
+            Duration::from_secs(60),
             AggregationType::Increase,
         );
     }
 
     #[test]
     fn binds_increase_to_exact_agg_increase() {
-        check_binds(
-            AggIntent::Increase {
-                window: Duration::from_secs(300),
-            },
+        check_binds_windowed(
+            AggIntent::Increase,
+            Duration::from_secs(300),
             AggregationType::Increase,
         );
     }
@@ -217,7 +251,7 @@ mod tests {
 
     #[test]
     fn does_not_bind_avg() {
-        let expr = agg_over(AggIntent::Avg, "test_metric");
+        let expr = agg_over(AggIntent::Avg { col: None }, "test_metric");
         assert!(BindExactAgg.apply(&expr, &AccuracyTarget::Exact).is_none());
     }
 
@@ -225,6 +259,7 @@ mod tests {
     fn does_not_bind_quantile() {
         let expr = agg_over(
             AggIntent::Quantile {
+                col: None,
                 q: 0.99,
                 accuracy: AccuracyTarget::Epsilon(0.01),
             },
@@ -242,6 +277,25 @@ mod tests {
         }
     }
 
+    fn windowed_agg_over_with_by(
+        intent: AggIntent,
+        metric: &str,
+        window: Duration,
+        by: Vec<usize>,
+    ) -> QueryExpr {
+        QueryExpr::Aggregate {
+            aggs: vec![intent],
+            child: Box::new(QueryExpr::Window {
+                kind: crate::intent_algebra::WindowKind::Sliding,
+                size: window,
+                slide: None,
+                child: Box::new(scan(metric)),
+            }),
+            by,
+            having: None,
+        }
+    }
+
     fn check_keyed_binds(intent: AggIntent, expected: AggregationType) {
         let expr = agg_over_with_by(intent, "test_metric", vec![0]);
         let bound = BindExactAgg
@@ -253,27 +307,36 @@ mod tests {
         }
     }
 
+    fn check_keyed_binds_windowed(intent: AggIntent, window: Duration, expected: AggregationType) {
+        let expr = windowed_agg_over_with_by(intent, "test_metric", window, vec![0]);
+        let bound = BindExactAgg
+            .apply(&expr, &AccuracyTarget::Exact)
+            .unwrap_or_else(|| panic!("rule didn't fire on keyed {expected:?}"));
+        match bound {
+            PhysicalExpr::ExactAgg { agg_type, .. } => assert_eq!(agg_type, expected),
+            other => panic!("expected ExactAgg, got {other:?}"),
+        }
+    }
+
     #[test]
     fn keyed_sum_binds_to_multiple_sum() {
-        check_keyed_binds(AggIntent::Sum, AggregationType::MultipleSum);
+        check_keyed_binds(AggIntent::Sum { col: None }, AggregationType::MultipleSum);
     }
 
     #[test]
     fn keyed_rate_binds_to_multiple_increase() {
-        check_keyed_binds(
-            AggIntent::Rate {
-                window: Duration::from_secs(60),
-            },
+        check_keyed_binds_windowed(
+            AggIntent::Rate,
+            Duration::from_secs(60),
             AggregationType::MultipleIncrease,
         );
     }
 
     #[test]
     fn keyed_increase_binds_to_multiple_increase() {
-        check_keyed_binds(
-            AggIntent::Increase {
-                window: Duration::from_secs(300),
-            },
+        check_keyed_binds_windowed(
+            AggIntent::Increase,
+            Duration::from_secs(300),
             AggregationType::MultipleIncrease,
         );
     }
@@ -296,18 +359,24 @@ mod tests {
     fn unkeyed_sum_still_binds_to_single_pop_sum() {
         // Regression guard: the keyed/unkeyed branch must still
         // dispatch correctly on `by.is_empty()`.
-        check_binds(AggIntent::Sum, AggregationType::Sum);
+        check_binds(AggIntent::Sum { col: None }, AggregationType::Sum);
     }
 
     #[test]
     fn rejects_zero_window_rate() {
         // Defensive — a zero-window Rate is semantically meaningless.
-        let expr = agg_over(
-            AggIntent::Rate {
-                window: Duration::ZERO,
-            },
-            "test_metric",
-        );
+        // `agg_over` (unwindowed child) exercises this directly: no
+        // enclosing `QueryExpr::Window` node means the window read
+        // defaults to `Duration::ZERO`.
+        let expr = agg_over(AggIntent::Rate, "test_metric");
+        assert!(BindExactAgg.apply(&expr, &AccuracyTarget::Exact).is_none());
+    }
+
+    #[test]
+    fn rejects_explicit_zero_window_rate() {
+        // Same rejection when the Window node is present but its size is
+        // literally zero (as opposed to no Window node at all).
+        let expr = windowed_agg_over(AggIntent::Rate, "test_metric", Duration::ZERO);
         assert!(BindExactAgg.apply(&expr, &AccuracyTarget::Exact).is_none());
     }
 
diff --git a/control_plane/src/sketch_algebra/rules/bind_hll_cardinality.rs b/control_plane/src/sketch_algebra/rules/bind_hll_cardinality.rs
index 17beb822..6fe4b700 100644
--- a/control_plane/src/sketch_algebra/rules/bind_hll_cardinality.rs
+++ b/control_plane/src/sketch_algebra/rules/bind_hll_cardinality.rs
@@ -36,7 +36,7 @@ impl Rule for BindHllOnCardinality {
             QueryExpr::Aggregate {
                 aggs, child, by, ..
             } if aggs.len() == 1 && by.is_empty() => match &aggs[0] {
-                AggIntent::Cardinality { accuracy } => (accuracy.clone(), child),
+                AggIntent::Cardinality { accuracy, .. } => (accuracy.clone(), child),
                 _ => return None,
             },
             _ => return None,
@@ -46,11 +46,13 @@ impl Rule for BindHllOnCardinality {
         let eps = match (accuracy, &intent_accuracy) {
             (AccuracyTarget::Exact, _) | (_, AccuracyTarget::Exact) => return None,
             (AccuracyTarget::Epsilon(a), AccuracyTarget::Epsilon(b)) => a.min(*b),
-            (AccuracyTarget::Epsilon(a), AccuracyTarget::EpsilonDelta { eps, .. })
-            | (AccuracyTarget::EpsilonDelta { eps, .. }, AccuracyTarget::Epsilon(a)) => a.min(*eps),
+            (AccuracyTarget::Epsilon(a), AccuracyTarget::EpsilonDelta { epsilon: eps, .. })
+            | (AccuracyTarget::EpsilonDelta { epsilon: eps, .. }, AccuracyTarget::Epsilon(a)) => {
+                a.min(*eps)
+            }
             (
-                AccuracyTarget::EpsilonDelta { eps: a, .. },
-                AccuracyTarget::EpsilonDelta { eps: b, .. },
+                AccuracyTarget::EpsilonDelta { epsilon: a, .. },
+                AccuracyTarget::EpsilonDelta { epsilon: b, .. },
             ) => a.min(*b),
         };
 
diff --git a/control_plane/src/sketch_algebra/rules/bind_kll_quantile.rs b/control_plane/src/sketch_algebra/rules/bind_kll_quantile.rs
index 29deb094..209275c9 100644
--- a/control_plane/src/sketch_algebra/rules/bind_kll_quantile.rs
+++ b/control_plane/src/sketch_algebra/rules/bind_kll_quantile.rs
@@ -48,7 +48,7 @@ impl Rule for BindKllOnQuantile {
             QueryExpr::Aggregate {
                 aggs, child, by, ..
             } if aggs.len() == 1 && by.is_empty() => match &aggs[0] {
-                AggIntent::Quantile { q, accuracy } => (*q, accuracy.clone(), child),
+                AggIntent::Quantile { q, accuracy, .. } => (*q, accuracy.clone(), child),
                 _ => return None,
             },
             _ => return None,
@@ -67,11 +67,13 @@ impl Rule for BindKllOnQuantile {
         let eps = match (accuracy, &intent_accuracy) {
             (AccuracyTarget::Exact, _) | (_, AccuracyTarget::Exact) => return None,
             (AccuracyTarget::Epsilon(a), AccuracyTarget::Epsilon(b)) => a.min(*b),
-            (AccuracyTarget::Epsilon(a), AccuracyTarget::EpsilonDelta { eps, .. })
-            | (AccuracyTarget::EpsilonDelta { eps, .. }, AccuracyTarget::Epsilon(a)) => a.min(*eps),
+            (AccuracyTarget::Epsilon(a), AccuracyTarget::EpsilonDelta { epsilon: eps, .. })
+            | (AccuracyTarget::EpsilonDelta { epsilon: eps, .. }, AccuracyTarget::Epsilon(a)) => {
+                a.min(*eps)
+            }
             (
-                AccuracyTarget::EpsilonDelta { eps: a, .. },
-                AccuracyTarget::EpsilonDelta { eps: b, .. },
+                AccuracyTarget::EpsilonDelta { epsilon: a, .. },
+                AccuracyTarget::EpsilonDelta { epsilon: b, .. },
             ) => a.min(*b),
         };
 
diff --git a/control_plane/src/sketch_algebra/tests.rs b/control_plane/src/sketch_algebra/tests.rs
index 44b9e83b..8a50a65f 100644
--- a/control_plane/src/sketch_algebra/tests.rs
+++ b/control_plane/src/sketch_algebra/tests.rs
@@ -57,7 +57,11 @@ fn windowed_scan() -> QueryExpr {
 fn agg_quantile(q: f64, accuracy: AccuracyTarget) -> QueryExpr {
     QueryExpr::Aggregate {
         by: vec![],
-        aggs: vec![AggIntent::Quantile { q, accuracy }],
+        aggs: vec![AggIntent::Quantile {
+            col: None,
+            q,
+            accuracy,
+        }],
         having: None,
         child: Box::new(windowed_scan()),
     }
@@ -261,7 +265,7 @@ fn topk_binding_family(bound: &PhysicalExpr) -> (SketchKind, bool, u32, u32) {
 #[test]
 fn bind_cms_topk_loose_recall_picks_cms_heap() {
     let acc = AccuracyTarget::EpsilonDelta {
-        eps: 0.01,
+        epsilon: 0.01,
         delta: 0.001,
     };
     let expr = agg_topk(10, acc.clone());
@@ -324,7 +328,11 @@ fn bind_cms_topk_picks_cost_min_meeting_sla() {
     let bound = bind_query_expr(&agg_topk(10, acc.clone()), acc).unwrap();
     let (kind, ..) = topk_binding_family(&bound);
     let chosen = table.for_kind(&kind).per_flush();
-    assert_eq!(chosen, cms.min(cs), "must pick the cost-min family that meets the SLA");
+    assert_eq!(
+        chosen,
+        cms.min(cs),
+        "must pick the cost-min family that meets the SLA"
+    );
     assert_eq!(kind, SketchKind::Cms);
 }
 
@@ -333,6 +341,7 @@ fn bind_hll_cardinality_basic() {
     let expr = QueryExpr::Aggregate {
         by: vec![],
         aggs: vec![AggIntent::Cardinality {
+            col: None,
             accuracy: AccuracyTarget::Epsilon(0.01),
         }],
         having: None,
@@ -375,7 +384,7 @@ fn sum_now_binds_to_exact_agg_after_pr_6_followup() {
     // exact-aggregation path can serve the intent.
     let expr = QueryExpr::Aggregate {
         by: vec![],
-        aggs: vec![AggIntent::Sum],
+        aggs: vec![AggIntent::Sum { col: None }],
         having: None,
         child: Box::new(windowed_scan()),
     };
@@ -462,6 +471,7 @@ fn phase_b_pattern_only_temporal_quantile_binds_to_sketch() {
     let expr = QueryExpr::Aggregate {
         by: vec![],
         aggs: vec![AggIntent::Quantile {
+            col: None,
             q: 0.99,
             accuracy: AccuracyTarget::Epsilon(0.01),
         }],
@@ -497,7 +507,7 @@ fn phase_b_pattern_only_temporal_quantile_binds_to_sketch() {
 fn phase_b_pattern_only_temporal_sum_binds_to_exact_agg() {
     let expr = QueryExpr::Aggregate {
         by: vec![],
-        aggs: vec![AggIntent::Sum],
+        aggs: vec![AggIntent::Sum { col: None }],
         having: None,
         child: Box::new(windowed_scan()),
     };
@@ -521,7 +531,7 @@ fn phase_b_pattern_only_temporal_sum_binds_to_exact_agg() {
 fn phase_b_pattern_only_spatial_aggregate_binds_to_multiple_sum() {
     let expr = QueryExpr::Aggregate {
         by: vec![1], // service column
-        aggs: vec![AggIntent::Sum],
+        aggs: vec![AggIntent::Sum { col: None }],
         having: None,
         child: Box::new(ts_scan()),
     };
@@ -542,9 +552,7 @@ fn phase_b_pattern_only_spatial_aggregate_binds_to_multiple_sum() {
 fn phase_b_pattern_temporal_and_spatial_combined_binds_to_multiple_increase() {
     let expr = QueryExpr::Aggregate {
         by: vec![1],
-        aggs: vec![AggIntent::Rate {
-            window: Duration::from_secs(300),
-        }],
+        aggs: vec![AggIntent::Rate],
         having: None,
         child: Box::new(windowed_scan()),
     };
@@ -573,7 +581,10 @@ fn phase_b_pattern_temporal_and_spatial_combined_binds_to_multiple_increase() {
 #[test]
 fn phase_b_pattern_archive_only_routes_to_archive() {
     let intent = AggIntent::Absent;
-    assert!(intent.archive_only(), "Phase β intent must flag archive");
+    assert!(
+        crate::intent_algebra::archive_only(&intent),
+        "Phase β intent must flag archive"
+    );
     let expr = QueryExpr::Aggregate {
         by: vec![],
         aggs: vec![intent.clone()],
@@ -664,7 +675,7 @@ fn collect_sketch_kinds(expr: &PhysicalExpr) -> Vec<SketchKind> {
 fn binding_is_archive(expr: &PhysicalExpr) -> bool {
     match expr {
         PhysicalExpr::Logical(QueryExpr::Aggregate { aggs, .. }) => {
-            aggs.iter().any(|a| a.archive_only())
+            aggs.iter().any(crate::intent_algebra::archive_only)
         }
         PhysicalExpr::Logical(_) => false,
         PhysicalExpr::SketchEstimate { child, .. } => binding_is_archive(child),
@@ -843,34 +854,22 @@ fn phase_b_e2e_archive_only_e2e_binding() {
 fn phase_b_archive_only_intents_round_trip_through_binder() {
     let intents = vec![
         AggIntent::Absent,
-        AggIntent::Present,
-        AggIntent::Delta {
-            window: Duration::from_secs(60),
-        },
-        AggIntent::Deriv {
-            window: Duration::from_secs(60),
-        },
-        AggIntent::PredictLinear {
-            window: Duration::from_secs(300),
-            ahead: Duration::from_secs(60),
-        },
-        AggIntent::HoltWinters {
-            window: Duration::from_secs(300),
-            smoothing_factor: 0.3,
-            trend_factor: 0.3,
-        },
-        AggIntent::Idelta {
-            window: Duration::from_secs(60),
-        },
-        AggIntent::Irate {
-            window: Duration::from_secs(60),
-        },
-        AggIntent::Resets {
-            window: Duration::from_secs(300),
-        },
-        AggIntent::Changes {
-            window: Duration::from_secs(300),
+        AggIntent::AbsentOverTime,
+        AggIntent::PresentOverTime,
+        AggIntent::Delta,
+        AggIntent::Deriv,
+        AggIntent::PredictLinear { seconds: 60.0 },
+        AggIntent::DoubleExpSmoothing {
+            smoothing: 0.3,
+            trend: 0.3,
         },
+        AggIntent::IDelta,
+        AggIntent::Resets,
+        AggIntent::Changes,
+        // Spot-check a couple of the Phase 1 IR merge's new archive-only
+        // intents through the same round-trip.
+        AggIntent::HistogramCount,
+        AggIntent::Group,
     ];
     for intent in intents {
         let expr = QueryExpr::Aggregate {
@@ -885,7 +884,7 @@ fn phase_b_archive_only_intents_round_trip_through_binder() {
             PhysicalExpr::Logical(QueryExpr::Aggregate { aggs, .. }) => {
                 assert_eq!(aggs.len(), 1);
                 assert!(
-                    aggs[0].archive_only(),
+                    crate::intent_algebra::archive_only(&aggs[0]),
                     "{intent:?} should preserve archive_only() flag through bind"
                 );
             }
diff --git a/control_plane/src/types_v2.rs b/control_plane/src/types_v2.rs
index f1a80836..5cad49aa 100644
--- a/control_plane/src/types_v2.rs
+++ b/control_plane/src/types_v2.rs
@@ -48,47 +48,31 @@ pub enum QueryLanguage {
 
 /// Per-target accuracy SLA, in the typed form `design.md` §6 calls for.
 ///
-/// Drives L4 sketch binding (`Exact` disables every `Bind*` rule, so the
-/// optimiser falls back to an exact `HashAgg` / `SortAgg`; `Epsilon` and
-/// `EpsilonDelta` set the ε / δ budget the cost model has to satisfy when
-/// it picks a sketch family + parameters).
-///
-/// The legacy `analyzer::QuerySpec.accuracy_sla: f64` field is preserved
-/// for back-compat — when a caller supplies a typed `accuracy: Some(…)`
-/// it takes precedence; otherwise the analyzer translates the legacy
-/// fraction to `Epsilon(1.0 - accuracy_sla)`.
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
-#[serde(tag = "kind", content = "value", rename_all = "snake_case")]
-pub enum AccuracyTarget {
-    /// No approximation allowed. L4 must pick an exact path; sketch
-    /// binding rules are skipped.
-    Exact,
-    /// Bound on relative error. The L4 cost model must pick sketch
-    /// parameters that satisfy `error ≤ eps` with whatever default
-    /// confidence the sketch family provides.
-    Epsilon(f64),
-    /// Bound on relative error and the probability of exceeding it
-    /// (Pr[error > eps] ≤ delta). Required for sketches whose
-    /// guarantees are inherently probabilistic (CMS, HLL).
-    EpsilonDelta {
-        /// Relative-error bound.
-        eps: f64,
-        /// Probability of exceeding the bound.
-        delta: f64,
-    },
-}
-
-impl AccuracyTarget {
-    /// Translate the legacy `accuracy_sla: f64` field — a fractional
-    /// "1.0 = exact, 0.0 = anything goes" SLA — into the typed form.
-    /// `accuracy_sla == 1.0` round-trips to `Exact`; everything else
-    /// becomes `Epsilon(1.0 - accuracy_sla)` (the implied error bound).
-    pub fn from_legacy_accuracy_sla(accuracy_sla: f64) -> Self {
-        if accuracy_sla >= 1.0 {
-            AccuracyTarget::Exact
-        } else {
-            AccuracyTarget::Epsilon((1.0 - accuracy_sla).max(0.0))
-        }
+/// Phase 1b (docs/migration-plan-backend-plan.md): re-exported from
+/// `asap_ir::types` rather than defined locally -- `AggIntent`'s
+/// `accuracy` fields are typed against ASAPController's `AccuracyTarget`,
+/// so keeping a separate local type here would force a conversion at
+/// every one of the ~400 `AggIntent` call sites. Two real differences
+/// from the pre-merge local type, both confirmed safe to fold on
+/// (no external YAML/JSON persists the old wire shape -- only one
+/// in-Rust test fixture, `pipeline.rs`, needed updating):
+/// - Wire shape: was `#[serde(tag = "kind", content = "value")]`
+///   (`{"kind": "epsilon", "value": 0.02}`); now serde's default
+///   externally-tagged representation (`{"Epsilon": 0.02}`).
+/// - `EpsilonDelta`'s second field is `epsilon`, not `eps`.
+pub use asap_ir::types::AccuracyTarget;
+
+/// Translate the legacy `accuracy_sla: f64` field -- a fractional
+/// "1.0 = exact, 0.0 = anything goes" SLA -- into the typed form.
+/// `accuracy_sla == 1.0` round-trips to `Exact`; everything else becomes
+/// `Epsilon(1.0 - accuracy_sla)` (the implied error bound). Free function,
+/// not `impl AccuracyTarget` -- Rust's orphan rules don't allow inherent
+/// impls on a foreign type.
+pub fn accuracy_target_from_legacy_accuracy_sla(accuracy_sla: f64) -> AccuracyTarget {
+    if accuracy_sla >= 1.0 {
+        AccuracyTarget::Exact
+    } else {
+        AccuracyTarget::Epsilon((1.0 - accuracy_sla).max(0.0))
     }
 }
 
@@ -276,7 +260,7 @@ mod tests {
             AccuracyTarget::Exact,
             AccuracyTarget::Epsilon(0.05),
             AccuracyTarget::EpsilonDelta {
-                eps: 0.01,
+                epsilon: 0.01,
                 delta: 0.001,
             },
         ];
@@ -291,11 +275,11 @@ mod tests {
     fn accuracy_target_from_legacy() {
         // 1.0 means exact in the legacy schema.
         assert_eq!(
-            AccuracyTarget::from_legacy_accuracy_sla(1.0),
+            accuracy_target_from_legacy_accuracy_sla(1.0),
             AccuracyTarget::Exact
         );
         // 0.99 SLA → ε = 0.01.
-        match AccuracyTarget::from_legacy_accuracy_sla(0.99) {
+        match accuracy_target_from_legacy_accuracy_sla(0.99) {
             AccuracyTarget::Epsilon(eps) => {
                 assert!((eps - 0.01).abs() < 1e-9, "got eps={eps}");
             }
@@ -303,7 +287,7 @@ mod tests {
         }
         // Out-of-range guard — analyzer rejects these upstream, but the
         // helper itself must not panic on a 0.0 SLA.
-        match AccuracyTarget::from_legacy_accuracy_sla(0.0) {
+        match accuracy_target_from_legacy_accuracy_sla(0.0) {
             AccuracyTarget::Epsilon(eps) => assert!((eps - 1.0).abs() < 1e-9),
             other => panic!("expected Epsilon, got {other:?}"),
         }