test(asapquery): prove collector-free compatibility end to end

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])");

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,

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(),
             }
         }

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

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(

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;

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

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!(