diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index d37872e8c..0e99bb4d3 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -1489,8 +1489,9 @@ impl BackendLocalPlanningSnapshot {
     }
 }
 
-/// Raw accumulators do not retain arbitrary source labels. Preserve native semantics
-/// unless the selected DAG explicitly authorizes pooling the source entities.
+/// Admit per-entity raw state only when the installed partition contract and
+/// scalar input evaluator preserve source rows. Composite updates still require
+/// an executable maintenance evaluator; a partition flag cannot authorize them.
 fn has_unsafe_raw_entity_leaf(
     node: &Rc<SummaryNode>,
     selected: &[Rc<SummaryNode>],
@@ -1513,9 +1514,12 @@ fn has_unsafe_raw_entity_leaf(
                         _
                     )
                 );
+                let scalar_series_input = matches!(&node.expr,
+                    SummaryExpr::SummaryAgg { input, .. }
+                    if input.item.is_none() && matches!(&input.weight, planner_types::post_asap::SummaryInputExpr::Column(planner_types::pre_asap::ColumnRef::SampleValue)));
                 return matches!(reduction, Reduction::PerEntity)
                     && !pooling
-                    && !preserves_series_state;
+                    && !(preserves_series_state || scalar_series_input);
             }
             let additive_reduction = matches!(reduction, Reduction::Reduce(_))
                 && matches!(family, SummaryFamilyType::ExactAggregate(ExactKind::Sum, _))
@@ -1871,10 +1875,8 @@ impl PhysicalCompiler {
                     aggregation_id.clone(),
                     environment.target,
                 );
-                let precompute_materialization = aggregation_config_for_materialization(
-                    &aggregation,
-                    asap_types::QueryLanguage::PromQl,
-                )?;
+                let precompute_materialization =
+                    scoped_materialization(&aggregation, &selected.node)?;
                 let materialization = precompute_materialization.policy_fingerprint();
                 let state_consumers = consumers[&materialization]
                     .iter()
