diff --git a/data_plane/src/precompute_engine/operators/count_min_sketch_with_heap_accumulator.rs b/data_plane/src/precompute_engine/operators/count_min_sketch_with_heap_accumulator.rs
index 9200e53a..59f12bbb 100644
--- a/data_plane/src/precompute_engine/operators/count_min_sketch_with_heap_accumulator.rs
+++ b/data_plane/src/precompute_engine/operators/count_min_sketch_with_heap_accumulator.rs
@@ -246,6 +246,46 @@ impl CountMinSketchWithHeapAccumulator {
         Err("deserialize_from_bytes for CountMinSketchWithHeapAccumulator not implemented".into())
     }
 
+    /// VALUE-WEIGHTED heavy-hitter update (FIX: CountSketch/CMS topk
+    /// recall-0). The default ingest path inserts `+1` per occurrence keyed
+    /// by the raw `item`, so the heap ranks groups by OCCURRENCE COUNT — the
+    /// wrong answer for `topk(k, sum by (label) (metric))`, which asks for
+    /// the top groups by SUM OF VALUE. This update adds the sample `value`
+    /// (not `+1`) into both the CMS matrix and the top-k heap, keyed by the
+    /// GROUP LABEL (e.g. the `host` / `zone` value), so the heap's ranking is
+    /// by summed value. Repeated calls for the same `group_label` accumulate,
+    /// so after folding a window the heap holds Σvalue per group.
+    ///
+    /// Delegates to the library's value-weighted `CountMinSketchWithHeap::
+    /// update(key, value)` (`sketchlib_cms_heap_update` → `insert_many(key,
+    /// round(value))`), which is the "separate update path" the evaluation
+    /// plan (Fig 3c) called for.
+    pub fn insert_value(&mut self, group_label: &str, value: f64) {
+        self.inner.update(group_label, value);
+    }
+
+    /// Read the top-`k` GROUPS ranked by summed VALUE (descending), keyed by
+    /// the group label. Pairs with [`Self::insert_value`]: the heap built by
+    /// value-weighted updates ranks by Σvalue, so this returns the
+    /// value-weighted top-k (not the occurrence-count top-k the raw `item`
+    /// heap would give). Sorted descending by value; ties broken by key for
+    /// determinism; truncated to `k`.
+    pub fn topk_by_value(&self, k: usize) -> Vec<(String, f64)> {
+        let mut items: Vec<(String, f64)> = self
+            .inner
+            .topk_heap_items()
+            .into_iter()
+            .map(|it| (it.key, it.value))
+            .collect();
+        items.sort_by(|a, b| {
+            b.1.partial_cmp(&a.1)
+                .unwrap_or(std::cmp::Ordering::Equal)
+                .then_with(|| a.0.cmp(&b.0))
+        });
+        items.truncate(k);
+        items
+    }
+
     /// Get all keys from the top-k heap.
     pub fn get_topk_keys(&self) -> Vec<KeyByLabelValues> {
         self.inner
@@ -681,4 +721,112 @@ mod tests {
         let w = W(true, (rows, cols, cells), heap_owned, heap_size);
         rmp_serde::to_vec(&w).expect("encode delta-heap")
     }
+
+    // ----------------------------------------------------------------
+    // FIX 1 — VALUE-WEIGHTED top-k (recall 0 → correct).
+    //
+    // `topk(k, sum by (host) (cpu_load))` asks for the top-k hosts by
+    // SUM OF VALUE. The heavy-hitter heap built by the default `+1`-per-
+    // occurrence update ranks by COUNT keyed by `item`, so its recall
+    // against the value-weighted ground truth is 0 when the busiest host
+    // (most samples) is NOT the heaviest host (largest Σvalue).
+    // `insert_value(group_label, value)` adds the sample VALUE keyed by the
+    // GROUP LABEL, so `topk_by_value` ranks by Σvalue — correct recall.
+    // ----------------------------------------------------------------
+
+    /// Crafted adversarial dataset: the host with the MOST samples
+    /// (`h_chatty`, 100 tiny samples) is NOT the host with the largest
+    /// value-sum (`h_heavy`, a handful of huge samples). A COUNT-ranked
+    /// heap would surface `h_chatty`; the value-weighted top-k must surface
+    /// the true heavy hitters by Σvalue, giving recall 1.0 against the
+    /// ground-truth top-k-by-value-sum.
+    #[test]
+    fn value_weighted_topk_has_full_recall_vs_count_topk() {
+        // (host, per-sample value, sample count) → true Σvalue:
+        //   h_heavy : 1000 × 3   = 3000   (few samples, huge value)
+        //   h_mid   : 200  × 5   = 1000
+        //   h_small : 50   × 6   = 300
+        //   h_chatty: 1    × 100 = 100    (MOST samples, tiny value)
+        let data: &[(&str, f64, usize)] = &[
+            ("h_heavy", 1000.0, 3),
+            ("h_mid", 200.0, 5),
+            ("h_small", 50.0, 6),
+            ("h_chatty", 1.0, 100),
+        ];
+
+        // Wide CMS + heap large enough to hold every group exactly (4 groups)
+        // so the estimate equals the true Σvalue with no hash collisions.
+        let mut acc = CountMinSketchWithHeapAccumulator::new(5, 4096, 16);
+        let mut truth: std::collections::HashMap<&str, f64> = std::collections::HashMap::new();
+        for (host, value, count) in data {
+            for _ in 0..*count {
+                acc.insert_value(host, *value);
+            }
+            *truth.entry(*host).or_insert(0.0) += value * (*count as f64);
+        }
+
+        // Ground-truth top-2 by value-sum: h_heavy (3000), h_mid (1000).
+        let mut truth_ranked: Vec<(&str, f64)> = truth.into_iter().collect();
+        truth_ranked.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap());
+        let truth_top2: std::collections::HashSet<&str> =
+            truth_ranked.iter().take(2).map(|(k, _)| *k).collect();
+        assert!(
+            truth_top2.contains("h_heavy") && truth_top2.contains("h_mid"),
+            "ground-truth top-2 by value-sum should be h_heavy + h_mid"
+        );
+
+        // Value-weighted top-2 from the heap.
+        let got = acc.topk_by_value(2);
+        assert_eq!(got.len(), 2, "k=2 → two groups: {got:?}");
+        let got_keys: std::collections::HashSet<&str> =
+            got.iter().map(|(k, _)| k.as_str()).collect();
+
+        // RECALL = |got ∩ truth| / |truth| must be 1.0.
+        let hits = got_keys.intersection(&truth_top2).count();
+        let recall = hits as f64 / truth_top2.len() as f64;
+        assert_eq!(
+            recall, 1.0,
+            "value-weighted top-k recall must be 1.0 (count-ranked heap would \
+             surface h_chatty and miss h_heavy → recall < 1): got={got:?}"
+        );
+
+        // The busiest-by-count host (h_chatty) must NOT be in the top-2,
+        // proving we rank by value-sum, not occurrence count.
+        assert!(
+            !got_keys.contains("h_chatty"),
+            "h_chatty (most samples, smallest value-sum) must be excluded: {got:?}"
+        );
+
+        // Estimates are exact here (no collisions, heap holds all groups):
+        // top-1 must be h_heavy with Σvalue 3000.
+        assert_eq!(got[0].0, "h_heavy");
+        assert!(
+            (got[0].1 - 3000.0).abs() < 1e-6,
+            "h_heavy value-sum estimate ≈ 3000, got {}",
+            got[0].1
+        );
+        assert_eq!(got[1].0, "h_mid");
+        assert!(
+            (got[1].1 - 1000.0).abs() < 1e-6,
+            "h_mid value-sum estimate ≈ 1000, got {}",
+            got[1].1
+        );
+    }
+
+    /// A single value-weighted insert must put the full value (not +1) into
+    /// the heap, and repeated inserts for the same group must accumulate.
+    #[test]
+    fn insert_value_accumulates_summed_value_in_heap() {
+        let mut acc = CountMinSketchWithHeapAccumulator::new(4, 1024, 8);
+        acc.insert_value("g", 10.0);
+        acc.insert_value("g", 25.0);
+        let top = acc.topk_by_value(1);
+        assert_eq!(top.len(), 1);
+        assert_eq!(top[0].0, "g");
+        assert!(
+            (top[0].1 - 35.0).abs() < 1e-6,
+            "summed value should be 35 (10+25), got {}",
+            top[0].1
+        );
+    }
 }
diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs
index 6f0dc21d..85246144 100644
--- a/data_plane/src/query_engines/asap_query_engine/engine.rs
+++ b/data_plane/src/query_engines/asap_query_engine/engine.rs
@@ -656,7 +656,7 @@ impl ASAPQueryEngine {
         query: &str,
         start_ms: u64,
         end_ms: u64,
-        _step_ms: u64,
+        step_ms: u64,
     ) -> Result<crate::query_engines::query_result::QueryResult, crate::query_engines::EngineError>
     {
         let Some(idx) = self.sketch_index.as_ref() else {
@@ -894,7 +894,37 @@ impl ASAPQueryEngine {
         })?;
 
         // Matrix shape — the range_query wire format requires it.
-        Ok(asap_tier_result_to_query_result(result, end_ms, true))
+        let warm_qr = asap_tier_result_to_query_result(result.clone(), end_ms, true);
+
+        // FIX 3 — coverage-aware warm+archive HYBRID STITCH for RANGE
+        // queries. The instant path (`execute`) already stitches when warm
+        // coverage is narrower than the request; the range path historically
+        // returned warm-only, so a request `[start_ms, end_ms]` whose warm
+        // sketches only cover a suffix `[cov_lo, cov_hi]` lost the
+        // prefix `[start_ms, cov_lo)` (the live "No result" / incomplete
+        // matrix symptom). When the reducer reports a coverage narrower than
+        // the requested range AND an archive engine is wired, fetch the
+        // archive's range answer over the SAME window and stitch them by
+        // (label_values, timestamp) — warm wins on overlap, archive fills the
+        // uncovered prefix/suffix. Mirrors the instant-path logic at the
+        // `execute` trait surface.
+        if let (Some((cov_lo, cov_hi)), Some(archive)) =
+            (result.coverage, self.archive_engine.as_ref())
+        {
+            if cov_lo > start_ms || cov_hi < end_ms {
+                if let Ok(archive_qr) = archive
+                    .execute_range(query, start_ms, end_ms, step_ms)
+                    .await
+                {
+                    return Ok(stitch_warm_and_archive(
+                        warm_qr, archive_qr, cov_lo, cov_hi,
+                    ));
+                }
+                // Archive error → fall back to warm-only (best effort).
+            }
+        }
+
+        Ok(warm_qr)
     }
 }
 
