diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
index 5004547db..2841bf626 100644
--- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs
+++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
@@ -190,7 +190,17 @@ impl PrometheusRemoteWriteReceiver {
             state.expiry_by_event_time.clear();
             state.max_event_timestamp_ms = None;
         }
+        let generation = self
+            .inner
+            .ingest
+            .physical_plan_snapshot()
+            .and_then(|plan| plan.precompute_plan.summary_catalog.clone())
+            .ok_or("finite completion requires a catalog generation")?;
         self.inner.ingest.router.drain().await?;
+        self.inner
+            .ingest
+            .sketch_index
+            .seal_finite_summary_input(&generation)?;
         trim_process_allocator();
         Ok(())
     }
@@ -322,10 +332,93 @@ impl PrometheusRemoteWriteReceiver {
         }
 
         let messages = route_messages(&new_samples, &self.inner.ingest, &physical_plan);
+        let generation = Arc::new(
+            physical_plan
+                .precompute_plan
+                .summary_catalog
+                .clone()
+                .ok_or(RemoteWriteError::InactivePhysicalPlan)?,
+        );
+        let snapshot = self.inner.ingest.hot_reload_config.snapshot();
+        let mut coordinates = std::collections::BTreeSet::new();
+        for message in &messages {
+            let WorkerMessage::GroupSamples {
+                policy_fp,
+                group_key,
+                samples,
+                ..
+            } = message
+            else {
+                continue;
+            };
+            let config = snapshot
+                .get_aggregation_config(policy_fp.as_u64())
+                .ok_or(RemoteWriteError::InactivePhysicalPlan)?;
+            let manager = crate::precompute_engine::window_manager::WindowManager::with_layout(
+                config.window_size,
+                config.slide_interval,
+                config.pane_origin_ms,
+                &config.window_layout,
+            );
+            let mut labels = group_key.as_population_labels();
+            if labels.is_empty() {
+                labels = config
+                    .grouping_labels
+                    .labels
+                    .iter()
+                    .cloned()
+                    .zip(group_key.values().labels)
+                    .collect();
+            }
+            let right_closed = config
+                .parameters
+                .get("promql_right_closed")
+                .and_then(serde_json::Value::as_bool)
+                .unwrap_or(false);
+            let affected = crate::precompute_engine::maintenance_runtime::affected_materializations(
+                &physical_plan.precompute_plan,
+                (*policy_fp).into(),
+            );
+            let mut starts = std::collections::BTreeSet::new();
+            for (_, timestamp, _) in samples {
+                let timestamp = if right_closed {
+                    timestamp.saturating_sub(1)
+                } else {
+                    *timestamp
+                };
+                starts.extend(manager.stored_bucket_starts(timestamp));
+            }
+            for start in starts {
+                let (start_ms, end_ms) = manager.stored_bucket_bounds(start);
+                for summary_definition_id in &affected {
+                    coordinates.insert(asap_types::sds::SummaryInstanceCoordinates {
+                        summary_definition_id: *summary_definition_id,
+                        time_range: asap_types::sds::HalfOpenTimeRange { start_ms, end_ms },
+                        group_values: labels.clone(),
+                    });
+                }
+            }
+        }
         self.inner
             .ingest
             .router
-            .try_route_group_batch_atomic(messages)?;
+            .try_route_group_batch_with_admission(messages, || {
+                if coordinates.is_empty() {
+                    return Ok(None);
+                }
+                let revision = self
+                    .inner
+                    .ingest
+                    .sketch_index
+                    .admit_summary_updates(&generation, coordinates)?;
+                Ok(Some(Arc::new(
+                    crate::storage_engines::types::SummaryInputRevision {
+                        generation,
+                        revision,
+                        first_revision: revision,
+                    },
+                )))
+            })?;
 
         // Commit dedup mutation only after the entire routed batch was
         // reserved successfully. A rejected/backpressured request must not