@@ -1905,10 +1907,8 @@ impl PhysicalCompiler {
                 // Preserve semantic window and evaluation cadence independently
                 // from the selected storage representation.
                 aggregation.window_secs = window_implementation.window_secs;
-                let mut runtime_materialization = aggregation_config_for_materialization(
-                    &aggregation,
-                    asap_types::QueryLanguage::PromQl,
-                )?;
+                let mut runtime_materialization =
+                    scoped_materialization(&aggregation, &selected.node)?;
                 runtime_materialization.window_size = window_implementation.window_secs;
                 runtime_materialization.slide_interval = window_implementation.slide_secs;
                 runtime_materialization.window_type =
@@ -2162,13 +2162,7 @@ impl PhysicalCompiler {
                         .materializations
                         .iter()
                         .find(|candidate| candidate.policy_fingerprint() == fingerprint)
-                        .map(|candidate| match &candidate.window_layout {
-                            asap_types::WindowMaterializationLayout::FullWindow => {
-                                candidate.window_size
-                            }
-                            layout => layout.base_pane_secs(),
-                        }
-                        .saturating_mul(1_000))
+                        .map(asap_types::PrecomputeMaterialization::stored_window_ms)
                         .ok_or_else(|| {
                             crate::query_plan::QueryPlanError::Invalid(format!(
                                 "compiled binding {} has no precompute materialization",
@@ -2836,10 +2830,12 @@ fn retained_partition_count(
     // Reset-aware and min/max state remains source-series scoped even with an
     // empty output grouping. Grouped states have at most one partition per
     // input series. Other empty groupings are the Reduce([]) global singleton.
-    if matches!(
-        materialization.aggregation_type,
-        A::Increase | A::MultipleIncrease | A::MinMax | A::MultipleMinMax
-    ) || !materialization.grouping_labels.labels.is_empty()
+    if materialization.partitioning == Some(asap_types::sds::PopulationPartitioning::PerEntity)
+        || matches!(
+            materialization.aggregation_type,
+            A::Increase | A::MultipleIncrease | A::MinMax | A::MultipleMinMax
+        )
+        || !materialization.grouping_labels.labels.is_empty()
     {
         u128::from(input_cardinality.unwrap_or(1).max(1))
     } else {
@@ -3190,6 +3186,29 @@ fn physical_aggregation(
 /// content-addressed materialization contract. This is the one conversion
 /// shared by the physical compiler and the compatibility replanner; it does
 /// not create a second registry or wire plan.
+fn scoped_materialization(
+    aggregation: &BackendAggregation,
+    node: &SummaryNode,
+) -> anyhow::Result<asap_types::PrecomputeMaterialization> {
+    let mut config =
+        aggregation_config_for_materialization(aggregation, asap_types::QueryLanguage::PromQl)?;
+    if !matches!(aggregation.aggregation_input, AggregationInput::Raw) {
+        return Ok(config);
+    }
+    let SummaryExpr::SummaryAgg { reduction, .. } = &node.expr else {
+        anyhow::bail!("materialization lacks SummaryAgg partition contract");
+    };
+    config.partitioning = Some(match reduction {
+        planner_types::pre_asap::Reduction::PerEntity => {
+            asap_types::sds::PopulationPartitioning::PerEntity
+        }
+        planner_types::pre_asap::Reduction::Reduce(_) => {
+            asap_types::sds::PopulationPartitioning::Grouped
+        }
+    });
+    Ok(config)
+}
+
 pub(crate) fn aggregation_config_for_materialization(
     aggregation: &BackendAggregation,
     language: asap_types::QueryLanguage,
@@ -3241,9 +3260,9 @@ fn materialization_consumers(
             {
                 continue;
             }
-            let config = aggregation_config_for_materialization(
+            let config = scoped_materialization(
                 &physical_aggregation(query, &state, query.query_id.clone(), target),
-                asap_types::QueryLanguage::PromQl,
+                &state.node,
             )?;
             consumers
                 .entry(config.policy_fingerprint())
@@ -3306,8 +3325,9 @@ fn shared_pane_origin_ms(
 /// serialized QueryPlan retains the merge edges. Unsupported operators are
 /// intentionally not traversed: QueryPlan lowers them to an explicit exact
 /// fallback node and no unused warm state is provisioned.
-/// Raw accumulators do not retain arbitrary source labels. Preserve native semantics
-/// unless the selected DAG explicitly authorizes pooling the source entities.
+/// Admit per-entity raw state only when the installed partition contract and
+/// scalar input evaluator preserve source rows. Composite updates still require
+/// an executable maintenance evaluator; a partition flag cannot authorize them.
 fn collect_selected_materializations(
     node: &Rc<SummaryNode>,
     composable: bool,
@@ -3590,7 +3610,51 @@ mod tests {
     use super::*;
 
     #[test]
-    fn raw_per_entity_state_requires_explicit_additive_reduction() {
+    fn installed_partition_must_match_the_bound_dag_reduction() {
+        let mut env = environment(10_000);
+        env.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite;
+        env.collector_ids.clear();
+        let mut plan = PhysicalCompiler
+            .compile(request("scope", "sum_over_time(m[1m])"), env)
+            .unwrap();
+        let installed = plan
+            .precompute_plan
+            .executable_dags
+            .values_mut()
+            .next()
+            .unwrap();
+        let mut dag = installed.document.decode().unwrap();
+        let node = dag
+            .nodes
+            .iter_mut()
+            .find(|node| {
+                matches!(
+                    node.payload,
+                    planner_types::post_asap::ExecutableOperatorPayload::SummaryAgg { .. }
+                )
+            })
+            .unwrap();
+        if let planner_types::post_asap::ExecutableOperatorPayload::SummaryAgg {
+            reduction, ..
+        } = &mut node.payload
+        {
+            *reduction = planner_types::pre_asap::Reduction::by(vec![]);
+        }
+        installed.document = asap_types::executable_plan::OwnedPostAsapDag::from_executable(
+            installed.document.query_id.clone(),
+            &dag,
+        )
+        .unwrap();
+        assert!(plan
+            .precompute_plan
+            .validate()
+            .unwrap_err()
+            .to_string()
+            .contains("partition"));
+    }
+
+    #[test]
+    fn raw_per_entity_state_carries_explicit_isolation() {
         for query in [
             "sum_over_time(m[1m])",
             "quantile_over_time(0.99, m[1m])",
@@ -3603,7 +3667,11 @@ mod tests {
                 .compile(request("per-entity", query), environment)
                 .unwrap();
             assert!(
-                plan.precompute_plan.materializations.is_empty(),
+                plan.precompute_plan
+                    .materializations
+                    .iter()
+                    .all(|state| state.partitioning
+                        == Some(asap_types::sds::PopulationPartitioning::PerEntity)),
                 "{query} pooled source entities"
             );
         }
@@ -4879,8 +4947,8 @@ mod tests {
     }
 
     #[test]
-    fn composable_per_entity_window_delegates_exact_subtree_to_prometheus() {
-        use crate::query_plan::{logical::LogicalOperator, QueryPlanNode};
+    fn composable_per_entity_window_installs_isolated_state() {
+        use crate::query_plan::QueryPlanNode;
         let mut snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!(
             "../../../docs/examples/asapquery-planning-snapshot.json"
         ))
@@ -4890,24 +4958,16 @@ mod tests {
         entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact);
         let (request, env) = snapshot.planning_request().unwrap();
         let plan = PhysicalCompiler.compile(request, env).unwrap();
-        assert!(plan.precompute_plan.materializations.is_empty());
+        assert_eq!(plan.precompute_plan.materializations.len(), 1);
+        assert_eq!(
+            plan.precompute_plan.materializations[0].partitioning,
+            Some(asap_types::sds::PopulationPartitioning::PerEntity)
+        );
         let entry = plan.query_plan.lookup("sum_over_time(m[1m])").unwrap();
-        assert!(entry.nodes.values().any(|node| matches!(
-            node,
-            QueryPlanNode::Logical {
-                operator: LogicalOperator::ExactSubquery { query },
-                ..
-            } if query == "sum_over_time(m[1m])"
-        )));
-        assert!(entry.nodes.values().all(|node| !matches!(
-            node,
-            QueryPlanNode::ReadMaterialization { .. }
-                | QueryPlanNode::ExactFallback { .. }
-                | QueryPlanNode::Logical {
-                    operator: LogicalOperator::Scan { .. },
-                    ..
-                }
-        )));
+        assert!(entry
+            .nodes
+            .values()
+            .any(|node| matches!(node, QueryPlanNode::ReadMaterialization { .. })));
     }
 
     // Each operand retains its source and semantic range; a smaller shared pane
@@ -5592,11 +5652,13 @@ mod tests {
             .compile()
             .expect("unquoted v1 compatibility startup remains available");
         let (local, env) = snapshot.clone().planning_request().unwrap();
-        assert!(PhysicalCompiler
-            .compile(local, env)
-            .unwrap_err()
-            .to_string()
-            .contains("native residual substitution requires an exact selected value"));
+        let isolated = PhysicalCompiler.compile(local, env).unwrap();
+        assert!(!isolated.precompute_plan.materializations.is_empty());
+        assert!(isolated
+            .precompute_plan
+            .materializations
+            .iter()
+            .all(|state| state.partitioning.is_some()));
         let (request, environment) = snapshot.planning_request().unwrap();
         let native = crate::physical::workload_cost::with_exact_alternative(request)
             .unwrap()
@@ -5626,11 +5688,13 @@ mod tests {
             .compile()
             .expect("unquoted v1 compatibility startup remains available");
         let (local, env) = snapshot.clone().planning_request().unwrap();
-        assert!(PhysicalCompiler
-            .compile(local, env)
-            .unwrap_err()
-            .to_string()
-            .contains("native residual substitution requires an exact selected value"));
+        let isolated = PhysicalCompiler.compile(local, env).unwrap();
+        assert!(!isolated.precompute_plan.materializations.is_empty());
+        assert!(isolated
+            .precompute_plan
+            .materializations
+            .iter()
+            .all(|state| state.partitioning.is_some()));
         let (request, environment) = snapshot.planning_request().unwrap();
         let native = crate::physical::workload_cost::with_exact_alternative(request)
             .unwrap()
diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs
index e2411a565..c2de97497 100644
--- a/crates/asap_types/src/aggregation_config.rs
+++ b/crates/asap_types/src/aggregation_config.rs
@@ -177,6 +177,15 @@ impl AggregationIdInfo {
 pub type AggregationConfig = PrecomputeMaterialization;
 
 impl PrecomputeMaterialization {
+    /// Temporal extent of one stored base state, independent of emission cadence.
+    pub fn stored_window_ms(&self) -> u64 {
+        match &self.window_layout {
+            WindowMaterializationLayout::FullWindow => self.window_size,
+            layout => layout.base_pane_secs(),
+        }
+        .saturating_mul(1_000)
+    }
+
     pub fn population_filter_canonical(&self) -> Result<String, String> {
         if let Some(column) = &self.table_timestamp_column {
             if self.table_name.is_none() || column.is_empty() {
diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs
index 8cff067df..a4453ebc7 100644
--- a/crates/asap_types/src/precompute_plan.rs
+++ b/crates/asap_types/src/precompute_plan.rs
@@ -343,6 +343,49 @@ impl PrecomputePlan {
             installed
                 .validate()
                 .map_err(PrecomputePlanError::CatalogContract)?;
+            let dag = installed
+                .document
+                .decode()
+                .map_err(PrecomputePlanError::CatalogContract)?;
+            for node in &dag.nodes {
+                let Some(crate::executable_plan::BackendNodeBinding::Materialization {
+                    summary_definition,
+                }) = installed.binding.node(node.id)
+                else {
+                    continue;
+                };
+                let Some(config) = self
+                    .materializations
+                    .iter()
+                    .find(|config| config.policy_fingerprint() == summary_definition.fingerprint())
+                else {
+                    return Err(PrecomputePlanError::CatalogContract(
+                        "DAG materialization has no runtime configuration".into(),
+                    ));
+                };
+                if let Some(partitioning) = config.partitioning {
+                    if let planner_types::post_asap::ExecutableOperatorPayload::SummaryAgg {
+                        reduction,
+                        ..
+                    } = &node.payload
+                    {
+                        let expected = match reduction {
+                            planner_types::pre_asap::Reduction::PerEntity => {
+                                crate::sds::PopulationPartitioning::PerEntity
+                            }
+                            planner_types::pre_asap::Reduction::Reduce(_) => {
+                                crate::sds::PopulationPartitioning::Grouped
+                            }
+                        };
+                        if partitioning != expected {
+                            return Err(PrecomputePlanError::CatalogContract(
+                                "runtime population partition disagrees with Planner reduction"
+                                    .into(),
+                            ));
+                        }
+                    }
+                }
+            }
         }
         let mut materializations = BTreeSet::new();
         for materialization in &self.materializations {
diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
index f3da96a76..5004547db 100644
--- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs
+++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
@@ -560,13 +560,16 @@ fn route_messages(
             // accumulator per source series. Their emitted label values still
             // follow the physical grouping, so query-time Reduce nodes can
             // combine those independent SDS instances safely.
-            let series_scoped = matches!(
-                config.aggregation_type,
-                asap_types::AggregationType::Increase
-                    | asap_types::AggregationType::MultipleIncrease
-                    | asap_types::AggregationType::MinMax
-                    | asap_types::AggregationType::MultipleMinMax
-            );
+            let series_scoped = config.partitioning
+                == Some(asap_types::sds::PopulationPartitioning::PerEntity)
+                || (config.partitioning.is_none()
+                    && matches!(
+                        config.aggregation_type,
+                        asap_types::AggregationType::Increase
+                            | asap_types::AggregationType::MultipleIncrease
+                            | asap_types::AggregationType::MinMax
+                            | asap_types::AggregationType::MultipleMinMax
+                    ));
             let grouping_pairs: Vec<(&str, &str)> = if series_scoped {
                 Vec::new()
             } else {
@@ -924,10 +927,21 @@ mod tests {
             vec!["job".into()],
         );
         let counter = config(AggregationType::Increase, vec!["job".into()], vec![]);
+        let mut kll = config(AggregationType::DatasketchesKLL, vec![], vec![]);
+        kll.partitioning = Some(asap_types::sds::PopulationPartitioning::PerEntity);
+        let kll_fp = kll.policy_fingerprint();
+        let mut pooled_kll = kll.clone();
+        pooled_kll.partitioning = Some(asap_types::sds::PopulationPartitioning::Grouped);
+        let pooled_kll_fp = pooled_kll.policy_fingerprint();
+        assert_ne!(kll_fp, pooled_kll_fp);
         let cms_fp = cms.policy_fingerprint();
         let counter_fp = counter.policy_fingerprint();
-        let streaming =
-            StreamingConfig::new(HashMap::from([(cms_fp.0, cms), (counter_fp.0, counter)]));
+        let streaming = StreamingConfig::new(HashMap::from([
+            (cms_fp.0, cms),
+            (counter_fp.0, counter),
+            (kll_fp.0, kll),
+            (pooled_kll_fp.0, pooled_kll),
+        ]));
         let hot_reload = physical_config(streaming);
         let physical_plan = hot_reload.physical_plan_snapshot().unwrap();
         let (sender, _worker) = mpsc::channel(8);
@@ -977,6 +991,8 @@ mod tests {
         let mut cms_buckets = 0;
         let mut counter_buckets = 0;
         let mut cms_samples = 0;
+        let mut kll_buckets = 0;
+        let mut pooled_kll_buckets = 0;
         for message in messages {
             let WorkerMessage::GroupSamples {
                 policy_fp, samples, ..
@@ -989,8 +1005,17 @@ mod tests {
                 cms_samples += samples.len();
             } else if policy_fp == counter_fp {
                 counter_buckets += 1;
+            } else if policy_fp == kll_fp {
+                kll_buckets += 1;
+            } else if policy_fp == pooled_kll_fp {
+                pooled_kll_buckets += 1;
             }
         }
+        assert_eq!(kll_buckets, 10, "PerEntity KLL keeps every source series");
+        assert_eq!(
+            pooled_kll_buckets, 1,
+            "Grouped empty keys intentionally pool"
+        );
         assert_eq!(cms_buckets, 1, "Reduce([]) has one global CMS SID");
         assert_eq!(cms_samples, 10, "global CMS receives every source series");
         assert_eq!(
diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs
index 897c25a1a..6c14192f7 100644
--- a/data_plane/src/drivers/query/servers/http.rs
+++ b/data_plane/src/drivers/query/servers/http.rs
@@ -6072,7 +6072,7 @@ pub fn build_active_physical_plan(
                 .iter()
                 .find(|config| config.policy_fingerprint() == binding.materialization.fingerprint())
                 .ok_or_else(|| "query binding has no precompute definition".to_string())?;
-            if binding.window_ms != materialization.slide_interval.saturating_mul(1_000) {
+            if binding.window_ms != materialization.stored_window_ms() {
                 return Err(
                     "query physical pane duration differs from installed precompute definition"
                         .into(),
diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs
index e1c994a0d..502960ae4 100644
--- a/data_plane/tests/asapquery_compatibility_process_e2e.rs
+++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs
@@ -18,6 +18,9 @@ use prost::Message;
 use serde_json::Value;
 use tokio::sync::Mutex;
 
+#[path = "support/erp_planning_process.rs"]
+mod erp_planning_process;
+
 struct ChildGuard(Child);
 
 impl Drop for ChildGuard {
@@ -94,6 +97,19 @@ async fn remote_write(client: &reqwest::Client, base: &str, request: &WriteReque
         .as_u16()
 }
 
+async fn drain_precompute(client: &reqwest::Client, backend: &str) {
+    let response = client
+        .post(format!("{backend}/api/v1/precompute/drain"))
+        .send()
+        .await
+        .unwrap();
+    assert!(
+        response.status().is_success(),
+        "drain failed: {}",
+        response.text().await.unwrap()
+    );
+}
+
 fn first_value(response: &Value, field: &str) -> Option<f64> {
     let samples = response["data"]["result"]
         .as_array()?
@@ -564,6 +580,7 @@ async fn registered_temporal_topk(algorithm: planner_types::post_asap::SketchAlg
     };
     assert_eq!(remote_write(&client, &backend, &watermark).await, 204);
     assert_eq!(remote_write(&client, &backend, &samples).await, 204);
+    drain_precompute(&client, &backend).await;
     let timestamp = (base + 5000) as f64 / 1000.0;
     let instant = tokio::time::timeout(Duration::from_secs(30), async {
         loop {
@@ -869,6 +886,7 @@ async fn run_shared_dashboard(multi_pane: bool) {
         };
         assert_eq!(remote_write(&client, &backend, &close_third).await, 204);
     }
+    drain_precompute(&client, &backend).await;
     let evaluation = base + if multi_pane { 10000 } else { 5000 };
     for (query, expected) in [
         (
@@ -1232,6 +1250,7 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
     );
     // A normal Prometheus retry must be accepted without changing sketches.
     assert_eq!(remote_write(&client, &backend, &request).await, 204);
+    drain_precompute(&client, &backend).await;
     let corrupt = client
         .post(format!("{backend}/api/v1/write"))
         .header("content-encoding", "snappy")
@@ -1268,40 +1287,22 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
         assert_eq!(range["status"], "success", "{query}: {range}");
         assert!(is_warm(&range), "{query}: {range}");
     }
-    // The bare per-series quantile has no producer binding and must forward
-    // the complete request to the exact backend.
-    for query in ["quantile_over_time(0.5, asap_demo_latency_ms[5s])"] {
-        let instant: Value = client
-            .get(format!("{backend}/api/v1/query"))
-            .query(&[
-                ("query", query.to_string()),
-                ("time", first_eval.to_string()),
-            ])
-            .send()
-            .await
-            .unwrap()
-            .json()
-            .await
-            .unwrap();
-        assert_eq!(instant["data"]["result"][0]["metric"]["fallback"], "true");
-        assert!(!is_warm(&instant));
-        let range: Value = client
-            .get(format!("{backend}/api/v1/query_range"))
-            .query(&[
-                ("query", query.to_string()),
-                ("start", first_eval.to_string()),
-                ("end", second_eval.to_string()),
-                ("step", "5".into()),
-            ])
-            .send()
-            .await
-            .unwrap()
-            .json()
-            .await
-            .unwrap();
-        assert_eq!(range["data"]["result"][0]["metric"]["fallback"], "true");
-        assert!(!is_warm(&range));
-    }
+    // Per-series quantile now has a population-isolated producer.
+    let quantile_query = "quantile_over_time(0.5, asap_demo_latency_ms[5s])";
+    let quantile =
+        wait_for_warm_instant(&client, &backend, quantile_query, first_eval, &backend_log).await;
+    assert!(is_warm(&quantile));
+    let quantile_range = wait_for_warm_range(
+        &client,
+        &backend,
+        quantile_query,
+        first_eval,
+        second_eval,
+        5,
+        &backend_log,
+    )
+    .await;
+    assert!(is_warm(&quantile_range));
     let sum = wait_for_warm_instant(
         &client,
         &backend,
diff --git a/data_plane/tests/support/erp_planning_process.rs b/data_plane/tests/support/erp_planning_process.rs
new file mode 100644
index 000000000..aa0cf5083
--- /dev/null
+++ b/data_plane/tests/support/erp_planning_process.rs
@@ -0,0 +1,264 @@
+use super::*;
+use control_plane::physical::{compiler::BackendLocalPlanningSnapshot, erp::ErpShapeObserver};
+
+fn measured_profiles(raw: &[f64]) -> Value {
+    let mut records = Vec::new();
+    for k in [32, 128] {
+        let mut error = 0.0f64;
+        let mut bytes = 0usize;
+        for seed in 0..10 {
+            let mut sketch = asap_sketchlib::KllSketch::with_seed(k, seed);
+            for value in raw {
+                sketch.update(*value);
+            }
+            bytes = bytes.max(sketch.sketch_bytes().len());
+            for q in 1..100 {
+                let q = q as f64 / 100.0;
+                let estimate = sketch.quantile(q);
+                let lower = raw.iter().filter(|v| **v < estimate).count() as f64 / raw.len() as f64;
+                let upper =
+                    raw.iter().filter(|v| **v <= estimate).count() as f64 / raw.len() as f64;
+                error = error.max((lower - q).max(q - upper).max(0.0));
+            }
+        }
+        records.push(serde_json::json!({
+            "id": format!("process-test-k{k}"), "sketch": "kll-percall", "implementation": "lib",
+            "parameters": {"k": k}, "trials": 10,
+            "distribution": {"erp_shape": {"family": "zipf", "parameters": {"exponent": 1.0},
+                "cardinality": 16, "benchmark_events": raw.len()}},
+            "error_metrics": {"max_rank_err": error},
+            "resources": {"memory_bytes": bytes, "update_cpu_seconds": 0.0,
+                "query_cpu_seconds": 0.0, "merge_cpu_seconds": 0.0}
+        }));
+    }
+    // This correctness fixture measures error and retained serialized state.
+    // CPU is not the objective or a performance claim in this process test.
+    serde_json::json!({"schema_version": 1,
+        "producer_version": "process-test-measured-error-and-serialized-state-cpu-not-measured",
+        "records": records})
+}
+
+#[tokio::test]
+async fn observed_shape_selects_installed_parameters_and_executes_remote_write() {
+    let fallback_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let fallback_url = format!("http://{}", fallback_listener.local_addr().unwrap());
+    let fallback_task = tokio::spawn(async move {
+        axum::serve(
+            fallback_listener,
+            Router::new().route("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/-/healthy", get(|| async { "healthy" })),
+        )
+        .await
+        .unwrap();
+    });
+    const QUERY: &str = "quantile_over_time(0.9, erp_latency[5s])";
+    let training: Vec<f64> = (1..=16)
+        .flat_map(|value| std::iter::repeat_n(value as f64, 512 / value))
+        .collect();
+    let raw: Vec<f64> = (1..=16)
+        .rev()
+        .flat_map(|value| std::iter::repeat_n(value as f64, 256 / value))
+        .collect();
+    let mut observer = ErpShapeObserver::new(16).unwrap();
+    for (index, value) in raw.iter().enumerate() {
+        observer.observe(&value.to_string(), index / 100).unwrap();
+    }
+    let observation = observer.snapshot().unwrap();
+    let artifact = measured_profiles(&training);
+    let mut chosen = Vec::new();
+    for only_large in [false, true] {
+        let mut evidence = artifact.clone();
+        if only_large {
+            evidence["records"]
+                .as_array_mut()
+                .unwrap()
+                .retain(|row| row["parameters"]["k"] == 128);
+        }
+        let mut fixture: Value = serde_json::from_str(include_str!(
+            "../../../docs/examples/asapquery-compatibility-demo-snapshot.json"
+        ))
+        .unwrap();
+        let mut entry = fixture["query_workload"]["repeating_queries"][3].clone();
+        entry["query"] = QUERY.into();
+        entry["requirements"]["accuracy"] = serde_json::json!({"explicit": {"Epsilon": 0.2}});
+        fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]);
+        fixture["implementation"]["erp"] = serde_json::json!({
+            "distribution": {"workload": {"external": {"dataset": "held-out-process-stream"}}},
+            "artifact": evidence, "implementation": "lib", "error_metric": "max_rank_err",
+            "min_trials": 10, "expected_updates": raw.len(), "expected_queries": 10.0,
+            "expected_merges": 0.0, "retention_seconds": 60.0, "cpu_weight": 0.0,
+            "byte_second_weight": 1e-9, "mode": "hybrid", "observed_shape": observation.observation,
+            "shape_match": {"minimum_benchmark_events": 1000, "max_log2_cardinality_distance": 0.0,
+                "max_parameter_distance": 0.1, "max_goodness_of_fit": 0.1,
+                "minimum_confidence": 0.8, "minimum_confidence_margin": 0.05},
+            "runtime": {"allowed_algorithms": ["Kll"], "max_memory_bytes": null}
+        });
+        let policy: control_plane::physical::erp::ErpPlanningInput =
+            serde_json::from_value(fixture["implementation"]["erp"].clone()).unwrap();
+        assert!(matches!(
+            policy.select(
+                planner_types::post_asap::SketchAlgorithm::Kll,
+                0.2,
+                planner_types::post_asap::SketchParams::Kll { k: 128 }
+            ),
+            control_plane::physical::erp::ErpParameterDecision::Empirical { .. }
+        ));
+        let snapshot: BackendLocalPlanningSnapshot =
+            serde_json::from_value(fixture.clone()).unwrap();
+        let plan = snapshot.compile().unwrap();
+        assert_eq!(
+            plan.precompute_plan.materializations.len(),
+            1,
+            "plan={plan:#?}; observation={observation:#?}; evidence={artifact}"
+        );
+        let expected_k = if only_large { 128 } else { 32 };
+        assert_eq!(
+            plan.precompute_plan.materializations[0].parameters["k"],
+            expected_k
+        );
+        chosen.push(plan.precompute_plan.materializations[0].policy_fingerprint());
+        eprintln!(
+            "ERP_PLANNED {}",
+            serde_json::json!({
+                "query": QUERY, "available_profiles": policy.artifact.records,
+            "parameter_decision": format!("{:?}", policy.select(planner_types::post_asap::SketchAlgorithm::Kll, 0.2, planner_types::post_asap::SketchParams::Kll { k: 128 })),
+            "lifecycle_estimates": plan.lifecycle_estimates,
+                "observation": policy.observed_shape,
+                "selected_parameters": plan.precompute_plan.materializations[0].parameters,
+                "materialization": chosen.last(),
+                "partitioning": plan.precompute_plan.materializations[0].partitioning,
+                "query_plan": plan.query_plan,
+            })
+        );
+        let output = tempfile::tempdir().unwrap();
+        let path = output.path().join("planning.json");
+        std::fs::write(&path, serde_json::to_vec(&fixture).unwrap()).unwrap();
+        let port = unused_port();
+        let mut child = ChildGuard(
+            Command::new(env!("CARGO_BIN_EXE_data_plane"))
+                .args([
+                    "--forward-unsupported-queries",
+                    "--prometheus-server",
+                    &fallback_url,
+                    "--profile",
+                    "asapquery",
+                    "--planning-snapshot",
+                ])
+                .arg(&path)
+                .args(["--http-port", &port.to_string(), "--output-dir"])
+                .arg(output.path())
+                .args([
+                    "--precompute-allowed-lateness-ms",
+                    "0",
+                    "--precompute-flush-interval-ms",
+                    "25",
+                ])
+                .stdout(Stdio::null())
+                .stderr(Stdio::inherit())
+                .spawn()
+                .unwrap(),
+        );
+        let client = reqwest::Client::new();
+        let backend = format!("http://127.0.0.1:{port}");
+        wait_until_ready(&client, &format!("{backend}/api/v1/health"), &mut child.0).await;
+        let config: Value = client
+            .get(format!("{backend}/api/v1/physical-plan/status"))
+            .send()
+            .await
+            .unwrap()
+            .json()
+            .await
+            .unwrap();
+        let config_text = serde_json::to_string(&config).unwrap();
+        assert!(
+            config_text.contains(&chosen.last().unwrap().0.to_string()),
+            "installed ERP identity missing: {config}"
+        );
+        let now = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_millis() as i64;
+        let base = now - now.rem_euclid(5000) - 20000;
+        let samples: Vec<_> = raw
+            .iter()
+            .enumerate()
+            .map(|(i, value)| (base + 1 + i as i64, *value))
+            .collect();
+        assert_eq!(
+            remote_write(
+                &client,
+                &backend,
+                &WriteRequest {
+                    timeseries: vec![
+                        series_with_labels("erp_latency", &[("instance", "a")], &samples),
+                        series_with_labels(
+                            "erp_latency",
+                            &[("instance", "b")],
+                            &samples
+                                .iter()
+                                .map(|(t, v)| (*t, *v + 1000.0))
+                                .collect::<Vec<_>>()
+                        ),
+                    ]
+                }
+            )
+            .await,
+            204
+        );
+        assert_eq!(
+            remote_write(
+                &client,
+                &backend,
+                &WriteRequest {
+                    timeseries: vec![
+                        series_with_labels(
+                            "erp_latency",
+                            &[("instance", "a")],
+                            &[(base + 15001, 1.0)]
+                        ),
+                        series_with_labels(
+                            "erp_latency",
+                            &[("instance", "b")],
+                            &[(base + 15001, 1001.0)]
+                        ),
+                    ]
+                }
+            )
+            .await,
+            204
+        );
+        drain_precompute(&client, &backend).await;
+        let result = wait_for_warm_instant(
+            &client,
+            &backend,
+            QUERY,
+            (base + 5000) as f64 / 1000.0,
+            &output.path().join("query_engine.log"),
+        )
+        .await;
+        let rows = result["data"]["result"].as_array().unwrap();
+        assert_eq!(
+            rows.len(),
+            2,
+            "per-series KLL states must not pool: {result}"
+        );
+        eprintln!(
+            "ERP_WARM {}",
+            serde_json::json!({"materialization": chosen.last(), "result": result})
+        );
+        for (instance, offset) in [("a", 0.0), ("b", 1000.0)] {
+            let row = rows
+                .iter()
+                .find(|row| row["metric"]["instance"] == instance)
+                .expect("source labels retained");
+            let estimate = row["value"][1].as_str().unwrap().parse::<f64>().unwrap() - offset;
+            let lower = raw.iter().filter(|v| **v < estimate).count() as f64 / raw.len() as f64;
+            let upper = raw.iter().filter(|v| **v <= estimate).count() as f64 / raw.len() as f64;
+            assert!((lower - 0.9).max(0.9 - upper).max(0.0) <= 0.2, "{result}");
+        }
+    }
+    fallback_task.abort();
+    assert_ne!(
+        chosen[0], chosen[1],
+        "changed evidence must change installed state identity"
+    );
+}
diff --git a/docs/design_docs/shape-aware-erp-v1.md b/docs/design_docs/shape-aware-erp-v1.md
index 04a8951b0..5d04d99af 100644
--- a/docs/design_docs/shape-aware-erp-v1.md
+++ b/docs/design_docs/shape-aware-erp-v1.md
@@ -57,3 +57,21 @@ numeric-distribution observation. Equal frequencies produce the canonical
 uniform fit only: Zipf exponent zero describes the same distribution and must
 not create a false ambiguity. Near-uniform, genuinely distinct fits still pass
 through the normal ambiguity policy.
+
+### Population isolation in backend-local execution
+
+A temporal scalar summary has one state per source series when its selected
+`SummaryAgg` uses `Reduction::PerEntity`. An explicit reduction with no grouping
+keys has one pooled population. These are different materializations even when
+source, sketch parameters, and the visible grouping-key list are identical.
+
+The compiler records shared `PopulationPartitioning` metadata in the runtime
+configuration and DataDescriptor. Both identities include the partitioning;
+installation checks it against the bound Planner DAG. Raw ingestion uses the
+full source labels for per-entity routing and the configured grouping for pooled
+routing. Memory estimates count per-entity states against source cardinality.
+Legacy configurations without this metadata retain their existing routing rules.
+
+This is a source-isolation contract, not permission to skip a maintenance update
+expression. Only already-supported scalar update expressions pass the compiler's
+per-entity admission check; other subDAG updates still require a real evaluator.
diff --git a/docs/developer_docs/erp-process-validation.md b/docs/developer_docs/erp-process-validation.md
new file mode 100644
index 000000000..9950e1ea5
--- /dev/null
+++ b/docs/developer_docs/erp-process-validation.md
@@ -0,0 +1,49 @@
+# ERP-selected KLL process validation
+
+The process test in [erp_planning_process.rs](../../data_plane/tests/support/erp_planning_process.rs)
+checks observation → Planner selection → physical installation → raw ingestion →
+ASAP query execution. This is a correctness fixture, not an o11ybench performance
+result or a claim that empirical rank error is a probabilistic bound.
+
+The fixture measures the runtime KLL implementation at `k=32` and `k=128` using
+10 seeds and 99 quantiles. It records maximum observed tie-aware rank error and
+serialized retained state size. The observer fits a held-out stream; Planner
+selects parameters from the available measured profiles. The first run offers
+both profiles; the second offers only the larger profile. No preselected physical
+plan is supplied to the data-plane process.
+
+Both processes compile the startup planning snapshot and accept remote-write
+samples for two source series. Their value domains differ by 1000, making an
+accidental pooled result detectable. A health-only fallback server has no query
+handler. The test requires two labeled warm ASAP results, each within the declared
+rank-error target of 0.2.
+
+One verified run produced:
+
+| Available profiles | Selected k | Materialization ID | Returned values (a, b) |
+| --- | ---: | ---: | --- |
+| 32, 128 | 32 | 17085259479989488410 | 12, 1012 |
+| 128 | 128 | 81115169662305743 | 12, 1012 |
+
+The measured fixture errors were 0.03336 and 0.00496, respectively. The objective
+uses serialized-state byte-seconds; CPU terms are disabled because this fixture
+does not benchmark CPU. These measurements are test evidence, not a production
+ERP artifact distributed for unrelated workloads. The observer describes ranked
+key frequencies, not arbitrary numeric value spacing.
+
+Run:
+
+```sh
+cargo test -p data_plane --test asapquery_compatibility_process_e2e erp_planning_process -- --nocapture
+```
+
+`ERP_PLANNED` records candidate profiles, fitted observation, selected parameters,
+estimated costs, installed identity, and QueryPlan. `ERP_WARM` records the actual
+HTTP result and execution path. The evaluation workspace retains the captured
+run at `/mydata/erp-production-study/process-evidence.json`; this host-local path
+is not a repository fixture.
+
+Separate compiler and adapter tests cover ERP miss → theoretical sizing → exact
+fallback. This process test establishes two ERP hits; it does not claim a live
+online feedback service, general sketch-family accuracy calibration, or latency,
+CPU, and memory improvements over Prometheus.