diff --git a/.github/workflows/mvp-ci.yml b/.github/workflows/mvp-ci.yml
index b06cb0e31..fdd9896e1 100644
--- a/.github/workflows/mvp-ci.yml
+++ b/.github/workflows/mvp-ci.yml
@@ -34,6 +34,8 @@ jobs:
         uses: actions/checkout@v4
         with:
           repository: ProjectASAP/asap_sketchlib
+          # PR #139: standard-update compatibility check used by UnivMon restore.
+          ref: c0de315754f9a6c77dd7a25aca0b2b62f0aec276
           path: asap_sketchlib
 
       - name: Install Rust components
diff --git a/Cargo.lock b/Cargo.lock
index f1cb47201..7d5b88241 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -373,7 +373,7 @@ dependencies = [
 [[package]]
 name = "asap-aware-mapping"
 version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=890504025e4f780880cc6c89d24edd6591af868a#890504025e4f780880cc6c89d24edd6591af868a"
+source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc#7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc"
 dependencies = [
  "asap-types",
  "serde",
@@ -384,7 +384,7 @@ dependencies = [
 [[package]]
 name = "asap-frontend-promql"
 version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=890504025e4f780880cc6c89d24edd6591af868a#890504025e4f780880cc6c89d24edd6591af868a"
+source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc#7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc"
 dependencies = [
  "asap-types",
  "promql-parser",
@@ -393,7 +393,7 @@ dependencies = [
 [[package]]
 name = "asap-frontend-sql"
 version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=890504025e4f780880cc6c89d24edd6591af868a#890504025e4f780880cc6c89d24edd6591af868a"
+source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc#7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc"
 dependencies = [
  "asap-sql-function-catalog",
  "asap-types",
@@ -415,12 +415,12 @@ dependencies = [
 [[package]]
 name = "asap-sql-function-catalog"
 version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=890504025e4f780880cc6c89d24edd6591af868a#890504025e4f780880cc6c89d24edd6591af868a"
+source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc#7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc"
 
 [[package]]
 name = "asap-types"
 version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=890504025e4f780880cc6c89d24edd6591af868a#890504025e4f780880cc6c89d24edd6591af868a"
+source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc#7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc"
 dependencies = [
  "serde",
  "serde_json",
@@ -970,6 +970,7 @@ dependencies = [
  "tokio",
  "tokio-stream",
  "tokio-tungstenite 0.21.0",
+ "toml",
  "tonic",
  "tonic-build",
  "tower 0.4.13",
@@ -1687,7 +1688,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
 dependencies = [
  "libc",
- "windows-sys 0.52.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -2246,7 +2247,7 @@ dependencies = [
  "libc",
  "percent-encoding",
  "pin-project-lite",
- "socket2 0.5.10",
+ "socket2 0.6.3",
  "tokio",
  "tower-service",
  "tracing",
@@ -2439,7 +2440,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
 dependencies = [
  "hermit-abi 0.5.2",
  "libc",
- "windows-sys 0.52.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -3258,7 +3259,7 @@ dependencies = [
 [[package]]
 name = "promql-parser"
 version = "0.10.0"
-source = "git+https://github.com/ProjectASAP/promql-parser?rev=d7606436bde05efe19c5398fe2122faa49062f7d#d7606436bde05efe19c5398fe2122faa49062f7d"
+source = "git+https://github.com/ProjectASAP/promql-parser?rev=9fede7eecca923c9882fe256484d00d37f8706cb#9fede7eecca923c9882fe256484d00d37f8706cb"
 dependencies = [
  "cfgrammar",
  "chrono",
@@ -3285,7 +3286,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf"
 dependencies = [
  "heck 0.5.0",
- "itertools 0.10.5",
+ "itertools 0.13.0",
  "log",
  "multimap",
  "once_cell",
@@ -3305,7 +3306,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d"
 dependencies = [
  "anyhow",
- "itertools 0.10.5",
+ "itertools 0.13.0",
  "proc-macro2",
  "quote",
  "syn 2.0.117",
@@ -3403,7 +3404,7 @@ dependencies = [
  "quinn-udp",
  "rustc-hash",
  "rustls 0.23.40",
- "socket2 0.5.10",
+ "socket2 0.6.3",
  "thiserror 2.0.18",
  "tokio",
  "tracing",
@@ -3440,7 +3441,7 @@ dependencies = [
  "cfg_aliases",
  "libc",
  "once_cell",
- "socket2 0.5.10",
+ "socket2 0.6.3",
  "tracing",
  "windows-sys 0.52.0",
 ]
@@ -3756,7 +3757,7 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys 0.12.1",
- "windows-sys 0.52.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -3947,6 +3948,15 @@ dependencies = [
  "serde_core",
 ]
 
+[[package]]
+name = "serde_spanned"
+version = "0.6.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bf41e0cfaf7226dca15e8197172c295a782857fcb97fad1808a166870dee75a3"
+dependencies = [
+ "serde",
+]
+
 [[package]]
 name = "serde_urlencoded"
 version = "0.7.1"
@@ -4290,7 +4300,7 @@ dependencies = [
  "getrandom 0.4.2",
  "once_cell",
  "rustix 1.1.4",
- "windows-sys 0.52.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -4535,6 +4545,47 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "toml"
+version = "0.8.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362"
+dependencies = [
+ "serde",
+ "serde_spanned",
+ "toml_datetime",
+ "toml_edit",
+]
+
+[[package]]
+name = "toml_datetime"
+version = "0.6.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c"
+dependencies = [
+ "serde",
+]
+
+[[package]]
+name = "toml_edit"
+version = "0.22.27"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a"
+dependencies = [
+ "indexmap 2.14.0",
+ "serde",
+ "serde_spanned",
+ "toml_datetime",
+ "toml_write",
+ "winnow",
+]
+
+[[package]]
+name = "toml_write"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801"
+
 [[package]]
 name = "tonic"
 version = "0.12.3"
@@ -5112,7 +5163,7 @@ version = "0.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
 dependencies = [
- "windows-sys 0.48.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -5328,6 +5379,15 @@ version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
 
+[[package]]
+name = "winnow"
+version = "0.7.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df79d97927682d2fd8adb29682d1140b343be4ac0f08fd68b7765d9c059d3945"
+dependencies = [
+ "memchr",
+]
+
 [[package]]
 name = "winreg"
 version = "0.50.0"
diff --git a/Cargo.toml b/Cargo.toml
index 6efc58309..a902149a1 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -40,7 +40,7 @@ clap = { version = "4.0", features = ["derive", "env"] }
 chrono = { version = "0.4", features = ["serde"] }
 # One parser source for backend parsing. Member crates inherit this immutable
 # ProjectASAP revision through `workspace = true`.
-promql-parser = { git = "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/ProjectASAP/promql-parser", rev = "d7606436bde05efe19c5398fe2122faa49062f7d" }
+promql-parser = { git = "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/ProjectASAP/promql-parser", rev = "9fede7eecca923c9882fe256484d00d37f8706cb" }
 tokio = { version = "1.0", features = ["full"] }
 arc-swap = "1.7"
 
diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml
index 303a2b400..32caa4e0e 100644
--- a/control_plane/Cargo.toml
+++ b/control_plane/Cargo.toml
@@ -76,8 +76,8 @@ asap_types.workspace = true
 # scaffolding, unaware that `data_plane`'s `summary_executor.rs` in *this*
 # repo is a real one. Vendored locally instead of chased upstream -- see
 # `data_plane/src/query_engines/asap_query_engine/summary_exec.rs`.
-planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "890504025e4f780880cc6c89d24edd6591af868a" }
-asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "890504025e4f780880cc6c89d24edd6591af868a" }
+planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc" }
+asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc" }
 
 # L1 adoption (design-target-architecture.md Part B): the PromQL front
 # end itself, replacing control_plane's own query_parser/promql.rs.
@@ -85,8 +85,8 @@ asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev =
 # `planner-types`/`asap-aware-mapping` above -- these three MUST move
 # together (two revs of the same upstream repo's types in one workspace
 # resolve to distinct Rust types that won't unify).
-asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "890504025e4f780880cc6c89d24edd6591af868a" }
-asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "890504025e4f780880cc6c89d24edd6591af868a" }
+asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc" }
+asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc" }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }
@@ -96,5 +96,6 @@ http-body-util = "0.1"
 tokio-tungstenite = "0.21"
 
 [build-dependencies]
+toml = "0.8"
 prost-build = "0.13"
 tonic-build = "0.12"
diff --git a/control_plane/build.rs b/control_plane/build.rs
index 8b028d671..7af15c0b3 100644
--- a/control_plane/build.rs
+++ b/control_plane/build.rs
@@ -29,6 +29,26 @@ fn git_path(repo: &Path, name: &str) -> Option<PathBuf> {
 fn main() -> Result<(), Box<dyn std::error::Error>> {
     let manifest_dir = PathBuf::from(std::env::var("CARGO_MANIFEST_DIR")?);
     let repo = manifest_dir.parent().unwrap_or(&manifest_dir);
+    let lock_path = repo.join("Cargo.lock");
+    println!("cargo:rerun-if-changed={}", lock_path.display());
+    let lock: toml::Value = std::fs::read_to_string(&lock_path)?.parse()?;
+    let planner_source = lock
+        .get("package")
+        .and_then(toml::Value::as_array)
+        .into_iter()
+        .flatten()
+        .find(|package| package.get("name").and_then(toml::Value::as_str) == Some("asap-types"))
+        .and_then(|package| package.get("source"))
+        .and_then(toml::Value::as_str)
+        .ok_or("Cargo.lock has no resolved ASAPPlanner types source")?;
+    let planner_revision = planner_source
+        .rsplit_once('#')
+        .filter(|(_, revision)| {
+            revision.len() == 40 && revision.bytes().all(|b| b.is_ascii_hexdigit())
+        })
+        .map(|(_, revision)| revision)
+        .ok_or("ASAPPlanner source must resolve to a Git commit")?;
+    println!("cargo:rustc-env=ASAPPLANNER_REVISION={planner_revision}");
     let revision = std::env::var("ASAPQUERY_BACKEND_REVISION")
         .ok()
         .or_else(|| git_output(repo, &["rev-parse", "HEAD"]))
diff --git a/control_plane/src/emit/otap.rs b/control_plane/src/emit/otap.rs
index 95a13f353..6d377c011 100644
--- a/control_plane/src/emit/otap.rs
+++ b/control_plane/src/emit/otap.rs
@@ -325,7 +325,7 @@ fn build_asap_sketches_config(sp: &EdgeSketchProcessor, window_secs: Option<u64>
             m.insert("delta".into(), Value::Number(delta.into()));
             m.insert("delta_transmission".into(), Value::Bool(true));
         }
-        SketchParams::Kmv { .. } | SketchParams::Theta { .. } => {
+        SketchParams::UnivMon { .. } | SketchParams::Kmv { .. } | SketchParams::Theta { .. } => {
             unreachable!(
                 "edge sketch processor config requested for a non-sketch or unsupported \
                  SketchAlgorithm; no Bind* rule in this repo produces one"
@@ -342,7 +342,7 @@ fn sketch_algorithm_tag(kind: &SketchAlgorithm) -> &'static str {
         SketchAlgorithm::Hll => "hll",
         SketchAlgorithm::Cms | SketchAlgorithm::CmsWithHeap => "cms",
         SketchAlgorithm::CountSketch | SketchAlgorithm::CountSketchWithHeap => "count_sketch",
-        SketchAlgorithm::Kmv | SketchAlgorithm::Theta => {
+        SketchAlgorithm::UnivMon | SketchAlgorithm::Kmv | SketchAlgorithm::Theta => {
             unreachable!(
                 "edge sketch processor config requested for a non-sketch or unsupported \
                  SketchAlgorithm; no Bind* rule in this repo produces one"
diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs
index 27a6d6477..6d94e1643 100644
--- a/control_plane/src/emit/stage_config.rs
+++ b/control_plane/src/emit/stage_config.rs
@@ -2096,7 +2096,11 @@ fn emit_edge_yaml_asap_edge(
                     e.insert("rows".into(), Value::Number((*depth as u64).into()));
                     e.insert("cols".into(), Value::Number((*width as u64).into()));
                 }
-                Some(SketchParams::Kmv { .. } | SketchParams::Theta { .. }) => unreachable!(
+                Some(
+                    SketchParams::UnivMon { .. }
+                    | SketchParams::Kmv { .. }
+                    | SketchParams::Theta { .. },
+                ) => unreachable!(
                     "5-sketch routing: non-sketch or unsupported SketchParams; \
                      no Bind* rule in this repo produces one"
                 ),
@@ -2860,10 +2864,12 @@ fn build_edge_processor_block(
             m.insert("encoding".into(), Value::String("msgpack".into()));
             m.insert("delta_transmission".into(), Value::Bool(true));
         }
-        SketchParams::Kmv { .. } | SketchParams::Theta { .. } => unreachable!(
-            "edge sketch processor config requested for a non-sketch or unsupported \
+        SketchParams::UnivMon { .. } | SketchParams::Kmv { .. } | SketchParams::Theta { .. } => {
+            unreachable!(
+                "edge sketch processor config requested for a non-sketch or unsupported \
              SketchAlgorithm; no Bind* rule in this repo produces one"
-        ),
+            )
+        }
     }
 
     Value::Mapping(m)
@@ -2903,7 +2909,7 @@ fn gateway_merge_processor_name(mp: &GatewayMergeProcessor) -> String {
         SketchAlgorithm::CountSketch | SketchAlgorithm::CountSketchWithHeap => {
             "countsketchmerge".to_string()
         }
-        SketchAlgorithm::Kmv | SketchAlgorithm::Theta => unreachable!(
+        SketchAlgorithm::UnivMon | SketchAlgorithm::Kmv | SketchAlgorithm::Theta => unreachable!(
             "gateway_merge_processor_name: non-sketch or unsupported SketchAlgorithm; \
              no Bind* rule in this repo produces one"
         ),
@@ -3050,6 +3056,8 @@ pub(crate) fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonVa
 /// `aggregations` list by the same `PolicyFingerprint` recipe.
 fn build_backend_readout_json(r: &BackendReadout) -> JsonValue {
     match &r.op {
+        SketchQuery::FrequencyL2 => json!({"op": "frequency_l2"}),
+        SketchQuery::FrequencyEntropy => json!({"op": "frequency_entropy"}),
         SketchQuery::Quantile { q } => json!({
             "op": "quantile",
             "q": q,
@@ -3121,6 +3129,7 @@ fn base_family(kind: &SketchAlgorithm) -> SketchAlgorithm {
 /// for `topk(...)` queries to bind to the right sids.
 fn sketch_algorithm_to_backend_type(kind: &SketchAlgorithm) -> &'static str {
     match kind {
+        SketchAlgorithm::UnivMon => "UnivMon",
         SketchAlgorithm::DDSketch => "DDSketch",
         SketchAlgorithm::Kll => "DatasketchesKLL",
         SketchAlgorithm::Hll => "HLL",
@@ -3147,7 +3156,7 @@ fn sketch_algorithm_tag(kind: &SketchAlgorithm) -> &'static str {
         SketchAlgorithm::Hll => "hll",
         SketchAlgorithm::Cms | SketchAlgorithm::CmsWithHeap => "cms",
         SketchAlgorithm::CountSketch | SketchAlgorithm::CountSketchWithHeap => "count_sketch",
-        SketchAlgorithm::Kmv | SketchAlgorithm::Theta => unreachable!(
+        SketchAlgorithm::UnivMon | SketchAlgorithm::Kmv | SketchAlgorithm::Theta => unreachable!(
             "sketch_algorithm_tag: non-sketch or unsupported SketchAlgorithm; \
              no Bind* rule in this repo produces one"
         ),
@@ -3159,6 +3168,14 @@ fn sketch_algorithm_tag(kind: &SketchAlgorithm) -> &'static str {
 /// internally-tagged enum form.
 fn sketch_params_to_json(p: &SketchParams) -> JsonValue {
     match p {
+        SketchParams::UnivMon {
+            heap_size,
+            sketch_rows,
+            sketch_cols,
+            layers,
+        } => json!({
+            "heap_size": heap_size, "sketch_rows": sketch_rows, "sketch_cols": sketch_cols, "layers": layers,
+        }),
         SketchParams::Kll { k } => json!({ "k": k }),
         SketchParams::DDSketch { alpha } => json!({ "alpha": alpha }),
         SketchParams::Hll { precision } => json!({ "precision": precision }),
@@ -3191,10 +3208,12 @@ fn sketch_params_to_json(p: &SketchParams) -> JsonValue {
         }),
         // Exact accumulators never reach here -- see
         // `sketch_kind_to_backend_type`'s doc.
-        SketchParams::Kmv { .. } | SketchParams::Theta { .. } => unreachable!(
-            "sketch_params_to_json: non-sketch or unsupported SummaryParams; \
+        SketchParams::Kmv { .. } | SketchParams::Theta { .. } => {
+            unreachable!(
+                "sketch_params_to_json: non-sketch or unsupported SummaryParams; \
              no Bind* rule in this repo produces one"
-        ),
+            )
+        }
     }
 }
 
diff --git a/control_plane/src/emit/telegraf.rs b/control_plane/src/emit/telegraf.rs
index 6fdd3a649..eb06c3d07 100644
--- a/control_plane/src/emit/telegraf.rs
+++ b/control_plane/src/emit/telegraf.rs
@@ -197,7 +197,7 @@ fn emit_processors_allsketches(
         // Exact-accumulator kinds (Sum/Count/MinMax/Increase/Rate) aren't
         // representable in `SketchParams` at all anymore -- they're
         // `ExactParams`, a distinct type post ASAPPlanner#218's split.
-        SketchParams::Kmv { .. } | SketchParams::Theta { .. } => {
+        SketchParams::UnivMon { .. } | SketchParams::Kmv { .. } | SketchParams::Theta { .. } => {
             unreachable!(
                 "edge sketch processor config requested for an unsupported SketchAlgorithm; \
                  no Bind* rule in this repo produces one"
@@ -214,7 +214,7 @@ fn sketch_algorithm_tag(kind: &SketchAlgorithm) -> &'static str {
         SketchAlgorithm::Hll => "hll",
         SketchAlgorithm::Cms | SketchAlgorithm::CmsWithHeap => "cms",
         SketchAlgorithm::CountSketch | SketchAlgorithm::CountSketchWithHeap => "count_sketch",
-        SketchAlgorithm::Kmv | SketchAlgorithm::Theta => {
+        SketchAlgorithm::UnivMon | SketchAlgorithm::Kmv | SketchAlgorithm::Theta => {
             unreachable!(
                 "edge sketch processor config requested for an unsupported \
                  SketchAlgorithm; no Bind* rule in this repo produces one"
diff --git a/control_plane/src/physical/allocator.rs b/control_plane/src/physical/allocator.rs
index 2b070f52c..9b6be02a4 100644
--- a/control_plane/src/physical/allocator.rs
+++ b/control_plane/src/physical/allocator.rs
@@ -774,6 +774,8 @@ fn canonical_intent_kind_str(intent: &AggIntent) -> &'static str {
         AggIntent::Quantile { .. } => "quantile",
         AggIntent::TopK { .. } => "topk",
         AggIntent::Cardinality { .. } => "cardinality",
+        AggIntent::FrequencyL2 { .. } => "frequency_l2",
+        AggIntent::FrequencyEntropy { .. } => "frequency_entropy",
         AggIntent::Rate => "rate",
         AggIntent::IRate => "irate",
         AggIntent::Increase => "increase",
diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs
index 9d230e89a..aa609da0b 100644
--- a/control_plane/src/physical/colored_dag/emitter.rs
+++ b/control_plane/src/physical/colored_dag/emitter.rs
@@ -1085,7 +1085,7 @@ pub(crate) fn edge_processor_name(kind: &SketchAlgorithm) -> Result<String, Emit
         // representable here anymore -- they're `ExactKind`, a distinct
         // type post ASAPPlanner#218's split, not a `SketchAlgorithm` variant
         // this function could even be called with.
-        SketchAlgorithm::Kmv | SketchAlgorithm::Theta => {
+        SketchAlgorithm::UnivMon | SketchAlgorithm::Kmv | SketchAlgorithm::Theta => {
             Err(EmitError::NoEdgeProcessor(kind.clone()))
         }
     }
diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index 7ae535234..460b5e531 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -37,7 +37,7 @@ use crate::query_plan::{
 use crate::types_v2::AccuracyTarget;
 use planner_types::pre_asap::Source;
 
-pub const PLANNER_REVISION: &str = "0deceda3e776216c5542d638d958b159f22e27ce";
+pub const PLANNER_REVISION: &str = env!("ASAPPLANNER_REVISION");
 pub const BACKEND_REVISION: &str = env!("ASAPQUERY_BACKEND_REVISION");
 pub use asap_types::precompute_plan::BACKEND_COMPAT;
 /// Matches the data plane's default persistence memory limit. A backend-local
@@ -1515,11 +1515,16 @@ fn has_unsafe_raw_entity_leaf(
                     )
                 );
                 let scalar_series_input = matches!(&node.expr,
-                    SummaryExpr::SummaryAgg { input, .. }
-                    if input.item.is_none() && matches!(&input.weight, planner_types::post_asap::SummaryInputExpr::Column(planner_types::pre_asap::ColumnRef::SampleValue)));
+                    SummaryExpr::SummaryAgg { input, family, .. }
+                    if !matches!(family, SummaryFamilyType::Sketch(kind, _) if matches!(kind.algorithm(), SketchAlgorithm::UnivMon))
+                        && asap_types::accumulator_spec::is_scalar_sample_value(input));
+                let frequency_series_input = matches!(&node.expr,
+                    SummaryExpr::SummaryAgg { input, family: SummaryFamilyType::Sketch(kind, _), .. }
+                    if matches!(kind.algorithm(), SketchAlgorithm::Hll | SketchAlgorithm::UnivMon)
+                        && asap_types::accumulator_spec::is_unit_sample_frequency(input));
                 return matches!(reduction, Reduction::PerEntity)
                     && !pooling
-                    && !(preserves_series_state || scalar_series_input);
+                    && !(preserves_series_state || scalar_series_input || frequency_series_input);
             }
             let additive_reduction = matches!(reduction, Reduction::Reduce(_))
                 && matches!(family, SummaryFamilyType::ExactAggregate(ExactKind::Sum, _))
@@ -2807,6 +2812,11 @@ fn retained_state_bytes(materialization: &asap_types::PrecomputeMaterialization)
                 + parameter(&["heap_size"], 1) * 256
         }
         A::DatasketchesKLL => parameter(&["k"], 200) * 32,
+        A::UnivMon => {
+            parameter(&["layers"], 4)
+                * (parameter(&["sketch_rows"], 5) * parameter(&["sketch_cols"], 1024) * 16
+                    + parameter(&["heap_size"], 32) * 256)
+        }
         A::HydraKLL => parameter(&["k"], 200) * parameter(&["col", "cols"], 1) * 32,
         A::HLL => 1u128 << parameter(&["precision", "p"], 14).min(24),
         A::DDSketch => 64 * 1024,
@@ -3432,6 +3442,14 @@ fn collect_selected_materializations(
                 input,
                 ..
             } => {
+                // Planner can model these families, but no backend state
+                // implementation exists. Leave an exact boundary unbound.
+                if matches!(
+                    kind.algorithm(),
+                    SketchAlgorithm::Kmv | SketchAlgorithm::Theta
+                ) {
+                    return Ok(());
+                }
                 if let Some(readout) = readout {
                     let mut parameters = sketch_params_json(kind.params());
                     let mut item_label = None;
@@ -3564,6 +3582,13 @@ pub(crate) fn physical_materialization_family(family: &SummaryFamilyType) -> Sum
 fn sketch_params_json(params: &planner_types::post_asap::SketchParams) -> Value {
     use planner_types::post_asap::SketchParams as P;
     match params {
+        P::UnivMon {
+            heap_size,
+            sketch_rows,
+            sketch_cols,
+            layers,
+        } => json!({"heap_size": heap_size, "sketch_rows": sketch_rows,
+                "sketch_cols": sketch_cols, "layers": layers}),
         P::Kll { k } => json!({"k": k}),
         P::Cms { width, depth } => json!({"width": width, "depth": depth}),
         P::Hll { precision } => json!({"precision": precision}),
@@ -4144,6 +4169,62 @@ mod tests {
         request_with_evidence(query_id, promql, None).expect("post-ASAP selection")
     }
 
+    /// Distinct range queries retain a per-series HLL selected by Planner.
+    #[test]
+    fn distinct_range_compiles_to_partitioned_hll() {
+        let mut deployment = environment(10_000);
+        deployment.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite;
+        deployment.collector_ids.clear();
+        let mut workload = request("distinct", "distinct_over_time(m{job=\"api\"}[1m])");
+        let query = &mut workload.queries[0];
+        // HLL's modeled RSE does not certify a failure probability.
+        query.accuracy = AccuracyTarget::Epsilon(0.05);
+        let parsed = crate::query_parser::parse_query_expr_canonical(
+            &query.query_string,
+            query.accuracy.clone(),
+        )
+        .unwrap();
+        query.post_asap =
+            select_post_asap(&parsed, query.accuracy.clone(), &query.lifecycle, None).unwrap();
+        let plan = PhysicalCompiler
+            .compile_metricsql(workload, deployment)
+            .unwrap();
+        assert_eq!(plan.precompute_plan.materializations.len(), 1);
+        let materialization = &plan.precompute_plan.materializations[0];
+        assert_eq!(
+            materialization.aggregation_type,
+            asap_types::AggregationType::HLL
+        );
+        assert_eq!(
+            materialization.partitioning,
+            Some(asap_types::sds::PopulationPartitioning::PerEntity)
+        );
+        plan.precompute_plan.validate().unwrap();
+        assert!(plan
+            .query_plan
+            .entries
+            .values()
+            .any(|entry| entry.nodes.values().any(|node| matches!(
+                node,
+                crate::query_plan::QueryPlanNode::SummaryEstimate {
+                    query: crate::query_plan::QueryReadout::Cardinality,
+                    ..
+                }
+            ))));
+    }
+
+    /// An unimplemented cardinality family fails admission rather than panicking in an emitter.
+    #[test]
+    fn unsupported_cardinality_family_fails_admission() {
+        let mut deployment = environment(10_000);
+        deployment.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite;
+        deployment.collector_ids.clear();
+        let mut workload = request("confidence", "distinct_over_time(m[1m])");
+        workload.hybrid_execution = true;
+        let result = PhysicalCompiler.compile_metricsql(workload, deployment);
+        assert!(matches!(result, Err(CompileError::QueryPlan(_))));
+    }
+
     #[test]
     fn duplicate_query_ids_cannot_overwrite_installed_dag_documents() {
         let mut workload = request("shared-id", "max_over_time(a[1m])");
@@ -5410,7 +5491,7 @@ mod tests {
     }
 
     #[test]
-    fn backend_local_hll_rejected_but_envelope_ingest_supported() {
+    fn backend_local_hll_and_envelope_ingest_are_supported() {
         let bundle = PhysicalCompiler
             .compile(
                 request("q", "quantile_over_time(0.99, m[1m])"),
@@ -5435,14 +5516,8 @@ mod tests {
             require_registered_producer: false,
         };
         envelope_plan.producers.clear();
-        assert!(matches!(
-            envelope_plan.validate(),
-            Err(PrecomputePlanError::UnsupportedFamily(_))
-        ));
-        assert!(matches!(
-            PrecomputePlan::build_backend_local(bundle.envelope, materializations),
-            Err(PrecomputePlanError::UnsupportedFamily(_))
-        ));
+        envelope_plan.validate().unwrap();
+        PrecomputePlan::build_backend_local(bundle.envelope, materializations).unwrap();
     }
 
     #[test]
diff --git a/control_plane/src/physical/deployment_cost/wire.rs b/control_plane/src/physical/deployment_cost/wire.rs
index dee12c6ec..0edb5d0f9 100644
--- a/control_plane/src/physical/deployment_cost/wire.rs
+++ b/control_plane/src/physical/deployment_cost/wire.rs
@@ -125,6 +125,11 @@ impl WireCostTable {
     // control_plane/docs/design-asapplanner-pin-migration.md).
     pub const fn for_algorithm(&self, algorithm: &SketchAlgorithm) -> SketchWireCost {
         match algorithm {
+            // No collector wire implementation is available for this family.
+            SketchAlgorithm::UnivMon => SketchWireCost {
+                state_bytes: u64::MAX,
+                envelope_bytes: 0,
+            },
             SketchAlgorithm::DDSketch => self.ddsketch_delta,
             SketchAlgorithm::Kll => self.kll_full,
             SketchAlgorithm::Hll => self.hll_delta,
diff --git a/control_plane/src/physical/post_asap/matcher.rs b/control_plane/src/physical/post_asap/matcher.rs
index db69c59c3..f5a671e8d 100644
--- a/control_plane/src/physical/post_asap/matcher.rs
+++ b/control_plane/src/physical/post_asap/matcher.rs
@@ -110,6 +110,7 @@ pub fn sketch_family_satisfied(required: &SketchAlgorithm, available: &SketchAlg
 /// this function never sees).
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum SummaryFamily {
+    Universal,
     Quantile,
     Cardinality,
     /// Bare per-item frequency point-query — no heavy-hitter heap.
@@ -131,6 +132,7 @@ impl SummaryFamily {
 
 fn summary_family(kind: &SketchAlgorithm) -> SummaryFamily {
     match kind {
+        SketchAlgorithm::UnivMon => SummaryFamily::Universal,
         SketchAlgorithm::Kll | SketchAlgorithm::DDSketch => SummaryFamily::Quantile,
         SketchAlgorithm::Hll | SketchAlgorithm::Theta | SketchAlgorithm::Kmv => {
             SummaryFamily::Cardinality
@@ -154,6 +156,12 @@ mod tests {
     /// never arise from real code.
     fn params_for(kind: &SketchAlgorithm) -> SketchParams {
         match kind {
+            SketchAlgorithm::UnivMon => SketchParams::UnivMon {
+                heap_size: 32,
+                sketch_rows: 5,
+                sketch_cols: 1024,
+                layers: 4,
+            },
             SketchAlgorithm::Kll => SketchParams::Kll { k: 200 },
             SketchAlgorithm::Cms => SketchParams::Cms {
                 width: 100,
diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs
index 1fbf7827f..0e2d1ceae 100644
--- a/control_plane/src/query_plan.rs
+++ b/control_plane/src/query_plan.rs
@@ -809,6 +809,8 @@ pub enum ExactReadout {
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 #[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
 pub enum QueryReadout {
+    FrequencyL2,
+    FrequencyEntropy,
     Quantile {
         q: f64,
     },
@@ -825,6 +827,8 @@ pub enum QueryReadout {
 impl From<SketchQuery> for QueryReadout {
     fn from(query: SketchQuery) -> Self {
         match query {
+            SketchQuery::FrequencyL2 => Self::FrequencyL2,
+            SketchQuery::FrequencyEntropy => Self::FrequencyEntropy,
             SketchQuery::Quantile { q } => Self::Quantile { q },
             SketchQuery::PointCount { key, value } => Self::PointCount { key, value },
             SketchQuery::Cardinality => Self::Cardinality,
@@ -836,6 +840,8 @@ impl From<SketchQuery> for QueryReadout {
 impl From<QueryReadout> for SketchQuery {
     fn from(query: QueryReadout) -> Self {
         match query {
+            QueryReadout::FrequencyL2 => Self::FrequencyL2,
+            QueryReadout::FrequencyEntropy => Self::FrequencyEntropy,
             QueryReadout::Quantile { q } => Self::Quantile { q },
             QueryReadout::PointCount { key, value } => Self::PointCount { key, value },
             QueryReadout::Cardinality => Self::Cardinality,
diff --git a/control_plane/src/query_planning.rs b/control_plane/src/query_planning.rs
index ccef785d4..87bc1704c 100644
--- a/control_plane/src/query_planning.rs
+++ b/control_plane/src/query_planning.rs
@@ -169,6 +169,7 @@ fn planned_capability(
             return None;
         };
         Some(match kind.algorithm() {
+            planner_types::post_asap::SketchAlgorithm::UnivMon => return None,
             planner_types::post_asap::SketchAlgorithm::DDSketch => SketchAlgorithm::DDSketch,
             planner_types::post_asap::SketchAlgorithm::Kll => SketchAlgorithm::Kll,
             planner_types::post_asap::SketchAlgorithm::Hll => SketchAlgorithm::Hll,
@@ -193,6 +194,7 @@ fn planned_capability(
                 return None;
             };
             match query {
+                SketchQuery::FrequencyL2 | SketchQuery::FrequencyEntropy => return None,
                 SketchQuery::Quantile { .. } => Capability::QuantileApprox(Some(handle(family)?)),
                 SketchQuery::Cardinality => Capability::CardinalityApprox,
                 SketchQuery::PointCount { .. } => {
diff --git a/control_plane/src/sketch_selection.rs b/control_plane/src/sketch_selection.rs
index eac452c97..75ea2353d 100644
--- a/control_plane/src/sketch_selection.rs
+++ b/control_plane/src/sketch_selection.rs
@@ -42,7 +42,7 @@ pub fn sketch_type_for_algorithm(h: SketchAlgorithm) -> Option<SketchType> {
             Some(SketchType::CountSketch)
         }
         SketchAlgorithm::Cms | SketchAlgorithm::CmsWithHeap => Some(SketchType::CountMinSketch),
-        SketchAlgorithm::Kmv | SketchAlgorithm::Theta => None,
+        SketchAlgorithm::UnivMon | SketchAlgorithm::Kmv | SketchAlgorithm::Theta => None,
     }
 }
 
diff --git a/control_plane/tests/offline_evidence.rs b/control_plane/tests/offline_evidence.rs
index a48a95c35..769bdd6bd 100644
--- a/control_plane/tests/offline_evidence.rs
+++ b/control_plane/tests/offline_evidence.rs
@@ -215,7 +215,7 @@ fn fixture(
         "distinct_count":100, "parameters":{"seed":42}});
     let environment = json!({"id":"test-machine", "cpu":"test", "os":"test", "runtime":"test",
         "implementation":"synthetic-test-fixture", "implementation_version":"v1"});
-    let records: Vec<_> = [(SketchAlgorithm::Cms,20.0), (SketchAlgorithm::CountSketch,10.0)].into_iter().map(|(algorithm,value)| {
+    let records: Vec<_> = [(SketchAlgorithm::Cms,20.0), (SketchAlgorithm::CountSketch,10.0), (SketchAlgorithm::UnivMon,30.0)].into_iter().map(|(algorithm,value)| {
         let params = model.size_params(algorithm.clone(), intent, 0.01, 0.01);
         json!({"id":format!("test-{algorithm:?}"), "algorithm":algorithm, "params":params,
             "distribution":distribution, "environment":environment,
diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml
index f553e5c7a..9a51159b1 100644
--- a/crates/asap_types/Cargo.toml
+++ b/crates/asap_types/Cargo.toml
@@ -32,4 +32,4 @@ sha2 = "0.10"
 # exactly (`control_plane/Cargo.toml`) -- two different revs of the same
 # git dependency in one workspace resolve to two distinct Rust types that
 # won't unify.
-planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "890504025e4f780880cc6c89d24edd6591af868a" }
+planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc" }
diff --git a/crates/asap_types/src/accumulator_spec.rs b/crates/asap_types/src/accumulator_spec.rs
index 23732563a..5e6202b6b 100644
--- a/crates/asap_types/src/accumulator_spec.rs
+++ b/crates/asap_types/src/accumulator_spec.rs
@@ -104,6 +104,34 @@ pub enum SampleUpdateRule {
     CounterDelta { scale: f64 },
 }
 
+/// The implemented raw frequency domain counts occurrences of sample values.
+pub fn is_unit_sample_frequency(update: &planner_types::post_asap::SummaryUpdate) -> bool {
+    use planner_types::post_asap::{NonNegativeWeightProof, SummaryInputExpr, WeightDomain};
+    matches!(
+        update.item,
+        Some(SummaryInputExpr::Column(
+            planner_types::pre_asap::ColumnRef::SampleValue
+        ))
+    ) && matches!(update.weight, SummaryInputExpr::Constant(1.0))
+        && matches!(
+            update.weight_domain,
+            WeightDomain::NonNegative {
+                proof: NonNegativeWeightProof::UnitCount
+            }
+        )
+}
+
+/// Raw HLL hashes the scalar sample; it does not interpret it as a frequency weight.
+pub fn is_scalar_sample_value(update: &planner_types::post_asap::SummaryUpdate) -> bool {
+    update.item.is_none()
+        && matches!(
+            update.weight,
+            planner_types::post_asap::SummaryInputExpr::Column(
+                planner_types::pre_asap::ColumnRef::SampleValue
+            )
+        )
+}
+
 impl AggregationConfig {
     pub fn sample_update_rule(&self) -> SampleUpdateRule {
         let scale = self
@@ -311,6 +339,32 @@ impl AggregationConfig {
                 ),
                 false,
             ),
+            UnivMon => {
+                let param =
+                    |name: &str, default: u64, max: u64| -> Result<u32, AccumulatorSpecError> {
+                        let value = self
+                            .parameters
+                            .get(name)
+                            .map_or(Some(default), |v| v.as_u64())
+                            .ok_or(AccumulatorSpecError::UnmappedAggregationType(UnivMon))?;
+                        if value == 0 || value > max {
+                            return Err(AccumulatorSpecError::UnmappedAggregationType(UnivMon));
+                        }
+                        Ok(value as u32)
+                    };
+                (
+                    independent_sketch(
+                        SketchAlgorithm::UnivMon,
+                        SketchParams::UnivMon {
+                            heap_size: param("heap_size", 32, u32::MAX as u64)?,
+                            sketch_rows: param("sketch_rows", 5, 20)?,
+                            sketch_cols: param("sketch_cols", 1024, u32::MAX as u64)?,
+                            layers: param("layers", 4, 64)? as u8,
+                        },
+                    ),
+                    false,
+                )
+            }
             HLL => {
                 let precision = match self.parameters.get("precision") {
                     None => 14,
diff --git a/crates/asap_types/src/accuracy.rs b/crates/asap_types/src/accuracy.rs
index ad57f101f..33c47b5b8 100644
--- a/crates/asap_types/src/accuracy.rs
+++ b/crates/asap_types/src/accuracy.rs
@@ -5,6 +5,7 @@ use serde::{Deserialize, Serialize};
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum AccuracyKind {
+    Uncalibrated,
     Exact,
     AdditiveFrequency,
     RelativeCardinality,
@@ -16,6 +17,7 @@ pub enum AccuracyKind {
 impl AccuracyKind {
     pub fn as_str(self) -> &'static str {
         match self {
+            Self::Uncalibrated => "uncalibrated",
             Self::Exact => "exact",
             Self::AdditiveFrequency => "additive_frequency",
             Self::RelativeCardinality => "relative_cardinality",
diff --git a/crates/asap_types/src/aggregation_type.rs b/crates/asap_types/src/aggregation_type.rs
index d98315479..f1e4f0f97 100644
--- a/crates/asap_types/src/aggregation_type.rs
+++ b/crates/asap_types/src/aggregation_type.rs
@@ -44,6 +44,7 @@ pub enum AggregationType {
     CountSketchWithHeap,
     // ---------- cardinality / set tracking ----------
     HLL,
+    UnivMon,
     DDSketch,
     // ---------- legacy config wrapper names ----------
     SingleSubpopulation,
@@ -66,6 +67,7 @@ impl AggregationType {
             AggregationType::CountSketch => "CountSketch",
             AggregationType::CountSketchWithHeap => "CountSketchWithHeap",
             AggregationType::HLL => "HLL",
+            AggregationType::UnivMon => "UnivMon",
             AggregationType::DDSketch => "DDSketch",
             AggregationType::SingleSubpopulation => "SingleSubpopulation",
             AggregationType::MultipleSubpopulation => "MultipleSubpopulation",
@@ -137,6 +139,7 @@ impl FromStr for AggregationType {
             "CountSketch" => Ok(AggregationType::CountSketch),
             "CountSketchWithHeap" => Ok(AggregationType::CountSketchWithHeap),
             "HLL" | "HyperLogLog" => Ok(AggregationType::HLL),
+            "UnivMon" => Ok(AggregationType::UnivMon),
             "DDSketch" | "DdSketch" => Ok(AggregationType::DDSketch),
             "SingleSubpopulation" => Ok(AggregationType::SingleSubpopulation),
             "MultipleSubpopulation" => Ok(AggregationType::MultipleSubpopulation),
diff --git a/crates/asap_types/src/enums.rs b/crates/asap_types/src/enums.rs
index 05bf9171d..9e65753c7 100644
--- a/crates/asap_types/src/enums.rs
+++ b/crates/asap_types/src/enums.rs
@@ -22,6 +22,8 @@ pub enum Statistic {
     Count,
     Sum,
     Cardinality,
+    FrequencyL2,
+    FrequencyEntropy,
     Increase,
     Rate,
     Min,
@@ -37,6 +39,8 @@ impl fmt::Display for Statistic {
             Statistic::Count => write!(f, "count"),
             Statistic::Sum => write!(f, "sum"),
             Statistic::Cardinality => write!(f, "cardinality"),
+            Statistic::FrequencyL2 => write!(f, "frequency_l2"),
+            Statistic::FrequencyEntropy => write!(f, "frequency_entropy"),
             Statistic::Increase => write!(f, "increase"),
             Statistic::Rate => write!(f, "rate"),
             Statistic::Min => write!(f, "min"),
@@ -55,6 +59,8 @@ impl Statistic {
             "count" => Some(Statistic::Count),
             "sum" => Some(Statistic::Sum),
             "cardinality" => Some(Statistic::Cardinality),
+            "frequency_l2" => Some(Statistic::FrequencyL2),
+            "frequency_entropy" => Some(Statistic::FrequencyEntropy),
             "increase" => Some(Statistic::Increase),
             "rate" => Some(Statistic::Rate),
             "min" => Some(Statistic::Min),
diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs
index 7f5128631..257d99dd4 100644
--- a/crates/asap_types/src/precompute_plan.rs
+++ b/crates/asap_types/src/precompute_plan.rs
@@ -363,6 +363,34 @@ impl PrecomputePlan {
                         "DAG materialization has no runtime configuration".into(),
                     ));
                 };
+                if self.ingest.protocol == IngestProtocol::PrometheusRemoteWriteV1
+                    && matches!(
+                        config.aggregation_type,
+                        crate::AggregationType::HLL | crate::AggregationType::UnivMon
+                    )
+                {
+                    if let planner_types::post_asap::ExecutableOperatorPayload::SummaryAgg {
+                        input,
+                        ..
+                    } = &node.payload
+                    {
+                        let supported = match config.aggregation_type {
+                            crate::AggregationType::HLL => {
+                                crate::accumulator_spec::is_scalar_sample_value(input)
+                                    || crate::accumulator_spec::is_unit_sample_frequency(input)
+                            }
+                            crate::AggregationType::UnivMon => {
+                                crate::accumulator_spec::is_unit_sample_frequency(input)
+                            }
+                            _ => unreachable!(),
+                        };
+                        if !supported {
+                            return Err(PrecomputePlanError::CatalogContract(
+                                "raw materialization input does not match its accumulator update semantics".into(),
+                            ));
+                        }
+                    }
+                }
                 if let Some(partitioning) = config.partitioning {
                     if let planner_types::post_asap::ExecutableOperatorPayload::SummaryAgg {
                         reduction,
@@ -419,15 +447,6 @@ impl PrecomputePlan {
                     reason: "window kind disagrees with size and slide".into(),
                 });
             }
-            // HLL is supported as an ingested sketch envelope, not as a raw
-            // accumulator. Validate here so external installs cannot bypass it.
-            if self.ingest.protocol == IngestProtocol::PrometheusRemoteWriteV1
-                && materialization.aggregation_type == crate::AggregationType::HLL
-            {
-                return Err(PrecomputePlanError::UnsupportedFamily(
-                    materialization.policy_fp_u64(),
-                ));
-            }
             if !materializations.insert(materialization.policy_fingerprint().into()) {
                 return Err(PrecomputePlanError::DuplicateMaterialization(
                     materialization.policy_fp_u64(),
@@ -554,7 +573,9 @@ pub(crate) fn state_encodings(family: &SummaryFamilyType) -> Vec<StateEncoding>
         SummaryFamilyType::Sketch(kind, _)
             if matches!(
                 kind.algorithm(),
-                SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap
+                SketchAlgorithm::CmsWithHeap
+                    | SketchAlgorithm::CountSketchWithHeap
+                    | SketchAlgorithm::UnivMon
             ) =>
         {
             vec![StateEncoding::SketchCoreMsgpackV1]
diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs
index d61ba2f7f..23c3cc48b 100644
--- a/crates/asap_types/src/sds.rs
+++ b/crates/asap_types/src/sds.rs
@@ -376,6 +376,14 @@ pub enum SummaryOperator {
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
 pub enum FidelityGuarantee {
+    /// Total unit-frequency count is exact. Distinct, L2 and entropy require
+    /// readout-specific evidence; dimensions alone certify no error bound.
+    UnivMonFrequency {
+        heap_size: u32,
+        sketch_rows: u32,
+        sketch_cols: u32,
+        layers: u8,
+    },
     Exact,
     /// Exact PromQL counter readout from fixed-size pane summaries. Each pane
     /// stores only `(first value/time, last value/time, reset-corrected delta,
@@ -440,6 +448,17 @@ impl FidelityGuarantee {
             },
             Ok(SummaryFamilyType::ExactAggregate(..)) => Self::Exact,
             Ok(SummaryFamilyType::Sketch(kind, _)) => match kind.params() {
+                SketchParams::UnivMon {
+                    heap_size,
+                    sketch_rows,
+                    sketch_cols,
+                    layers,
+                } => Self::UnivMonFrequency {
+                    heap_size: *heap_size,
+                    sketch_rows: *sketch_rows,
+                    sketch_cols: *sketch_cols,
+                    layers: *layers,
+                },
                 SketchParams::Kll { k } => Self::KllRankError {
                     k: *k,
                     model: if config.aggregation_type == AggregationType::HydraKLL {
@@ -479,6 +498,17 @@ impl FidelityGuarantee {
     }
     fn validate(&self) -> Result<(), SdsError> {
         let valid = match self {
+            Self::UnivMonFrequency {
+                heap_size,
+                sketch_rows,
+                sketch_cols,
+                layers,
+            } => {
+                *heap_size > 0
+                    && *sketch_cols > 0
+                    && (1..=20).contains(sketch_rows)
+                    && (1..=64).contains(layers)
+            }
             Self::Exact => true,
             Self::ExactCounter {
                 model,
@@ -618,6 +648,7 @@ impl FidelityGuarantee {
         use SketchAlgorithm as S;
 
         let configured = |aggregation_type| match (aggregation_type, self) {
+            (A::UnivMon, UnivMonFrequency { .. }) => true,
             (A::Sum | A::MultipleSum | A::MinMax | A::MultipleMinMax, Exact) => true,
             (A::Increase | A::MultipleIncrease, ExactCounter { .. }) => true,
             (A::DatasketchesKLL | A::HydraKLL, KllRankError { .. }) => true,
@@ -643,7 +674,8 @@ impl FidelityGuarantee {
             } => {
                 matches!(
                     (algorithm, self),
-                    (S::Kll, KllRankError { .. })
+                    (S::UnivMon, UnivMonFrequency { .. })
+                        | (S::Kll, KllRankError { .. })
                         | (S::DDSketch, DdSketchRelativeError { .. })
                         | (S::Hll, HllCardinalityError { .. })
                         | (S::Cms | S::CmsWithHeap, CmsFrequencyError { .. })
@@ -665,6 +697,17 @@ impl FidelityGuarantee {
                 .all(|value| value.as_u64() == Some(u64::from(expected)))
         };
         match self {
+            Self::UnivMonFrequency {
+                heap_size,
+                sketch_rows,
+                sketch_cols,
+                layers,
+            } => {
+                u32_parameter(&["heap_size"], *heap_size)
+                    && u32_parameter(&["sketch_rows"], *sketch_rows)
+                    && u32_parameter(&["sketch_cols"], *sketch_cols)
+                    && u32_parameter(&["layers"], u32::from(*layers))
+            }
             Self::KllRankError { k, .. } => u32_parameter(&["k", "K"], *k),
             Self::HllCardinalityError { precision, .. } => {
                 u32_parameter(&["precision", "p"], *precision)
diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml
index 972f993cb..d33c3b29e 100644
--- a/data_plane/Cargo.toml
+++ b/data_plane/Cargo.toml
@@ -39,8 +39,8 @@ sha2 = "0.10"
 # reduction: Reduction, .. }`) are `pre_asap` types, in the same crate now
 # (not a separate `asap-ir` import). Query serving consumes the compiled
 # QueryPlan; these types are used at physical-plan compilation boundaries.
-planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "890504025e4f780880cc6c89d24edd6591af868a" }
-asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "890504025e4f780880cc6c89d24edd6591af868a" }
+planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc" }
+asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc" }
 
 # Shared external (workspace)
 serde.workspace = true
@@ -133,7 +133,7 @@ fs2 = "0.4"
 # none of them.
 
 [dev-dependencies]
-asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "890504025e4f780880cc6c89d24edd6591af868a" }
+asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc" }
 tempfile = "3.20.0"
 criterion = { version = "0.5", features = ["html_reports"] }
 tokio-tungstenite = "0.21"
diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs
index ca4c32547..a8d6e415b 100644
--- a/data_plane/src/drivers/ingest/otel.rs
+++ b/data_plane/src/drivers/ingest/otel.rs
@@ -1380,7 +1380,9 @@ async fn route_modified_otlp_sketches_to_precompute(
                                 SketchAlgorithm::DDSketch | SketchAlgorithm::Kll => {
                                     Capability::QuantileApprox(Some(algorithm.clone()))
                                 }
-                                SketchAlgorithm::Hll => Capability::CardinalityApprox,
+                                SketchAlgorithm::Hll | SketchAlgorithm::UnivMon => {
+                                    Capability::CardinalityApprox
+                                }
                                 // Heap-LESS frequency sketches answer bare
                                 // frequency point queries (no top-k); index
                                 // them as FrequencyEstimate so a `topk(...)`
@@ -1936,6 +1938,7 @@ fn aggregation_type_for_sketch_algorithm(
         SketchAlgorithm::DDSketch => Some(AggregationType::DDSketch),
         SketchAlgorithm::Kll => Some(AggregationType::DatasketchesKLL),
         SketchAlgorithm::Hll => Some(AggregationType::HLL),
+        SketchAlgorithm::UnivMon => Some(AggregationType::UnivMon),
         SketchAlgorithm::CountSketch => Some(AggregationType::CountSketch),
         SketchAlgorithm::CountSketchWithHeap => Some(AggregationType::CountSketchWithHeap),
         SketchAlgorithm::Cms => Some(AggregationType::CountMinSketch),
@@ -1959,6 +1962,17 @@ fn sketch_config_to_params(
     use crate::storage_engines::sketch_db::data::SketchConfig;
     let mut params = std::collections::HashMap::new();
     match cfg {
+        SketchConfig::UnivMon {
+            heap_size,
+            sketch_rows,
+            sketch_cols,
+            layers,
+        } => {
+            params.insert("heap_size".into(), serde_json::json!(heap_size));
+            params.insert("sketch_rows".into(), serde_json::json!(sketch_rows));
+            params.insert("sketch_cols".into(), serde_json::json!(sketch_cols));
+            params.insert("layers".into(), serde_json::json!(layers));
+        }
         SketchConfig::DDSketch { relative_accuracy } => {
             params.insert(
                 "relative_accuracy".to_string(),
diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs
index 6a390db68..c456d8e3c 100644
--- a/data_plane/src/drivers/query/servers/http.rs
+++ b/data_plane/src/drivers/query/servers/http.rs
@@ -2694,12 +2694,20 @@ mod tests {
             planner_revision: PLANNER_REVISION.into(),
             capability_snapshot_id: "test".into(),
         };
+        let catalog = Arc::new(
+            asap_types::summary_catalog::SummaryCatalog::from_materializations(7, 1, &[]).unwrap(),
+        );
+        let generation = catalog.reference().unwrap();
+        let sketch_index = Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new());
+        sketch_index
+            .install_summary_catalog(Arc::clone(&catalog))
+            .unwrap();
         let active = crate::storage_engines::types::HotReloadActivePhysicalPlan::new(
             crate::storage_engines::types::ActivePhysicalPlan {
                 envelope: envelope.clone(),
-                summary_catalog: None,
+                summary_catalog: Some(Arc::clone(&catalog)),
                 precompute_plan: PrecomputePlan {
-                    summary_catalog: None,
+                    summary_catalog: Some(generation.clone()),
                     envelope: envelope.clone(),
                     ingest: IngestContract {
                         protocol: IngestProtocol::PrometheusRemoteWriteV1,
@@ -2715,7 +2723,7 @@ mod tests {
                     materializations: Vec::new(),
                 },
                 transmission_plan: TransmissionPlan {
-                    summary_catalog: None,
+                    summary_catalog: Some(generation),
                     envelope,
                     frame_identity: FrameIdentityContract {
                         identity_version: 1,
@@ -2747,7 +2755,7 @@ mod tests {
             pass_raw_samples: false,
             sketch_snapshots: dashmap::DashMap::new(),
             series_resolver: Arc::new(crate::drivers::ingest::SeriesIdResolver::new()),
-            sketch_index: Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()),
+            sketch_index: Arc::clone(&sketch_index),
             observability: IngestObservability::default(),
         });
         let receiver =
@@ -2760,7 +2768,7 @@ mod tests {
                 adapter_config,
             },
             Arc::new(ASAPQueryEngine::new(15_000)),
-            Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()),
+            sketch_index,
         )
         .with_active_physical_plan(active)
         .with_remote_write(receiver.clone());
@@ -3055,9 +3063,8 @@ aggregations:
     }
 
     #[tokio::test]
-    async fn test_streaming_config_hot_reload_missing_handle_503() {
-        // setup_test_server() passes `None` for hot_reload → both
-        // endpoints should return 503 with a clear error message.
+    async fn test_streaming_config_route_is_absent_without_legacy_handle() {
+        // A server without a legacy hot-reload handle does not expose this route.
         let server_port = setup_test_server().await;
         let client = Client::new();
 
@@ -3068,7 +3075,7 @@ aggregations:
             .send()
             .await
             .unwrap();
-        assert_eq!(get_resp.status(), reqwest::StatusCode::SERVICE_UNAVAILABLE);
+        assert_eq!(get_resp.status(), reqwest::StatusCode::NOT_FOUND);
 
         let post_resp = client
             .post(format!(
@@ -3078,7 +3085,7 @@ aggregations:
             .send()
             .await
             .unwrap();
-        assert_eq!(post_resp.status(), reqwest::StatusCode::SERVICE_UNAVAILABLE);
+        assert_eq!(post_resp.status(), reqwest::StatusCode::NOT_FOUND);
     }
 
     #[tokio::test]
@@ -7278,7 +7285,8 @@ mod catalog_install_tests {
     #[test]
     fn catalog_install_applies_pane_origin_validation_to_metricsql_entries() {
         let mut request = request();
-        let entry = request.query_plan.entries.values_mut().next().unwrap();
+        let key = request.query_plan.entries.keys().next().unwrap().clone();
+        let mut entry = request.query_plan.entries.remove(&key).unwrap();
         entry.language = control_plane::query_plan::QueryLanguage::MetricsQl;
         let binding = entry
             .nodes
@@ -7291,6 +7299,12 @@ mod catalog_install_tests {
             })
             .expect("demo has maintained summaries");
         binding.pane_origin_ms = Some(1);
-        assert!(install(request).unwrap_err().contains("pane origin"));
+        let key = control_plane::query_plan::QueryPlan::catalog_key(
+            entry.language,
+            &entry.canonical_query,
+        );
+        request.query_plan.entries.insert(key, entry);
+        let error = install(request).unwrap_err();
+        assert!(error.contains("pane origin"), "{error}");
     }
 }
diff --git a/data_plane/src/precompute_engine/accumulator_factory.rs b/data_plane/src/precompute_engine/accumulator_factory.rs
index cf3aa6b76..f7e9ca500 100644
--- a/data_plane/src/precompute_engine/accumulator_factory.rs
+++ b/data_plane/src/precompute_engine/accumulator_factory.rs
@@ -21,6 +21,8 @@ use asap_types::aggregation_config::AggregationConfig;
 // former local helpers (`kll_k_param`, `heap_size_param`,
 // `ddsketch_alpha_param`) is that module's own `AccumulatorSpec`
 // construction, so they aren't re-imported here.
+use super::operators::hll_sketch_accumulator::HllSketchAccumulator;
+use super::operators::univmon_accumulator::UnivMonAccumulator;
 use asap_types::accumulator_spec::{cms_params, AccumulatorSpecError};
 use planner_types::post_asap::{ExactKind, SketchAlgorithm, SketchParams, SummaryFamilyType};
 
@@ -1144,8 +1146,43 @@ pub fn create_accumulator_updater(config: &AggregationConfig) -> Box<dyn Accumul
             )))
         }
 
-        // HLL has an envelope catalog identity but no raw updater; the
-        // backend-local plan builder rejects it before reaching this factory.
+        (SummaryFamilyType::Sketch(kind, _), false)
+            if kind.algorithm() == &SketchAlgorithm::UnivMon =>
+        {
+            let SketchParams::UnivMon {
+                heap_size,
+                sketch_rows,
+                sketch_cols,
+                layers,
+            } = kind.params()
+            else {
+                unreachable!("validated UnivMon family parameters")
+            };
+            Box::new(UnivMonUpdater {
+                acc: UnivMonAccumulator::new(
+                    *heap_size as usize,
+                    *sketch_rows as usize,
+                    *sketch_cols as usize,
+                    *layers as usize,
+                )
+                .expect("validated UnivMon dimensions"),
+            })
+        }
+
+        (SummaryFamilyType::Sketch(kind, _), false)
+            if kind.algorithm() == &SketchAlgorithm::Hll =>
+        {
+            let SketchParams::Hll { precision } = kind.params() else {
+                unreachable!("validated HLL family parameters")
+            };
+            Box::new(HllUpdater {
+                acc: HllSketchAccumulator::new(
+                    asap_sketchlib::HllVariant::Regular,
+                    u32::from(*precision),
+                ),
+            })
+        }
+
         // Other unsupported families retain the legacy warning fallback.
         (other_family, keyed) => {
             tracing::warn!(
@@ -1158,12 +1195,111 @@ pub fn create_accumulator_updater(config: &AggregationConfig) -> Box<dyn Accumul
     }
 }
 
+struct UnivMonUpdater {
+    acc: UnivMonAccumulator,
+}
+
+struct HllUpdater {
+    acc: HllSketchAccumulator,
+}
+
+impl AccumulatorUpdater for HllUpdater {
+    fn is_keyed(&self) -> bool {
+        false
+    }
+    fn memory_usage_bytes(&self) -> usize {
+        self.acc.approx_memory_bytes()
+    }
+    fn update_single(&mut self, value: f64, _: i64) {
+        if !value.is_nan() {
+            let bits = if value == 0.0 { 0 } else { value.to_bits() };
+            self.acc.inner.update(&bits.to_le_bytes());
+        }
+    }
+    fn update_keyed(&mut self, _: &KeyByLabelValues, value: f64, timestamp_ms: i64) {
+        self.update_single(value, timestamp_ms);
+    }
+    impl_clone_accumulator_methods!(acc);
+    fn reset(&mut self) {
+        self.acc.reset_to_empty();
+    }
+}
+
+impl AccumulatorUpdater for UnivMonUpdater {
+    fn is_keyed(&self) -> bool {
+        false
+    }
+    fn memory_usage_bytes(&self) -> usize {
+        self.acc.approx_memory_bytes()
+    }
+    fn update_single(&mut self, value: f64, _: i64) {
+        self.acc
+            .insert_sample(value)
+            .expect("UnivMon sample counter overflow");
+    }
+    fn update_keyed(&mut self, _: &KeyByLabelValues, value: f64, timestamp_ms: i64) {
+        self.update_single(value, timestamp_ms);
+    }
+    impl_clone_accumulator_methods!(acc);
+    fn reset(&mut self) {
+        self.acc.reset_to_empty();
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use asap_types::enums::WindowKind;
     use asap_types::AggregationType;
 
+    /// Both cardinality implementations consume values, with a single signed-zero identity.
+    #[test]
+    fn hll_and_univmon_raw_updates_share_value_identity() {
+        for family in [AggregationType::HLL, AggregationType::UnivMon] {
+            let config = AggregationConfig::new(
+                family,
+                String::new(),
+                Default::default(),
+                asap_types::KeyByLabelNames::new(vec![]),
+                asap_types::KeyByLabelNames::new(vec![]),
+                asap_types::KeyByLabelNames::new(vec![]),
+                String::new(),
+                60,
+                60,
+                WindowKind::Tumbling,
+                "m".into(),
+                "m".into(),
+                None,
+                None,
+                None,
+            );
+            let mut updater = create_accumulator_updater(&config);
+            for value in [0.0, -0.0, 2.0, 2.0, f64::NAN] {
+                updater.update_single(value, 1000);
+            }
+            let state = updater.take_accumulator();
+            assert_eq!(state.get_accumulator_type(), family);
+            let estimate = state
+                .query_statistic(
+                    asap_types::Statistic::Cardinality,
+                    &None,
+                    &Default::default(),
+                )
+                .unwrap();
+            assert!((estimate - 2.0).abs() < 0.05, "{family:?}: {estimate}");
+            assert!(updater.memory_usage_bytes() >= 4096);
+            let empty = updater
+                .snapshot_accumulator()
+                .query_statistic(
+                    asap_types::Statistic::Cardinality,
+                    &None,
+                    &Default::default(),
+                )
+                .unwrap();
+            assert_eq!(empty, 0.0);
+        }
+    }
+
     #[test]
     fn test_sum_updater() {
         let mut updater = SumAccumulatorUpdater::new();
diff --git a/data_plane/src/precompute_engine/operators/hll_sketch_accumulator.rs b/data_plane/src/precompute_engine/operators/hll_sketch_accumulator.rs
index b2172c7ed..b4c3b4d0c 100644
--- a/data_plane/src/precompute_engine/operators/hll_sketch_accumulator.rs
+++ b/data_plane/src/precompute_engine/operators/hll_sketch_accumulator.rs
@@ -265,6 +265,9 @@ impl SerializableToSink for HllSketchAccumulator {
 }
 
 impl AggregateCore for HllSketchAccumulator {
+    fn approx_memory_bytes(&self) -> usize {
+        std::mem::size_of::<Self>().saturating_add(self.inner.registers.capacity())
+    }
     fn clone_boxed_core(&self) -> Box<dyn AggregateCore> {
         Box::new(self.clone())
     }
diff --git a/data_plane/src/precompute_engine/operators/mod.rs b/data_plane/src/precompute_engine/operators/mod.rs
index 458812717..7d7c84666 100644
--- a/data_plane/src/precompute_engine/operators/mod.rs
+++ b/data_plane/src/precompute_engine/operators/mod.rs
@@ -14,6 +14,7 @@ pub mod multiple_min_max_accumulator;
 pub mod multiple_sum_accumulator;
 pub mod sketch_envelope_accumulator;
 pub mod sum_accumulator;
+pub mod univmon_accumulator;
 
 pub use count_min_sketch_accumulator::*;
 pub use count_min_sketch_with_heap_accumulator::*;
diff --git a/data_plane/src/precompute_engine/operators/univmon_accumulator.rs b/data_plane/src/precompute_engine/operators/univmon_accumulator.rs
new file mode 100644
index 000000000..b216f8f54
--- /dev/null
+++ b/data_plane/src/precompute_engine/operators/univmon_accumulator.rs
@@ -0,0 +1,236 @@
+//! One frequency state shared by count, distinct, L2 and entropy readouts.
+
+use crate::storage_engines::types::{
+    AggregateCore, AuxStats, KeyByLabelValues, SerializableToSink,
+};
+use asap_sketchlib::{DataInput, UnivMon};
+use asap_types::{AggregationType, Statistic};
+use serde_json::Value;
+use std::collections::HashMap;
+
+type Error = Box<dyn std::error::Error + Send + Sync>;
+
+#[derive(Debug, Clone)]
+pub struct UnivMonAccumulator {
+    inner: UnivMon,
+}
+
+impl UnivMonAccumulator {
+    pub fn new(heap_size: usize, rows: usize, cols: usize, layers: usize) -> Result<Self, Error> {
+        if heap_size == 0 || cols == 0 || !(1..=20).contains(&rows) || !(1..=64).contains(&layers) {
+            return Err("invalid UnivMon dimensions".into());
+        }
+        rows.checked_mul(cols)
+            .and_then(|n| n.checked_mul(layers))
+            .ok_or("UnivMon dimensions overflow")?;
+        Ok(Self {
+            inner: UnivMon::init_univmon(heap_size, rows, cols, layers),
+        })
+    }
+
+    /// Each non-NaN sample is one occurrence. Signed zero has one identity.
+    pub fn insert_sample(&mut self, value: f64) -> Result<(), Error> {
+        if value.is_nan() {
+            return Ok(());
+        }
+        self.inner
+            .bucket_size
+            .checked_add(1)
+            .ok_or("UnivMon count overflow")?;
+        let bits = if value == 0.0 { 0 } else { value.to_bits() };
+        self.inner.insert(&DataInput::U64(bits), 1);
+        Ok(())
+    }
+
+    pub fn from_bytes(bytes: &[u8]) -> Result<Self, Error> {
+        let inner = UnivMon::deserialize_from_bytes(bytes)
+            .map_err(|e| format!("invalid UnivMon state: {e}"))?;
+        if !inner.accepts_standard_updates() {
+            return Err(
+                "terminal-mode UnivMon state cannot enter the standard-update accumulator".into(),
+            );
+        }
+        Ok(Self { inner })
+    }
+
+    fn compatible(&self, other: &Self) -> bool {
+        (
+            self.inner.heap_size,
+            self.inner.sketch_row,
+            self.inner.sketch_col,
+            self.inner.layer_size,
+        ) == (
+            other.inner.heap_size,
+            other.inner.sketch_row,
+            other.inner.sketch_col,
+            other.inner.layer_size,
+        )
+    }
+
+    pub fn dimensions(&self) -> (usize, usize, usize, usize) {
+        (
+            self.inner.heap_size,
+            self.inner.sketch_row,
+            self.inner.sketch_col,
+            self.inner.layer_size,
+        )
+    }
+
+    pub fn merge_in_place(&mut self, other: &Self) -> Result<(), Error> {
+        if !self.compatible(other) {
+            return Err("incompatible UnivMon dimensions".into());
+        }
+        self.inner
+            .bucket_size
+            .checked_add(other.inner.bucket_size)
+            .ok_or("UnivMon count overflow")?;
+        self.inner.merge(&other.inner);
+        Ok(())
+    }
+}
+
+impl SerializableToSink for UnivMonAccumulator {
+    fn serialize_to_json(&self) -> Value {
+        serde_json::json!({"count": self.inner.bucket_size})
+    }
+
+    fn serialize_to_bytes(&self) -> Vec<u8> {
+        self.inner
+            .serialize_to_bytes()
+            .expect("validated unit-frequency UnivMon state")
+    }
+}
+
+impl AggregateCore for UnivMonAccumulator {
+    fn approx_memory_bytes(&self) -> usize {
+        std::mem::size_of::<Self>().saturating_add(
+            self.inner.layer_size.saturating_mul(
+                self.inner
+                    .sketch_row
+                    .saturating_mul(self.inner.sketch_col)
+                    .saturating_mul(16)
+                    .saturating_add(self.inner.heap_size.saturating_mul(256)),
+            ),
+        )
+    }
+    fn clone_boxed_core(&self) -> Box<dyn AggregateCore> {
+        Box::new(self.clone())
+    }
+    fn type_name(&self) -> &'static str {
+        "UnivMonAccumulator"
+    }
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+    fn as_any_mut(&mut self) -> &mut dyn std::any::Any {
+        self
+    }
+    fn get_accumulator_type(&self) -> AggregationType {
+        AggregationType::UnivMon
+    }
+    fn get_keys(&self) -> Option<Vec<KeyByLabelValues>> {
+        None
+    }
+    fn reset_to_empty(&mut self) {
+        self.inner.free();
+    }
+
+    fn merge_with(&self, other: &dyn AggregateCore) -> Result<Box<dyn AggregateCore>, Error> {
+        let other = other
+            .as_any()
+            .downcast_ref::<Self>()
+            .ok_or("expected UnivMon state")?;
+        let mut merged = self.clone();
+        merged.merge_in_place(other)?;
+        Ok(Box::new(merged))
+    }
+
+    fn query_statistic(
+        &self,
+        statistic: Statistic,
+        key: &Option<KeyByLabelValues>,
+        _: &HashMap<String, String>,
+    ) -> Result<f64, Error> {
+        if key.is_some() {
+            return Err("UnivMon population is selected by the catalog binding".into());
+        }
+        match statistic {
+            Statistic::Count => Ok(self.inner.calc_l1()),
+            Statistic::Cardinality => Ok(self.inner.calc_card()),
+            Statistic::FrequencyL2 => Ok(self.inner.calc_l2()),
+            Statistic::FrequencyEntropy => Ok(self.inner.calc_entropy()),
+            _ => Err("unsupported UnivMon readout".into()),
+        }
+    }
+
+    fn aux_stats(&self) -> AuxStats {
+        AuxStats {
+            count: Some(self.inner.bucket_size as u64),
+            ..AuxStats::empty()
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn read(state: &dyn AggregateCore, stat: Statistic) -> f64 {
+        state.query_statistic(stat, &None, &HashMap::new()).unwrap()
+    }
+
+    /// Duplicate samples affect frequency but not cardinality, including signed zero.
+    #[test]
+    fn shared_readouts_survive_serialization() {
+        let mut state = UnivMonAccumulator::new(32, 5, 1024, 4).unwrap();
+        for value in [0.0, -0.0, 2.0, 2.0, f64::NAN] {
+            state.insert_sample(value).unwrap();
+        }
+        let restored = UnivMonAccumulator::from_bytes(&state.serialize_to_bytes()).unwrap();
+        for stat in [
+            Statistic::Count,
+            Statistic::Cardinality,
+            Statistic::FrequencyL2,
+            Statistic::FrequencyEntropy,
+        ] {
+            assert_eq!(read(&state, stat), read(&restored, stat));
+        }
+        assert_eq!(read(&restored, Statistic::Count), 4.0);
+        assert!((read(&restored, Statistic::Cardinality) - 2.0).abs() < 0.01);
+        assert!((read(&restored, Statistic::FrequencyL2) - 8.0f64.sqrt()).abs() < 0.01);
+        assert!((read(&restored, Statistic::FrequencyEntropy) - 1.0).abs() < 0.01);
+    }
+
+    /// Terminal-mode serialization is valid sketchlib state but not this accumulator's update domain.
+    #[test]
+    fn terminal_state_is_rejected_before_ingestion_or_merge() {
+        let mut state = UnivMon::init_univmon(4, 3, 16, 2);
+        state.fast_insert(&DataInput::U64(1), 1);
+        let bytes = state.serialize_to_bytes().unwrap();
+        assert!(UnivMonAccumulator::from_bytes(&bytes).is_err());
+        state.free();
+        assert!(UnivMonAccumulator::from_bytes(&state.serialize_to_bytes().unwrap()).is_ok());
+    }
+
+    /// Pane merge preserves overlapping keys and reset removes the previous window.
+    #[test]
+    fn merge_and_reset_preserve_frequency_semantics() {
+        let mut left = UnivMonAccumulator::new(32, 5, 1024, 4).unwrap();
+        let mut right = left.clone();
+        for value in [1.0, 2.0] {
+            left.insert_sample(value).unwrap();
+        }
+        for value in [2.0, 3.0] {
+            right.insert_sample(value).unwrap();
+        }
+        let merged = left.merge_with(&right).unwrap();
+        assert_eq!(read(merged.as_ref(), Statistic::Count), 4.0);
+        assert!((read(merged.as_ref(), Statistic::Cardinality) - 3.0).abs() < 0.01);
+        left.reset_to_empty();
+        assert_eq!(read(&left, Statistic::Count), 0.0);
+        assert_eq!(read(&left, Statistic::FrequencyEntropy), 0.0);
+        assert!(left
+            .merge_with(&UnivMonAccumulator::new(16, 5, 1024, 4).unwrap())
+            .is_err());
+    }
+}
diff --git a/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs b/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs
index 187f67aa7..4ae79947a 100644
--- a/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs
+++ b/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs
@@ -91,7 +91,13 @@ impl ResolvedMaterialization<'_> {
                         && (0.0..=1.0).contains(q)
                         && matches!(aggregation_type, DatasketchesKLL | HydraKLL | DDSketch)
                 }
-                QueryReadout::Cardinality => *aggregation_type == HLL,
+                QueryReadout::Cardinality => matches!(aggregation_type, HLL | UnivMon),
+                QueryReadout::FrequencyL2 | QueryReadout::FrequencyEntropy => {
+                    *aggregation_type == UnivMon
+                }
+                QueryReadout::PointCount { value: None, .. } if *aggregation_type == UnivMon => {
+                    true
+                }
                 QueryReadout::PointCount { .. } => matches!(
                     aggregation_type,
                     CountMinSketch | CountMinSketchWithHeap | CountSketch | CountSketchWithHeap
diff --git a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs
index 4132ddb90..75ce06ad1 100644
--- a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs
+++ b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs
@@ -1021,7 +1021,30 @@ fn readout_per_window(
 /// `SketchQuery` -- shared by both the cumulative and per-window readout
 /// paths.
 fn sketch_query_value(rs: &SummaryState, query: &SketchQuery) -> Result<f64, SummaryExecutorError> {
+    if let SummaryState::UnivMon(state) = rs {
+        use crate::storage_engines::types::AggregateCore;
+        let statistic = match query {
+            SketchQuery::Cardinality => asap_types::Statistic::Cardinality,
+            SketchQuery::FrequencyL2 => asap_types::Statistic::FrequencyL2,
+            SketchQuery::FrequencyEntropy => asap_types::Statistic::FrequencyEntropy,
+            SketchQuery::PointCount {
+                key: ColumnRef::SampleValue,
+                value: None,
+            } => asap_types::Statistic::Count,
+            _ => {
+                return Err(SummaryExecutorError::Unsupported(
+                    "unsupported UnivMon readout",
+                ))
+            }
+        };
+        return state
+            .query_statistic(statistic, &None, &Default::default())
+            .map_err(|_| SummaryExecutorError::Unsupported("UnivMon readout failed"));
+    }
     match query {
+        SketchQuery::FrequencyL2 | SketchQuery::FrequencyEntropy => Err(
+            SummaryExecutorError::Unsupported("frequency moment readout requires UnivMon"),
+        ),
         SketchQuery::Quantile { q } => Ok(rs.quantile(*q)),
         SketchQuery::Cardinality => Ok(rs.cardinality()),
         // `key: ColumnRef::SampleValue, value: None` means "no specific
@@ -1102,6 +1125,22 @@ fn summary_family_matches_sketch(
         return false;
     };
     match (sketch.algorithm(), sketch.params(), kind, config) {
+        (
+            SketchAlgorithm::UnivMon,
+            SketchParams::UnivMon {
+                heap_size,
+                sketch_rows,
+                sketch_cols,
+                layers,
+            },
+            SketchAlgorithm::UnivMon,
+            SketchConfig::UnivMon {
+                heap_size: h,
+                sketch_rows: r,
+                sketch_cols: c,
+                layers: l,
+            },
+        ) => heap_size == h && sketch_rows == r && sketch_cols == c && layers == l,
         (
             SketchAlgorithm::DDSketch,
             SketchParams::DDSketch { alpha },
@@ -1253,6 +1292,20 @@ fn to_delta_kind(kind: SketchAlgorithm, config: &SketchConfig) -> Option<DeltaSk
     // heap_size-absent default (`accuracy.rs`).
     const DEFAULT_HEAP_SIZE: usize = 100;
     match (kind, config) {
+        (
+            SketchAlgorithm::UnivMon,
+            SketchConfig::UnivMon {
+                heap_size,
+                sketch_rows,
+                sketch_cols,
+                layers,
+            },
+        ) => Some(DeltaSketchKind::UnivMon {
+            heap_size: *heap_size,
+            sketch_rows: *sketch_rows,
+            sketch_cols: *sketch_cols,
+            layers: *layers,
+        }),
         (SketchAlgorithm::DDSketch, SketchConfig::DDSketch { relative_accuracy }) => {
             Some(DeltaSketchKind::DDSketch {
                 alpha: *relative_accuracy,
@@ -1760,6 +1813,92 @@ mod tests {
     const T0: u64 = 1_000_000;
     const T1: u64 = 2_000_000;
 
+    /// One installed frequency summary merges panes before all four readouts.
+    #[test]
+    fn bound_univmon_merges_panes_for_four_readouts() {
+        use crate::precompute_engine::operators::univmon_accumulator::UnivMonAccumulator;
+        use crate::storage_engines::sketch_db::index::SketchEncoding;
+        use crate::storage_engines::types::SerializableToSink;
+        use control_plane::query_plan::{MaterializationBinding, PhysicalGrouping};
+        let index = SketchStore::new();
+        let fp = asap_types::PolicyFingerprint(701);
+        let mut meta = kll_meta(1, "m", &["job"]);
+        meta.policy_fp = fp;
+        meta.agg_kind = AggKind::Sketch {
+            algorithm: SketchAlgorithm::UnivMon,
+            config: SketchConfig::UnivMon {
+                heap_size: 32,
+                sketch_rows: 5,
+                sketch_cols: 1024,
+                layers: 4,
+            },
+            spatial_filter_canonical: String::new(),
+        };
+        meta.accuracy = None;
+        meta.capability = Some(Capability::CardinalityApprox);
+        index.register(meta);
+        for (start, values) in [(0, [1.0, 2.0]), (1000, [2.0, 3.0])] {
+            let mut state = UnivMonAccumulator::new(32, 5, 1024, 4).unwrap();
+            for value in values {
+                state.insert_sample(value).unwrap();
+            }
+            index.append_sample(
+                1,
+                BTreeMap::from([("job".into(), "a".into())]),
+                (start, start + 1000),
+                SketchSampleState {
+                    bytes: state.serialize_to_bytes(),
+                    encoding: SketchEncoding::MsgpackFull,
+                },
+            );
+        }
+        let context = QueryExecutionContext {
+            index: &index,
+            t0_ms: 0,
+            t1_ms: 2000,
+            is_cumulative: true,
+            allowed_materializations: Some(BTreeSet::from([fp])),
+        };
+        let binding = MaterializationBinding {
+            materialization: fp.into(),
+            output_grouping: PhysicalGrouping::PerEntity,
+            item_labels: vec![],
+            window_ms: 1000,
+            pane_origin_ms: Some(0),
+            readout_lookback_ms: Some(2000),
+        };
+        let states = context.read_bound_materialization(&binding).unwrap();
+        assert_eq!(states.len(), 1);
+        assert_eq!(states[0].0.get("job").unwrap(), "a");
+        for (query, expected) in [
+            (
+                SketchQuery::PointCount {
+                    key: ColumnRef::SampleValue,
+                    value: None,
+                },
+                4.0,
+            ),
+            (SketchQuery::Cardinality, 3.0),
+            (SketchQuery::FrequencyL2, 6.0f64.sqrt()),
+            (SketchQuery::FrequencyEntropy, 1.5),
+        ] {
+            let SummaryValue::Points(points, _) =
+                context.readout_bound(&states[0].1, &query).unwrap()
+            else {
+                panic!("expected scalar points")
+            };
+            assert_eq!(points.len(), 1);
+            assert!(
+                (points[0].1 - expected).abs() < 0.05,
+                "{query:?}: {:?}",
+                points
+            );
+        }
+        let mut unknown = binding;
+        unknown.materialization = asap_types::PolicyFingerprint(702).into();
+        assert!(context.read_bound_materialization(&unknown).is_err());
+    }
+
     fn ctx(index: &SketchStore) -> QueryExecutionContext<'_> {
         QueryExecutionContext {
             index,
diff --git a/data_plane/src/storage_engines/sketch_db/accuracy.rs b/data_plane/src/storage_engines/sketch_db/accuracy.rs
index d9d2e82ed..6134e7695 100644
--- a/data_plane/src/storage_engines/sketch_db/accuracy.rs
+++ b/data_plane/src/storage_engines/sketch_db/accuracy.rs
@@ -85,6 +85,11 @@ impl BackendAccuracyProfile for AccuracyProfile {
     /// The sketch's own theoretical bound, without the GOS staleness term.
     fn derive_sketch_only(config: &AggregationConfig) -> Self {
         match config.aggregation_type {
+            AggregationType::UnivMon => Self {
+                epsilon: f64::MAX,
+                delta: 1.0,
+                kind: AccuracyKind::Uncalibrated,
+            },
             // Exact aggregates. (The `SetAggregator` /
             // `DeltaSetAggregator` exact-set-membership family lived
             // here too before its retirement.)
diff --git a/data_plane/src/storage_engines/sketch_db/data/mod.rs b/data_plane/src/storage_engines/sketch_db/data/mod.rs
index 7c8cf3ef7..d5c423311 100644
--- a/data_plane/src/storage_engines/sketch_db/data/mod.rs
+++ b/data_plane/src/storage_engines/sketch_db/data/mod.rs
@@ -70,11 +70,29 @@ pub use asap_types::AggregationType;
 /// `SketchInstanceMetadata.agg_kind`.
 #[derive(Debug, Clone)]
 pub enum SketchConfig {
-    DDSketch { relative_accuracy: f64 },
-    Kll { k: u32 },
-    Hll { precision: u32 },
-    CountSketch { rows: i32, cols: i32 },
-    CountMin { rows: i32, cols: i32 },
+    UnivMon {
+        heap_size: u32,
+        sketch_rows: u32,
+        sketch_cols: u32,
+        layers: u8,
+    },
+    DDSketch {
+        relative_accuracy: f64,
+    },
+    Kll {
+        k: u32,
+    },
+    Hll {
+        precision: u32,
+    },
+    CountSketch {
+        rows: i32,
+        cols: i32,
+    },
+    CountMin {
+        rows: i32,
+        cols: i32,
+    },
 }
 
 /// What kind of aggregation a `sid` identifies. M2.3 generalization
@@ -172,6 +190,20 @@ pub fn agg_kind_for_config(config: &asap_types::aggregation_config::AggregationC
                 };
                 let algorithm = kind.algorithm().clone();
                 let physical = match (kind.algorithm(), kind.params()) {
+                    (
+                        Algorithm::UnivMon,
+                        SketchParams::UnivMon {
+                            heap_size,
+                            sketch_rows,
+                            sketch_cols,
+                            layers,
+                        },
+                    ) => SketchConfig::UnivMon {
+                        heap_size: *heap_size,
+                        sketch_rows: *sketch_rows,
+                        sketch_cols: *sketch_cols,
+                        layers: *layers,
+                    },
                     (Algorithm::DDSketch, SketchParams::DDSketch { alpha }) => {
                         SketchConfig::DDSketch {
                             relative_accuracy: *alpha,
@@ -257,7 +289,9 @@ impl AggKind {
                     SketchAlgorithm::DDSketch | SketchAlgorithm::Kll => {
                         Capability::QuantileApprox(Some(algorithm.clone()))
                     }
-                    SketchAlgorithm::Hll => Capability::CardinalityApprox,
+                    SketchAlgorithm::Hll | SketchAlgorithm::UnivMon => {
+                        Capability::CardinalityApprox
+                    }
                     SketchAlgorithm::Cms | SketchAlgorithm::CountSketch => {
                         Capability::FrequencyEstimate(Some(algorithm.clone()))
                     }
@@ -348,6 +382,7 @@ fn sketch_algorithm_canonical(k: SketchAlgorithm) -> &'static str {
         SketchAlgorithm::DDSketch => "DDSketch",
         SketchAlgorithm::Kll => "Kll",
         SketchAlgorithm::Hll => "Hll",
+        SketchAlgorithm::UnivMon => "UnivMon",
         SketchAlgorithm::CountSketch => "CountSketch",
         SketchAlgorithm::Cms => "CountMin",
         SketchAlgorithm::CmsWithHeap => "CmsWithHeap",
@@ -363,6 +398,12 @@ fn sketch_algorithm_canonical(k: SketchAlgorithm) -> &'static str {
 
 fn sketch_config_canonical(cfg: &SketchConfig) -> String {
     match cfg {
+        SketchConfig::UnivMon {
+            heap_size,
+            sketch_rows,
+            sketch_cols,
+            layers,
+        } => format!("U:{heap_size}:{sketch_rows}:{sketch_cols}:{layers}"),
         SketchConfig::DDSketch { relative_accuracy } => {
             format!("D:{relative_accuracy}")
         }
@@ -403,6 +444,11 @@ impl AccuracyBound {
     /// error envelope.
     pub fn from_config(cfg: &SketchConfig) -> Self {
         match cfg {
+            // Family dimensions alone do not establish a readout error bound.
+            SketchConfig::UnivMon { .. } => Self {
+                epsilon: f64::MAX,
+                confidence: 0.0,
+            },
             SketchConfig::DDSketch { relative_accuracy } => Self {
                 epsilon: *relative_accuracy,
                 confidence: 1.0,
diff --git a/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs b/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs
index 25b085e13..d3574d782 100644
--- a/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs
+++ b/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs
@@ -228,8 +228,21 @@ fn encode_agg_kind(agg_kind: &AggKind, buf: &mut Vec<u8>) {
                 SketchAlgorithm::CountSketchWithHeap => 7,
                 SketchAlgorithm::Kmv => 8,
                 SketchAlgorithm::Theta => 9,
+                SketchAlgorithm::UnivMon => 10,
             });
             match config {
+                SketchConfig::UnivMon {
+                    heap_size,
+                    sketch_rows,
+                    sketch_cols,
+                    layers,
+                } => {
+                    buf.push(b'U');
+                    buf.extend_from_slice(&heap_size.to_le_bytes());
+                    buf.extend_from_slice(&sketch_rows.to_le_bytes());
+                    buf.extend_from_slice(&sketch_cols.to_le_bytes());
+                    buf.push(*layers);
+                }
                 SketchConfig::DDSketch { relative_accuracy } => {
                     buf.push(b'D');
                     buf.extend_from_slice(&relative_accuracy.to_le_bytes());
diff --git a/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs b/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs
index 51e6e683d..65c0e144c 100644
--- a/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs
+++ b/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs
@@ -68,11 +68,29 @@ pub const SERIES_ID_METADATA_FILE: &str = "sid_metadata.json";
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 #[serde(tag = "type", rename_all = "snake_case")]
 enum SketchConfigRec {
-    DdSketch { relative_accuracy: f64 },
-    Kll { k: u32 },
-    Hll { precision: u32 },
-    CountSketch { rows: i32, cols: i32 },
-    CountMin { rows: i32, cols: i32 },
+    UnivMon {
+        heap_size: u32,
+        sketch_rows: u32,
+        sketch_cols: u32,
+        layers: u8,
+    },
+    DdSketch {
+        relative_accuracy: f64,
+    },
+    Kll {
+        k: u32,
+    },
+    Hll {
+        precision: u32,
+    },
+    CountSketch {
+        rows: i32,
+        cols: i32,
+    },
+    CountMin {
+        rows: i32,
+        cols: i32,
+    },
 }
 
 impl From<&SketchConfig> for SketchConfigRec {
@@ -82,6 +100,17 @@ impl From<&SketchConfig> for SketchConfigRec {
                 relative_accuracy: *relative_accuracy,
             },
             SketchConfig::Kll { k } => SketchConfigRec::Kll { k: *k },
+            SketchConfig::UnivMon {
+                heap_size,
+                sketch_rows,
+                sketch_cols,
+                layers,
+            } => SketchConfigRec::UnivMon {
+                heap_size: *heap_size,
+                sketch_rows: *sketch_rows,
+                sketch_cols: *sketch_cols,
+                layers: *layers,
+            },
             SketchConfig::Hll { precision } => SketchConfigRec::Hll {
                 precision: *precision,
             },
@@ -104,6 +133,17 @@ impl From<&SketchConfigRec> for SketchConfig {
                 relative_accuracy: *relative_accuracy,
             },
             SketchConfigRec::Kll { k } => SketchConfig::Kll { k: *k },
+            SketchConfigRec::UnivMon {
+                heap_size,
+                sketch_rows,
+                sketch_cols,
+                layers,
+            } => SketchConfig::UnivMon {
+                heap_size: *heap_size,
+                sketch_rows: *sketch_rows,
+                sketch_cols: *sketch_cols,
+                layers: *layers,
+            },
             SketchConfigRec::Hll { precision } => SketchConfig::Hll {
                 precision: *precision,
             },
@@ -127,6 +167,7 @@ fn sketch_algorithm_to_str(k: SketchAlgorithm) -> &'static str {
         SketchAlgorithm::DDSketch => "DDSketch",
         SketchAlgorithm::Kll => "Kll",
         SketchAlgorithm::Hll => "Hll",
+        SketchAlgorithm::UnivMon => "UnivMon",
         SketchAlgorithm::CountSketch => "CountSketch",
         SketchAlgorithm::Cms => "CountMin",
         SketchAlgorithm::CmsWithHeap => "CmsWithHeap",
@@ -141,6 +182,7 @@ fn sketch_algorithm_from_str(s: &str) -> Option<SketchAlgorithm> {
         "DDSketch" => SketchAlgorithm::DDSketch,
         "Kll" => SketchAlgorithm::Kll,
         "Hll" => SketchAlgorithm::Hll,
+        "UnivMon" => SketchAlgorithm::UnivMon,
         "CountSketch" => SketchAlgorithm::CountSketch,
         "CountMin" => SketchAlgorithm::Cms,
         "CmsWithHeap" => SketchAlgorithm::CmsWithHeap,
@@ -322,7 +364,7 @@ impl SidMetaRecord {
                 SketchAlgorithm::DDSketch | SketchAlgorithm::Kll => {
                     Capability::QuantileApprox(Some(kind))
                 }
-                SketchAlgorithm::Hll => Capability::CardinalityApprox,
+                SketchAlgorithm::Hll | SketchAlgorithm::UnivMon => Capability::CardinalityApprox,
                 SketchAlgorithm::CountSketch | SketchAlgorithm::Cms => {
                     Capability::FrequencyEstimate(Some(kind))
                 }
diff --git a/data_plane/src/storage_engines/sketch_db/query/delta_apply.rs b/data_plane/src/storage_engines/sketch_db/query/delta_apply.rs
index 58006c5c3..593aeb635 100644
--- a/data_plane/src/storage_engines/sketch_db/query/delta_apply.rs
+++ b/data_plane/src/storage_engines/sketch_db/query/delta_apply.rs
@@ -58,6 +58,12 @@ use crate::storage_engines::sketch_db::query::decoders::{
 /// so those two need the params known up front to allocate it.
 #[derive(Debug, Clone, Copy)]
 pub enum DeltaSketchKind {
+    UnivMon {
+        heap_size: u32,
+        sketch_rows: u32,
+        sketch_cols: u32,
+        layers: u8,
+    },
     DDSketch {
         alpha: f64,
     },
@@ -103,6 +109,20 @@ impl DeltaSketchKind {
     /// (delta-from-empty ⊕ empty = window state).
     fn bootstrap_empty(&self) -> SummaryState {
         match self {
+            Self::UnivMon {
+                heap_size,
+                sketch_rows,
+                sketch_cols,
+                layers,
+            } => SummaryState::UnivMon(
+                crate::precompute_engine::operators::univmon_accumulator::UnivMonAccumulator::new(
+                    *heap_size as usize,
+                    *sketch_rows as usize,
+                    *sketch_cols as usize,
+                    *layers as usize,
+                )
+                .expect("validated UnivMon catalog dimensions"),
+            ),
             DeltaSketchKind::DDSketch { alpha } => SummaryState::Dd(DdSketch::new(*alpha)),
             DeltaSketchKind::Kll { k } => SummaryState::Kll(KllSketch::new(*k as u16)),
             DeltaSketchKind::Hll { precision } => {
@@ -138,6 +158,29 @@ fn decode_full(
     encoding: SketchEncoding,
 ) -> Result<SummaryState, String> {
     match (kind, encoding) {
+        (
+            DeltaSketchKind::UnivMon {
+                heap_size,
+                sketch_rows,
+                sketch_cols,
+                layers,
+            },
+            SketchEncoding::MsgpackFull,
+        ) => {
+            let state = crate::precompute_engine::operators::univmon_accumulator::UnivMonAccumulator::from_bytes(bytes)
+                .map_err(|e| e.to_string())?;
+            if state.dimensions()
+                != (
+                    *heap_size as usize,
+                    *sketch_rows as usize,
+                    *sketch_cols as usize,
+                    *layers as usize,
+                )
+            {
+                return Err("UnivMon payload dimensions differ from installed catalog".into());
+            }
+            Ok(SummaryState::UnivMon(state))
+        }
         (DeltaSketchKind::DDSketch { .. }, SketchEncoding::ProtoFull) => {
             let sk = dd_from_proto(bytes)?;
             Ok(SummaryState::Dd(sk))
@@ -201,6 +244,7 @@ fn decode_full(
 /// folded across a window (or several) via delta application, or merged
 /// in from another sid's own reconstruction.
 pub enum SummaryState {
+    UnivMon(crate::precompute_engine::operators::univmon_accumulator::UnivMonAccumulator),
     Dd(DdSketch),
     Hll(HllSketch),
     Kll(KllSketch),
@@ -235,6 +279,7 @@ impl SummaryState {
             ));
         }
         match self {
+            SummaryState::UnivMon(_) => Err("UnivMon requires full pane snapshots".into()),
             SummaryState::Dd(sk) => {
                 match encoding {
                     // PROTO_DELTA: dispatch on the payload SHAPE, mirroring the
@@ -492,6 +537,9 @@ impl SummaryState {
     /// on `DeltaSketchKind`.
     pub fn merge_same_family(&mut self, other: &SummaryState) -> Result<(), String> {
         match (self, other) {
+            (SummaryState::UnivMon(a), SummaryState::UnivMon(b)) => {
+                a.merge_in_place(b).map_err(|e| e.to_string())
+            }
             (SummaryState::Dd(a), SummaryState::Dd(b)) => {
                 a.merge(b).map_err(|e| format!("merge DDSketch: {e}"))
             }
@@ -523,6 +571,7 @@ impl SummaryState {
     /// Diagnostic family name for error messages — not used for dispatch.
     fn family_name(&self) -> &'static str {
         match self {
+            SummaryState::UnivMon(_) => "UnivMon",
             SummaryState::Dd(_) => "DDSketch",
             SummaryState::Hll(_) => "Hll",
             SummaryState::Kll(_) => "Kll",
diff --git a/data_plane/src/storage_engines/sketch_db/query/timeline.rs b/data_plane/src/storage_engines/sketch_db/query/timeline.rs
index 44364613f..51329ad48 100644
--- a/data_plane/src/storage_engines/sketch_db/query/timeline.rs
+++ b/data_plane/src/storage_engines/sketch_db/query/timeline.rs
@@ -264,11 +264,24 @@ fn sketch_algorithm_byte(k: SketchAlgorithm) -> u8 {
         SketchAlgorithm::CountSketchWithHeap => 7,
         SketchAlgorithm::Kmv => 8,
         SketchAlgorithm::Theta => 9,
+        SketchAlgorithm::UnivMon => 10,
     }
 }
 
 fn encode_sketch_config(cfg: &SketchConfig, buf: &mut Vec<u8>) {
     match cfg {
+        SketchConfig::UnivMon {
+            heap_size,
+            sketch_rows,
+            sketch_cols,
+            layers,
+        } => {
+            buf.push(b'U');
+            buf.extend_from_slice(&heap_size.to_le_bytes());
+            buf.extend_from_slice(&sketch_rows.to_le_bytes());
+            buf.extend_from_slice(&sketch_cols.to_le_bytes());
+            buf.push(*layers);
+        }
         SketchConfig::DDSketch { relative_accuracy } => {
             buf.push(b'D');
             buf.extend_from_slice(&relative_accuracy.to_le_bytes());
diff --git a/data_plane/src/storage_engines/sketch_db/query/timeline_dispatch.rs b/data_plane/src/storage_engines/sketch_db/query/timeline_dispatch.rs
index dca578231..56c68f9dd 100644
--- a/data_plane/src/storage_engines/sketch_db/query/timeline_dispatch.rs
+++ b/data_plane/src/storage_engines/sketch_db/query/timeline_dispatch.rs
@@ -129,6 +129,8 @@ pub fn combine_statistic(
             .fold(None, |acc, v| Some(acc.map_or(v, |a: f64| a.max(v)))),
         // Non-combinable at the scalar level — see module doc.
         Statistic::Cardinality
+        | Statistic::FrequencyL2
+        | Statistic::FrequencyEntropy
         | Statistic::Increase
         | Statistic::Rate
         | Statistic::Quantile
diff --git a/data_plane/src/tests/prometheus_forwarding_tests.rs b/data_plane/src/tests/prometheus_forwarding_tests.rs
index 7e11342b3..e35c227fc 100644
--- a/data_plane/src/tests/prometheus_forwarding_tests.rs
+++ b/data_plane/src/tests/prometheus_forwarding_tests.rs
@@ -10,7 +10,7 @@ use tokio::net::TcpListener;
 use tokio::time::{sleep, Duration};
 
 /// Mock Prometheus server for testing
-async fn start_mock_prometheus_server(port: u16) -> Result<(), Box<dyn std::error::Error>> {
+async fn start_mock_prometheus_server() -> Result<u16, Box<dyn std::error::Error>> {
     use axum::{extract::Query, response::Json, routing::get, Router};
     use serde_json::json;
     use std::collections::HashMap;
@@ -51,7 +51,8 @@ async fn start_mock_prometheus_server(port: u16) -> Result<(), Box<dyn std::erro
 
     let app = Router::new().route("/api/v1/query", get(mock_query_handler));
 
-    let listener = TcpListener::bind(format!("127.0.0.1:{port}")).await?;
+    let listener = TcpListener::bind("127.0.0.1:0").await?;
+    let port = listener.local_addr()?.port();
 
     tokio::spawn(async move {
         axum::serve(listener, app).await.unwrap();
@@ -59,7 +60,7 @@ async fn start_mock_prometheus_server(port: u16) -> Result<(), Box<dyn std::erro
 
     // Give the server time to start
     sleep(Duration::from_millis(100)).await;
-    Ok(())
+    Ok(port)
 }
 
 async fn setup_test_server(prometheus_port: u16) -> (HttpServer, u16) {
@@ -88,8 +89,7 @@ async fn setup_test_server(prometheus_port: u16) -> (HttpServer, u16) {
 #[tokio::test]
 async fn test_prometheus_forwarding_instant_query() {
     // Start mock Prometheus server
-    let prometheus_port = 19090;
-    start_mock_prometheus_server(prometheus_port).await.unwrap();
+    let prometheus_port = start_mock_prometheus_server().await.unwrap();
 
     // Start our HTTP server with forwarding enabled
     let (_server, server_port) = setup_test_server(prometheus_port).await;
@@ -119,8 +119,7 @@ async fn test_prometheus_forwarding_instant_query() {
 #[tokio::test]
 async fn test_prometheus_forwarding_error_handling() {
     // Start mock Prometheus server
-    let prometheus_port = 19092;
-    start_mock_prometheus_server(prometheus_port).await.unwrap();
+    let prometheus_port = start_mock_prometheus_server().await.unwrap();
 
     // Start our HTTP server with forwarding enabled
     let (_server, server_port) = setup_test_server(prometheus_port).await;
diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs
index d5f536111..5b5e1e8a5 100644
--- a/data_plane/tests/asapquery_compatibility_process_e2e.rs
+++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs
@@ -21,6 +21,8 @@ use tokio::sync::Mutex;
 #[path = "support/erp_planning_process.rs"]
 mod erp_planning_process;
 
+#[path = "support/distinct_planning_process.rs"]
+mod distinct_planning_process;
 #[path = "support/durable_summary_process.rs"]
 mod durable_summary_process;
 
diff --git a/data_plane/tests/support/distinct_planning_process.rs b/data_plane/tests/support/distinct_planning_process.rs
new file mode 100644
index 000000000..123624db7
--- /dev/null
+++ b/data_plane/tests/support/distinct_planning_process.rs
@@ -0,0 +1,168 @@
+use super::*;
+use control_plane::physical::compiler::BackendLocalPlanningSnapshot;
+
+/// The production compiler, ingest engine and query DAG preserve distinct populations.
+#[tokio::test]
+async fn distinct_range_uses_planner_selected_hll_and_source_labels() {
+    const QUERY: &str = "distinct_over_time(distinct_values{job=\"api\"}[5s])";
+    let fallback_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let fallback_url = format!("http://{}", fallback_listener.local_addr().unwrap());
+    let fallback_task = tokio::spawn(async move {
+        axum::serve(
+            fallback_listener,
+            Router::new().route("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/-/healthy", get(|| async { "healthy" })),
+        )
+        .await
+        .unwrap();
+    });
+    let mut fixture: Value = serde_json::from_str(include_str!(
+        "../../../docs/examples/asapquery-compatibility-demo-snapshot.json"
+    ))
+    .unwrap();
+    let mut entry = fixture["query_workload"]["repeating_queries"][3].clone();
+    entry["query"] = QUERY.into();
+    entry["requirements"]["accuracy"] = serde_json::json!({"explicit": {"Epsilon": 0.05}});
+    fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]);
+    let plan = serde_json::from_value::<BackendLocalPlanningSnapshot>(fixture.clone())
+        .unwrap()
+        .compile()
+        .unwrap();
+    assert_eq!(plan.precompute_plan.materializations.len(), 1);
+    assert_eq!(
+        plan.precompute_plan.materializations[0].aggregation_type,
+        asap_types::AggregationType::HLL
+    );
+    eprintln!(
+        "DISTINCT_PLANNED {}",
+        serde_json::json!({"materializations": plan.precompute_plan.materializations, "query_plan": plan.query_plan, "lifecycle_estimates": plan.lifecycle_estimates})
+    );
+    let output = tempfile::tempdir().unwrap();
+    let path = output.path().join("planning.json");
+    std::fs::write(&path, serde_json::to_vec(&fixture).unwrap()).unwrap();
+    let port = unused_port();
+    let mut vm_port = unused_port();
+    while vm_port == port {
+        vm_port = unused_port();
+    }
+    let mut child = ChildGuard(
+        Command::new(env!("CARGO_BIN_EXE_data_plane"))
+            .args([
+                "--forward-unsupported-queries",
+                "--prometheus-server",
+                &fallback_url,
+                "--profile",
+                "asapquery",
+                "--planning-snapshot",
+            ])
+            .arg(&path)
+            .args(["--http-port", &port.to_string(), "--output-dir"])
+            .arg(output.path())
+            .args([
+                "--victoriametrics-http-port",
+                &vm_port.to_string(),
+                "--victoriametrics-url",
+                &fallback_url,
+            ])
+            .args([
+                "--precompute-allowed-lateness-ms",
+                "0",
+                "--precompute-flush-interval-ms",
+                "25",
+            ])
+            .stdout(Stdio::null())
+            .stderr(Stdio::inherit())
+            .spawn()
+            .unwrap(),
+    );
+    let client = reqwest::Client::new();
+    let backend = format!("http://127.0.0.1:{port}");
+    wait_until_ready(&client, &format!("{backend}/api/v1/health"), &mut child.0).await;
+    // Source syntax uses the shared parser fork; serving semantics and exact
+    // routing belong to the MetricsQL adapter and its installed query entries.
+    let snapshot = serde_json::from_value::<BackendLocalPlanningSnapshot>(fixture).unwrap();
+    let (mut request, mut environment) = snapshot.planning_request().unwrap();
+    request.hybrid_execution = false;
+    environment.plan_version = 2;
+    let compiled = control_plane::physical::compiler::PhysicalCompiler
+        .compile_metricsql(request, environment)
+        .unwrap();
+    let identity = serde_json::json!({"plan_id": compiled.envelope.plan_id, "plan_version": compiled.envelope.plan_version});
+    let install = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest {
+        summary_catalog: compiled.summary_catalog,
+        collector_plans: compiled.collector_plans,
+        precompute_plan: compiled.precompute_plan,
+        transmission_plan: compiled.transmission_plan,
+        query_plan: compiled.query_plan,
+        storage_routing: None,
+        adaptation_evidence: vec![],
+    };
+    eprintln!(
+        "DISTINCT_INSTALLED {}",
+        serde_json::to_string(&install).unwrap()
+    );
+    let response = client
+        .post(format!("{backend}/api/v1/physical-plan"))
+        .json(&install)
+        .send()
+        .await
+        .unwrap();
+    assert!(
+        response.status().is_success(),
+        "{}",
+        response.text().await.unwrap()
+    );
+    let response = client
+        .post(format!("{backend}/api/v1/physical-plan/activate"))
+        .json(&identity)
+        .send()
+        .await
+        .unwrap();
+    assert!(
+        response.status().is_success(),
+        "{}",
+        response.text().await.unwrap()
+    );
+    let now = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_millis() as i64;
+    let base = now - now.rem_euclid(5000) - 20000;
+    let mut series = Vec::new();
+    for (instance, distinct, job) in [("a", 5, "api"), ("b", 13, "api"), ("excluded", 23, "other")]
+    {
+        let mut samples: Vec<_> = (0..100)
+            .map(|i| (base + 1 + i, (i % distinct) as f64))
+            .collect();
+        samples.push((base + 15001, 1000.0));
+        series.push(series_with_labels(
+            "distinct_values",
+            &[("instance", instance), ("job", job)],
+            &samples,
+        ));
+    }
+    assert_eq!(
+        remote_write(&client, &backend, &WriteRequest { timeseries: series }).await,
+        204
+    );
+    drain_precompute(&client, &backend).await;
+    let result = wait_for_warm_instant(
+        &client,
+        &format!("http://127.0.0.1:{vm_port}"),
+        QUERY,
+        (base + 5000) as f64 / 1000.0,
+        &output.path().join("query_engine.log"),
+    )
+    .await;
+    let rows = result["data"]["result"].as_array().unwrap();
+    assert_eq!(rows.len(), 2, "{result}");
+    for (instance, exact) in [("a", 5.0), ("b", 13.0)] {
+        let row = rows
+            .iter()
+            .find(|row| row["metric"]["instance"] == instance)
+            .unwrap();
+        let estimate = row["value"][1].as_str().unwrap().parse::<f64>().unwrap();
+        assert!((estimate - exact).abs() / exact <= 0.05, "{result}");
+    }
+    eprintln!("DISTINCT_WARM {result}");
+    fallback_task.abort();
+}
diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md
index 4d8df5a09..048dcbac4 100644
--- a/docs/design_docs/summary-catalog-sds-architecture.md
+++ b/docs/design_docs/summary-catalog-sds-architecture.md
@@ -272,6 +272,22 @@ contract separately identifies the supported rank-error bound or versioned
 bound derivation, probability of failure, readout scope and required conditions.
 If that guarantee is unavailable, fidelity is explicitly `Unknown`.
 
+For a shared UnivMon state, `heap_size`, `sketch_rows`, `sketch_cols`, and
+`layers` describe one configuration. They do not establish one error bound for
+all readouts. The backend's `UnivMonFrequency` contract records these parameters
+and the unit-frequency update domain: each sample value contributes one
+occurrence. Total count is exact in that domain; distinct count, frequency L2,
+and frequency entropy require their own accuracy evidence. Frequency L2 means
+`sqrt(sum(frequency(key)^2))`; entropy is measured in bits.
+
+ERP evidence must state the readout's units: relative error for distinct and L2,
+and absolute bits error for entropy. A measured error is not a certified failure
+probability. Readouts may share state only when their configuration and data
+population match and each readout's accuracy requirements are satisfied. A
+small configuration suitable for L2 may therefore be unsuitable for entropy.
+Completeness of the input window remains a separate requirement for every
+readout, including exact count.
+
 A `FidelityGuarantee` contains:
 
 - The applicable operation and error quantity, such as quantile rank error.
diff --git a/docs/developer_docs/univmon-dependency.md b/docs/developer_docs/univmon-dependency.md
new file mode 100644
index 000000000..ae0df030c
--- /dev/null
+++ b/docs/developer_docs/univmon-dependency.md
@@ -0,0 +1,15 @@
+# UnivMon dependency checkout
+
+The UnivMon runtime requires asap_sketchlib commit
+`c0de315754f9a6c77dd7a25aca0b2b62f0aec276` from PR #139. CI checks
+out this exact revision. This dependency is not yet merged into sketchlib main.
+
+For local builds, use that revision in the sibling `asap_sketchlib` checkout
+referenced by the workspace Cargo patch and ASAPCollector's precompute crate.
+Both consumers must resolve to the same checkout so their state types agree.
+Use an isolated checkout when other work depends on a different revision.
+
+The dependency exposes the standard-update compatibility predicate. Restoring
+a terminal-mode UnivMon state into an ingest accumulator is rejected before
+mutation. After #139 merges, update the pin explicitly and rerun restore and
+real-process tests; do not silently substitute an unverified main revision.