@@ -349,10 +442,12 @@ impl PrometheusRemoteWriteReceiver {
             .iter()
             .filter(|sample| sample.value.is_none())
             .count() as u64;
-        self.inner.stats.samples.fetch_add(
-            (new_samples.len() as u64).saturating_sub(stale_count),
-            Ordering::Relaxed,
-        );
+        let accepted_numeric_samples = (new_samples.len() as u64).saturating_sub(stale_count);
+        self.inner
+            .stats
+            .samples
+            .fetch_add(accepted_numeric_samples, Ordering::Relaxed);
+        crate::precompute_engine::metrics::record_accepted_samples(accepted_numeric_samples);
         self.inner
             .stats
             .stale_markers
@@ -786,11 +881,27 @@ mod tests {
             planner_revision: PLANNER_REVISION.into(),
             capability_snapshot_id: "test".into(),
         };
+        let configs = streaming
+            .aggregation_configs
+            .values()
+            .cloned()
+            .collect::<Vec<_>>();
+        let catalog = Arc::new(
+            asap_types::summary_catalog::SummaryCatalog::from_materializations(7, 3, &configs)
+                .unwrap(),
+        );
+        let reference = catalog.reference().unwrap();
+        let generation = asap_types::sds::CatalogGeneration {
+            schema_version: reference.schema_version,
+            plan_id: reference.plan_id,
+            plan_version: reference.plan_version,
+            snapshot_sha256: reference.snapshot_sha256,
+        };
         let active = ActivePhysicalPlan {
             envelope: envelope.clone(),
-            summary_catalog: None,
+            summary_catalog: Some(catalog),
             precompute_plan: PrecomputePlan {
-                summary_catalog: None,
+                summary_catalog: Some(generation),
                 envelope: envelope.clone(),
                 ingest: IngestContract {
                     protocol: IngestProtocol::PrometheusRemoteWriteV1,
@@ -879,6 +990,18 @@ mod tests {
             sketch_index: Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()),
             observability: IngestObservability::default(),
         });
+        ingest
+            .sketch_index
+            .install_summary_catalog(
+                ingest
+                    .physical_plan_snapshot()
+                    .unwrap()
+                    .summary_catalog
+                    .as_ref()
+                    .unwrap()
+                    .clone(),
+            )
+            .unwrap();
         (
             PrometheusRemoteWriteReceiver::new(PrometheusRemoteWriteConfig::default(), ingest),
             receiver,
@@ -896,10 +1019,13 @@ mod tests {
                 aggregation_sub_type: String::new(),
                 parameters: match aggregation_type {
                     AggregationType::CountMinSketchWithHeap => HashMap::from([
-                        ("width".into(), serde_json::json!(128)),
-                        ("depth".into(), serde_json::json!(5)),
+                        ("w".into(), serde_json::json!(128)),
+                        ("d".into(), serde_json::json!(5)),
                         ("heap_size".into(), serde_json::json!(2)),
                     ]),
+                    AggregationType::DatasketchesKLL => {
+                        HashMap::from([("k".into(), serde_json::json!(200))])
+                    }
                     _ => HashMap::new(),
                 },
                 grouping_labels: KeyByLabelNames::new(grouping),
@@ -1078,7 +1204,7 @@ mod tests {
         let drain = tokio::spawn(async move { handle.drain().await });
         assert!(matches!(
             worker.recv().await.unwrap(),
-            WorkerMessage::GroupSamples { .. }
+            WorkerMessage::Admitted { .. }
         ));
         let WorkerMessage::Drain(reply) = worker.recv().await.unwrap() else {
             panic!("expected barrier")
@@ -1260,11 +1386,131 @@ mod tests {
         assert_eq!(receiver.stats().duplicates.load(Ordering::Relaxed), 1);
     }
 
+    #[tokio::test]
+    async fn queued_population_blocks_partial_warm_read_until_both_workers_publish() {
+        use crate::precompute_engine::{
+            config::LateDataPolicy,
+            output_sink::SketchStoreSink,
+            worker::{Worker, WorkerRuntimeConfig},
+        };
+        use crate::query_engines::asap_query_engine::summary_executor::{
+            QueryExecutionContext, SummaryExecutorError,
+        };
+        use std::sync::atomic::{AtomicI64, AtomicUsize};
+        let (receiver, mut queued) = configured_receiver();
+        let request = WriteRequest {
+            timeseries: ["a", "b"]
+                .into_iter()
+                .map(|job| TimeSeries {
+                    labels: vec![
+                        Label {
+                            name: "__name__".into(),
+                            value: "requests_total".into(),
+                        },
+                        Label {
+                            name: "job".into(),
+                            value: job.into(),
+                        },
+                    ],
+                    samples: vec![Sample {
+                        timestamp: 100,
+                        value: 4.0,
+                    }],
+                    exemplars: vec![],
+                    histograms: vec![],
+                })
+                .collect(),
+        };
+        receiver.accept(&compressed(request)).unwrap();
+        let first = queued.recv().await.unwrap();
+        let second = queued.recv().await.unwrap();
+        let ingest = &receiver.inner.ingest;
+        let sink = Arc::new(SketchStoreSink::new(
+            ingest.sketch_index.clone(),
+            ingest.hot_reload_config.clone(),
+            ingest.series_resolver.clone(),
+        ));
+        let start_worker = |id| {
+            let (tx, rx) = mpsc::channel(4);
+            let worker = Worker::new(
+                id,
+                rx,
+                sink.clone(),
+                ingest.hot_reload_config.clone(),
+                WorkerRuntimeConfig {
+                    max_buffer_per_series: 100,
+                    allowed_lateness_ms: 0,
+                    pass_raw_samples: false,
+                    raw_mode_aggregation_id: 0,
+                    late_data_policy: LateDataPolicy::Drop,
+                    wall_clock_idle_grace_period_ms: i64::MAX,
+                    wall_clock_max_open_grace_period_ms: i64::MAX,
+                },
+                Arc::new(AtomicUsize::new(0)),
+                Arc::new(AtomicI64::new(i64::MIN)),
+            );
+            (tx, tokio::spawn(worker.run()))
+        };
+        let (fast, fast_task) = start_worker(0);
+        let (slow, slow_task) = start_worker(1);
+        fast.send(first).await.unwrap();
+        let (done, result) = tokio::sync::oneshot::channel();
+        fast.send(WorkerMessage::Drain(done)).await.unwrap();
+        result.await.unwrap().unwrap();
+        let policy = *ingest
+            .hot_reload_config
+            .snapshot()
+            .aggregation_configs
+            .keys()
+            .next()
+            .unwrap();
+        let binding = control_plane::query_plan::MaterializationBinding {
+            materialization: asap_types::PolicyFingerprint(policy).into(),
+            output_grouping: control_plane::query_plan::PhysicalGrouping::Reduce(
+                vec!["job".into()],
+            ),
+            item_labels: vec![],
+            window_ms: 60_000,
+            pane_origin_ms: Some(0),
+            readout_lookback_ms: None,
+        };
+        let context = QueryExecutionContext {
+            index: &ingest.sketch_index,
+            t0_ms: 0,
+            t1_ms: 60_000,
+            is_cumulative: true,
+            allowed_materializations: None,
+        };
+        assert!(matches!(
+            context.read_bound_materialization(&binding),
+            Err(SummaryExecutorError::Unsupported(
+                "materialization population has unpublished input"
+            ))
+        ));
+        slow.send(second).await.unwrap();
+        let (done, result) = tokio::sync::oneshot::channel();
+        slow.send(WorkerMessage::Drain(done)).await.unwrap();
+        result.await.unwrap().unwrap();
+        assert_eq!(
+            context.read_bound_materialization(&binding).unwrap().len(),
+            2
+        );
+        fast.send(WorkerMessage::Shutdown).await.unwrap();
+        slow.send(WorkerMessage::Shutdown).await.unwrap();
+        fast_task.await.unwrap();
+        slow_task.await.unwrap();
+    }
+
     #[tokio::test]
     async fn valid_request_routes_canonical_sample_to_installed_plan() {
         let (receiver, mut worker) = configured_receiver();
         receiver.accept(&one_sample(4.0)).unwrap();
         let message = worker.recv().await.expect("routed worker message");
+        let WorkerMessage::Admitted { input, revision } = message else {
+            panic!("missing admission receipt")
+        };
+        assert!(revision.revision > 0);
+        let message = *input;
         let WorkerMessage::GroupSamples {
             group_key, samples, ..
         } = message
diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs
index c751ddf9f..345479b29 100644
--- a/data_plane/src/main.rs
+++ b/data_plane/src/main.rs
@@ -672,6 +672,14 @@ async fn main() -> Result<()> {
         Arc::new(data_plane::drivers::ingest::series_resolver::SeriesIdResolver::new())
     };
     let sketch_index = Arc::new(data_plane::storage_engines::sketch_db::index::SketchStore::new());
+    if let Some(catalog) = startup_physical_plan
+        .as_ref()
+        .and_then(|plan| plan.summary_catalog.as_ref())
+    {
+        sketch_index
+            .install_summary_catalog(Arc::clone(catalog))
+            .map_err(std::io::Error::other)?;
+    }
 
     // M2.3.6c — also start a persistence layer behind the SketchStore
     // when --persistence-enabled. SketchStore is now where all
diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs
index 7a284b480..d94911ce9 100644
--- a/data_plane/src/precompute_engine/maintenance_runtime.rs
+++ b/data_plane/src/precompute_engine/maintenance_runtime.rs
@@ -80,6 +80,7 @@ struct CommitRegistryState {
     frontiers: BTreeMap<asap_types::sds::SummaryDefinitionId, (i64, u64)>,
     pending_batch: Option<[u8; 32]>,
     batch_has_published: bool,
+    admitted_keys: BTreeSet<MaterializationCommitKey>,
 }
 
 impl CommitRegistryState {
@@ -90,13 +91,14 @@ impl CommitRegistryState {
         {
             return Err("maintenance retry belongs to an obsolete plan generation".into());
         }
-        if self
-            .frontiers
-            .get(&key.summary_definition)
-            .is_some_and(|(latest, horizon)| {
-                key.window_end_ms
-                    <= latest.saturating_sub(i64::try_from(*horizon).unwrap_or(i64::MAX))
-            })
+        if !self.admitted_keys.contains(key)
+            && self
+                .frontiers
+                .get(&key.summary_definition)
+                .is_some_and(|(latest, horizon)| {
+                    key.window_end_ms
+                        <= latest.saturating_sub(i64::try_from(*horizon).unwrap_or(i64::MAX))
+                })
         {
             return Err(
                 "maintenance retry is outside the materialization retention horizon".into(),
@@ -123,6 +125,7 @@ impl CommitRegistry {
             .map(|plan| (plan.plan_id(), plan.plan_version()));
         if state.generation != generation {
             state.entries.clear();
+            state.admitted_keys.clear();
             state.frontiers.clear();
             state.pending_batch = None;
             state.batch_has_published = false;
@@ -157,6 +160,7 @@ impl CommitRegistry {
             if !state.batch_has_published {
                 state.entries.retain(|_, entry| entry.published);
                 state.pending_batch = None;
+                state.admitted_keys.clear();
             }
         }
     }
@@ -195,10 +199,22 @@ impl CommitRegistry {
                 })
         });
         state.pending_batch = None;
+        state.admitted_keys.clear();
         state.batch_has_published = false;
         Ok(())
     }
 
+    fn pin_admitted(&self, key: &MaterializationCommitKey) -> Result<(), String> {
+        let mut state = self.0.lock().map_err(|_| "commit registry poisoned")?;
+        if state.pending_batch.is_none()
+            || state.generation != Some((key.plan_id, key.plan_version))
+        {
+            return Err("admitted maintenance output has no current batch".into());
+        }
+        state.admitted_keys.insert(key.clone());
+        Ok(())
+    }
+
     fn is_published(&self, key: &MaterializationCommitKey) -> Result<bool, String> {
         let state = self.0.lock().map_err(|_| "commit registry poisoned")?;
         state.validate_key(key)?;
@@ -295,6 +311,15 @@ impl MaintenanceDagSink {
         lineage.update(b"asap-maintenance-lineage-v1");
         let definition_bytes = source_definition.0 .0.to_be_bytes();
         lineage.update(definition_bytes);
+        if let Some(input) = &output.input_revision {
+            if input.generation.plan_id != plan.plan_id()
+                || input.generation.plan_version != plan.plan_version()
+            {
+                return Err("maintenance input belongs to an obsolete generation".into());
+            }
+            lineage.update(input.first_revision.to_be_bytes());
+            lineage.update(input.revision.to_be_bytes());
+        }
         let group_bytes = output
             .key
             .as_ref()
@@ -386,6 +411,12 @@ impl MaintenanceDagSink {
                     .saturating_mul(config.slide_interval)
                     .max(config.window_size)
                     .saturating_mul(1_000);
+                // This output was admitted before another worker advanced the
+                // replay floor. The store validates its exact consumed receipt
+                // before publication; it is not an unsolicited expired replay.
+                if output.input_revision.is_some() {
+                    self.commits.pin_admitted(&key)?;
+                }
                 if self.commits.is_published(&key)? {
                     continue;
                 }
@@ -504,6 +535,10 @@ impl OutputSink for MaintenanceDagSink {
                         .into(),
                 );
             }
+            if let Some(input) = &output.input_revision {
+                digest.update(input.first_revision.to_be_bytes());
+                digest.update(input.revision.to_be_bytes());
+            }
             digest.update(output.policy_fp.0.to_be_bytes());
             digest.update(output.start_timestamp.to_be_bytes());
             digest.update(output.end_timestamp.to_be_bytes());
@@ -638,6 +673,40 @@ mod tests {
         assert!(matches!(error, Err(reason) if reason.contains("typed update evaluator")));
     }
 
+    #[test]
+    fn admitted_slow_worker_can_publish_behind_another_workers_replay_floor() {
+        let commits = CommitRegistry::default();
+        commits.0.lock().unwrap().generation = Some((7, 1));
+        let key = |end| MaterializationCommitKey {
+            plan_id: 7,
+            plan_version: 1,
+            summary_definition: definition(2),
+            window_start_ms: end - 10,
+            window_end_ms: end,
+            input_lineage: vec![0; 32],
+        };
+        let fast = key(1000);
+        commits.begin_batch([1; 32]).unwrap();
+        commits
+            .commit_if_absent(fast.clone(), Arc::new(sum(2.0)))
+            .unwrap();
+        commits.publish(&fast, || Ok(())).unwrap();
+        commits.complete_batch([1; 32], &[(fast, 30)]).unwrap();
+        let slow = key(10);
+        assert!(commits.is_published(&slow).is_err());
+        commits.begin_batch([2; 32]).unwrap();
+        commits.pin_admitted(&slow).unwrap();
+        commits
+            .commit_if_absent(slow.clone(), Arc::new(sum(3.0)))
+            .unwrap();
+        commits.publish(&slow, || Ok(())).unwrap();
+        commits
+            .complete_batch([2; 32], &[(slow.clone(), 30)])
+            .unwrap();
+        assert!(commits.is_published(&slow).is_err());
+        assert!(commits.0.lock().unwrap().admitted_keys.is_empty());
+    }
+
     // Receipts follow the declared event-time horizon and never retain accepted
     // summary payloads; expired retries fail instead of becoming duplicate writes.
     #[test]
@@ -1009,3 +1078,36 @@ mod tests {
         assert!(commits.get(&key).unwrap().is_none());
     }
 }
+
+/// Materializations affected by an admitted source update, following installed
+/// semantic dependencies rather than assuming source and output identities match.
+pub(crate) fn affected_materializations(
+    plan: &asap_types::precompute_plan::PrecomputePlan,
+    source: asap_types::sds::SummaryDefinitionId,
+) -> BTreeSet<asap_types::sds::SummaryDefinitionId> {
+    use asap_types::executable_plan::BackendNodeBinding;
+    let mut affected = BTreeSet::from([source]);
+    for installed in plan.executable_dags.values() {
+        let mut reachable = installed.binding.nodes.iter().filter_map(|(node, binding)| {
+            matches!(binding, BackendNodeBinding::Materialization { summary_definition } if *summary_definition == source).then_some(*node)
+        }).collect::<BTreeSet<_>>();
+        let mut frontier = reachable.iter().copied().collect::<Vec<_>>();
+        while let Some(producer) = frontier.pop() {
+            for edge in &installed.document.edges {
+                if edge.producer == producer && reachable.insert(edge.consumer) {
+                    frontier.push(edge.consumer);
+                }
+            }
+        }
+        for sink in &installed.binding.precompute_sinks {
+            if reachable.contains(sink) {
+                if let Some(BackendNodeBinding::Materialization { summary_definition }) =
+                    installed.binding.nodes.get(sink)
+                {
+                    affected.insert(*summary_definition);
+                }
+            }
+        }
+    }
+    affected
+}
diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs
index 768a08291..51d4ebafb 100644
--- a/data_plane/src/precompute_engine/output_sink.rs
+++ b/data_plane/src/precompute_engine/output_sink.rs
@@ -161,14 +161,57 @@ impl SketchStoreSink {
         };
         let agg_cfg = &agg_cfg;
         let resolver = self.series_resolver.clone();
-        self.sketch_index
-            .ingest_precompute_for_agg_config(
-                |metric, fp, ak| resolver.resolve(metric, fp, ak),
-                agg_cfg,
-                output,
-                accumulator,
-            )
-            .is_some()
+        let persist = || {
+            self.sketch_index
+                .ingest_precompute_for_agg_config(
+                    |metric, fp, ak| resolver.resolve(metric, fp, ak),
+                    agg_cfg,
+                    output,
+                    accumulator,
+                )
+                .inspect(|_| crate::precompute_engine::metrics::record_materialized_outputs(1))
+        };
+        if let Some(revision) = &output.input_revision {
+            let group_values = output.population_labels.clone().unwrap_or_else(|| {
+                agg_cfg
+                    .grouping_labels
+                    .labels
+                    .iter()
+                    .cloned()
+                    .zip(output.key.clone().unwrap_or_default().labels)
+                    .collect()
+            });
+            let (Ok(start_ms), Ok(end_ms)) = (
+                i64::try_from(output.start_timestamp),
+                i64::try_from(output.end_timestamp),
+            ) else {
+                return false;
+            };
+            let coordinate = asap_types::sds::SummaryInstanceCoordinates {
+                summary_definition_id: output.policy_fp.into(),
+                time_range: asap_types::sds::HalfOpenTimeRange { start_ms, end_ms },
+                group_values,
+            };
+            if let Err(error) = self.sketch_index.publish_admitted_summary_update(
+                &revision.generation,
+                &coordinate,
+                revision.first_revision,
+                revision.revision,
+                agg_cfg
+                    .num_aggregates_to_retain
+                    .unwrap_or(1)
+                    .saturating_mul(agg_cfg.slide_interval)
+                    .max(agg_cfg.window_size)
+                    .saturating_mul(1_000),
+                persist,
+            ) {
+                warn!(%error, "summary input revision publication failed");
+                return false;
+            }
+            true
+        } else {
+            persist().is_some()
+        }
     }
 }
 
diff --git a/data_plane/src/precompute_engine/series_router.rs b/data_plane/src/precompute_engine/series_router.rs
index 6c2d83549..b0de9df62 100644
--- a/data_plane/src/precompute_engine/series_router.rs
+++ b/data_plane/src/precompute_engine/series_router.rs
@@ -25,6 +25,11 @@ use xxhash_rust::xxh64::xxh64;
 /// by sid without losing the data the legacy `(agg_id, group_key)` shape
 /// carried.
 pub enum WorkerMessage {
+    /// Receipt allocated after queue reservation and before any input is visible.
+    Admitted {
+        input: Box<WorkerMessage>,
+        revision: Arc<crate::storage_engines::types::SummaryInputRevision>,
+    },
     /// A batch of samples for the same series, routed by series key.
     /// Used in `pass_raw_samples` mode where no aggregation is needed.
     RawSamples {
@@ -96,6 +101,11 @@ pub enum WorkerMessage {
 impl fmt::Debug for WorkerMessage {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
+            Self::Admitted { input, revision } => f
+                .debug_struct("Admitted")
+                .field("input", input)
+                .field("revision", &revision.revision)
+                .finish(),
             Self::RawSamples {
                 series_key,
                 samples,
@@ -166,6 +176,9 @@ impl SeriesRouter {
         let mut per_worker: HashMap<usize, Vec<WorkerMessage>> = HashMap::new();
         for msg in messages {
             let worker_idx = match &msg {
+                WorkerMessage::Admitted { .. } => {
+                    return Err("input must be admitted by the router".into())
+                }
                 WorkerMessage::GroupSamples { sid, .. } => self.worker_for_sid(*sid),
                 WorkerMessage::AccumulatorInput { sid, .. } => self.worker_for_sid(*sid),
                 WorkerMessage::RawSamples { series_key, .. } => self.worker_for(series_key),
@@ -203,6 +216,17 @@ impl SeriesRouter {
     pub fn try_route_group_batch_atomic(
         &self,
         messages: Vec<WorkerMessage>,
+    ) -> Result<(), TryRouteError> {
+        self.try_route_group_batch_with_admission(messages, || Ok(None))
+    }
+
+    pub fn try_route_group_batch_with_admission(
+        &self,
+        messages: Vec<WorkerMessage>,
+        admit: impl FnOnce() -> Result<
+            Option<Arc<crate::storage_engines::types::SummaryInputRevision>>,
+            String,
+        >,
     ) -> Result<(), TryRouteError> {
         let mut pending = Vec::with_capacity(messages.len());
         for message in messages {
@@ -210,6 +234,9 @@ impl SeriesRouter {
                 WorkerMessage::GroupSamples { sid, .. }
                 | WorkerMessage::AccumulatorInput { sid, .. } => self.worker_for_sid(*sid),
                 WorkerMessage::RawSamples { series_key, .. } => self.worker_for(series_key),
+                WorkerMessage::Admitted { .. } => {
+                    return Err(TryRouteError::Admission("input already admitted".into()))
+                }
                 WorkerMessage::Flush | WorkerMessage::Drain(_) | WorkerMessage::Shutdown => 0,
             };
             let permit = self.senders[worker_idx]
@@ -221,8 +248,15 @@ impl SeriesRouter {
                 })?;
             pending.push((permit, message));
         }
+        let revision = admit().map_err(TryRouteError::Admission)?;
         for (permit, message) in pending {
-            permit.send(message);
+            permit.send(match &revision {
+                Some(revision) => WorkerMessage::Admitted {
+                    input: Box::new(message),
+                    revision: Arc::clone(revision),
+                },
+                None => message,
+            });
         }
         Ok(())
     }
@@ -283,8 +317,10 @@ impl SeriesRouter {
     }
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
+#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
 pub enum TryRouteError {
+    #[error("summary admission rejected: {0}")]
+    Admission(String),
     #[error("precompute queue is full")]
     Full,
     #[error("precompute worker is unavailable")]
@@ -343,10 +379,18 @@ mod tests {
                 ingest_received_at: Instant::now(),
             },
         ];
+        let mut admitted = false;
         assert_eq!(
-            router.try_route_group_batch_atomic(messages),
+            router.try_route_group_batch_with_admission(messages, || {
+                admitted = true;
+                Ok(None)
+            }),
             Err(TryRouteError::Full)
         );
+        assert!(
+            !admitted,
+            "failed queue reservation must not mutate admission"
+        );
         assert!(matches!(
             receiver.try_recv(),
             Err(mpsc::error::TryRecvError::Empty)
diff --git a/data_plane/src/precompute_engine/window_manager.rs b/data_plane/src/precompute_engine/window_manager.rs
index 89dd5bbde..ed0b84b97 100644
--- a/data_plane/src/precompute_engine/window_manager.rs
+++ b/data_plane/src/precompute_engine/window_manager.rs
@@ -12,6 +12,7 @@ pub struct WindowManager {
     pane_interval_ms: i64,
     /// Planned event-time phase of this materialization definition.
     origin_ms: Option<i64>,
+    stores_full_windows: bool,
 }
 
 impl WindowManager {
@@ -39,6 +40,7 @@ impl WindowManager {
             slide_interval_ms,
             pane_interval_ms: slide_interval_ms,
             origin_ms,
+            stores_full_windows: true,
         }
     }
 
@@ -49,12 +51,32 @@ impl WindowManager {
         layout: &asap_types::WindowMaterializationLayout,
     ) -> Self {
         let mut manager = Self::with_origin(window_size_secs, slide_interval_secs, origin_ms);
-        if !matches!(layout, asap_types::WindowMaterializationLayout::FullWindow) {
+        manager.stores_full_windows =
+            matches!(layout, asap_types::WindowMaterializationLayout::FullWindow);
+        if !manager.stores_full_windows {
             manager.pane_interval_ms = (layout.base_pane_secs() * 1_000) as i64;
         }
         manager
     }
 
+    /// Resolve the physical buckets updated by an input timestamp.
+    /// Their extent may differ from the semantic window or emission cadence.
+    pub fn stored_bucket_starts(&self, timestamp_ms: i64) -> Vec<i64> {
+        if self.stores_full_windows {
+            self.window_starts_containing(timestamp_ms)
+        } else {
+            vec![self.pane_start_for(timestamp_ms)]
+        }
+    }
+
+    pub fn stored_bucket_bounds(&self, start_ms: i64) -> (i64, i64) {
+        if self.stores_full_windows {
+            self.window_bounds(start_ms)
+        } else {
+            self.pane_bounds(start_ms)
+        }
+    }
+
     pub fn window_size_ms(&self) -> i64 {
         self.window_size_ms
     }
@@ -171,6 +193,29 @@ impl WindowManager {
 mod tests {
     use super::*;
 
+    #[test]
+    fn stored_bucket_assignment_distinguishes_full_windows_from_base_panes() {
+        // Admission and execution need the stored extent, not only slide cadence.
+        let full = WindowManager::with_layout(
+            10,
+            5,
+            Some(1_000),
+            &asap_types::WindowMaterializationLayout::FullWindow,
+        );
+        let mut starts = full.stored_bucket_starts(7_000);
+        starts.sort_unstable();
+        assert_eq!(starts, vec![1_000, 6_000]);
+        assert_eq!(full.stored_bucket_bounds(1_000), (1_000, 11_000));
+        let panes = WindowManager::with_layout(
+            10,
+            5,
+            Some(1_000),
+            &asap_types::WindowMaterializationLayout::Pane { pane_secs: 1 },
+        );
+        assert_eq!(panes.stored_bucket_starts(7_000), vec![7_000]);
+        assert_eq!(panes.stored_bucket_bounds(7_000), (7_000, 8_000));
+    }
+
     #[test]
     fn test_tumbling_window_start() {
         // 60-second (60000ms) tumbling windows
diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs
index 70514c845..6fdbeed50 100644
--- a/data_plane/src/precompute_engine/worker.rs
+++ b/data_plane/src/precompute_engine/worker.rs
@@ -37,6 +37,7 @@ use tracing::{debug, debug_span, info, warn};
 /// producing one output per (sid, window) — exactly like Arroyo's
 /// `GROUP BY window, key`.
 struct GroupState {
+    input_revisions: BTreeMap<i64, Arc<crate::storage_engines::types::SummaryInputRevision>>,
     config: Arc<AggregationConfig>,
     /// Source policy fingerprint that minted this sid. Held so
     /// `evict_orphaned_groups` can check liveness against the streaming
@@ -91,11 +92,7 @@ impl GroupState {
     }
 
     fn bucket_starts_for(&self, timestamp_ms: i64) -> Vec<i64> {
-        if self.stores_full_windows() {
-            self.window_manager.window_starts_containing(timestamp_ms)
-        } else {
-            vec![self.window_manager.pane_start_for(timestamp_ms)]
-        }
+        self.window_manager.stored_bucket_starts(timestamp_ms)
     }
 
     fn closed_buckets(&self, previous_ms: i64, current_ms: i64) -> Vec<i64> {
@@ -107,11 +104,7 @@ impl GroupState {
     }
 
     fn bucket_bounds(&self, start_ms: i64) -> (i64, i64) {
-        if self.stores_full_windows() {
-            self.window_manager.window_bounds(start_ms)
-        } else {
-            self.window_manager.pane_bounds(start_ms)
-        }
+        self.window_manager.stored_bucket_bounds(start_ms)
     }
 
     fn touch_pane(&mut self, pane_start_ms: i64, now_ms: i64) {
@@ -129,6 +122,8 @@ impl GroupState {
         let sketch = &self.sketch_panes;
         self.pane_wall_clock
             .retain(|ps, _| active.contains_key(ps) || sketch.contains_key(ps));
+        self.input_revisions
+            .retain(|ps, _| active.contains_key(ps) || sketch.contains_key(ps));
     }
 }
 
@@ -153,6 +148,7 @@ pub struct WorkerRuntimeConfig {
 /// `(metric, attrs_fingerprint, agg_kind_canonical)` identity contract on
 /// `SeriesIdResolver`, so one sid uniquely names one bucket.
 pub struct Worker {
+    current_input_revision: Option<Arc<crate::storage_engines::types::SummaryInputRevision>>,
     id: usize,
     receiver: mpsc::Receiver<WorkerMessage>,
     output_sink: Arc<dyn OutputSink>,
@@ -208,6 +204,7 @@ impl Worker {
             wall_clock_max_open_grace_period_ms,
         } = runtime_config;
         Self {
+            current_input_revision: None,
             id,
             receiver,
             output_sink,
@@ -241,7 +238,20 @@ impl Worker {
 
         let mut processing_error: Option<String> = None;
         while let Some(msg) = self.receiver.recv().await {
+            let msg = match msg {
+                WorkerMessage::Admitted { input, revision } => {
+                    self.current_input_revision = Some(revision);
+                    *input
+                }
+                message => {
+                    self.current_input_revision = None;
+                    message
+                }
+            };
             match msg {
+                WorkerMessage::Admitted { .. } => {
+                    processing_error = Some("nested admission receipt".into());
+                }
                 WorkerMessage::GroupSamples {
                     sid,
                     policy_fp,
@@ -395,6 +405,7 @@ impl Worker {
             let cfg = snap.get_aggregation_config(policy_fp.as_u64())?;
             let config = Arc::new(cfg.clone());
             let gs = GroupState {
+                input_revisions: BTreeMap::new(),
                 window_manager: WindowManager::with_layout(
                     config.window_size,
                     config.slide_interval,
@@ -433,6 +444,7 @@ impl Worker {
         group_key: &Arc<GroupKey>,
         samples: Vec<(String, i64, f64)>, // (series_key, timestamp_ms, value)
     ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        let input_revision = self.current_input_revision.clone();
         let worker_id = self.id;
         let allowed_lateness_ms = self.allowed_lateness_ms;
         let late_data_policy = self.late_data_policy;
@@ -497,19 +509,26 @@ impl Worker {
             let too_late = previous_event_time != i64::MIN
                 && pane_timestamp(*ts)
                     < watermark_for_event_time(previous_event_time, allowed_lateness_ms);
-            let value = if let SampleUpdateRule::CounterDelta { .. } =
-                state.config.sample_update_rule()
-            {
-                let Some(delta) =
+            let value =
+                if let SampleUpdateRule::CounterDelta { .. } = state.config.sample_update_rule() {
                     reset_aware_counter_delta(&mut state.counter_previous, series_key, *val, *ts)
-                else {
-                    continue;
+                } else {
+                    Some(*val)
                 };
-                delta
-            } else {
-                *val
-            };
             for bucket_start in state.bucket_starts_for(pane_timestamp(*ts)) {
+                if let Some(revision) = &input_revision {
+                    state
+                        .input_revisions
+                        .entry(bucket_start)
+                        .and_modify(|existing| {
+                            if existing.revision < revision.revision {
+                                let mut combined = (**revision).clone();
+                                combined.first_revision = existing.first_revision;
+                                *existing = Arc::new(combined);
+                            }
+                        })
+                        .or_insert_with(|| Arc::clone(revision));
+                }
                 let (_, bucket_end) = state.bucket_bounds(bucket_start);
                 let bucket_closed = !state.active_panes.contains_key(&bucket_start)
                     && previous_closure_watermark >= bucket_end;
@@ -519,6 +538,9 @@ impl Worker {
                     let window_end = bucket_end;
                     match late_data_policy {
                         LateDataPolicy::Drop => {
+                            if let Some(input) = state.input_revisions.get_mut(&bucket_start) {
+                                Arc::make_mut(input).first_revision = 0;
+                            }
                             record_late_input("drop", "raw_sample");
                             debug!(
                                 "Worker {} dropping late sample for sid={} (group={}): \
@@ -543,6 +565,9 @@ impl Worker {
                                 state.config.sample_update_rule(),
                                 SampleUpdateRule::CounterDelta { .. }
                             ) {
+                                if let Some(input) = state.input_revisions.get_mut(&bucket_start) {
+                                    Arc::make_mut(input).first_revision = 0;
+                                }
                                 record_late_input("drop", "counter_delta_membership");
                                 continue;
                             }
@@ -556,6 +581,7 @@ impl Worker {
                                 key,
                                 PolicyFingerprint::from_config(&state.config),
                                 group_key,
+                                &state.input_revisions,
                             );
                             emit_batch.push((output, updater.take_accumulator()));
                             debug!(
@@ -576,7 +602,9 @@ impl Worker {
                     .active_panes
                     .entry(bucket_start)
                     .or_insert_with(|| create_accumulator_updater(&state.config));
-                apply_sample(&mut **updater, series_key, value, *ts, &state.config);
+                if let Some(value) = value {
+                    apply_sample(&mut **updater, series_key, value, *ts, &state.config);
+                }
             }
         }
 
@@ -601,6 +629,7 @@ impl Worker {
                     key,
                     PolicyFingerprint::from_config(&state.config),
                     group_key,
+                    &state.input_revisions,
                 );
                 emit_batch.push((output, accumulator));
             }
@@ -610,6 +639,19 @@ impl Worker {
         if event_watermark > state.closure_watermark_ms {
             state.closure_watermark_ms = event_watermark;
         }
+        // A skipped admitted sample makes this coordinate incomplete, including
+        // corrections prepared earlier in this same batch.
+        for (output, _) in &mut emit_batch {
+            if state
+                .input_revisions
+                .get(&(output.start_timestamp as i64))
+                .is_some_and(|input| input.first_revision == 0)
+            {
+                if let Some(input) = &mut output.input_revision {
+                    Arc::make_mut(input).first_revision = 0;
+                }
+            }
+        }
         state.prune_pane_wall_clock();
 
         // Emit to output sink
@@ -621,7 +663,8 @@ impl Worker {
                 sid,
                 group_key
             );
-            self.output_sink.emit_batch(emit_batch)?;
+            self.output_sink
+                .emit_batch(coalesce_admitted_outputs(emit_batch)?)?;
         }
 
         crate::precompute_engine::metrics::record_processed_updates(samples.len() as u64);
@@ -715,6 +758,7 @@ impl Worker {
                             key,
                             PolicyFingerprint::from_config(&state.config),
                             group_key,
+                            &state.input_revisions,
                         );
                         emit_batch.push((output, incoming.clone_boxed_core()));
                     }
@@ -760,6 +804,7 @@ impl Worker {
                     key,
                     PolicyFingerprint::from_config(&state.config),
                     group_key,
+                    &state.input_revisions,
                 );
                 emit_batch.push((output, accumulator));
             }
@@ -775,6 +820,7 @@ impl Worker {
                     key,
                     PolicyFingerprint::from_config(&state.config),
                     group_key,
+                    &state.input_revisions,
                 );
                 emit_batch.push((output, accumulator));
             }
@@ -964,6 +1010,7 @@ impl Worker {
                         key,
                         PolicyFingerprint::from_config(&state.config),
                         &group_key,
+                        &state.input_revisions,
                     );
                     emit_batch.push((output, accumulator));
                 }
@@ -978,6 +1025,7 @@ impl Worker {
                         key,
                         PolicyFingerprint::from_config(&state.config),
                         &group_key,
+                        &state.input_revisions,
                     );
                     emit_batch.push((output, accumulator));
                 }
@@ -1064,6 +1112,7 @@ impl Worker {
                         key,
                         PolicyFingerprint::from_config(&state.config),
                         &group_key,
+                        &state.input_revisions,
                     );
                     emit_batch.push((output, accumulator));
                 }
@@ -1078,6 +1127,7 @@ impl Worker {
                         key,
                         PolicyFingerprint::from_config(&state.config),
                         &group_key,
+                        &state.input_revisions,
                     );
                     emit_batch.push((output, accumulator));
                 }
@@ -1139,15 +1189,72 @@ fn population_labels_from_group_key(group_key: &GroupKey) -> Option<BTreeMap<Str
     (!labels.is_empty()).then_some(labels)
 }
 
+/// One admission receipt covers all fragments for a physical coordinate.
+/// Combine late corrections before publishing that receipt once.
+fn coalesce_admitted_outputs(
+    outputs: Vec<(PrecomputedOutput, Box<dyn AggregateCore>)>,
+) -> Result<
+    Vec<(PrecomputedOutput, Box<dyn AggregateCore>)>,
+    Box<dyn std::error::Error + Send + Sync>,
+> {
+    let mut merged: Vec<(PrecomputedOutput, Box<dyn AggregateCore>)> = Vec::new();
+    let mut coordinates = BTreeMap::new();
+    for (output, state) in outputs {
+        let index = output.input_revision.as_ref().and_then(|revision| {
+            let key = (
+                output.policy_fp.0,
+                output.start_timestamp,
+                output.end_timestamp,
+                output
+                    .key
+                    .as_ref()
+                    .map(|key| key.serialize_to_bytes())
+                    .unwrap_or_default(),
+                output.population_labels.clone(),
+                revision.generation.plan_id,
+                revision.generation.plan_version,
+                revision.revision,
+            );
+            match coordinates.entry(key) {
+                std::collections::btree_map::Entry::Occupied(entry) => Some(*entry.get()),
+                std::collections::btree_map::Entry::Vacant(entry) => {
+                    entry.insert(merged.len());
+                    None
+                }
+            }
+        });
+        if let Some(index) = index {
+            merged[index].1 = merged[index].1.merge_with(state.as_ref())?;
+            if output
+                .input_revision
+                .as_ref()
+                .is_some_and(|input| input.first_revision == 0)
+            {
+                if let Some(input) = merged[index].0.input_revision.as_mut() {
+                    Arc::make_mut(input).first_revision = 0;
+                }
+            }
+        } else {
+            merged.push((output, state));
+        }
+    }
+    Ok(merged)
+}
+
 fn precomputed_output_for_group(
     start_timestamp: u64,
     end_timestamp: u64,
     key: KeyByLabelValues,
     policy_fp: PolicyFingerprint,
     group_key: &GroupKey,
+    input_revisions: &BTreeMap<i64, Arc<crate::storage_engines::types::SummaryInputRevision>>,
 ) -> PrecomputedOutput {
-    PrecomputedOutput::new(start_timestamp, end_timestamp, Some(key), policy_fp)
-        .with_population_labels(population_labels_from_group_key(group_key))
+    let mut output = PrecomputedOutput::new(start_timestamp, end_timestamp, Some(key), policy_fp)
+        .with_population_labels(population_labels_from_group_key(group_key));
+    output.input_revision = i64::try_from(start_timestamp)
+        .ok()
+        .and_then(|start| input_revisions.get(&start).cloned());
+    output
 }
 
 /// Extract the metric name from a series key like `"metric_name{key1=\"val1\"}"`.
@@ -1462,6 +1569,37 @@ fn merge_sketch_panes_for_window(
 mod tests {
     use super::*;
 
+    #[test]
+    fn admitted_corrections_publish_all_fragments_under_one_receipt() {
+        let revision = Arc::new(crate::storage_engines::types::SummaryInputRevision {
+            generation: Arc::new(asap_types::sds::CatalogGeneration {
+                schema_version: 1,
+                plan_id: 1,
+                plan_version: 1,
+                snapshot_sha256: "test".into(),
+            }),
+            first_revision: 1,
+            revision: 1,
+        });
+        let mut output = PrecomputedOutput::new(0, 1000, None, PolicyFingerprint(1));
+        output.input_revision = Some(revision);
+        let merged = coalesce_admitted_outputs(vec![
+            (output.clone(), Box::new(SumAccumulator::with_sum(2.0))),
+            (output, Box::new(SumAccumulator::with_sum(3.0))),
+        ])
+        .unwrap();
+        assert_eq!(merged.len(), 1);
+        assert_eq!(
+            merged[0]
+                .1
+                .as_any()
+                .downcast_ref::<SumAccumulator>()
+                .unwrap()
+                .sum,
+            5.0
+        );
+    }
+
     fn test_group_key(value: &str) -> Arc<GroupKey> {
         if value.is_empty() {
             return crate::precompute_engine::group_key::intern_pairs(std::iter::empty::<(
@@ -1706,6 +1844,56 @@ mod tests {
     // Test: raw mode — each sample forwarded as SumAccumulator with sum==value
     // -----------------------------------------------------------------------
 
+    #[test]
+    fn initial_counter_observation_publishes_an_empty_delta_window() {
+        let mut config = make_agg_config(1, "counter", AggregationType::Sum, "sum", 1, 1, vec![]);
+        config
+            .parameters
+            .insert("weight_mode".into(), serde_json::json!("counter_delta"));
+        let fingerprint = config.policy_fingerprint();
+        let sink = Arc::new(CapturingOutputSink::new());
+        let mut worker = make_worker(
+            HashMap::from([(fingerprint.0, config)]),
+            sink.clone(),
+            false,
+            0,
+            LateDataPolicy::Drop,
+        );
+        worker.current_input_revision = Some(Arc::new(
+            crate::storage_engines::types::SummaryInputRevision {
+                generation: Arc::new(asap_types::sds::CatalogGeneration {
+                    schema_version: 1,
+                    plan_id: 1,
+                    plan_version: 1,
+                    snapshot_sha256: "test".into(),
+                }),
+                first_revision: 1,
+                revision: 1,
+            },
+        ));
+        worker
+            .process_group_samples(
+                1,
+                fingerprint,
+                &test_group_key(""),
+                vec![("counter".into(), 1000, 7.0)],
+            )
+            .unwrap();
+        worker.force_close_all().unwrap();
+        let captured = sink.drain();
+        assert_eq!(captured.len(), 1);
+        assert_eq!(captured[0].0.input_revision.as_ref().unwrap().revision, 1);
+        assert_eq!(
+            captured[0]
+                .1
+                .as_any()
+                .downcast_ref::<SumAccumulator>()
+                .unwrap()
+                .sum,
+            0.0
+        );
+    }
+
     #[test]
     fn test_raw_mode_forwarding() {
         let sink = Arc::new(CapturingOutputSink::new());
@@ -2554,13 +2742,20 @@ aggregations:
             crate::precompute_engine::group_key::intern_pairs([("instance", "a"), ("job", "api")]);
         let key = build_group_key_label_values(&group);
         assert_eq!(key.labels, vec!["a".to_string(), "api".to_string()]);
-        let output = precomputed_output_for_group(0, 5_000, key, PolicyFingerprint(7), &group);
+        let output = precomputed_output_for_group(
+            0,
+            5_000,
+            key,
+            PolicyFingerprint(7),
+            &group,
+            &BTreeMap::new(),
+        );
         assert_eq!(
             output.population_labels,
             Some(BTreeMap::from([
                 ("instance".to_string(), "a".to_string()),
                 ("job".to_string(), "api".to_string()),
-            ]))
+            ])),
         );
     }
 
diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs
index 8215af484..3ed21015a 100644
--- a/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs
+++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs
@@ -230,6 +230,33 @@ pub fn execute_sql_dag_with_external(
     t0_ms: u64,
     t1_ms: u64,
     is_cumulative: bool,
+) -> ClickHouseDagOutcome {
+    let revision = index.summary_update_revision();
+    let result = execute_sql_dag_with_external_unfenced(
+        index,
+        entry,
+        sds,
+        prepared,
+        t0_ms,
+        t1_ms,
+        is_cumulative,
+    );
+    if !revision.matches(index.summary_update_revision()) {
+        return ClickHouseDagOutcome::Fallback(ClickHouseDagFallback::UnsupportedPlan(
+            "summary input changed during SQL DAG evaluation".into(),
+        ));
+    }
+    result
+}
+
+fn execute_sql_dag_with_external_unfenced(
+    index: &SketchStore,
+    entry: &QueryPlanEntry,
+    sds: &SummaryCatalog,
+    prepared: &PreparedExternalLeaves,
+    t0_ms: u64,
+    t1_ms: u64,
+    is_cumulative: bool,
 ) -> ClickHouseDagOutcome {
     if let Err(error) = validate_payload(Some(sds), entry, sds.plan_id, sds.plan_version) {
         return ClickHouseDagOutcome::Fallback(ClickHouseDagFallback::UnsupportedPlan(
diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs
index 766947b8c..12d31e207 100644
--- a/data_plane/src/query_engines/asap_query_engine/engine.rs
+++ b/data_plane/src/query_engines/asap_query_engine/engine.rs
@@ -277,103 +277,134 @@ impl ASAPQueryEngine {
         crate::query_engines::EngineError,
     > {
         use crate::query_engines::EngineError;
-        super::logical_dag::execute_installed(entry, leaves, at, |root, evaluation_ms| {
-            let mut subtree = entry.clone();
-            subtree.root = root;
-            let reachable = subtree.topological_order().map_err(|e| {
-                EngineError::capability_miss("installed_logical_dag", e.to_string())
-            })?;
-            subtree.nodes.retain(|id, _| reachable.contains(id));
-            let bindings: Vec<_> = subtree
-                .materialization_bindings()
-                .into_iter()
-                .cloned()
-                .collect();
-            let windows: std::collections::BTreeSet<Option<u64>> =
-                bindings.iter().map(|b| b.readout_lookback_ms).collect();
-            if windows.len() != 1 || windows.contains(&None) || windows.contains(&Some(0)) {
-                return Err(EngineError::capability_miss(
-                    "installed_logical_dag",
-                    "bound subtree requires one explicit positive window",
-                ));
-            }
-            subtree.instant.lookback_ms = windows
-                .first()
-                .copied()
-                .flatten()
-                .expect("explicit semantic lookback checked");
-            subtree.instant.full_history = false;
-            subtree.instant.cumulative_readout = true;
-            let requirement = readiness_requirement(&subtree);
-            let index = self.sketch_index.as_ref().ok_or_else(|| {
-                EngineError::capability_miss("installed_logical_dag", "summary store unavailable")
-            })?;
-            let (result, t0) =
-                super::live_serve::serve_instant_from_query_plan(index, &subtree, evaluation_ms)
-                    .map_err(|e| {
-                        EngineError::capability_miss(
-                            "installed_logical_dag",
-                            format!("bound readout failed: {e:?}"),
-                        )
-                    })?;
-            let active = self.active_physical_plan.as_ref().ok_or_else(|| {
-                EngineError::capability_miss(
-                    "installed_logical_dag",
-                    "readiness registry unavailable",
-                )
-            })?;
-            let plan_id = physical.plan_id();
-            let version = physical.plan_version();
-            // Exact range accumulators preserve their actual first/last sample
-            // timestamps. Sparse counter series may legitimately begin after
-            // the range boundary; Prometheus evaluates the samples that exist.
-            // The physical plan's retention bound guarantees stored panes were
-            // not evicted, so requiring a sample at t0 would reject valid data.
-            let exact_accumulator_bindings =
-                physical.summary_catalog.as_deref().is_some_and(|catalog| {
-                    bindings.iter().all(|binding| {
-                        super::catalog_resolver::resolve(catalog, binding.materialization)
-                            .is_ok_and(|resolved| resolved.is_exact())
-                    })
-                });
-            let sparse_exact_coverage = exact_accumulator_bindings
-                && result
-                    .coverage
-                    .is_some_and(|(_, coverage_end)| coverage_end >= evaluation_ms);
-            if !sparse_exact_coverage
-                && !complete_window_coverage(
-                    result.coverage,
-                    t0,
+        let revision = self
+            .sketch_index
+            .as_ref()
+            .map(|index| index.summary_update_revision());
+        let result =
+            super::logical_dag::execute_installed(entry, leaves, at, |root, evaluation_ms| {
+                let mut subtree = entry.clone();
+                subtree.root = root;
+                let reachable = subtree.topological_order().map_err(|e| {
+                    EngineError::capability_miss("installed_logical_dag", e.to_string())
+                })?;
+                subtree.nodes.retain(|id, _| reachable.contains(id));
+                let bindings: Vec<_> = subtree
+                    .materialization_bindings()
+                    .into_iter()
+                    .cloned()
+                    .collect();
+                let windows: std::collections::BTreeSet<Option<u64>> =
+                    bindings.iter().map(|b| b.readout_lookback_ms).collect();
+                if windows.len() != 1 || windows.contains(&None) || windows.contains(&Some(0)) {
+                    return Err(EngineError::capability_miss(
+                        "installed_logical_dag",
+                        "bound subtree requires one explicit positive window",
+                    ));
+                }
+                subtree.instant.lookback_ms = windows
+                    .first()
+                    .copied()
+                    .flatten()
+                    .expect("explicit semantic lookback checked");
+                subtree.instant.full_history = false;
+                subtree.instant.cumulative_readout = true;
+                let requirement = readiness_requirement(&subtree);
+                let index = self.sketch_index.as_ref().ok_or_else(|| {
+                    EngineError::capability_miss(
+                        "installed_logical_dag",
+                        "summary store unavailable",
+                    )
+                })?;
+                let (result, t0) = super::live_serve::serve_instant_from_query_plan(
+                    index,
+                    &subtree,
                     evaluation_ms,
-                    requirement.max_window_ms,
                 )
-            {
-                active.mark_materializing(
-                    plan_id,
-                    version,
-                    &requirement.materializations,
-                    result.coverage,
-                );
-                return Err(EngineError::capability_miss(
-                    "installed_logical_dag",
-                    format!("bound readout incomplete at {evaluation_ms}"),
-                ));
-            }
-            let coverage = result.coverage.expect("coverage checked");
-            if !active.mark_ready(plan_id, version, &requirement.materializations, coverage)
-                || !active.mark_serving(plan_id, version, &requirement.materializations, coverage)
-            {
-                return Err(EngineError::capability_miss(
-                    "installed_logical_dag",
-                    "physical generation changed during bound readout",
-                ));
-            }
-            Ok(asap_tier_result_to_query_result(
-                result,
-                evaluation_ms,
-                false,
-            ))
-        })
+                .map_err(|e| {
+                    EngineError::capability_miss(
+                        "installed_logical_dag",
+                        format!("bound readout failed: {e:?}"),
+                    )
+                })?;
+                let active = self.active_physical_plan.as_ref().ok_or_else(|| {
+                    EngineError::capability_miss(
+                        "installed_logical_dag",
+                        "readiness registry unavailable",
+                    )
+                })?;
+                let plan_id = physical.plan_id();
+                let version = physical.plan_version();
+                // Exact range accumulators preserve their actual first/last sample
+                // timestamps. Sparse counter series may legitimately begin after
+                // the range boundary; Prometheus evaluates the samples that exist.
+                // The physical plan's retention bound guarantees stored panes were
+                // not evicted, so requiring a sample at t0 would reject valid data.
+                let exact_accumulator_bindings =
+                    physical.summary_catalog.as_deref().is_some_and(|catalog| {
+                        bindings.iter().all(|binding| {
+                            super::catalog_resolver::resolve(catalog, binding.materialization)
+                                .is_ok_and(|resolved| resolved.is_exact())
+                        })
+                    });
+                let sparse_exact_coverage = exact_accumulator_bindings
+                    && result
+                        .coverage
+                        .is_some_and(|(_, coverage_end)| coverage_end >= evaluation_ms);
+                if !sparse_exact_coverage
+                    && !complete_window_coverage(
+                        result.coverage,
+                        t0,
+                        evaluation_ms,
+                        requirement.max_window_ms,
+                    )
+                {
+                    active.mark_materializing(
+                        plan_id,
+                        version,
+                        &requirement.materializations,
+                        result.coverage,
+                    );
+                    return Err(EngineError::capability_miss(
+                        "installed_logical_dag",
+                        format!("bound readout incomplete at {evaluation_ms}"),
+                    ));
+                }
+                let coverage = result.coverage.expect("coverage checked");
+                if !active.mark_ready(plan_id, version, &requirement.materializations, coverage)
+                    || !active.mark_serving(
+                        plan_id,
+                        version,
+                        &requirement.materializations,
+                        coverage,
+                    )
+                {
+                    return Err(EngineError::capability_miss(
+                        "installed_logical_dag",
+                        "physical generation changed during bound readout",
+                    ));
+                }
+                Ok(asap_tier_result_to_query_result(
+                    result,
+                    evaluation_ms,
+                    false,
+                ))
+            });
+        let current = self
+            .sketch_index
+            .as_ref()
+            .map(|index| index.summary_update_revision());
+        if match (revision, current) {
+            (Some(before), Some(after)) => !before.matches(after),
+            (None, None) => false,
+            _ => true,
+        } {
+            return Err(EngineError::capability_miss(
+                "installed_logical_dag",
+                "summary input changed during query DAG evaluation",
+            ));
+        }
+        result
     }
 
     async fn execute_logical_range(
diff --git a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs
index a908595dd..7a8e8bf52 100644
--- a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs
+++ b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs
@@ -535,40 +535,49 @@ fn execute_physical_query_payload(
     t1_ms: u64,
     is_cumulative: bool,
 ) -> Result<PostAsapReadoutOutcome, LoweringSkip> {
-    let runtime = PhysicalQueryRuntime {
-        context: QueryExecutionContext {
-            index,
-            t0_ms,
-            t1_ms,
-            is_cumulative,
-            allowed_materializations: None,
-        },
-    };
-    let output = physical_dag::execute_from(entry, root, &runtime)
-        .map_err(|error| LoweringSkip::ExecuteFailed(format!("{error:?}")))?;
-    match output {
-        PhysicalQueryOutput::Scalar(_) => Err(LoweringSkip::ExecuteFailed(
-            "scalar-only query is not a warm vector result".into(),
-        )),
-        PhysicalQueryOutput::Value(values, coverage) => {
-            let mut series = Vec::new();
-            for (group_key, value) in &values {
-                series.extend(summary_value_to_series(group_key, value));
+    let revision = index.summary_update_revision();
+    let result = (|| {
+        let runtime = PhysicalQueryRuntime {
+            context: QueryExecutionContext {
+                index,
+                t0_ms,
+                t1_ms,
+                is_cumulative,
+                allowed_materializations: None,
+            },
+        };
+        let output = physical_dag::execute_from(entry, root, &runtime)
+            .map_err(|error| LoweringSkip::ExecuteFailed(format!("{error:?}")))?;
+        match output {
+            PhysicalQueryOutput::Scalar(_) => Err(LoweringSkip::ExecuteFailed(
+                "scalar-only query is not a warm vector result".into(),
+            )),
+            PhysicalQueryOutput::Value(values, coverage) => {
+                let mut series = Vec::new();
+                for (group_key, value) in &values {
+                    series.extend(summary_value_to_series(group_key, value));
+                }
+                Ok(PostAsapReadoutOutcome { series, coverage })
             }
-            Ok(PostAsapReadoutOutcome { series, coverage })
-        }
-        PhysicalQueryOutput::State { groups, .. } => {
-            let mut coverage = None;
-            let mut series = Vec::new();
-            for (group_key, state) in &groups {
-                fold_coverage(&mut coverage, state.exact_coverage());
-                if let Some(value) = state.exact_value(&None) {
-                    series.push((group_key.clone(), vec![(t1_ms as i64, value)]));
+            PhysicalQueryOutput::State { groups, .. } => {
+                let mut coverage = None;
+                let mut series = Vec::new();
+                for (group_key, state) in &groups {
+                    fold_coverage(&mut coverage, state.exact_coverage());
+                    if let Some(value) = state.exact_value(&None) {
+                        series.push((group_key.clone(), vec![(t1_ms as i64, value)]));
+                    }
                 }
+                Ok(PostAsapReadoutOutcome { series, coverage })
             }
-            Ok(PostAsapReadoutOutcome { series, coverage })
         }
+    })();
+    if !revision.matches(index.summary_update_revision()) {
+        return Err(LoweringSkip::ExecuteFailed(
+            "summary input changed during query DAG evaluation".into(),
+        ));
     }
+    result
 }
 
 /// Plan and execute an instant query without consulting the legacy candidate
diff --git a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs
index 8d8261ff8..4132ddb90 100644
--- a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs
+++ b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs
@@ -439,6 +439,23 @@ impl QueryExecutionContext<'_> {
     ) -> Result<Vec<(BTreeMap<String, String>, GroupState)>, SummaryExecutorError> {
         use control_plane::query_plan::PhysicalGrouping;
 
+        let inventory_revision = self.index.summary_update_revision();
+        let query_range = asap_types::sds::HalfOpenTimeRange {
+            start_ms: i64::try_from(self.t0_ms).map_err(|_| {
+                SummaryExecutorError::Unsupported("query start exceeds signed event time")
+            })?,
+            end_ms: i64::try_from(self.t1_ms).map_err(|_| {
+                SummaryExecutorError::Unsupported("query end exceeds signed event time")
+            })?,
+        };
+        if self
+            .index
+            .has_pending_summary_updates(binding.materialization, query_range)
+        {
+            return Err(SummaryExecutorError::Unsupported(
+                "materialization population has unpublished input",
+            ));
+        }
         validate_binding_phase(binding, self.t1_ms)?;
 
         enum Candidate {
@@ -488,6 +505,12 @@ impl QueryExecutionContext<'_> {
         let mut by_group: BTreeMap<BTreeMap<String, String>, Vec<GroupState>> = BTreeMap::new();
 
         for sid in sids.iter().copied() {
+            if self
+                .index
+                .summary_window_known_empty(binding.materialization, sid, query_range)
+            {
+                continue;
+            }
             let candidate = self
                 .index
                 .with_instance(sid, |meta| {
@@ -617,12 +640,18 @@ impl QueryExecutionContext<'_> {
             );
             return Err(SummaryExecutorError::NoCandidates);
         }
-        by_group
+        let result = by_group
             .into_iter()
             .map(|(key, states)| {
                 <Self as SummaryExecutor>::merge_states(self, states).map(|state| (key, state))
             })
-            .collect()
+            .collect();
+        if !inventory_revision.matches(self.index.summary_update_revision()) {
+            return Err(SummaryExecutorError::Unsupported(
+                "summary input changed during read",
+            ));
+        }
+        result
     }
 
     pub fn readout_bound(
diff --git a/data_plane/src/storage_engines/sketch_db/index/admission.rs b/data_plane/src/storage_engines/sketch_db/index/admission.rs
new file mode 100644
index 000000000..f0d6eceed
--- /dev/null
+++ b/data_plane/src/storage_engines/sketch_db/index/admission.rs
@@ -0,0 +1,468 @@
+//! Store-owned tracking of accepted, not necessarily published summary updates.
+//! This records known work; it does not infer an event-time watermark.
+
+use asap_types::sds::SummaryDefinitionId;
+use asap_types::sds::{CatalogGeneration, HalfOpenTimeRange, SummaryInstanceCoordinates};
+use std::collections::{BTreeMap, BTreeSet};
+
+#[derive(Default)]
+struct WindowRevision {
+    admitted: u64,
+    published: u64,
+    series_id: Option<u64>,
+    pending: BTreeSet<u64>,
+}
+
+#[derive(Default)]
+pub(super) struct AdmissionInventory {
+    generation: Option<CatalogGeneration>,
+    revision: u64,
+    windows: BTreeMap<SummaryInstanceCoordinates, WindowRevision>,
+    metadata_bytes: usize,
+    replay_floors: BTreeMap<SummaryDefinitionId, i64>,
+    observed_extent: Option<HalfOpenTimeRange>,
+    finite_complete: bool,
+    published_series: BTreeSet<u64>,
+    pending_revisions: usize,
+}
+
+impl AdmissionInventory {
+    const MAX_WINDOWS: usize = 262_144;
+    const MAX_METADATA_BYTES: usize = 64 * 1024 * 1024;
+
+    pub(super) fn install(&mut self, generation: CatalogGeneration) {
+        if self.generation.as_ref() != Some(&generation) {
+            self.generation = Some(generation);
+            self.windows.clear();
+            self.metadata_bytes = 0;
+            self.replay_floors.clear();
+            self.observed_extent = None;
+            self.finite_complete = false;
+            self.published_series.clear();
+            self.pending_revisions = 0;
+            self.revision = self.revision.saturating_add(1);
+        }
+    }
+
+    /// All validation and capacity checks precede mutation. The caller reserves
+    /// the whole queue batch before invoking this operation.
+    pub(super) fn admit(
+        &mut self,
+        generation: &CatalogGeneration,
+        coordinates: BTreeSet<SummaryInstanceCoordinates>,
+    ) -> Result<u64, String> {
+        if self.generation.as_ref() != Some(generation) {
+            return Err("summary admission catalog generation differs".into());
+        }
+        if self.finite_complete {
+            return Err("finite summary input is closed".into());
+        }
+        let mut added = 0usize;
+        let mut bytes = 0usize;
+        for coordinate in &coordinates {
+            if self
+                .replay_floors
+                .get(&coordinate.summary_definition_id)
+                .is_some_and(|floor| coordinate.time_range.end_ms <= *floor)
+            {
+                return Err("summary input precedes retained replay horizon".into());
+            }
+            coordinate
+                .time_range
+                .validate()
+                .map_err(|e| e.to_string())?;
+            if !self.windows.contains_key(coordinate) {
+                added += 1;
+                bytes = bytes.saturating_add(Self::coordinate_bytes(coordinate));
+            }
+        }
+        if self.pending_revisions.saturating_add(coordinates.len()) > Self::MAX_WINDOWS
+            || self.windows.len().saturating_add(added) > Self::MAX_WINDOWS
+            || self.metadata_bytes.saturating_add(bytes) > Self::MAX_METADATA_BYTES
+        {
+            return Err("summary admission inventory capacity exceeded".into());
+        }
+        let revision = self
+            .revision
+            .checked_add(1)
+            .ok_or("summary admission revision exhausted")?;
+        for coordinate in coordinates {
+            self.observed_extent = Some(match self.observed_extent {
+                Some(range) => HalfOpenTimeRange {
+                    start_ms: range.start_ms.min(coordinate.time_range.start_ms),
+                    end_ms: range.end_ms.max(coordinate.time_range.end_ms),
+                },
+                None => coordinate.time_range,
+            });
+            let window = self.windows.entry(coordinate).or_default();
+            window.admitted = revision;
+            window.pending.insert(revision);
+            self.pending_revisions += 1;
+        }
+        self.metadata_bytes += bytes;
+        self.revision = revision;
+        Ok(revision)
+    }
+
+    pub(super) fn validate_publication(
+        &self,
+        generation: &CatalogGeneration,
+        coordinate: &SummaryInstanceCoordinates,
+        first_revision: u64,
+        revision: u64,
+    ) -> Result<bool, String> {
+        if self.generation.as_ref() != Some(generation) {
+            return Err("summary publication catalog generation differs".into());
+        }
+        let window = self
+            .windows
+            .get(coordinate)
+            .ok_or("summary publication was not admitted")?;
+        if revision == 0 || revision > window.admitted {
+            return Err("summary publication revision was not admitted".into());
+        }
+        if window.published >= revision {
+            return Ok(true);
+        }
+        if window.series_id.is_none() && self.published_series.len() >= Self::MAX_WINDOWS {
+            return Err("summary admission series capacity exceeded".into());
+        }
+        if self.revision == u64::MAX {
+            return Err("summary admission revision exhausted".into());
+        }
+        if !window.pending.contains(&revision) {
+            return Err("summary output revision was not admitted to this coordinate".into());
+        }
+        if first_revision == 0
+            || first_revision > revision
+            || window
+                .pending
+                .first()
+                .is_some_and(|pending| first_revision > *pending)
+        {
+            return Err("summary output omitted an earlier unpublished input revision".into());
+        }
+        Ok(false)
+    }
+
+    pub(super) fn acknowledge(
+        &mut self,
+        generation: &CatalogGeneration,
+        coordinate: &SummaryInstanceCoordinates,
+        revision: u64,
+    ) -> Result<(), String> {
+        if self.generation.as_ref() != Some(generation) {
+            return Err("summary publication catalog generation differs".into());
+        }
+        let window = self
+            .windows
+            .get_mut(coordinate)
+            .ok_or("summary publication was not admitted")?;
+        if revision == 0 || revision > window.admitted {
+            return Err("summary publication revision was not admitted".into());
+        }
+        window.published = window.published.max(revision);
+        let before = window.pending.len();
+        window.pending.retain(|pending| *pending > revision);
+        self.pending_revisions -= before - window.pending.len();
+        // A newer admission remains pending even if an old write completes now.
+        self.revision = self
+            .revision
+            .checked_add(1)
+            .ok_or("summary admission revision exhausted")?;
+        Ok(())
+    }
+
+    pub(super) fn record_series(
+        &mut self,
+        generation: &CatalogGeneration,
+        coordinate: &SummaryInstanceCoordinates,
+        series_id: u64,
+    ) -> Result<(), String> {
+        if self.generation.as_ref() != Some(generation) {
+            return Err("summary series publication catalog generation differs".into());
+        }
+        if !self.published_series.contains(&series_id)
+            && self.published_series.len() >= Self::MAX_WINDOWS
+        {
+            return Err("summary admission series capacity exceeded".into());
+        }
+        let window = self
+            .windows
+            .get_mut(coordinate)
+            .ok_or("published window was not admitted")?;
+        if window
+            .series_id
+            .is_some_and(|previous| previous != series_id)
+        {
+            return Err("summary coordinate changed series identity".into());
+        }
+        window.series_id = Some(series_id);
+        self.published_series.insert(series_id);
+        Ok(())
+    }
+
+    pub(super) fn seal_finite(&mut self, generation: &CatalogGeneration) -> Result<(), String> {
+        if self.generation.as_ref() != Some(generation) {
+            return Err("finite completion catalog generation differs".into());
+        }
+        if self
+            .windows
+            .values()
+            .any(|window| window.published < window.admitted)
+        {
+            return Err("finite source has unpublished summary windows".into());
+        }
+        self.finite_complete = true;
+        self.revision = self
+            .revision
+            .checked_add(1)
+            .ok_or("summary admission revision exhausted")?;
+        Ok(())
+    }
+
+    pub(super) fn known_empty(
+        &self,
+        definition: SummaryDefinitionId,
+        series_id: u64,
+        range: HalfOpenTimeRange,
+    ) -> bool {
+        self.finite_complete
+            && self.published_series.contains(&series_id)
+            && self.observed_extent.is_some_and(|extent| {
+                range.start_ms >= extent.start_ms && range.end_ms <= extent.end_ms
+            })
+            && self
+                .replay_floors
+                .get(&definition)
+                .is_none_or(|floor| range.start_ms >= *floor)
+            && !self.windows.iter().any(|(coordinate, state)| {
+                coordinate.summary_definition_id == definition
+                    && coordinate.time_range.start_ms < range.end_ms
+                    && coordinate.time_range.end_ms > range.start_ms
+                    && state.series_id == Some(series_id)
+            })
+    }
+
+    pub(super) fn revision(&self) -> u64 {
+        self.revision
+    }
+
+    pub(super) fn has_pending(
+        &self,
+        definition: SummaryDefinitionId,
+        range: HalfOpenTimeRange,
+    ) -> bool {
+        self.windows.iter().any(|(coordinate, state)| {
+            coordinate.summary_definition_id == definition
+                && coordinate.time_range.start_ms < range.end_ms
+                && coordinate.time_range.end_ms > range.start_ms
+                && state.published < state.admitted
+        })
+    }
+
+    /// Advancing this configured replay floor also rejects future old admission.
+    /// Pending work is never forgotten because a different series ran ahead.
+    pub(super) fn retire_completed_before(
+        &mut self,
+        definition: SummaryDefinitionId,
+        frontier_ms: i64,
+    ) {
+        let floor = self.replay_floors.entry(definition).or_insert(i64::MIN);
+        *floor = (*floor).max(frontier_ms);
+        let frontier_ms = *floor;
+        let pending: BTreeSet<_> = self
+            .windows
+            .values()
+            .flat_map(|state| state.pending.iter().copied())
+            .collect();
+        self.windows.retain(|coordinate, state| {
+            let remove = coordinate.summary_definition_id == definition
+                && coordinate.time_range.end_ms <= frontier_ms
+                && state.published >= state.admitted
+                && !pending.contains(&state.admitted);
+            if remove {
+                self.metadata_bytes = self
+                    .metadata_bytes
+                    .saturating_sub(Self::coordinate_bytes(coordinate));
+            }
+            !remove
+        });
+    }
+
+    fn coordinate_bytes(coordinate: &SummaryInstanceCoordinates) -> usize {
+        std::mem::size_of::<SummaryInstanceCoordinates>()
+            + coordinate
+                .group_values
+                .iter()
+                .map(|(key, value)| key.len() + value.len() + 64)
+                .sum::<usize>()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    fn generation(version: u64) -> CatalogGeneration {
+        CatalogGeneration {
+            schema_version: 1,
+            plan_id: 1,
+            plan_version: version,
+            snapshot_sha256: format!("snapshot-{version}"),
+        }
+    }
+    fn window(series: &str) -> SummaryInstanceCoordinates {
+        SummaryInstanceCoordinates {
+            summary_definition_id: SummaryDefinitionId(asap_types::PolicyFingerprint(7)),
+            time_range: HalfOpenTimeRange {
+                start_ms: 0,
+                end_ms: 1000,
+            },
+            group_values: BTreeMap::from([("instance".into(), series.into())]),
+        }
+    }
+
+    #[test]
+    fn fast_series_publication_cannot_hide_a_queued_series() {
+        let generation = generation(1);
+        let mut inventory = AdmissionInventory::default();
+        inventory.install(generation.clone());
+        let a = window("a");
+        let b = window("b");
+        let revision = inventory
+            .admit(&generation, BTreeSet::from([a.clone(), b.clone()]))
+            .unwrap();
+        inventory.acknowledge(&generation, &a, revision).unwrap();
+        assert!(inventory.has_pending(a.summary_definition_id, a.time_range));
+        inventory.acknowledge(&generation, &b, revision).unwrap();
+        assert!(!inventory.has_pending(a.summary_definition_id, a.time_range));
+    }
+
+    #[test]
+    fn old_publication_cannot_acknowledge_a_newer_update() {
+        let generation = generation(1);
+        let mut inventory = AdmissionInventory::default();
+        inventory.install(generation.clone());
+        let coordinate = window("a");
+        let first = inventory
+            .admit(&generation, BTreeSet::from([coordinate.clone()]))
+            .unwrap();
+        let second = inventory
+            .admit(&generation, BTreeSet::from([coordinate.clone()]))
+            .unwrap();
+        let before = inventory.revision();
+        inventory
+            .acknowledge(&generation, &coordinate, first)
+            .unwrap();
+        assert_ne!(before, inventory.revision());
+        assert!(inventory.has_pending(coordinate.summary_definition_id, coordinate.time_range));
+        inventory.retire_completed_before(coordinate.summary_definition_id, 1000);
+        assert_eq!(inventory.windows.len(), 1);
+        inventory
+            .acknowledge(&generation, &coordinate, second)
+            .unwrap();
+        inventory.retire_completed_before(coordinate.summary_definition_id, 1000);
+        assert!(inventory.windows.is_empty());
+    }
+
+    #[test]
+    fn later_output_cannot_hide_an_unpublished_earlier_input() {
+        let generation = generation(1);
+        let mut inventory = AdmissionInventory::default();
+        inventory.install(generation.clone());
+        let coordinate = window("a");
+        let first = inventory
+            .admit(&generation, BTreeSet::from([coordinate.clone()]))
+            .unwrap();
+        let second = inventory
+            .admit(&generation, BTreeSet::from([coordinate.clone()]))
+            .unwrap();
+        assert!(inventory
+            .validate_publication(&generation, &coordinate, second, second)
+            .is_err());
+        assert_eq!(
+            inventory.validate_publication(&generation, &coordinate, first, second),
+            Ok(false)
+        );
+        inventory
+            .acknowledge(&generation, &coordinate, first)
+            .unwrap();
+        assert_eq!(
+            inventory.validate_publication(&generation, &coordinate, second, second),
+            Ok(false)
+        );
+    }
+
+    #[test]
+    fn generation_switch_rejects_old_completion_and_admission() {
+        let mut inventory = AdmissionInventory::default();
+        let old = generation(1);
+        inventory.install(old.clone());
+        let coordinate = window("a");
+        let revision = inventory
+            .admit(&old, BTreeSet::from([coordinate.clone()]))
+            .unwrap();
+        inventory.install(generation(2));
+        assert!(inventory.acknowledge(&old, &coordinate, revision).is_err());
+        assert!(inventory.admit(&old, BTreeSet::from([coordinate])).is_err());
+    }
+
+    #[test]
+    fn only_finite_completion_proves_an_inactive_series_window_empty() {
+        let generation = generation(1);
+        let mut inventory = AdmissionInventory::default();
+        inventory.install(generation.clone());
+        let first = window("a");
+        let mut second = window("b");
+        second.time_range = HalfOpenTimeRange {
+            start_ms: 1000,
+            end_ms: 2000,
+        };
+        let revision = inventory
+            .admit(&generation, BTreeSet::from([first.clone(), second.clone()]))
+            .unwrap();
+        inventory.record_series(&generation, &first, 1).unwrap();
+        inventory
+            .acknowledge(&generation, &first, revision)
+            .unwrap();
+        assert!(inventory.seal_finite(&generation).is_err());
+        inventory.record_series(&generation, &second, 2).unwrap();
+        inventory
+            .acknowledge(&generation, &second, revision)
+            .unwrap();
+        assert!(!inventory.known_empty(first.summary_definition_id, 1, second.time_range));
+        inventory.seal_finite(&generation).unwrap();
+        assert!(inventory.known_empty(first.summary_definition_id, 1, second.time_range));
+        assert!(!inventory.known_empty(first.summary_definition_id, 2, second.time_range));
+        assert!(!inventory.known_empty(first.summary_definition_id, 999, second.time_range));
+        assert!(!inventory.known_empty(
+            first.summary_definition_id,
+            1,
+            HalfOpenTimeRange {
+                start_ms: 2000,
+                end_ms: 3000
+            }
+        ));
+        assert!(inventory
+            .admit(&generation, BTreeSet::from([first]))
+            .is_err());
+    }
+
+    #[test]
+    fn metadata_budget_rejects_admission_without_partial_mutation() {
+        let generation = generation(1);
+        let mut inventory = AdmissionInventory::default();
+        inventory.install(generation.clone());
+        let before = inventory.revision();
+        let mut coordinate = window("a");
+        coordinate.group_values.insert(
+            "large".into(),
+            "x".repeat(AdmissionInventory::MAX_METADATA_BYTES),
+        );
+        assert!(inventory
+            .admit(&generation, BTreeSet::from([window("b"), coordinate]))
+            .is_err());
+        assert!(inventory.windows.is_empty());
+        assert_eq!(before, inventory.revision());
+    }
+}
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 8353d646c..8a6074f5c 100644
--- a/data_plane/src/storage_engines/sketch_db/index/mod.rs
+++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs
@@ -557,8 +557,59 @@ impl From<&control_plane::physical::compiler::SummaryFrameIdentity> for Incomple
 /// `instances` is keyed under a `RwLock<HashMap>` because the registration
 /// rate is low (one write per first-seen sid) and reads dominate;
 /// `series` is a `DashMap` because per-sid writes happen on every DP.
+#[derive(Clone, Copy)]
+pub(crate) struct SummaryReadRevision {
+    admission: u64,
+    mutation: u64,
+    in_flight: usize,
+}
+impl SummaryReadRevision {
+    fn capture(
+        admission: u64,
+        mutation: &std::sync::atomic::AtomicU64,
+        active: &std::sync::atomic::AtomicUsize,
+        between_reads: impl FnOnce(),
+    ) -> Self {
+        use std::sync::atomic::Ordering::SeqCst;
+        let before = mutation.load(SeqCst);
+        between_reads();
+        let in_flight = active.load(SeqCst);
+        let after = mutation.load(SeqCst);
+        Self {
+            admission,
+            mutation: after,
+            in_flight: if before == after {
+                in_flight
+            } else {
+                in_flight.max(1)
+            },
+        }
+    }
+
+    pub(crate) fn matches(self, other: Self) -> bool {
+        self.in_flight == 0
+            && other.in_flight == 0
+            && self.admission == other.admission
+            && self.mutation == other.mutation
+    }
+}
+
+struct StateMutation<'a>(&'a SketchStore);
+impl Drop for StateMutation<'_> {
+    fn drop(&mut self) {
+        use std::sync::atomic::Ordering::SeqCst;
+        self.0.mutation_revision.fetch_add(1, SeqCst);
+        self.0.active_mutations.fetch_sub(1, SeqCst);
+    }
+}
+
 #[derive(Default)]
 pub struct SketchStore {
+    admission: RwLock<admission::AdmissionInventory>,
+    mutation_revision: std::sync::atomic::AtomicU64,
+    active_mutations: std::sync::atomic::AtomicUsize,
+    admitted_mutations: std::sync::atomic::AtomicU64,
+    finite_mutation_revision: std::sync::atomic::AtomicU64,
     /// sid → metadata. May contain ghost sids (registered identities
     /// whose state was merged away by an upstream gateway before
     /// reaching this backend).
@@ -747,9 +798,126 @@ impl SketchStore {
         &self,
         catalog: Arc<asap_types::summary_catalog::SummaryCatalog>,
     ) -> Result<(), String> {
+        let reference = catalog.reference().map_err(|error| error.to_string())?;
+        let mut inventory = self.admission.write().unwrap();
         self.descriptors
             .install_catalog(Arc::clone(&catalog))
-            .map_err(|error| error.to_string())
+            .map_err(|error| error.to_string())?;
+        inventory.install(CatalogGeneration {
+            schema_version: reference.schema_version,
+            plan_id: reference.plan_id,
+            plan_version: reference.plan_version,
+            snapshot_sha256: reference.snapshot_sha256,
+        });
+        Ok(())
+    }
+
+    pub(crate) fn admit_summary_updates(
+        &self,
+        generation: &CatalogGeneration,
+        coordinates: BTreeSet<asap_types::sds::SummaryInstanceCoordinates>,
+    ) -> Result<u64, String> {
+        let catalog = self
+            .descriptors
+            .authoritative_catalog()
+            .ok_or("summary admission requires an installed catalog")?;
+        if coordinates.iter().any(|coordinate| {
+            !catalog
+                .materializations
+                .contains_key(&coordinate.summary_definition_id)
+        }) {
+            return Err("summary admission references an uninstalled definition".into());
+        }
+        self.admission
+            .write()
+            .unwrap()
+            .admit(generation, coordinates)
+    }
+
+    pub(crate) fn publish_admitted_summary_update(
+        &self,
+        generation: &CatalogGeneration,
+        coordinate: &asap_types::sds::SummaryInstanceCoordinates,
+        first_revision: u64,
+        revision: u64,
+        replay_horizon_ms: u64,
+        persist: impl FnOnce() -> Option<u64>,
+    ) -> Result<(), String> {
+        // Fence installation and read validation across the state write: an old
+        // producer cannot mutate a new generation before its receipt is rejected.
+        let mut inventory = self.admission.write().unwrap();
+        if inventory.validate_publication(generation, coordinate, first_revision, revision)? {
+            return Ok(());
+        }
+        let series_id = persist().ok_or("summary state publication failed")?;
+        inventory.record_series(generation, coordinate, series_id)?;
+        inventory.acknowledge(generation, coordinate, revision)?;
+        self.admitted_mutations
+            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        let floor = coordinate
+            .time_range
+            .end_ms
+            .saturating_sub(i64::try_from(replay_horizon_ms).unwrap_or(i64::MAX));
+        inventory.retire_completed_before(coordinate.summary_definition_id, floor);
+        Ok(())
+    }
+
+    pub(crate) fn seal_finite_summary_input(
+        &self,
+        generation: &CatalogGeneration,
+    ) -> Result<(), String> {
+        use std::sync::atomic::Ordering::SeqCst;
+        let mutation = self.mutation_revision.load(SeqCst);
+        if self.active_mutations.load(SeqCst) != 0
+            || mutation != self.admitted_mutations.load(SeqCst)
+        {
+            return Err("finite summary completion cannot certify untracked state writes".into());
+        }
+        self.admission.write().unwrap().seal_finite(generation)?;
+        self.finite_mutation_revision.store(mutation, SeqCst);
+        Ok(())
+    }
+
+    pub(crate) fn summary_window_known_empty(
+        &self,
+        definition: SummaryDefinitionId,
+        series_id: u64,
+        range: HalfOpenTimeRange,
+    ) -> bool {
+        use std::sync::atomic::Ordering::SeqCst;
+        self.active_mutations.load(SeqCst) == 0
+            && self.finite_mutation_revision.load(SeqCst) == self.mutation_revision.load(SeqCst)
+            && self
+                .admission
+                .read()
+                .unwrap()
+                .known_empty(definition, series_id, range)
+    }
+
+    pub(crate) fn summary_update_revision(&self) -> SummaryReadRevision {
+        SummaryReadRevision::capture(
+            self.admission.read().unwrap().revision(),
+            &self.mutation_revision,
+            &self.active_mutations,
+            || {},
+        )
+    }
+
+    fn begin_state_mutation(&self) -> StateMutation<'_> {
+        self.active_mutations
+            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        StateMutation(self)
+    }
+
+    pub(crate) fn has_pending_summary_updates(
+        &self,
+        definition: SummaryDefinitionId,
+        range: HalfOpenTimeRange,
+    ) -> bool {
+        self.admission
+            .read()
+            .unwrap()
+            .has_pending(definition, range)
     }
 
     /// Record that `sid` is a per-item (item_label-mode) frequency sketch
@@ -1025,6 +1193,7 @@ impl SketchStore {
         window: TimestampRange,
         sample: SketchSampleState,
     ) {
+        let _mutation = self.begin_state_mutation();
         let store = self
             .series
             .entry(sid)
@@ -1067,6 +1236,7 @@ impl SketchStore {
         window: TimestampRange,
         payload: Box<dyn crate::storage_engines::types::AggregateCore>,
     ) {
+        let _mutation = self.begin_state_mutation();
         let max_value = payload
             .as_any()
             .downcast_ref::<crate::precompute_engine::operators::MinMaxAccumulator>()
@@ -2243,6 +2413,7 @@ impl SketchStore {
     /// operator / debug-endpoint use so eviction can be observed in
     /// e2e tests without waiting out retirement retention.
     pub fn force_expire(&self, sid: u64) -> Option<Arc<SketchInstanceMetadata>> {
+        let _mutation = self.begin_state_mutation();
         let mut map = self.instances.write().ok()?;
         let instance = map.get_mut(&sid)?;
         let meta = Arc::make_mut(&mut instance.metadata);
@@ -2268,6 +2439,7 @@ impl SketchStore {
     /// (it is independently keyed and not part of the metadata-index
     /// invariant).
     pub fn remove_instance(&self, sid: u64) -> Option<Arc<SketchInstanceMetadata>> {
+        let _mutation = self.begin_state_mutation();
         let removed = {
             // Fixed lock order: instances → policy_to_series_ids → metric_to_series_ids.
             let mut instances = self.instances.write().ok()?;
@@ -2954,6 +3126,38 @@ mod tests {
         }
     }
 
+    #[test]
+    fn completing_writer_between_revision_loads_cannot_certify_a_snapshot() {
+        let store = SketchStore::new();
+        let before = store.summary_update_revision();
+        let writer = store.begin_state_mutation();
+        let after = SummaryReadRevision::capture(
+            0,
+            &store.mutation_revision,
+            &store.active_mutations,
+            || drop(writer),
+        );
+        assert!(!before.matches(after));
+        assert!(!after.matches(after), "capture crossed a writer completion");
+    }
+
+    #[test]
+    fn direct_store_writes_invalidate_query_snapshots_even_without_admission() {
+        let store = SketchStore::new();
+        let before = store.summary_update_revision();
+        let mutation = store.begin_state_mutation();
+        let during = store.summary_update_revision();
+        assert!(
+            !during.matches(during),
+            "an in-flight write cannot certify a snapshot"
+        );
+        drop(mutation);
+        assert!(!before.matches(store.summary_update_revision()));
+        let before = store.summary_update_revision();
+        store.append_sample(1, BTreeMap::new(), (0, 1000), sample(1));
+        assert!(!before.matches(store.summary_update_revision()));
+    }
+
     #[test]
     fn store_lookups_and_equivalent_sids_share_sds_allocations() {
         let store = SketchStore::new();
@@ -4876,6 +5080,7 @@ mod tests {
 
 // 2026-05 reorg: generic epoch-partitioned columnar storage lives
 // alongside the store that uses it.
+mod admission;
 pub mod epoch_columnar;
 
 // `persistence` moved up to `sketch_db::persistence`. Re-exported here
diff --git a/data_plane/src/storage_engines/types/precomputed_output.rs b/data_plane/src/storage_engines/types/precomputed_output.rs
index 596fc03d7..5a84cc215 100644
--- a/data_plane/src/storage_engines/types/precomputed_output.rs
+++ b/data_plane/src/storage_engines/types/precomputed_output.rs
@@ -42,8 +42,18 @@ pub enum Origin {
 /// `AggregationConfig::aggregation_id`; this PR completes the cleanup
 /// by retiring the field on `PrecomputedOutput` too. Sinks resolve the
 /// source config via `PolicyRegistry::get(policy_fp)`.
+/// In-process proof of which accepted input revision a window includes.
+#[derive(Debug, Clone)]
+pub struct SummaryInputRevision {
+    pub generation: std::sync::Arc<asap_types::sds::CatalogGeneration>,
+    pub revision: u64,
+    pub first_revision: u64,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct PrecomputedOutput {
+    #[serde(skip)]
+    pub input_revision: Option<std::sync::Arc<SummaryInputRevision>>,
     pub start_timestamp: u64,
     pub end_timestamp: u64,
     pub key: Option<KeyByLabelValues>,
@@ -81,6 +91,7 @@ impl PrecomputedOutput {
         policy_fp: PolicyFingerprint,
     ) -> Self {
         Self {
+            input_revision: None,
             start_timestamp,
             end_timestamp,
             key,
@@ -107,6 +118,7 @@ impl PrecomputedOutput {
         policy_fp: PolicyFingerprint,
     ) -> Self {
         Self {
+            input_revision: None,
             start_timestamp,
             end_timestamp,
             key,
diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs
index 502960ae4..24482f17c 100644
--- a/data_plane/tests/asapquery_compatibility_process_e2e.rs
+++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs
@@ -84,7 +84,7 @@ async fn remote_write(client: &reqwest::Client, base: &str, request: &WriteReque
     let body = snap::raw::Encoder::new()
         .compress_vec(&request.encode_to_vec())
         .expect("snappy encode");
-    client
+    let response = client
         .post(format!("{base}/api/v1/write"))
         .header("content-encoding", "snappy")
         .header("content-type", "application/x-protobuf")
@@ -92,9 +92,15 @@ async fn remote_write(client: &reqwest::Client, base: &str, request: &WriteReque
         .body(body)
         .send()
         .await
-        .expect("send Remote Write")
-        .status()
-        .as_u16()
+        .expect("send Remote Write");
+    let status = response.status().as_u16();
+    if status >= 400 {
+        eprintln!(
+            "Remote Write {status}: {}",
+            response.text().await.unwrap_or_default()
+        );
+    }
+    status
 }
 
 async fn drain_precompute(client: &reqwest::Client, backend: &str) {
@@ -1577,9 +1583,19 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
     let planned_snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot =
         serde_json::from_str(&std::fs::read_to_string(snapshot).unwrap()).unwrap();
     let planned = planned_snapshot.compile().unwrap();
-    // All four installed states, including the reset-aware counter state, have
-    // closed phase-aligned panes and are available to their query bindings.
-    assert_eq!(materializations.len(), 4, "{materializations:?}");
+    // Every selected state must be serving; the Planner may share or separate
+    // physical populations, so compare identities rather than a frozen count.
+    let expected = planned
+        .precompute_plan
+        .materializations
+        .iter()
+        .map(|config| config.policy_fp_u64())
+        .collect::<std::collections::BTreeSet<_>>();
+    let actual = materializations
+        .iter()
+        .map(|entry| entry["materialization"].as_u64().unwrap())
+        .collect::<std::collections::BTreeSet<_>>();
+    assert_eq!(actual, expected, "{materializations:?}");
     for entry in materializations {
         assert_eq!(entry["phase"], "serving", "{entry}");
         assert!(
diff --git a/docs/design_docs/continuous-summary-completeness.md b/docs/design_docs/continuous-summary-completeness.md
new file mode 100644
index 000000000..4be73e780
--- /dev/null
+++ b/docs/design_docs/continuous-summary-completeness.md
@@ -0,0 +1,13 @@
+# Summary publication completeness for accepted input
+
+The SummaryStore owns an in-memory admission inventory keyed by catalog generation, summary definition, group population and physical time window. Remote Write reserves every worker queue slot before admitting any coordinates, then sends the same immutable input revision with the queued work. Queue rejection leaves neither messages nor admission records behind.
+
+Workers carry the first and last consumed input revisions through materialized source and maintenance DAG outputs. State writes validate the generation and consumed revisions before modifying SummaryStore; a later output cannot acknowledge an earlier missing update. Corrections within one admitted coordinate are combined before publication. Dropped admitted input leaves the coordinate incomplete. Accepted raw-sample throughput counts once per request, without materialization fanout; materialized-output throughput counts new successful store publications, without receipt retries.
+
+A query rejects overlapping admitted but unpublished input. The whole QueryPlan DAG is fenced by the store revision, so branches cannot combine different publication snapshots. Both direct sketch writes (including OTLP) and exact-state writes (including SQL backfill) participate in the mutation fence; in-flight writes cannot certify a read snapshot. Finite absence proof also requires that every state mutation was admitted. The initial fence is global: unrelated updates can cause conservative exact fallback. Read-set-scoped revisions are required before claiming sustained continuous-query performance.
+
+Finite drain closes input, waits for all workers and certifies that every accepted coordinate published. Only this closed-input proof can establish that a known series has no samples in a retained window. Missing state, unknown series, unknown time coverage and incomplete coordinates never imply an empty result. Live Remote Write provides no source watermark, so the backend does not infer global event-time completeness from the fastest series.
+
+Admission metadata has bounded coordinate, pending-revision and byte budgets. Completed receipts expire with configured materialization retention; pending work and the published prefixes of pending admissions remain protected. Already admitted slow-worker outputs can finish behind another worker's maintenance replay frontier. Unsolicited expired input and expired untagged replay remain rejected.
+
+This inventory is not durable and does not establish exactly-once execution across crashes. Startup installs the authoritative catalog before persistence recovery, but persisted series metadata still needs a separate migration to preserve summary definition identity and catalog provenance. General multi-input maintenance transforms and durable producer watermarks remain separate work.