@@ -1703,6 +1733,25 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu
                             now_ms,
                             accumulate_windows,
                         ),
+                        // FIX 2 — GLOBAL HLL distinct rollup. `count(hll_metric)`
+                        // with NO `by (...)` (empty group_by_keys + outer Count)
+                        // is the distinct-UNION-cardinality idiom: MERGE the
+                        // per-series HLL registers (register-wise max) across all
+                        // matched sids and estimate ONCE. The per-series
+                        // `evaluate_for_capability` path would otherwise emit one
+                        // estimate per series (double-counting overlaps / never
+                        // producing the single global number). Only the GLOBAL
+                        // (no-`by`) shape is rerouted; `count by (zone) (...)`
+                        // keeps the per-group per-series path below.
+                        crate::storage_engines::sketch_db::index::Capability::CardinalityApprox
+                            if candidate.group_by_keys.is_empty()
+                                && matches!(
+                                    candidate.outer_agg,
+                                    control_plane::asap_tier_analysis::OuterAgg::Count(_)
+                                ) =>
+                        {
+                            reducer.evaluate_cardinality_global(&hit_sids, t0_ms, now_ms)
+                        }
                         // P2-4 (typed dispatch): route off the typed
                         // `required_capability` rather than the
                         // function-name-string detour.
@@ -2957,6 +3006,120 @@ mod asap_tier_classify_tests {
         );
     }
 
