From 906e08c7e1c0e6acc09d66773893cfcfd5d6eac5 Mon Sep 17 00:00:00 2001
From: zzylol <50204836+zzylol@users.noreply.github.com>
Date: Wed, 21 Oct 2026 19:19:11 +0000
Subject: [PATCH 1/3] test: drop duplicate and stale tests, keep slow variants
 out of CI, run the DAG viewer tests in CI

Delete tests that repeat stronger ones, assertions that cannot fail, the
unused cse::dag_node_count helper, and ignored tests whose premise no longer
holds. Move the monthly Example 4 variants and the exhaustive-limit check
out of the debug CI run, ratchet the metrics-observability corpus coverage,
point ignore reasons at open issues, rename post-ASAP-named test helpers to
the physical ASAP DAG terms, and run the DAG viewer's Python tests in CI.
---
 .github/workflows/rust.yml                    |   9 +
 crates/devtools/src/bin/stage_pipeline.rs     |   6 +
 crates/devtools/tests/stage_pipeline.rs       |  17 --
 crates/executor/tests/physical_dag.rs         |   4 +-
 crates/executor/tests/summary_projection.rs   |   4 +-
 .../executor/tests/weighted_topk_binding.rs   |   6 +-
 crates/frontend-promql/src/error.rs           |   1 -
 .../frontend-promql/tests/count_planning.rs   |   8 +-
 .../tests/maintained_population_horizon.rs    |   2 +-
 .../awesome_prometheus_alerts.rs              |  17 +-
 .../observability/metrics_observability.rs    |  43 ++++-
 .../tests/promql_conformance.rs               | 173 ------------------
 .../frontend-promql/tests/promql_lowering.rs  |  69 +------
 crates/frontend-promql/tests/support.rs       |   4 +-
 .../tests/univmon_candidates.rs               |   4 +-
 crates/integration-tests/src/lib.rs           |  12 +-
 .../tests/exact_composition.rs                |   4 +-
 .../tests/operator_design_examples.rs         |   2 -
 .../tests/operator_sharing.rs                 |  61 +-----
 .../tests/planner_layering_example1.rs        |  19 +-
 .../tests/planner_layering_example2.rs        |   6 +-
 .../tests/planner_layering_example3.rs        |   4 +-
 .../tests/planner_layering_example4.rs        | 114 ++++++------
 .../tests/promql_numeric_regressions.rs       |  14 +-
 .../tests/promql_to_post_asap.rs              |  10 +-
 .../tests/sql_to_post_asap.rs                 |  10 +-
 crates/types/src/ir/cse.rs                    |  55 +-----
 docs/design_docs/decisions/cse-cost-model.md  |   2 +-
 .../planner-layering-example2-acceptance.md   |   6 +-
 .../planner-layering-example3-acceptance.md   |   4 +-
 .../planner-layering-example4-acceptance.md   |   8 +-
 .../metrics-observability-corpora.md          |   5 +-
 32 files changed, 192 insertions(+), 511 deletions(-)

diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
index 26c9febe5..4a599e886 100644
--- a/.github/workflows/rust.yml
+++ b/.github/workflows/rust.yml
@@ -108,6 +108,15 @@ 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.
+    - 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
+
     # 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.
 

From b66d6d9723a4fdcb20b7950f29998b7507319f7e Mon Sep 17 00:00:00 2001
From: zzylol <50204836+zzylol@users.noreply.github.com>
Date: Wed, 21 Oct 2026 19:40:13 +0000
Subject: [PATCH 2/3] ci: run the DAG viewer's Node tests

---
 .github/workflows/rust.yml | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
index 4a599e886..1b96be58e 100644
--- a/.github/workflows/rust.yml
+++ b/.github/workflows/rust.yml
@@ -117,6 +117,9 @@ jobs:
         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: |

From 7483501b821a752a96467dca000f0b0b2ebd2655 Mon Sep 17 00:00:00 2001
From: zzylol <50204836+zzylol@users.noreply.github.com>
Date: Wed, 21 Oct 2026 19:44:06 +0000
Subject: [PATCH 3/3] ci: pin Python and Node for the DAG viewer tests

---
 .github/workflows/rust.yml | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
index 1b96be58e..8890f03eb 100644
--- a/.github/workflows/rust.yml
+++ b/.github/workflows/rust.yml
@@ -111,6 +111,18 @@ jobs:
     # 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: |