diff --git a/crates/asap-aware-mapping/src/erp.rs b/crates/asap-aware-mapping/src/erp.rs
index ab45601f..b5ce518d 100644
--- a/crates/asap-aware-mapping/src/erp.rs
+++ b/crates/asap-aware-mapping/src/erp.rs
@@ -76,6 +76,53 @@ pub struct ErpSelection<'a> {
     pub accuracy_mode: AccuracyMode,
 }
 
+/// Runtime-observable input shape used to match a benchmark scenario. Input
+/// volume is a sufficiency gate, not a distance axis: once the benchmark has
+/// enough samples, repeating the same stationary distribution adds little
+/// information about sketch error.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct ErpDataShape {
+    pub cardinality: u64,
+    /// Stable distribution family, for example `uniform`, `zipf`,
+    /// `power_law`, or `empirical`. Different families are never interpolated.
+    pub family: String,
+    /// Family-specific numeric parameters. Zipf uses `exponent`; a continuous
+    /// power law may use `alpha` and `minimum`. Uniform has no parameters.
+    #[serde(default)]
+    pub parameters: BTreeMap<String, f64>,
+    pub benchmark_events: u64,
+}
+
+#[derive(Debug, Clone)]
+pub struct ErpNearestSelectionRequest {
+    pub selection: ErpSelectionRequest,
+    pub observed: ErpDataShape,
+    pub minimum_benchmark_events: u64,
+    pub max_log2_cardinality_distance: f64,
+    /// Maximum normalized distance for every common distribution parameter.
+    pub max_parameter_distance: f64,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct ErpWindowWorkload {
+    pub input_updates: u64,
+    pub query_executions: u64,
+    pub panes_per_query: u64,
+    pub retained_panes: u64,
+    pub materializations: u64,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct ErpOperationCounts {
+    pub updates: u64,
+    pub merges: u64,
+    pub queries: u64,
+    pub retained_sketches: u64,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
 pub enum ErpError {
     #[error("unsupported ERP schema version {0}")]
@@ -155,6 +202,132 @@ impl ErpArtifact {
             })
             .ok_or(ErpError::NoApplicableConfiguration)
     }
+
+    /// Select against the nearest compatible measured shape. Shape metadata is
+    /// read from `distribution.erp_shape`, keeping ERP v1 wire compatibility.
+    pub fn select_nearest(
+        &self,
+        request: &ErpNearestSelectionRequest,
+    ) -> Result<ErpSelection<'_>, ErpError> {
+        self.validate()?;
+        if !request.selection.valid() || !request.valid() {
+            return Err(ErpError::Invalid("invalid nearest-shape request"));
+        }
+        self.records
+            .iter()
+            .filter_map(|row| Some((row, data_shape(&row.distribution)?)))
+            .filter(|(_, shape)| shape.benchmark_events >= request.minimum_benchmark_events)
+            .filter_map(|(row, shape)| Some((row, request.distance(shape)?)))
+            .filter(|(_, distance)| *distance <= 1.0)
+            .filter(|(row, _)| {
+                request
+                    .selection
+                    .implementation
+                    .as_ref()
+                    .is_none_or(|wanted| &row.implementation == wanted)
+                    && (request.selection.allowed_sketches.is_empty()
+                        || request
+                            .selection
+                            .allowed_sketches
+                            .iter()
+                            .any(|name| name == &row.sketch))
+                    && row.trials >= request.selection.min_trials
+            })
+            .filter_map(|(row, distance)| {
+                let observed_error = *row.error_metrics.get(&request.selection.error_metric)?;
+                (observed_error <= request.selection.max_error).then_some((
+                    distance,
+                    ErpSelection {
+                        record: row,
+                        observed_error,
+                        estimated_cost: request.selection.cost(&row.resources),
+                        accuracy_mode: request.selection.mode,
+                    },
+                ))
+            })
+            .min_by(|(left_distance, left), (right_distance, right)| {
+                left_distance
+                    .total_cmp(right_distance)
+                    .then_with(|| left.estimated_cost.total_cmp(&right.estimated_cost))
+                    .then_with(|| left.record.id.cmp(&right.record.id))
+            })
+            .map(|(_, selection)| selection)
+            .ok_or(ErpError::NoApplicableConfiguration)
+    }
+}
+
+fn data_shape(distribution: &serde_json::Value) -> Option<ErpDataShape> {
+    serde_json::from_value(distribution.get("erp_shape")?.clone()).ok()
+}
+
+impl ErpNearestSelectionRequest {
+    fn valid(&self) -> bool {
+        self.minimum_benchmark_events > 0
+            && self.observed.cardinality > 0
+            && self.max_log2_cardinality_distance.is_finite()
+            && self.max_log2_cardinality_distance > 0.0
+            && !self.observed.family.trim().is_empty()
+            && self
+                .observed
+                .parameters
+                .values()
+                .all(|value| value.is_finite())
+            && self.max_parameter_distance.is_finite()
+            && self.max_parameter_distance > 0.0
+    }
+
+    fn distance(&self, candidate: ErpDataShape) -> Option<f64> {
+        if candidate.cardinality == 0
+            || candidate.family != self.observed.family
+            || candidate
+                .parameters
+                .keys()
+                .ne(self.observed.parameters.keys())
+        {
+            return None;
+        }
+        let cardinality = ((candidate.cardinality as f64).log2()
+            - (self.observed.cardinality as f64).log2())
+        .abs()
+            / self.max_log2_cardinality_distance;
+        let parameters = candidate
+            .parameters
+            .iter()
+            .map(|(name, value)| {
+                let observed = self.observed.parameters.get(name)?;
+                (value.is_finite() && observed.is_finite())
+                    .then_some((value - observed).abs() / self.max_parameter_distance)
+            })
+            .collect::<Option<Vec<_>>>()?
+            .into_iter()
+            .fold(0.0_f64, f64::max);
+        Some(cardinality.max(parameters))
+    }
+}
+
+impl ErpOperationCounts {
+    /// Compose atomic benchmark costs with a pane-based window plan. A query
+    /// over one pane needs no merge; N panes need N-1 merges.
+    pub fn from_window(workload: ErpWindowWorkload) -> Self {
+        Self {
+            updates: workload
+                .input_updates
+                .saturating_mul(workload.materializations),
+            merges: workload
+                .query_executions
+                .saturating_mul(workload.panes_per_query.saturating_sub(1)),
+            queries: workload.query_executions,
+            retained_sketches: workload
+                .retained_panes
+                .saturating_mul(workload.materializations),
+        }
+    }
+
+    pub fn cpu_seconds(self, resources: &ErpResourceProfile) -> f64 {
+        self.updates as f64 * resources.update_cpu_seconds
+            + self.merges as f64 * resources.merge_cpu_seconds
+            + self.queries as f64 * resources.query_cpu_seconds
+    }
 }
 
 impl ErpResourceProfile {
@@ -295,4 +468,81 @@ mod tests {
         assert_eq!(selected.record.id, "erp-0");
         assert_eq!(selected.accuracy_mode, AccuracyMode::Hybrid);
     }
+
+    #[test]
+    fn nearest_shape_prefers_cardinality_and_skew_then_cost() {
+        let mut close = row("close", 512, 0.009, 12_288.0);
+        close.distribution = serde_json::json!({"erp_shape": {
+            "cardinality": 1000, "family": "zipf", "parameters": {"exponent": 1.2}, "benchmark_events": 100000
+        }});
+        let mut cheap_but_far = row("far", 256, 0.009, 1.0);
+        cheap_but_far.distribution = serde_json::json!({"erp_shape": {
+            "cardinality": 8000, "family": "zipf", "parameters": {"exponent": 1.2}, "benchmark_events": 100000
+        }});
+        let artifact = ErpArtifact {
+            schema_version: ERP_SCHEMA_VERSION,
+            producer_version: "bench-1".into(),
+            records: vec![cheap_but_far, close],
+        };
+        let selected = artifact
+            .select_nearest(&ErpNearestSelectionRequest {
+                selection: request(),
+                observed: ErpDataShape {
+                    cardinality: 1200,
+                    family: "zipf".into(),
+                    parameters: BTreeMap::from([("exponent".into(), 1.1)]),
+                    benchmark_events: 0,
+                },
+                minimum_benchmark_events: 10_000,
+                max_log2_cardinality_distance: 4.0,
+                max_parameter_distance: 0.5,
+            })
+            .unwrap();
+        assert_eq!(selected.record.id, "close");
+    }
+
+    #[test]
+    fn nearest_shape_rejects_distribution_family_and_small_benchmarks() {
+        let mut row = row("uniform", 512, 0.009, 12_288.0);
+        row.distribution = serde_json::json!({"erp_shape": {
+            "cardinality": 1000, "family": "uniform", "parameters": {}, "benchmark_events": 999
+        }});
+        let artifact = ErpArtifact {
+            schema_version: ERP_SCHEMA_VERSION,
+            producer_version: "bench-1".into(),
+            records: vec![row],
+        };
+        let nearest = ErpNearestSelectionRequest {
+            selection: request(),
+            observed: ErpDataShape {
+                cardinality: 1000,
+                family: "zipf".into(),
+                parameters: BTreeMap::from([("exponent".into(), 1.0)]),
+                benchmark_events: 0,
+            },
+            minimum_benchmark_events: 1_000,
+            max_log2_cardinality_distance: 1.0,
+            max_parameter_distance: 0.5,
+        };
+        assert_eq!(
+            artifact.select_nearest(&nearest),
+            Err(ErpError::NoApplicableConfiguration)
+        );
+    }
+
+    #[test]
+    fn pane_window_composes_atomic_operation_costs() {
+        let counts = ErpOperationCounts::from_window(ErpWindowWorkload {
+            input_updates: 1_000,
+            query_executions: 10,
+            panes_per_query: 12,
+            retained_panes: 24,
+            materializations: 2,
+        });
+        assert_eq!(counts.updates, 2_000);
+        assert_eq!(counts.merges, 110);
+        assert_eq!(counts.queries, 10);
+        assert_eq!(counts.retained_sketches, 48);
+        assert!((counts.cpu_seconds(&row("cost", 1, 0.0, 0.0).resources) - 0.00131).abs() < 1e-12);
+    }
 }