+    /// Encode an HLL FULL proto frame over an EXPLICIT set of string items,
+    /// so a test can craft overlapping / disjoint distinct sets across
+    /// series and compute the TRUE union cardinality.
+    fn encode_hll_from_items(precision: u32, items: &[String]) -> Vec<u8> {
+        use asap_sketchlib::proto::sketchlib::{
+            sketch_envelope, HllVariant as ProtoVariant, HyperLogLogState, SketchEnvelope,
+        };
+        use asap_sketchlib::{HllSketch, HllVariant};
+        use prost::Message;
+        let mut sk = HllSketch::new(HllVariant::Regular, precision);
+        for it in items {
+            sk.update(it.as_bytes());
+        }
+        let state = HyperLogLogState {
+            variant: ProtoVariant::Regular as i32,
+            precision: sk.precision,
+            registers: sk.registers.clone(),
+            hip_kxq0: sk.hip_kxq0,
+            hip_kxq1: sk.hip_kxq1,
+            hip_est: sk.hip_est,
+            registers_sparse: None,
+        };
+        SketchEnvelope {
+            sketch_state: Some(sketch_envelope::SketchState::Hll(state)),
+            ..Default::default()
+        }
+        .encode_to_vec()
+    }
+
+    /// FIX 2 — GLOBAL HLL distinct rollup. `count(hll_metric)` with no `by`
+    /// must MERGE the per-series HLL registers (register-wise max) across ALL
+    /// matched series and estimate ONCE — the distinct UNION cardinality. Two
+    /// series share an overlapping prefix of items and each carry disjoint
+    /// items, so summing per-series estimates would over-count the overlap.
+    /// The merged global estimate must land within HLL error of the true
+    /// union, and be strictly below the naive per-series sum.
+    #[tokio::test]
+    async fn execute_count_hll_global_merges_registers_across_series() {
+        let idx = Arc::new(SketchStore::new());
+        let now_ms = std::time::SystemTime::now()
+            .duration_since(std::time::SystemTime::UNIX_EPOCH)
+            .map(|d| d.as_millis() as u64)
+            .unwrap_or(0);
+        let w_start = now_ms.saturating_sub(3_000);
+        let w_end = now_ms.saturating_sub(2_000);
+
+        // Series A: items 0..600. Series B: items 400..1000.
+        // Overlap = [400,600) = 200 items; true union = [0,1000) = 1000.
+        let precision = 12u32; // ~1.6% standard error
+        let a_items: Vec<String> = (0..600).map(|i| format!("u-{i}")).collect();
+        let b_items: Vec<String> = (400..1000).map(|i| format!("u-{i}")).collect();
+        let true_union = 1000.0_f64;
+
+        for (sid, items) in [(8200u64, &a_items), (8201u64, &b_items)] {
+            let mut meta = hll_meta(sid, "unique_users_global");
+            meta.agg_kind = crate::storage_engines::sketch_db::index::AggKind::Sketch {
+                kind: SketchKindHandle::Hll,
+                config: SketchConfig::Hll { precision },
+                spatial_filter_canonical: String::new(),
+            };
+            idx.register(meta);
+            idx.append_sample(
+                sid,
+                BTreeMap::new(),
+                (w_start, w_end),
+                SketchSampleState {
+                    bytes: encode_hll_from_items(precision, items),
+                    encoding: crate::storage_engines::sketch_db::index::SketchEncoding::ProtoFull,
+                },
+            );
+        }
+
+        let engine = build_engine_with_index(idx);
+        let result = engine
+            .execute("count(unique_users_global)")
+            .await
+            .expect("global count(hll_metric) must answer, not capability-miss");
+
+        // GLOBAL distinct is a single scalar — exactly one element.
+        let est = match &result {
+            crate::query_engines::query_result::QueryResult::Vector(v) => {
+                assert_eq!(
+                    v.values.len(),
+                    1,
+                    "global count() must collapse to ONE merged estimate, got {} \
+                     (per-series leak): {v:?}",
+                    v.values.len()
+                );
+                v.values[0].value
+            }
+            crate::query_engines::query_result::QueryResult::Matrix(m) => {
+                assert_eq!(m.values.len(), 1, "one merged series");
+                m.values[0].samples.last().map(|s| s.value).unwrap_or(0.0)
+            }
+        };
+
+        // Within HLL error of the true union (p=12 → ~1.04/sqrt(2^12) ≈ 1.6%;
+        // allow a generous 8% band for the estimator's finite-sample noise).
+        let rel_err = (est - true_union).abs() / true_union;
+        assert!(
+            rel_err < 0.08,
+            "global merged estimate {est} must be within HLL error of the \
+             true union {true_union} (rel_err {rel_err:.4})"
+        );
+
+        // And strictly below the naive per-series sum (600 + 600 = 1200),
+        // proving registers were MERGED (max), not the estimates SUMMED.
+        assert!(
+            est < 1150.0,
+            "merged global estimate {est} must be well below the per-series \
+             sum (~1200) — proves register-merge, not estimate-sum"
+        );
+    }
+
     /// REPRODUCTION (root-cause hunt): `quantile_over_time(0.99,
     /// http_requests_total_latency_ms[30s])` end-to-end via
     /// `execute(&str)` against a now-anchored KLL sid carrying real
@@ -4453,3 +4616,276 @@ mod hybrid_stitch_tests {
         assert_eq!(b.samples.len(), 2);
     }
 }
