feat(precompute): install ingest, schema, and producer contracts

control_plane/src/emit/backend_push.rs

@@ -442,41 +442,52 @@ async fn push_cumulative_entries(
     // succeeds only when all three are accepted.
     let plan_id = PLAN_ID_COUNTER.fetch_add(1, Ordering::Relaxed);
     let generated_at_unix_ms = now_unix_ms();
-    let plan_bytes = match crate::backend_plan::from_stage_config(
+    let backend_plan = match crate::backend_plan::from_stage_config(
         &cumulative_be,
         monitors,
         plan_id,
         generated_at_unix_ms,
     ) {
-        Ok(plan) => plan.encode_to_vec(),
+        Ok(plan) => plan,
         Err(e) => {
             warn!(error = %e, "backend_plan::from_stage_config failed; refusing partial publication");
             return PushOutcome::EmitFailed;
         }
     };
-    let precompute_plan = PrecomputePlan {
-        envelope: PlanEnvelope {
-            plan_id,
-            plan_version: 1,
-            generated_at_unix_ms,
-            activation_unix_ms: generated_at_unix_ms,
-            expiry_unix_ms: None,
-            backend_compat: "asap-query-backend.v1".into(),
-            planner_revision: crate::physical::compiler::PLANNER_REVISION.into(),
-            capability_snapshot_id: "replanner".into(),
-        },
-        materializations: match cumulative_be
-            .aggregations
-            .iter()
-            .map(crate::backend_plan::aggregation_config_for_materialization)
-            .collect::<anyhow::Result<Vec<_>>>()
-        {
-            Ok(materializations) => materializations,
-            Err(error) => {
-                warn!(%error, "failed to build typed PrecomputePlan");
-                return PushOutcome::EmitFailed;
-            }
-        },
+    let plan_bytes = backend_plan.encode_to_vec();
+    let precompute_envelope = PlanEnvelope {
+        plan_id,
+        plan_version: 1,
+        generated_at_unix_ms,
+        activation_unix_ms: generated_at_unix_ms,
+        expiry_unix_ms: None,
+        backend_compat: "asap-query-backend.v1".into(),
+        planner_revision: crate::physical::compiler::PLANNER_REVISION.into(),
+        capability_snapshot_id: "replanner".into(),
+    };
+    let materializations = match cumulative_be
+        .aggregations
+        .iter()
+        .map(crate::backend_plan::aggregation_config_for_materialization)
+        .collect::<anyhow::Result<Vec<_>>>()
+    {
+        Ok(materializations) => materializations,
+        Err(error) => {
+            warn!(%error, "failed to build typed PrecomputePlan");
+            return PushOutcome::EmitFailed;
+        }
+    };
+    let precompute_plan = match PrecomputePlan::build(
+        precompute_envelope,
+        materializations,
+        &backend_plan,
+        &["legacy-replanner".into()],
+    ) {
+        Ok(plan) => plan,
+        Err(error) => {
+            warn!(%error, "failed to validate typed PrecomputePlan");
+            return PushOutcome::EmitFailed;
+        }
     };
 
     // Storage-routing: the routing classifier (`build_routing_entry` in

control_plane/src/emit/stage_config.rs

@@ -3142,24 +3142,33 @@ fn sketch_params_to_json(p: &SketchParams) -> JsonValue {
         SketchParams::Kll { k } => json!({ "k": k }),
         SketchParams::DDSketch { alpha } => json!({ "alpha": alpha }),
         SketchParams::Hll { precision } => json!({ "precision": precision }),
-        // Cms/CmsWithHeap: matches the pre-split shape exactly — the old
-        // `SketchParams::Cms` arm never emitted `with_heap` in JSON even
-        // though `CmsParams.with_heap` existed as a field; heap-bearing
-        // and bare CMS produced identical wire JSON. `heap_size` is a
-        // new field with no wire representation here (nothing on the
-        // real backend wire path reads it — see `bind_cms_topk.rs`).
-        SketchParams::Cms { width, depth } | SketchParams::CmsWithHeap { width, depth, .. } => {
-            json!({ "w": width, "d": depth })
-        }
+        SketchParams::Cms { width, depth } => json!({ "w": width, "d": depth }),
+        SketchParams::CmsWithHeap {
+            width,
+            depth,
+            heap_size,
+        } => json!({
+            "w": width,
+            "d": depth,
+            "with_heap": true,
+            "heap_size": heap_size,
+        }),
         // CountSketch/CountSketchWithHeap: the old arm always emitted
         // `with_heap` (from `CountSketchParams.with_heap: bool`);
         // that boolean is now the kind identity itself.
         SketchParams::CountSketch { width, depth } => {
             json!({ "w": width, "d": depth, "with_heap": false })
         }
-        SketchParams::CountSketchWithHeap { width, depth, .. } => {
-            json!({ "w": width, "d": depth, "with_heap": true })
-        }
+        SketchParams::CountSketchWithHeap {
+            width,
+            depth,
+            heap_size,
+        } => json!({
+            "w": width,
+            "d": depth,
+            "with_heap": true,
+            "heap_size": heap_size,
+        }),
         // Exact accumulators never reach here -- see
         // `sketch_kind_to_backend_type`'s doc.
         SketchParams::Kmv { .. } | SketchParams::Theta { .. } => unreachable!(

control_plane/src/opamp/mod.rs

@@ -1016,6 +1016,7 @@ mod tests {
             },
             materializations: vec![crate::physical::compiler::CollectorMaterialization {
                 query_id: "q".into(),
+                materialization: asap_types::PolicyFingerprint(1),
                 metric: "requests".into(),
                 algorithm: "hll".into(),
                 parameters: serde_json::json!({"precision": 14}),