diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index 6dd6b7f91..dce551793 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -2790,6 +2790,28 @@ mod tests {
         assert_eq!(plan.query_plan.entries.len(), 1);
     }
 
+    #[test]
+    fn compatibility_demo_snapshot_compiles_the_complete_query_matrix() {
+        let source =
+            include_str!("../../../docs/examples/asapquery-compatibility-demo-snapshot.json");
+        let snapshot: BackendLocalPlanningSnapshot =
+            serde_json::from_str(source).expect("strict compatibility demo fixture");
+        let plan = snapshot.compile().expect("compatibility demo compiles");
+
+        assert!(plan.collector_plans.is_empty());
+        assert!(plan.transmission_plan.rules.is_empty());
+        assert_eq!(plan.query_plan.entries.len(), 4);
+        assert_eq!(plan.precompute_plan.materializations.len(), 3);
+        for query in [
+            "rate(asap_demo_counter_total[5s])",
+            "increase(asap_demo_counter_total[5s])",
+            "sum_over_time(asap_demo_gauge[5s])",
+            "quantile_over_time(0.5, asap_demo_latency_ms[5s])",
+        ] {
+            assert!(plan.query_plan.lookup(query).is_ok(), "missing {query}");
+        }
+    }
+
     #[test]
     fn multiple_readouts_share_one_precompute_materialization() {
         let mut planning_request = request("q-p90", "quantile_over_time(0.90, m[1m])");
diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
index b04fd2885..07901031f 100644
--- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs
+++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
@@ -461,13 +461,7 @@ fn route_messages(
                 })
                 .collect();
             let attrs_fp = super::canonical_attrs_fingerprint(&grouping_pairs);