+
+// ---------------------------------------------------------------------------
+// FIX 3 — RANGE-query warm+archive hybrid stitch.
+//
+// The instant path already stitches; the range path historically returned
+// warm-only, so a `[start, end]` request whose warm sketches only cover a
+// suffix lost the prefix. These tests drive `execute_range_promql_modern`
+// with an archive engine wired and warm coverage narrower than the request,
+// and assert the stitched matrix covers the FULL range (prefix from archive,
+// suffix from warm).
+// ---------------------------------------------------------------------------
+#[cfg(test)]
+mod range_stitch_tests {
+    use super::*;
+    use crate::query_engines::query_result::{
+        QueryResult, RangeVectorElement, Sample,
+    };
+    use crate::query_engines::routing::query_engine_routing::{
+        EngineCapabilities, QueryEngine,
+    };
+    use crate::query_engines::EngineError;
+    use crate::storage_engines::sketch_db::index::{
+        AccuracyBound, Capability, SketchConfig, SketchEncoding, SketchInstanceMetadata,
+        SketchKindHandle, SketchSampleState, SketchStore,
+    };
+    use crate::storage_engines::types::{HotReloadStreamingConfig, KeyByLabelValues};
+    use async_trait::async_trait;
+    use std::collections::{BTreeMap, BTreeSet};
+
+    /// Mock archive engine: returns a fixed full-range matrix for any range
+    /// query, so the stitch can pull the uncovered prefix from it.
+    struct FakeArchive {
+        matrix: QueryResult,
+    }
+
+    #[async_trait]
+    impl QueryEngine for FakeArchive {
+        async fn execute(&self, _query: &str) -> Result<QueryResult, EngineError> {
+            Ok(self.matrix.clone())
+        }
+        async fn execute_range(
+            &self,
+            _query: &str,
+            _start_ms: u64,
+            _end_ms: u64,
+            _step_ms: u64,
+        ) -> Result<QueryResult, EngineError> {
+            Ok(self.matrix.clone())
+        }
+        fn capabilities(&self) -> EngineCapabilities {
+            EngineCapabilities {
+                data_source_id: asap_types::StorageBackend::GorillaObjectStore.data_source_id(),
+                storage_backend: asap_types::StorageBackend::GorillaObjectStore,
+                supports_streams_above_bytes: usize::MAX,
+            }
+        }
+    }
+
+    /// Encode a CountMin FULL proto frame whose row 0 sums to `total`
+    /// (the per-window frequency TOTAL the `count_over_time` reducer reads).
+    fn cms_bytes(total: i64) -> Vec<u8> {
+        use asap_sketchlib::proto::sketchlib::{
+            sketch_envelope, CountMinState, CounterType, SketchEnvelope,
+        };
+        use prost::Message;
+        let (rows, cols) = (2u32, 4u32);
+        let mut counts_int = vec![0i64; (rows * cols) as usize];
+        counts_int[0] = total;
+        let state = CountMinState {
+            rows,
+            cols,
+            counter_type: CounterType::Int64 as i32,
+            counts_int,
+            ..Default::default()
+        };
+        SketchEnvelope {
+            sketch_state: Some(sketch_envelope::SketchState::CountMin(state)),
+            ..Default::default()
+        }
+        .encode_to_vec()
+    }
+
+    /// A CountMin FrequencyEstimate sid — `count_over_time` over it emits one
+    /// PER-WINDOW sample (not a single cumulative scalar), which is what the
+    /// range stitch needs so warm contributes one value per covered window.
+    fn cms_meta(sid: u64, metric: &str) -> SketchInstanceMetadata {
+        let cfg = SketchConfig::CountMin { rows: 2, cols: 4 };
+        SketchInstanceMetadata {
+            sid,
+            metric_name: metric.to_string(),
+            group_by_keys: BTreeSet::new(),
+            capability: Some(Capability::FrequencyEstimate(SketchKindHandle::CountMin)),
+            agg_kind: crate::storage_engines::sketch_db::index::AggKind::Sketch {
+                kind: SketchKindHandle::CountMin,
+                config: cfg.clone(),
+                spatial_filter_canonical: String::new(),
+            },
+            accuracy: Some(AccuracyBound::from_config(&cfg)),
+            first_seen_unix_ms: 0,
+            retired_at_ms: None,
+            expires_at_ms: None,
+            policy_fp: asap_types::PolicyFingerprint::UNSET,
+        }
+    }
+
+    /// Warm DDSketch covers only the SUFFIX of the requested range
+    /// (two windows near `end`); archive returns a full-range matrix
+    /// including the prefix. The stitched matrix must span the FULL request:
+    /// prefix timestamps come from archive, suffix from warm (warm wins on
+    /// any overlap).
+    #[tokio::test]
+    async fn range_stitches_archive_prefix_with_warm_suffix() {
+        let idx = Arc::new(SketchStore::new());
+        let now_ms = std::time::SystemTime::now()
+            .duration_since(std::time::SystemTime::UNIX_EPOCH)
+            .map(|d| d.as_millis() as u64)
+            .unwrap_or(0);
+
+        // Requested range: [now-600s, now].
+        let start_ms = now_ms.saturating_sub(600_000);
+        let end_ms = now_ms;
+
+        // Warm windows only in the suffix: [now-200s], [now-100s].
+        // `count_over_time` over a CountMin sid emits one PER-WINDOW total,
+        // so warm contributes a value at BOTH window-ends.
+        let warm_w1_end = now_ms.saturating_sub(200_000);
+        let warm_w2_end = now_ms.saturating_sub(100_000);
+        let warm_w1_total = 100.0_f64;
+        let sid = 9100u64;
+        idx.register(cms_meta(sid, "req_count"));
+        for (w_end, total) in [(warm_w1_end, 100i64), (warm_w2_end, 200i64)] {
+            idx.append_sample(
+                sid,
+                BTreeMap::new(),
+                (w_end.saturating_sub(30_000), w_end),
+                SketchSampleState {
+                    bytes: cms_bytes(total),
+                    encoding: SketchEncoding::ProtoFull,
+                },
+            );
+        }
+
+        // Archive provides the WHOLE range, including the prefix the warm
+        // tier can't cover. Use the bare empty-label series the DD reducer
+        // emits (so labels line up for the stitch merge).
+        let labels = KeyByLabelValues::new_with_labels(Vec::new());
+        let mut arch_el = RangeVectorElement::new(labels);
+        // Prefix samples (before warm coverage) + a suffix sample warm will win.
+        let prefix_ts = now_ms.saturating_sub(500_000) as i64;
+        let mid_ts = now_ms.saturating_sub(300_000) as i64;
+        arch_el.samples.push(Sample::new(prefix_ts as u64, 999.0));
+        arch_el.samples.push(Sample::new(mid_ts as u64, 998.0));
+        arch_el
+            .samples
+            .push(Sample::new(warm_w1_end, 1.0)); // overlap: warm should win
+        let archive = Arc::new(FakeArchive {
+            matrix: QueryResult::matrix(vec![arch_el]),
+        });
+
+        let streaming_config = Arc::new(crate::storage_engines::types::StreamingConfig::default());
+        let hot_reload = HotReloadStreamingConfig::from_arc(streaming_config);
+        let engine = ASAPQueryEngine::new_with_hot_reload(hot_reload, 15000)
+            .with_sketch_index(idx)
+            .with_archive_engine(archive);
+
+        let result = engine
+            .execute_range_promql_modern(
+                "count_over_time(req_count[5m])",
+                start_ms,
+                end_ms,
+                15_000,
+            )
+            .await
+            .expect("range query must answer (stitched), not error");
+
+        let m = match result {
+            QueryResult::Matrix(m) => m,
+            other => panic!("expected Matrix, got {other:?}"),
+        };
+        assert_eq!(m.values.len(), 1, "one merged series: {m:?}");
+        let samples = &m.values[0].samples;
+        let ts: std::collections::BTreeSet<i64> =
+            samples.iter().map(|s| s.timestamp as i64).collect();
+
+        // The PREFIX timestamps (only the archive has them) must be present —
+        // this is the whole point of the fix (warm-only would have dropped
+        // them).
+        assert!(
+            ts.contains(&prefix_ts),
+            "archive prefix sample (t={prefix_ts}) must survive the stitch: {ts:?}"
+        );
+        assert!(
+            ts.contains(&mid_ts),
+            "archive mid sample (t={mid_ts}) must survive the stitch: {ts:?}"
+        );
+        // The SUFFIX warm windows must be present too.
+        assert!(
+            ts.contains(&(warm_w1_end as i64)) && ts.contains(&(warm_w2_end as i64)),
+            "warm suffix windows must be present: {ts:?}"
+        );
+
+        // Warm wins on the overlapping timestamp: at warm_w1_end the value
+        // must be the warm per-window total (100), NOT the archive sentinel 1.0.
+        let overlap = samples
+            .iter()
+            .find(|s| s.timestamp == warm_w1_end)
+            .expect("overlap sample present");
+        assert!(
+            (overlap.value - warm_w1_total).abs() < 1e-6,
+            "warm must win on overlap (expected warm total {warm_w1_total}, got {})",
+            overlap.value
+        );
+    }
+
+    /// Control: when warm coverage already spans the request exactly
+    /// (`cov_lo == start_ms && cov_hi == end_ms`), no stitch is needed and the
+    /// warm-only matrix is returned unchanged — the archive is NOT consulted
+    /// even though it's wired. Per-window coverage is window-end-point-based,
+    /// Control: with NO archive engine wired, the range path returns the
+    /// warm-only matrix (no stitch, no error) even when warm coverage is
+    /// narrower than the request — the stitch is gated on a configured
+    /// archive. This pins that the fix doesn't disturb the archive-less
+    /// deployment (the warm tier answers what it can).
+    #[tokio::test]
+    async fn range_warm_only_when_no_archive_engine() {
+        let idx = Arc::new(SketchStore::new());
+        let now_ms = std::time::SystemTime::now()
+            .duration_since(std::time::SystemTime::UNIX_EPOCH)
+            .map(|d| d.as_millis() as u64)
+            .unwrap_or(0);
+        let start_ms = now_ms.saturating_sub(600_000);
+        let end_ms = now_ms;
+        // Warm covers only one suffix window — narrower than the request.
+        let w_end = now_ms.saturating_sub(100_000);
+        let sid = 9200u64;
+        idx.register(cms_meta(sid, "req_count"));
+        idx.append_sample(
+            sid,
+            BTreeMap::new(),
+            (w_end.saturating_sub(30_000), w_end),
+            SketchSampleState {
+                bytes: cms_bytes(42),
+                encoding: SketchEncoding::ProtoFull,
+            },
+        );
+
+        // No `.with_archive_engine(...)` — stitch must NOT fire.
+        let streaming_config = Arc::new(crate::storage_engines::types::StreamingConfig::default());
+        let hot_reload = HotReloadStreamingConfig::from_arc(streaming_config);
+        let engine = ASAPQueryEngine::new_with_hot_reload(hot_reload, 15000).with_sketch_index(idx);
+
+        let result = engine
+            .execute_range_promql_modern("count_over_time(req_count[5m])", start_ms, end_ms, 15_000)
+            .await
+            .expect("range query must answer warm-only");
+        let m = match result {
+            QueryResult::Matrix(m) => m,
+            other => panic!("expected Matrix, got {other:?}"),
+        };
+        // Warm-only: exactly the single warm window-end sample, no archive
+        // prefix injected.
+        let ts: Vec<u64> = m
+            .values
+            .iter()
+            .flat_map(|el| el.samples.iter().map(|s| s.timestamp))
+            .collect();
+        assert_eq!(
+            ts,
+            vec![w_end],
+            "warm-only result must carry just the warm window sample: {ts:?}"
+        );
+    }
+}
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 21f42fb1..93177b63 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
@@ -267,6 +267,60 @@ impl RollingState {
             _ => 0.0,
         }
     }