diff --git a/docs/design_docs/error-resource-profile.md b/docs/design_docs/error-resource-profile.md
index 6627e9af..14f63634 100644
--- a/docs/design_docs/error-resource-profile.md
+++ b/docs/design_docs/error-resource-profile.md
@@ -2,7 +2,7 @@
 
 ## Status
 
-ERP v1 is a discrete, distribution-conditioned profile exchanged between
+ERP v1 is a discrete, shape- and distribution-conditioned profile exchanged between
 `sketch-bench` and ASAPPlanner. It is an empirical planning input, not a proof
 of a worst-case sketch guarantee. The implementation lives in
 `asap-aware-mapping::erp`; `approxbench erp` exports the producer artifact.
@@ -33,7 +33,8 @@ sketch-bench parameter/distribution sweep
     -> flat MergedRecord rows
     -> `approxbench erp`
     -> versioned ERP artifact
-    -> deployment supplies an ErpSelectionRequest
+    -> backend observes live cardinality/distribution shape
+    -> deployment supplies an exact or nearest ErpSelectionRequest
     -> ASAPPlanner filters applicable and accurate records
     -> least estimated workload cost
     -> deployment maps the selected identity to runtime configuration
@@ -56,10 +57,25 @@ Each record contains:
 - named observed error metrics; and
 - memory plus per-operation update, merge, and query CPU.
 
