diff --git a/data_plane/benches/sketch_db.rs b/data_plane/benches/sketch_db.rs
index 78c770cb..13625786 100644
--- a/data_plane/benches/sketch_db.rs
+++ b/data_plane/benches/sketch_db.rs
@@ -456,6 +456,45 @@ fn bench_reconcile_per_batch(c: &mut Criterion) {
     g.finish();
 }
 
+fn bench_group_key_projection(c: &mut Criterion) {
+    use data_plane::precompute_engine::group_key::intern_pairs;
+
+    let cardinality = 100_000usize;
+    let labels = (0..cardinality)
+        .map(|index| (format!("region;{index}"), format!("service={index}")))
+        .collect::<Vec<_>>();
+    let mut group = c.benchmark_group("group_key_projection");
+    group.throughput(Throughput::Elements(cardinality as u64));
+    group.bench_function("cold_high_cardinality", |b| {
+        b.iter(|| {
+            for (region, service) in &labels {
+                black_box(intern_pairs([
+                    ("region", region.as_str()),
+                    ("service", service.as_str()),
+                ]));
+            }
+        });
+    });
+    // Warm the bounded interner, then measure shared-DAG reuse.
+    for (region, service) in &labels {
+        black_box(intern_pairs([
+            ("region", region.as_str()),
+            ("service", service.as_str()),
+        ]));
+    }
+    group.bench_function("warm_shared_projection", |b| {
+        b.iter(|| {
+            for (region, service) in &labels {
+                black_box(intern_pairs([
+                    ("region", region.as_str()),
+                    ("service", service.as_str()),
+                ]));
+            }
+        });
+    });
+    group.finish();
+}
+
 criterion_group!(
     benches,
     bench_append_sample,
@@ -463,5 +502,6 @@ criterion_group!(
     bench_query_range,
     bench_query_precomputes_by_agg,
     bench_reconcile_per_batch,
+    bench_group_key_projection,
 );
 criterion_main!(benches);
diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs
index bcf3e308..73a7056d 100644
--- a/data_plane/src/drivers/ingest/otel.rs
+++ b/data_plane/src/drivers/ingest/otel.rs
@@ -660,7 +660,11 @@ async fn route_otlp_to_precompute(
     // alongside the sid in the WorkerMessage so the worker can resolve
     // the source config and render emit-time labels without consulting
     // the sid → attrs reverse mapping.
-    type BucketTuple = (u64, asap_types::PolicyFingerprint, String); // (sid, policy_fp, group_key)
+    type BucketTuple = (
+        u64,
+        asap_types::PolicyFingerprint,
+        Arc<crate::precompute_engine::group_key::GroupKey>,
+    ); // (sid, policy_fp, group_key)
     type SampleTuple = (String, i64, f64);
     let mut by_bucket: HashMap<u64, (BucketTuple, Vec<SampleTuple>)> = HashMap::new();
     let mut raw_matched = 0usize;
@@ -4601,7 +4605,7 @@ mod sid_bucketing_tests {
         let groups: Vec<(
             u64,
             asap_types::PolicyFingerprint,
-            String,
+            Arc<crate::precompute_engine::group_key::GroupKey>,
             Vec<(String, i64, f64)>,
         )> = messages
             .into_iter()
diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
index 573018d8..05c47a2c 100644
--- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs
+++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
@@ -524,7 +524,11 @@ fn route_messages(
     ingest: &Arc<IngestState>,
     physical_plan: &crate::storage_engines::types::ActivePhysicalPlan,
 ) -> Vec<WorkerMessage> {
-    type Bucket = (u64, asap_types::PolicyFingerprint, String);
+    type Bucket = (
+        u64,
+        asap_types::PolicyFingerprint,
+        Arc<crate::precompute_engine::group_key::GroupKey>,
+    );
     type RoutedSample = (String, i64, f64);
     let snapshot = physical_plan.runtime_config.clone();
     let _ = crate::storage_engines::sketch_db::lifecycle::reconcile_if_config_changed(
@@ -579,7 +583,14 @@ fn route_messages(
                     .collect()
             };
             let group_key = if series_scoped {
-                sample.population_key.to_string()
+                let mut names = sample.labels.keys().map(String::as_str).collect::<Vec<_>>();
+                names.sort_unstable();
+                crate::precompute_engine::group_key::intern_pairs(names.into_iter().map(|name| {
+                    (
+                        name,
+                        sample.labels.get(name).map(String::as_str).unwrap_or(""),
+                    )
+                }))
             } else {
                 group_key
             };
@@ -1229,7 +1240,7 @@ mod tests {
         else {
             panic!("expected GroupSamples");
         };
-        assert_eq!(group_key, "api");
+        assert_eq!(group_key.values().labels, vec!["api"]);
         assert_eq!(
             samples,
             vec![("requests_total{job=\"api\"}".into(), 100, 4.0)]
diff --git a/data_plane/src/precompute_engine/group_key.rs b/data_plane/src/precompute_engine/group_key.rs
new file mode 100644
index 00000000..dd3c5684
--- /dev/null
+++ b/data_plane/src/precompute_engine/group_key.rs
@@ -0,0 +1,142 @@
+use std::collections::HashMap;
+use std::sync::{Arc, OnceLock};
+
+use moka::sync::Cache;
+
+use crate::storage_engines::types::KeyByLabelValues;
+
+/// Collision-free, positional identity for a physical GROUP BY partition.
+/// Label names are retained so different DAG projections cannot alias merely
+/// because their values happen to be equal.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct GroupKey {
+    labels: Arc<[(Arc<str>, Arc<str>)]>,
+    canonical: Arc<[u8]>,
+}
+
+impl GroupKey {
+    pub fn project<'a>(
+        names: impl IntoIterator<Item = &'a str>,
+        labels: &HashMap<String, String>,
+    ) -> Self {
+        Self::new(
+            names
+                .into_iter()
+                .map(|name| (name, labels.get(name).map(String::as_str).unwrap_or(""))),
+        )
+    }
+
+    pub fn new<'a>(pairs: impl IntoIterator<Item = (&'a str, &'a str)>) -> Self {
+        let labels: Arc<[(Arc<str>, Arc<str>)]> = pairs
+            .into_iter()
+            .map(|(name, value)| (Arc::from(name), Arc::from(value)))
+            .collect::<Vec<_>>()
+            .into();
+        let mut canonical = Vec::new();
+        canonical.extend_from_slice(b"ASAPGK\x01");
+        canonical.extend_from_slice(&(labels.len() as u32).to_be_bytes());
+        for (name, value) in labels.iter() {
+            put_component(&mut canonical, name.as_bytes());
+            put_component(&mut canonical, value.as_bytes());
+        }
+        Self {
+            labels,
+            canonical: canonical.into(),
+        }
+    }
+
+    pub fn canonical_bytes(&self) -> &[u8] {
+        &self.canonical
+    }
+
+    pub fn values(&self) -> KeyByLabelValues {
+        KeyByLabelValues::new_with_labels(
+            self.labels
+                .iter()
+                .map(|(_, value)| value.to_string())
+                .collect(),
+        )
+    }
+
+    pub fn as_population_labels(&self) -> std::collections::BTreeMap<String, String> {
+        self.labels
+            .iter()
+            .map(|(name, value)| (name.to_string(), value.to_string()))
+            .collect()
+    }
+}
+
+impl std::fmt::Display for GroupKey {
+    fn fmt(&self, output: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        output.write_str("{")?;
+        for (index, (name, value)) in self.labels.iter().enumerate() {
+            if index != 0 {
+                output.write_str(",")?;
+            }
+            write!(output, "{name}={value:?}")?;
+        }
+        output.write_str("}")
+    }
+}
+
+fn put_component(target: &mut Vec<u8>, value: &[u8]) {
+    target.extend_from_slice(&(value.len() as u32).to_be_bytes());
+    target.extend_from_slice(value);
+}
+
+/// Bounded interner shared by ingress adapters. Repeated DAG consumers with
+/// the same projection carry one immutable key allocation into workers.
+pub fn intern(key: GroupKey) -> Arc<GroupKey> {
+    intern_pairs(
+        key.labels
+            .iter()
+            .map(|(name, value)| (name.as_ref(), value.as_ref())),
+    )
+}
+
+/// Project and intern from borrowed labels. The hot path builds only the
+/// compact lookup bytes on a cache hit; label/name strings are allocated once
+/// when a new high-cardinality group first appears.
+pub fn intern_pairs<'a>(pairs: impl IntoIterator<Item = (&'a str, &'a str)>) -> Arc<GroupKey> {
+    let pairs = pairs.into_iter().collect::<Vec<_>>();
+    let mut encoded = Vec::with_capacity(
+        11 + pairs
+            .iter()
+            .map(|(name, value)| name.len() + value.len() + 8)
+            .sum::<usize>(),
+    );
+    encoded.extend_from_slice(b"ASAPGK\x01");
+    encoded.extend_from_slice(&(pairs.len() as u32).to_be_bytes());
+    for (name, value) in &pairs {
+        put_component(&mut encoded, name.as_bytes());
+        put_component(&mut encoded, value.as_bytes());
+    }
+    static INTERNER: OnceLock<Cache<Vec<u8>, Arc<GroupKey>>> = OnceLock::new();
+    let interner = INTERNER.get_or_init(|| Cache::builder().max_capacity(131_072).build());
+    interner.get_with(encoded, || Arc::new(GroupKey::new(pairs)))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn delimiters_names_order_and_missing_values_cannot_alias() {
+        let a = GroupKey::new([("x", "a;b"), ("y", "c")]);
+        let b = GroupKey::new([("x", "a"), ("y", "b;c")]);
+        let reordered = GroupKey::new([("y", "c"), ("x", "a;b")]);
+        let other_names = GroupKey::new([("p", "a;b"), ("q", "c")]);
+        let missing = GroupKey::new([("x", ""), ("y", "c")]);
+        for other in [&b, &reordered, &other_names, &missing] {
+            assert_ne!(a.canonical_bytes(), other.canonical_bytes());
+        }
+        assert_eq!(a.values().labels, vec!["a;b", "c"]);
+    }
+
+    #[test]
+    fn interner_reuses_equal_projection() {
+        let left = intern(GroupKey::new([("region", "us-east"), ("job", "api")]));
+        let right = intern(GroupKey::new([("region", "us-east"), ("job", "api")]));
+        assert!(Arc::ptr_eq(&left, &right));
+    }
+}
diff --git a/data_plane/src/precompute_engine/ingest_handler.rs b/data_plane/src/precompute_engine/ingest_handler.rs
index b0fc300a..c6a7c2af 100644
--- a/data_plane/src/precompute_engine/ingest_handler.rs
+++ b/data_plane/src/precompute_engine/ingest_handler.rs
@@ -260,7 +260,10 @@ impl IngestState {
     /// Extract the group key for a series key against a given aggregation
     /// config. Re-exports the module-private helper so that out-of-module
     /// ingest sources (e.g. OTLP) can reuse it.
-    pub fn extract_group_key_for(series_key: &str, config: &AggregationConfig) -> String {
+    pub fn extract_group_key_for(
+        series_key: &str,
+        config: &AggregationConfig,
+    ) -> Arc<crate::precompute_engine::group_key::GroupKey> {
         extract_group_key(series_key, config)
     }
 
@@ -274,33 +277,33 @@ impl IngestState {
     pub fn extract_group_key_from_labels(
         labels: &std::collections::HashMap<String, String>,
         config: &AggregationConfig,
-    ) -> String {
-        let mut values = Vec::with_capacity(config.grouping_labels.labels.len());
-        for label_name in &config.grouping_labels.labels {
-            values.push(
-                labels
-                    .get(label_name.as_str())
-                    .map(|s| s.as_str())
-                    .unwrap_or(""),
-            );
-        }
-        values.join(";")
+    ) -> Arc<crate::precompute_engine::group_key::GroupKey> {
+        crate::precompute_engine::group_key::intern_pairs(config.grouping_labels.labels.iter().map(
+            |name| {
+                (
+                    name.as_str(),
+                    labels.get(name).map(String::as_str).unwrap_or(""),
+                )
+            },
+        ))
     }
 }
 
 /// Extract the group key (grouping label values joined by semicolons)
 /// for a given series key and aggregation config.
-fn extract_group_key(series_key: &str, config: &AggregationConfig) -> String {
+fn extract_group_key(
+    series_key: &str,
+    config: &AggregationConfig,
+) -> Arc<crate::precompute_engine::group_key::GroupKey> {
     let labels = parse_labels_from_series_key(series_key);
-    let mut values = Vec::new();
-    for label_name in &config.grouping_labels.labels {
-        if let Some(val) = labels.get(label_name.as_str()) {
-            values.push(*val);
-        } else {
-            values.push("");
-        }
-    }
-    values.join(";")
+    crate::precompute_engine::group_key::intern_pairs(config.grouping_labels.labels.iter().map(
+        |name| {
+            (
+                name.as_str(),
+                labels.get(name.as_str()).copied().unwrap_or(""),
+            )
+        },
+    ))
 }
 
 #[cfg(test)]
diff --git a/data_plane/src/precompute_engine/mod.rs b/data_plane/src/precompute_engine/mod.rs
index 3a7e128b..55bf282d 100644
--- a/data_plane/src/precompute_engine/mod.rs
+++ b/data_plane/src/precompute_engine/mod.rs
@@ -2,6 +2,7 @@ pub mod accumulator_factory;
 pub mod config;
 mod engine;
 pub mod frame_lineage;
+pub mod group_key;
 pub mod ingest_handler;
 pub(crate) mod metrics;
 pub mod operators;
diff --git a/data_plane/src/precompute_engine/series_router.rs b/data_plane/src/precompute_engine/series_router.rs
index 475620b8..6c2d8354 100644
--- a/data_plane/src/precompute_engine/series_router.rs
+++ b/data_plane/src/precompute_engine/series_router.rs
@@ -1,8 +1,10 @@
+use crate::precompute_engine::group_key::GroupKey;
 use crate::storage_engines::types::AggregateCore;
 use asap_types::PolicyFingerprint;
 use futures::future::try_join_all;
 use std::collections::HashMap;
 use std::fmt;
+use std::sync::Arc;
 use std::time::Instant;
 use tokio::sync::mpsc;
 use xxhash_rust::xxh64::xxh64;
@@ -48,7 +50,7 @@ pub enum WorkerMessage {
         /// Grouping label values joined by semicolons (e.g. "constant").
         /// Empty string if the aggregation has no grouping labels. Used
         /// at emit time to render the output's `KeyByLabelValues`.
-        group_key: String,
+        group_key: Arc<GroupKey>,
         /// Each entry: (series_key, timestamp_ms, value).
         /// series_key is needed for keyed (MultipleSubpopulation) accumulators
         /// to extract the aggregated-label key.
@@ -75,7 +77,7 @@ pub enum WorkerMessage {
         /// Grouping label values joined by semicolons, matching the
         /// format produced by `IngestState::extract_group_key_for`.
         /// Used at emit time to render the output's `KeyByLabelValues`.
-        group_key: String,
+        group_key: Arc<GroupKey>,
         /// Wall-clock timestamp the sketch refers to (millis since epoch).
         /// Used to place the sketch into the correct pane.
         timestamp_ms: i64,
diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs
index 6cec5063..70514c84 100644
--- a/data_plane/src/precompute_engine/worker.rs
+++ b/data_plane/src/precompute_engine/worker.rs
@@ -2,6 +2,7 @@ use crate::precompute_engine::accumulator_factory::{
     create_accumulator_updater, AccumulatorUpdater,
 };
 use crate::precompute_engine::config::LateDataPolicy;
+use crate::precompute_engine::group_key::GroupKey;
 use crate::precompute_engine::metrics::record_late_input;
 use crate::precompute_engine::operators::sum_accumulator::SumAccumulator;
 use crate::precompute_engine::output_sink::OutputSink;
@@ -21,8 +22,6 @@ use std::sync::Arc;
 use tokio::sync::mpsc;
 use tracing::{debug, debug_span, info, warn};
 
-const POPULATION_GROUP_KEY_PREFIX: &str = "__asap_population__";
-
 /// Per-bucket aggregation state: window manager + active pane accumulators.
 ///
 /// B7.6 (schema-retirement #5): one `GroupState` per `sid`, where `sid` is
@@ -50,7 +49,7 @@ struct GroupState {
     /// can render the output's `KeyByLabelValues` without consulting the
     /// sid → attrs reverse mapping. Format matches the input messages'
     /// `group_key` field.
-    group_key: String,
+    group_key: Arc<GroupKey>,
     window_manager: WindowManager,
     /// Active panes for raw-sample accumulation, keyed by pane_start_ms.
     active_panes: BTreeMap<i64, Box<dyn AccumulatorUpdater>>,
@@ -389,7 +388,7 @@ impl Worker {
         &mut self,
         sid: u64,
         policy_fp: PolicyFingerprint,
-        group_key: &str,
+        group_key: &Arc<GroupKey>,
     ) -> Option<&mut GroupState> {
         if !self.group_states.contains_key(&sid) {
             let snap = self.hot_reload.snapshot();
@@ -404,7 +403,7 @@ impl Worker {
                 ),
                 config,
                 policy_fp,
-                group_key: group_key.to_string(),
+                group_key: Arc::clone(group_key),
                 active_panes: BTreeMap::new(),
                 counter_previous: HashMap::new(),
                 sketch_panes: BTreeMap::new(),
@@ -431,7 +430,7 @@ impl Worker {
         &mut self,
         sid: u64,
         policy_fp: PolicyFingerprint,
-        group_key: &str,
+        group_key: &Arc<GroupKey>,
         samples: Vec<(String, i64, f64)>, // (series_key, timestamp_ms, value)
     ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         let worker_id = self.id;
@@ -648,7 +647,7 @@ impl Worker {
         &mut self,
         sid: u64,
         policy_fp: PolicyFingerprint,
-        group_key: &str,
+        group_key: &Arc<GroupKey>,
         timestamp_ms: i64,
         incoming: Box<dyn AggregateCore>,
     ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
@@ -1131,17 +1130,13 @@ fn watermark_for_event_time(max_event_time_ms: i64, allowed_lateness_ms: i64) ->
     }
 }
 
-fn build_group_key_label_values(group_key: &str) -> KeyByLabelValues {
-    if let Some(labels) = population_labels_from_group_key(group_key) {
-        return KeyByLabelValues::new_with_labels(labels.into_values().collect());
-    }
-    let labels: Vec<String> = group_key.split(';').map(|s| s.to_string()).collect();
-    KeyByLabelValues::new_with_labels(labels)
+fn build_group_key_label_values(group_key: &GroupKey) -> KeyByLabelValues {
+    group_key.values()
 }
 
-fn population_labels_from_group_key(group_key: &str) -> Option<BTreeMap<String, String>> {
-    let encoded = group_key.strip_prefix(POPULATION_GROUP_KEY_PREFIX)?;
-    serde_json::from_str(encoded).ok()
+fn population_labels_from_group_key(group_key: &GroupKey) -> Option<BTreeMap<String, String>> {
+    let labels = group_key.as_population_labels();
+    (!labels.is_empty()).then_some(labels)
 }
 
 fn precomputed_output_for_group(
@@ -1149,7 +1144,7 @@ fn precomputed_output_for_group(
     end_timestamp: u64,
     key: KeyByLabelValues,
     policy_fp: PolicyFingerprint,
-    group_key: &str,
+    group_key: &GroupKey,
 ) -> PrecomputedOutput {
     PrecomputedOutput::new(start_timestamp, end_timestamp, Some(key), policy_fp)
         .with_population_labels(population_labels_from_group_key(group_key))
@@ -1467,6 +1462,16 @@ fn merge_sketch_panes_for_window(
 mod tests {
     use super::*;
 
+    fn test_group_key(value: &str) -> Arc<GroupKey> {
+        if value.is_empty() {
+            return crate::precompute_engine::group_key::intern_pairs(std::iter::empty::<(
+                &str,
+                &str,
+            )>());
+        }
+        crate::precompute_engine::group_key::intern_pairs([("group", value)])
+    }
+
     #[test]
     fn counter_delta_is_reset_aware_series_local_and_cross_pane_safe() {
         let mut previous = HashMap::new();
@@ -1759,19 +1764,39 @@ mod tests {
         // All go to the same bucket (sid=1, group_key="")
         let pf = PolicyFingerprint(1);
         worker
-            .process_group_samples(1, pf, "", group_samples("cpu", vec![(1000, 1.0)]))
+            .process_group_samples(
+                1,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(1000, 1.0)]),
+            )
             .unwrap();
         worker
-            .process_group_samples(1, pf, "", group_samples("cpu", vec![(5000, 2.0)]))
+            .process_group_samples(
+                1,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(5000, 2.0)]),
+            )
             .unwrap();
         worker
-            .process_group_samples(1, pf, "", group_samples("cpu", vec![(9000, 3.0)]))
+            .process_group_samples(
+                1,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(9000, 3.0)]),
+            )
             .unwrap();
         assert_eq!(sink.len(), 0);
 
         // Sample at t=10000ms closes [0, 10000)
         worker
-            .process_group_samples(1, pf, "", group_samples("cpu", vec![(10000, 100.0)]))
+            .process_group_samples(
+                1,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(10000, 100.0)]),
+            )
             .unwrap();
 
         let captured = sink.drain();
@@ -1826,7 +1851,7 @@ mod tests {
             .process_group_samples(
                 1,
                 pf,
-                "",
+                &test_group_key(""),
                 vec![
                     ("cpu{host=\"A\"}".to_string(), 1000, 10.0),
                     ("cpu{host=\"B\"}".to_string(), 2000, 20.0),
@@ -1840,7 +1865,7 @@ mod tests {
             .process_group_samples(
                 1,
                 pf,
-                "",
+                &test_group_key(""),
                 group_samples("cpu{host=\"A\"}", vec![(10000, 0.0)]),
             )
             .unwrap();
@@ -1898,7 +1923,7 @@ mod tests {
             .process_group_samples(
                 sid_constant,
                 pf,
-                "constant",
+                &test_group_key("constant"),
                 group_samples("cpu{pattern=\"constant\"}", vec![(1000, 5.0)]),
             )
             .unwrap();
@@ -1907,7 +1932,7 @@ mod tests {
             .process_group_samples(
                 sid_sine,
                 pf,
-                "sine",
+                &test_group_key("sine"),
                 group_samples("cpu{pattern=\"sine\"}", vec![(2000, 7.0)]),
             )
             .unwrap();
@@ -1917,7 +1942,7 @@ mod tests {
             .process_group_samples(
                 sid_constant,
                 pf,
-                "constant",
+                &test_group_key("constant"),
                 group_samples("cpu{pattern=\"constant\"}", vec![(10000, 0.0)]),
             )
             .unwrap();
@@ -1925,7 +1950,7 @@ mod tests {
             .process_group_samples(
                 sid_sine,
                 pf,
-                "sine",
+                &test_group_key("sine"),
                 group_samples("cpu{pattern=\"sine\"}", vec![(10000, 0.0)]),
             )
             .unwrap();
@@ -1973,7 +1998,7 @@ mod tests {
             .process_group_samples(
                 1,
                 pf,
-                "constant",
+                &test_group_key("constant"),
                 vec![
                     (
                         "latency{pattern=\"constant\",host=\"a\"}".to_string(),
@@ -1999,7 +2024,7 @@ mod tests {
             .process_group_samples(
                 1,
                 pf,
-                "constant",
+                &test_group_key("constant"),
                 group_samples(
                     "latency{pattern=\"constant\",host=\"a\"}",
                     vec![(10000, 0.0)],
@@ -2051,14 +2076,24 @@ mod tests {
         // Sample at t=15000ms → goes to pane 10000ms
         let pf = PolicyFingerprint(2);
         worker
-            .process_group_samples(2, pf, "", group_samples("cpu", vec![(15_000, 42.0)]))
+            .process_group_samples(
+                2,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(15_000, 42.0)]),
+            )
             .unwrap();
         assert_eq!(sink.len(), 0);
 
         // Sample at t=45000ms → advances watermark to 45000ms
         // Closes windows [0, 30000) and [10000, 40000)
         worker
-            .process_group_samples(2, pf, "", group_samples("cpu", vec![(45_000, 0.0)]))
+            .process_group_samples(
+                2,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(45_000, 0.0)]),
+            )
             .unwrap();
 
         let captured = sink.drain();
@@ -2103,10 +2138,20 @@ mod tests {
         );
 
         worker
-            .process_group_samples(2, policy, "", group_samples("cpu", vec![(15_000, 42.0)]))
+            .process_group_samples(
+                2,
+                policy,
+                &test_group_key(""),
+                group_samples("cpu", vec![(15_000, 42.0)]),
+            )
             .unwrap();
         worker
-            .process_group_samples(2, policy, "", group_samples("cpu", vec![(45_000, 0.0)]))
+            .process_group_samples(
+                2,
+                policy,
+                &test_group_key(""),
+                group_samples("cpu", vec![(45_000, 0.0)]),
+            )
             .unwrap();
 
         let captured = sink.drain();
@@ -2163,7 +2208,7 @@ mod tests {
             .process_group_samples(
                 3,
                 pf,
-                "",
+                &test_group_key(""),
                 vec![
                     ("cpu{host=\"A\"}".to_string(), 1000, 10.0),
                     ("cpu{host=\"B\"}".to_string(), 2000, 20.0),
@@ -2176,7 +2221,7 @@ mod tests {
             .process_group_samples(
                 3,
                 pf,
-                "",
+                &test_group_key(""),
                 group_samples("cpu{host=\"A\"}", vec![(10000, 0.0)]),
             )
             .unwrap();
@@ -2251,13 +2296,23 @@ mod tests {
         // Establish watermark at t=20000ms
         let pf = PolicyFingerprint(4);
         worker
-            .process_group_samples(4, pf, "", group_samples("cpu", vec![(20_000, 1.0)]))
+            .process_group_samples(
+                4,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(20_000, 1.0)]),
+            )
             .unwrap();
         let _ = sink.drain();
 
         // Send a late sample
         worker
-            .process_group_samples(4, pf, "", group_samples("cpu", vec![(5_000, 99.0)]))
+            .process_group_samples(
+                4,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(5_000, 99.0)]),
+            )
             .unwrap();
 
         assert_eq!(sink.len(), 0, "late sample should be dropped");
@@ -2306,16 +2361,31 @@ mod tests {
         // budget permits closing [0, 10s).
         let pf = PolicyFingerprint(5);
         worker
-            .process_group_samples(5, pf, "", group_samples("cpu", vec![(500, 1.0)]))
+            .process_group_samples(
+                5,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(500, 1.0)]),
+            )
             .unwrap();
         worker
-            .process_group_samples(5, pf, "", group_samples("cpu", vec![(30_000, 0.0)]))
+            .process_group_samples(
+                5,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(30_000, 0.0)]),
+            )
             .unwrap();
         let _ = sink.drain();
 
         // Send late sample for evicted pane
         worker
-            .process_group_samples(5, pf, "", group_samples("cpu", vec![(8_000, 55.0)]))
+            .process_group_samples(
+                5,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(8_000, 55.0)]),
+            )
             .unwrap();
 
         let captured = sink.drain();
@@ -2382,7 +2452,7 @@ aggregations:
             .process_group_samples(
                 sid,
                 pf,
-                "",
+                &test_group_key(""),
                 group_samples("requests_total", vec![(1_000, 3.0)]),
             )
             .unwrap();
@@ -2390,7 +2460,7 @@ aggregations:
             .process_group_samples(
                 sid,
                 pf,
-                "",
+                &test_group_key(""),
                 group_samples("requests_total", vec![(5_000, 4.0)]),
             )
             .unwrap();
@@ -2398,7 +2468,7 @@ aggregations:
             .process_group_samples(
                 sid,
                 pf,
-                "",
+                &test_group_key(""),
                 group_samples("requests_total", vec![(9_000, 5.0)]),
             )
             .unwrap();
@@ -2408,7 +2478,7 @@ aggregations:
             .process_group_samples(
                 sid,
                 pf,
-                "",
+                &test_group_key(""),
                 group_samples("requests_total", vec![(10_000, 0.0)]),
             )
             .unwrap();
@@ -2462,19 +2532,29 @@ aggregations:
 
     #[test]
     fn test_build_group_key_label_values() {
-        let key = build_group_key_label_values("constant");
+        let single = test_group_key("constant");
+        let key = build_group_key_label_values(&single);
         assert_eq!(key.labels, vec!["constant".to_string()]);
 
-        let key = build_group_key_label_values("us-east;svc-a");
-        assert_eq!(key.labels, vec!["us-east".to_string(), "svc-a".to_string()]);
+        let delimited = crate::precompute_engine::group_key::intern_pairs([
+            ("region", "us-east;1"),
+            ("service", "svc-a"),
+        ]);
+        let key = build_group_key_label_values(&delimited);
+        assert_eq!(
+            key.labels,
+            vec!["us-east;1".to_string(), "svc-a".to_string()]
+        );
 
-        let key = build_group_key_label_values("");
-        assert_eq!(key.labels, vec!["".to_string()]);
+        let empty = test_group_key("");
+        let key = build_group_key_label_values(&empty);
+        assert!(key.labels.is_empty());
 
-        let group = r#"__asap_population__{"instance":"a","job":"api"}"#;
-        let key = build_group_key_label_values(group);
+        let group =
+            crate::precompute_engine::group_key::intern_pairs([("instance", "a"), ("job", "api")]);
+        let key = build_group_key_label_values(&group);
         assert_eq!(key.labels, vec!["a".to_string(), "api".to_string()]);
-        let output = precomputed_output_for_group(0, 5_000, key, PolicyFingerprint(7), group);
+        let output = precomputed_output_for_group(0, 5_000, key, PolicyFingerprint(7), &group);
         assert_eq!(
             output.population_labels,
             Some(BTreeMap::from([
@@ -2517,7 +2597,7 @@ aggregations:
             .process_group_samples(
                 sid_a,
                 pf,
-                "groupA",
+                &test_group_key("groupA"),
                 group_samples("cpu", vec![(5_000, 1.0)]),
             )
             .unwrap();
@@ -2526,7 +2606,7 @@ aggregations:
             .process_group_samples(
                 sid_b,
                 pf,
-                "groupB",
+                &test_group_key("groupB"),
                 group_samples("cpu", vec![(5_000, 2.0)]),
             )
             .unwrap();
@@ -2537,7 +2617,7 @@ aggregations:
             .process_group_samples(
                 sid_a,
                 pf,
-                "groupA",
+                &test_group_key("groupA"),
                 group_samples("cpu", vec![(100_000, 3.0)]),
             )
             .unwrap();
@@ -2566,7 +2646,7 @@ aggregations:
             .process_group_samples(
                 sid_b,
                 pf,
-                "groupB",
+                &test_group_key("groupB"),
                 group_samples("cpu", vec![(5_000, 4.0)]),
             )
             .unwrap();
@@ -2612,12 +2692,22 @@ aggregations:
         let pf = PolicyFingerprint(1);
 
         worker
-            .process_group_samples(1, pf, "", group_samples("cpu", vec![(0, 1.0)]))
+            .process_group_samples(
+                1,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(0, 1.0)]),
+            )
             .unwrap();
         worker.flush_all().unwrap();
         worker.flush_all().unwrap();
         worker
-            .process_group_samples(1, pf, "", group_samples("cpu", vec![(0, 2.0)]))
+            .process_group_samples(
+                1,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(0, 2.0)]),
+            )
             .unwrap();
         worker.force_close_all().unwrap();
 
@@ -2654,10 +2744,20 @@ aggregations:
         let pf = PolicyFingerprint(1);
 
         worker
-            .process_group_samples(1, pf, "", group_samples("cpu", vec![(5_000, 1.0)]))
+            .process_group_samples(
+                1,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(5_000, 1.0)]),
+            )
             .unwrap();
         worker
-            .process_group_samples(1, pf, "", group_samples("cpu", vec![(10_000, 2.0)]))
+            .process_group_samples(
+                1,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(10_000, 2.0)]),
+            )
             .unwrap();
         assert_eq!(
             sink.len(),
@@ -2666,7 +2766,12 @@ aggregations:
         );
 
         worker
-            .process_group_samples(1, pf, "", group_samples("cpu", vec![(15_000, 3.0)]))
+            .process_group_samples(
+                1,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(15_000, 3.0)]),
+            )
             .unwrap();
         let emitted = sink.drain();
         assert_eq!(emitted.len(), 1);
@@ -2699,7 +2804,7 @@ aggregations:
             .process_group_samples(
                 1,
                 PolicyFingerprint(1),
-                "",
+                &test_group_key(""),
                 group_samples(
                     "cpu",
                     vec![
@@ -2759,7 +2864,12 @@ aggregations:
         // Send data at t=50s
         let pf = PolicyFingerprint(1);
         worker
-            .process_group_samples(1, pf, "", group_samples("cpu", vec![(50_000, 1.0)]))
+            .process_group_samples(
+                1,
+                pf,
+                &test_group_key(""),
+                group_samples("cpu", vec![(50_000, 1.0)]),
+            )
             .unwrap();
 
         // Flush should publish worker watermark
@@ -2835,7 +2945,7 @@ aggregations:
         for i in 0..10 {
             let s = make_ddsketch(0.01, &[1.0 + i as f64, 2.0, 3.0]);
             worker
-                .process_accumulator_input(sid, pf, "us-east", 60_000, Box::new(s))
+                .process_accumulator_input(sid, pf, &test_group_key("us-east"), 60_000, Box::new(s))
                 .expect("first batch must process");
         }
         assert_eq!(
@@ -2851,7 +2961,7 @@ aggregations:
         // and the output is emitted.
         let s2 = make_ddsketch(0.01, &[5.0, 6.0]);
         worker
-            .process_accumulator_input(sid, pf, "us-east", 120_000, Box::new(s2))
+            .process_accumulator_input(sid, pf, &test_group_key("us-east"), 120_000, Box::new(s2))
             .expect("second batch must process");
 
         let captured = sink.drain();
@@ -2930,25 +3040,49 @@ aggregations:
         for i in 0..3 {
             let s = make_ddsketch(0.01, &[100.0 + i as f64]);
             worker
-                .process_accumulator_input(sid_east, pf, "us-east", 60_000, Box::new(s))
+                .process_accumulator_input(
+                    sid_east,
+                    pf,
+                    &test_group_key("us-east"),
+                    60_000,
+                    Box::new(s),
+                )
                 .unwrap();
         }
         // Two sketches in a different zone.
         for i in 0..2 {
             let s = make_ddsketch(0.01, &[200.0 + i as f64]);
             worker
-                .process_accumulator_input(sid_west, pf, "us-west", 60_000, Box::new(s))
+                .process_accumulator_input(
+                    sid_west,
+                    pf,
+                    &test_group_key("us-west"),
+                    60_000,
+                    Box::new(s),
+                )
                 .unwrap();
         }
 
         // Advance the watermark past 90_000 to close window [60_000, 90_000).
         let s = make_ddsketch(0.01, &[1.0]);
         worker
-            .process_accumulator_input(sid_east, pf, "us-east", 120_000, Box::new(s))
+            .process_accumulator_input(
+                sid_east,
+                pf,
+                &test_group_key("us-east"),
+                120_000,
+                Box::new(s),
+            )
             .unwrap();
         let s = make_ddsketch(0.01, &[1.0]);
         worker
-            .process_accumulator_input(sid_west, pf, "us-west", 120_000, Box::new(s))
+            .process_accumulator_input(
+                sid_west,
+                pf,
+                &test_group_key("us-west"),
+                120_000,
+                Box::new(s),
+            )
             .unwrap();
 
         let captured = sink.drain();
@@ -3078,7 +3212,7 @@ aggregations:
         for i in 0..10 {
             let s = make_ddsketch(0.01, &[1.0 + i as f64]);
             worker
-                .process_accumulator_input(sid, pf, "us-east", 0, Box::new(s))
+                .process_accumulator_input(sid, pf, &test_group_key("us-east"), 0, Box::new(s))
                 .expect("ingest must accept frozen-event-time sketches");
         }
         assert_eq!(
@@ -3183,7 +3317,12 @@ aggregations:
             let value = 1.0 + i as f64;
             expected_sum += value;
             worker
-                .process_group_samples(7, pf, "", group_samples("netflow_bytes", vec![(0, value)]))
+                .process_group_samples(
+                    7,
+                    pf,
+                    &test_group_key(""),
+                    group_samples("netflow_bytes", vec![(0, value)]),
+                )
                 .unwrap();
         }
 
@@ -3196,7 +3335,12 @@ aggregations:
         wall_clock.store(1_007_000, Ordering::Relaxed);
         expected_sum += 8.0;
         worker
-            .process_group_samples(7, pf, "", group_samples("netflow_bytes", vec![(0, 8.0)]))
+            .process_group_samples(
+                7,
+                pf,
+                &test_group_key(""),
+                group_samples("netflow_bytes", vec![(0, 8.0)]),
+            )
             .unwrap();
 
         wall_clock.store(1_013_001, Ordering::Relaxed);
@@ -3241,7 +3385,7 @@ aggregations:
                 .process_group_samples(
                     9,
                     pf,
-                    "",
+                    &test_group_key(""),
                     group_samples("netflow_bytes", vec![(0, 1.0 + i as f64)]),
                 )
                 .unwrap();
@@ -3264,7 +3408,12 @@ aggregations:
         // mergeable correction instead of being silently dropped.
         wall_clock.store(3_007_000, Ordering::Relaxed);
         worker
-            .process_group_samples(9, pf, "", group_samples("netflow_bytes", vec![(0, 8.0)]))
+            .process_group_samples(
+                9,
+                pf,
+                &test_group_key(""),
+                group_samples("netflow_bytes", vec![(0, 8.0)]),
+            )
             .unwrap();
         let mut corrections = sink.drain();
         assert_eq!(corrections.len(), 1, "late input must emit a correction");
@@ -3306,7 +3455,7 @@ aggregations:
             .process_group_samples(
                 11,
                 PolicyFingerprint(11),
-                "",
+                &test_group_key(""),
                 group_samples("netflow_bytes", vec![(0, 1.0)]),
             )
             .unwrap();
@@ -3349,7 +3498,7 @@ aggregations:
                 .process_accumulator_input(
                     80,
                     pf,
-                    "us-east",
+                    &test_group_key("us-east"),
                     0,
                     Box::new(make_ddsketch(0.01, &[1.0 + i as f64])),
                 )
@@ -3362,7 +3511,13 @@ aggregations:
 
         wall_clock.store(2_007_000, Ordering::Relaxed);
         worker
-            .process_accumulator_input(80, pf, "us-east", 0, Box::new(make_ddsketch(0.01, &[8.0])))
+            .process_accumulator_input(
+                80,
+                pf,
+                &test_group_key("us-east"),
+                0,
+                Box::new(make_ddsketch(0.01, &[8.0])),
+            )
             .unwrap();
 
         wall_clock.store(2_013_001, Ordering::Relaxed);
@@ -3407,7 +3562,7 @@ aggregations:
                 .process_accumulator_input(
                     100,
                     pf,
-                    "us-east",
+                    &test_group_key("us-east"),
                     0,
                     Box::new(make_ddsketch(0.01, &[1.0 + i as f64])),
                 )
@@ -3422,7 +3577,13 @@ aggregations:
 
         wall_clock.store(4_007_000, Ordering::Relaxed);
         worker
-            .process_accumulator_input(100, pf, "us-east", 0, Box::new(make_ddsketch(0.01, &[8.0])))
+            .process_accumulator_input(
+                100,
+                pf,
+                &test_group_key("us-east"),
+                0,
+                Box::new(make_ddsketch(0.01, &[8.0])),
+            )
             .unwrap();
         let mut corrections = sink.drain();
         assert_eq!(corrections.len(), 1, "late sketch must emit a correction");
@@ -3470,7 +3631,13 @@ aggregations:
 
         let s = make_ddsketch(0.01, &[42.0]);
         worker
-            .process_accumulator_input(1, PolicyFingerprint(1), "us-east", 0, Box::new(s))
+            .process_accumulator_input(
+                1,
+                PolicyFingerprint(1),
+                &test_group_key("us-east"),
+                0,
+                Box::new(s),
+            )
             .unwrap();
 
         // Even after a wall-clock eternity, no emit happens with
@@ -3520,7 +3687,7 @@ aggregations:
             .process_group_samples(
                 1,
                 PolicyFingerprint(1),
-                "",
+                &test_group_key(""),
                 vec![
                     ("gauge{job=\"api\"}".into(), 1000, 10.0),
                     ("gauge{job=\"api\"}".into(), 2000, 30.0),
@@ -3574,7 +3741,7 @@ aggregations:
             .process_group_samples(
                 1,
                 PolicyFingerprint(1),
-                "",
+                &test_group_key(""),
                 vec![
                     ("requests_total{instance=\"a\"}".into(), 1000, 100.0),
                     ("requests_total{instance=\"b\"}".into(), 1000, 50.0),
@@ -3622,7 +3789,7 @@ aggregations:
         tx.send(WorkerMessage::GroupSamples {
             sid: 1,
             policy_fp: PolicyFingerprint(1),
-            group_key: "".into(),
+            group_key: test_group_key(""),
             samples: group_samples("cpu", vec![(1000, 2.0), (2000, 3.0)]),
             ingest_received_at: std::time::Instant::now(),
         })
@@ -3685,7 +3852,7 @@ aggregations:
         tx.send(WorkerMessage::GroupSamples {
             sid: 1,
             policy_fp: PolicyFingerprint(1),
-            group_key: "".into(),
+            group_key: test_group_key(""),
             samples: group_samples("cpu", vec![(1000, 2.0)]),
             ingest_received_at: std::time::Instant::now(),
         })
@@ -3730,7 +3897,7 @@ aggregations:
                 .process_group_samples(
                     1,
                     pf,
-                    "",
+                    &test_group_key(""),
                     group_samples("cpu", vec![(1_000 + i * 100, 1.0)]),
                 )
                 .unwrap();
@@ -3800,7 +3967,7 @@ aggregations:
         for i in 0..10 {
             let s = make_ddsketch(0.01, &[1.0 + i as f64]);
             worker
-                .process_accumulator_input(51, pf, "us-east", 0, Box::new(s))
+                .process_accumulator_input(51, pf, &test_group_key("us-east"), 0, Box::new(s))
                 .unwrap();
         }