+
+    /// Borrow the inner HLL sketch when this rolling state is HLL-backed.
+    /// Used by the GLOBAL cardinality rollup (`count(hll_metric)` with no
+    /// `by`), which must MERGE the per-series HLL registers (register-wise
+    /// max) across all matched series and estimate ONCE — summing per-series
+    /// distinct estimates would double-count items present in multiple series.
+    pub fn as_hll(&self) -> Option<&HllSketch> {
+        match self {
+            RollingState::Hll(sk) => Some(sk),
+            _ => None,
+        }
+    }
+}
+
+/// Fold every in-range window's frames for ONE series into a single merged
+/// `HllSketch` (cumulative over `[t0, t1]`), returning `None` if no Full
+/// HLL frame ever landed (every sample was a leading delta). This is the
+/// per-series building block for the GLOBAL `count(hll_metric)` rollup: the
+/// reducer merges the returned sketches across series (register-wise max)
+/// before estimating, so the answer is the distinct UNION cardinality, not
+/// the sum of per-series cardinalities.
+pub fn cumulative_hll_state(
+    samples: &[(i64, &SketchSampleState)],
+    precision: u32,
+) -> Result<Option<HllSketch>, String> {
+    let kind = DeltaSketchKind::Hll { precision };
+    let mut rolling: Option<RollingState> = None;
+    for (_window_end, state) in samples {
+        match state.encoding {
+            SketchEncoding::ProtoFull | SketchEncoding::MsgpackFull => {
+                let new_state = decode_full(&kind, &state.bytes, state.encoding)?;
+                rolling = Some(match (rolling.take(), new_state) {
+                    (None, n) => n,
+                    (Some(RollingState::Hll(mut a)), RollingState::Hll(b)) => {
+                        a.merge(&b).map_err(|e| format!("cum merge HLL: {e}"))?;
+                        RollingState::Hll(a)
+                    }
+                    (Some(prev), _) => prev,
+                });
+            }
+            SketchEncoding::ProtoDelta | SketchEncoding::MsgpackDelta => {
+                if rolling.is_none() {
+                    rolling = Some(kind.bootstrap_empty());
+                }
+                if let Some(rs) = rolling.as_mut() {
+                    rs.apply_delta_bytes(&state.bytes, state.encoding)?;
+                }
+            }
+        }
+    }
+    Ok(rolling.and_then(|rs| match rs {
+        RollingState::Hll(sk) => Some(sk),
+        _ => None,
+    }))
 }
 
 /// Walk a sorted-by-window-end slice of samples in time order and