-            let agg_kind = crate::storage_engines::sketch_db::data::AggKind::ExactAgg {
-                agg_type: config.aggregation_type,
-                parameters_canonical: crate::storage_engines::sketch_db::data::canonical_parameters(
-                    &config.parameters,
-                ),
-                spatial_filter_canonical: config.spatial_filter_normalized.clone(),
-            };
+            let agg_kind = crate::storage_engines::sketch_db::data::agg_kind_for_config(config);
             let sid = ingest.series_resolver.resolve(
                 &config.metric,
                 &attrs_fp,
diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs
index fac8b854c..37d75e43c 100644
--- a/data_plane/src/drivers/query/servers/http.rs
+++ b/data_plane/src/drivers/query/servers/http.rs
@@ -749,6 +749,22 @@ async fn process_query_request(
 /// single-target metrics keep their original semantics — every shape
 /// resolves to the one configured backend.
 fn resolve_metric_storage(state: &AppState, query: &str, tenant: &str) -> StorageBackend {
+    // A non-bootstrap atomic PhysicalPlan owns routing. Every request first
+    // enters the ASAP engine, where QueryPlan lookup either executes its
+    // compiler-bound DAG or returns an explicit fallback reason. Consulting
+    // the legacy shape/SID candidate heuristics here would bypass QueryPlan
+    // (and can also discard the request's explicit evaluation timestamp).
+    if state.active_physical_plan.as_ref().is_some_and(|active| {
+        let snapshot = active.snapshot();
+        snapshot.query_plan.plan_id != 0 && !snapshot.query_plan.entries.is_empty()
+    }) {
+        debug!(
+            tenant,
+            query, "resolve_metric_storage: active QueryPlan owns warm/fallback routing"
+        );
+        return StorageBackend::SketchStore;
+    }
+
     if let Some(routing_handle) = state.backend_storage_routing.as_ref() {
         // Phase α: snapshot the hot-reload handle once per request,
         // scoped to this request's tenant. The snapshot resolves to
@@ -1197,11 +1213,10 @@ async fn process_via_simple_engine(
                 Err(status) => status.into_response(),
             }
         }
-        Err(_) => {
+        Err(error) => {
             debug!(
-                "Modern execute() returned CapabilityMiss for query='{}', \
-                 falling through to fallback / unsupported",
-                parsed_request.query
+                "Modern execute() returned {error} for query='{}', falling through to fallback / unsupported",
+                parsed_request.query,
             );
             let total_duration = start_time.elapsed();
             debug!(
@@ -1971,7 +1986,7 @@ async fn process_range_query_request(
 
     let router_result = state
         .query_router
-        .execute_range_for_tier(
+        .execute_range_for_tier_routed(
             &parsed_request.query,
             stat,
             accuracy,
@@ -1984,7 +1999,7 @@ async fn process_range_query_request(
         .await;
 
     match router_result {
-        Ok(query_result) => {
+        Ok((query_result, data_source_id)) => {
             let query_duration = query_start_time.elapsed();
             debug!(
                 "EngineRouter range dispatch took: {:.2}ms",
@@ -2003,7 +2018,9 @@ async fn process_range_query_request(
                 )
                 .await
             {
-                Ok(response) => response.into_response(),
+                Ok(response) => {
+                    annotate_data_source(response.into_response(), data_source_id).await
+                }
                 Err(status) => status.into_response(),
             }
         }
diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs
index 3cd03239c..b139f7411 100644
--- a/data_plane/src/precompute_engine/output_sink.rs
+++ b/data_plane/src/precompute_engine/output_sink.rs
@@ -241,7 +241,7 @@ impl OutputSink for NoopOutputSink {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::precompute_engine::operators::SumAccumulator;
+    use crate::precompute_engine::operators::{DDSketchAccumulator, SumAccumulator};
     use crate::storage_engines::sketch_db::index::{AggKind, SidLookup};
     use crate::storage_engines::types::{KeyByLabelValues, StreamingConfig};
     use asap_types::aggregation_config::AggregationConfig;
@@ -332,6 +332,48 @@ mod tests {
         );
     }
 
+    #[test]
+    fn sketch_policy_is_registered_and_stored_as_sketch_state() {
+        let mut cfg = sum_agg_config(8, "latency", &[]);
+        cfg.aggregation_type = AggregationType::DDSketch;
+        cfg.parameters
+            .insert("alpha".into(), serde_json::json!(0.01));
+        let policy_fp = cfg.policy_fp_u64();
+        let hot_reload =
+            HotReloadStreamingConfig::new(StreamingConfig::new(HashMap::from([(policy_fp, cfg)])));
+        let sketch_index = Arc::new(SketchStore::new());
+        let sink = SketchStoreSink::new(
+            sketch_index.clone(),
+            hot_reload,
+            Arc::new(SeriesIdResolver::new()),
+        );
+        let mut accumulator = DDSketchAccumulator::new(0.01);
+        accumulator.inner.update(42.0);
+
+        sink.emit_batch(vec![(
+            PrecomputedOutput::new(1_000, 2_000, None, asap_types::PolicyFingerprint(policy_fp)),
+            Box::new(accumulator),
+        )])
+        .expect("emit sketch");
+
+        let meta = sketch_index
+            .list_by_status(crate::storage_engines::sketch_db::lifecycle::AggStatus::Active)
+            .into_iter()
+            .next()
+            .expect("registered sketch SID");
+        assert!(matches!(
+            meta.agg_kind,
+            AggKind::Sketch {
+                algorithm: crate::storage_engines::sketch_db::data::SketchAlgorithm::DDSketch,
+                ..
+            }
+        ));
+        assert_eq!(sketch_index.query_range(meta.sid, 1_000, 2_000).len(), 1);
+        assert!(sketch_index
+            .query_exact_agg_range(meta.sid, 1_000, 2_000)
+            .is_empty());
+    }
+
     #[test]
     fn sketch_index_sink_skips_unknown_agg_id_gracefully() {
         // Streaming config does NOT contain agg_id=99 — the sink
diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs
index 72828eeed..37aa7009c 100644
--- a/data_plane/src/precompute_engine/worker.rs
+++ b/data_plane/src/precompute_engine/worker.rs
@@ -395,7 +395,15 @@ impl Worker {
         }
         let state = self.group_states.get_mut(&sid).unwrap();
 
-        // Find the max timestamp in this batch to advance the watermark
+        // Find the timestamp span in this batch. A first batch may contain
+        // several windows (Prometheus commonly sends catch-up samples after
+        // startup), so its minimum timestamp is also the initial closure
+        // scan boundary.
+        let batch_min_ts = samples
+            .iter()
+            .map(|(_, ts, _)| *ts)
+            .min()
+            .unwrap_or(i64::MIN);
         let batch_max_ts = samples
             .iter()
             .map(|(_, ts, _)| *ts)
@@ -475,9 +483,14 @@ impl Worker {
         }
 
         // Check for closed windows
+        let closure_scan_start = if previous_closure_watermark == i64::MIN {
+            batch_min_ts
+        } else {
+            previous_closure_watermark
+        };
         let closed = state
             .window_manager
-            .closed_windows(previous_closure_watermark, event_watermark);
+            .closed_windows(closure_scan_start, event_watermark);
 
         for window_start in &closed {
             let (_, window_end) = state.window_manager.window_bounds(*window_start);
@@ -2413,6 +2426,53 @@ aggregations:
         assert_eq!(emitted[0].0.end_timestamp, 10_000);
     }
 
+    #[test]
+    fn first_catch_up_batch_closes_every_complete_window() {
+        let config = make_agg_config(
+            1,
+            "cpu",
+            AggregationType::SingleSubpopulation,
+            "Sum",
+            5,
+            0,
+            vec![],
+        );
+        let sink = Arc::new(CapturingOutputSink::new());
+        let mut worker = make_worker_with_lateness(
+            HashMap::from([(1, config)]),
+            sink.clone(),
+            false,
+            0,
+            LateDataPolicy::Drop,
+            0,
+        );
+
+        worker
+            .process_group_samples(
+                1,
+                PolicyFingerprint(1),
+                "",
+                group_samples(
+                    "cpu",
+                    vec![
+                        (500, 1.0),
+                        (4_200, 2.0),
+                        (5_400, 3.0),
+                        (9_400, 4.0),
+                        (10_500, 5.0),
+                    ],
+                ),
+            )
+            .unwrap();
+
+        let emitted = sink.drain();
+        assert_eq!(emitted.len(), 2);
+        assert_eq!(emitted[0].0.start_timestamp, 0);
+        assert_eq!(emitted[0].0.end_timestamp, 5_000);
+        assert_eq!(emitted[1].0.start_timestamp, 5_000);
+        assert_eq!(emitted[1].0.end_timestamp, 10_000);
+    }
+
     #[test]
     fn test_flush_publishes_worker_watermark() {
         let config = make_agg_config(
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 cdaa7a4fb..b9a83c797 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
@@ -251,7 +251,7 @@ fn execute_physical_query_plan(
         },
     };
     let output = physical_dag::execute(entry, &runtime)
-        .map_err(|error| LoweringSkip::ExecuteFailed(error.to_string()))?;
+        .map_err(|error| LoweringSkip::ExecuteFailed(format!("{error:?}")))?;
     match output {
         PhysicalQueryOutput::Value(values) => {
             let mut coverage = None;
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 3510bc9ff..025005cf2 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
@@ -369,9 +369,10 @@ impl QueryExecutionContext<'_> {
         let mut sids = self.index.sids_for_policy(binding.materialization);
         sids.sort_unstable();
         sids.dedup();
+        let mut matched_metadata = 0usize;
         let mut by_group: BTreeMap<BTreeMap<String, String>, Vec<GroupState>> = BTreeMap::new();
 
-        for sid in sids {
+        for sid in sids.iter().copied() {
             let candidate = self
                 .index
                 .with_instance(sid, |meta| {
@@ -390,6 +391,7 @@ impl QueryExecutionContext<'_> {
                 })
                 .flatten();
             let Some(candidate) = candidate else { continue };
+            matched_metadata += 1;
             match candidate {
                 Candidate::Sketch(kind) => {
                     let Some(series) = self
@@ -432,6 +434,15 @@ impl QueryExecutionContext<'_> {
             }
         }
         if by_group.is_empty() {
+            tracing::debug!(
+                metric = %binding.metric,
+                materialization = %binding.materialization,
+                ?sids,
+                matched_metadata,
+                t0_ms = self.t0_ms,
+                t1_ms = self.t1_ms,
+                "bound materialization produced no readable state"
+            );
             return Err(SummaryExecutorError::NoCandidates);
         }
         by_group
diff --git a/data_plane/src/query_engines/routing/query_engine_routing.rs b/data_plane/src/query_engines/routing/query_engine_routing.rs
index 59a6ab955..96a21636e 100644
--- a/data_plane/src/query_engines/routing/query_engine_routing.rs
+++ b/data_plane/src/query_engines/routing/query_engine_routing.rs
@@ -366,6 +366,35 @@ impl EngineRouter {
         step_ms: u64,
         tier: RangeTier,
     ) -> Result<QueryResult, EngineRouterError> {
+        self.execute_range_for_tier_routed(
+            query,
+            stat,
+            accuracy,
+            metric_storage,
+            start_ms,
+            end_ms,
+            step_ms,
+            tier,
+        )
+        .await
+        .map(|(result, _)| result)
+    }
+
+    /// Range dispatch with the identity of the engine that produced the
+    /// successful result. Transport adapters use this to annotate responses
+    /// without guessing whether the router took its fallback leg.
+    #[allow(clippy::too_many_arguments)]
+    pub async fn execute_range_for_tier_routed(
+        &self,
+        query: &str,
+        stat: Statistic,
+        accuracy: AccuracyTarget,
+        metric_storage: StorageBackend,
+        start_ms: u64,
+        end_ms: u64,
+        step_ms: u64,
+        tier: RangeTier,
+    ) -> Result<(QueryResult, &'static str), EngineRouterError> {
         let mut backends = compatible_storage_backends(stat, &accuracy, metric_storage);
         if matches!(tier, RangeTier::WarmOnly) {
             // Drop the archive leg: a range fully inside warm retention
@@ -405,7 +434,7 @@ impl EngineRouter {
                         backend = ?backend,
                         "router: range dispatch succeeded",
                     );
-                    return Ok(result);
+                    return Ok((result, id));
                 }
                 Err(e) => {
                     warn!(
diff --git a/data_plane/src/storage_engines/sketch_db/data/mod.rs b/data_plane/src/storage_engines/sketch_db/data/mod.rs
index 7e12e4a19..1d8aec4d8 100644
--- a/data_plane/src/storage_engines/sketch_db/data/mod.rs
+++ b/data_plane/src/storage_engines/sketch_db/data/mod.rs
@@ -128,6 +128,106 @@ pub enum AggKind {
     },
 }
 
+/// Resolve the physical state family produced by a precompute policy. This is
+/// shared by SID minting and store registration so a sketch policy can never
+/// be minted as `ExactAgg` and later registered as `Sketch` (or vice versa).
+pub fn agg_kind_for_config(config: &asap_types::aggregation_config::AggregationConfig) -> AggKind {
+    use planner_types::post_asap::{SketchAlgorithm as Algorithm, SketchParams, SummaryFamilyType};
+
+    // HLL is intentionally absent from raw-value accumulator dispatch because
+    // it arrives through SketchEnvelope ingest. It is still a sketch for SID
+    // identity and store registration, so classify it before consulting the
+    // accumulator factory contract.
+    let sketch = (config.aggregation_type == AggregationType::HLL)
+        .then(|| {
+            let precision = config
+                .parameters
+                .get("precision")
+                .or_else(|| config.parameters.get("p"))
+                .and_then(serde_json::Value::as_u64)
+                .and_then(|value| u32::try_from(value).ok())
+                .unwrap_or(14);
+            AggKind::Sketch {
+                algorithm: SketchAlgorithm::Hll,
+                config: SketchConfig::Hll { precision },
+                spatial_filter_canonical: config.spatial_filter_normalized.clone(),
+            }
+        })
+        .or_else(|| {
+            config.accumulator_spec().ok().and_then(|spec| {
+                let SummaryFamilyType::Sketch(kind, _) = spec.family else {
+                    return None;
+                };
+                let algorithm = kind.algorithm().clone();
+                let physical = match (kind.algorithm(), kind.params()) {
+                    (Algorithm::DDSketch, SketchParams::DDSketch { alpha }) => {
+                        SketchConfig::DDSketch {
+                            relative_accuracy: *alpha,
+                        }
+                    }
+                    (Algorithm::Kll, SketchParams::Kll { k }) => SketchConfig::Kll { k: *k },
+                    (Algorithm::Hll, SketchParams::Hll { precision }) => SketchConfig::Hll {
+                        precision: (*precision).into(),
+                    },
+                    (Algorithm::Cms, SketchParams::Cms { width, depth })
+                    | (Algorithm::CmsWithHeap, SketchParams::CmsWithHeap { width, depth, .. }) => {
+                        SketchConfig::CountMin {
+                            rows: *depth as i32,
+                            cols: *width as i32,
+                        }
+                    }
+                    (Algorithm::CountSketch, SketchParams::CountSketch { width, depth })
+                    | (
+                        Algorithm::CountSketchWithHeap,
+                        SketchParams::CountSketchWithHeap { width, depth, .. },
+                    ) => SketchConfig::CountSketch {
+                        rows: *depth as i32,
+                        cols: *width as i32,
+                    },
+                    _ => return None,
+                };
+                Some(AggKind::Sketch {
+                    algorithm,
+                    config: physical,
+                    spatial_filter_canonical: config.spatial_filter_normalized.clone(),
+                })
+            })
+        });
+
+    sketch.unwrap_or_else(|| AggKind::ExactAgg {
+        agg_type: config.aggregation_type,
+        parameters_canonical: canonical_parameters(&config.parameters),
+        spatial_filter_canonical: config.spatial_filter_normalized.clone(),
+    })
+}
+
+impl AggKind {
+    /// Runtime query capability and accuracy metadata implied by this state.
+    pub fn capability_and_accuracy(&self) -> (Capability, Option<AccuracyBound>) {
+        match self {
+            Self::ExactAgg { agg_type, .. } => (Capability::ExactAgg(*agg_type), None),
+            Self::Sketch {
+                algorithm, config, ..
+            } => {
+                let capability = match algorithm {
+                    SketchAlgorithm::DDSketch | SketchAlgorithm::Kll => {
+                        Capability::QuantileApprox(Some(algorithm.clone()))
+                    }
+                    SketchAlgorithm::Hll => Capability::CardinalityApprox,
+                    SketchAlgorithm::Cms | SketchAlgorithm::CountSketch => {
+                        Capability::FrequencyEstimate(Some(algorithm.clone()))
+                    }
+                    SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap => {
+                        Capability::FrequencyTopk(Some(algorithm.clone()))
+                    }
+                    SketchAlgorithm::Kmv | SketchAlgorithm::Theta => Capability::CardinalityApprox,
+                };
+                (capability, Some(AccuracyBound::from_config(config)))
+            }
+        }
+    }
+}
+
 /// Render a `HashMap<String, Value>` of parameters into the canonical
 /// string form `AggKind::ExactAgg::parameters_canonical` expects.
 /// Keys sorted lexicographically; each value via `serde_json`.
@@ -406,3 +506,40 @@ impl AggPayload {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use asap_types::{enums::WindowKind, KeyByLabelNames};
+    use std::collections::HashMap;
+
+    #[test]
+    fn hll_envelope_config_is_registered_as_a_sketch() {
+        let config = asap_types::aggregation_config::AggregationConfig::new(
+            AggregationType::HLL,
+            String::new(),
+            HashMap::from([("precision".to_string(), serde_json::json!(12))]),
+            KeyByLabelNames::empty(),
+            KeyByLabelNames::empty(),
+            KeyByLabelNames::empty(),
+            String::new(),
+            60,
+            60,
+            WindowKind::Tumbling,
+            String::new(),
+            "unique_users".to_string(),
+            None,
+            None,
+            None,
+        );
+
+        assert!(matches!(
+            agg_kind_for_config(&config),
+            AggKind::Sketch {
+                algorithm: SketchAlgorithm::Hll,
+                config: SketchConfig::Hll { precision: 12 },
+                ..
+            }
+        ));
+    }
+}
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 533658730..02f028040 100644
--- a/data_plane/src/storage_engines/sketch_db/index/mod.rs
+++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs
@@ -1855,16 +1855,7 @@ impl SketchStore {
         // samples by sid up-front) skip the resolver round-trip by
         // invoking the sid-direct sibling.
         let (attrs_fp, _label_values_map) = build_attrs_fp_and_label_map(agg_cfg, output);
-        let agg_kind = AggKind::ExactAgg {
-            agg_type: agg_cfg.aggregation_type,
-            parameters_canonical: canonical_parameters(&agg_cfg.parameters),
-            // The canonical spatial-filter participates in sid identity
-            // so filter-distinct policies don't collide on the same
-            // (metric, attrs, agg_kind) tuple. `spatial_filter_normalized`
-            // is the canonicalized form produced by
-            // `asap_types::utils::normalize_spatial_filter`.
-            spatial_filter_canonical: agg_cfg.spatial_filter_normalized.clone(),
-        };
+        let agg_kind = crate::storage_engines::sketch_db::data::agg_kind_for_config(agg_cfg);
         // Sid mint delegated to the caller's closure — typically
         // `|m, fp, ak| series_resolver.resolve(m, fp, ak)`. Keeps the
         // SketchStore free of any layer-inverted dependency on the
@@ -1899,11 +1890,8 @@ impl SketchStore {
     ) -> Option<u64> {
         let (_attrs_fp, label_values_map) = build_attrs_fp_and_label_map(agg_cfg, output);
         let key_names = &agg_cfg.grouping_labels.labels;
-        let agg_kind = AggKind::ExactAgg {
-            agg_type: agg_cfg.aggregation_type,
-            parameters_canonical: canonical_parameters(&agg_cfg.parameters),
-            spatial_filter_canonical: agg_cfg.spatial_filter_normalized.clone(),
-        };
+        let agg_kind = crate::storage_engines::sketch_db::data::agg_kind_for_config(agg_cfg);
+        let (capability, accuracy) = agg_kind.capability_and_accuracy();
 
         match self.instance(sid) {
             None => {
@@ -1920,9 +1908,9 @@ impl SketchStore {
                     sid,
                     metric_name: agg_cfg.metric.clone(),
                     group_by_keys,
-                    capability: Some(Capability::ExactAgg(agg_cfg.aggregation_type)),
+                    capability: Some(capability),
                     agg_kind,
-                    accuracy: None,
+                    accuracy,
                     first_seen_unix_ms: output.start_timestamp as i64,
                     retired_at_ms: None,
                     expires_at_ms: None,
@@ -1941,12 +1929,23 @@ impl SketchStore {
         }
 
         let window = (output.start_timestamp, output.end_timestamp);
-        self.append_precompute(
-            sid,
-            label_values_map,
-            window,
-            accumulator.clone_boxed_core(),
-        );
+        match crate::storage_engines::sketch_db::data::agg_kind_for_config(agg_cfg) {
+            AggKind::Sketch { .. } => self.append_sample(
+                sid,
+                label_values_map,
+                window,
+                SketchSampleState {
+                    bytes: accumulator.serialize_to_bytes(),
+                    encoding: SketchEncoding::MsgpackFull,
+                },
+            ),
+            AggKind::ExactAgg { .. } => self.append_precompute(
+                sid,
+                label_values_map,
+                window,
+                accumulator.clone_boxed_core(),
+            ),
+        }
         Some(sid)
     }
 
diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs
new file mode 100644
index 000000000..eefc1b34c
--- /dev/null
+++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs
@@ -0,0 +1,487 @@
+//! Black-box acceptance test for the collector-free ASAPQuery profile.
+//!
+//! Starts the production binary from a canonical workload snapshot, ingests
+//! only Prometheus Remote Write v1, exercises every declared warm query family
+//! through instant and range APIs, and verifies exact fallback request parity.
+
+use std::collections::HashMap;
+use std::net::TcpListener;
+use std::process::{Child, Command, Stdio};
+use std::sync::Arc;
+use std::time::Duration;
+
+use axum::{extract::Query, http::HeaderMap, routing::get, Json, Router};
+use data_plane::drivers::ingest::prometheus_remote_write::{
+    Label, Sample, TimeSeries, WriteRequest,
+};
+use prost::Message;
+use serde_json::Value;
+use tokio::sync::Mutex;
+
+struct ChildGuard(Child);
+
+impl Drop for ChildGuard {
+    fn drop(&mut self) {
+        let _ = self.0.kill();
+        let _ = self.0.wait();
+    }
+}
+
+fn unused_port() -> u16 {
+    let listener = TcpListener::bind("127.0.0.1:0").expect("reserve loopback port");
+    listener.local_addr().expect("read loopback address").port()
+}
+
+async fn wait_until_ready(client: &reqwest::Client, url: &str, child: &mut Child) {
+    for _ in 0..120 {
+        if let Some(status) = child.try_wait().expect("inspect backend process") {
+            panic!("backend exited before readiness: {status}");
+        }
+        if client
+            .get(url)
+            .send()
+            .await
+            .is_ok_and(|response| response.status().is_success())
+        {
+            return;
+        }
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    panic!("backend did not become ready at {url}");
+}
+
+fn series(metric: &str, samples: &[(i64, f64)]) -> TimeSeries {
+    TimeSeries {
+        labels: vec![Label {
+            name: "__name__".into(),
+            value: metric.into(),
+        }],
+        samples: samples
+            .iter()
+            .map(|(timestamp, value)| Sample {
+                value: *value,
+                timestamp: *timestamp,
+            })
+            .collect(),
+        exemplars: Vec::new(),
+        histograms: Vec::new(),
+    }
+}
+
+async fn remote_write(client: &reqwest::Client, base: &str, request: &WriteRequest) -> u16 {
+    let body = snap::raw::Encoder::new()
+        .compress_vec(&request.encode_to_vec())
+        .expect("snappy encode");
+    client
+        .post(format!("{base}/api/v1/write"))
+        .header("content-encoding", "snappy")
+        .header("content-type", "application/x-protobuf")
+        .header("x-prometheus-remote-write-version", "0.1.0")
+        .body(body)
+        .send()
+        .await
+        .expect("send Remote Write")
+        .status()
+        .as_u16()
+}
+
+fn first_value(response: &Value, field: &str) -> Option<f64> {
+    let samples = response["data"]["result"]
+        .as_array()?
+        .first()?
+        .get(field)?
+        .as_array()?;
+    let value = if field == "value" {
+        samples.get(1)?
+    } else {
+        samples.last()?.as_array()?.get(1)?
+    };
+    value.as_str()?.parse().ok()
+}
+
+fn is_warm(response: &Value) -> bool {
+    response["infos"].as_array().is_some_and(|infos| {
+        infos.iter().any(|info| {
+            info.as_str()
+                .is_some_and(|line| line == "data_source: asap_query")
+        })
+    })
+}
+
+async fn wait_for_warm_instant(
+    client: &reqwest::Client,
+    base: &str,
+    query: &str,
+    evaluation_seconds: f64,
+    log_path: &std::path::Path,
+) -> Value {
+    let mut last = Value::Null;
+    for _ in 0..80 {
+        last = client
+            .get(format!("{base}/api/v1/query"))
+            .query(&[
+                ("query", query.to_string()),
+                ("time", evaluation_seconds.to_string()),
+            ])
+            .send()
+            .await
+            .expect("instant query")
+            .json()
+            .await
+            .expect("instant JSON");
+        if is_warm(&last) && first_value(&last, "value").is_some() {
+            return last;
+        }
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    let log = std::fs::read_to_string(log_path)
+        .unwrap_or_else(|error| format!("log unavailable: {error}"));
+    panic!("query never became warm: {query}: {last}\nbackend log:\n{log}");
+}
+
+#[tokio::test]
+async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() {
+    let fallback_calls = Arc::new(Mutex::new(Vec::<(
+        String,
+        HashMap<String, String>,
+        HeaderMap,
+    )>::new()));
+    let instant_calls = Arc::clone(&fallback_calls);
+    let range_calls = Arc::clone(&fallback_calls);
+    let fallback_app = Router::new()
+        .route("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/-/healthy", get(|| async { "Prometheus is Healthy." }))
+        .route(
+            "/api/v1/query",
+            get(
+                move |Query(params): Query<HashMap<String, String>>, headers: HeaderMap| {
+                    let calls = Arc::clone(&instant_calls);
+                    async move {
+                        calls
+                            .lock()
+                            .await
+                            .push(("instant".into(), params.clone(), headers));
+                        let timestamp = params
+                            .get("time")
+                            .and_then(|value| value.parse::<f64>().ok())
+                            .unwrap_or_default();
+                        Json(serde_json::json!({
+                            "status": "success",
+                            "data": {"resultType": "vector", "result": [{
+                                "metric": {"fallback": "true"},
+                                "value": [timestamp, "42"]
+                            }]}
+                        }))
+                    }
+                },
+            ),
+        )
+        .route(
+            "/api/v1/query_range",
+            get(
+                move |Query(params): Query<HashMap<String, String>>, headers: HeaderMap| {
+                    let calls = Arc::clone(&range_calls);
+                    async move {
+                        calls
+                            .lock()
+                            .await
+                            .push(("range".into(), params.clone(), headers));
+                        let start = params
+                            .get("start")
+                            .and_then(|value| value.parse::<f64>().ok())
+                            .unwrap_or_default();
+                        let end = params
+                            .get("end")
+                            .and_then(|value| value.parse::<f64>().ok())
+                            .unwrap_or_default();
+                        Json(serde_json::json!({
+                            "status": "success",
+                            "data": {"resultType": "matrix", "result": [{
+                                "metric": {"fallback": "true"},
+                                "values": [[start, "42"], [end, "43"]]
+                            }]}
+                        }))
+                    }
+                },
+            ),
+        );
+    let fallback_listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("bind fallback");
+    let fallback_address = fallback_listener.local_addr().expect("fallback address");
+    tokio::spawn(async move {
+        axum::serve(fallback_listener, fallback_app)
+            .await
+            .expect("serve fallback")
+    });
+
+    let backend_port = unused_port();
+    let output_dir = tempfile::tempdir().expect("backend output directory");
+    let snapshot = concat!(
+        env!("CARGO_MANIFEST_DIR"),
+        "/../docs/examples/asapquery-compatibility-demo-snapshot.json"
+    );
+    let child = Command::new(env!("CARGO_BIN_EXE_data_plane"))
+        .arg("--profile")
+        .arg("asapquery")
+        .arg("--planning-snapshot")
+        .arg(snapshot)
+        .arg("--prometheus-server")
+        .arg(format!("http://{fallback_address}"))
+        .arg("--forward-unsupported-queries")
+        .arg("--http-port")
+        .arg(backend_port.to_string())
+        .arg("--output-dir")
+        .arg(output_dir.path())
+        .arg("--precompute-allowed-lateness-ms")
+        .arg("0")
+        .arg("--precompute-flush-interval-ms")
+        .arg("25")
+        .stdout(Stdio::null())
+        .stderr(Stdio::inherit())
+        .spawn()
+        .expect("start production backend");
+    let mut child = ChildGuard(child);
+    let client = reqwest::Client::new();
+    let backend = format!("http://127.0.0.1:{backend_port}");
+    wait_until_ready(&client, &format!("{backend}/api/v1/health"), &mut child.0).await;
+
+    let now_ms = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .expect("system time")
+        .as_millis() as i64;
+    let base = now_ms - now_ms.rem_euclid(5_000) - 20_000;
+    let request = WriteRequest {
+        timeseries: vec![
+            series(
+                "asap_demo_counter_total",
+                &[
+                    (base + 500, 10.0),
+                    (base + 1_700, 20.0),
+                    (base + 2_900, 3.0),
+                    (base + 4_200, 13.0),
+                    (base + 5_400, 13.0),
+                    (base + 6_600, 21.0),
+                    (base + 8_100, 2.0),
+                    (base + 9_400, 12.0),
+                ],
+            ),
+            series(
+                "asap_demo_gauge",
+                &[
+                    (base + 500, 1.0),
+                    (base + 1_700, 2.0),
+                    (base + 2_900, 3.0),
+                    (base + 4_200, 4.0),
+                    (base + 5_400, 5.0),
+                    (base + 6_600, 6.0),
+                    (base + 8_100, 7.0),
+                    (base + 9_400, 8.0),
+                ],
+            ),
+            series(
+                "asap_demo_latency_ms",
+                &[
+                    (base + 500, 10.0),
+                    (base + 1_700, 20.0),
+                    (base + 2_900, 30.0),
+                    (base + 4_200, 40.0),
+                    (base + 5_400, 15.0),
+                    (base + 6_600, 25.0),
+                    (base + 8_100, 35.0),
+                    (base + 9_400, 45.0),
+                ],
+            ),
+        ],
+    };
+    assert_eq!(remote_write(&client, &backend, &request).await, 204);
+    let watermark_advance = WriteRequest {
+        timeseries: vec![
+            series("asap_demo_counter_total", &[(base + 10_500, 15.0)]),
+            series("asap_demo_gauge", &[(base + 10_500, 9.0)]),
+            series("asap_demo_latency_ms", &[(base + 10_500, 55.0)]),
+        ],
+    };
+    assert_eq!(
+        remote_write(&client, &backend, &watermark_advance).await,
+        204
+    );
+    // A normal Prometheus retry must be accepted without changing sketches.
+    assert_eq!(remote_write(&client, &backend, &request).await, 204);
+    let corrupt = client
+        .post(format!("{backend}/api/v1/write"))
+        .header("content-encoding", "snappy")
+        .header("content-type", "application/x-protobuf")
+        .body(vec![1, 2, 3])
+        .send()
+        .await
+        .expect("send corrupt request");
+    assert_eq!(corrupt.status().as_u16(), 400);
+
+    let first_eval = (base + 5_000) as f64 / 1_000.0;
+    let second_eval = (base + 10_000) as f64 / 1_000.0;
+    let backend_log = output_dir.path().join("query_engine.log");
+    let rate = wait_for_warm_instant(
+        &client,
+        &backend,
+        "rate(asap_demo_counter_total[5s])",
+        first_eval,
+        &backend_log,
+    )
+    .await;
+    let increase = wait_for_warm_instant(
+        &client,
+        &backend,
+        "increase(asap_demo_counter_total[5s])",
+        first_eval,
+        &backend_log,
+    )
+    .await;
+    let sum = wait_for_warm_instant(
+        &client,
+        &backend,
+        "sum_over_time(asap_demo_gauge[5s])",
+        first_eval,
+        &backend_log,
+    )
+    .await;
+    let quantile = wait_for_warm_instant(
+        &client,
+        &backend,
+        "quantile_over_time(0.5, asap_demo_latency_ms[5s])",
+        first_eval,
+        &backend_log,
+    )
+    .await;
+    let rate_value = first_value(&rate, "value").expect("rate value");
+    let increase_value = first_value(&increase, "value").expect("increase value");
+    assert!((rate_value * 5.0 - increase_value).abs() < 1e-9);
+    assert!((first_value(&sum, "value").expect("sum value") - 10.0).abs() < 1e-9);
+    let quantile_value = first_value(&quantile, "value").expect("quantile value");
+    assert!(
+        (19.0..=31.0).contains(&quantile_value),
+        "unexpected p50: {quantile_value}; response={quantile}"
+    );
+
+    for query in [
+        "rate(asap_demo_counter_total[5s])",
+        "increase(asap_demo_counter_total[5s])",
+        "sum_over_time(asap_demo_gauge[5s])",
+        "quantile_over_time(0.5, asap_demo_latency_ms[5s])",
+    ] {
+        let response: Value = client
+            .get(format!("{backend}/api/v1/query_range"))
+            .query(&[
+                ("query", query.to_string()),
+                ("start", first_eval.to_string()),
+                ("end", second_eval.to_string()),
+                ("step", "5".into()),
+            ])
+            .send()
+            .await
+            .expect("range query")
+            .json()
+            .await
+            .expect("range JSON");
+        assert_eq!(response["status"], "success", "{query}: {response}");
+        assert!(
+            is_warm(&response),
+            "{query} did not use warm tier: {response}"
+        );
+        let values = response["data"]["result"][0]["values"]
+            .as_array()
+            .unwrap_or_else(|| panic!("missing range values for {query}: {response}"));
+        assert_eq!(values.len(), 2, "wrong step count for {query}: {response}");
+        assert_eq!(values[0][0], first_eval);
+        assert_eq!(values[1][0], second_eval);
+    }
+
+    let fallback_instant: Value = client
+        .get(format!("{backend}/api/v1/query"))
+        .header("authorization", "Bearer demo")
+        .header("x-scope-orgid", "tenant-demo")
+        .query(&[
+            ("query", "max(asap_unplanned)"),
+            ("time", &first_eval.to_string()),
+            ("timeout", "7s"),
+        ])
+        .send()
+        .await
+        .expect("fallback instant")
+        .json()
+        .await
+        .expect("fallback instant JSON");
+    assert_eq!(
+        fallback_instant["data"]["result"][0]["metric"]["fallback"],
+        "true"
+    );
+    let fallback_range: Value = client
+        .get(format!("{backend}/api/v1/query_range"))
+        .header("authorization", "Bearer demo")
+        .header("x-scope-orgid", "tenant-demo")
+        .query(&[
+            ("query", "max(asap_unplanned)"),
+            ("start", &first_eval.to_string()),
+            ("end", &second_eval.to_string()),
+            ("step", "5"),
+            ("timeout", "9s"),
+        ])
+        .send()
+        .await
+        .expect("fallback range")
+        .json()
+        .await
+        .expect("fallback range JSON");
+    assert_eq!(
+        fallback_range["data"]["result"][0]["metric"]["fallback"],
+        "true"
+    );
+
+    let calls = fallback_calls.lock().await;
+    assert_eq!(
+        calls.len(),
+        2,
+        "planned queries unexpectedly fell back: {calls:?}"
+    );
+    assert_eq!(calls[0].0, "instant");
+    assert_eq!(calls[0].1["query"], "max(asap_unplanned)");
+    assert_eq!(calls[0].1["time"], first_eval.to_string());
+    assert_eq!(calls[0].1["timeout"], "7s");
+    assert_eq!(calls[0].2["authorization"], "Bearer demo");
+    assert_eq!(calls[0].2["x-scope-orgid"], "tenant-demo");
+    assert_eq!(calls[1].0, "range");
+    assert_eq!(calls[1].1["start"], first_eval.to_string());
+    assert_eq!(calls[1].1["end"], second_eval.to_string());
+    assert_eq!(calls[1].1["step"], "5");
+    assert_eq!(calls[1].1["timeout"], "9s");
+    drop(calls);
+
+    let status: Value = client
+        .get(format!("{backend}/api/v1/physical-plan/status"))
+        .send()
+        .await
+        .expect("physical status")
+        .json()
+        .await
+        .expect("physical status JSON");
+    assert_eq!(status["status"], "success");
+    let materializations = status["materializations"]
+        .as_array()
+        .expect("materialization statuses");
+    assert_eq!(materializations.len(), 3);
+    assert!(materializations
+        .iter()
+        .all(|entry| entry["phase"] == "serving"));
+
+    let metrics = client
+        .get(format!("{backend}/metrics"))
+        .send()
+        .await
+        .expect("metrics request")
+        .text()
+        .await
+        .expect("metrics body");
+    assert!(metrics.contains("asap_remote_write_requests_total 4"));
+    assert!(metrics.contains("asap_remote_write_samples_total 27"));
+    assert!(metrics.contains("asap_remote_write_duplicates_total 24"));
+    assert!(metrics.contains("asap_remote_write_rejected_requests_total 1"));
+}
diff --git a/demos/asapquery/prometheus.yml b/demos/asapquery/prometheus.yml
new file mode 100644
index 000000000..7977c8725
--- /dev/null
+++ b/demos/asapquery/prometheus.yml
@@ -0,0 +1,16 @@
+global:
+  scrape_interval: 1s
+  evaluation_interval: 1s
+
+scrape_configs:
+  - job_name: asapquery-demo-source
+    honor_labels: true
+    static_configs:
+      - targets: ["127.0.0.1:19092"]
+
+remote_write:
+  - url: http://127.0.0.1:19091/api/v1/write
+    queue_config:
+      min_shards: 1
+      max_shards: 1
+      batch_send_deadline: 1s
diff --git a/demos/asapquery/run.sh b/demos/asapquery/run.sh
new file mode 100755
index 000000000..addad6b5c
--- /dev/null
+++ b/demos/asapquery/run.sh
@@ -0,0 +1,171 @@
+#!/usr/bin/env bash
+
+set -Eeuo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_DIR="$(cd "${SCRIPT_DIR}/../.." && pwd)"
+PROM_IMAGE="${ASAPQUERY_PROMETHEUS_IMAGE:-prom/prometheus:v2.55.1}"
+PUSH_IMAGE="${ASAPQUERY_PUSHGATEWAY_IMAGE:-prom/pushgateway:v1.9.0}"
+RUN_ID="asapquery-demo-$$"
+PROM_CONTAINER="${RUN_ID}-prometheus"
+PUSH_CONTAINER="${RUN_ID}-pushgateway"
+EVIDENCE_DIR="${ASAPQUERY_DEMO_EVIDENCE_DIR:-${REPO_DIR}/target/asapquery-demo-evidence}"
+BACKEND_PID=""
+
+cleanup() {
+    if [[ -n "${BACKEND_PID}" ]]; then
+        kill "${BACKEND_PID}" 2>/dev/null || true
+        wait "${BACKEND_PID}" 2>/dev/null || true
+    fi
+    docker rm -f "${PROM_CONTAINER}" "${PUSH_CONTAINER}" >/dev/null 2>&1 || true
+}
+trap cleanup EXIT
+
+for command in cargo curl docker python3; do
+    command -v "${command}" >/dev/null 2>&1 || {
+        echo "missing required command: ${command}" >&2
+        exit 1
+    }
+done
+docker info >/dev/null
+mkdir -p "${EVIDENCE_DIR}"
+
+echo "Building the production backend..."
+cargo build --locked -p data_plane --bin data_plane
+
+docker run -d --rm --name "${PUSH_CONTAINER}" --network host \
+    "${PUSH_IMAGE}" --web.listen-address=:19092 >/dev/null
+docker run -d --rm --name "${PROM_CONTAINER}" --network host \
+    -v "${SCRIPT_DIR}/prometheus.yml:/etc/prometheus/prometheus.yml:ro" \
+    "${PROM_IMAGE}" \
+    --config.file=/etc/prometheus/prometheus.yml \
+    --storage.tsdb.path=/prometheus \
+    --web.listen-address=:19090 >/dev/null
+
+for _ in $(seq 1 120); do
+    if curl -fsS http://127.0.0.1:19090/-/healthy >/dev/null; then
+        break
+    fi
+    sleep 0.25
+done
+curl -fsS http://127.0.0.1:19090/-/healthy >/dev/null
+
+"${REPO_DIR}/target/debug/data_plane" \
+    --profile asapquery \
+    --planning-snapshot "${REPO_DIR}/docs/examples/asapquery-compatibility-demo-snapshot.json" \
+    --prometheus-server http://127.0.0.1:19090 \
+    --forward-unsupported-queries \
+    --http-port 19091 \
+    --precompute-allowed-lateness-ms 0 \
+    --precompute-flush-interval-ms 100 \
+    --output-dir "${EVIDENCE_DIR}/backend" &
+BACKEND_PID=$!
+
+for _ in $(seq 1 120); do
+    if curl -fsS http://127.0.0.1:19091/api/v1/health >/dev/null; then
+        break
+    fi
+    sleep 0.25
+done
+curl -fsS http://127.0.0.1:19091/api/v1/health >/dev/null
+
+echo "Publishing raw metrics through Prometheus for complete 5s windows..."
+counter=10
+for sample in $(seq 1 18); do
+    if [[ "${sample}" == "7" ]]; then
+        counter=2
+    else
+        counter=$((counter + sample))
+    fi
+    latency=$((10 + (sample % 6) * 10))
+    curl -fsS --data-binary @- \
+        http://127.0.0.1:19092/metrics/job/asapquery-demo <<EOF >/dev/null
+# TYPE asap_demo_counter_total counter
+asap_demo_counter_total ${counter}
+# TYPE asap_demo_gauge gauge
+asap_demo_gauge ${sample}
+# TYPE asap_demo_latency_ms gauge
+asap_demo_latency_ms ${latency}
+EOF
+    sleep 1
+done
+sleep 2
+
+query_backend() {
+    local query=$1
+    local output=$2
+    shift 2
+    curl -fsS --get http://127.0.0.1:19091/api/v1/query \
+        --data-urlencode "query=${query}" "$@" >"${output}"
+}
+
+assert_warm_response() {
+    python3 - "$1" <<'PY'
+import json
+import sys
+
+with open(sys.argv[1], encoding="utf-8") as source:
+    response = json.load(source)
+assert response.get("status") == "success", response
+assert response.get("data", {}).get("result"), response
+assert "data_source: asap_query" in response.get("infos", []), response
+PY
+}
+
+end=$(( $(date +%s) / 5 * 5 - 5 ))
+start=$((end - 5))
+for query in \
+    'rate(asap_demo_counter_total[5s])' \
+    'increase(asap_demo_counter_total[5s])' \
+    'sum_over_time(asap_demo_gauge[5s])' \
+    'quantile_over_time(0.5, asap_demo_latency_ms[5s])'; do
+    slug="$(printf '%s' "${query}" | tr -cs '[:alnum:]' '_')"
+    query_backend "${query}" "${EVIDENCE_DIR}/${slug}-instant.json" \
+        --data-urlencode "time=${end}"
+    assert_warm_response "${EVIDENCE_DIR}/${slug}-instant.json"
+
+    curl -fsS --get http://127.0.0.1:19091/api/v1/query_range \
+        --data-urlencode "query=${query}" \
+        --data-urlencode "start=${start}" \
+        --data-urlencode "end=${end}" \
+        --data-urlencode 'step=5' >"${EVIDENCE_DIR}/${slug}-range.json"
+    assert_warm_response "${EVIDENCE_DIR}/${slug}-range.json"
+done
+
+fallback_query='max(asap_demo_gauge)'
+query_backend "${fallback_query}" "${EVIDENCE_DIR}/fallback-backend.json" \
+    --data-urlencode "time=${end}" --data-urlencode 'timeout=7s'
+curl -fsS --get http://127.0.0.1:19090/api/v1/query \
+    --data-urlencode "query=${fallback_query}" \
+    --data-urlencode "time=${end}" --data-urlencode 'timeout=7s' \
+    >"${EVIDENCE_DIR}/fallback-direct.json"
+python3 - "${EVIDENCE_DIR}/fallback-backend.json" \
+    "${EVIDENCE_DIR}/fallback-direct.json" <<'PY'
+import json
+import sys
+
+with open(sys.argv[1], encoding="utf-8") as source:
+    backend = json.load(source)
+with open(sys.argv[2], encoding="utf-8") as source:
+    direct = json.load(source)
+assert backend.get("data") == direct.get("data"), (backend, direct)
+PY
+
+curl -fsS http://127.0.0.1:19091/api/v1/physical-plan/status \
+    >"${EVIDENCE_DIR}/physical-plan-status.json"
+curl -fsS http://127.0.0.1:19091/metrics >"${EVIDENCE_DIR}/backend.metrics"
+python3 - "${EVIDENCE_DIR}/physical-plan-status.json" <<'PY'
+import json
+import sys
+
+with open(sys.argv[1], encoding="utf-8") as source:
+    status = json.load(source)
+assert status.get("status") == "success", status
+materializations = status.get("materializations", [])
+assert len(materializations) == 3, materializations
+assert all(item.get("phase") == "serving" for item in materializations), materializations
+PY
+grep -q '^asap_remote_write_samples_total [1-9]' "${EVIDENCE_DIR}/backend.metrics"
+
+echo "PASS: Prometheus Remote Write -> PrecomputePlan -> QueryPlan DAG -> warm result"
+echo "Evidence: ${EVIDENCE_DIR}"
diff --git a/docs/design_docs/asapquery-compatibility-profile.md b/docs/design_docs/asapquery-compatibility-profile.md
index 136bd21cf..41dbb0948 100644
--- a/docs/design_docs/asapquery-compatibility-profile.md
+++ b/docs/design_docs/asapquery-compatibility-profile.md
@@ -1,6 +1,6 @@
 # ASAPQuery compatibility profile
 
-> Status: proposed MVP architecture and implementation contract
+> Status: implemented by the compatibility stack (pending merge)
 >
 > Reference: [ProjectASAP/ASAPQuery at `9fb051a`](https://github.com/ProjectASAP/ASAPQuery/tree/9fb051aa798361fca8e3012835412cb6fa338a0c)
 >
@@ -15,10 +15,8 @@ external collectors, accept materialized summaries over modified OTLP, use
 multiple storage tiers, and compile distributed physical plans. This profile
 does not remove those capabilities. It defines the target configuration required
 for ASAPQuery-compatible behavior and gives that configuration an independent
-end-to-end acceptance target. It is not yet a strict subset of the implemented
-runtime because Remote Write ingestion must first be restored as an optional
-backend component. After that work lands, selecting this profile is a strict
-configuration subset of the broader product.
+end-to-end acceptance target. Selecting `--profile asapquery` is now a strict,
+startup-validated subset of the broader runtime.
 
 The user-visible goal is the same drop-in shape as ASAPQuery:
 
@@ -98,18 +96,20 @@ configured QueryWorkload + DataWorkload
                     │
                     ▼
       backend-only physical compiler
-             │              │
-             ▼              ▼
-       PrecomputePlan   BackendPlan
-             │              │
-             ▼              ▼
-        precompute     store + query router
+        │          │          │
+        ▼          ▼          ▼
+ PrecomputePlan BackendPlan QueryPlan DAG
+        │          │          │
+        ▼          ▼          ▼
+   precompute    catalog   SID-bound executor
 ```
 
-The plan has no collector projection. One compile produces a backend-local
-`PrecomputePlan` and `BackendPlan` from the same selected Post-ASAP candidate.
-`PrecomputePlan` is an internal typed projection/section of `BackendPlan`, not a
-separately published protocol. They share plan and materialization identities.
+The plan has no collector projection. One compile produces an atomic
+`PhysicalPlan` containing sibling backend-local `PrecomputePlan`, `BackendPlan`,
+and `QueryPlan` sections from the same selected Post-ASAP candidate.
+`PrecomputePlan` directly is the runtime precompute contract; there is no second
+streaming-config semantic model or lossy conversion step. All sections share
+plan and materialization identities.
 One immutable version installs the precompute configuration, store catalog, and
 inactive query routes atomically. Materialization readiness is runtime state:
 each route becomes eligible for summary serving only after its required windows
@@ -139,8 +139,8 @@ For the first MVP, operators provide immutable `QueryWorkload` and
    `DataWorkload`;
 3. enumerates only backend-local implementations for Planner candidates;
 4. returns implementation-cost evidence needed for selection;
-5. compiles the selected candidate into matching PrecomputePlan and BackendPlan
-   views; and
+5. compiles the selected candidate into matching PrecomputePlan, BackendPlan,
+   and QueryPlan views; and
 6. stages and atomically activates those views.
 
 It does not enumerate SDK or Collector placements in this profile. A Planner
@@ -245,7 +245,8 @@ and `timeout`, plus configured tenant and authorization context; it does not
 require byte-for-byte reproduction of the incoming HTTP request. Routing has two
 successful outcomes:
 
-1. execute the active BackendPlan readout when compatible summary state has
+1. execute the active QueryPlan DAG, whose materialization bindings were
+   resolved from BackendPlan, when compatible summary state has
    complete and fresh coverage; or
 2. forward a semantically equivalent request to the configured Prometheus
    endpoint.
@@ -272,7 +273,7 @@ start backend
   -> verify Prometheus fallback health
   -> load configured QueryWorkload and DataWorkload snapshots
   -> run Planner candidate search and selection
-  -> compile PrecomputePlan + BackendPlan
+  -> compile one PhysicalPlan (PrecomputePlan + BackendPlan + QueryPlan)
   -> atomically install precompute + catalog + inactive routes under one version
   -> accept Remote Write and forward every query to Prometheus
   -> enter Materializing state
@@ -297,40 +298,34 @@ same atomic cutover and warmup rules.
 | --- | --- | --- |
 | Ingest source | Prometheus Remote Write raw samples | Collector materializations over modified OTLP and other explicit profiles |
 | Summary construction | Backend-local only | Collector or backend placement |
-| Physical outputs | PrecomputePlan + BackendPlan | SDKPlan/CollectorPlan/TransmissionPlan/BackendPlan views as applicable |
+| Physical outputs | PrecomputePlan + BackendPlan + QueryPlan | CollectorPlan/PrecomputePlan/TransmissionPlan/BackendPlan/QueryPlan views as applicable |
 | Query protocol | Prometheus HTTP / PromQL | Additional protocols may be supported |
 | Exact fallback | Upstream Prometheus | Prometheus or another compiled storage/query route |
 | Storage required for MVP | In-process warm summary state | Warm, durable, archive, and remote tiers |
 | Sampling and delta | Disabled | Optional physical mechanisms |
 
-This table describes the intended product boundary, not the current
-implementation state. The compatibility profile is a restricted target profile,
-but it is not a strict subset of the backend executable today because some of
-its required adapters were removed. Once those adapters are restored behind the
-explicit profile, every enabled component belongs to ASAPQuery-backend and the
-selected runtime configuration is a strict subset of the broader product.
+The compatibility profile is a restricted runtime configuration. Every enabled
+component belongs to ASAPQuery-backend; broader distributed features remain
+available only outside this profile.
 
 The profile is also not a literal subset of historical ASAPQuery internals. It
 preserves the relevant external behavior while adding the current canonical
 ASAPPlanner types, versioned physical compilation, readiness evidence, and
 stronger activation and retry contracts.
 
-## Current implementation gap
+## Implementation status
 
-Against ASAPQuery-backend
-[`d1498fd`](https://github.com/ProjectASAP/ASAPQuery-backend/tree/d1498fd191b5f782e743c2d0f27a382b5c69f432):
-
-| Area | Reusable today | Required change |
+| Area | Implemented contract | Executable evidence |
 | --- | --- | --- |
-| Prometheus query adapter and fallback client | Present | Bind them to the compatibility profile and its BackendPlan readiness checks. |
-| Streaming precompute workers and accumulators | Present, with substantial divergence and newer backend fixes | Complete the ASAPQuery bug-fix parity audit, migrate applicable fixes with regression tests, and explicitly reject fixes for intentionally retired features. The known missing active-ingest wall-clock fix must measure pane idleness from last touch rather than pane creation. Then admit raw Remote Write samples through a dedicated adapter and add Prometheus semantic conformance coverage. |
-| Hot-reload plan/store/query snapshots | Partial | Install PrecomputePlan and BackendPlan as one atomic version. |
-| Prometheus Remote Write decoder/listener | Removed from the current backend path | Restore the narrow v1 adapter from the reference behavior without restoring other legacy connectors. Preserve stale-marker semantics and retry-safe batch application. |
-| Workload input | Canonical Planner integration is present | Load deterministic `QueryWorkload` and `DataWorkload` snapshots at startup; online observation is optional after the MVP. |
-| Collector/OTLP path | Present in the broader product | Disable it in this profile; do not make it a test or startup dependency. |
-| Compatibility E2E | Missing | Add a Prometheus + backend + synthetic writer/query test and demo. |
+| Startup/profile | Collector-free startup, excluded-component validation, fallback health gate | `data_plane` profile tests and production-process E2E |
+| Physical planning | Canonical workload snapshot to one atomic PrecomputePlan/BackendPlan/QueryPlan bundle | `compatibility_demo_snapshot_compiles_the_complete_query_matrix` |
+| Remote Write | Strict v1 decoding, stale handling, limits, retry-safe deduplication and backpressure | receiver unit tests plus process E2E replay/corrupt-batch assertions |
+| Precompute/store | Raw samples use the planned family; first catch-up batches close all complete windows; sketch and exact payloads share canonical SID semantics | worker/store tests and four-family process matrix |
+| Query execution | QueryPlan-only serving-time lookup, node-level materialization binding, generic DAG traversal, exact fallback | instant/range process matrix and fallback request capture |
+| Atomic activation | Versioned stage/activate snapshot and materialization readiness state | physical-plan endpoint tests and `/physical-plan/status` assertions |
+| Real deployment | Prometheus remote_write with no Collector | `./scripts/e2e.sh asapquery-demo` |
 
-## Phased implementation
+## Implemented phases
 
 ### Phase A: profile and startup contract
 
@@ -357,12 +352,13 @@ overloaded requests cannot leave untracked mutations while returning success.
 ### Phase C: backend-only planning
 
 Load the configured query and data workload snapshots, call the pinned
-ASAPPlanner, enumerate backend-local implementations, and compile one
-PrecomputePlan plus BackendPlan. Do not create or wait for CollectorPlan.
+ASAPPlanner, enumerate backend-local implementations, and compile one atomic
+PhysicalPlan with PrecomputePlan, BackendPlan, and QueryPlan. Do not create or
+wait for CollectorPlan.
 
 Acceptance: captured Planner input, selected Post-ASAP candidate,
-PrecomputePlan, and BackendPlan are deterministic golden artifacts with matching
-plan/materialization/window/family/parameter identities.
+PrecomputePlan, BackendPlan, and QueryPlan are deterministic golden artifacts
+with matching plan/materialization/window/family/parameter identities.
 
 ### Phase D: atomic activation and warmup
 
@@ -389,8 +385,9 @@ are equivalent to direct Prometheus calls.
 
 ### Phase F: compatibility demo
 
-Run Prometheus with `remote_write` configured to the backend, start the backend
-with fixed workload snapshots, send their corresponding repeating queries
+`./scripts/e2e.sh asapquery-demo` runs Prometheus with `remote_write` configured
+to the backend, starts the backend with fixed workload snapshots, sends their
+corresponding repeating queries
 through the backend, wait for planning and warmup, and capture route decisions
 and resource measurements.
 
@@ -437,8 +434,13 @@ dependencies of the Remote Write profile.
 
 ## MVP completion criterion
 
-The profile is complete when a clean checkout can run one documented command
-that starts Prometheus and ASAPQuery-backend without ASAPCollector, ingests only
+The executable completion command is:
+
+```bash
+./scripts/e2e.sh asapquery-demo
+```
+
+It starts Prometheus and ASAPQuery-backend without ASAPCollector, ingests only
 through Prometheus Remote Write, plans from the configured workloads, activates a
 backend-local summary, serves both the declared sum and sketch-backed quantile
 compatibility cases plus Prometheus `rate` and `increase` through both instant
diff --git a/docs/examples/asapquery-compatibility-demo-snapshot.json b/docs/examples/asapquery-compatibility-demo-snapshot.json
new file mode 100644
index 000000000..2870e5438
--- /dev/null
+++ b/docs/examples/asapquery-compatibility-demo-snapshot.json
@@ -0,0 +1,92 @@
+{
+  "snapshot_version": 1,
+  "query_workload": {
+    "language": "prom_q_l",
+    "query_batch": null,
+    "repeating_queries": [
+      {
+        "query": "rate(asap_demo_counter_total[5s])",
+        "demand": { "fixed_interval": 1000 },
+        "requirements": { "accuracy": "implicit_exact", "response_latency": "unspecified" },
+        "predictability": { "predictable": { "known_at": null } },
+        "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null }
+      },
+      {
+        "query": "increase(asap_demo_counter_total[5s])",
+        "demand": { "fixed_interval": 1000 },
+        "requirements": { "accuracy": "implicit_exact", "response_latency": "unspecified" },
+        "predictability": { "predictable": { "known_at": null } },
+        "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null }
+      },
+      {
+        "query": "sum_over_time(asap_demo_gauge[5s])",
+        "demand": { "fixed_interval": 1000 },
+        "requirements": { "accuracy": "implicit_exact", "response_latency": "unspecified" },
+        "predictability": { "predictable": { "known_at": null } },
+        "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null }
+      },
+      {
+        "query": "quantile_over_time(0.5, asap_demo_latency_ms[5s])",
+        "demand": { "fixed_interval": 1000 },
+        "requirements": {
+          "accuracy": { "explicit": { "EpsilonDelta": { "epsilon": 0.01, "delta": 0.01 } } },
+          "response_latency": "unspecified"
+        },
+        "predictability": { "predictable": { "known_at": null } },
+        "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null }
+      }
+    ],
+    "data_workload": {
+      "arrival": "continuously_ingesting",
+      "ingestion_volume": { "value": null, "source": "unknown", "observed_at_ms": null, "valid_for_ms": null },
+      "ingestion_rate": { "value": 100.0, "source": "declared", "observed_at_ms": null, "valid_for_ms": null },
+      "input_cardinality": { "value": null, "source": "unknown", "observed_at_ms": null, "valid_for_ms": null },
+      "distribution": { "value": null, "source": "unknown", "observed_at_ms": null, "valid_for_ms": null }
+    }
+  },
+  "data_workload": {
+    "arrival": "continuously_ingesting",
+    "ingestion_volume": { "value": null, "source": "unknown", "observed_at_ms": null, "valid_for_ms": null },
+    "ingestion_rate": { "value": 100.0, "source": "declared", "observed_at_ms": null, "valid_for_ms": null },
+    "input_cardinality": { "value": null, "source": "unknown", "observed_at_ms": null, "valid_for_ms": null },
+    "distribution": { "value": null, "source": "unknown", "observed_at_ms": null, "valid_for_ms": null }
+  },
+  "implementation": {
+    "lifecycle_costs": {
+      "build": 10.0,
+      "maintenance_per_update": 0.001,
+      "read": 0.1,
+      "retention_per_second": 0.001,
+      "retirement": 1.0
+    },
+    "evidence_observed_at_unix_ms": 9500,
+    "evidence_valid_for_ms": 60000,
+    "horizon_seconds": 300.0,
+    "window_implementation_id": "backend-tumbling-v1",
+    "state_layout": "anchored-pane-v1",
+    "implementation_cost": {
+      "model_version": "compat-cost-v1",
+      "workload_fingerprint": "asapquery-compatibility-demo",
+      "observed_at_unix_ms": 9500,
+      "valid_for_ms": 60000,
+      "horizon_seconds": 300.0,
+      "cpu_cost": 1.0,
+      "peak_memory_bytes": 4096,
+      "network_bytes": 0,
+      "storage_bytes": 2048,
+      "source_scan_bytes": 0,
+      "weighted_cost": 1.0
+    }
+  },
+  "environment": {
+    "target": "backend_local_remote_write",
+    "collector_ids": [],
+    "capability_snapshot_id": "asapquery-compatibility-demo-v1",
+    "observed_at_unix_ms": 10000,
+    "max_evidence_age_ms": 60000,
+    "plan_version": 1,
+    "activation_unix_ms": 10000,
+    "expiry_unix_ms": null,
+    "backend_compat": "asap-query-backend.v1"
+  }
+}
diff --git a/docs/user_guide/asapquery-profile.md b/docs/user_guide/asapquery-profile.md
index 6f0b80646..e61ca2e6c 100644
--- a/docs/user_guide/asapquery-profile.md
+++ b/docs/user_guide/asapquery-profile.md
@@ -4,6 +4,24 @@ The `asapquery` runtime profile runs without ASAPCollector. It accepts
 Prometheus Remote Write v1 and sends unsupported or not-yet-ready PromQL to the
 same Prometheus server as an exact fallback.
 
+For a complete, self-checking run with a real Prometheus server:
+
+```bash
+./scripts/e2e.sh asapquery-demo
+```
+
+The command builds the production backend, starts pinned Prometheus and
+Pushgateway containers, drives raw samples (including a counter reset), checks
+all declared instant and range warm queries, compares an unplanned query with a
+direct Prometheus fallback response, and writes status/results/metrics under
+`target/asapquery-demo-evidence`. It requires Docker, `curl`, and Python 3.
+
+For the hermetic production-process conformance suite (no Docker):
+
+```bash
+./scripts/e2e.sh asapquery
+```
+
 Start Prometheus first, configure it to write to the backend, and keep its
 normal local storage enabled:
 
diff --git a/scripts/e2e.sh b/scripts/e2e.sh
index 4417953e7..ae240e875 100755
--- a/scripts/e2e.sh
+++ b/scripts/e2e.sh
@@ -26,6 +26,8 @@ Targets:
   contracts       Shared Rust type/protobuf wire contracts
   control-plane   Planner HTTP, OpAMP, publication, and runtime feedback
   data-plane      Query, routing, storage, ingest adapter, and lifecycle tests
+  asapquery       Collector-free Remote Write -> QueryPlan process conformance
+  asapquery-demo  Run the real Prometheus compatibility demo (requires Docker)
   differential    Production DDSketch PromQL vs deterministic raw-value oracle
   sketch-oracles  Every sketch via production binary + independent raw oracle
   monitor         Real monitor gRPC transport tests
@@ -107,6 +109,22 @@ data_plane() {
     differential
 }
 
+asapquery() {
+    CURRENT_STAGE="asapquery/physical-compile"
+    say "asapquery: canonical workload -> backend-local atomic PhysicalPlan"
+    rust_test control_plane compatibility_demo_snapshot_compiles_the_complete_query_matrix
+
+    CURRENT_STAGE="asapquery/production-process"
+    say "asapquery: Remote Write -> precompute/store -> QueryPlan DAG -> fallback"
+    rust_test data_plane --test asapquery_compatibility_process_e2e
+}
+
+asapquery_demo() {
+    CURRENT_STAGE="asapquery/real-prometheus-demo"
+    say "asapquery: real Prometheus Remote Write compatibility demo"
+    "${REPO_DIR}/demos/asapquery/run.sh"
+}
+
 differential() {
     CURRENT_STAGE="data-plane/promql-differential"
     say "data-plane: production DDSketch PromQL -> raw oracle + range endpoint consistency"
@@ -204,6 +222,8 @@ main() {
         contracts) need cargo; contracts ;;
         control-plane) need cargo; control_plane ;;
         data-plane) need cargo; data_plane ;;
+        asapquery) need cargo; asapquery ;;
+        asapquery-demo) need cargo; asapquery_demo ;;
         differential) need cargo; differential ;;
         sketch-oracles) need cargo; sketch_oracles ;;
         monitor) need cargo; monitor ;;