diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
index 26c9febe5..8890f03eb 100644
--- a/.github/workflows/rust.yml
+++ b/.github/workflows/rust.yml
@@ -108,6 +108,30 @@ jobs:
       env:
         RUSTC_WRAPPER: sccache
 
+    # The viewer's Python tests run its JavaScript in V8 through py_mini_racer;
+    # the ones that plan a query run target/debug/stage_pipeline, which the
+    # workspace test build above already linked for devtools' tests.
+    # A pinned Python and Node: the runner's system Python refuses a global
+    # pip install (PEP 668).
+    - name: Install Python
+      uses: actions/setup-python@v5
+      with:
+        python-version: '3.12'
+
+    - name: Install Node
+      uses: actions/setup-node@v4
+      with:
+        node-version: '22'
+
+    - name: Run the DAG viewer's Python tests
+      working-directory: tools/dag-viewer
+      run: |
+        python3 -m pip install py-mini-racer==0.6.0
+        python3 -m unittest -v test_viewer test_server
+
+    - name: Run the DAG viewer's Node tests
+      run: node --test tools/dag-viewer/stages.test.mjs
+
     # A PromQL-only runtime builds asap-planner without DataFusion or MetricsQL.
     - name: Check and test asap-planner with the PromQL frontend alone
       run: |