diff --git a/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs b/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs
index 5b73f8bf..8e752ad5 100644
--- a/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs
+++ b/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs
@@ -411,6 +411,117 @@ impl<'a> SketchReducer<'a> {
         )
     }
 
+    /// GLOBAL HLL distinct rollup (FIX: sealed-window global `count()`
+    /// empty). `count(hll_metric)` with NO `by (...)` asks for the distinct
+    /// count across ALL matched series. The per-series Cardinality path emits
+    /// one estimate per series; summing them double-counts any item present
+    /// in more than one series, and the engine otherwise returns a multi-row
+    /// vector rather than the single global number. This method MERGES the
+    /// per-series HLL registers (HLL merge = element-wise max of registers)
+    /// across every matched sid, then estimates ONCE — the correct distinct
+    /// UNION cardinality.
+    ///
+    /// Returns a single-series `ASAPTierResult` (empty label set) carrying the
+    /// merged estimate at the latest covered window-end, plus the merged
+    /// coverage range. `ASAPTierError::NoData` if no sid had an in-window HLL
+    /// Full frame; `UnsupportedCapability` if a matched sid isn't HLL-backed.
+    pub fn evaluate_cardinality_global(
+        &self,
+        sids: &[u64],
+        t0_ms: u64,
+        t1_ms: u64,
+    ) -> Result<ASAPTierResult, ASAPTierError> {
+        use super::delta_apply::cumulative_hll_state;
+        use crate::storage_engines::sketch_db::data::SketchConfig;
+        use asap_sketchlib::HllSketch;
+
+        let mut merged: Option<HllSketch> = None;
+        let mut metric_name_for_err = String::new();
+        let mut cov_lo: u64 = u64::MAX;
+        let mut cov_hi: u64 = 0;
+        let mut any_window = false;
+
+        for &sid in sids {
+            let meta = match self.index.instance(sid) {
+                Some(m) => m,
+                None => continue,
+            };
+            metric_name_for_err = meta.metric_name.clone();
+            // Only HLL sids answer cardinality via register merge.
+            match meta
+                .sketch_kind()
+                .expect("ASAP-tier reducer only handles sketch-backed sids")
+            {
+                SketchKindHandle::Hll => {}
+                _ => {
+                    return Err(ASAPTierError::UnsupportedCapability {
+                        function: "cardinality_global".to_string(),
+                        capability: Capability::CardinalityApprox,
+                    });
+                }
+            }
+            let precision = match meta.sketch_config() {
+                Some(SketchConfig::Hll { precision }) => *precision,
+                _ => 14,
+            };
+
+            let series_list = self.index.query_range(sid, t0_ms, t1_ms);
+            for ts in series_list {
+                let samples_vec: Vec<(i64, &SketchSampleState)> = ts
+                    .samples
+                    .iter()
+                    .flat_map(|(t, frames)| frames.iter().map(move |s| (*t, s)))
+                    .collect();
+                for (w_end, _) in &samples_vec {
+                    any_window = true;
+                    let w = if *w_end >= 0 { *w_end as u64 } else { 0 };
+                    cov_lo = cov_lo.min(w);
+                    cov_hi = cov_hi.max(w);
+                }
+                let series_state =
+                    cumulative_hll_state(&samples_vec, precision).map_err(|e| {
+                        ASAPTierError::DeserializeFailure {
+                            sid,
+                            encoding: SketchEncoding::ProtoFull,
+                            reason: e,
+                        }
+                    })?;
+                if let Some(sk) = series_state {
+                    merged = Some(match merged.take() {
+                        None => sk,
+                        Some(mut acc) => {
+                            acc.merge(&sk).map_err(|e| ASAPTierError::DeserializeFailure {
+                                sid,
+                                encoding: SketchEncoding::ProtoFull,
+                                reason: format!("global HLL merge: {e}"),
+                            })?;
+                            acc
+                        }
+                    });
+                }
+            }
+        }
+
+        let Some(merged) = merged else {
+            return Err(ASAPTierError::NoData {
+                metric_name: metric_name_for_err,
+            });
+        };
+        let _ = any_window;
+        let estimate = merged.estimate();
+        let window_end = if cov_hi > 0 { cov_hi as i64 } else { t1_ms as i64 };
+        let coverage = if cov_lo <= cov_hi {
+            Some((cov_lo, cov_hi))
+        } else {
+            None
+        };
+        Ok(ASAPTierResult {
+            // Empty label set — a global distinct count has no group labels.
+            series: vec![(BTreeMap::new(), vec![(window_end, estimate)])],
+            coverage,
+        })
+    }
+
     /// Shared evaluation core for the string ([`Self::evaluate`]) and
     /// typed ([`Self::evaluate_for_capability`]) entry points. `family`
     /// + `is_cumulative` are already resolved by the caller;