-The distribution is an opaque wire value to ASAPPlanner and is equality-matched
-in v1. This deliberately prevents accidental interpolation. A future schema may
-add a typed distribution signature (cardinality, entropy, skew, moments, tail
-mass, quantile density) and a conservative distance model.
+The canonical workload remains available as an opaque wire value for exact
+matching. Shape-aware records additionally carry:
+
+```text
+erp_shape = {
+  cardinality,
+  family,              // uniform | zipf | power_law | normal | empirical | ...
+  parameters,          // family-specific numeric parameter map
+  benchmark_events
+}
+```
+
+The family is explicit rather than encoding uniform as a missing Zipf
+parameter. Profiles from different families are never interpolated. Parameter
+keys must match before a distance is computed; this keeps the contract open to
+Zipf exponent, continuous power-law alpha/minimum, normal mean/deviation, and
+future synthetic or fitted families. Empirical/custom traces carry a stable
+family and descriptor and normally use exact matching rather than synthetic
+interpolation. `benchmark_events` is a sufficiency gate, not a distance axis.
 
 ## Selection
 
@@ -83,6 +99,12 @@ cpu_weight * (
 + byte_second_weight * retention_seconds * memory_bytes
 ```
 
+For shape-aware selection, records must first satisfy the benchmark-event floor
+and distribution-family constraint. Their normalized distance is the maximum
+of log2-cardinality distance and every family-specific parameter distance. Only
+candidates within all caller-supplied bounds are eligible; cost selects among
+those candidates.
+
 The least-cost accepted record wins. Missing error metrics, missing contexts,
 invalid values, and insufficient trials make a record inapplicable. Cost ties
 are broken by record ID; producers should avoid duplicate physical points.
@@ -116,13 +138,32 @@ or runtime drift invalidates the context, the deployment falls back to formal
 sizing or exact execution. V1 returns `NoApplicableConfiguration`; the caller
 performs this fallback explicitly so it cannot be mistaken for a measured zero.
 
+## Window cost composition
+
+The benchmark provides atomic unit costs. Planner derives operation counts from
+the selected materialization and window model:
+
+```text
+updates = input_updates * materializations
+merges = query_executions * (panes_per_query - 1)
+queries = query_executions
+retained_sketches = retained_panes * materializations
+cpu = updates*C_update + merges*C_merge + queries*C_query
+```
+
+A tumbling window has one pane per query and no merge. A shared sliding-window
+plan has one materialization; a natural per-query deployment has one per
+distinct window. Retained memory is measured bytes per sketch multiplied by
+retained sketches, outside the CPU equation.
+
 ## Distribution drift
 
-ERP selection is valid only while the deployment's distribution descriptor
-matches the evidence context. Production integration should periodically derive
-a versioned signature, compare it with the selected profile, and trigger
-replanning on mismatch. Until typed conservative matching exists, a changed
-descriptor must fail closed.
+ERP selection is valid only while the observed shape remains inside the
+configured profile distance. The deployment periodically derives a signature
+and triggers replanning on mismatch. Insufficient benchmark volume, excessive
+shape distance, absent metrics, unsupported runtime parameters, and drift all
+fail closed. Bursts are explicit benchmark scenario provenance and are not
+silently inferred or interpolated.
 
 Recommended future signatures are family-specific:
 
@@ -164,7 +205,7 @@ selection, but must identify that as capability beyond query-local sizing.
 
 ## Future work
 
-- typed distribution signatures and conservative nearest-profile matching;
+- richer family-specific signatures beyond cardinality and Zipf exponent;
 - confidence/quantile error summaries from independent trials;
 - environment descriptors and validity intervals in the ERP wire format;
 - multi-state error composition for merged windows and query DAGs;