From 2752c31fb3a768fb00ddcb375bc130717ca4063e Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Fri, 11 Sep 2026 11:57:20 -0600
Subject: [PATCH 1/8] Acquire complete immutable input cohorts before
 maintenance publication

---
 .../precompute_engine/maintenance_runtime.rs  |   8 +-
 .../sketch_db/index/maintenance.rs            | 180 ++++++++++++++++++
 2 files changed, 186 insertions(+), 2 deletions(-)

diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs
index cbdaf19f..08b0298f 100644
--- a/data_plane/src/precompute_engine/maintenance_runtime.rs
+++ b/data_plane/src/precompute_engine/maintenance_runtime.rs
@@ -608,8 +608,12 @@ pub fn execute_completed_maintenance(
     let generation = store
         .active_catalog_generation()
         .ok_or("maintenance requires an authoritative catalog")?;
-    let frozen =
-        store.read_frozen_exact_windows(source_sid, source, &generation, &expected, group)?;
+    let mut cohort = store.read_frozen_exact_cohort(
+        &generation,
+        &derived.inputs,
+        &[(source_sid, source, expected, group.clone())],
+    )?;
+    let frozen = cohort.pop().ok_or("immutable input cohort is empty")?;
     let (dag, key, states) =
         prepare_frozen_maintenance_sink(installed, configs, sink, &frozen, window)?;
     let digest = key
diff --git a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs
index 53e6064f..c49af73a 100644
--- a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs
+++ b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs
@@ -16,6 +16,49 @@ pub(crate) struct FrozenExactWindows {
 }
 
 impl SketchStore {
+    /// Acquire a complete immutable read set before any output is reserved.
+    /// Individual payloads are immutable; checking the shared generation again
+    /// after all reads prevents a catalog transition from mixing incarnations.
+    pub(crate) fn read_frozen_exact_cohort(
+        &self,
+        generation: &Arc<CatalogGeneration>,
+        expected_definitions: &BTreeSet<SummaryDefinitionId>,
+        requests: &[(
+            u64,
+            SummaryDefinitionId,
+            BTreeSet<(u64, u64)>,
+            BTreeMap<String, String>,
+        )],
+    ) -> Result<Vec<FrozenExactWindows>, String> {
+        if requests.is_empty() || requests.len() > 65_536 {
+            return Err("immutable input cohort has invalid population count".into());
+        }
+        let supplied = requests
+            .iter()
+            .map(|(_, definition, _, _)| *definition)
+            .collect();
+        if expected_definitions != &supplied {
+            return Err("immutable input cohort differs from installed input definitions".into());
+        }
+        let mut ordered: Vec<_> = requests.iter().collect();
+        ordered.sort_by(|left, right| (left.1, left.0, &left.3).cmp(&(right.1, right.0, &right.3)));
+        if ordered
+            .windows(2)
+            .any(|pair| (pair[0].1, pair[0].0, &pair[0].3) == (pair[1].1, pair[1].0, &pair[1].3))
+        {
+            return Err("immutable input cohort repeats a physical population".into());
+        }
+        self.validate_routed_catalog_generation(Some(generation.as_ref()))?;
+        let inputs = ordered
+            .into_iter()
+            .map(|(sid, definition, windows, group)| {
+                self.read_frozen_exact_windows(*sid, *definition, generation, windows, group)
+            })
+            .collect::<Result<Vec<_>, _>>()?;
+        self.validate_routed_catalog_generation(Some(generation.as_ref()))?;
+        Ok(inputs)
+    }
+
     fn durable_maintenance_population_ids(
         &self,
         definition: SummaryDefinitionId,
@@ -411,6 +454,102 @@ mod tests {
     use crate::storage_engines::types::PrecomputedOutput;
     use asap_types::traits::SerializableToSink;
 
+    #[test]
+    fn cohort_requires_every_durable_source_in_one_catalog_generation() {
+        // Neither a missing second window nor a new catalog may yield a
+        // partially acquired cohort, even when the first source is complete.
+        let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot =
+            serde_json::from_str(include_str!(
+                "../../../../../docs/examples/asapquery-planning-snapshot.json"
+            ))
+            .unwrap();
+        let plan = snapshot.compile().unwrap();
+        let mut first = plan.precompute_plan.materializations[0].clone();
+        first.aggregation_type = asap_types::AggregationType::Sum;
+        first.aggregation_sub_type = "sum".into();
+        first.grouping_labels = std::iter::empty::<String>().collect();
+        let mut second = first.clone();
+        second.metric = "cohort_second".into();
+        let configs = [first, second];
+        let catalog =
+            asap_types::summary_catalog::SummaryCatalog::from_materializations(1, 1, &configs)
+                .unwrap();
+        let store = Arc::new(SketchStore::new());
+        store
+            .install_summary_catalog(Arc::new(catalog.clone()))
+            .unwrap();
+        let generation = store.active_catalog_generation().unwrap();
+        let directory = tempfile::tempdir().unwrap();
+        let mut config = persistence::config::SketchStorePersistenceConfig::with_memory_limit(
+            1 << 24,
+            directory.path().to_path_buf(),
+        );
+        config.delete_older_than_ms = None;
+        config.hot_window_ms = None;
+        let mut persistence = store.start_persistence(config).unwrap();
+        let mut requests = Vec::new();
+        for (index, config) in configs.iter().enumerate() {
+            let definition = config.policy_fingerprint().into();
+            let coordinate = asap_types::sds::SummaryInstanceCoordinates {
+                summary_definition_id: definition,
+                time_range: HalfOpenTimeRange {
+                    start_ms: 0,
+                    end_ms: 1000,
+                },
+                group_values: BTreeMap::new(),
+            };
+            let revision = store
+                .admit_summary_updates(&generation, BTreeSet::from([coordinate.clone()]))
+                .unwrap();
+            let mut output = PrecomputedOutput::new(0, 1000, None, config.policy_fingerprint());
+            output.catalog_generation = Some(Arc::clone(&generation));
+            let mut sum = SumAccumulator::new();
+            sum.update(5.0 + index as f64);
+            let sid = 900 + index as u64;
+            store
+                .publish_admitted_summary_update(
+                    &generation,
+                    &coordinate,
+                    revision,
+                    revision,
+                    2000,
+                    |writer| writer.ingest_precompute_with_series_id(sid, config, &output, &sum),
+                )
+                .unwrap();
+            requests.push((
+                sid,
+                definition,
+                BTreeSet::from([(0, 1000)]),
+                BTreeMap::new(),
+            ));
+        }
+        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+        while !store.seal_finite_summary_input(&generation).unwrap() {
+            assert!(std::time::Instant::now() < deadline);
+            std::thread::sleep(std::time::Duration::from_millis(5));
+        }
+        let definitions = requests.iter().map(|request| request.1).collect();
+        let parts = persistence.manifest.live_parts().len();
+        let cohort = store
+            .read_frozen_exact_cohort(&generation, &definitions, &requests)
+            .unwrap();
+        assert_eq!(cohort.len(), 2);
+        assert!(cohort.iter().all(|input| input.generation == generation));
+        requests[1].2.insert((1000, 2000));
+        assert!(store
+            .read_frozen_exact_cohort(&generation, &definitions, &requests)
+            .is_err());
+        assert_eq!(persistence.manifest.live_parts().len(), parts);
+        let mut next = catalog;
+        next.plan_version += 1;
+        store.install_summary_catalog(Arc::new(next)).unwrap();
+        requests[1].2.remove(&(1000, 2000));
+        assert!(store
+            .read_frozen_exact_cohort(&generation, &definitions, &requests)
+            .is_err());
+        persistence.shutdown();
+    }
+
     #[test]
     fn one_sid_with_two_populations_cannot_publish_a_partial_global_summary() {
         let mut fixture: serde_json::Value = serde_json::from_str(include_str!(
@@ -502,6 +641,47 @@ mod tests {
             )
             .unwrap();
         assert!(!frozen.singleton_population_complete);
+        // The read set is deterministic and all-or-nothing, independently of
+        // the later routing decision (which still rejects this global reduce).
+        let request = |name: &str| {
+            (
+                700,
+                source_id,
+                BTreeSet::from([(0, 60_000)]),
+                BTreeMap::from([("instance".to_string(), name.to_string())]),
+            )
+        };
+        let cohort = store
+            .read_frozen_exact_cohort(
+                &generation,
+                &BTreeSet::from([source_id]),
+                &[request("b"), request("a")],
+            )
+            .unwrap();
+        assert_eq!(cohort.len(), 2);
+        assert_eq!(cohort[0].group["instance"], "a");
+        assert_eq!(cohort[1].group["instance"], "b");
+        assert!(store
+            .read_frozen_exact_cohort(
+                &generation,
+                &BTreeSet::from([source_id]),
+                &[request("a"), request("a")]
+            )
+            .is_err());
+        assert!(store
+            .read_frozen_exact_cohort(
+                &generation,
+                &BTreeSet::from([source_id, target.policy_fingerprint().into()]),
+                &[request("a")]
+            )
+            .is_err());
+        assert!(store
+            .read_frozen_exact_cohort(
+                &generation,
+                &BTreeSet::from([source_id]),
+                &[request("a"), request("absent")]
+            )
+            .is_err());
         let parts_before = persistence.manifest.live_parts().len();
         let resolver = crate::drivers::ingest::series_resolver::SeriesIdResolver::new();
         for _ in 0..2 {

From 190964c8a2c0eeb6306225d6a862bf4b7f281bda Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Fri, 11 Sep 2026 12:05:10 -0600
Subject: [PATCH 2/8] Hold cohort lifetimes through immutable registration and
 publication

---
 .../precompute_engine/maintenance_runtime.rs  |   4 +-
 .../sketch_db/index/maintenance.rs            | 149 ++++++++++++++----
 .../storage_engines/sketch_db/index/mod.rs    |  67 +++++---
 3 files changed, 169 insertions(+), 51 deletions(-)

diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs
index 08b0298f..290f064a 100644
--- a/data_plane/src/precompute_engine/maintenance_runtime.rs
+++ b/data_plane/src/precompute_engine/maintenance_runtime.rs
@@ -635,7 +635,7 @@ pub fn execute_completed_maintenance(
     if store.recover_frozen_maintenance_output(
         target_sid,
         target_config,
-        &frozen,
+        std::slice::from_ref(&frozen),
         digest,
         window,
     )? {
@@ -665,7 +665,7 @@ pub fn execute_completed_maintenance(
         target_config,
         &output,
         state.as_ref(),
-        &frozen,
+        std::slice::from_ref(&frozen),
         digest,
     )
 }
diff --git a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs
index c49af73a..c48680b4 100644
--- a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs
+++ b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs
@@ -277,24 +277,71 @@ impl SketchStore {
 }
 
 impl SketchStore {
+    fn validate_frozen_cohort_bindings<'a>(
+        &self,
+        config: &asap_types::PrecomputeMaterialization,
+        sources: &'a [FrozenExactWindows],
+        instances: &HashMap<u64, SdsBinding>,
+    ) -> Result<&'a Arc<CatalogGeneration>, String> {
+        let generation = &sources
+            .first()
+            .ok_or("immutable publication input cohort is empty")?
+            .generation;
+        let definitions = sources.iter().map(|source| source.definition).collect();
+        if config
+            .derived_input
+            .as_ref()
+            .is_none_or(|input| input.inputs != definitions)
+        {
+            return Err(
+                "immutable publication input definitions differ from installed identity".into(),
+            );
+        }
+        let mut populations = BTreeSet::new();
+        for source in sources {
+            if &source.generation != generation
+                || !populations.insert((source.definition, source.sid, &source.group))
+            {
+                return Err(
+                    "immutable publication input cohort has mixed generations or duplicates".into(),
+                );
+            }
+            let binding = instances
+                .get(&source.sid)
+                .ok_or("immutable source was removed")?;
+            if binding.metadata.policy_fp != source.definition.fingerprint()
+                || binding.catalog_generation.as_deref() != Some(generation.as_ref())
+                || !binding.metadata.is_writable()
+            {
+                return Err("immutable source identity or lifetime changed".into());
+            }
+        }
+        self.validate_routed_catalog_generation(Some(generation.as_ref()))?;
+        Ok(generation)
+    }
+
     pub(crate) fn recover_frozen_maintenance_output(
         &self,
         sid: u64,
         config: &asap_types::PrecomputeMaterialization,
-        source: &FrozenExactWindows,
+        sources: &[FrozenExactWindows],
         digest: [u8; 32],
         window: (u64, u64),
     ) -> Result<bool, String> {
-        self.validate_routed_catalog_generation(Some(source.generation.as_ref()))?;
+        let _admission = self
+            .admission
+            .read()
+            .map_err(|_| "admission registry poisoned")?;
         let instances = self
             .instances
             .read()
             .map_err(|_| "instance registry poisoned")?;
+        let generation = self.validate_frozen_cohort_bindings(config, sources, &instances)?;
         let Some(binding) = instances.get(&sid) else {
             return Ok(false);
         };
         if binding.metadata.policy_fp != config.policy_fingerprint()
-            || binding.catalog_generation.as_deref() != Some(source.generation.as_ref())
+            || binding.catalog_generation.as_deref() != Some(generation.as_ref())
             || !binding.metadata.is_writable()
         {
             return Err("immutable output identity or lifetime changed".into());
@@ -345,41 +392,37 @@ impl SketchStore {
         config: &asap_types::PrecomputeMaterialization,
         output: &crate::storage_engines::types::PrecomputedOutput,
         state: &dyn AggregateCore,
-        source: &FrozenExactWindows,
+        sources: &[FrozenExactWindows],
         input_digest: [u8; 32],
     ) -> Result<bool, String> {
         use persistence::source::{EpochSnapshot, EpochSnapshotEntry};
-        if config
-            .derived_input
-            .as_ref()
-            .is_none_or(|derived| derived.inputs != BTreeSet::from([source.definition]))
-            || output.policy_fp != config.policy_fingerprint()
-            || output.catalog_generation.as_deref() != Some(source.generation.as_ref())
+        // One guard excludes source retirement through target registration
+        // and commit; admission excludes catalog transitions in the same span.
+        let _admission = self
+            .admission
+            .read()
+            .map_err(|_| "admission registry poisoned")?;
+        let mut instances = self
+            .instances
+            .write()
+            .map_err(|_| "instance registry poisoned")?;
+        let generation = self.validate_frozen_cohort_bindings(config, sources, &instances)?;
+        if output.policy_fp != config.policy_fingerprint()
+            || output.catalog_generation.as_deref() != Some(generation.as_ref())
         {
             return Err("derived publication differs from its installed input identity".into());
         }
         let labels = self
-            .register_precompute_output(sid, config, output)
+            .register_precompute_output_with_instances(sid, config, output, &mut instances)
             .ok_or("derived output registration failed")?;
         let _mutation = self.begin_state_mutation();
-        let instances = self
-            .instances
-            .read()
-            .map_err(|_| "instance registry poisoned")?;
-        let source_binding = instances
-            .get(&source.sid)
-            .ok_or("immutable source was removed")?;
         let binding = instances.get(&sid).ok_or("derived output was removed")?;
-        if source_binding.metadata.policy_fp != source.definition.fingerprint()
-            || source_binding.catalog_generation.as_deref() != Some(source.generation.as_ref())
-            || source_binding.metadata.status() == AggStatus::Expired
-            || !binding.metadata.is_writable()
+        if !binding.metadata.is_writable()
             || binding.metadata.policy_fp != output.policy_fp
-            || binding.catalog_generation.as_deref() != Some(source.generation.as_ref())
+            || binding.catalog_generation.as_deref() != Some(generation.as_ref())
         {
             return Err("derived publication physical lifetime changed".into());
         }
-        self.validate_routed_catalog_generation(Some(source.generation.as_ref()))?;
         let mut completed = self
             .completed_windows
             .write()
@@ -470,7 +513,15 @@ mod tests {
         first.grouping_labels = std::iter::empty::<String>().collect();
         let mut second = first.clone();
         second.metric = "cohort_second".into();
-        let configs = [first, second];
+        let mut target = first.clone();
+        target.derived_input = Some(asap_types::derived_input::DerivedInputIdentity {
+            inputs: BTreeSet::from([
+                first.policy_fingerprint().into(),
+                second.policy_fingerprint().into(),
+            ]),
+            program_sha256: "0".repeat(64),
+        });
+        let configs = [first, second, target];
         let catalog =
             asap_types::summary_catalog::SummaryCatalog::from_materializations(1, 1, &configs)
                 .unwrap();
@@ -488,7 +539,7 @@ mod tests {
         config.hot_window_ms = None;
         let mut persistence = store.start_persistence(config).unwrap();
         let mut requests = Vec::new();
-        for (index, config) in configs.iter().enumerate() {
+        for (index, config) in configs.iter().take(2).enumerate() {
             let definition = config.policy_fingerprint().into();
             let coordinate = asap_types::sds::SummaryInstanceCoordinates {
                 summary_definition_id: definition,
@@ -540,6 +591,37 @@ mod tests {
             .read_frozen_exact_cohort(&generation, &definitions, &requests)
             .is_err());
         assert_eq!(persistence.manifest.live_parts().len(), parts);
+        let target = &configs[2];
+        let mut output = PrecomputedOutput::new(0, 1000, None, target.policy_fingerprint());
+        output.catalog_generation = Some(Arc::clone(&generation));
+        let mut sum = SumAccumulator::new();
+        sum.update(11.0);
+        assert!(store
+            .publish_frozen_maintenance_output(902, target, &output, &sum, &cohort, [42; 32])
+            .unwrap());
+        assert!(store
+            .recover_frozen_maintenance_output(902, target, &cohort, [42; 32], (0, 1000))
+            .unwrap());
+        let published_parts = persistence.manifest.live_parts().len();
+        store.force_expire(901).unwrap();
+        assert!(store
+            .publish_frozen_maintenance_output(903, target, &output, &sum, &cohort, [42; 32])
+            .is_err());
+        assert!(store
+            .recover_frozen_maintenance_output(902, target, &cohort, [42; 32], (0, 1000))
+            .is_err());
+        assert!(!store.instances.read().unwrap().contains_key(&903));
+        assert_eq!(persistence.manifest.live_parts().len(), published_parts);
+        assert!(!store
+            .persistence_metadata
+            .read()
+            .unwrap()
+            .as_ref()
+            .unwrap()
+            .load_strict()
+            .unwrap()
+            .iter()
+            .any(|record| record.sid == 903));
         let mut next = catalog;
         next.plan_version += 1;
         store.install_summary_catalog(Arc::new(next)).unwrap();
@@ -831,7 +913,7 @@ mod tests {
         let catalog = asap_types::summary_catalog::SummaryCatalog::from_materializations(
             1,
             1,
-            &[source_config, target.clone()],
+            &[source_config.clone(), target.clone()],
         )
         .unwrap();
         let store = Arc::new(SketchStore::new());
@@ -847,6 +929,11 @@ mod tests {
         let generation = store.active_catalog_generation().unwrap();
         let mut output = PrecomputedOutput::new(0, 1000, None, target.policy_fingerprint());
         output.catalog_generation = Some(Arc::clone(&generation));
+        let mut source_output = output.clone();
+        source_output.policy_fp = source_config.policy_fingerprint();
+        store
+            .register_precompute_output(600, &source_config, &source_output)
+            .unwrap();
         store
             .register_precompute_output(601, &target, &output)
             .unwrap();
@@ -883,7 +970,13 @@ mod tests {
             singleton_population_complete: false,
         };
         assert!(store
-            .recover_frozen_maintenance_output(601, &target, &input, [7; 32], (0, 1000))
+            .recover_frozen_maintenance_output(
+                601,
+                &target,
+                std::slice::from_ref(&input),
+                [7; 32],
+                (0, 1000)
+            )
             .unwrap());
         assert_eq!(
             store.completed_windows.read().unwrap().get(&601),
diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs
index e9ef1234..c7c29c90 100644
--- a/data_plane/src/storage_engines/sketch_db/index/mod.rs
+++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs
@@ -818,6 +818,15 @@ impl SketchStore {
     /// where the two indexes were written under separate sequential
     /// locks). See the index-field doc comments for the full invariant.
     pub fn register(&self, meta: SketchInstanceMetadata) {
+        let mut instances = self.instances.write().unwrap();
+        self.register_with_instances(meta, &mut instances);
+    }
+
+    fn register_with_instances(
+        &self,
+        meta: SketchInstanceMetadata,
+        instances: &mut HashMap<u64, SdsBinding>,
+    ) -> bool {
         let sid = meta.sid;
         let policy_fp = meta.policy_fp;
         // A non-legacy materialization must resolve through the installed
@@ -827,7 +836,7 @@ impl SketchStore {
             Ok(instance) => instance,
             Err(error) => {
                 tracing::warn!(sid, %error, "rejecting SummaryStore registration outside the active SummaryCatalog");
-                return;
+                return false;
             }
         };
         let metric_name = instance
@@ -835,10 +844,9 @@ impl SketchStore {
             .time_series_metric()
             .map(str::to_owned);
         // Fixed lock order: instances → policy_to_series_ids → metric_to_series_ids.
-        let mut instances = self.instances.write().unwrap();
         if self.removed_sids.read().unwrap().contains_key(&sid) {
             tracing::warn!(sid, "rejecting reuse of a removed summary instance ID");
-            return;
+            return false;
         }
         let mut policy_idx = self.policy_to_series_ids.write().unwrap();
         let mut metric_idx = self.metric_to_series_ids.write().unwrap();
@@ -849,6 +857,7 @@ impl SketchStore {
         if let Some(metric_name) = metric_name {
             metric_idx.entry(metric_name).or_default().insert(sid);
         }
+        true
     }
 
     /// Install one authoritative catalog snapshot for future registrations.
@@ -2973,13 +2982,24 @@ impl SketchStore {
         sid: u64,
         agg_cfg: &asap_types::PrecomputeMaterialization,
         output: &crate::storage_engines::types::PrecomputedOutput,
+    ) -> Option<BTreeMap<String, String>> {
+        let mut instances = self.instances.write().ok()?;
+        self.register_precompute_output_with_instances(sid, agg_cfg, output, &mut instances)
+    }
+
+    fn register_precompute_output_with_instances(
+        &self,
+        sid: u64,
+        agg_cfg: &asap_types::PrecomputeMaterialization,
+        output: &crate::storage_engines::types::PrecomputedOutput,
+        instances: &mut HashMap<u64, SdsBinding>,
     ) -> Option<BTreeMap<String, String>> {
         let (_attrs_fp, label_values_map) = build_attrs_fp_and_label_map(agg_cfg, output);
         let key_names = &agg_cfg.grouping_labels.names();
         let agg_kind = crate::storage_engines::sketch_db::data::agg_kind_for_config(agg_cfg);
         let (capability, accuracy) = agg_kind.capability_and_accuracy();
 
-        match self.instance(sid) {
+        match instances.get(&sid) {
             None => {
                 if self.active_catalog_generation().as_deref()
                     != output.catalog_generation.as_deref()
@@ -3009,23 +3029,28 @@ impl SketchStore {
                 // unconditionally `None`, which meant ASAP-tier ExactAgg
                 // state was reachable only through the legacy precompute
                 // query path; capability-matching couldn't see it.
-                self.register(SketchInstanceMetadata {
-                    sid,
-                    metric_name: agg_cfg.metric.clone(),
-                    group_by_keys,
-                    capability: Some(capability),
-                    agg_kind,
-                    accuracy,
-                    first_seen_unix_ms: output.start_timestamp as i64,
-                    retired_at_ms: None,
-                    expires_at_ms: None,
-                    // Trust the caller's output — it carries the
-                    // policy fingerprint computed at emit time
-                    // (precompute worker / backfill processor).
-                    // Falling back to `from_config(&agg_cfg)` here
-                    // would also be correct but redundant.
-                    policy_fp: output.policy_fp,
-                });
+                if !self.register_with_instances(
+                    SketchInstanceMetadata {
+                        sid,
+                        metric_name: agg_cfg.metric.clone(),
+                        group_by_keys,
+                        capability: Some(capability),
+                        agg_kind,
+                        accuracy,
+                        first_seen_unix_ms: output.start_timestamp as i64,
+                        retired_at_ms: None,
+                        expires_at_ms: None,
+                        // Trust the caller's output — it carries the
+                        // policy fingerprint computed at emit time
+                        // (precompute worker / backfill processor).
+                        // Falling back to `from_config(&agg_cfg)` here
+                        // would also be correct but redundant.
+                        policy_fp: output.policy_fp,
+                    },
+                    instances,
+                ) {
+                    return None;
+                }
             }
             Some(existing) if !existing.is_writable() || existing.policy_fp != output.policy_fp => {
                 return None;

From 537bc86265b0ce0f0a8729285c20d43cbfe7d437 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Fri, 11 Sep 2026 12:09:30 -0600
Subject: [PATCH 3/8] Bind immutable receipts to canonical complete input
 lineage

---
 .../precompute_engine/maintenance_runtime.rs  | 151 +++++++++++++++---
 1 file changed, 129 insertions(+), 22 deletions(-)

diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs
index 290f064a..bb8d3079 100644
--- a/data_plane/src/precompute_engine/maintenance_runtime.rs
+++ b/data_plane/src/precompute_engine/maintenance_runtime.rs
@@ -381,6 +381,67 @@ fn merge_inputs(inputs: &[Arc<MaintenanceValue>]) -> Result<MaintenanceValue, St
     })
 }
 
+fn frozen_cohort_lineage(
+    inputs: &[crate::storage_engines::sketch_db::index::FrozenExactWindows],
+    expected: &asap_types::derived_input::DerivedInputIdentity,
+) -> Result<[u8; 32], String> {
+    let generation = &inputs
+        .first()
+        .ok_or("immutable input cohort is empty")?
+        .generation;
+    if inputs
+        .iter()
+        .map(|input| input.definition)
+        .collect::<BTreeSet<_>>()
+        != expected.inputs
+    {
+        return Err("immutable lineage differs from installed input definitions".into());
+    }
+    let mut ordered: Vec<_> = inputs.iter().collect();
+    ordered.sort_by(|left, right| {
+        (left.definition, left.sid, &left.group).cmp(&(right.definition, right.sid, &right.group))
+    });
+    if ordered.windows(2).any(|pair| {
+        (pair[0].definition, pair[0].sid, &pair[0].group)
+            == (pair[1].definition, pair[1].sid, &pair[1].group)
+    }) {
+        return Err("immutable lineage repeats a physical population".into());
+    }
+    let multiple = ordered.len() > 1;
+    let mut lineage = Sha256::new();
+    if multiple {
+        lineage.update(b"immutable-maintenance-input-v2");
+        lineage.update((ordered.len() as u64).to_be_bytes());
+    } else {
+        // Preserve the existing durable single-input receipt identity.
+        lineage.update(b"immutable-maintenance-input-v1");
+    }
+    for input in ordered {
+        if &input.generation != generation || input.windows.is_empty() {
+            return Err("immutable lineage has mixed generations or empty windows".into());
+        }
+        lineage.update(input.sid.to_be_bytes());
+        let metadata =
+            serde_json::to_vec(&(&input.definition, &input.generation, &input.group, expected))
+                .map_err(|error| error.to_string())?;
+        if multiple {
+            lineage.update((metadata.len() as u64).to_be_bytes());
+        }
+        lineage.update(metadata);
+        if multiple {
+            lineage.update((input.windows.len() as u64).to_be_bytes());
+        }
+        for ((start, end), state) in &input.windows {
+            lineage.update(start.to_be_bytes());
+            lineage.update(end.to_be_bytes());
+            let bytes = state.serialize_to_bytes();
+            lineage.update((bytes.len() as u64).to_be_bytes());
+            lineage.update(bytes);
+        }
+    }
+    Ok(lineage.finalize().into())
+}
+
 /// Evaluate one installed maintenance sink over an immutable physical source
 /// incarnation. Durable publication is a separate existing-store transaction;
 /// the in-memory scheduler cache here never claims durable exactly-once writes.
@@ -453,28 +514,12 @@ fn prepare_frozen_maintenance_sink(
     if &actual_input != expected_input {
         return Err("immutable sink input program differs from its catalog identity".into());
     }
-    let mut lineage = Sha256::new();
-    lineage.update(b"immutable-maintenance-input-v1");
-    lineage.update(input.sid.to_be_bytes());
-    lineage.update(
-        serde_json::to_vec(&(
-            &input.definition,
-            &input.generation,
-            &input.group,
-            expected_input,
-        ))
-        .map_err(|error| error.to_string())?,
-    );
-    let mut states = Vec::with_capacity(input.windows.len());
-    for ((start, end), state) in &input.windows {
-        lineage.update(start.to_be_bytes());
-        lineage.update(end.to_be_bytes());
-        let bytes = state.serialize_to_bytes();
-        lineage.update((bytes.len() as u64).to_be_bytes());
-        lineage.update(&bytes);
-        states.push((*end as i64, Arc::clone(state)));
-    }
-    let digest: [u8; 32] = lineage.finalize().into();
+    let digest = frozen_cohort_lineage(std::slice::from_ref(input), expected_input)?;
+    let states = input
+        .windows
+        .iter()
+        .map(|((_, end), state)| (*end as i64, Arc::clone(state)))
+        .collect();
     let key = MaterializationCommitKey {
         plan_id: input.generation.plan_id,
         plan_version: input.generation.plan_version,
@@ -1356,6 +1401,68 @@ mod tests {
         asap_types::PolicyFingerprint(value).into()
     }
 
+    #[test]
+    fn cohort_lineage_is_order_independent_and_binds_every_input() {
+        use crate::storage_engines::sketch_db::index::FrozenExactWindows;
+        let make = |sid, id, value| {
+            let mut state = crate::precompute_engine::operators::SumAccumulator::new();
+            state.update(value);
+            FrozenExactWindows {
+                sid,
+                definition: definition(id),
+                generation: Arc::new(asap_types::sds::CatalogGeneration {
+                    schema_version: 2,
+                    plan_id: 1,
+                    plan_version: 1,
+                    snapshot_sha256: "0".repeat(64),
+                }),
+                group: BTreeMap::from([("instance".into(), sid.to_string())]),
+                windows: BTreeMap::from([((0, 1000), Arc::new(state) as SummaryState)]),
+                singleton_population_complete: false,
+            }
+        };
+        let expected = asap_types::derived_input::DerivedInputIdentity {
+            inputs: BTreeSet::from([definition(1), definition(2)]),
+            program_sha256: "1".repeat(64),
+        };
+        let baseline =
+            frozen_cohort_lineage(&[make(10, 1, 3.0), make(20, 2, 5.0)], &expected).unwrap();
+        assert_eq!(
+            baseline,
+            frozen_cohort_lineage(&[make(20, 2, 5.0), make(10, 1, 3.0)], &expected).unwrap()
+        );
+        assert_ne!(
+            baseline,
+            frozen_cohort_lineage(&[make(10, 1, 3.0), make(20, 2, 6.0)], &expected).unwrap()
+        );
+        assert_ne!(
+            baseline,
+            frozen_cohort_lineage(&[make(10, 1, 3.0), make(21, 2, 5.0)], &expected).unwrap()
+        );
+        let mut changed = make(20, 2, 5.0);
+        changed.group.insert("instance".into(), "other".into());
+        assert_ne!(
+            baseline,
+            frozen_cohort_lineage(&[make(10, 1, 3.0), changed], &expected).unwrap()
+        );
+        let mut changed = make(20, 2, 5.0);
+        let state = changed.windows.remove(&(0, 1000)).unwrap();
+        changed.windows.insert((1000, 2000), state);
+        assert_ne!(
+            baseline,
+            frozen_cohort_lineage(&[make(10, 1, 3.0), changed], &expected).unwrap()
+        );
+        let mut changed = make(20, 2, 5.0);
+        Arc::make_mut(&mut changed.generation).plan_version += 1;
+        assert!(frozen_cohort_lineage(&[make(10, 1, 3.0), changed], &expected).is_err());
+        assert!(frozen_cohort_lineage(&[make(10, 1, 3.0)], &expected).is_err());
+        assert!(frozen_cohort_lineage(
+            &[make(10, 1, 3.0), make(10, 1, 3.0), make(20, 2, 5.0)],
+            &expected
+        )
+        .is_err());
+    }
+
     fn node(id: u32) -> ExecutableDagNode {
         ExecutableDagNode {
             id: PostAsapNodeId(id),

From b761197ce55835e4836b8953d6dcb774fa51ae40 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Fri, 11 Sep 2026 11:54:45 -0600
Subject: [PATCH 4/8] feat: validate shared derived source window cohorts

---
 control_plane/src/physical/compiler.rs   |  20 ++---
 crates/asap_types/src/precompute_plan.rs | 108 +++++++++++++++++++++--
 2 files changed, 110 insertions(+), 18 deletions(-)

diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index f5d91043..b3788499 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -1239,18 +1239,18 @@ impl PhysicalCompiler {
                                 query_id: query.query_id.clone(),
                                 reason: "derived source config missing".into(),
                             })?;
-                    if runtime_materialization.window_size != source_config.window_size
-                        || runtime_materialization.slide_interval != source_config.slide_interval
-                        || runtime_materialization.pane_origin_ms != source_config.pane_origin_ms
-                        || runtime_materialization.window_size
-                            != runtime_materialization.slide_interval
-                        || runtime_materialization.stored_window_ms()
-                            != source_config.stored_window_ms()
-                        || source_config.stored_window_ms()
-                            != source_config.window_size.saturating_mul(1_000)
+                    asap_types::precompute_plan::validated_source_window_cohort(
+                        &runtime_materialization,
+                        &[source_config],
+                    )
+                    .map_err(|error| CompileError::Query {
+                        query_id: query.query_id.clone(),
+                        reason: error.to_string(),
+                    })?;
+                    if runtime_materialization.window_size != runtime_materialization.slide_interval
                     {
                         return Err(CompileError::Query { query_id: query.query_id.clone(),
-                            reason: "immutable scalar composition requires matching full nonoverlapping windows".into() });
+                            reason: "immutable scalar composition runtime requires nonoverlapping windows".into() });
                     }
                     let SummaryExpr::SummaryAgg { child, .. } = &selected.node.expr else {
                         unreachable!()
diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs
index 543fae8d..efc95582 100644
--- a/crates/asap_types/src/precompute_plan.rs
+++ b/crates/asap_types/src/precompute_plan.rs
@@ -9,6 +9,55 @@ use serde::{Deserialize, Serialize};
 use std::collections::{BTreeMap, BTreeSet, HashMap};
 use thiserror::Error;
 
+/// Validate the physical windows consumed by one derived input program.
+/// This returns the existing source definitions, not a second serialized
+/// contract. It does not prove completion or authorize runtime execution.
+/// Full-window sliding is lossless here; pane merging requires a separate
+/// explicit operator and is deliberately not inferred from window sizes.
+pub fn validated_source_window_cohort<'a>(
+    target: &crate::PrecomputeMaterialization,
+    sources: &[&'a crate::PrecomputeMaterialization],
+) -> Result<Vec<&'a crate::PrecomputeMaterialization>, PrecomputePlanError> {
+    let invalid = || {
+        PrecomputePlanError::CatalogContract(
+            "derived inputs require matching explicit full stored windows".into(),
+        )
+    };
+    let full_ms = target.window_size.checked_mul(1000).ok_or_else(invalid)?;
+    if sources.is_empty()
+        || full_ms == 0
+        || target.slide_interval == 0
+        || target.slide_interval > target.window_size
+        || target.pane_origin_ms.is_none()
+        || target.stored_window_ms() != full_ms
+        || (target.slide_interval < target.window_size
+            && !matches!(
+                target.window_layout,
+                crate::WindowMaterializationLayout::FullWindow
+            ))
+    {
+        return Err(invalid());
+    }
+    let mut identities = BTreeSet::new();
+    for source in sources {
+        if source.derived_input.is_some()
+            || !identities.insert(source.policy_fingerprint())
+            || source.window_size != target.window_size
+            || source.slide_interval != target.slide_interval
+            || source.pane_origin_ms != target.pane_origin_ms
+            || source.stored_window_ms() != full_ms
+            || (source.slide_interval < source.window_size
+                && !matches!(
+                    source.window_layout,
+                    crate::WindowMaterializationLayout::FullWindow
+                ))
+        {
+            return Err(invalid());
+        }
+    }
+    Ok(sources.to_vec())
+}
+
 pub const BACKEND_COMPAT: &str = "asap-query-backend.v1";
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -381,14 +430,11 @@ impl PrecomputePlan {
                 .iter()
                 .find(|candidate| candidate.policy_fingerprint() == source_id.fingerprint())
                 .ok_or_else(invalid)?;
-            if source.derived_input.is_some()
-                || source.window_size != config.window_size
-                || source.slide_interval != config.slide_interval
-                || source.window_size != source.slide_interval
-                || source.pane_origin_ms != config.pane_origin_ms
-                || source.stored_window_ms() != config.stored_window_ms()
-                || source.window_size.checked_mul(1000) != Some(source.stored_window_ms())
-            {
+            validated_source_window_cohort(config, &[source])?;
+            // Current installed runtime capability remains nonoverlapping.
+            // The shared cohort contract also describes explicit full-window
+            // sliding for consumers which separately prove its completion.
+            if source.window_size != source.slide_interval {
                 return Err(invalid());
             }
             let mut matched = false;
@@ -700,3 +746,49 @@ pub(crate) fn state_encodings(family: &SummaryFamilyType) -> Vec<StateEncoding>
         _ => Vec::new(),
     }
 }
+
+#[cfg(test)]
+mod source_window_cohort_tests {
+    use super::*;
+    fn full_window() -> crate::PrecomputeMaterialization {
+        serde_json::from_value(serde_json::json!({
+            "aggregation_type":"Sum", "aggregation_sub_type":"", "parameters":{},
+            "grouping_labels":{"labels":[]}, "aggregated_labels":{"labels":[]},
+            "rollup_labels":{"labels":[]}, "original_yaml":"",
+            "window_size":60, "slide_interval":10, "window_type":"sliding",
+            "window_layout":{"kind":"full_window"}, "pane_origin_ms":0,
+            "spatial_filter":"", "spatial_filter_normalized":"", "metric":"m",
+            "num_aggregates_to_retain":null, "table_name":null, "value_projection":null
+        }))
+        .unwrap()
+    }
+    #[test]
+    fn full_sliding_cohort_preserves_explicit_windows_and_identity() {
+        let target = full_window();
+        let source = full_window();
+        let result = validated_source_window_cohort(&target, &[&source]).unwrap();
+        assert!(std::ptr::eq(result[0], &source));
+        let mut other = source.clone();
+        other.metric = "other".into();
+        assert_eq!(
+            validated_source_window_cohort(&target, &[&source, &other])
+                .unwrap()
+                .len(),
+            2
+        );
+        assert!(validated_source_window_cohort(&target, &[&source, &source]).is_err());
+        for mutation in 0..3 {
+            let mut changed = source.clone();
+            match mutation {
+                0 => changed.pane_origin_ms = Some(1),
+                1 => changed.slide_interval = 20,
+                _ => {
+                    changed.window_layout =
+                        crate::WindowMaterializationLayout::Pane { pane_secs: 10 }
+                }
+            }
+            assert_ne!(source.policy_fingerprint(), changed.policy_fingerprint());
+            assert!(validated_source_window_cohort(&target, &[&changed]).is_err());
+        }
+    }
+}

From 0e96c09e06c68ad92a7b20c6383f8f77bca75f9d Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Fri, 11 Sep 2026 11:55:57 -0600
Subject: [PATCH 5/8] fix: retain legacy nonoverlapping origin contract

---
 crates/asap_types/src/precompute_plan.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs
index efc95582..f48df61a 100644
--- a/crates/asap_types/src/precompute_plan.rs
+++ b/crates/asap_types/src/precompute_plan.rs
@@ -28,7 +28,7 @@ pub fn validated_source_window_cohort<'a>(
         || full_ms == 0
         || target.slide_interval == 0
         || target.slide_interval > target.window_size
-        || target.pane_origin_ms.is_none()
+        || (target.slide_interval < target.window_size && target.pane_origin_ms.is_none())
         || target.stored_window_ms() != full_ms
         || (target.slide_interval < target.window_size
             && !matches!(

From 91f380b9edea083ff63f6e228a3d0d003e7cb590 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Fri, 11 Sep 2026 11:57:01 -0600
Subject: [PATCH 6/8] test: distinguish explicit sliding and legacy origins

---
 crates/asap_types/src/precompute_plan.rs | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs
index f48df61a..818195a3 100644
--- a/crates/asap_types/src/precompute_plan.rs
+++ b/crates/asap_types/src/precompute_plan.rs
@@ -766,6 +766,11 @@ mod source_window_cohort_tests {
     fn full_sliding_cohort_preserves_explicit_windows_and_identity() {
         let target = full_window();
         let source = full_window();
+        let mut missing_origin = source.clone();
+        missing_origin.pane_origin_ms = None;
+        assert!(validated_source_window_cohort(&missing_origin, &[&missing_origin]).is_err());
+        missing_origin.slide_interval = missing_origin.window_size;
+        assert!(validated_source_window_cohort(&missing_origin, &[&missing_origin]).is_ok());
         let result = validated_source_window_cohort(&target, &[&source]).unwrap();
         assert!(std::ptr::eq(result[0], &source));
         let mut other = source.clone();

From 8db331968027b3a27b07a0939168bac17be624af Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Fri, 11 Sep 2026 12:16:55 -0600
Subject: [PATCH 7/8] Resolve frozen maintenance inputs at each materialized
 frontier

---
 .../precompute_engine/maintenance_runtime.rs  | 307 +++++++++++++-----
 1 file changed, 222 insertions(+), 85 deletions(-)

diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs
index bb8d3079..9e0c7467 100644
--- a/data_plane/src/precompute_engine/maintenance_runtime.rs
+++ b/data_plane/src/precompute_engine/maintenance_runtime.rs
@@ -56,13 +56,18 @@ type PendingOutput = (
     Box<dyn AggregateCore>,
 );
 
+enum MaintenanceInputs<'a> {
+    Live {
+        definition: asap_types::sds::SummaryDefinitionId,
+        state: SummaryState,
+    },
+    Frozen(&'a [crate::storage_engines::sketch_db::index::FrozenExactWindows]),
+}
+
 struct OperatorAdapter<'a> {
     binding: &'a BackendExecutableBinding,
-    source_definition: asap_types::sds::SummaryDefinitionId,
-    source: SummaryState,
+    inputs: MaintenanceInputs<'a>,
     configs: &'a [asap_types::aggregation_config::AggregationConfig],
-    immutable_windows: Option<Arc<[(i64, SummaryState)]>>,
-    singleton_population_complete: bool,
 }
 
 impl PrecomputeOperatorRegistry<MaintenanceValue> for OperatorAdapter<'_> {
@@ -72,13 +77,10 @@ impl PrecomputeOperatorRegistry<MaintenanceValue> for OperatorAdapter<'_> {
         &self,
         node: &ExecutableDagNode,
     ) -> Result<Option<MaintenanceValue>, String> {
-        if !matches!(
-            self.binding.node(node.id),
-            Some(BackendNodeBinding::Materialization { summary_definition })
-                if *summary_definition == self.source_definition
-        ) {
-            return Ok(None);
-        }
+        let definition = match self.binding.node(node.id) {
+            Some(BackendNodeBinding::Materialization { summary_definition }) => *summary_definition,
+            _ => return Ok(None),
+        };
         let family = node.output_schema.fields.iter().find_map(|field| {
             (!matches!(
                 field.dtype,
@@ -86,16 +88,34 @@ impl PrecomputeOperatorRegistry<MaintenanceValue> for OperatorAdapter<'_> {
             ))
             .then(|| field.dtype.clone())
         });
-        if let Some(states) = &self.immutable_windows {
-            return Ok(Some(MaintenanceValue::SummaryWindows {
-                states: Arc::clone(states),
-                family: family.ok_or("immutable source lacks a summary schema")?,
-            }));
+        match &self.inputs {
+            MaintenanceInputs::Live {
+                definition: source,
+                state,
+            } if definition == *source => Ok(Some(MaintenanceValue::Summary {
+                state: Arc::clone(state),
+                family,
+            })),
+            MaintenanceInputs::Live { .. } => Ok(None),
+            MaintenanceInputs::Frozen(inputs) => {
+                let mut matching = inputs.iter().filter(|input| input.definition == definition);
+                let Some(input) = matching.next() else {
+                    return Ok(None);
+                };
+                if matching.next().is_some() {
+                    return Err("maintenance frontier requires explicit population routing".into());
+                }
+                Ok(Some(MaintenanceValue::SummaryWindows {
+                    states: input
+                        .windows
+                        .iter()
+                        .map(|((_, end), state)| (*end as i64, Arc::clone(state)))
+                        .collect::<Vec<_>>()
+                        .into(),
+                    family: family.ok_or("immutable source lacks a summary schema")?,
+                }))
+            }
         }
-        Ok(Some(MaintenanceValue::Summary {
-            state: Arc::clone(&self.source),
-            family,
-        }))
     }
 
     fn execute(
@@ -109,7 +129,7 @@ impl PrecomputeOperatorRegistry<MaintenanceValue> for OperatorAdapter<'_> {
                 operation: planner_types::post_asap::ValueOperation::FinalizeExactAccumulator,
                 timing: planner_types::post_asap::ExecutionTiming::MaintenanceTime,
             } => {
-                if self.immutable_windows.is_none() {
+                if !matches!(&self.inputs, MaintenanceInputs::Frozen(_)) {
                     return Err(
                         "maintenance finalization requires immutable completed input windows"
                             .into(),
@@ -124,7 +144,7 @@ impl PrecomputeOperatorRegistry<MaintenanceValue> for OperatorAdapter<'_> {
                 let MaintenanceValue::Rows { values, name } = value.as_ref() else {
                     return Err("maintenance SummaryAgg requires a typed update evaluator; finalize summary state before applying an update".into());
                 };
-                if self.immutable_windows.is_none() {
+                if !matches!(&self.inputs, MaintenanceInputs::Frozen(_)) {
                     return Err(
                         "maintenance aggregation requires immutable completed input windows".into(),
                     );
@@ -144,19 +164,24 @@ impl PrecomputeOperatorRegistry<MaintenanceValue> for OperatorAdapter<'_> {
                     .iter()
                     .find(|config| config.policy_fingerprint() == target.fingerprint())
                     .ok_or("maintenance SummaryAgg lacks installed accumulator configuration")?;
-                let source_config = self
-                    .configs
-                    .iter()
-                    .find(|config| {
-                        config.policy_fingerprint() == self.source_definition.fingerprint()
-                    })
-                    .ok_or("maintenance input lacks installed source configuration")?;
-                validate_maintenance_grouping(
-                    config,
-                    source_config,
-                    node,
-                    self.singleton_population_complete,
-                )?;
+                let MaintenanceInputs::Frozen(sources) = &self.inputs else {
+                    return Err("maintenance aggregation requires frozen sources".into());
+                };
+                for source in *sources {
+                    let source_config = self
+                        .configs
+                        .iter()
+                        .find(|config| {
+                            config.policy_fingerprint() == source.definition.fingerprint()
+                        })
+                        .ok_or("maintenance input lacks installed source configuration")?;
+                    validate_maintenance_grouping(
+                        config,
+                        source_config,
+                        node,
+                        source.singleton_population_complete,
+                    )?;
+                }
                 if config.accumulator_spec().map_err(|e| e.to_string())?.family != *family {
                     return Err(
                         "maintenance SummaryAgg family differs from installed configuration".into(),
@@ -449,13 +474,12 @@ fn prepare_frozen_maintenance_sink(
     installed: &asap_types::executable_plan::InstalledPostAsapDag,
     configs: &[asap_types::PrecomputeMaterialization],
     sink: PostAsapNodeId,
-    input: &crate::storage_engines::sketch_db::index::FrozenExactWindows,
+    inputs: &[crate::storage_engines::sketch_db::index::FrozenExactWindows],
     output_window: (u64, u64),
 ) -> Result<
     (
         planner_types::post_asap::ExecutableDag,
         MaterializationCommitKey,
-        Vec<(i64, SummaryState)>,
     ),
     String,
 > {
@@ -472,15 +496,17 @@ fn prepare_frozen_maintenance_sink(
         .derived_input
         .as_ref()
         .ok_or("immutable sink is not a derived materialization")?;
-    if expected_input.inputs != BTreeSet::from([input.definition]) {
+    if expected_input.inputs != inputs.iter().map(|input| input.definition).collect() {
         return Err("immutable sink requires synchronized input definitions".into());
     }
     if output_window.0 >= output_window.1
         || output_window.1 - output_window.0 != config.stored_window_ms()
-        || input
-            .windows
-            .keys()
-            .any(|(start, end)| *start < output_window.0 || *end > output_window.1)
+        || inputs.iter().any(|input| {
+            input
+                .windows
+                .keys()
+                .any(|(start, end)| *start < output_window.0 || *end > output_window.1)
+        })
     {
         return Err("immutable inputs do not fit the installed output window".into());
     }
@@ -499,7 +525,7 @@ fn prepare_frozen_maintenance_sink(
         .iter()
         .filter_map(|(node, binding)| match binding {
             BackendNodeBinding::Materialization { summary_definition }
-                if *summary_definition == input.definition =>
+                if expected_input.inputs.contains(summary_definition) =>
             {
                 Some((*node, *summary_definition))
             }
@@ -514,15 +540,14 @@ fn prepare_frozen_maintenance_sink(
     if &actual_input != expected_input {
         return Err("immutable sink input program differs from its catalog identity".into());
     }
-    let digest = frozen_cohort_lineage(std::slice::from_ref(input), expected_input)?;
-    let states = input
-        .windows
-        .iter()
-        .map(|((_, end), state)| (*end as i64, Arc::clone(state)))
-        .collect();
+    let digest = frozen_cohort_lineage(inputs, expected_input)?;
+    let generation = &inputs
+        .first()
+        .ok_or("immutable input cohort is empty")?
+        .generation;
     let key = MaterializationCommitKey {
-        plan_id: input.generation.plan_id,
-        plan_version: input.generation.plan_version,
+        plan_id: generation.plan_id,
+        plan_version: generation.plan_version,
         summary_definition: target,
         window_start_ms: i64::try_from(output_window.0)
             .map_err(|_| "output window exceeds timestamp range")?,
@@ -530,26 +555,21 @@ fn prepare_frozen_maintenance_sink(
             .map_err(|_| "output window exceeds timestamp range")?,
         input_lineage: digest.to_vec(),
     };
-    Ok((dag, key, states))
+    Ok((dag, key))
 }
 
 fn execute_prepared_frozen_sink(
     installed: &asap_types::executable_plan::InstalledPostAsapDag,
     configs: &[asap_types::PrecomputeMaterialization],
     sink: PostAsapNodeId,
-    input: &crate::storage_engines::sketch_db::index::FrozenExactWindows,
+    inputs: &[crate::storage_engines::sketch_db::index::FrozenExactWindows],
     dag: &planner_types::post_asap::ExecutableDag,
     key: MaterializationCommitKey,
-    states: Vec<(i64, SummaryState)>,
 ) -> Result<SummaryState, String> {
-    let first = states.first().ok_or("immutable input is empty")?;
     let adapter = OperatorAdapter {
         binding: &installed.binding,
-        source_definition: input.definition,
-        source: Arc::clone(&first.1),
+        inputs: MaintenanceInputs::Frozen(inputs),
         configs,
-        immutable_windows: Some(states.into()),
-        singleton_population_complete: input.singleton_population_complete,
     };
     let value = execute_precompute_sink(
         dag,
@@ -571,14 +591,26 @@ fn evaluate_frozen_maintenance_sink(
     input: &crate::storage_engines::sketch_db::index::FrozenExactWindows,
     output_window: (u64, u64),
 ) -> Result<(SummaryState, [u8; 32]), String> {
-    let (dag, key, states) =
-        prepare_frozen_maintenance_sink(installed, configs, sink, input, output_window)?;
+    let (dag, key) = prepare_frozen_maintenance_sink(
+        installed,
+        configs,
+        sink,
+        std::slice::from_ref(input),
+        output_window,
+    )?;
     let digest = key
         .input_lineage
         .as_slice()
         .try_into()
         .map_err(|_| "invalid input digest")?;
-    let state = execute_prepared_frozen_sink(installed, configs, sink, input, &dag, key, states)?;
+    let state = execute_prepared_frozen_sink(
+        installed,
+        configs,
+        sink,
+        std::slice::from_ref(input),
+        &dag,
+        key,
+    )?;
     Ok((state, digest))
 }
 
@@ -659,8 +691,13 @@ pub fn execute_completed_maintenance(
         &[(source_sid, source, expected, group.clone())],
     )?;
     let frozen = cohort.pop().ok_or("immutable input cohort is empty")?;
-    let (dag, key, states) =
-        prepare_frozen_maintenance_sink(installed, configs, sink, &frozen, window)?;
+    let (dag, key) = prepare_frozen_maintenance_sink(
+        installed,
+        configs,
+        sink,
+        std::slice::from_ref(&frozen),
+        window,
+    )?;
     let digest = key
         .input_lineage
         .as_slice()
@@ -686,7 +723,14 @@ pub fn execute_completed_maintenance(
     )? {
         return Ok(false);
     }
-    let state = execute_prepared_frozen_sink(installed, configs, sink, &frozen, &dag, key, states)?;
+    let state = execute_prepared_frozen_sink(
+        installed,
+        configs,
+        sink,
+        std::slice::from_ref(&frozen),
+        &dag,
+        key,
+    )?;
     let mut output = crate::storage_engines::types::PrecomputedOutput::new(
         window.0,
         window.1,
@@ -1124,11 +1168,11 @@ impl MaintenanceDagSink {
             }
             let adapter = OperatorAdapter {
                 binding: &installed.binding,
-                source_definition,
-                source: Arc::clone(&source),
+                inputs: MaintenanceInputs::Live {
+                    definition: source_definition,
+                    state: Arc::clone(&source),
+                },
                 configs: &plan.precompute_plan.materializations,
-                immutable_windows: None,
-                singleton_population_complete: false,
             };
             for sink_node in &installed.binding.precompute_sinks {
                 // Derived summaries consume complete immutable windows at the
@@ -1498,6 +1542,87 @@ mod tests {
         Arc::new(accumulator)
     }
 
+    #[test]
+    fn frozen_adapter_resolves_each_materialized_frontier_without_aliasing() {
+        use planner_types::post_asap::{ExactKind, ExactParams, SummaryFamilyType, SummaryField};
+        let make = |id| crate::storage_engines::sketch_db::index::FrozenExactWindows {
+            sid: id,
+            definition: definition(id),
+            generation: Arc::new(asap_types::sds::CatalogGeneration {
+                schema_version: 2,
+                plan_id: 1,
+                plan_version: 1,
+                snapshot_sha256: "0".repeat(64),
+            }),
+            group: BTreeMap::new(),
+            windows: BTreeMap::from([((0, 1000), sum(id as f64))]),
+            singleton_population_complete: false,
+        };
+        let inputs = [make(1), make(2)];
+        let binding = BackendExecutableBinding {
+            nodes: [(1, definition(1)), (2, definition(2)), (3, definition(3))]
+                .into_iter()
+                .map(|(id, summary_definition)| {
+                    (
+                        PostAsapNodeId(id),
+                        BackendNodeBinding::Materialization { summary_definition },
+                    )
+                })
+                .collect(),
+            query_sink: PostAsapNodeId(3),
+            query_plan_sink: asap_types::query_plan::QueryNodeId(3),
+            precompute_sinks: vec![PostAsapNodeId(3)],
+        };
+        let adapter = OperatorAdapter {
+            binding: &binding,
+            inputs: MaintenanceInputs::Frozen(&inputs),
+            configs: &[],
+        };
+        let source_node = |id| {
+            let mut source = node(id);
+            source.output_schema.fields = vec![SummaryField {
+                name: "state".into(),
+                dtype: SummaryFamilyType::ExactAggregate(ExactKind::Sum, ExactParams::Sum),
+                nullable: false,
+            }];
+            source
+        };
+        let first = adapter
+            .materialized_input(&source_node(1))
+            .unwrap()
+            .unwrap();
+        let second = adapter
+            .materialized_input(&source_node(2))
+            .unwrap()
+            .unwrap();
+        assert!(adapter
+            .materialized_input(&source_node(3))
+            .unwrap()
+            .is_none());
+        let merged = adapter
+            .execute(&node(3), &[Arc::new(first), Arc::new(second)])
+            .unwrap();
+        assert_eq!(
+            merged
+                .state()
+                .unwrap()
+                .query_statistic(
+                    asap_types::Statistic::Sum,
+                    &None,
+                    &std::collections::HashMap::new()
+                )
+                .unwrap(),
+            3.0
+        );
+        let ambiguous = [make(1), make(1)];
+        let adapter = OperatorAdapter {
+            binding: &binding,
+            inputs: MaintenanceInputs::Frozen(&ambiguous),
+            configs: &[],
+        };
+        assert!(adapter.materialized_input(&source_node(1)).is_err());
+    }
+
     #[test]
     fn finalized_summary_update_builds_a_different_installed_family() {
         use planner_types::post_asap::{
@@ -1551,13 +1676,25 @@ mod tests {
             query_plan_sink: control_plane::query_plan::QueryNodeId(3),
             precompute_sinks: vec![PostAsapNodeId(3)],
         };
+        let frozen_inputs = [
+            crate::storage_engines::sketch_db::index::FrozenExactWindows {
+                sid: 1,
+                definition: source_definition,
+                generation: Arc::new(asap_types::sds::CatalogGeneration {
+                    schema_version: 2,
+                    plan_id: 1,
+                    plan_version: 1,
+                    snapshot_sha256: "0".repeat(64),
+                }),
+                group: BTreeMap::new(),
+                windows: BTreeMap::from([((0, 1000), sum(7.0))]),
+                singleton_population_complete: false,
+            },
+        ];
         let adapter = OperatorAdapter {
             binding: &binding,
-            source_definition,
-            source: sum(7.0),
+            inputs: MaintenanceInputs::Frozen(&frozen_inputs),
             configs: &configs,
-            immutable_windows: Some(vec![(1000, sum(7.0))].into()),
-            singleton_population_complete: false,
         };
         let mut read = node(2);
         read.payload = ExecutableOperatorPayload::Value {
@@ -2106,11 +2243,11 @@ mod tests {
         };
         let adapter = OperatorAdapter {
             binding: &binding,
-            source_definition: definition(1),
-            source: sum(7.0),
+            inputs: MaintenanceInputs::Live {
+                definition: definition(1),
+                state: sum(7.0),
+            },
             configs: &[],
-            immutable_windows: None,
-            singleton_population_complete: false,
         };
         let mut aggregate = node(1);
         aggregate.operator = ExecutableOperator::SummaryAgg;
@@ -2431,11 +2568,11 @@ mod tests {
         let source = sum(2.0);
         let adapter = OperatorAdapter {
             binding: &binding,
-            source_definition: definition(1),
-            source,
+            inputs: MaintenanceInputs::Live {
+                definition: definition(1),
+                state: source,
+            },
             configs: &[],
-            immutable_windows: None,
-            singleton_population_complete: false,
         };
         let commits = CommitRegistry::default();
         let key = MaterializationCommitKey {
@@ -2506,11 +2643,11 @@ mod tests {
         };
         let adapter = OperatorAdapter {
             binding: &binding,
-            source_definition: definition(1),
-            source: sum(2.0),
+            inputs: MaintenanceInputs::Live {
+                definition: definition(1),
+                state: sum(2.0),
+            },
             configs: &[],
-            immutable_windows: None,
-            singleton_population_complete: false,
         };
         let commits = CommitRegistry::default();
         let key = MaterializationCommitKey {

From c71c801cfee5a8dab577e4a33f9f5c7fbc4d17c9 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Fri, 11 Sep 2026 12:22:55 -0600
Subject: [PATCH 8/8] Execute complete aligned source cohorts through the
 installed maintenance DAG

---
 .../precompute_engine/maintenance_runtime.rs  | 425 +++++++++++++++---
 1 file changed, 359 insertions(+), 66 deletions(-)

diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs
index 9e0c7467..0cfbcdb7 100644
--- a/data_plane/src/precompute_engine/maintenance_runtime.rs
+++ b/data_plane/src/precompute_engine/maintenance_runtime.rs
@@ -629,6 +629,43 @@ pub fn execute_completed_maintenance(
     target_sid: u64,
     window: (u64, u64),
     group: &BTreeMap<String, String>,
+) -> Result<bool, String> {
+    let target = match installed.binding.node(sink) {
+        Some(BackendNodeBinding::Materialization { summary_definition }) => *summary_definition,
+        _ => return Err("maintenance sink lacks an installed output identity".into()),
+    };
+    let derived = configs
+        .iter()
+        .find(|config| config.policy_fingerprint() == target.fingerprint())
+        .and_then(|config| config.derived_input.as_ref())
+        .ok_or("maintenance output has no derived input")?;
+    if derived.inputs.len() != 1 {
+        return Err("maintenance execution requires synchronized multi-source scheduling".into());
+    }
+    execute_completed_maintenance_cohort(
+        store,
+        installed,
+        configs,
+        sink,
+        &BTreeMap::from([(*derived.inputs.first().unwrap(), source_sid)]),
+        target_sid,
+        window,
+        group,
+    )
+}
+
+/// Execute one completed population per input definition at matching full
+/// windows. All source populations must share the same explicit label map.
+#[allow(clippy::too_many_arguments)]
+pub(crate) fn execute_completed_maintenance_cohort(
+    store: &crate::storage_engines::sketch_db::index::SketchStore,
+    installed: &asap_types::executable_plan::InstalledPostAsapDag,
+    configs: &[asap_types::PrecomputeMaterialization],
+    sink: PostAsapNodeId,
+    source_sids: &BTreeMap<asap_types::sds::SummaryDefinitionId, u64>,
+    target_sid: u64,
+    window: (u64, u64),
+    group: &BTreeMap<String, String>,
 ) -> Result<bool, String> {
     use asap_types::executable_plan::BackendNodeBinding;
     let target = match installed.binding.node(sink) {
@@ -643,61 +680,73 @@ pub fn execute_completed_maintenance(
         .derived_input
         .as_ref()
         .ok_or("maintenance output has no derived input")?;
-    if derived.inputs.len() != 1 {
-        return Err("maintenance execution requires synchronized multi-source scheduling".into());
-    }
-    let source = *derived.inputs.first().unwrap();
-    let source_config = configs
-        .iter()
-        .find(|config| config.policy_fingerprint() == source.fingerprint())
-        .ok_or("maintenance source configuration is absent")?;
-    let source_width = source_config.stored_window_ms();
-    let target_width = target_config.stored_window_ms();
-    let origin = source_config.pane_origin_ms.unwrap_or(0);
-    if source_width == 0
-        || target_width == 0
-        || window.0 >= window.1
-        || window.1 - window.0 != target_width
-        || target_width % source_width != 0
-        || target_width / source_width > 65_536
-        || source_config
-            .slide_interval
-            .checked_mul(1000)
-            .is_none_or(|slide| slide < source_width)
-        || target_config
-            .slide_interval
-            .checked_mul(1000)
-            .is_none_or(|slide| slide < target_width)
-        || window.1 > i64::MAX as u64
-        || (window.0 as i128 - origin as i128).rem_euclid(source_width as i128) != 0
-        || (window.0 as i128 - target_config.pane_origin_ms.unwrap_or(0) as i128)
-            .rem_euclid(target_width as i128)
-            != 0
-    {
-        return Err("maintenance window requires unsupported overlap, phase, or extent".into());
+    if derived.inputs != source_sids.keys().copied().collect() {
+        return Err("maintenance source set differs from installed input definitions".into());
     }
-    let expected = (0..target_width / source_width)
-        .map(|index| {
-            let start = window.0 + index * source_width;
-            (start, start + source_width)
+    let source_configs = source_sids
+        .keys()
+        .map(|source| {
+            configs
+                .iter()
+                .find(|config| config.policy_fingerprint() == source.fingerprint())
+                .ok_or("maintenance source configuration is absent")
         })
-        .collect();
+        .collect::<Result<Vec<_>, _>>()?;
+    if source_configs.len() > 1 {
+        asap_types::precompute_plan::validated_source_window_cohort(target_config, &source_configs)
+            .map_err(|error| error.to_string())?;
+    }
+    let mut requests = Vec::with_capacity(source_sids.len());
+    for source_config in &source_configs {
+        let source = source_config.policy_fingerprint().into();
+        let source_sid = source_sids[&source];
+        let source_width = source_config.stored_window_ms();
+        let target_width = target_config.stored_window_ms();
+        let origin = source_config.pane_origin_ms.unwrap_or(0);
+        if source_width == 0
+            || target_width == 0
+            || window.0 >= window.1
+            || window.1 - window.0 != target_width
+            || target_width % source_width != 0
+            || target_width / source_width > 65_536
+            || source_config
+                .slide_interval
+                .checked_mul(1000)
+                .is_none_or(|slide| slide < source_width)
+            || target_config
+                .slide_interval
+                .checked_mul(1000)
+                .is_none_or(|slide| slide < target_width)
+            || window.1 > i64::MAX as u64
+            || (window.0 as i128 - origin as i128).rem_euclid(source_width as i128) != 0
+            || (window.0 as i128 - target_config.pane_origin_ms.unwrap_or(0) as i128)
+                .rem_euclid(target_width as i128)
+                != 0
+        {
+            return Err("maintenance window requires unsupported overlap, phase, or extent".into());
+        }
+        let expected = (0..target_width / source_width)
+            .map(|index| {
+                let start = window.0 + index * source_width;
+                (start, start + source_width)
+            })
+            .collect();
+        requests.push((source_sid, source, expected, group.clone()));
+    }
     let generation = store
         .active_catalog_generation()
         .ok_or("maintenance requires an authoritative catalog")?;
-    let mut cohort = store.read_frozen_exact_cohort(
-        &generation,
-        &derived.inputs,
-        &[(source_sid, source, expected, group.clone())],
-    )?;
-    let frozen = cohort.pop().ok_or("immutable input cohort is empty")?;
-    let (dag, key) = prepare_frozen_maintenance_sink(
-        installed,
-        configs,
-        sink,
-        std::slice::from_ref(&frozen),
-        window,
-    )?;
+    let cohort = store.read_frozen_exact_cohort(&generation, &derived.inputs, &requests)?;
+    if cohort.len() > 1
+        && cohort
+            .iter()
+            .any(|source| !source.singleton_population_complete)
+    {
+        return Err(
+            "multi-source maintenance requires a complete single population per source".into(),
+        );
+    }
+    let (dag, key) = prepare_frozen_maintenance_sink(installed, configs, sink, &cohort, window)?;
     let digest = key
         .input_lineage
         .as_slice()
@@ -708,29 +757,28 @@ pub fn execute_completed_maintenance(
         .iter()
         .find(|node| node.id == sink)
         .ok_or("maintenance target node is absent")?;
-    validate_maintenance_grouping(
-        target_config,
-        source_config,
-        target_node,
-        frozen.singleton_population_complete,
-    )?;
+    for frozen in &cohort {
+        let source_config = source_configs
+            .iter()
+            .find(|config| config.policy_fingerprint() == frozen.definition.fingerprint())
+            .ok_or("maintenance source configuration is absent")?;
+        validate_maintenance_grouping(
+            target_config,
+            source_config,
+            target_node,
+            frozen.singleton_population_complete,
+        )?;
+    }
     if store.recover_frozen_maintenance_output(
         target_sid,
         target_config,
-        std::slice::from_ref(&frozen),
+        &cohort,
         digest,
         window,
     )? {
         return Ok(false);
     }
-    let state = execute_prepared_frozen_sink(
-        installed,
-        configs,
-        sink,
-        std::slice::from_ref(&frozen),
-        &dag,
-        key,
-    )?;
+    let state = execute_prepared_frozen_sink(installed, configs, sink, &cohort, &dag, key)?;
     let mut output = crate::storage_engines::types::PrecomputedOutput::new(
         window.0,
         window.1,
@@ -754,7 +802,7 @@ pub fn execute_completed_maintenance(
         target_config,
         &output,
         state.as_ref(),
-        std::slice::from_ref(&frozen),
+        &cohort,
         digest,
     )
 }
@@ -2115,6 +2163,7 @@ mod tests {
             .series_ids_for_policy(durable_configs[1].policy_fingerprint())
             .is_empty());
         persistence.shutdown();
+        exercise_two_source_completed_sink(&dag, &configs, &scheduled_binding);
         assert!(evaluate_weight(
             &SummaryInputExpr::Column(planner_types::pre_asap::ColumnRef::Named("missing".into())),
             7.0,
@@ -2123,6 +2172,250 @@ mod tests {
         .is_err());
     }
 
+    fn exercise_two_source_completed_sink(
+        template: &ExecutableDag,
+        configs: &[asap_types::PrecomputeMaterialization],
+        binding: &BackendExecutableBinding,
+    ) {
+        // A bound operator fixture, not a claim that a frontend selected this
+        // composition. Both actual durable sources are required before output.
+        use crate::storage_engines::sketch_db::index::{
+            persistence::config::SketchStorePersistenceConfig, SketchStore,
+        };
+        use asap_types::executable_plan::{InstalledPostAsapDag, OwnedPostAsapDag};
+        let mut first = configs[0].clone();
+        first.window_layout = asap_types::WindowMaterializationLayout::FullWindow;
+        let mut second = first.clone();
+        second.metric = "second_maintenance_source".into();
+        let first_id = first.policy_fingerprint().into();
+        let second_id = second.policy_fingerprint().into();
+        let mut dag = template.clone();
+        let mut second_node = dag
+            .nodes
+            .iter()
+            .find(|node| node.id == PostAsapNodeId(1))
+            .unwrap()
+            .clone();
+        second_node.id = PostAsapNodeId(5);
+        let mut merge = second_node.clone();
+        merge.id = PostAsapNodeId(6);
+        merge.operator = ExecutableOperator::SummaryMerge;
+        merge.payload = ExecutableOperatorPayload::SummaryMerge;
+        dag.nodes.extend([second_node, merge]);
+        let original = dag
+            .edges
+            .iter()
+            .find(|edge| edge.producer == PostAsapNodeId(1) && edge.consumer == PostAsapNodeId(2))
+            .unwrap()
+            .clone();
+        dag.edges.retain(|edge| {
+            !(edge.producer == PostAsapNodeId(1) && edge.consumer == PostAsapNodeId(2))
+        });
+        for (producer, consumer) in [(1, 6), (5, 6), (6, 2)] {
+            let mut edge = original.clone();
+            edge.producer = PostAsapNodeId(producer);
+            edge.consumer = PostAsapNodeId(consumer);
+            dag.edges.push(edge);
+        }
+        if let ExecutableOperatorPayload::SummaryAgg { input, .. } = &mut dag
+            .nodes
+            .iter_mut()
+            .find(|node| node.id == PostAsapNodeId(3))
+            .unwrap()
+            .payload
+        {
+            input.weight = planner_types::post_asap::SummaryInputExpr::Column(
+                planner_types::pre_asap::ColumnRef::SampleValue,
+            );
+        }
+        let document =
+            OwnedPostAsapDag::from_executable("two-source-fixture".into(), &dag).unwrap();
+        let mut target = configs[1].clone();
+        target.derived_input = Some(
+            asap_types::derived_input::DerivedInputIdentity::from_dag(
+                &document,
+                PostAsapNodeId(2),
+                &BTreeMap::from([
+                    (PostAsapNodeId(1), first_id),
+                    (PostAsapNodeId(5), second_id),
+                ]),
+            )
+            .unwrap(),
+        );
+        let mut binding = binding.clone();
+        for (node, summary_definition) in [
+            (1, first_id),
+            (5, second_id),
+            (3, target.policy_fingerprint().into()),
+        ] {
+            binding.nodes.insert(
+                PostAsapNodeId(node),
+                BackendNodeBinding::Materialization { summary_definition },
+            );
+        }
+        binding
+            .nodes
+            .insert(PostAsapNodeId(6), BackendNodeBinding::MaintenanceInput);
+        let installed = InstalledPostAsapDag { document, binding };
+        let configs = [first, second, target];
+        let catalog = Arc::new(
+            asap_types::summary_catalog::SummaryCatalog::from_materializations(2, 1, &configs)
+                .unwrap(),
+        );
+        let directory = tempfile::tempdir().unwrap();
+        let persistence_config = || {
+            let mut config =
+                SketchStorePersistenceConfig::with_memory_limit(1 << 24, directory.path().into());
+            config.delete_older_than_ms = None;
+            config.hot_window_ms = None;
+            config
+        };
+        let store = Arc::new(SketchStore::new());
+        store.install_summary_catalog(Arc::clone(&catalog)).unwrap();
+        let mut persistence = store.start_persistence(persistence_config()).unwrap();
+        let generation = store.active_catalog_generation().unwrap();
+        let sources = BTreeMap::from([(first_id, 800), (second_id, 801)]);
+        for (index, value) in [2.0, 7.0].into_iter().enumerate() {
+            let config = &configs[index];
+            let coordinate = asap_types::sds::SummaryInstanceCoordinates {
+                summary_definition_id: config.policy_fingerprint().into(),
+                time_range: asap_types::sds::HalfOpenTimeRange {
+                    start_ms: 0,
+                    end_ms: 2000,
+                },
+                group_values: BTreeMap::new(),
+            };
+            let revision = store
+                .admit_summary_updates(&generation, BTreeSet::from([coordinate.clone()]))
+                .unwrap();
+            let mut output = PrecomputedOutput::new(0, 2000, None, config.policy_fingerprint());
+            output.catalog_generation = Some(Arc::clone(&generation));
+            store
+                .publish_admitted_summary_update(
+                    &generation,
+                    &coordinate,
+                    revision,
+                    revision,
+                    4000,
+                    |writer| {
+                        writer.ingest_precompute_with_series_id(
+                            800 + index as u64,
+                            config,
+                            &output,
+                            sum(value).as_ref(),
+                        )
+                    },
+                )
+                .unwrap();
+        }
+        assert!(execute_completed_maintenance_cohort(
+            &store,
+            &installed,
+            &configs,
+            PostAsapNodeId(3),
+            &sources,
+            802,
+            (0, 2000),
+            &BTreeMap::new()
+        )
+        .is_err());
+        assert!(store
+            .series_ids_for_policy(configs[2].policy_fingerprint())
+            .is_empty());
+        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+        while !store.seal_finite_summary_input(&generation).unwrap() {
+            assert!(std::time::Instant::now() < deadline);
+            std::thread::sleep(std::time::Duration::from_millis(5));
+        }
+        let requests = sources
+            .iter()
+            .map(|(definition, sid)| {
+                (
+                    *sid,
+                    *definition,
+                    BTreeSet::from([(0, 2000)]),
+                    BTreeMap::new(),
+                )
+            })
+            .collect::<Vec<_>>();
+        let cohort = store
+            .read_frozen_exact_cohort(
+                &generation,
+                &configs[2].derived_input.as_ref().unwrap().inputs,
+                &requests,
+            )
+            .unwrap();
+        let (dag, key) = prepare_frozen_maintenance_sink(
+            &installed,
+            &configs,
+            PostAsapNodeId(3),
+            &cohort,
+            (0, 2000),
+        )
+        .unwrap();
+        let result = execute_prepared_frozen_sink(
+            &installed,
+            &configs,
+            PostAsapNodeId(3),
+            &cohort,
+            &dag,
+            key,
+        )
+        .unwrap();
+        assert_eq!(
+            result
+                .query_statistic(
+                    asap_types::Statistic::Quantile,
+                    &None,
+                    &std::collections::HashMap::from([("quantile".into(), "0.5".into())])
+                )
+                .unwrap(),
+            9.0
+        );
+        assert!(execute_completed_maintenance_cohort(
+            &store,
+            &installed,
+            &configs,
+            PostAsapNodeId(3),
+            &sources,
+            802,
+            (0, 2000),
+            &BTreeMap::new()
+        )
+        .unwrap());
+        let parts = persistence.manifest.live_parts().len();
+        assert!(!execute_completed_maintenance_cohort(
+            &store,
+            &installed,
+            &configs,
+            PostAsapNodeId(3),
+            &sources,
+            802,
+            (0, 2000),
+            &BTreeMap::new()
+        )
+        .unwrap());
+        assert_eq!(persistence.manifest.live_parts().len(), parts);
+        persistence.shutdown();
+        drop(store);
+        let restored = Arc::new(SketchStore::new());
+        restored.install_summary_catalog(catalog).unwrap();
+        let mut persistence = restored.start_persistence(persistence_config()).unwrap();
+        assert!(!execute_completed_maintenance_cohort(
+            &restored,
+            &installed,
+            &configs,
+            PostAsapNodeId(3),
+            &sources,
+            802,
+            (0, 2000),
+            &BTreeMap::new()
+        )
+        .unwrap());
+        assert_eq!(persistence.manifest.live_parts().len(), parts);
+        persistence.shutdown();
+    }
+
     #[test]
     fn finalization_preserves_windows_until_an_explicit_merge() {
         use planner_types::post_asap::{ExactKind, ExactParams, SummaryFamilyType, SummaryField};