diff --git a/crates/devtools/src/bin/stage_pipeline.rs b/crates/devtools/src/bin/stage_pipeline.rs
index 4d9184229..595792ecd 100644
--- a/crates/devtools/src/bin/stage_pipeline.rs
+++ b/crates/devtools/src/bin/stage_pipeline.rs
@@ -1188,6 +1188,12 @@ mod tests {
     use super::*;
     use asap_devtools::planning_stages::run_planning_stages;
 
+    /// The exhaustive search's default `--max-combinations` is 10^4.
+    #[test]
+    fn exhaustive_default_candidate_limit_is_ten_thousand() {
+        assert_eq!(MAX_COMBINATIONS, 10_000);
+    }
+
     /// With no valid candidate, the document is still written: `selected`
     /// is null and every rejection is listed.
     #[test]
diff --git a/crates/devtools/tests/stage_pipeline.rs b/crates/devtools/tests/stage_pipeline.rs
index d77ace433..db0ac3b6a 100644
--- a/crates/devtools/tests/stage_pipeline.rs
+++ b/crates/devtools/tests/stage_pipeline.rs
@@ -415,23 +415,6 @@ fn nine_quantile_queries() -> Vec<String> {
     args
 }
 
-/// The exhaustive search's default limit is 10^4: the nine quantile
-/// queries' 19683 combinations are rejected once the count passes it, and
-/// the message names the exhaustive search.
-#[test]
-fn exhaustive_default_candidate_limit_is_ten_thousand() {
-    let mut args = nine_quantile_queries();
-    args.push("--exhaustive".into());
-    let args: Vec<&str> = args.iter().map(String::as_str).collect();
-    let (status, stderr) = failure(&args);
-    assert_eq!(status, Some(3));
-    assert!(
-        stderr.contains("workload has more than 10000 Stage 1 candidates"),
-        "{stderr}"
-    );
-    assert!(stderr.contains("--exhaustive"), "{stderr}");
-}
-
 /// Without `--exhaustive`, the same nine queries plan with the default
 /// search, `plan_installable`'s: each query, reading its own metric, is an
 /// independent component searched on its own, so no limit is reached. The
diff --git a/crates/executor/tests/physical_dag.rs b/crates/executor/tests/physical_dag.rs
index 433233cb5..970e6f15e 100644
--- a/crates/executor/tests/physical_dag.rs
+++ b/crates/executor/tests/physical_dag.rs
@@ -497,7 +497,7 @@ fn exact_state_and_family_validation() {
 
 // Planner binding rejects unknown computation instead of accepting a fallback.
 #[test]
-fn bind_post_asap_before_execution() {
+fn bind_physical_asap_dag_before_execution() {
     use asap_executor::dag::planner::bind;
     use planner_types::ir::properties::ExecutionDataState;
     use planner_types::ir::scalar::{ArithmeticOpKind, ScalarValue};
@@ -567,7 +567,7 @@ fn bind_post_asap_before_execution() {
     };
     let native = bind(&dag, sources(), &[1]).unwrap();
     assert_eq!(floats(&run(&native, 1, query()), 0), vec![3.]);
-    // A literal Fallback needs no deployment input.
+    // A literal `Values` source needs no deployment input.
     let literal = bind(&dag, BTreeMap::new(), &[1]).unwrap();
     assert_eq!(floats(&run(&literal, 1, query()), 0), vec![3.]);
     dag.nodes[1].payload = PhysicalASAPOperatorPayload::ASAP(ASAPOp::Extension {
diff --git a/crates/executor/tests/summary_projection.rs b/crates/executor/tests/summary_projection.rs
index 4d8f2c2bc..d54987a12 100644
--- a/crates/executor/tests/summary_projection.rs
+++ b/crates/executor/tests/summary_projection.rs
@@ -19,10 +19,10 @@ use planner_types::ir::ASAPOp;
 use planner_types::ir::NonASAPOp;
 use std::{collections::BTreeMap, sync::Arc};
 
-// A Post-ASAP projection may reorder/rename summary columns; recovery must retain
+// A projection over a summary may reorder/rename its columns; recovery must retain
 // the family and pass through the same immutable state, without decoding the payload.
 #[test]
-fn post_asap_summary_projection_survives_recovery() {
+fn summary_projection_survives_recovery() {
     let family = FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum);
     let schema = Arc::new(planner_types::ir::schema::Schema {
         unique_keys: vec![],
diff --git a/crates/executor/tests/weighted_topk_binding.rs b/crates/executor/tests/weighted_topk_binding.rs
index 1b41024ef..342cfdb4d 100644
--- a/crates/executor/tests/weighted_topk_binding.rs
+++ b/crates/executor/tests/weighted_topk_binding.rs
@@ -14,8 +14,10 @@ use planner_types::ir::schema::{DataType, *};
 use planner_types::ir::ASAPOp;
 use std::{collections::BTreeMap, rc::Rc, sync::Arc};
 #[test]
-#[ignore = "the whole-expression heap's SummaryEstimate{TopK} output is not the keyed-evaluation \
-            shape the physical planner builds (#579)"]
+#[ignore = "the whole-expression heap keeps the outer top-k's keys indexed against the absorbed \
+            aggregate, so the physical planner rejects its keyed evaluation shape (#639); the \
+            unconfirmed Count-Min heap also has no non-negative weight proof until plan selection \
+            confirms one"]
 fn planner_weighted_topk_binds_at_either_deployment_phase() {
     assert_weighted_binding(SketchAlgorithm::CmsWithHeap);
     assert_weighted_binding(SketchAlgorithm::CountSketchWithHeap);
diff --git a/crates/frontend-promql/src/error.rs b/crates/frontend-promql/src/error.rs
index ebbcf49c2..0d7ae0d2c 100644
--- a/crates/frontend-promql/src/error.rs
+++ b/crates/frontend-promql/src/error.rs
@@ -76,6 +76,5 @@ mod tests {
         // must not hardcode "PromQL".
         let msg = PromqlError::UnsupportedFeature("subquery".into()).to_string();
         assert_eq!(msg, "unsupported feature: subquery");
-        assert!(!msg.contains("PromQL"), "got: {msg}");
     }
 }
diff --git a/crates/frontend-promql/tests/count_planning.rs b/crates/frontend-promql/tests/count_planning.rs
index fafadc70d..428cdabae 100644
--- a/crates/frontend-promql/tests/count_planning.rs
+++ b/crates/frontend-promql/tests/count_planning.rs
@@ -8,7 +8,7 @@ use asap_types::ir::schema::{
 };
 use asap_types::ir::{ASAPOp, Operator, OperatorNode};
 use std::rc::Rc;
-use support::{logical_candidates, lower_promql, post_asap_dag, selected_dag};
+use support::{logical_candidates, lower_promql, physical_asap_dag, selected_dag};
 
 /// Whether `node` reads a Hydra state (one sketch shared by every group).
 fn reads_hydra(node: &Rc<OperatorNode>) -> bool {
@@ -24,7 +24,7 @@ fn reads_hydra(node: &Rc<OperatorNode>) -> bool {
 }
 
 #[test]
-#[ignore = "Pass 1 offers no Hydra state for a grouped count yet (#605)"]
+#[ignore = "Pass 1 offers no Hydra state for a grouped count yet (#703)"]
 fn grouped_count_offers_hydra_but_selection_keeps_per_group_state() {
     let target = AccuracyTarget::max_error(0.01).failure_probability(0.01);
     // Hydra hashes a non-null item per row: the series identity (PromQL
@@ -99,7 +99,7 @@ fn frequency_count_candidates_use_unit_weights() {
             ) {
                 continue;
             }
-            let dag = post_asap_dag(node);
+            let dag = physical_asap_dag(node);
             assert!(
                 dag.nodes.iter().any(|node| matches!(
                     &node.payload,
@@ -233,7 +233,7 @@ fn cms_count_updates_total_ten_for_zero_positive_and_negative_samples() {
     let dag = candidates
         .iter()
         .find_map(|node| {
-            let dag = post_asap_dag(node);
+            let dag = physical_asap_dag(node);
             dag.nodes
                 .iter()
                 .any(|node| {
diff --git a/crates/frontend-promql/tests/maintained_population_horizon.rs b/crates/frontend-promql/tests/maintained_population_horizon.rs
index baff0db16..cdceec60e 100644
--- a/crates/frontend-promql/tests/maintained_population_horizon.rs
+++ b/crates/frontend-promql/tests/maintained_population_horizon.rs
@@ -22,7 +22,7 @@ fn population_preserves_selector_horizon() {
         panic!()
     };
     assert_eq!(spec.lookback_ms, 1_000);
-    support::post_asap_dag(&candidate);
+    support::physical_asap_dag(&candidate);
     let NonASAPOp::Aggregate { child: source, .. } = root.expect_non_asap() else {
         panic!()
     };
diff --git a/crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs b/crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs
index eb1795f20..b0ea6464d 100644
--- a/crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs
+++ b/crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs
@@ -33,8 +33,8 @@ use std::rc::Rc;
 
 use asap_types::ir::operator::{AggIntent, BinaryOpKind, Reduction};
 use asap_types::ir::properties::AccuracyTarget;
-use asap_types::ir::scalar::{CompareOpKind, ScalarValue};
-use asap_types::ir::{BinaryOperator, NonASAPOp, OperatorNode, ScalarExpr};
+use asap_types::ir::scalar::CompareOpKind;
+use asap_types::ir::{BinaryOperator, NonASAPOp, OperatorNode};
 use support::lower_promql;
 
 const CORPUS: &str = include_str!("data/awesome_prometheus_alerts.txt");
@@ -294,19 +294,6 @@ fn predict_linear_function_body_lowers_with_horizon() {
     ));
 }
 
-#[test]
-fn vector_literal_lowers_to_a_labelless_vector() {
-    // `vector(1)` — used in dead-man's-switch ("always firing") alerts. Now
-    // lowers to a `PromqlVectorFromScalar` over the scalar `1` (issue #48).
-    let qe = ok("vector(1)");
-    let NonASAPOp::PromqlVectorFromScalar(inner) = qe.expect_non_asap() else {
-        panic!("expected PromqlVectorFromScalar, got {qe:?}");
-    };
-    assert!(matches!(inner, ScalarExpr::Literal(ScalarValue::Float64(v)) if *v == 1.0));
-    // The result is a vector: it carries a time index (unlike a bare scalar).
-    assert!(qe.schema.time_index.is_some());
-}
-
 #[test]
 fn without_grouping_lowers_to_the_exclusion_form() {
     // `(min without (cpu) (rate(node_cpu_seconds_total{mode="idle"}[1h]))) > 0.8`
diff --git a/crates/frontend-promql/tests/observability/metrics_observability.rs b/crates/frontend-promql/tests/observability/metrics_observability.rs
index e080dfe38..46d8cf7a4 100644
--- a/crates/frontend-promql/tests/observability/metrics_observability.rs
+++ b/crates/frontend-promql/tests/observability/metrics_observability.rs
@@ -10,42 +10,69 @@ mod support;
 use asap_types::ir::properties::AccuracyTarget;
 use support::{logical_plan, lower_promql};
 
-const CORPORA: &[(&str, &str)] = &[
+/// Today's coverage of one corpus, as a ratchet: a change that plans fewer
+/// queries with an ASAP operator, or fails to plan more of them, fails the
+/// test. Raise `min_asap_candidates` (or lower `max_planning_errors`) when
+/// coverage improves.
+struct CoverageRatchet {
+    /// Queries whose logical plan binds an ASAP operator below the root.
+    min_asap_candidates: usize,
+    /// Lowered queries whose logical planning returns an error.
+    max_planning_errors: usize,
+}
+
+const fn ratchet(min_asap_candidates: usize, max_planning_errors: usize) -> CoverageRatchet {
+    CoverageRatchet {
+        min_asap_candidates,
+        max_planning_errors,
+    }
+}
+
+const CORPORA: &[(&str, &str, CoverageRatchet)] = &[
     (
         "CMU query logs",
         include_str!("data/metrics_observability/cmu_chad_logs.txt"),
+        ratchet(162, 0),
     ),
     (
         "CMU dashboards",
         include_str!("data/metrics_observability/cmu_chad_dashboards.txt"),
+        ratchet(174, 0),
     ),
     (
         "CMU rules",
         include_str!("data/metrics_observability/cmu_chad_rules.txt"),
+        ratchet(202, 0),
     ),
     (
         "Claude query logs",
         include_str!("data/metrics_observability/from_claude_logs.txt"),
+        ratchet(71, 0),
     ),
     (
         "Grafana dashboards",
         include_str!("data/metrics_observability/grafana_dashboards.txt"),
+        ratchet(164, 0),
     ),
     (
         "Kubernetes mixin dashboards",
         include_str!("data/metrics_observability/kubernetes_mixin_dashboards.txt"),
+        ratchet(147, 0),
     ),
     (
         "Kubernetes mixin alerts",
         include_str!("data/metrics_observability/kubernetes_mixin_alerts.txt"),
+        ratchet(27, 2),
     ),
     (
         "Kubernetes mixin rules",
         include_str!("data/metrics_observability/kubernetes_mixin_rules.txt"),
+        ratchet(66, 0),
     ),
     (
         "Promset",
         include_str!("data/metrics_observability/promset.txt"),
+        ratchet(1505, 4),
     ),
 ];
 
@@ -57,8 +84,8 @@ fn queries(corpus: &str) -> impl Iterator<Item = &str> {
 }
 
 #[test]
-fn benchmark_corpora_are_total_and_report_coverage() {
-    for (name, corpus) in CORPORA {
+fn benchmark_corpora_are_total_and_keep_their_coverage() {
+    for (name, corpus, ratchet) in CORPORA {
         let mut total = 0;
         let mut lowered = 0;
         let mut parse_failures = 0;
@@ -103,6 +130,16 @@ fn benchmark_corpora_are_total_and_report_coverage() {
         // counted above so extraction drift is visible, but are not expected
         // to parse as PromQL.
         assert_eq!(total, lowered + rejected + parse_failures);
+        assert!(
+            post_asap_candidates >= ratchet.min_asap_candidates,
+            "{name}: {post_asap_candidates} queries bind an ASAP operator, fewer than {}",
+            ratchet.min_asap_candidates
+        );
+        assert!(
+            post_asap_errors <= ratchet.max_planning_errors,
+            "{name}: {post_asap_errors} queries fail logical planning, more than {}",
+            ratchet.max_planning_errors
+        );
     }
 }
 
diff --git a/crates/frontend-promql/tests/promql_conformance.rs b/crates/frontend-promql/tests/promql_conformance.rs
index 8faca896c..b8f4ab99e 100644
--- a/crates/frontend-promql/tests/promql_conformance.rs
+++ b/crates/frontend-promql/tests/promql_conformance.rs
@@ -208,17 +208,6 @@ fn selector_without_metric_name_is_rejected__GAP() {
     }
 }
 
-#[test]
-fn range_vector_selector_is_time_range() {
-    // SEMANTICS: `[5m]` turns an instant vector into a range vector,
-    // represented in the canonical DAG as a dedicated `TimeRange` node.
-    let qe = ok("node_cpu_seconds_total[5m]");
-    let NonASAPOp::TimeRange { range, .. } = qe.expect_non_asap() else {
-        panic!("expected TimeRange for a range-vector selector, got {qe:?}");
-    };
-    assert_eq!(*range, Duration::from_secs(300));
-}
-
 // ─────────────────────────────────────────────────────────────────────────────
 // B. Counters: rate / irate / increase     (cheat sheet "Rates of Increase";
 //    functions.test)
@@ -249,24 +238,6 @@ fn selector_time_ranges_carry_their_kind() {
     );
 }
 
-#[test]
-fn rate_range_lives_in_time_range_node() {
-    // SEMANTICS: per-second average rate; the temporal range lives on the
-    // enclosing `TimeRange` node, not inside the intent.
-    let qe = ok("rate(http_requests_total[5m])");
-    let NonASAPOp::Aggregate {
-        measures, child, ..
-    } = qe.expect_non_asap()
-    else {
-        panic!("expected Aggregate, got {qe:?}");
-    };
-    assert!(matches!(measures.as_slice(), [AggIntent::Rate]));
-    let NonASAPOp::TimeRange { range, .. } = child.expect_non_asap() else {
-        panic!("expected TimeRange child, got {child:?}");
-    };
-    assert_eq!(*range, Duration::from_secs(300));
-}
-
 #[test]
 fn irate_maps_to_its_own_intent() {
     assert!(has(&ok("irate(http_requests_total[1m])"), |i| matches!(
@@ -275,22 +246,6 @@ fn irate_maps_to_its_own_intent() {
     )));
 }
 
-#[test]
-fn increase_range_lives_in_time_range_node() {
-    let qe = ok("increase(http_requests_total[1h])");
-    let NonASAPOp::Aggregate {
-        measures, child, ..
-    } = qe.expect_non_asap()
-    else {
-        panic!("expected Aggregate, got {qe:?}");
-    };
-    assert!(matches!(measures.as_slice(), [AggIntent::Increase]));
-    let NonASAPOp::TimeRange { range, .. } = child.expect_non_asap() else {
-        panic!("expected TimeRange child, got {child:?}");
-    };
-    assert_eq!(*range, Duration::from_secs(3600));
-}
-
 // ─────────────────────────────────────────────────────────────────────────────
 // C. Aggregation across series             (cheat sheet "Aggregating Over
 //    Multiple Series"; aggregators.test)
@@ -411,46 +366,6 @@ fn group_aggregator_lowers_to_a_distinct_intent() {
 //    `sum(rate(...))` shape; aggregators.test + functions.test)
 // ─────────────────────────────────────────────────────────────────────────────
 
-#[test]
-fn sum_of_rate_is_two_levels() {
-    // SEMANTICS: per-series rate, THEN cross-series sum. Both must survive.
-    let qe = ok("sum(rate(http_requests_total[5m]))");
-    let NonASAPOp::Aggregate {
-        measures, child, ..
-    } = qe.expect_non_asap()
-    else {
-        panic!("expected outer Aggregate{{Sum}}, got {qe:?}");
-    };
-    assert!(matches!(measures.as_slice(), [AggIntent::Sum { .. }]));
-    assert!(matches!(
-        child.expect_non_asap(),
-        NonASAPOp::Aggregate { measures, .. } if matches!(measures.as_slice(), [AggIntent::Rate])
-    ));
-}
-
-#[test]
-fn sum_by_of_rate_groups_outer_level() {
-    // Outer cross-series Sum grouped on positional `Aggregate.by` over the
-    // label-preserving inner Rate. Leaf = [ts, value, instance] → by = [2].
-    let qe = ok("sum by(instance) (rate(node_network_receive_bytes_total[5m]))");
-    let NonASAPOp::Aggregate {
-        reduction,
-        measures,
-        child,
-        ..
-    } = qe.expect_non_asap()
-    else {
-        panic!("expected outer Aggregate grouped by instance, got {qe:?}");
-    };
-    assert_eq!(reduction, &Reduction::by(vec![2]));
-    assert!(matches!(measures.as_slice(), [AggIntent::Sum { .. }]));
-    // child is the inner per-series Rate aggregate.
-    assert!(matches!(
-        child.expect_non_asap(),
-        NonASAPOp::Aggregate { measures, .. } if matches!(measures.as_slice(), [AggIntent::Rate])
-    ));
-}
-
 #[test]
 fn sum_by_of_over_time_groups_outer_level() {
     // Outer cross-series Sum grouped on positional `Aggregate.by` over an inner
@@ -516,16 +431,6 @@ fn over_time_functions_reduce_over_time_range() {
     }
 }
 
-#[test]
-fn quantile_over_time_is_aggregate_over_time_range() {
-    let qe = ok("quantile_over_time(0.9, request_latency_seconds[5m])");
-    assert!(matches!(qe.expect_non_asap(), NonASAPOp::Aggregate { .. }));
-    assert!(has(
-        &qe,
-        |i| matches!(i, AggIntent::Quantile { q, .. } if (*q - 0.9).abs() < 1e-9)
-    ));
-}
-
 // ─────────────────────────────────────────────────────────────────────────────
 // F. Histograms                            (cheat sheet "Quantiles from
 //    Histograms"; histograms.test)
@@ -545,39 +450,6 @@ fn histogram_quantile_over_rate() {
     assert!(has(&qe, |i| matches!(i, AggIntent::Rate)));
 }
 
-#[test]
-fn histogram_quantile_over_sum_by_le_preserves_le_grouping() {
-    // SEMANTICS: the standard pattern — bucket rates summed by `le`, then the
-    // quantile. The `sum by (le)` aggregation must survive into the
-    // canonical DAG.
-    let qe = ok(
-        "histogram_quantile(0.99, sum by(le) (rate(demo_api_request_duration_seconds_bucket[5m])))",
-    );
-    let NonASAPOp::Aggregate {
-        measures, child, ..
-    } = qe.expect_non_asap()
-    else {
-        panic!("expected outer Aggregate{{HistogramQuantile}}, got {qe:?}");
-    };
-    // `by (le)` marks the classic cumulative-bucket form → `HistogramQuantile`.
-    assert!(matches!(
-        measures.as_slice(),
-        [AggIntent::HistogramQuantile { .. }]
-    ));
-    // `sum by(le)` now survives as a positional Aggregate (by = [2], `le`), over
-    // the inner Rate — no name-based Partition.
-    let NonASAPOp::Aggregate {
-        reduction,
-        measures,
-        ..
-    } = child.expect_non_asap()
-    else {
-        panic!("expected `sum by(le)` as a positional Aggregate, got {child:?}");
-    };
-    assert_eq!(reduction, &Reduction::by(vec![2]));
-    assert!(matches!(measures.as_slice(), [AggIntent::Sum { .. }]));
-}
-
 // ─────────────────────────────────────────────────────────────────────────────
 // G. Binary ops: math, matching, comparison  (cheat sheet "Math Between
 //    Series" / "Filtering Series by Value"; operators.test)
@@ -788,16 +660,6 @@ fn set_ops_lower_to_binaryop() {
 //    functions.test, limit.test)
 // ─────────────────────────────────────────────────────────────────────────────
 
-#[test]
-fn topk_over_count_is_heavy_hitter() {
-    // SEMANTICS: top-k by frequency → first-class heavy-hitter `TopK` intent.
-    let qe = ok("topk(10, count_over_time(http_requests_total[1m]))");
-    assert!(has(
-        &qe,
-        |i| matches!(i, AggIntent::TopK { k, .. } if *k == 10)
-    ));
-}
-
 #[test]
 fn bottomk_is_generic_sort_limit() {
     // SEMANTICS: bottom-k → generic ascending order + limit (no sketch).
@@ -880,41 +742,6 @@ fn outer_aggregate_over_nested_aggregate_nests() {
     assert!(has(&qe, |i| matches!(i, AggIntent::Rate)), "rate preserved");
 }
 
-#[test]
-fn outer_group_key_absent_from_nested_aggregate_is_dropped() {
-    // SEMANTICS (PromQL, issue #53): aggregating `by` a label that no input
-    // series carries is valid — every series lands in one group and the
-    // (empty) label is omitted from the output. Here the inner `sum by (group)`
-    // collapses `job` away (its closed output schema is `[group, sum]`), so the
-    // outer `by (job)` groups everything into a single global partition:
-    // the query lowers with the provably-absent key dropped, exactly
-    // `sum(sum by (group)(…))`.
-    let qe = ok(r#"sum(sum by (group)(http_requests{job="api-server"})) by (job)"#);
-    let NonASAPOp::Aggregate {
-        reduction,
-        measures,
-        child,
-        ..
-    } = qe.expect_non_asap()
-    else {
-        panic!("expected outer Aggregate, got {qe:?}");
-    };
-    assert_eq!(
-        reduction,
-        &Reduction::by(vec![]),
-        "absent `job` key dropped → global aggregate"
-    );
-    assert!(matches!(measures.as_slice(), [AggIntent::Sum { .. }]));
-    let NonASAPOp::Aggregate { reduction, .. } = child.expect_non_asap() else {
-        panic!("expected inner `sum by (group)` Aggregate, got {child:?}");
-    };
-    assert_eq!(
-        reduction,
-        &Reduction::by(vec![2]),
-        "inner grouping on `group` survives"
-    );
-}
-
 #[test]
 fn outer_group_key_present_after_inner_aggregate_still_resolves() {
     // The counterpart guard for #53: when the outer key IS in the inner
diff --git a/crates/frontend-promql/tests/promql_lowering.rs b/crates/frontend-promql/tests/promql_lowering.rs
index 6c0f39eb6..3ffbef8cc 100644
--- a/crates/frontend-promql/tests/promql_lowering.rs
+++ b/crates/frontend-promql/tests/promql_lowering.rs
@@ -200,47 +200,6 @@ fn avg_over_time_maps_to_avg_intent() {
     assert_eq!(*range, Duration::from_secs(600));
 }
 
-#[test]
-fn stddev_and_stdvar_over_time() {
-    let qe = lower("stddev_over_time(m[5m])");
-    let NonASAPOp::Aggregate {
-        measures, child, ..
-    } = qe.expect_non_asap()
-    else {
-        panic!("expected Aggregate");
-    };
-    assert!(matches!(
-        measures.as_slice(),
-        [AggIntent::StdDev {
-            population: true,
-            ..
-        }]
-    ));
-    assert!(matches!(
-        child.expect_non_asap(),
-        NonASAPOp::TimeRange { .. }
-    ));
-
-    let qe = lower("stdvar_over_time(m[5m])");
-    let NonASAPOp::Aggregate {
-        measures, child, ..
-    } = qe.expect_non_asap()
-    else {
-        panic!("expected Aggregate");
-    };
-    assert!(matches!(
-        measures.as_slice(),
-        [AggIntent::Variance {
-            population: true,
-            ..
-        }]
-    ));
-    assert!(matches!(
-        child.expect_non_asap(),
-        NonASAPOp::TimeRange { .. }
-    ));
-}
-
 #[test]
 fn histogram_quantile_wraps_inner_in_quantile() {
     // The argument's structure (here `rate`) is preserved *under* the quantile,
@@ -566,22 +525,6 @@ fn count_never_lowers_to_distinct_sample_values() {
     }
 }
 
-#[test]
-fn count_over_time_is_count_intent() {
-    let qe = lower("count_over_time(m[5m])");
-    let NonASAPOp::Aggregate {
-        measures, child, ..
-    } = qe.expect_non_asap()
-    else {
-        panic!("expected Aggregate");
-    };
-    assert!(matches!(measures.as_slice(), [AggIntent::Count { .. }]));
-    assert!(matches!(
-        child.expect_non_asap(),
-        NonASAPOp::TimeRange { .. }
-    ));
-}
-
 #[test]
 fn outer_count_counts_series() {
     // `count by (symbol) (count_over_time(...))`: inner per-series sample count
@@ -1489,13 +1432,11 @@ fn explicit_range_selector_is_a_range_time_range() {
     assert_eq!(*range, Duration::from_secs(300));
 
     let qe = lower("m[5m]");
-    assert!(matches!(
-        qe.expect_non_asap(),
-        NonASAPOp::TimeRange {
-            kind: TimeRangeKind::Range,
-            ..
-        }
-    ));
+    let NonASAPOp::TimeRange { range, kind, .. } = qe.expect_non_asap() else {
+        panic!("expected TimeRange for a range-vector selector, got {qe:?}");
+    };
+    assert_eq!(*kind, TimeRangeKind::Range);
+    assert_eq!(*range, Duration::from_secs(300));
 }
 
 #[test]
diff --git a/crates/frontend-promql/tests/support.rs b/crates/frontend-promql/tests/support.rs
index 686a68a50..dd5f55468 100644
--- a/crates/frontend-promql/tests/support.rs
+++ b/crates/frontend-promql/tests/support.rs
@@ -70,7 +70,9 @@ pub fn promql_scalar(node: &ScalarExpr) -> Option<f64> {
 /// at query time) and export the physical DAG — export needs every node
 /// timed first.
 #[allow(dead_code)]
-pub fn post_asap_dag(root: &Rc<OperatorNode>) -> asap_types::ir::physical_export::PhysicalASAPDAG {
+pub fn physical_asap_dag(
+    root: &Rc<OperatorNode>,
+) -> asap_types::ir::physical_export::PhysicalASAPDAG {
     use asap_types::ir::{apply_materialization_timings, MaterializationAssignment, TimingMemo};
     let timed = apply_materialization_timings(
         root,
diff --git a/crates/frontend-promql/tests/univmon_candidates.rs b/crates/frontend-promql/tests/univmon_candidates.rs
index 56a22bdd7..bae05b125 100644
--- a/crates/frontend-promql/tests/univmon_candidates.rs
+++ b/crates/frontend-promql/tests/univmon_candidates.rs
@@ -8,7 +8,7 @@ use asap_types::ir::properties::AccuracyMetric;
 use asap_types::ir::properties::AccuracyTarget;
 use asap_types::ir::schema::{FieldDataType, SketchAlgorithm, SketchStatistic, SummaryInputExpr};
 use asap_types::ir::{ASAPOp, Operator, OperatorNode};
-use support::{logical_candidates, lower_promql, post_asap_dag, run_stage1_to_stage3};
+use support::{logical_candidates, lower_promql, physical_asap_dag, run_stage1_to_stage3};
 
 /// The summary evaluation over a UnivMon state that `root` reads, if any.
 fn univmon_estimate(root: &Rc<OperatorNode>) -> Option<Rc<OperatorNode>> {
@@ -113,7 +113,7 @@ fn four_evaluations_size_their_value_frequency_states_and_keep_honest_guarantees
                 assert!(local_guarantee(family, query).is_none());
             }
         }
-        post_asap_dag(root);
+        physical_asap_dag(root);
     }
 }
 
diff --git a/crates/integration-tests/src/lib.rs b/crates/integration-tests/src/lib.rs
index ca388eb37..73636bf79 100644
--- a/crates/integration-tests/src/lib.rs
+++ b/crates/integration-tests/src/lib.rs
@@ -94,8 +94,8 @@ pub mod fixtures {
     }
 }
 
-/// Timing and export helpers for post-ASAP plans.
-pub mod post_asap {
+/// Timing and physical ASAP DAG export helpers for planned candidates.
+pub mod physical_export {
     use asap_types::ir::physical_export::{compile_physical_asap_dag, PhysicalASAPDAG};
     use asap_types::ir::{
         apply_materialization_timings, MaterializationAssignment, OperatorNode, TimingMemo,
@@ -128,14 +128,14 @@ pub mod post_asap {
     }
 
     /// Time `root` (default assignment), then export the physical DAG.
-    pub fn post_asap_dag(root: &Rc<OperatorNode>) -> PhysicalASAPDAG {
-        compile_physical_asap_dag(&timed(root)).expect("post-ASAP DAG export failed")
+    pub fn physical_asap_dag(root: &Rc<OperatorNode>) -> PhysicalASAPDAG {
+        compile_physical_asap_dag(&timed(root)).expect("physical ASAP DAG export failed")
     }
 
     /// Time `root` with every sub-DAG `MatAtIngestion` where the DAG allows
     /// it, then export the physical DAG.
-    pub fn at_ingestion_post_asap_dag(root: &Rc<OperatorNode>) -> PhysicalASAPDAG {
-        compile_physical_asap_dag(&at_ingestion(root)).expect("post-ASAP DAG export failed")
+    pub fn at_ingestion_physical_asap_dag(root: &Rc<OperatorNode>) -> PhysicalASAPDAG {
+        compile_physical_asap_dag(&at_ingestion(root)).expect("physical ASAP DAG export failed")
     }
 }
 
diff --git a/crates/integration-tests/tests/exact_composition.rs b/crates/integration-tests/tests/exact_composition.rs
index b8ecdefed..ba30f6793 100644
--- a/crates/integration-tests/tests/exact_composition.rs
+++ b/crates/integration-tests/tests/exact_composition.rs
@@ -9,8 +9,8 @@
 
 use std::rc::Rc;
 
+use asap_integration_tests::physical_export::{at_ingestion, physical_asap_dag, timed};
 use asap_integration_tests::planning::logical_candidates;
-use asap_integration_tests::post_asap::{at_ingestion, post_asap_dag, timed};
 use asap_types::ir::operator::agg_intent::{default_quantile, AggIntent};
 use asap_types::ir::operator::operator_properties::{Reduction, Source};
 use asap_types::ir::physical_export::PhysicalASAPOperatorPayload;
@@ -322,7 +322,7 @@ fn physical_export_carries_explicit_timing_for_a_composed_plan() {
     let composed = composed(&root);
     // Timing is explicit in the wire-6 DAG: the root finalizes the exact
     // `max` state over the quantile evaluation, at query time.
-    let wire = post_asap_dag(&composed);
+    let wire = physical_asap_dag(&composed);
     let wire_root = wire.nodes.iter().find(|n| n.id == wire.roots[0]).unwrap();
     assert!(matches!(
         wire_root.payload,
diff --git a/crates/integration-tests/tests/operator_design_examples.rs b/crates/integration-tests/tests/operator_design_examples.rs
index e7aa1966d..f2dc4ec59 100644
--- a/crates/integration-tests/tests/operator_design_examples.rs
+++ b/crates/integration-tests/tests/operator_design_examples.rs
@@ -114,8 +114,6 @@ async fn sql_sum_projection_before_and_after_summary_rewrite() {
     let wire = physical_common::compile_physical_asap_dag(&rewritten).unwrap();
     wire.validate().unwrap();
     assert_eq!(wire.nodes.len(), dag.len());
-    let json = serde_json::to_string(&wire).unwrap();
-    assert!(!json.contains("KeepPreAsap") && !json.contains("ScalarBridge"));
 }
 
 /// Scalar subqueries survive normalization with shared, visible producers.
diff --git a/crates/integration-tests/tests/operator_sharing.rs b/crates/integration-tests/tests/operator_sharing.rs
index 5aa27f9e7..fa9369b44 100644
--- a/crates/integration-tests/tests/operator_sharing.rs
+++ b/crates/integration-tests/tests/operator_sharing.rs
@@ -9,11 +9,11 @@
 use std::rc::Rc;
 
 use asap_frontend_sql::{lower_sql, SqlCatalog};
+use asap_integration_tests::physical_export::physical_asap_dag;
 use asap_integration_tests::planning::selected_dag;
-use asap_integration_tests::post_asap::post_asap_dag;
 use asap_types::ir::properties::AccuracyTarget;
 use asap_types::ir::schema::{DataType, Field, Schema};
-use asap_types::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode};
+use asap_types::ir::{ASAPOp, NonASAPOp, OperatorNode};
 
 fn col(name: &str, dtype: DataType) -> Field {
     Field::plain(name, dtype, false)
@@ -64,63 +64,10 @@ fn find_all(
         .collect()
 }
 
-fn is_summary_evaluation(node: &OperatorNode) -> bool {
-    matches!(
-        node.operator,
-        Operator::ASAP(
-            ASAPOp::SummaryEstimate { .. }
-                | ASAPOp::FinalizeExactAccumulator { .. }
-                | ASAPOp::EvaluatePopulation { .. }
-        )
-    )
-}
-
 fn is_scan(node: &OperatorNode) -> bool {
     matches!(node.non_asap(), Some(NonASAPOp::Scan { .. }))
 }
 
-/// The first node reached through single-input non-ASAP operators below
-/// `node` (inclusive) that is not one: where an operator chain meets a
-/// summary or a multi-input operator.
-fn through_unary_non_asap(node: &Rc<OperatorNode>) -> &Rc<OperatorNode> {
-    match node.non_asap().map(|op| op.children()) {
-        Some(children) if children.len() == 1 => through_unary_non_asap(children[0]),
-        _ => node,
-    }
-}
-
-// #468 problem 1: the Project above the summary evaluation and the Scan below
-// it are both plain NonASAP nodes (no post-ASAP-only wrapper variant).
-#[ignore = "planner chooses no summary here: Avg has no summary realization, so the Aggregate stays a logical pass-through"]
-#[tokio::test]
-async fn project_above_and_scan_below_a_summary_are_both_non_asap_nodes() {
-    let root = plan(
-        "WITH metric AS (SELECT avg(CASE WHEN l_quantity BETWEEN 1 AND 50 THEN 1.0 ELSE 0.0 END) \
-         AS in_range FROM lineitem) SELECT in_range, in_range = 1.0 AS ok FROM metric",
-        AccuracyTarget::exact(),
-    )
-    .await;
-    // The root is the outer SELECT list: a NonASAP Project.
-    assert!(
-        matches!(root.operator, Operator::NonASAP(NonASAPOp::Project { .. })),
-        "root must be the outer Project, got {:?}",
-        root.operator
-    );
-    // A summary evaluation sits below the Project chain.
-    let evaluation = through_unary_non_asap(&root);
-    assert!(
-        is_summary_evaluation(evaluation),
-        "the Project chain must read a summary, got {:?}",
-        evaluation.operator
-    );
-    // Below the summary the Scan is the same NonASAP operator a front end emits.
-    let scans = find_all(evaluation, is_scan);
-    assert_eq!(scans.len(), 1, "one lineitem Scan below the summary");
-    assert!(!scans[0].is_asap());
-    // The flat plan exports (time first, wire 6).
-    post_asap_dag(&root);
-}
-
 // #468 problem 2: the exact aggregate and the sketch read one shared Scan
 // (`Rc::ptr_eq`), not two copies.
 #[ignore = "waits for the binding rule splitting multi-measure aggregates"]
@@ -157,7 +104,7 @@ async fn exact_aggregate_and_sketch_share_one_scan() {
 // #468 problem 3: a summary can sit below a set operator — each side of the
 // UNION ALL holds its own SummaryEstimate.
 #[tokio::test]
-#[ignore = "plan selection picks the exact plan: a query-time summary never costs less until physical ASAP-aware optimization plans materialization (#604)"]
+#[ignore = "plan selection picks the exact plan: a query-time summary on each side never costs less than the exact aggregate until Pass 2 shares it (#580)"]
 async fn each_side_of_union_all_holds_a_summary_estimate() {
     let root = plan(
         "SELECT approx_distinct(l_partkey) FROM lineitem \
@@ -185,5 +132,5 @@ async fn each_side_of_union_all_holds_a_summary_estimate() {
             side.operator
         );
     }
-    post_asap_dag(&root);
+    physical_asap_dag(&root);
 }
diff --git a/crates/integration-tests/tests/planner_layering_example1.rs b/crates/integration-tests/tests/planner_layering_example1.rs
index 54b27f858..aa6b510fd 100644
--- a/crates/integration-tests/tests/planner_layering_example1.rs
+++ b/crates/integration-tests/tests/planner_layering_example1.rs
@@ -278,7 +278,7 @@ fn stage1_logical_asap_optimization_q2_summary_families() {
 
 /// Pass 1 also offers Hydra (HydraCms over all (job, series) keys, a heap per job) for Q2's top-k.
 #[test]
-#[ignore = "needs Pass 1 Hydra for grouped approximate counts (#605)"]
+#[ignore = "needs Pass 1 Hydra for grouped approximate counts (#703)"]
 fn stage1_logical_asap_optimization_q2_offers_hydra() {
     let run = pipeline();
     assert!(run
@@ -568,23 +568,6 @@ fn compile_in_runtime(p: &Physical) -> Result<(), String> {
         .map_err(|e| format!("{}: {e}", p.id))
 }
 
-/// Runtime capability check (added by the implementer, not part of the
-/// spec): the physical planner compiles every physical candidate.
-#[test]
-#[ignore = "runtime gaps: a physical candidate's CMS+heap weights are an open requirement \
-            that only the selected plan's confirmation answers; the heaps' \
-            SummaryEstimate{TopK} output is not the keyed-evaluation shape the physical \
-            planner builds (#579)"]
-fn stage2_physical_asap_optimization_every_candidate_compiles_in_the_physical_planner() {
-    let run = pipeline();
-    let failures: Vec<_> = run
-        .physical
-        .iter()
-        .filter_map(|p| compile_in_runtime(p).err())
-        .collect();
-    assert!(failures.is_empty(), "{failures:#?}");
-}
-
 // ── Plan selection ───────────────────────────────────────────────────────
 
 /// Plan selection selects one candidate and gives every other one a reason.
diff --git a/crates/integration-tests/tests/planner_layering_example2.rs b/crates/integration-tests/tests/planner_layering_example2.rs
index 0757ece3c..32cc8712e 100644
--- a/crates/integration-tests/tests/planner_layering_example2.rs
+++ b/crates/integration-tests/tests/planner_layering_example2.rs
@@ -363,7 +363,7 @@ fn stage1_logical_asap_optimization_pass1_offers_a_distinct_count_summary() {
 
 /// Pass 1 offers a specialized entropy summary for Q2 and a norm summary for Q3.
 #[test]
-#[ignore = "needs Pass 1 specialized entropy and L2 summary families (no PR yet)"]
+#[ignore = "needs Pass 1 specialized entropy and L2 summary families (#444)"]
 fn stage1_logical_asap_optimization_pass1_offers_specialized_entropy_and_l2_summaries() {
     let run = pipeline();
     for q in [1, 2] {
@@ -564,7 +564,7 @@ fn stage3_plan_selection_has_no_theta_candidate() {
 
 /// A UnivMon estimate has a guarantee, so plan selection judges it against its target instead of rejecting it for having none.
 #[test]
-#[ignore = "needs a UnivMon entropy accuracy model; L2 and the distinct count have one"]
+#[ignore = "needs a UnivMon entropy accuracy model (#788); L2 and the distinct count have one"]
 fn stage3_plan_selection_judges_univmon_with_an_accuracy_model() {
     let run = pipeline();
     let invalid = run.invalid();
@@ -877,7 +877,7 @@ async fn stage3_plan_selection_accepts_sql_univmon_l2_beside_its_exact_populatio
 
 /// One materialized UnivMon for all three costs no more than three separate UnivMons.
 #[test]
-#[ignore = "needs a UnivMon entropy accuracy model and materializing the shared UnivMon (#604): not materialized, it is charged once per consuming query"]
+#[ignore = "needs a UnivMon entropy accuracy model (#788): without one, the shared UnivMon candidate is left unpriced"]
 fn stage3_plan_selection_shared_univmon_costs_no_more_than_three() {
     let run = pipeline();
     let cost = |c: &Logical| run.cost(&run.physical_of(c).next().unwrap().id);
diff --git a/crates/integration-tests/tests/planner_layering_example3.rs b/crates/integration-tests/tests/planner_layering_example3.rs
index 32fc43311..68d643a5a 100644
--- a/crates/integration-tests/tests/planner_layering_example3.rs
+++ b/crates/integration-tests/tests/planner_layering_example3.rs
@@ -338,7 +338,7 @@ fn stage1_logical_asap_optimization_b_pass1_offers_exact_and_kll() {
 
 /// Each Pass 1 option gets three window forms: none, a 5-min sliding window with a 1-min slide, and 1-min tumbling windows.
 #[test]
-#[ignore = "needs Pass 2 window composition: sliding windows (no PR yet); tumbling windows pass in \
+#[ignore = "needs Pass 2 window composition: sliding windows (#522); tumbling windows pass in \
             stage1_logical_asap_optimization_b_window_composition_adds_tumbling_per_mergeable_option"]
 fn stage1_logical_asap_optimization_b_window_composition_adds_sliding_and_tumbling_per_option() {
     let run = run_b();
@@ -444,7 +444,7 @@ fn stage1_logical_asap_optimization_b_window_parameters_are_legal() {
 
 /// A sliding window with L = W reads one completed window (no merge); tumbling windows merge W / length of them.
 #[test]
-#[ignore = "needs Pass 2 window composition: sliding windows (no PR yet); tumbling merges pass in \
+#[ignore = "needs Pass 2 window composition: sliding windows (#522); tumbling merges pass in \
             stage1_logical_asap_optimization_b_window_composition_adds_tumbling_per_mergeable_option"]
 fn stage1_logical_asap_optimization_b_merge_only_where_the_window_form_needs_it() {
     let run = run_b();
diff --git a/crates/integration-tests/tests/planner_layering_example4.rs b/crates/integration-tests/tests/planner_layering_example4.rs
index 136618a57..bfd7ce6a7 100644
--- a/crates/integration-tests/tests/planner_layering_example4.rs
+++ b/crates/integration-tests/tests/planner_layering_example4.rs
@@ -127,13 +127,24 @@ use MaterializationOption::{MatAtIngestion, MatAtQuery, NotMaterialized};
 // ── Constraints over every candidate ─────────────────────────────────────
 
 /// Every node upstream of an ingestion-time node also runs at ingestion time, in every workload variant.
+/// The monthly variant runs in
+/// `stage2_physical_asap_optimization_monthly_ingestion_time_upstream_is_ingestion_time`.
 #[test]
 fn stage2_physical_asap_optimization_ingestion_time_upstream_is_ingestion_time() {
-    for workload in [once(), monthly(), at_rest(), pattern_b()] {
+    for workload in [once(), at_rest(), pattern_b()] {
         assert_ingestion_upstream_is_ingestion(&run_promql(&workload));
     }
 }
 
+/// The monthly variant of
+/// `stage2_physical_asap_optimization_ingestion_time_upstream_is_ingestion_time`:
+/// it plans 67,228 candidates, too slow for a debug build in CI.
+#[test]
+#[ignore = "slow: run in release by hand"]
+fn stage2_physical_asap_optimization_monthly_ingestion_time_upstream_is_ingestion_time() {
+    assert_ingestion_upstream_is_ingestion(&run_promql(&monthly()));
+}
+
 /// Data at rest has no ingestion, so no candidate runs anything at ingestion time.
 #[test]
 fn stage2_physical_asap_optimization_at_rest_runs_nothing_at_ingestion_time() {
@@ -145,36 +156,59 @@ fn stage2_physical_asap_optimization_at_rest_runs_nothing_at_ingestion_time() {
     }
 }
 
+/// Each Pattern A query's range window and offset, as its PromQL text states them.
+fn pattern_a_windows_and_offsets() -> (Vec<u64>, Vec<u64>) {
+    let windows = PATTERN_A.iter().map(|p| p.window_ms).collect();
+    let offsets = PATTERN_A.iter().map(|p| p.offset_before_t_ms).collect();
+    (windows, offsets)
+}
+
+/// Every materialized output of every candidate of `workload` is kept for at
+/// least its longest reader's `windows[q] + offsets[q]`.
+fn assert_materialized_outputs_cover_their_consumers(
+    workload: &PlanningWorkload,
+    windows: &[u64],
+    offsets: &[u64],
+) {
+    let run = run_promql(workload);
+    for p in &run.physical {
+        for n in &p.dag().nodes {
+            if materialization(p, n.id) == NotMaterialized {
+                continue;
+            }
+            let need = readers(p.dag(), &p.query_roots, n.id)
+                .into_iter()
+                .map(|q| windows[q] + offsets[q])
+                .max()
+                .unwrap_or(0);
+            let kept = retention_ms(p, n.id).unwrap_or_else(|| {
+                panic!("{}: {:?} is materialized without a retention", p.id, n.id)
+            });
+            assert!(kept >= need, "{}: {:?} kept {kept} < {need}", p.id, n.id);
+        }
+    }
+}
+
 /// A materialized output is kept for as long as its consumers read: it covers every reader's lookback and offset.
+/// The monthly variant runs in
+/// `stage2_physical_asap_optimization_monthly_materialized_output_covers_its_consumers`.
 #[test]
 fn stage2_physical_asap_optimization_materialized_output_covers_its_consumers() {
-    // Each query's range window and offset, as its PromQL text states them.
-    let windows: Vec<u64> = PATTERN_A.iter().map(|p| p.window_ms).collect();
-    let offsets: Vec<u64> = PATTERN_A.iter().map(|p| p.offset_before_t_ms).collect();
-    for (workload, windows, offsets) in [
-        (once(), windows.clone(), offsets.clone()),
-        (monthly(), windows.clone(), offsets.clone()),
-        (at_rest(), windows, offsets),
-        (pattern_b(), vec![PATTERN_B_WINDOW_MS], vec![0]),
-    ] {
-        let run = run_promql(&workload);
-        for p in &run.physical {
-            for n in &p.dag().nodes {
-                if materialization(p, n.id) == NotMaterialized {
-                    continue;
-                }
-                let need = readers(p.dag(), &p.query_roots, n.id)
-                    .into_iter()
-                    .map(|q| windows[q] + offsets[q])
-                    .max()
-                    .unwrap_or(0);
-                let kept = retention_ms(p, n.id).unwrap_or_else(|| {
-                    panic!("{}: {:?} is materialized without a retention", p.id, n.id)
-                });
-                assert!(kept >= need, "{}: {:?} kept {kept} < {need}", p.id, n.id);
-            }
-        }
+    let (windows, offsets) = pattern_a_windows_and_offsets();
+    for workload in [once(), at_rest()] {
+        assert_materialized_outputs_cover_their_consumers(&workload, &windows, &offsets);
     }
+    assert_materialized_outputs_cover_their_consumers(&pattern_b(), &[PATTERN_B_WINDOW_MS], &[0]);
+}
+
+/// The monthly variant of
+/// `stage2_physical_asap_optimization_materialized_output_covers_its_consumers`:
+/// it plans 67,228 candidates, too slow for a debug build in CI.
+#[test]
+#[ignore = "slow: run in release by hand"]
+fn stage2_physical_asap_optimization_monthly_materialized_output_covers_its_consumers() {
+    let (windows, offsets) = pattern_a_windows_and_offsets();
+    assert_materialized_outputs_cover_their_consumers(&monthly(), &windows, &offsets);
 }
 
 // ── Pattern A ────────────────────────────────────────────────────────────
@@ -354,7 +388,7 @@ fn stage2_physical_asap_optimization_b_b3_keeps_query_time_windows() {
 
 /// The sliding-window KLL is kept from ingestion time or from query time; not materializing it is the no-window plan.
 #[test]
-#[ignore = "needs sliding windows (no PR yet)"]
+#[ignore = "needs sliding windows (#522)"]
 fn stage2_physical_asap_optimization_b_sliding_kll_has_two_materialization_options() {
     let found: BTreeSet<_> = options_of(&run_promql(&pattern_b()), sliding(), 1)
         .into_keys()
@@ -362,30 +396,6 @@ fn stage2_physical_asap_optimization_b_sliding_kll_has_two_materialization_optio
     assert_eq!(found, BTreeSet::from([MatAtIngestion, MatAtQuery]));
 }
 
-/// B2 rebuilds all five tumbling KLLs at every evaluation, so it costs at least B1 and B3.
-#[test]
-#[ignore = "the built-in model contradicts it: with query-time retention, B1 keeps 6 panes of 1M per-series KLLs for 13,326.5 cost/s and B3 keeps 4 for 8,884.5, while B2 rebuilds them for 2.183; needs the model-relative rewrite (Q49)"]
-fn stage3_plan_selection_b_rebuilding_every_window_costs_most() {
-    let run = run_promql_for(&pattern_b(), &keeping_query_time_results());
-    let options = options_of(&run, tumbling(), 1);
-    let b2 = cost(&run, &options[&NotMaterialized].0);
-    for m in [MatAtIngestion, MatAtQuery] {
-        assert!(cost(&run, &options[&m].0) <= b2, "{m:?} vs B2 {b2}");
-    }
-}
-
-/// Repeating over arriving data, the built-in models pick B1 among the tumbling options.
-#[test]
-#[ignore = "the built-in model contradicts it: B1 keeps 6 panes of 1M per-series KLLs (107 GB) for 13,326.5 cost/s, B2 rebuilds them for 2.183; needs the model-relative rewrite (Q49)"]
-fn stage3_plan_selection_b_prefers_ingestion_time_tumbling_windows() {
-    let run = run_promql(&pattern_b());
-    let options = options_of(&run, tumbling(), 1);
-    let b1 = cost(&run, &options[&MatAtIngestion].0);
-    for (m, (id, _)) in &options {
-        assert!(b1 <= cost(&run, id), "B1 {b1} vs {m:?} {}", cost(&run, id));
-    }
-}
-
 // ── Cost models (#683, #660) ─────────────────────────────────────────────
 
 /// Pattern B under `cost_model`, with the default resource speed and horizon.
diff --git a/crates/integration-tests/tests/promql_numeric_regressions.rs b/crates/integration-tests/tests/promql_numeric_regressions.rs
index 425efc46d..c9257fb49 100644
--- a/crates/integration-tests/tests/promql_numeric_regressions.rs
+++ b/crates/integration-tests/tests/promql_numeric_regressions.rs
@@ -1,8 +1,8 @@
 //! Numeric regression fixtures: actual PromQL lowering plus numeric update/evaluation checks.
 //! The count/sum interpreter below verifies planner update semantics, not a deployed backend.
 use asap_integration_tests::fixtures::lower_promql;
+use asap_integration_tests::physical_export::physical_asap_dag;
 use asap_integration_tests::planning::logical_candidates;
-use asap_integration_tests::post_asap::post_asap_dag;
 use asap_types::ir::operator::Reduction;
 use asap_types::ir::properties::AccuracyTarget;
 use asap_types::ir::scalar::ColumnRef;
@@ -67,7 +67,7 @@ fn count_up_counts_targets_even_when_values_repeat_or_change_sign() {
             3.
         );
     }
-    post_asap_dag(&node);
+    physical_asap_dag(&node);
 }
 
 /// Ten samples give count ten, whereas sum retains the signed sample values.
@@ -85,7 +85,7 @@ fn window_counts_and_sums_distinguish_one_zero_three_and_negative_values() {
             let got: f64 = (0..10).map(|_| contribution(family, update, value)).sum();
             assert_eq!(got, if is_count { 10. } else { value * 10. });
         }
-        post_asap_dag(&node);
+        physical_asap_dag(&node);
     }
 }
 
@@ -101,7 +101,7 @@ fn sum_rate_and_increase_have_real_exact_accumulator_nodes() {
         let (family, _, _) = aggregate(&node);
         assert!(matches!(family, FieldDataType::ExactAggregate(k, _) if *k == kind));
         assert!(node.guarantee.as_ref().unwrap().is_exact());
-        post_asap_dag(&node);
+        physical_asap_dag(&node);
     }
 }
 
@@ -124,7 +124,7 @@ fn checked_ratio_must_not_certify_cross_zero_interpolation() {
         "direct quantile ratio should remain an available candidate"
     );
     assert!(node.guarantee.is_none());
-    post_asap_dag(&node);
+    physical_asap_dag(&node);
     // Keep the actual signed-sketch counterexample: division guards alone pass
     // even though the quantile interpolation does not preserve relative error.
     // The sketch reads its minimum and maximum exactly, so the interpolated
@@ -181,7 +181,7 @@ fn quantile_over_temporal_average_keeps_a_legal_candidate() {
                 })
             })
             .unwrap_or_else(|| panic!("outer sketch candidate must survive: {query}"));
-        post_asap_dag(&node);
+        physical_asap_dag(&node);
     }
 }
 
@@ -273,7 +273,7 @@ fn sketch_counts_use_unit_weights_and_signed_sums_keep_value_weights() {
             };
             assert_eq!(got, if is_count { 10. } else { 10. * value });
         }
-        post_asap_dag(node);
+        physical_asap_dag(node);
     }
 }
 
diff --git a/crates/integration-tests/tests/promql_to_post_asap.rs b/crates/integration-tests/tests/promql_to_post_asap.rs
index ef15068ff..3c068fbd4 100644
--- a/crates/integration-tests/tests/promql_to_post_asap.rs
+++ b/crates/integration-tests/tests/promql_to_post_asap.rs
@@ -9,8 +9,10 @@
 use std::rc::Rc;
 
 use asap_integration_tests::fixtures::lower_promql;
+use asap_integration_tests::physical_export::{
+    at_ingestion, at_ingestion_physical_asap_dag, physical_asap_dag,
+};
 use asap_integration_tests::planning::logical_candidates;
-use asap_integration_tests::post_asap::{at_ingestion, at_ingestion_post_asap_dag, post_asap_dag};
 use asap_logical_asap_optimization::accuracy::OperandDomain;
 use asap_physical_asap_optimization::PhysicalCandidate;
 use asap_plan_selection::AccuracyEvidenceProvider;
@@ -647,7 +649,7 @@ fn nested_summary_explicitly_finalizes_exact_child_at_ingestion_time() {
         .iter()
         .any(|field| matches!(field.dtype, FieldDataType::Plain(DataType::Float64))));
     // Explicit boundary is a valid post-ASAP DAG.
-    post_asap_dag(&plan);
+    physical_asap_dag(&plan);
 }
 
 #[test]
@@ -660,7 +662,7 @@ fn physical_node_owns_phase_independently_of_binary_payload() {
     let expected = ExecutionTiming::QueryTime;
     let input = lower_promql(query, AccuracyTarget::max_error(0.05)).unwrap();
     let plan = realize(&input, AccuracyTarget::max_error(0.05));
-    let dag = at_ingestion_post_asap_dag(&plan);
+    let dag = at_ingestion_physical_asap_dag(&plan);
     let node = dag
         .nodes
         .iter()
@@ -861,7 +863,7 @@ fn without_aggregation_candidates_export_valid_dags() {
             let candidates = logical_candidates(root, accuracy.clone());
             assert!(!candidates.is_empty(), "{query}");
             for node in &candidates {
-                post_asap_dag(node);
+                physical_asap_dag(node);
             }
         }
     }
diff --git a/crates/integration-tests/tests/sql_to_post_asap.rs b/crates/integration-tests/tests/sql_to_post_asap.rs
index c663b70d5..7b77904a2 100644
--- a/crates/integration-tests/tests/sql_to_post_asap.rs
+++ b/crates/integration-tests/tests/sql_to_post_asap.rs
@@ -20,8 +20,8 @@
 use std::rc::Rc;
 
 use asap_frontend_sql::{lower_sql, lower_sql_dialect, SqlCatalog};
+use asap_integration_tests::physical_export::physical_asap_dag;
 use asap_integration_tests::planning::logical_candidates;
-use asap_integration_tests::post_asap::post_asap_dag;
 use asap_types::ir::operator::operator_properties::Reduction;
 use asap_types::ir::physical_export::{EdgeRole, PhysicalASAPNodeId, PhysicalASAPOperatorPayload};
 use asap_types::ir::properties::AccuracyTarget;
@@ -226,7 +226,7 @@ async fn clickhouse_outer_sum_recursively_binds_inner_temporal_aggregate() {
             family,
             &FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum)
         );
-        let dag = post_asap_dag(&root);
+        let dag = physical_asap_dag(&root);
         assert_eq!(
             dag.nodes
                 .iter()
@@ -374,7 +374,7 @@ async fn sql_join_recursively_binds_both_temporal_aggregate_children() {
         .guarantee
         .as_ref()
         .is_some_and(|value| value.is_exact()));
-    let dag = post_asap_dag(&root);
+    let dag = physical_asap_dag(&root);
     let join_id = dag
         .nodes
         .iter()
@@ -522,7 +522,7 @@ async fn sql_filter_keeps_read_predicate_and_summary_population_selection() {
     }
     assert_eq!(retained_read_predicate, Some(expected_read_predicate));
 
-    let dag = post_asap_dag(&root);
+    let dag = physical_asap_dag(&root);
     let expected_wire = wire_pred(retained_read_predicate.as_ref().unwrap());
     assert!(dag.nodes.iter().any(|node| matches!(
         &node.payload,
@@ -695,7 +695,7 @@ async fn map_projection_export_keeps_the_map_over_an_avg_of_exact_states() {
         &catalog(), SqlDialect::ClickhouseSQL, AccuracyTarget::exact(),
     ).await.unwrap();
     let root = realize(&pre, AccuracyTarget::exact());
-    let dag = post_asap_dag(&root);
+    let dag = physical_asap_dag(&root);
     assert!(dag.nodes.iter().any(|node| matches!(&node.payload,
         PhysicalASAPOperatorPayload::NonASAP(NonASAPOp::Project { cols, .. })
         if cols.iter().any(|item| matches!(&item.expr, ScalarExpr::FunctionCall { name, .. } if name == "map"))
diff --git a/crates/types/src/ir/cse.rs b/crates/types/src/ir/cse.rs
index 8ab2541bb..c5c7227d8 100644
--- a/crates/types/src/ir/cse.rs
+++ b/crates/types/src/ir/cse.rs
@@ -223,13 +223,6 @@ impl InternTable {
     }
 }
 
-/// Count of *unique* nodes reachable from `root` (pointer identity,
-/// following [`OperatorNode::children`]): the real size of the DAG, not a
-/// tree-walk count that re-counts a shared descendant once per parent.
-pub fn dag_node_count(root: &Rc<OperatorNode>) -> usize {
-    OperatorNode::reachable(root).len()
-}
-
 /// Input pointer → (input `Rc`, interned result). The input `Rc` is retained
 /// so its address cannot be freed and reused by a fresh allocation while the
 /// memo still maps it.
@@ -662,52 +655,6 @@ mod tests {
         );
     }
 
-    // ── dag_node_count ───────────────────────────────────────────────────
-
-    #[test]
-    fn dag_node_count_is_the_naive_count_when_nothing_is_shared() {
-        assert_eq!(dag_node_count(&scan()), 1);
-        assert_eq!(dag_node_count(&quantile_agg(vec![1], Some(2), 0.5)), 2);
-        assert_eq!(
-            dag_node_count(&scalar_of_vector()),
-            3,
-            "follows scalar references"
-        );
-    }
-
-    #[test]
-    fn dag_node_count_deduplicates_an_internally_shared_sub_dag() {
-        let root = compare(
-            quantile_agg(vec![1], Some(2), 0.5),
-            quantile_agg(vec![1], Some(2), 0.5),
-        );
-        assert_eq!(
-            dag_node_count(&root),
-            5,
-            "fixture sanity: nothing shared yet"
-        );
-        let shared = share_common_sub_dags(vec![("q", root)]);
-        let [(_, root)] = shared.as_slice() else {
-            panic!("expected 1 root");
-        };
-        assert_eq!(
-            dag_node_count(root),
-            3,
-            "BinaryOp + one Aggregate + its Scan"
-        );
-    }
-
-    #[test]
-    fn dag_node_count_deduplicates_across_two_workload_roots() {
-        let (ra, rb) = two_roots(
-            quantile_agg(vec![1], Some(2), 0.5),
-            quantile_agg(vec![1], Some(2), 0.5),
-        );
-        assert!(Rc::ptr_eq(&ra, &rb), "fixture sanity: the two roots merged");
-        assert_eq!(dag_node_count(&ra), 2);
-        assert_eq!(dag_node_count(&rb), 2);
-    }
-
     #[test]
     fn dedup_gates_sharing_the_same_as_aggregate() {
         // `Dedup { cols }` adds `cols` as a unique key, so two identical
@@ -753,7 +700,7 @@ mod tests {
             panic!("expected BinaryOp root");
         };
         assert!(Rc::ptr_eq(lhs, rhs));
-        assert_eq!(dag_node_count(&shared[0].1), 3);
+        assert_eq!(OperatorNode::reachable(&shared[0].1).len(), 3);
     }
 
     // Comparing a shareable node whose equal-but-unshareable children form a
diff --git a/docs/design_docs/decisions/cse-cost-model.md b/docs/design_docs/decisions/cse-cost-model.md
index 2952e39ce..60feabbe8 100644
--- a/docs/design_docs/decisions/cse-cost-model.md
+++ b/docs/design_docs/decisions/cse-cost-model.md
@@ -103,7 +103,7 @@ does not prune them.
 
 ## Defaults
 
-`cse_recompute_cost`'s default is a structural-size proxy: `cse::dag_node_count`,
+`cse_recompute_cost`'s default is a structural-size proxy: `OperatorNode::reachable(root).len()`,
 the number of *unique* nodes in the sub-DAG's DAG (deduplicated by `Rc`
 pointer identity), not a raw serialization length. This distinction matters
 here specifically — a `CseCandidate`'s sub-DAG is, by definition, something
diff --git a/docs/design_docs/proposals/planner-layering-example2-acceptance.md b/docs/design_docs/proposals/planner-layering-example2-acceptance.md
index ff2594c51..0ba0a9972 100644
--- a/docs/design_docs/proposals/planner-layering-example2-acceptance.md
+++ b/docs/design_docs/proposals/planner-layering-example2-acceptance.md
@@ -68,7 +68,7 @@ workload itself.
 | Once recognized, the SQL workload has the stand-in's Pass 1 options | `stage1_logical_asap_optimization_sql_has_the_standin_options` | passes |
 | Each statistic has an exact and a UnivMon option | `stage1_logical_asap_optimization_pass1_offers_exact_and_univmon_for_each_statistic` | passes |
 | Q1 has a specialized distinct-count summary (HLL or KMV) | `stage1_logical_asap_optimization_pass1_offers_a_distinct_count_summary` | passes |
-| Q2 and Q3 have a specialized entropy and norm summary | `stage1_logical_asap_optimization_pass1_offers_specialized_entropy_and_l2_summaries` | ignored: no PR yet |
+| Q2 and Q3 have a specialized entropy and norm summary | `stage1_logical_asap_optimization_pass1_offers_specialized_entropy_and_l2_summaries` | ignored: needs Pass 1 specialized entropy and L2 summary families (#444) |
 | Every combination of the queries' own options is kept, with no shared summary | `stage1_logical_asap_optimization_keeps_every_independent_combination` | passes |
 | A candidate has one UnivMon build read by all three queries, feeding a distinct-count, an entropy and an L2 estimate | `stage1_logical_asap_optimization_summary_capability_adds_one_univmon_for_all_three` | passes |
 | For each pair, a candidate shares one UnivMon between the two while the third takes each of its own options | `stage1_logical_asap_optimization_summary_capability_adds_pairwise_shared_univmons` | passes |
@@ -98,11 +98,11 @@ evaluation.
 | The selected plan is the cheapest valid one | `stage3_plan_selection_selects_cheapest_valid` | passes |
 | Each node of a valid candidate is charged once per consuming query while `NotMaterialized`, once when materialized | `stage3_plan_selection_charges_each_node_per_consuming_query` | passes |
 | No candidate builds a Theta sketch | `stage3_plan_selection_has_no_theta_candidate` | passes |
-| A UnivMon estimate has a guarantee, so it is judged against its target instead of rejected for having none | `stage3_plan_selection_judges_univmon_with_an_accuracy_model` | ignored: needs a UnivMon entropy accuracy model; L2 and the distinct count have one |
+| A UnivMon estimate has a guarantee, so it is judged against its target instead of rejected for having none | `stage3_plan_selection_judges_univmon_with_an_accuracy_model` | ignored: needs a UnivMon entropy accuracy model (#788); L2 and the distinct count have one |
 | Given a distinct bound for the source, classic HLL's failure probability is bound and Q1's HLL is accepted; a larger bound enlarges its precision, also in a tumbling-pane chain | `stage3_plan_selection_accepts_hll_given_a_distinct_bound`, `…_enlarges_hll_for_a_larger_distinct_bound`, `…_enlarges_tumbling_pane_hll` | passes |
 | KMV certifies Q1 for any failure probability, without a distinct bound | `stage3_plan_selection_accepts_kmv_for_any_failure_probability_without_a_bound` | passes |
 | UnivMon's distinct count is certified given a distinct bound its layers cover; its L2 is certified from layer 0 alone, also through SQL Q3's exact population count and `CASE` | `stage3_plan_selection_accepts_univmon_distinct_given_a_distinct_bound`, `…_rejects_univmon_distinct_past_its_layers`, `…_accepts_univmon_l2` | passes |
-| One materialized UnivMon for all three costs no more than three separate UnivMons | `stage3_plan_selection_shared_univmon_costs_no_more_than_three` | ignored: needs a UnivMon entropy accuracy model and keeping the shared UnivMon materialized |
+| One materialized UnivMon for all three costs no more than three separate UnivMons | `stage3_plan_selection_shared_univmon_costs_no_more_than_three` | ignored: needs a UnivMon entropy accuracy model (#788) |
 | Selection direction: the selected plan has at most one UnivMon, because one shared UnivMon dominates several over the same input | `stage3_plan_selection_selected_plan_has_at_most_one_univmon` | passes |
 | The selected KMV plan's Q1 runs in the executor | `stage3_plan_selection_selected_kmv_plan_executes_end_to_end` | passes |
 
diff --git a/docs/design_docs/proposals/planner-layering-example3-acceptance.md b/docs/design_docs/proposals/planner-layering-example3-acceptance.md
index b0b30f810..acd85dc73 100644
--- a/docs/design_docs/proposals/planner-layering-example3-acceptance.md
+++ b/docs/design_docs/proposals/planner-layering-example3-acceptance.md
@@ -71,10 +71,10 @@ Shared data workload, except Pattern A's `arrival` is `Mixed`.
 |---|---|---|
 | The query lowers to scan → range 5m → quantile, no summary | `stage0_query_language_frontends_b_lowers_to_one_range_quantile` | passes |
 | Exact and KLL options | `stage1_logical_asap_optimization_b_pass1_offers_exact_and_kll` | passes |
-| Each summary option appears with no window, a 5-min sliding window with a 1-min slide, and 1-min tumbling windows | `stage1_logical_asap_optimization_b_window_composition_adds_sliding_and_tumbling_per_option` | ignored: sliding, no PR yet |
+| Each summary option appears with no window, a 5-min sliding window with a 1-min slide, and 1-min tumbling windows | `stage1_logical_asap_optimization_b_window_composition_adds_sliding_and_tumbling_per_option` | ignored: sliding (#522) |
 | Each summary option appears with no window and as 1-min tumbling panes, 5 merged before the estimate | `stage1_logical_asap_optimization_b_window_composition_adds_tumbling_per_mergeable_option` | passes |
 | Window parameters are legal: L divides W; s divides L and the evaluation interval; a tumbling length divides W and the interval | `stage1_logical_asap_optimization_b_window_parameters_are_legal` | passes |
-| Sliding with L = W has no merge; tumbling merges before the estimate | `stage1_logical_asap_optimization_b_merge_only_where_the_window_form_needs_it` | ignored: sliding, no PR yet |
+| Sliding with L = W has no merge; tumbling merges before the estimate | `stage1_logical_asap_optimization_b_merge_only_where_the_window_form_needs_it` | ignored: sliding (#522) |
 | A window form that merges uses a mergeable summary (KLL, DDSketch), in both patterns | `stage1_logical_asap_optimization_window_merges_use_mergeable_summaries` | passes |
 | One selected, cheapest valid, each node charged once per consuming query | `stage3_plan_selection_b_selects_cheapest_valid` | passes |
 | Panes rebuilt at query time cost the whole build plus the merge; whole-window KLL is selected | `stage3_plan_selection_b_panes_rebuilt_at_query_time_add_the_merge` | passes |
diff --git a/docs/design_docs/proposals/planner-layering-example4-acceptance.md b/docs/design_docs/proposals/planner-layering-example4-acceptance.md
index 5ecacc468..74d1cae97 100644
--- a/docs/design_docs/proposals/planner-layering-example4-acceptance.md
+++ b/docs/design_docs/proposals/planner-layering-example4-acceptance.md
@@ -78,9 +78,9 @@ it.
 
 | Invariant | Test | Status |
 |---|---|---|
-| Every node upstream of an ingestion-time node also runs at ingestion time (all variants) | `stage2_physical_asap_optimization_ingestion_time_upstream_is_ingestion_time` | passes: B1 and the monthly variant's maintained panes |
+| Every node upstream of an ingestion-time node also runs at ingestion time (all variants) | `stage2_physical_asap_optimization_ingestion_time_upstream_is_ingestion_time`; the monthly variant in `stage2_physical_asap_optimization_monthly_ingestion_time_upstream_is_ingestion_time` | passes: B1 and the monthly variant's maintained panes; the monthly test is ignored as slow (run in release by hand) |
 | With data at rest, no node runs at ingestion time | `stage2_physical_asap_optimization_at_rest_runs_nothing_at_ingestion_time` | passes |
-| A materialized output covers its consumers: retention ≥ max over its readers of `lookback` + offset | `stage2_physical_asap_optimization_materialized_output_covers_its_consumers` | passes: B1 keeps 6 min ≥ 5 min |
+| A materialized output covers its consumers: retention ≥ max over its readers of `lookback` + offset | `stage2_physical_asap_optimization_materialized_output_covers_its_consumers`; the monthly variant in `stage2_physical_asap_optimization_monthly_materialized_output_covers_its_consumers` | passes: B1 keeps 6 min ≥ 5 min; the monthly test is ignored as slow (run in release by hand) |
 
 ### Pattern A
 
@@ -101,9 +101,7 @@ it.
 | Until B3: the tumbling KLL candidate yields B1 and B2, both priced within the 200 ms latency bound (implementer test) | `stage2_physical_asap_optimization_b_tumbling_kll_has_b1_and_b2` | passes |
 | B1 builds at ingestion time; its merge and estimate run at query time | `stage2_physical_asap_optimization_b_b1_builds_at_ingestion_and_merges_at_query_time` | passes |
 | B3 runs nothing at ingestion time | `stage2_physical_asap_optimization_b_b3_keeps_query_time_windows` | passes |
-| The sliding KLL yields exactly ingestion time and query time, kept | `stage2_physical_asap_optimization_b_sliding_kll_has_two_materialization_options` | ignored: sliding windows (no PR yet) |
-| B2 costs at least B1 and B3 | `stage3_plan_selection_b_rebuilding_every_window_costs_most` | ignored: cost model 1 contradicts it. With query-time results retained, B3 keeps 4 panes for 8,884.5 cost/s and B1 6 for 13,326.5, while B2 rebuilds them for 2.183. Needs a rewrite relative to the cost model |
-| With the built-in models, B1 is the cheapest of the three | `stage3_plan_selection_b_prefers_ingestion_time_tumbling_windows` | ignored: cost model 1 contradicts it. B1 keeps 6 panes of 1M per-series KLLs (about 107 GB) for 13,326.5 cost/s; B2 rebuilds them for 2.183, and the whole-window KLL (P2, 2.100) is selected. Needs B3 and a rewrite relative to the cost model |
+| The sliding KLL yields exactly ingestion time and query time, kept | `stage2_physical_asap_optimization_b_sliding_kll_has_two_materialization_options` | ignored: sliding windows (#522) |
 | Each cost model chooses what its deployment pays for: cost model 1 the whole-window KLL; cost model 2 the exact plan (smallest peak memory); cost model 3 with lowest latency B1 in 256 GB, and the whole-window KLL in 64 GB, where B1 is invalid on peak memory | `stage3_plan_selection_b_cost_models_choose_differently` | passes |
 | Under cost model 3, the Pareto front over peak memory, disk and latency is {exact, whole-window KLL, B1} and B1 is selected; B2 is dominated by the whole-window KLL; least memory selects the exact plan | `stage3_plan_selection_b_pareto_front_over_memory_disk_and_latency` | passes |
 
diff --git a/docs/develop_docs/metrics-observability-corpora.md b/docs/develop_docs/metrics-observability-corpora.md
index f9a928529..6811f6477 100644
--- a/docs/develop_docs/metrics-observability-corpora.md
+++ b/docs/develop_docs/metrics-observability-corpora.md
@@ -43,7 +43,10 @@ o11y-bench, and awesome-prometheus-alerts. They are not duplicated here.
   PromQL.
 - `metrics_observability.rs` is the executable corpus test. It runs the
   PromQL parser/lowerer and Stage 1 (logical ASAP-aware optimization) on each
-  query, and prints the current measurements.
+  query, and prints the current measurements. Each corpus also has a coverage
+  ratchet: the test fails when fewer queries get an ASAP candidate, or more
+  fail Stage 1, than the counts committed in the test. Raise the ratchet when
+  coverage improves.
 - `summarize_misses.py` is an optional report formatter. It consumes the
   test's opt-in `POST_ASAP_MISS` lines and groups misses by coarse root shape.