diff --git a/asap-precompute-rs/src/answers.rs b/asap-precompute-rs/src/answers.rs
index 65a3fab..7b0baf6 100644
--- a/asap-precompute-rs/src/answers.rs
+++ b/asap-precompute-rs/src/answers.rs
@@ -184,6 +184,47 @@ pub fn exact_answers(
     out
 }
 
+#[cfg(test)]
+mod classifier_tests {
+    use super::*;
+
+    fn attr(key: &str) -> SketchEnvelope {
+        SketchEnvelope {
+            metric_name: "m".into(),
+            labels: vec![KeyValue::new("attribute.key", key)],
+            ..Default::default()
+        }
+    }
+
+    fn state(out: &[SketchEnvelope]) -> String {
+        out.iter()
+            .find(|e| e.metric_name == "asap.cardinality.state" && e.value == 1.0)
+            .and_then(|e| e.labels.iter().find(|kv| kv.key == "state"))
+            .map(|kv| kv.value.clone())
+            .unwrap()
+    }
+
+    #[test]
+    fn identifier_growing_and_stable() {
+        let mut c = CardinalityClassifier::default();
+        let (out, event) = c.observe(&attr("user.id"), "sketch", 980.0, 1000);
+        assert_eq!(state(&out), "identifier-like");
+        assert!(event.unwrap().body.contains("identifier-like"));
+        for _ in 0..4 {
+            let (out, _) = c.observe(&attr("pod"), "sketch", 50.0, 1000);
+            assert_eq!(state(&out), "stable");
+        }
+        let (out, event) = c.observe(&attr("pod"), "sketch", 70.0, 1000);
+        assert_eq!(state(&out), "growing");
+        assert_eq!(event.unwrap().attributes[1].value, "growing");
+        // No event while the state does not change.
+        let (_, event) = c.observe(&attr("pod"), "sketch", 90.0, 1000);
+        assert!(event.is_none());
+        // Too few points: no answer.
+        assert!(c.observe(&attr("rare"), "sketch", 5.0, 5).0.is_empty());
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -282,3 +323,176 @@ mod tests {
             .any(|kv| kv.key == "method" && kv.value == "sketch")));
     }
 }
+
+/// A detection result worth telling a person about, emitted as a log record.
+#[derive(Clone, Debug, PartialEq)]
+pub struct Event {
+    /// Window the event refers to.
+    pub window_end_ms: u64,
+    /// `sketch` or `exact`.
+    pub method: String,
+    /// Event kind, e.g. `cardinality`.
+    pub kind: String,
+    /// Series labels plus event fields.
+    pub attributes: Vec<KeyValue>,
+    /// One readable sentence.
+    pub body: String,
+}
+
+/// Cardinality class of an attribute key in one window.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum CardinalityState {
+    /// Distinct values close to the baseline.
+    Stable,
+    /// Distinct values well above the baseline mean.
+    Growing,
+    /// Almost every point carries a new value, like an identifier.
+    IdentifierLike,
+}
+
+impl CardinalityState {
+    /// Name used in labels and events.
+    pub fn name(self) -> &'static str {
+        match self {
+            Self::Stable => "stable",
+            Self::Growing => "growing",
+            Self::IdentifierLike => "identifier-like",
+        }
+    }
+}
+
+/// Thresholds for [`CardinalityClassifier`].
+#[derive(Clone, Debug, serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct CardinalityRules {
+    /// Windows kept as the baseline.
+    #[serde(default = "default_baseline")]
+    pub baseline_windows: usize,
+    /// distinct / points at or above which a key is identifier-like.
+    #[serde(default = "default_identifier_ratio")]
+    pub identifier_ratio: f64,
+    /// Fraction above the baseline mean at which a key is growing.
+    #[serde(default = "default_growth")]
+    pub growth: f64,
+    /// Points needed before classifying.
+    #[serde(default = "default_min_points")]
+    pub min_points: u64,
+}
+
+fn default_baseline() -> usize {
+    10
+}
+fn default_identifier_ratio() -> f64 {
+    0.9
+}
+fn default_growth() -> f64 {
+    0.2
+}
+fn default_min_points() -> u64 {
+    100
+}
+
+impl Default for CardinalityRules {
+    fn default() -> Self {
+        Self {
+            baseline_windows: default_baseline(),
+            identifier_ratio: default_identifier_ratio(),
+            growth: default_growth(),
+            min_points: default_min_points(),
+        }
+    }
+}
+
+/// Classifies each attribute key against its own recent windows. The same
+/// rules run on sketch and exact answers so their labels can be compared.
+#[derive(Default)]
+pub struct CardinalityClassifier {
+    rules: CardinalityRules,
+    history: HashMap<String, (std::collections::VecDeque<f64>, Option<CardinalityState>)>,
+}
+
+impl CardinalityClassifier {
+    /// Classifier with the given thresholds.
+    pub fn new(rules: CardinalityRules) -> Self {
+        Self {
+            rules,
+            history: HashMap::new(),
+        }
+    }
+
+    /// Classifies one window of `series` (its distinct count and points), and
+    /// returns its state gauge plus an event when the state changed.
+    pub fn observe(
+        &mut self,
+        template: &SketchEnvelope,
+        method: &str,
+        distinct: f64,
+        points: u64,
+    ) -> (Vec<SketchEnvelope>, Option<Event>) {
+        if points < self.rules.min_points {
+            return (Vec::new(), None);
+        }
+        let key = format!("{}|{:?}", template.metric_name, template.labels);
+        let (window, previous) = self.history.entry(key).or_default();
+        let baseline =
+            (window.len() >= 3).then(|| window.iter().sum::<f64>() / window.len() as f64);
+        let ratio = distinct / points as f64;
+        let state = if ratio >= self.rules.identifier_ratio {
+            CardinalityState::IdentifierLike
+        } else if baseline.is_some_and(|b| distinct > b * (1.0 + self.rules.growth)) {
+            CardinalityState::Growing
+        } else {
+            CardinalityState::Stable
+        };
+        window.push_back(distinct);
+        if window.len() > self.rules.baseline_windows {
+            window.pop_front();
+        }
+        let mut out = vec![gauge(template, "asap.distinct.ratio", method, &[], ratio)];
+        out.extend(
+            [
+                CardinalityState::Stable,
+                CardinalityState::Growing,
+                CardinalityState::IdentifierLike,
+            ]
+            .into_iter()
+            .map(|s| {
+                let v = if s == state { 1.0 } else { 0.0 };
+                gauge(
+                    template,
+                    "asap.cardinality.state",
+                    method,
+                    &[("state", s.name().into())],
+                    v,
+                )
+            }),
+        );
+        let changed = *previous != Some(state);
+        *previous = Some(state);
+        let attribute = template
+            .labels
+            .iter()
+            .find(|kv| kv.key == crate::matchers::ATTRIBUTE_KEY_LABEL)
+            .map(|kv| kv.value.clone())
+            .unwrap_or_default();
+        let event = changed.then(|| Event {
+            window_end_ms: template.window_end_ms,
+            method: method.into(),
+            kind: "cardinality".into(),
+            attributes: vec![
+                KeyValue::new("attribute.key", attribute.clone()),
+                KeyValue::new("state", state.name()),
+                KeyValue::new("distinct", format!("{distinct:.0}")),
+                KeyValue::new("points", points.to_string()),
+            ],
+            body: format!(
+                "attribute {attribute} of {} is {}: {distinct:.0} distinct values in {points} points ({:.0}%){}",
+                template.metric_name,
+                state.name(),
+                ratio * 100.0,
+                baseline.map(|b| format!(", baseline {b:.0}")).unwrap_or_default()
+            ),
+        });
+        (out, event)
+    }
+}
diff --git a/asap-precompute-rs/src/matchers.rs b/asap-precompute-rs/src/matchers.rs
index 4f2c99c..bc04473 100644
--- a/asap-precompute-rs/src/matchers.rs
+++ b/asap-precompute-rs/src/matchers.rs
@@ -205,6 +205,29 @@ pub fn series_attrs(labels: &[KeyValue], aggregate_by: &[String]) -> Vec<KeyValu
     out
 }
 
