Reject maintenance updates without an implemented evaluator

data_plane/src/precompute_engine/maintenance_runtime.rs

@@ -42,8 +42,10 @@ impl PrecomputeOperatorRegistry<SummaryState> for OperatorAdapter<'_> {
             return Ok(Arc::clone(&self.source));
         }
         match &node.payload {
-            ExecutableOperatorPayload::SummaryAgg { .. }
-            | ExecutableOperatorPayload::SummaryMerge => merge_inputs(inputs),
+            ExecutableOperatorPayload::SummaryMerge => merge_inputs(inputs),
+            ExecutableOperatorPayload::SummaryAgg { .. } => Err(
+                "maintenance SummaryAgg requires a typed update evaluator; merging input state does not execute its update expression".into(),
+            ),
             payload => Err(format!(
                 "maintenance operator {:?} has no summary-state implementation",
                 payload.operator()
@@ -596,6 +598,36 @@ mod tests {
         Arc::new(accumulator)
     }
 
+    #[test]
+    fn summary_aggregation_does_not_silently_reuse_input_family() {
+        use planner_types::post_asap::{
+            ExactKind, ExactParams, GroupingStrategy, SummaryFamilyType, SummaryUpdate,
+        };
+        use planner_types::pre_asap::{ColumnRef, Reduction};
+
+        let binding = BackendExecutableBinding {
+            nodes: BTreeMap::new(),
+            query_sink: PostAsapNodeId(2),
+            query_plan_sink: control_plane::query_plan::QueryNodeId(2),
+            precompute_sinks: vec![PostAsapNodeId(1)],
+        };
+        let adapter = OperatorAdapter {
+            binding: &binding,
+            source_definition: definition(1),
+            source: sum(7.0),
+        };
+        let mut aggregate = node(1);
+        aggregate.operator = ExecutableOperator::SummaryAgg;
+        aggregate.payload = ExecutableOperatorPayload::SummaryAgg {
+            family: SummaryFamilyType::ExactAggregate(ExactKind::Count, ExactParams::Count),
+            input: SummaryUpdate::column(ColumnRef::SampleValue),
+            reduction: Reduction::by(vec![]),
+            grouping: GroupingStrategy::default(),
+        };
+        let error = adapter.execute(&aggregate, &[Arc::new(sum(7.0))]);
+        assert!(matches!(error, Err(reason) if reason.contains("typed update evaluator")));
+    }
+
     // Receipts follow the declared event-time horizon and never retain accepted
     // summary payloads; expired retries fail instead of becoming duplicate writes.
     #[test]

docs/developer_docs/maintenance-replay.md

@@ -1,5 +1,13 @@
 # Maintenance replay and retention
 
+The production maintenance adapter accepts an already computed source summary
+and executes homogeneous `SummaryMerge` dependencies. An additional `SummaryAgg`
+requires evaluation of its declared update expression and target family; it is
+rejected until that evaluator exists. Copying or merging its input would silently
+ignore those semantics (for example, count over a sum state is not that sum).
+The scheduler's ability to traverse a DAG does not imply every operator is
+implemented by the production adapter.
+
 The maintenance sink retains in-process publication receipts for the active
 physical plan. Receipt identity includes the plan generation, target summary
 definition, output window, and a SHA-256 digest of the source definition,