fix(promql): bind temporal TopK update semantics on the shared workload plan

control_plane/src/backend_plan/from_stage_config.rs

@@ -264,6 +264,7 @@ mod tests {
             spatial_filter: String::new(),
             grouping,
             item_label: None,
+            heap_update_mode: None,
             aggregation_input: AggregationInput::SketchEnvelope,
         }
     }
@@ -514,6 +515,7 @@ mod tests {
             spatial_filter: String::new(),
             grouping: vec!["zone".to_string()],
             item_label: None,
+            heap_update_mode: None,
             aggregation_input: AggregationInput::Raw,
         };
         let cfg = BackendStageConfig {

control_plane/src/emit/backend_push.rs

@@ -746,6 +746,7 @@ mod tests {
                 ),
                 grouping: vec![],
                 item_label: None,
+                heap_update_mode: None,
                 spatial_filter: String::new(),
                 window_secs: 60,
                 aggregation_input: AggregationInput::SketchEnvelope,

control_plane/src/emit/stage_config.rs

@@ -2986,6 +2986,11 @@ pub(crate) fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonVa
             obj.insert("item_label".to_string(), JsonValue::String(label.clone()));
         }
     }
+    if let Some(mode) = agg.heap_update_mode {
+        if let Some(obj) = parameters.as_object_mut() {
+            obj.insert("weight_mode".into(), JsonValue::String(mode.into()));
+        }
+    }
     let aggregation_input = match agg.aggregation_input {
         AggregationInput::SketchEnvelope => "sketch_envelope",
         AggregationInput::Raw => "raw",
@@ -3206,6 +3211,7 @@ mod tests {
             spatial_filter: String::new(),
             grouping: Vec::new(),
             item_label: None,
+            heap_update_mode: None,
             aggregation_input,
         }
     }

control_plane/src/physical/colored_dag/emitter.rs

