fix: finalize exact state before maintained value consumers

crates/asap-aware-mapping/src/replacement.rs

@@ -1817,6 +1817,14 @@ fn realize_binary(
 fn finalize_exact_accumulator(
     node: Rc<SummaryNode>,
     logical_output: &QueryExpr,
+) -> Result<Rc<SummaryNode>, ImplementError> {
+    finalize_exact_accumulator_at(node, logical_output, ExecutionTiming::ReadTime)
+}
+
+fn finalize_exact_accumulator_at(
+    node: Rc<SummaryNode>,
+    logical_output: &QueryExpr,
+    timing: ExecutionTiming,
 ) -> Result<Rc<SummaryNode>, ImplementError> {
     let is_exact_state = matches!(
         node.expr,
@@ -1838,7 +1846,7 @@ fn finalize_exact_accumulator(
         expr: SummaryExpr::ValueOperation {
             child: node,
             operation: ValueOperation::FinalizeExactAccumulator,
-            timing: ExecutionTiming::ReadTime,
+            timing,
         },
         schema,
         guarantee,
@@ -2228,6 +2236,11 @@ fn construct_summary_agg(
     }
 
     let bound_child = realize_child_with(&input.child, models, child_target)?;
+    // A maintained parent consumes finalized values, never the child's
+    // accumulator representation. Keep the read boundary explicit even when
+    // an exact scalar accumulator currently stores its value directly.
+    let bound_child =
+        finalize_exact_accumulator_at(bound_child, &input.child, ExecutionTiming::MaintenanceTime)?;
 
     // ── Guarantee (issue #172) ──────────────────────────────────────────
     // Derived *before* the node exists, so an illegal composition is never
@@ -7954,6 +7967,14 @@ mod tests {
             family,
             SummaryFamilyType::Sketch(kind, _) if kind.algorithm() == &SketchAlgorithm::Kll
         ));
+        let SummaryExpr::ValueOperation {
+            child,
+            operation: ValueOperation::FinalizeExactAccumulator,
+            timing: ExecutionTiming::MaintenanceTime,
+        } = &child.expr
+        else {
+            panic!("expected explicit maintenance readout");
+        };
         let SummaryExpr::SummaryAgg {
             family: inner_family,
             child: leaf,

crates/integration-tests/tests/exact_composition.rs

@@ -226,7 +226,7 @@ fn names(node: &SummaryNode) -> Vec<&str> {
 // ── step 1: pin every already-supported exact-accumulator nesting ───────
 
 #[test]
-fn every_exact_accumulator_nests_directly_under_an_outer_sketch() {
+fn every_exact_accumulator_is_finalized_before_an_outer_sketch() {
     use std::time::Duration;
     let cases: Vec<(Rc<QueryExpr>, ExactKind)> = vec![
         (
@@ -297,12 +297,20 @@ fn every_exact_accumulator_nests_directly_under_an_outer_sketch() {
         let SummaryExpr::SummaryAgg { child, .. } = &summary_input.expr else {
             panic!("expected outer SummaryAgg");
         };
+        let SummaryExpr::ValueOperation {
+            child,
+            operation: asap_types::post_asap::ValueOperation::FinalizeExactAccumulator,
+            timing: ExecutionTiming::MaintenanceTime,
+        } = &child.expr
+        else {
+            panic!("{kind:?}: missing maintenance finalization");
+        };
         assert!(
             matches!(
                 &child.expr,
                 SummaryExpr::SummaryAgg { family: SummaryFamilyType::ExactAggregate(k, _), .. } if *k == kind
             ),
-            "{kind:?}: expected the exact accumulator directly under the outer sketch, got {:?}",
+            "{kind:?}: expected the exact accumulator under its finalization, got {:?}",
             child.expr
         );
         validate_execution_data_states(root).expect("accumulator state composes under maintenance");

crates/integration-tests/tests/promql_to_post_asap.rs

@@ -744,6 +744,15 @@ fn promql_quantile_of_rate_binds_kll_over_rate_accumulator() {
         )
     );
 
+    let SummaryExpr::ValueOperation {
+        child,
+        operation: ValueOperation::FinalizeExactAccumulator,
+        timing: asap_types::post_asap::ExecutionTiming::MaintenanceTime,
+    } = &child.expr
+    else {
+        panic!("rate needs a maintenance readout");
+    };
+
     // The rate: exact counter-reset-aware accumulator, per-series (labels
     // and time axis preserved), no estimate wrapper. `rate(...)` has no
     // grouping concept at all — every entity stays its own summary.
@@ -875,3 +884,62 @@ fn promql_sum_of_count_over_time_is_composed_by_default_search() {
             if range.as_secs() == 300 && matches!(child.as_ref(), QueryExpr::Scan { .. })
     ));
 }
+
+#[test]
+fn nested_summary_explicitly_finalizes_exact_child_at_maintenance_time() {
+    // Real workload selection must expose the state-to-value edge; an outer
+    // sketch must not interpret exact accumulator bytes as input samples.
+    let pre = Rc::new(
+        lower_promql(
+            "quantile(0.9, sum_over_time(m[1m]))",
+            AccuracyTarget::Epsilon(0.05),
+        )
+        .unwrap(),
+    );
+    let space = search_workload(vec![("query", pre)]);
+    let selected = space.global_selection(&DefaultCostModel);
+    let plan = selected.materialize(&space.roots[0].1).unwrap().unwrap();
+    let SummaryExpr::SummaryEstimate { summary_input, .. } = &plan.expr else {
+        panic!("expected selected quantile summary");
+    };
+    let SummaryExpr::SummaryAgg { child, .. } = &summary_input.expr else {
+        panic!("expected maintained outer summary");
+    };
+    let SummaryExpr::ValueOperation {
+        child: source,
+        operation,
+        timing,
+    } = &child.expr
+    else {
+        panic!(
+            "missing explicit accumulator finalization: {:?}",
+            child.expr
+        );
+    };
+    assert!(matches!(
+        operation,
+        ValueOperation::FinalizeExactAccumulator
+    ));
+    assert_eq!(
+        *timing,
+        asap_types::post_asap::ExecutionTiming::MaintenanceTime
+    );
+    assert!(matches!(
+        source.expr,
+        SummaryExpr::SummaryAgg {
+            family: SummaryFamilyType::ExactAggregate(ExactKind::Sum, _),
+            ..
+        }
+    ));
+    assert!(child
+        .schema
+        .fields
+        .iter()
+        .all(|field| matches!(field.dtype, SummaryFamilyType::Plain(_))));
+    assert!(child
+        .schema
+        .fields
+        .iter()
+        .any(|field| matches!(field.dtype, SummaryFamilyType::Plain(DataType::Float64))));
+    compile_executable_dag(&plan).expect("explicit boundary is a valid executable DAG");
+}

crates/types/src/post_asap/execution_data_state.rs

@@ -23,7 +23,7 @@
 //! | `SummaryEstimate.summary_input` | `MAINTENANCE_SUMMARY` (any family). Produces `READ_ROWS`. |
 //! | `SummaryJoin.outer/inner` | `MAINTENANCE_ROWS` or `MAINTENANCE_SUMMARY`; never a read-time data_state. |
 //! | `SummarySubtract`/`SummaryDelete`/`SummaryMerge` | `MAINTENANCE_SUMMARY`. |
-//! | `ValueOperation.child` with `MaintenanceTime` | `MAINTENANCE_ROWS`. Produces `MAINTENANCE_ROWS`. |
+//! | `ValueOperation.child` with `MaintenanceTime` | `MAINTENANCE_ROWS`; explicit `FinalizeExactAccumulator` also accepts exact accumulator state. Produces `MAINTENANCE_ROWS`. |
 //! | `ValueOperation.child` with `ReadTime` | `READ_ROWS`. Produces `READ_ROWS`. |
 //!
 //! ## `KeepPreAsap` declares its data_state through the derivation
@@ -421,7 +421,8 @@ fn visit(
                 ExecutionTiming::ReadTime => ExecutionDataState::READ_ROWS,
             };
             let s = produced_data_state(&child.expr).unwrap_or(required);
-            let exact_readout = *timing == ExecutionTiming::ReadTime
+            let exact_readout = (*timing == ExecutionTiming::ReadTime
+                || matches!(operation, ValueOperation::FinalizeExactAccumulator))
                 && s == ExecutionDataState::MAINTENANCE_SUMMARY
                 && is_exact_accumulator_state(&child.schema).is_ok();
             if s != required && !exact_readout {