+/// Label carrying the attribute key in per-attribute series.
+pub const ATTRIBUTE_KEY_LABEL: &str = "attribute.key";
+
+/// Splits one data point into one observation per attribute key in `keys`
+/// (`"*"` = every attribute it has): labels `attribute.key=<key>`, value =
+/// the attribute's value as bytes. A cardinality or frequency sketch over
+/// these series answers "how many distinct values" and "which values
+/// dominate" per attribute key.
+pub fn per_attribute(obs: &Observation, keys: &[String]) -> Vec<Observation> {
+    let all = keys.iter().any(|k| k == "*");
+    obs.labels
+        .iter()
+        .filter(|kv| all || keys.contains(&kv.key))
+        .map(|kv| Observation {
+            timestamp_ms: obs.timestamp_ms,
+            metric: obs.metric.clone(),
+            resource_labels: obs.resource_labels.clone(),
+            labels: vec![KeyValue::new(ATTRIBUTE_KEY_LABEL, kv.key.clone())],
+            value: crate::observation::ObservationValue::bytes(kv.value.as_bytes().to_vec()),
+        })
+        .collect()
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/asap-precompute-rs/src/otap/answer_nodes.rs b/asap-precompute-rs/src/otap/answer_nodes.rs
index c7a08c3..c7e6286 100644
--- a/asap-precompute-rs/src/otap/answer_nodes.rs
+++ b/asap-precompute-rs/src/otap/answer_nodes.rs
@@ -26,12 +26,14 @@ use linkme::distributed_slice;
 use otel_arrow_dfe_config::node::NodeUserConfig;
 use otel_arrow_dfe_engine::config::ProcessorConfig;
 use otel_arrow_dfe_engine::context::PipelineContext;
+use otel_arrow_dfe_engine::control::AckMsg;
 use otel_arrow_dfe_engine::control::NodeControlMsg;
 use otel_arrow_dfe_engine::error::Error;
 use otel_arrow_dfe_engine::local::processor as local;
 use otel_arrow_dfe_engine::message::Message;
 use otel_arrow_dfe_engine::node::NodeId;
 use otel_arrow_dfe_engine::processor::ProcessorWrapper;
+use otel_arrow_dfe_engine::ConsumerEffectHandlerExtension;
 use otel_arrow_dfe_engine::{MessageSourceLocalEffectHandlerExtension, ProcessorFactory};
 use otel_arrow_dfe_otap::pdata::OtapPdata;
 use otel_arrow_dfe_otap::OTAP_PROCESSOR_FACTORIES;
@@ -40,6 +42,9 @@ use serde::Deserialize;
 use otel_arrow_dfe_pdata::proto::opentelemetry::common::v1::{
     AnyValue, InstrumentationScope, KeyValue as OtlpKeyValue,
 };
+use otel_arrow_dfe_pdata::proto::opentelemetry::logs::v1::{
+    LogRecord, LogsData, ResourceLogs, ScopeLogs, SeverityNumber,
+};
 use otel_arrow_dfe_pdata::proto::opentelemetry::metrics::v1::{
     Gauge, Metric, MetricsData, NumberDataPoint, ResourceMetrics, ScopeMetrics,
 };
@@ -48,7 +53,9 @@ use otel_arrow_dfe_pdata::proto::OtlpProtoMessage;
 use otel_arrow_dfe_pdata::OtapPayload;
 
 use super::codec::decode_pdata_to_observations;
-use crate::answers::{exact_answers, sketch_answers, ExactWindow};
+use crate::answers::{
+    exact_answers, sketch_answers, CardinalityClassifier, CardinalityRules, Event, ExactWindow,
+};
 use crate::envelope::SketchEnvelope;
 use crate::observation::{KeyValue, Observation, ObservationValueKind};
 
@@ -80,6 +87,9 @@ pub struct DetectorConfig {
     /// Heavy hitters answered from CMS + heap windows.
     #[serde(default = "default_k")]
     pub k: usize,
+    /// Thresholds for classifying per-attribute cardinality.
+    #[serde(default)]
+    pub cardinality: CardinalityRules,
 }
 
 /// `exact_answers` configuration.
@@ -104,6 +114,9 @@ pub struct ExactConfig {
     /// Heavy hitters per attribute key.
     #[serde(default = "default_k")]
     pub k: usize,
+    /// Thresholds for classifying per-attribute cardinality.
+    #[serde(default)]
+    pub cardinality: CardinalityRules,
 }
 
 fn now_ms() -> u64 {
@@ -154,16 +167,66 @@ fn gauges_pdata(envs: &[SketchEnvelope]) -> Result<OtapPdata, String> {
     Ok(OtapPdata::new_todo_context(payload))
 }
 
+/// Detection events as OTLP log records (for Loki or any log backend).
+fn events_pdata(events: &[Event]) -> Result<OtapPdata, String> {
+    let records = events
+        .iter()
+        .map(|e| {
+            let mut attributes = vec![
+                OtlpKeyValue::new("event.kind", AnyValue::new_string(e.kind.clone())),
+                OtlpKeyValue::new("method", AnyValue::new_string(e.method.clone())),
+            ];
+            attributes.extend(e.attributes.iter().map(|kv| {
+                OtlpKeyValue::new(kv.key.clone(), AnyValue::new_string(kv.value.clone()))
+            }));
+            LogRecord::build()
+                .time_unix_nano(e.window_end_ms * 1_000_000)
+                .observed_time_unix_nano(e.window_end_ms * 1_000_000)
+                .severity_number(SeverityNumber::Info)
+                .severity_text("INFO")
+                .body(AnyValue::new_string(e.body.clone()))
+                .attributes(attributes)
+                .finish()
+        })
+        .collect::<Vec<_>>();
+    let data = LogsData::new(vec![ResourceLogs::new(
+        Resource::build()
+            .attributes(vec![OtlpKeyValue::new(
+                "service.name",
+                AnyValue::new_string("asap-detector"),
+            )])
+            .finish(),
+        vec![ScopeLogs::new(
+            InstrumentationScope::build().name("asap.events").finish(),
+            records,
+        )],
+    )]);
+    let payload: OtapPayload = OtlpProtoMessage::Logs(data)
+        .try_into()
+        .map_err(|e| format!("{e:?}"))?;
+    Ok(OtapPdata::new_todo_context(payload))
+}
+
 async fn emit(
     envs: &[SketchEnvelope],
+    events: &[Event],
     effects: &mut local::EffectHandler<OtapPdata>,
 ) -> Result<(), Error> {
-    if envs.is_empty() {
-        return Ok(());
+    if !envs.is_empty() {
+        match gauges_pdata(envs) {
+            Ok(pdata) => effects.send_message_with_source_node(pdata).await?,
+            Err(e) => effects.info(&format!("answers: encode failed: {e}")).await,
+        }
     }
-    match gauges_pdata(envs) {
-        Ok(pdata) => effects.send_message_with_source_node(pdata).await?,
-        Err(e) => effects.info(&format!("answers: encode failed: {e}")).await,
+    if !events.is_empty() {
+        match events_pdata(events) {
+            Ok(pdata) => effects.send_message_with_source_node(pdata).await?,
+            Err(e) => {
+                effects
+                    .info(&format!("answers: event encode failed: {e}"))
+                    .await
+            }
+        }
     }
     Ok(())
 }
@@ -185,6 +248,20 @@ async fn observations(
 
 struct SketchDetector {
     config: DetectorConfig,
+    cardinality: CardinalityClassifier,
+}
+
+/// The `bound="estimate"` value of `metric` among `answers`.
+fn estimate_of(answers: &[SketchEnvelope], metric: &str) -> Option<f64> {
+    answers
+        .iter()
+        .find(|e| {
+            e.metric_name == metric
+                && e.labels
+                    .iter()
+                    .any(|kv| kv.key == "bound" && kv.value == "estimate")
+        })
+        .map(|e| e.value)
 }
 
 #[async_trait(?Send)]
@@ -197,17 +274,27 @@ impl local::Processor<OtapPdata> for SketchDetector {
         let Message::PData(pdata) = msg else {
             return Ok(());
         };
-        let mut out = Vec::new();
+        // Consumed into state here: ack so upstream (e.g. fanout) can move on.
+        effects.notify_ack(AckMsg::new(pdata.clone())).await?;
+        let (mut out, mut events) = (Vec::new(), Vec::new());
         for obs in observations(pdata, effects).await {
             let Some(env) = obs.value.envelope.filter(|e| !e.payload.is_empty()) else {
                 continue;
             };
             match sketch_answers(&env, &self.config.quantiles, self.config.k) {
-                Ok(answers) => out.extend(answers),
+                Ok(answers) => {
+                    if let Some(d) = estimate_of(&answers, "asap.distinct") {
+                        let (gauges, event) =
+                            self.cardinality.observe(&env, "sketch", d, env.count);
+                        out.extend(gauges);
+                        events.extend(event);
+                    }
+                    out.extend(answers);
+                }
                 Err(e) => effects.info(&format!("sketch_detector: {e}")).await,
             }
         }
-        emit(&out, effects).await
+        emit(&out, &events, effects).await
     }
 }
 
@@ -219,6 +306,7 @@ struct ExactState {
 
 struct ExactAnswers {
     config: ExactConfig,
+    cardinality: CardinalityClassifier,
     /// Open windows by window end.
     windows: BTreeMap<u64, ExactState>,
     emitted_until: u64,
@@ -270,7 +358,13 @@ impl ExactAnswers {
             w.values.push(obs.value.float);
             w.points += 1;
         }
-        for attr in &self.config.per_attribute {
+        let all = self.config.per_attribute.iter().any(|k| k == "*");
+        let keys: Vec<String> = if all {
+            obs.labels.iter().map(|kv| kv.key.clone()).collect()
+        } else {
+            self.config.per_attribute.clone()
+        };
+        for attr in &keys {
             let Some(value) = label(attr) else { continue };
             let key = format!("a|{}|{attr}", obs.metric);
             let (_, w) = state.series.entry(key).or_insert_with(|| {
@@ -286,15 +380,23 @@ impl ExactAnswers {
         }
     }
 
-    fn due(&mut self, now: u64) -> Vec<SketchEnvelope> {
+    fn due(&mut self, now: u64) -> (Vec<SketchEnvelope>, Vec<Event>) {
         let cutoff = now.saturating_sub(self.config.grace.as_millis() as u64);
         let open = self.windows.split_off(&(cutoff + 1));
         let ready = std::mem::replace(&mut self.windows, open);
-        let mut out = Vec::new();
+        let (mut out, mut events) = (Vec::new(), Vec::new());
         for (end, state) in ready {
             self.emitted_until = self.emitted_until.max(end);
             for (_, (mut template, mut w)) in state.series {
                 template.count = w.points;
+                if !w.distinct.is_empty() {
+                    let distinct = w.distinct.len() as f64;
+                    let (gauges, event) = self
+                        .cardinality
+                        .observe(&template, "exact", distinct, w.points);
+                    out.extend(gauges);
+                    events.extend(event);
+                }
                 out.extend(exact_answers(
                     &template,
                     &mut w,
@@ -303,7 +405,7 @@ impl ExactAnswers {
                 ));
             }
         }
-        out
+        (out, events)
     }
 }
 
@@ -322,18 +424,20 @@ impl local::Processor<OtapPdata> for ExactAnswers {
         }
         match msg {
             Message::PData(pdata) => {
+                // Consumed into state here: ack so upstream (e.g. fanout) can move on.
+                effects.notify_ack(AckMsg::new(pdata.clone())).await?;
                 for obs in observations(pdata, effects).await {
                     self.observe(&obs);
                 }
                 Ok(())
             }
             Message::Control(NodeControlMsg::TimerTick { .. }) => {
-                let out = self.due(now_ms());
-                emit(&out, effects).await
+                let (out, events) = self.due(now_ms());
+                emit(&out, &events, effects).await
             }
             Message::Control(NodeControlMsg::Shutdown { .. }) => {
-                let out = self.due(u64::MAX);
-                emit(&out, effects).await
+                let (out, events) = self.due(u64::MAX);
+                emit(&out, &events, effects).await
             }
             _ => Ok(()),
         }
@@ -358,7 +462,10 @@ static SKETCH_DETECTOR: ProcessorFactory<OtapPdata> = ProcessorFactory {
         let config: DetectorConfig =
             serde_json::from_value(node_config.config.clone()).map_err(invalid)?;
         Ok(ProcessorWrapper::local(
-            SketchDetector { config },
+            SketchDetector {
+                cardinality: CardinalityClassifier::new(config.cardinality.clone()),
+                config,
+            },
             node,
             node_config,
             processor_config,
@@ -381,6 +488,7 @@ static EXACT_ANSWERS: ProcessorFactory<OtapPdata> = ProcessorFactory {
             serde_json::from_value(node_config.config.clone()).map_err(invalid)?;
         Ok(ProcessorWrapper::local(
             ExactAnswers {
+                cardinality: CardinalityClassifier::new(config.cardinality.clone()),
                 config,
                 windows: BTreeMap::new(),
                 emitted_until: 0,
@@ -415,11 +523,13 @@ mod tests {
 
     #[test]
     fn exact_windows_by_event_time_after_grace() {
+        let config: ExactConfig = serde_json::from_value(serde_json::json!({
+            "group_by": ["http.route"], "per_attribute": ["user.id"], "k": 2
+        }))
+        .unwrap();
         let mut exact = ExactAnswers {
-            config: serde_json::from_value(serde_json::json!({
-                "group_by": ["http.route"], "per_attribute": ["user.id"], "k": 2
-            }))
-            .unwrap(),
+            cardinality: CardinalityClassifier::new(config.cardinality.clone()),
+            config,
             windows: BTreeMap::new(),
             emitted_until: 0,
             timer_started: false,
@@ -428,8 +538,8 @@ mod tests {
             exact.observe(&point(1_000 + i, "/a", &format!("u{}", i % 4), i as f64));
         }
         exact.observe(&point(61_000, "/a", "u0", 1.0));
-        assert!(exact.due(64_000).is_empty());
-        let out = exact.due(65_000);
+        assert!(exact.due(64_000).0.is_empty());
+        let (out, _) = exact.due(65_000);
         let get = |metric: &str, extra: (&str, &str)| {
             out.iter()
                 .find(|e| {
@@ -451,6 +561,7 @@ mod tests {
         exact.observe(&point(2_000, "/a", "u9", 1.0));
         assert!(exact
             .due(125_000)
+            .0
             .iter()
             .all(|e| e.window_end_ms == 120_000));
     }
diff --git a/asap-precompute-rs/src/otap/config.rs b/asap-precompute-rs/src/otap/config.rs
index d268d2e..3ad6245 100644
--- a/asap-precompute-rs/src/otap/config.rs
+++ b/asap-precompute-rs/src/otap/config.rs
@@ -116,6 +116,8 @@ pub struct PluginConfig {
     pub quantiles: Vec<f64>,
     /// Labels that identify a series; empty keys series by every label.
     pub aggregate_by: Vec<String>,
+    /// Attribute keys sketched as their own series; `"*"` = every attribute.
+    pub per_attribute: Vec<String>,
 }
 
 impl Default for PluginConfig {
@@ -134,6 +136,7 @@ impl Default for PluginConfig {
             transmit_sketch: true,
             quantiles: Vec::new(),
             aggregate_by: Vec::new(),
+            per_attribute: Vec::new(),
         }
     }
 }
diff --git a/asap-precompute-rs/src/otap/processor.rs b/asap-precompute-rs/src/otap/processor.rs
index 52a107f..0adaf84 100644
--- a/asap-precompute-rs/src/otap/processor.rs
+++ b/asap-precompute-rs/src/otap/processor.rs
@@ -74,6 +74,8 @@ use std::time::Duration;
 
 use async_trait::async_trait;
 use linkme::distributed_slice;
+use otel_arrow_dfe_engine::control::AckMsg;
+use otel_arrow_dfe_engine::ConsumerEffectHandlerExtension;
 use serde::{Deserialize, Serialize};
 
 use otel_arrow_dfe_config::error::Error as OtapConfigError;
@@ -93,6 +95,7 @@ use otel_arrow_dfe_otap::OTAP_PROCESSOR_FACTORIES;
 
 use crate::config::PrecomputeConfigSet;
 use crate::envelope::SketchEnvelope;
+use crate::matchers::per_attribute;
 use crate::observation::ObservationValueKind;
 use crate::otap::config::resolve as resolve_plugin_config;
 use crate::otap::PluginConfig;
@@ -155,6 +158,10 @@ pub struct AsapSketchesUserConfig {
     /// Labels that identify a series; empty keys series by every label.
     #[serde(default)]
     pub aggregate_by: Vec<String>,
+    /// Sketch each listed attribute's values as its own series
+    /// (`attribute.key=<key>`); `"*"` = every attribute.
+    #[serde(default)]
+    pub per_attribute: Vec<String>,
 }
 
 impl AsapSketchesUserConfig {
@@ -193,6 +200,7 @@ impl AsapSketchesUserConfig {
             transmit_sketch: self.transmit_sketch,
             quantiles: self.quantiles,
             aggregate_by: self.aggregate_by,
+            per_attribute: self.per_attribute,
         };
         // Bounce the config through the resolver so config mistakes
         // (unsupported sketch_type, zero window) surface here rather
@@ -411,6 +419,10 @@ impl local::Processor<OtapPdata> for AsapSketchesProcessor {
 
         match msg {
             Message::PData(pdata) => {
+                // Consumed into state here: ack so upstream (e.g. fanout) can move on.
+                effect_handler
+                    .notify_ack(AckMsg::new(pdata.clone()))
+                    .await?;
                 let outcome = match decode_pdata_to_observations(pdata) {
                     Ok(outcome) => outcome,
                     Err(error) => {
@@ -442,7 +454,16 @@ impl local::Processor<OtapPdata> for AsapSketchesProcessor {
                         self.emit_envelopes(&envs, effect_handler).await?;
                     }
                 }
-                for obs in &outcome.observations {
+                let observations = if self.plugin_config.per_attribute.is_empty() {
+                    outcome.observations
+                } else {
+                    outcome
+                        .observations
+                        .iter()
+                        .flat_map(|o| per_attribute(o, &self.plugin_config.per_attribute))
+                        .collect()
+                };
+                for obs in &observations {
                     // Note: `observe` here handles both "genuine OTLP
                     // metric" and "sketch shipped as binary inside an
                     // OTAP metric" (an upstream asap_sketches node's
diff --git a/asap-precompute-rs/src/otap/window_merge.rs b/asap-precompute-rs/src/otap/window_merge.rs
index 6a5534b..e61db4a 100644
--- a/asap-precompute-rs/src/otap/window_merge.rs
+++ b/asap-precompute-rs/src/otap/window_merge.rs
@@ -20,12 +20,14 @@ use linkme::distributed_slice;
 use otel_arrow_dfe_config::node::NodeUserConfig;
 use otel_arrow_dfe_engine::config::ProcessorConfig;
 use otel_arrow_dfe_engine::context::PipelineContext;
+use otel_arrow_dfe_engine::control::AckMsg;
 use otel_arrow_dfe_engine::control::NodeControlMsg;
 use otel_arrow_dfe_engine::error::Error;
 use otel_arrow_dfe_engine::local::processor as local;
 use otel_arrow_dfe_engine::message::Message;
 use otel_arrow_dfe_engine::node::NodeId;
 use otel_arrow_dfe_engine::processor::ProcessorWrapper;
+use otel_arrow_dfe_engine::ConsumerEffectHandlerExtension;
 use otel_arrow_dfe_engine::{MessageSourceLocalEffectHandlerExtension, ProcessorFactory};
 use otel_arrow_dfe_otap::pdata::OtapPdata;
 use otel_arrow_dfe_otap::OTAP_PROCESSOR_FACTORIES;
@@ -181,6 +183,8 @@ impl local::Processor<OtapPdata> for WindowMergeProcessor {
         }
         match msg {
             Message::PData(pdata) => {
+                // Consumed into state here: ack so upstream (e.g. fanout) can move on.
+                effects.notify_ack(AckMsg::new(pdata.clone())).await?;
                 let observations = match decode_pdata_to_observations(pdata) {
                     Ok(outcome) => outcome.observations,
                     Err(e) => {
diff --git a/deploy/telemetry-functions/cardinality/branch.yaml b/deploy/telemetry-functions/cardinality/branch.yaml
new file mode 100644
index 0000000..52b74c1
--- /dev/null
+++ b/deploy/telemetry-functions/cardinality/branch.yaml
@@ -0,0 +1,55 @@
+# Stage 1 at each source: one HLL per attribute key per 1-minute window goes
+# to merge; the raw points are copied to the exact baseline.
+version: otel_dataflow/v1
+engine:
+  http_admin:
+    bind_address: 0.0.0.0:8085
+policies:
+  resources:
+    core_allocation:
+      type: core_count
+      count: 1
+groups:
+  default:
+    pipelines:
+      main:
+        nodes:
+          receiver:
+            type: receiver:otlp
+            config:
+              protocols:
+                http: {listening_addr: "0.0.0.0:4318", wait_for_result: false}
+          fanout:
+            type: processor:fanout
+            outputs: [sketch, raw]
+            config:
+              destinations:
+                - {port: sketch}
+                - {port: raw}
+          sketch:
+            type: urn:asap:processor:asap_sketches
+            config:
+              sketch_type: hll
+              window_size: 60s
+              output_metric_name: http.server.request.duration
+              per_attribute: ["*"]
+              omit_resource_attrs: true
+          to_merge:
+            type: exporter:otlp_http
+            config:
+              endpoint: http://merge:4318
+              client_pool_size: 1
+              max_in_flight: 1
+              http: {timeout: 30s, compression: none}
+          to_exact:
+            type: exporter:otlp_http
+            config:
+              endpoint: http://exact:4318
+              client_pool_size: 1
+              max_in_flight: 1
+              http: {timeout: 30s, compression: none}
+        connections:
+          - {from: receiver, to: fanout}
+          - {from: 'fanout["sketch"]', to: sketch}
+          - {from: sketch, to: to_merge}
+          - {from: 'fanout["raw"]', to: to_exact}
diff --git a/deploy/telemetry-functions/cardinality/detector.yaml b/deploy/telemetry-functions/cardinality/detector.yaml
new file mode 100644
index 0000000..e57e3cb
--- /dev/null
+++ b/deploy/telemetry-functions/cardinality/detector.yaml
@@ -0,0 +1,37 @@
+# Distinct values per attribute key (with bounds) and its class (stable,
+# growing, identifier-like) to Prometheus; class changes as events to Loki.
+version: otel_dataflow/v1
+engine:
+  http_admin:
+    bind_address: 0.0.0.0:8085
+policies:
+  resources:
+    core_allocation:
+      type: core_count
+      count: 1
+groups:
+  default:
+    pipelines:
+      main:
+        nodes:
+          receiver:
+            type: receiver:otlp
+            config:
+              protocols:
+                http: {listening_addr: "0.0.0.0:4318", wait_for_result: false}
+          detector:
+            type: urn:asap:processor:sketch_detector
+            config:
+              cardinality: {baseline_windows: 10, identifier_ratio: 0.9, growth: 0.2}
+          to_prometheus:
+            type: exporter:otlp_http
+            config:
+              endpoint: http://prometheus:9090
+              metrics_endpoint: http://prometheus:9090/api/v1/otlp/v1/metrics
+              logs_endpoint: http://loki:3100/otlp/v1/logs
+              client_pool_size: 1
+              max_in_flight: 1
+              http: {timeout: 30s, compression: none}
+        connections:
+          - {from: receiver, to: detector}
+          - {from: detector, to: to_prometheus}
diff --git a/deploy/telemetry-functions/cardinality/exact.yaml b/deploy/telemetry-functions/cardinality/exact.yaml
new file mode 100644
index 0000000..af1d888
--- /dev/null
+++ b/deploy/telemetry-functions/cardinality/exact.yaml
@@ -0,0 +1,38 @@
+# Exact baseline and ground truth: every raw point, exact answers per window.
+version: otel_dataflow/v1
+engine:
+  http_admin:
+    bind_address: 0.0.0.0:8085
+policies:
+  resources:
+    core_allocation:
+      type: core_count
+      count: 1
+groups:
+  default:
+    pipelines:
+      main:
+        nodes:
+          receiver:
+            type: receiver:otlp
+            config:
+              protocols:
+                http: {listening_addr: "0.0.0.0:4318", wait_for_result: false}
+          exact:
+            type: urn:asap:processor:exact_answers
+            config:
+              window: 60s
+              per_attribute: ["*"]
+              cardinality: {baseline_windows: 10, identifier_ratio: 0.9, growth: 0.2}
+          to_prometheus:
+            type: exporter:otlp_http
+            config:
+              endpoint: http://prometheus:9090
+              metrics_endpoint: http://prometheus:9090/api/v1/otlp/v1/metrics
+              logs_endpoint: http://loki:3100/otlp/v1/logs
+              client_pool_size: 1
+              max_in_flight: 1
+              http: {timeout: 30s, compression: none}
+        connections:
+          - {from: receiver, to: exact}
+          - {from: exact, to: to_prometheus}
diff --git a/deploy/telemetry-functions/cardinality/generator-a.yaml b/deploy/telemetry-functions/cardinality/generator-a.yaml
new file mode 100644
index 0000000..38f54c1
--- /dev/null
+++ b/deploy/telemetry-functions/cardinality/generator-a.yaml
@@ -0,0 +1,37 @@
+# Source a: four attributes with known cardinality behaviour.
+version: otel_dataflow/v1
+engine:
+  http_admin:
+    bind_address: 0.0.0.0:8085
+policies:
+  resources:
+    core_allocation:
+      type: core_count
+      count: 1
+groups:
+  default:
+    pipelines:
+      main:
+        nodes:
+          generator:
+            type: urn:asap:receiver:scenario_generator
+            config:
+              seed: 1
+              signals_per_second: 20000
+              metric_name: http.server.request.duration
+              resource_attributes: {service.name: checkout, source: "a"}
+              attributes:
+                - {key: http.route, cardinality: 20, zipf: 1.1}      # stable
+                - {key: http.method, cardinality: 5}                  # stable
+                - {key: k8s.pod.name, cardinality: 50, growth_per_minute: 0.05}  # growing
+                - {key: user.id, unique: true}                        # identifier-like
+              value: {lognormal: {median: 50.0, sigma: 0.6}}
+          export:
+            type: exporter:otlp_http
+            config:
+              endpoint: http://branch-a:4318
+              client_pool_size: 1
+              max_in_flight: 1
+              http: {timeout: 30s, compression: none}
+        connections:
+          - {from: generator, to: export}
diff --git a/deploy/telemetry-functions/cardinality/generator-b.yaml b/deploy/telemetry-functions/cardinality/generator-b.yaml
new file mode 100644
index 0000000..cd76e55
--- /dev/null
+++ b/deploy/telemetry-functions/cardinality/generator-b.yaml
@@ -0,0 +1,37 @@
+# Source b: four attributes with known cardinality behaviour.
+version: otel_dataflow/v1
+engine:
+  http_admin:
+    bind_address: 0.0.0.0:8085
+policies:
+  resources:
+    core_allocation:
+      type: core_count
+      count: 1
+groups:
+  default:
+    pipelines:
+      main:
+        nodes:
+          generator:
+            type: urn:asap:receiver:scenario_generator
+            config:
+              seed: 2
+              signals_per_second: 20000
+              metric_name: http.server.request.duration
+              resource_attributes: {service.name: checkout, source: "b"}
+              attributes:
+                - {key: http.route, cardinality: 20, zipf: 1.1}      # stable
+                - {key: http.method, cardinality: 5}                  # stable
+                - {key: k8s.pod.name, cardinality: 50, growth_per_minute: 0.05}  # growing
+                - {key: user.id, unique: true}                        # identifier-like
+              value: {lognormal: {median: 50.0, sigma: 0.6}}
+          export:
+            type: exporter:otlp_http
+            config:
+              endpoint: http://branch-b:4318
+              client_pool_size: 1
+              max_in_flight: 1
+              http: {timeout: 30s, compression: none}
+        connections:
+          - {from: generator, to: export}
diff --git a/deploy/telemetry-functions/pipelines/merge.yaml b/deploy/telemetry-functions/cardinality/merge.yaml
similarity index 100%
rename from deploy/telemetry-functions/pipelines/merge.yaml
rename to deploy/telemetry-functions/cardinality/merge.yaml
diff --git a/deploy/telemetry-functions/compose.yaml b/deploy/telemetry-functions/compose.yaml
index 672aa11..33aa284 100644
--- a/deploy/telemetry-functions/compose.yaml
+++ b/deploy/telemetry-functions/compose.yaml
@@ -1,6 +1,7 @@
 # Every pipeline stage is the official OTAP engine (asap-collector) in its
 # own container, pinned to one CPU, talking OTLP/HTTP over the compose
 # network. Prometheus, Grafana and Loki are the observers.
+# USE_CASE picks the pipelines directory (latency, cardinality, ...);
 # CPU_* variables move a stage to another core.
 x-collector: &collector
   image: asap-collector:local
@@ -10,43 +11,43 @@ services:
   generator-a:
     <<: *collector
     command: ["--config", "/pipelines/generator-a.yaml"]
-    volumes: ["./pipelines:/pipelines:ro"]
+    volumes: ["./${USE_CASE:-latency}:/pipelines:ro"]
     cpuset: "${CPU_GENERATOR_A:-0}"
     depends_on: [branch-a]
   generator-b:
     <<: *collector
     command: ["--config", "/pipelines/generator-b.yaml"]
-    volumes: ["./pipelines:/pipelines:ro"]
+    volumes: ["./${USE_CASE:-latency}:/pipelines:ro"]
     cpuset: "${CPU_GENERATOR_B:-1}"
     depends_on: [branch-b]
   branch-a:
     <<: *collector
     command: ["--config", "/pipelines/branch.yaml"]
-    volumes: ["./pipelines:/pipelines:ro"]
+    volumes: ["./${USE_CASE:-latency}:/pipelines:ro"]
     cpuset: "${CPU_BRANCH_A:-2}"
     depends_on: [merge, exact]
   branch-b:
     <<: *collector
     command: ["--config", "/pipelines/branch.yaml"]
-    volumes: ["./pipelines:/pipelines:ro"]
+    volumes: ["./${USE_CASE:-latency}:/pipelines:ro"]
     cpuset: "${CPU_BRANCH_B:-3}"
     depends_on: [merge, exact]
   merge:
     <<: *collector
     command: ["--config", "/pipelines/merge.yaml"]
-    volumes: ["./pipelines:/pipelines:ro"]
+    volumes: ["./${USE_CASE:-latency}:/pipelines:ro"]
     cpuset: "${CPU_MERGE:-4}"
     depends_on: [detector]
   detector:
     <<: *collector
     command: ["--config", "/pipelines/detector.yaml"]
-    volumes: ["./pipelines:/pipelines:ro"]
+    volumes: ["./${USE_CASE:-latency}:/pipelines:ro"]
     cpuset: "${CPU_DETECTOR:-5}"
     depends_on: [prometheus]
   exact:
     <<: *collector
     command: ["--config", "/pipelines/exact.yaml"]
-    volumes: ["./pipelines:/pipelines:ro"]
+    volumes: ["./${USE_CASE:-latency}:/pipelines:ro"]
     cpuset: "${CPU_EXACT:-6}"
     depends_on: [prometheus]
 
diff --git a/deploy/telemetry-functions/grafana/dashboards/cardinality.json b/deploy/telemetry-functions/grafana/dashboards/cardinality.json
new file mode 100644
index 0000000..8a20ef5
--- /dev/null
+++ b/deploy/telemetry-functions/grafana/dashboards/cardinality.json
@@ -0,0 +1,376 @@
+{
+  "uid": "asap-cardinality",
+  "title": "Attribute cardinality: sketch vs exact (real collector containers)",
+  "schemaVersion": 39,
+  "version": 1,
+  "refresh": "10s",
+  "time": {
+    "from": "now-30m",
+    "to": "now"
+  },
+  "templating": {
+    "list": [
+      {
+        "name": "attribute",
+        "type": "query",
+        "datasource": {
+          "type": "prometheus",
+          "uid": "asap-prometheus"
+        },
+        "query": {
+          "query": "label_values(asap_distinct, attribute_key)",
+          "refId": "attribute"
+        },
+        "refresh": 2,
+        "current": {
+          "text": "k8s.pod.name",
+          "value": "k8s.pod.name"
+        }
+      }
+    ]
+  },
+  "panels": [
+    {
+      "type": "timeseries",
+      "title": "Distinct values of $attribute per minute: sketch with 95% bound vs exact",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "asap-prometheus"
+      },
+      "gridPos": {
+        "h": 8,
+        "w": 12,
+        "x": 0,
+        "y": 0
+      },
+      "targets": [
+        {
+          "refId": "A",
+          "expr": "asap_distinct{attribute_key=\"$attribute\",method=\"sketch\",bound=\"estimate\"}",
+          "legendFormat": "sketch"
+        },
+        {
+          "refId": "B",
+          "expr": "asap_distinct{attribute_key=\"$attribute\",method=\"sketch\",bound=\"lower\"}",
+          "legendFormat": "sketch lower"
+        },
+        {
+          "refId": "C",
+          "expr": "asap_distinct{attribute_key=\"$attribute\",method=\"sketch\",bound=\"upper\"}",
+          "legendFormat": "sketch upper"
+        },
+        {
+          "refId": "D",
+          "expr": "asap_distinct{attribute_key=\"$attribute\",method=\"exact\"}",
+          "legendFormat": "exact"
+        }
+      ],
+      "fieldConfig": {
+        "defaults": {
+          "custom": {
+            "spanNulls": true
+          }
+        },
+        "overrides": [
+          {
+            "matcher": {
+              "id": "byName",
+              "options": "sketch upper"
+            },
+            "properties": [
+              {
+                "id": "custom.fillBelowTo",
+                "value": "sketch lower"
+              },
+              {
+                "id": "custom.lineWidth",
+                "value": 0
+              },
+              {
+                "id": "custom.fillOpacity",
+                "value": 20
+              },
+              {
+                "id": "color",
+                "value": {
+                  "mode": "fixed",
+                  "fixedColor": "blue"
+                }
+              }
+            ]
+          },
+          {
+            "matcher": {
+              "id": "byName",
+              "options": "sketch lower"
+            },
+            "properties": [
+              {
+                "id": "custom.lineWidth",
+                "value": 0
+              },
+              {
+                "id": "color",
+                "value": {
+                  "mode": "fixed",
+                  "fixedColor": "blue"
+                }
+              }
+            ]
+          },
+          {
+            "matcher": {
+              "id": "byName",
+              "options": "sketch"
+            },
+            "properties": [
+              {
+                "id": "color",
+                "value": {
+                  "mode": "fixed",
+                  "fixedColor": "blue"
+                }
+              }
+            ]
+          },
+          {
+            "matcher": {
+              "id": "byName",
+              "options": "exact"
+            },
+            "properties": [
+              {
+                "id": "color",
+                "value": {
+                  "mode": "fixed",
+                  "fixedColor": "orange"
+                }
+              },
+              {
+                "id": "custom.lineStyle",
+                "value": {
+                  "fill": "dash",
+                  "dash": [
+                    6,
+                    4
+                  ]
+                }
+              }
+            ]
+          }
+        ]
+      },
+      "id": 1
+    },
+    {
+      "type": "table",
+      "title": "Class of every attribute key (latest window)",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "asap-prometheus"
+      },
+      "gridPos": {
+        "h": 10,
+        "w": 12,
+        "x": 12,
+        "y": 0
+      },
+      "targets": [
+        {
+          "refId": "A",
+          "expr": "asap_cardinality_state == 1",
+          "format": "table",
+          "instant": true
+        }
+      ],
+      "transformations": [
+        {
+          "id": "organize",
+          "options": {
+            "excludeByName": {
+              "Time": true,
+              "Value": true,
+              "__name__": true,
+              "job": true,
+              "instance": true,
+              "metric": true
+            },
+            "renameByName": {
+              "attribute_key": "attribute",
+              "method": "method",
+              "state": "class"
+            }
+          }
+        },
+        {
+          "id": "sortBy",
+          "options": {
+            "sort": [
+              {
+                "field": "class",
+                "desc": false
+              },
+              {
+                "field": "attribute"
+              }
+            ]
+          }
+        }
+      ],
+      "id": 2
+    },
+    {
+      "type": "timeseries",
+      "title": "Distinct / points per attribute key (1.0 = every point new)",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "asap-prometheus"
+      },
+      "gridPos": {
+        "h": 8,
+        "w": 12,
+        "x": 0,
+        "y": 10
+      },
+      "targets": [
+        {
+          "refId": "A",
+          "expr": "asap_distinct_ratio",
+          "legendFormat": "{{attribute_key}} {{method}}"
+        }
+      ],
+      "fieldConfig": {
+        "defaults": {
+          "unit": "percentunit",
+          "max": 1,
+          "min": 0
+        },
+        "overrides": []
+      },
+      "id": 3
+    },
+    {
+      "type": "logs",
+      "title": "Cardinality events (Loki)",
+      "datasource": {
+        "type": "loki",
+        "uid": "asap-loki"
+      },
+      "gridPos": {
+        "h": 8,
+        "w": 12,
+        "x": 12,
+        "y": 10
+      },
+      "targets": [
+        {
+          "refId": "A",
+          "expr": "{service_name=\"asap-detector\"} | event_kind=\"cardinality\""
+        }
+      ],
+      "options": {
+        "showTime": true,
+        "wrapLogMessage": true,
+        "sortOrder": "Descending"
+      },
+      "id": 4
+    },
+    {
+      "id": 5,
+      "type": "timeseries",
+      "title": "CPU cores per pinned stage",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "asap-prometheus"
+      },
+      "gridPos": {
+        "h": 8,
+        "w": 8,
+        "x": 0,
+        "y": 20
+      },
+      "targets": [
+        {
+          "refId": "A",
+          "expr": "sum by (stage) (rate(cpu_time_total{job=\"collectors\",otel_scope_pipeline_id=\"main\"}[1m]))",
+          "legendFormat": "{{stage}}"
+        }
+      ],
+      "fieldConfig": {
+        "defaults": {
+          "custom": {
+            "spanNulls": true
+          },
+          "unit": "short"
+        },
+        "overrides": []
+      }
+    },
+    {
+      "id": 6,
+      "type": "timeseries",
+      "title": "Bytes received per stage, log scale (sketches at merge, raw points at exact)",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "asap-prometheus"
+      },
+      "gridPos": {
+        "h": 8,
+        "w": 8,
+        "x": 8,
+        "y": 20
+      },
+      "targets": [
+        {
+          "refId": "A",
+          "expr": "sum by (stage) (rate(payload_size_bytes_total{job=\"collectors\",otel_scope_node_type=\"receiver\"}[1m])) * 8",
+          "legendFormat": "{{stage}}"
+        }
+      ],
+      "fieldConfig": {
+        "defaults": {
+          "custom": {
+            "spanNulls": true,
+            "scaleDistribution": {
+              "type": "log",
+              "log": 10
+            }
+          },
+          "unit": "bps"
+        },
+        "overrides": []
+      }
+    },
+    {
+      "id": 7,
+      "type": "timeseries",
+      "title": "Resident memory per stage",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "asap-prometheus"
+      },
+      "gridPos": {
+        "h": 8,
+        "w": 8,
+        "x": 16,
+        "y": 20
+      },
+      "targets": [
+        {
+          "refId": "A",
+          "expr": "max by (stage) (memory_rss{job=\"collectors\"})",
+          "legendFormat": "{{stage}}"
+        }
+      ],
+      "fieldConfig": {
+        "defaults": {
+          "custom": {
+            "spanNulls": true
+          },
+          "unit": "bytes"
+        },
+        "overrides": []
+      }
+    }
+  ]
+}
diff --git a/deploy/telemetry-functions/pipelines/branch.yaml b/deploy/telemetry-functions/latency/branch.yaml
similarity index 100%
rename from deploy/telemetry-functions/pipelines/branch.yaml
rename to deploy/telemetry-functions/latency/branch.yaml
diff --git a/deploy/telemetry-functions/pipelines/detector.yaml b/deploy/telemetry-functions/latency/detector.yaml
similarity index 100%
rename from deploy/telemetry-functions/pipelines/detector.yaml
rename to deploy/telemetry-functions/latency/detector.yaml
diff --git a/deploy/telemetry-functions/pipelines/exact.yaml b/deploy/telemetry-functions/latency/exact.yaml
similarity index 100%
rename from deploy/telemetry-functions/pipelines/exact.yaml
rename to deploy/telemetry-functions/latency/exact.yaml
diff --git a/deploy/telemetry-functions/pipelines/generator-a.yaml b/deploy/telemetry-functions/latency/generator-a.yaml
similarity index 100%
rename from deploy/telemetry-functions/pipelines/generator-a.yaml
rename to deploy/telemetry-functions/latency/generator-a.yaml
diff --git a/deploy/telemetry-functions/pipelines/generator-b.yaml b/deploy/telemetry-functions/latency/generator-b.yaml
similarity index 100%
rename from deploy/telemetry-functions/pipelines/generator-b.yaml
rename to deploy/telemetry-functions/latency/generator-b.yaml
diff --git a/deploy/telemetry-functions/latency/merge.yaml b/deploy/telemetry-functions/latency/merge.yaml
new file mode 100644
index 0000000..cb392f2
--- /dev/null
+++ b/deploy/telemetry-functions/latency/merge.yaml
@@ -0,0 +1,33 @@
+# Stage 2: merge every source's sketch for a window, 5 s after it ends.
+version: otel_dataflow/v1
+engine:
+  http_admin:
+    bind_address: 0.0.0.0:8085
+policies:
+  resources:
+    core_allocation:
+      type: core_count
+      count: 1
+groups:
+  default:
+    pipelines:
+      main:
+        nodes:
+          receiver:
+            type: receiver:otlp
+            config:
+              protocols:
+                http: {listening_addr: "0.0.0.0:4318", wait_for_result: false}
+          merge:
+            type: urn:asap:processor:window_merge
+            config: {grace: 5s}
+          to_detector:
+            type: exporter:otlp_http
+            config:
+              endpoint: http://detector:4318
+              client_pool_size: 1
+              max_in_flight: 1
+              http: {timeout: 30s, compression: none}
+        connections:
+          - {from: receiver, to: merge}
+          - {from: merge, to: to_detector}
diff --git a/deploy/telemetry-functions/run.sh b/deploy/telemetry-functions/run.sh
index f4cb2fc..50afed8 100755
--- a/deploy/telemetry-functions/run.sh
+++ b/deploy/telemetry-functions/run.sh
@@ -1,5 +1,6 @@
 #!/usr/bin/env bash
 # Build asap-collector, package it, and start every stage plus the observers.
+# USE_CASE=latency (default) or cardinality picks the scenario.
 # DOCKER_COMMAND="sudo -n docker" when Docker needs sudo. Stop with:
 #   $DOCKER_COMMAND compose -f deploy/telemetry-functions/compose.yaml down
 set -euo pipefail
@@ -13,5 +14,6 @@ trap 'rm -rf "$context"' EXIT
 cp "$target/release/asap-collector" "$context/"
 strip --strip-debug "$context/asap-collector"
 "${docker[@]}" build -q -f deploy/collector.Dockerfile -t asap-collector:local "$context"
+export USE_CASE=${USE_CASE:-latency}
 "${docker[@]}" compose -f deploy/telemetry-functions/compose.yaml up -d
-echo "Grafana: http://localhost:${GRAFANA_PORT:-3000}/d/asap-latency"
+echo "Grafana: http://localhost:${GRAFANA_PORT:-3000}/d/asap-$USE_CASE"
diff --git a/docs/images/cardinality-grafana.png b/docs/images/cardinality-grafana.png
new file mode 100644
index 0000000..48412be
Binary files /dev/null and b/docs/images/cardinality-grafana.png differ