@@ -702,6 +702,9 @@ pub struct BackendAggregation {
     /// `parameters["item_label"]` so the data-plane ingest records it on the
     /// CMS sid and can answer per-item `estimate(key)` (FrequencyEstimate).
     pub item_label: Option<String>,
+    /// Runtime accumulator mode derived from SummaryAgg.input.weight, never
+    /// from the TopK readout. None retains the legacy value-update default.
+    pub heap_update_mode: Option<&'static str>,
     /// Phase ε.1 — what shape the backend ingests for this
     /// aggregation. Mode 1 (sketch at edge) / sketch_envelope is the
     /// default (the wire payload is a sketch state already). Mode 2
@@ -899,6 +902,7 @@ impl Emitter for ThreeStageEmitter {
                     });
                     backend_aggregations.push(BackendAggregation {
                         item_label: None,
+                        heap_update_mode: None,
                         aggregation_id,
                         metric_name: edge.source_metric.clone().unwrap_or_default(),
                         family: SummaryFamilyType::Sketch(
@@ -983,6 +987,7 @@ impl Emitter for ThreeStageEmitter {
                     next_agg_index += 1;
                     backend_aggregations.push(BackendAggregation {
                         item_label: None,
+                        heap_update_mode: None,
                         aggregation_id: aid,
                         metric_name: edge.source_metric.clone().unwrap_or_default(),
                         family: SummaryFamilyType::Sketch(

control_plane/src/physical/compiler.rs

@@ -187,6 +187,10 @@ pub struct BackendLocalImplementation {
     pub window_implementation_id: String,
     pub state_layout: String,
     pub implementation_cost: ImplementationCostEvidence,
+    /// Certificates keyed by exact registered PromQL; converted to root IDs
+    /// before workload selection so one query cannot borrow another's evidence.
+    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
+    pub topk_evidence: HashMap<String, TopKMembershipEvidence>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -1582,6 +1586,7 @@ impl BackendLocalPlanningSnapshot {
         }
         let mut queries = Vec::with_capacity(entries.len());
         let mut canonical_roots = Vec::with_capacity(entries.len());
+        let mut topk_evidence_by_id = HashMap::new();
         for (index, entry) in entries.into_iter().enumerate() {
             let evaluation_interval_ms = match entry.recurrence {
                 QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(interval)) => interval.0,
@@ -1634,8 +1639,12 @@ impl BackendLocalPlanningSnapshot {
             cost.workload_fingerprint =
                 canonical_promql(&query_string).map_err(CompileError::QueryPlan)?;
             cost.horizon_seconds = self.implementation.horizon_seconds;
+            let query_id = format!("compat-query-{index}");
+            if let Some(evidence) = self.implementation.topk_evidence.get(&query_string) {
+                topk_evidence_by_id.insert(query_id.clone(), evidence.clone());
+            }
             queries.push(PlanningQuery {
-                query_id: format!("compat-query-{index}"),
+                query_id,
                 query_string,
                 post_asap,
                 source: Source::TimeSeries {
@@ -1656,11 +1665,11 @@ impl BackendLocalPlanningSnapshot {
                 runtime_policy: RuntimeRulePolicy::default(),
             });
         }
-        select_workload_roots(&mut queries, canonical_roots, &HashMap::new())?;
+        select_workload_roots(&mut queries, canonical_roots, &topk_evidence_by_id)?;
         Ok((
             PlanningRequest {
                 queries,
-                evidence: HashMap::new(),
+                evidence: topk_evidence_by_id,
                 planner_revision: PLANNER_REVISION.into(),
             },
             self.environment,
@@ -2508,6 +2517,13 @@ fn physical_aggregation(
         spatial_filter: String::new(),
         grouping: query.group_by.clone(),
         item_label: None,
+        heap_update_mode: selected.parameters.get("weight_mode").and_then(|mode| {
+            match mode.as_str() {
+                Some("count") => Some("count"),
+                Some("value") => Some("value"),
+                _ => None,
+            }
+        }),
         aggregation_input: match target {
             PhysicalDeploymentTarget::DistributedCollectors => AggregationInput::SketchEnvelope,
             PhysicalDeploymentTarget::BackendLocalRemoteWrite => AggregationInput::Raw,
@@ -2589,9 +2605,22 @@ fn collect_selected_materializations(
             }
             SummaryExpr::SummaryAgg {
                 family: SummaryFamilyType::Sketch(kind, _),
+                input,
                 ..
             } => {
                 if let Some(readout) = readout {
+                    let mut parameters = sketch_params_json(kind.params());
+                    if matches!(readout, SketchQuery::TopK { .. }) {
+                        use planner_types::post_asap::SummaryInputExpr;
+                        let mode = match &input.weight {
+                            SummaryInputExpr::Constant(value) if *value == 1.0 => "count",
+                            SummaryInputExpr::Column(
+                                planner_types::pre_asap::ColumnRef::SampleValue,
+                            ) => "value",
+                            _ => return Err("unsupported TopK SummaryUpdate weight".into()),
+                        };
+                        parameters["weight_mode"] = mode.into();
+                    }
                     let metric = summary_agg_metric(node).ok_or_else(|| {
                         "SummaryAgg has no unique time-series source in post-ASAP IR".to_string()
                     })?;
@@ -2605,7 +2634,7 @@ fn collect_selected_materializations(
                         ),
                         readout: Some(readout.clone()),
                         algorithm: format!("{:?}", kind.algorithm()).to_ascii_lowercase(),
-                        parameters: sketch_params_json(kind.params()),
+                        parameters,
                     });
                 }
             }
@@ -2709,6 +2738,32 @@ fn stable_workload_plan_id(
 mod tests {
     use super::*;
 
+    // Count and value rankings must configure different state update contracts.
+    #[test]
+    fn temporal_topk_binds_planner_update_weight() {
+        for (query, mode) in [
+            ("topk(1, sum_over_time(m[1m]))", "value"),
+            ("topk(1, count_over_time(m[1m]))", "count"),
+        ] {
+            let evidence = TopKMembershipEvidence {
+                selected_lower_bound: 101.0,
+                excluded_upper_bound: 100.0,
+                interval_failure_probability: 0.001,
+                observed_at_unix_ms: 9500,
+                source: "unit-fixture".into(),
+            };
+            let request = request_with_evidence("topk", query, Some(evidence)).unwrap();
+            let plan = PhysicalCompiler
+                .compile(request, environment(10000))
+                .unwrap();
+            assert_eq!(plan.precompute_plan.materializations.len(), 1, "{query}");
+            assert_eq!(
+                plan.precompute_plan.materializations[0].parameters["weight_mode"], mode,
+                "{query}"
+            );
+        }
+    }
+
     fn environment(now: u64) -> DeploymentEnvironment {
         DeploymentEnvironment {
             target: PhysicalDeploymentTarget::DistributedCollectors,
@@ -3293,6 +3348,7 @@ mod tests {
                 window_implementation_id: "backend-tumbling-v1".into(),
                 state_layout: "anchored-pane-v1".into(),
                 implementation_cost: template.window_implementations[0].cost.clone(),
+                topk_evidence: HashMap::new(),
             },
             environment,
         };
@@ -3405,13 +3461,15 @@ mod tests {
 
         assert!(plan.collector_plans.is_empty());
         assert!(plan.transmission_plan.rules.is_empty());
-        assert_eq!(plan.query_plan.entries.len(), 4);
-        assert_eq!(plan.precompute_plan.materializations.len(), 3);
+        assert_eq!(plan.query_plan.entries.len(), 6);
+        assert_eq!(plan.precompute_plan.materializations.len(), 5);
         for query in [
             "rate(asap_demo_counter_total[5s])",
             "increase(asap_demo_counter_total[5s])",
             "sum_over_time(asap_demo_gauge[5s])",
             "quantile_over_time(0.5, asap_demo_latency_ms[5s])",
+            "topk(1, sum_over_time(asap_demo_gauge[5s]))",
+            "topk(1, count_over_time(asap_demo_gauge[5s]))",
         ] {
             assert!(plan.query_plan.lookup(query).is_ok(), "missing {query}");
         }

control_plane/src/replan.rs

@@ -680,6 +680,7 @@ impl Replanner {
         Some(BackendStageConfig {
             aggregations: vec![BackendAggregation {
                 item_label: None,
+                heap_update_mode: None,
                 aggregation_id: format!("exact-{}-{}", workload.metric_name, role),
                 metric_name: workload.metric_name.clone(),
                 family: SummaryFamilyType::ExactAggregate(exact_kind, exact_params),
@@ -1310,6 +1311,7 @@ mod tests {
                 BackendStageConfig {
                     aggregations: vec![BackendAggregation {
                         item_label: None,
+                        heap_update_mode: None,
                         aggregation_id: "exact-http_requests_total-sum".to_string(),
                         metric_name: "http_requests_total".to_string(),
                         family: planner_types::post_asap::SummaryFamilyType::ExactAggregate(

data_plane/src/drivers/ingest/otel.rs

@@ -597,13 +597,9 @@ fn flush_barrier_drops(_state: &IngestState, drops: &HashMap<u64, u64>, driver_t
 /// distinct `(rack, node, pod)` tuple under a `grouping_labels=[zone]`
 /// policy would mint its own sid and never roll up.
 ///
-/// `agg_kind` is `ExactAgg { ... }` for both raw-sample and opaque-
-/// envelope sketch paths so the resolver key matches the signature
-/// `reconcile_from_streaming_config` derives from the same config; the
-/// modified-OTLP first-class sketch path takes a different sid-
-/// resolution route inside `route_modified_otlp_sketches_to_precompute`
-/// because it carries per-DP `(SketchAlgorithm, SketchConfig)` and
-/// must distinguish (e.g.) DDSketch vs Kll over the same series.
+/// Configured ingest shares the policy-aware physical identity used by the
+/// live storage sink and backfill. Unbound modified-OTLP sketches retain
+/// their separate wire-level identity protocol.
 fn resolve_bucket_sid_for_agg_config(
     ingest_state: &Arc<IngestState>,
     config: &asap_types::aggregation_config::AggregationConfig,
@@ -619,14 +615,8 @@ fn resolve_bucket_sid_for_agg_config(
         })
         .collect();
     let fp = crate::drivers::ingest::canonical_attrs_fingerprint(&grouping_pairs);
-    let agg_kind = crate::storage_engines::sketch_db::data::AggKind::ExactAgg {
-        agg_type: config.aggregation_type,
-        parameters_canonical: crate::storage_engines::sketch_db::data::canonical_parameters(
-            &config.parameters,
-        ),
-        spatial_filter_canonical: config.spatial_filter_normalized.clone(),
-    };
-    let agg_kind_canonical = agg_kind.canonical_string();
+    let agg_kind_canonical =
+        crate::storage_engines::sketch_db::data::materialization_kind_for_config(config);
     let sid = ingest_state
         .series_resolver
         .resolve(&config.metric, &fp, &agg_kind_canonical);
@@ -4656,14 +4646,8 @@ mod sid_bucketing_tests {
         // regardless of group_key shape — the test pin is on sid
         // assignment, not on group_key content), then verify the
         // sid matches the resolver mint for THAT zone.
-        let agg_kind = crate::storage_engines::sketch_db::data::AggKind::ExactAgg {
-            agg_type: cfg.aggregation_type,
-            parameters_canonical: crate::storage_engines::sketch_db::data::canonical_parameters(
-                &cfg.parameters,
-            ),
-            spatial_filter_canonical: cfg.spatial_filter_normalized.clone(),
-        };
-        let agg_kind_canonical = agg_kind.canonical_string();
+        let agg_kind_canonical =
+            crate::storage_engines::sketch_db::data::materialization_kind_for_config(&cfg);
         for (sid, _, _, samples) in &groups {
             let mut vals: Vec<f64> = samples.iter().map(|(_, _, v)| *v).collect();
             vals.sort_by(|a, b| a.partial_cmp(b).unwrap());

data_plane/src/drivers/ingest/prometheus_remote_write.rs

@@ -461,13 +461,17 @@ fn route_messages(
                 })
                 .collect();
             let attrs_fp = super::canonical_attrs_fingerprint(&grouping_pairs);
-            let agg_kind = crate::storage_engines::sketch_db::data::agg_kind_for_config(config);
-            let sid = ingest.series_resolver.resolve(
-                &config.metric,
-                &attrs_fp,
-                &agg_kind.canonical_string(),
-            );
             let policy_fp = asap_types::PolicyFingerprint(config.policy_fp_u64());
+            // A sketch family is not a complete physical identity. Two
+            // materializations may use the same family and grouping while
+            // differing in update semantics (for example count- versus
+            // value-weighted Top-K). Keep those states on distinct SIDs.
+            let materialization_kind =
+                crate::storage_engines::sketch_db::data::materialization_kind_for_config(config);
+            let sid =
+                ingest
+                    .series_resolver
+                    .resolve(&config.metric, &attrs_fp, &materialization_kind);
             buckets
                 .entry(sid)
                 .or_insert_with(|| ((sid, policy_fp, group_key), Vec::new()))

data_plane/src/precompute_engine/operators/sum_accumulator.rs

@@ -49,12 +49,15 @@ impl SumAccumulator {
     }
 
     pub fn deserialize_from_bytes(buffer: &[u8]) -> Result<Self, Box<dyn std::error::Error>> {
-        if buffer.len() < 4 {
-            return Err("Buffer too short for f32".into());
+        match buffer.len() {
+            // Legacy Python scalar sums carry no sample-count evidence.
+            4 => Ok(Self::with_sum(f32::from_le_bytes(buffer.try_into()?) as f64)),
+            // Counted sums use the same fixed layout as the Collector Sum payload.
+            16 => Self::from_sum_bytes(buffer),
+            len => {
+                Err(format!("Invalid persisted Sum payload length: {len} (want 4 or 16)").into())
+            }
         }
-        // Python uses struct.pack("<f", self.sum) which is 4-byte little-endian float
-        let sum = f32::from_le_bytes([buffer[0], buffer[1], buffer[2], buffer[3]]) as f64;
-        Ok(Self::with_sum(sum))
     }
 
     /// Decode the fixed Sum payload produced by the first-class Sum
@@ -96,8 +99,15 @@ impl SerializableToSink for SumAccumulator {
     }
 
     fn serialize_to_bytes(&self) -> Vec<u8> {
-        // Match Python's struct.pack("<f", self.sum) - 4-byte little-endian float
-        (self.sum as f32).to_le_bytes().to_vec()
+        match self.observation_count {
+            Some(count) => {
+                let mut bytes = Vec::with_capacity(16);
+                bytes.extend_from_slice(&self.sum.to_le_bytes());
+                bytes.extend_from_slice(&count.to_le_bytes());
+                bytes
+            }
+            None => (self.sum as f32).to_le_bytes().to_vec(),
+        }
     }
 }
 
@@ -272,6 +282,40 @@ mod tests {
         assert_eq!(merged.aux_stats().count, None);
     }
 
+    // Persistence retains known counts, including zero and the full u64 range.
+    #[test]
+    fn counted_sum_binary_round_trip() {
+        for count in [0, 3, u64::MAX] {
+            let acc = SumAccumulator {
+                sum: 1.0000000000001,
+                observation_count: Some(count),
+            };
+            let bytes = acc.serialize_to_bytes();
+            assert_eq!(bytes.len(), 16);
+            let restored = SumAccumulator::deserialize_from_bytes(&bytes).unwrap();
+            assert_eq!(restored.sum, acc.sum);
+            assert_eq!(restored.observation_count, Some(count));
+        }
+    }
+
+    // Existing scalar-only files remain readable without inventing counts.
+    #[test]
+    fn legacy_binary_sum_has_unknown_count() {
+        let bytes = 42.5f32.to_le_bytes();
+        let restored = SumAccumulator::deserialize_from_bytes(&bytes).unwrap();
+        assert_eq!(restored.sum, 42.5);
+        assert_eq!(restored.observation_count, None);
+        assert_eq!(restored.serialize_to_bytes(), bytes);
+    }
+
+    // Truncated counted payloads must not silently decode as scalar sums.
+    #[test]
+    fn persisted_sum_rejects_invalid_lengths() {
+        for len in [0, 3, 5, 8, 15, 17] {
+            assert!(SumAccumulator::deserialize_from_bytes(&vec![0; len]).is_err());
+        }
+    }
+
     #[test]
     fn test_sum_accumulator_creation() {
         let acc = SumAccumulator::new();