diff --git a/Cargo.lock b/Cargo.lock
index d4ecdd64..93db6508 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -340,6 +340,16 @@ dependencies = [
  "regex-syntax",
 ]
 
+[[package]]
+name = "asap-frontend-promql"
+version = "0.1.0"
+source = "git+https://github.com/ProjectASAP/ASAPController?rev=cc18c9872bbaf0cadf43566892c3974ec9948eba#cc18c9872bbaf0cadf43566892c3974ec9948eba"
+dependencies = [
+ "asap-ir",
+ "asap-l2",
+ "promql-parser 0.9.0 (git+https://github.com/ProjectASAP/promql-parser?branch=asap)",
+]
+
 [[package]]
 name = "asap-ir"
 version = "0.1.0"
@@ -791,6 +801,7 @@ name = "control_plane"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "asap-frontend-promql",
  "asap-ir",
  "asap-l2",
  "asap-plan",
@@ -803,7 +814,7 @@ dependencies = [
  "http-body-util",
  "parking_lot",
  "prometheus",
- "promql-parser 0.9.0",
+ "promql-parser 0.9.0 (git+https://github.com/ProjectASAP/promql-parser?rev=c51beafb361af4cc95ed62ae377862c660ceb757)",
  "prost",
  "prost-build",
  "reqwest 0.12.28",
@@ -2480,6 +2491,19 @@ dependencies = [
  "regex",
 ]
 
+[[package]]
+name = "promql-parser"
+version = "0.9.0"
+source = "git+https://github.com/ProjectASAP/promql-parser?branch=asap#c51beafb361af4cc95ed62ae377862c660ceb757"
+dependencies = [
+ "cfgrammar",
+ "chrono",
+ "lazy_static",
+ "lrlex",
+ "lrpar",
+ "regex",
+]
+
 [[package]]
 name = "promql-parser"
 version = "0.9.0"
diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml
index 62e9dbc9..d5e4e234 100644
--- a/control_plane/Cargo.toml
+++ b/control_plane/Cargo.toml
@@ -29,6 +29,19 @@ chrono      = { version = "0.4",  features = ["serde"] }
 # limit_ratio, a batch of experimental Prometheus functions) upstream
 # crates.io 0.8 doesn't have. Pinned to a commit, not the `asap` branch,
 # for the same reproducibility reason as the asap-ir pin below.
+#
+# Known duplication: `asap-frontend-promql` (below) references this same
+# repo via a floating `branch = "asap"` rather than `rev =`, so cargo
+# resolves two distinct package instances today (verified: `cargo tree -i
+# promql-parser` lists both). They happen to be the same commit right
+# now. Tried unifying via a workspace `[patch]` -- cargo rejects patching
+# a source with a different ref of itself ("patches must point to
+# different sources"), and no local checkout exists to patch to path=
+# instead (unlike the asap_sketchlib precedent in the root Cargo.toml).
+# Accepted as a known, low-severity tradeoff (extra compile time, a
+# latent divergence risk if either pin moves) rather than forcing a
+# fragile workaround -- revisit if `asap-frontend-promql` ever pins via
+# `rev` itself.
 promql-parser  = { git = "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/ProjectASAP/promql-parser", rev = "c51beafb361af4cc95ed62ae377862c660ceb757" }
 prost          = "0.13"
 bytes          = "1"
@@ -73,6 +86,16 @@ asap-l2 = { git = "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/ProjectASAP/ASAPController", rev = "cc18c9
 asap-sketch = { git = "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/ProjectASAP/ASAPController", rev = "cc18c9872bbaf0cadf43566892c3974ec9948eba" }
 asap-plan = { git = "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/ProjectASAP/ASAPController", rev = "cc18c9872bbaf0cadf43566892c3974ec9948eba" }
 
+# L1 adoption (design-target-architecture.md Part B): the PromQL front
+# end itself, replacing control_plane's own query_parser/promql.rs.
+# Pinned via `rev`, not ASAPController's own `branch = "asap"` reference
+# on its internal `promql-parser` dependency -- keeps this crate's own
+# `promql-parser` pin (above) as the single source of truth for that
+# transitive dependency's exact commit, avoiding two git-sourced copies
+# of the same crate diverging over time. Same rev as the other
+# ASAPController crates above -- no separate re-pin history yet.
+asap-frontend-promql = { git = "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/ProjectASAP/ASAPController", rev = "cc18c9872bbaf0cadf43566892c3974ec9948eba" }
+
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }
 tower = { version = "0.4", features = ["util"] }
diff --git a/control_plane/docs/design-target-architecture.md b/control_plane/docs/design-target-architecture.md
index 550fa1d5..26f15b16 100644
--- a/control_plane/docs/design-target-architecture.md
+++ b/control_plane/docs/design-target-architecture.md
@@ -257,22 +257,24 @@ deployment-specific detour from it.
 
 | Layer | Target | Current gap |
 |---|---|---|
-| L1 | `asap_frontend_promql::lower_promql` called directly; no local parser | `control_plane/src/query_parser/{mod,promql}.rs` (1793 lines) is still fully local — zero `asap-frontend-promql` dependency. **Full gap.** |
+| L1 | `asap_frontend_promql::lower_promql` called directly; no local parser | `query_parser::parse_query_expr_canonical`/`parse_query` call `lower_promql` directly; `query_parser/promql.rs` (the local parser, ~1187 lines) is deleted. No reconciliation pass — classification (e.g. bare selectors no longer implying `Aggregate{Sum}`) follows `asap-l2`'s lowering as-is. **Closed** (#428). |
 | L2 | `Binder::default()` / `convert_root` used via L1, no local schema logic | Already true in substance — `intent_algebra/{binder,column_resolution}.rs` are thin re-export shims. **Effectively closed.** |
 | L3 | Zero local `QueryExpr`/`AggIntent`/`Schema` definitions | Already true — `intent_algebra/{agg_intent,query_expr,relational,schema,expr_ir}.rs` are thin re-export shims with only genuinely-local residues (`Frequency` extension helpers, `PerPartitionWrap`, PromQL-ergonomic `LabelFilter`). `intent_algebra/lower.rs` (~1000 lines) remains real local code — deliberately, for two documented reasons with no ASAPController equivalent (multi-agg fusion, the windowed-Count-as-Frequency heuristic). **Effectively closed modulo `lower.rs`'s two documented exceptions.** |
 | L4 | One `CostModel` impl; `Rc<L4Node>` used directly | `sketch_algebra::cost_model::ControlPlaneCostModel` + `sketch_algebra::lower::bind_query_expr` (delegating to `implement_tree_in_with`) already match this shape. `sketch_algebra::matcher::SummaryFamilyMatcher` is the `Matcher` impl this section's serving-time §3 depends on. **Effectively closed** — `PhysicalExpr`/`L4Plan` is a thin, acceptable L5-placement wrapper around `Rc<L4Node>`, not a competing L4 algebra. |
 | L5 | Full local `PhysicalPlanner`/`TopologyDescriptor`/`StageAllocator` impl | `physical/colored_dag/*` + `emit/*` already implement this shape structurally, just not against the trait names above (no literal `PhysicalPlanner` trait exists in this repo — the free functions/structs are the de facto impl). Low-priority gap: naming/trait-alignment, not missing functionality. |
 | Serving | Single `SummaryExecutor` impl is the live path | `data_plane`'s `summary_executor.rs` implements the trait fully and is **now the default-on live path** (`ASAP_SUMMARY_EXECUTOR_LIVE` default flipped from off to on — the grouping-ambiguity blocker below is resolved via `Reduction`, and both unit + e2e tests already proved correctness for the covered shapes). `sketch_reducer.rs` remains the permanent fallback for shapes this executor self-excludes before binding (`rate()`/`irate()`, `topk(K, sum by(...)(rate(...)))`, keyed-CMS point-estimate) — **not** legacy debt pending deletion, an intentional, indefinite split. |
 
-**Net reading**: L2–L4 are substantially already at target — the earlier
-instinct that "`intent_algebra`/`sketch_algebra` should be unnecessary
-once connected to ASAPController" is correct and largely *already true*
-for L2–L4, not a still-open gap. The serving-time cutover is done for the
-shapes `SummaryExecutor` covers (default-on); the one real, still-open
-item is L1 (adopt `asap-frontend-promql`, retiring `query_parser/`
-outright). L5 should **not** shrink — it's this deployment's own,
-permanent responsibility per ASAPController's own "no `asap-physical`
-crate" status.
+**Net reading**: L1–L4 and the serving-time cutover are all now at
+target. The earlier instinct that "`intent_algebra`/`sketch_algebra`
+should be unnecessary once connected to ASAPController" is correct and
+largely *already true* for L2–L4; L1 has since closed the same way
+(#428), and the serving-time cutover is done for the shapes
+`SummaryExecutor` covers (default-on, #427). `sketch_reducer.rs` is not
+pending deletion — it's the permanent, intentional fallback for shapes
+`SummaryExecutor` self-excludes before binding. L5 should **not** shrink
+— it's this deployment's own, permanent responsibility per
+ASAPController's own "no `asap-physical` crate" status; its only
+remaining gap is the low-priority naming/trait-alignment noted above.
 
 ## 5. Open questions (carried from `data_plane/docs/l4node-plan-executor-design.md`, mostly resolved)
 
diff --git a/control_plane/src/asap_tier_analysis.rs b/control_plane/src/asap_tier_analysis.rs
index 9bec25b2..9a17bab2 100644
--- a/control_plane/src/asap_tier_analysis.rs
+++ b/control_plane/src/asap_tier_analysis.rs
@@ -52,11 +52,22 @@ use promql_parser::parser::{self, Expr, VectorSelector};
 use crate::intent_algebra::agg_intent::AggIntent;
 use crate::intent_algebra::query_expr::QueryExpr;
 use crate::query_parser::{parse_query_expr_canonical, parsed_query_from_canonical};
+use crate::types_v2::AccuracyTarget;
 
 pub use crate::sketch_algebra::capability::{
     capability_for, Capability, OuterAgg, OuterFn, SketchKindHandle,
 };
 
+/// Fixed accuracy target for warm-tier shape analysis (L1 adoption,
+/// design-target-architecture.md Part B) -- this walk only inspects
+/// `AggIntent` kinds, not their accuracy; the converter needs SOME
+/// non-exact epsilon to pin sketch-eligible intents at, but the real
+/// per-query accuracy bound comes from `QueryWorkload` further
+/// downstream (see this module's own doc). `0.01` matches the same
+/// deployment-wide default `data_plane`'s serving-time code uses
+/// (`live_serve::LIVE_ACCURACY`, `shadow_compare::SHADOW_ACCURACY`).
+const WARM_TIER_ANALYSIS_ACCURACY: AccuracyTarget = AccuracyTarget::Epsilon(0.01);
+
 // ── Public types ─────────────────────────────────────────────────────────────
 
 /// One sub-expression of the input PromQL that CAN be served from the
@@ -188,7 +199,7 @@ pub fn analyze_promql_for_asap_tier(metricsql: &str) -> ASAPTierAnalysis {
     // sketch-eligible intents at a non-exact epsilon, which is all
     // warm-tier analysis needs (the real per-query accuracy bound comes
     // from QueryWorkload further downstream).
-    let expr = match parse_query_expr_canonical(metricsql) {
+    let expr = match parse_query_expr_canonical(metricsql, WARM_TIER_ANALYSIS_ACCURACY) {
         Ok(e) => e,
         Err(e) => {
             return ASAPTierAnalysis {
@@ -230,14 +241,29 @@ pub fn analyze_promql_for_asap_tier(metricsql: &str) -> ASAPTierAnalysis {
         };
     }
 
-    // Step 4: for each intent, look up its capability. The first
-    // intent that returns `None` aborts the analysis — the warm
-    // tier can't answer this query (the router falls over to archive).
+    // Step 4: for each intent, look up its capability.
+    //
+    // L1 adoption (design-target-architecture.md Part B) resilience fix:
+    // an unsupported intent used to abort the WHOLE analysis immediately,
+    // discarding any candidates already collected from other Aggregate
+    // nodes in the tree. That was fine when the (retired) local parser
+    // fused compositions into a single intent per query, but
+    // `lower_promql` genuinely produces multiple `Aggregate` nodes for
+    // compositions like `avg by (zone) (quantile_over_time(...))` — an
+    // outer `Avg` (no capability mapping) wrapping an inner `Quantile`
+    // (sketchable). Aborting on the outer miss would throw away the
+    // inner candidate the warm tier CAN answer. Skip an unsupported
+    // intent and keep collecting instead; only report `unsupported` if
+    // NOTHING in the whole tree was answerable. `data_plane`'s
+    // `engine.rs` already loops over every returned candidate (not just
+    // index 0), so a partial list from a composed query is an
+    // already-supported case, not a new invariant.
     let metric_name = parsed.metric_name.clone();
     let group_by_keys: BTreeSet<String> = parsed.group_by_labels.iter().cloned().collect();
     let spatial_filter_canonical = render_spatial_filter(&parsed.label_filters);
 
     let mut out = ASAPTierAnalysis::default();
+    let mut last_unsupported: Option<UnsupportedReason> = None;
     for intent in &intents {
         match capability_for(intent) {
             Some(cap) => {
@@ -271,13 +297,15 @@ pub fn analyze_promql_for_asap_tier(metricsql: &str) -> ASAPTierAnalysis {
                 });
             }
             None => {
-                out.unsupported = Some(UnsupportedReason::UnsupportedAggIntent(
+                last_unsupported = Some(UnsupportedReason::UnsupportedAggIntent(
                     intent_kind_label(intent).to_string(),
                 ));
-                return out;
             }
         }
     }
+    if out.candidates.is_empty() {
+        out.unsupported = last_unsupported;
+    }
     out
 }
 
@@ -870,11 +898,11 @@ mod tests {
             "increase(errors_total[2m])",
         ];
         for q in queries {
-            let canonical = parse_query_expr_canonical(q)
+            let canonical = parse_query_expr_canonical(q, WARM_TIER_ANALYSIS_ACCURACY)
                 .unwrap_or_else(|e| panic!("canonical parse failed for {q:?}: {e}"));
             let derived = parsed_query_from_canonical(&canonical);
-            let direct =
-                parse_query(q).unwrap_or_else(|e| panic!("parse_query failed for {q:?}: {e}"));
+            let direct = parse_query(q, WARM_TIER_ANALYSIS_ACCURACY)
+                .unwrap_or_else(|e| panic!("parse_query failed for {q:?}: {e}"));
 
             assert_eq!(
                 derived.metric_name, direct.metric_name,
@@ -995,28 +1023,65 @@ mod tests {
     fn analyze_quantile_over_time_with_sum_by_group_keys() {
         // PromQL `sum by (host) (quantile_over_time(...))` populates
         // group_by_keys with `host`.
+        //
+        // L1 adoption (design-target-architecture.md Part B): `lower_promql`
+        // doesn't fuse "outer aggregation wraps an inner range function"
+        // into one shape the way the retired local parser did -- this
+        // composition is genuinely two operations (sum the per-series
+        // quantiles, grouped by host), so it lowers to two `Aggregate`
+        // nodes and `analyze_promql_for_asap_tier` (which walks every
+        // `Aggregate` node in the tree) now reports two candidates: the
+        // outer `ExactAgg(Sum)` and the inner `QuantileApprox`. Both carry
+        // `group_by_keys: {host}` (the trace-derived group-by is shared
+        // context across all candidates from one analysis call, not
+        // per-node). `data_plane`'s `engine.rs` already loops over every
+        // candidate (not just index 0), so multi-candidate composed
+        // shapes are an already-supported case, not a new invariant.
         let a = analyze_promql_for_asap_tier(
             "sum by (host) (quantile_over_time(0.99, http_latency_ms[5m]))",
         );
         assert!(a.unsupported.is_none(), "{a:?}");
-        assert_eq!(a.candidates.len(), 1);
-        assert_eq!(a.candidates[0].group_by_keys, keys(&["host"]));
+        assert_eq!(a.candidates.len(), 2, "{a:?}");
+        for c in &a.candidates {
+            assert_eq!(c.group_by_keys, keys(&["host"]), "{a:?}");
+        }
+        assert!(
+            a.candidates
+                .iter()
+                .any(|c| c.required_capability == Capability::QuantileApprox(SketchKindHandle::Any)),
+            "{a:?}"
+        );
     }
 
     #[test]
-    fn analyze_histogram_quantile_is_rejected() {
-        // `histogram_quantile(...)` is a PromQL/MetricsQL language-level
-        // operator, NOT an L3 intent. Per Step γ5, the PromQL parser
-        // substitutes it into a plain `Aggregate { Quantile(φ) }` so
-        // downstream sees the canonical Quantile intent. The inner argument
-        // shape requires a `rate(bucket[r])` which the analyzer rejects as
-        // an exact-counter intent, so the analyzer returns SOME unsupported
-        // reason; the bucket-aware physical reduction is not yet wired into
-        // the ASAP-tier path.
+    fn analyze_histogram_quantile_partially_answerable_via_inner_composition() {
+        // Renamed (was `..._is_rejected`): L1 adoption
+        // (design-target-architecture.md Part B), accepted behavior
+        // change + the resilience fix above compounding usefully here.
+        //
+        // The retired local parser unconditionally substituted
+        // `histogram_quantile(...)` into a plain sketchable `Quantile`
+        // intent (Step γ5). `lower_promql` instead correctly detects this
+        // is the classic `_bucket` + `by (le)` shape and produces the
+        // real, exact-only `AggIntent::HistogramQuantile` -- which
+        // `capability_for` has no mapping for. Under the old
+        // abort-on-first-miss loop this would have rejected the whole
+        // query; the resilience fix means that outer miss is skipped and
+        // the inner `sum(rate(...)) by (le)` composition's two real
+        // candidates (`ExactAgg(Sum)`, `ExactAgg(Increase)`) still
+        // surface -- a partial answer where the old fused behavior gave
+        // a full one, but not a full rejection either.
         let a = analyze_promql_for_asap_tier(
             "histogram_quantile(0.99, sum(rate(http_latency_bucket[5m])) by (le))",
         );
-        assert!(a.unsupported.is_some(), "{a:?}");
+        assert!(a.unsupported.is_none(), "{a:?}");
+        assert_eq!(a.candidates.len(), 2, "{a:?}");
+        assert!(
+            a.candidates
+                .iter()
+                .any(|c| c.required_capability == Capability::ExactAgg(AggregationType::Increase)),
+            "{a:?}"
+        );
     }
 
     #[test]
@@ -1046,15 +1111,17 @@ mod tests {
     // the `ExactAgg` capability and routing everything to archive.)
 
     #[test]
-    fn bare_vector_selector_binds_to_exact_agg() {
-        // The PromQL parser models a bare selector as `Aggregate { Sum }`
-        // over the sample value; `Sum` carries an `ExactAgg` capability.
+    fn bare_vector_selector_is_no_longer_asap_tier_answerable() {
+        // Superseded by `bare_selector_is_no_longer_asap_tier_answerable`
+        // below -- see that test's comment (L1 adoption,
+        // design-target-architecture.md Part B, accepted behavior
+        // change): `lower_promql` doesn't implicitly wrap a bare selector
+        // in `Aggregate { Sum }` the way the retired local parser did.
         let a = analyze_promql_for_asap_tier("http_requests_total{zone=\"z0\"}");
-        assert!(a.unsupported.is_none(), "{a:?}");
-        assert_eq!(a.candidates.len(), 1);
-        assert_eq!(
-            a.candidates[0].required_capability,
-            Capability::ExactAgg(AggregationType::Sum)
+        assert!(a.candidates.is_empty(), "{a:?}");
+        assert!(
+            matches!(a.unsupported, Some(UnsupportedReason::NoCallNodeFound)),
+            "{a:?}"
         );
     }
 
@@ -1182,10 +1249,22 @@ mod tests {
     }
 
     #[test]
-    fn bare_selector_candidate_carries_outer_fn_plain() {
+    fn bare_selector_is_no_longer_asap_tier_answerable() {
+        // L1 adoption (design-target-architecture.md Part B), accepted
+        // behavior change: the retired local parser wrapped a bare
+        // selector in an implicit `Aggregate { Sum }` specifically so the
+        // warm tier could still answer it via `ExactAgg(Sum)`.
+        // `lower_promql` doesn't add that wrapper -- a bare selector
+        // stays a plain `Scan` with zero `Aggregate` nodes, so this now
+        // returns `NoCallNodeFound` / no candidates and falls through to
+        // archive. Accepted as-is rather than reintroducing the implicit
+        // wrap locally.
         let a = analyze_promql_for_asap_tier("http_requests_total{zone=\"z0\"}");
-        assert!(a.unsupported.is_none(), "{a:?}");
-        assert_eq!(a.candidates[0].outer_fn, OuterFn::Plain, "{a:?}");
+        assert!(a.candidates.is_empty(), "{a:?}");
+        assert!(
+            matches!(a.unsupported, Some(UnsupportedReason::NoCallNodeFound)),
+            "{a:?}"
+        );
     }
 
     #[test]
@@ -1197,16 +1276,25 @@ mod tests {
         // the case that motivated the original `query_contains_rate_call`
         // walker — now satisfied by walking the AST once in the
         // analyzer and emitting the typed `OuterFn::Rate` flag.
+        //
+        // L1 adoption (design-target-architecture.md Part B): this is
+        // now two real `Aggregate` nodes (outer sum-by-zone, inner rate),
+        // so `analyze_promql_for_asap_tier` reports two candidates --
+        // `ExactAgg(Sum)` (the outer reduction) and `ExactAgg(Increase)`
+        // (the inner rate) -- both sharing the trace-derived `outer_fn:
+        // Rate` and `range_seconds: 300`. Find the `Increase` one rather
+        // than assuming index 0.
         let a = analyze_promql_for_asap_tier("sum by (zone) (rate(http_requests_total[5m]))");
         assert!(a.unsupported.is_none(), "{a:?}");
-        assert_eq!(
-            a.candidates[0].required_capability,
-            Capability::ExactAgg(AggregationType::Increase),
-            "{a:?}"
-        );
-        assert_eq!(a.candidates[0].outer_fn, OuterFn::Rate, "{a:?}");
+        assert_eq!(a.candidates.len(), 2, "{a:?}");
+        let inc = a
+            .candidates
+            .iter()
+            .find(|c| c.required_capability == Capability::ExactAgg(AggregationType::Increase))
+            .unwrap_or_else(|| panic!("expected an ExactAgg(Increase) candidate: {a:?}"));
+        assert_eq!(inc.outer_fn, OuterFn::Rate, "{a:?}");
         // Range is lifted from the inner rate's matrix selector.
-        assert_eq!(a.candidates[0].range_seconds, 300, "{a:?}");
+        assert_eq!(inc.range_seconds, 300, "{a:?}");
     }
 
     #[test]
@@ -1273,20 +1361,21 @@ mod tests {
 
     #[test]
     fn max_by_quantile_over_time_carries_outer_agg_max() {
+        // L1 adoption (design-target-architecture.md Part B): two real
+        // `Aggregate` nodes now, not one fused shape -- the outer `max`
+        // itself is a real `AggIntent::Max` (`ExactAgg(MinMax)`), plus the
+        // inner `Quantile`. Both candidates share the trace-derived
+        // `outer_agg: Max([zone])` (shared context per analysis call).
         let a = analyze_promql_for_asap_tier(
             "max by (zone) (quantile_over_time(0.99, http_latency_ms[5m]))",
         );
         assert!(a.unsupported.is_none(), "{a:?}");
-        assert_eq!(a.candidates.len(), 1);
-        let c = &a.candidates[0];
-        // Inner function still binds to QuantileApprox — outer_agg
-        // doesn't alter the candidate's required_capability (the
-        // engine's fold runs over the inner result).
-        assert_eq!(
-            c.required_capability,
-            Capability::QuantileApprox(SketchKindHandle::Any),
-            "{c:?}"
-        );
+        assert_eq!(a.candidates.len(), 2, "{a:?}");
+        let c = a
+            .candidates
+            .iter()
+            .find(|c| c.required_capability == Capability::QuantileApprox(SketchKindHandle::Any))
+            .unwrap_or_else(|| panic!("expected a QuantileApprox candidate: {a:?}"));
         // OuterAgg captured.
         match &c.outer_agg {
             OuterAgg::Max(labels) => {
@@ -1298,10 +1387,24 @@ mod tests {
 
     #[test]
     fn avg_by_quantile_over_time_carries_outer_agg_avg() {
+        // L1 adoption (design-target-architecture.md Part B): the outer
+        // `avg` is now a real `AggIntent::Avg`, which `capability_for`
+        // does NOT map to any capability (`Avg` has never been
+        // supported -- see `capability_for`'s own tests). Per the
+        // resilience fix above, that unsupported intent is skipped
+        // rather than aborting the whole analysis, so only the inner
+        // `Quantile` candidate survives -- still carrying the
+        // trace-derived `outer_agg: Avg([zone])`.
         let a = analyze_promql_for_asap_tier(
             "avg by (zone) (quantile_over_time(0.99, http_latency_ms[5m]))",
         );
         assert!(a.unsupported.is_none(), "{a:?}");
+        assert_eq!(a.candidates.len(), 1, "{a:?}");
+        assert_eq!(
+            a.candidates[0].required_capability,
+            Capability::QuantileApprox(SketchKindHandle::Any),
+            "{a:?}"
+        );
         match &a.candidates[0].outer_agg {
             OuterAgg::Avg(labels) => assert_eq!(labels, &vec!["zone".to_string()]),
             other => panic!("expected OuterAgg::Avg([zone]), got {other:?}"),
@@ -1412,11 +1515,14 @@ mod tests {
     }
 
     #[test]
-    fn is_asap_tier_answerable_true_for_bare_selector() {
-        // A bare selector lowers to `Aggregate { Sum }`, which carries an
-        // `ExactAgg` capability — so it is ASAP-tier-answerable.
+    fn is_asap_tier_answerable_false_for_bare_selector() {
+        // Renamed + inverted (was `..._true_..`): L1 adoption
+        // (design-target-architecture.md Part B, accepted behavior
+        // change) -- see `bare_selector_is_no_longer_asap_tier_answerable`'s
+        // comment. A bare selector no longer lowers to an implicit
+        // `Aggregate { Sum }`, so it is NOT ASAP-tier-answerable anymore.
         let a = analyze_promql_for_asap_tier("m{zone=\"z0\"}");
-        assert!(a.is_asap_tier_answerable());
+        assert!(!a.is_asap_tier_answerable());
     }
 
     // ── Cardinality / count_over_time real-PromQL acceptance ────────────
diff --git a/control_plane/src/asap_tier_implement.rs b/control_plane/src/asap_tier_implement.rs
index fe7ac7f4..08ab54f1 100644
--- a/control_plane/src/asap_tier_implement.rs
+++ b/control_plane/src/asap_tier_implement.rs
@@ -78,6 +78,14 @@ use asap_sketch::L4Node;
 
 use crate::intent_algebra::query_expr::{BindingScope, QueryExpr};
 use crate::query_parser::parse_query_expr_canonical;
+use crate::types_v2::AccuracyTarget;
+
+/// Fixed accuracy target for this L1 call site (L1 adoption,
+/// design-target-architecture.md Part B) -- matches
+/// `asap_tier_analysis::WARM_TIER_ANALYSIS_ACCURACY`; this module has no
+/// per-query accuracy bound available either (see its own module doc:
+/// "not yet a drop-in replacement for `analyze_promql_for_asap_tier`").
+const IMPLEMENT_PROMQL_ACCURACY: AccuracyTarget = AccuracyTarget::Epsilon(0.01);
 
 /// Find every independently-realizable `Aggregate` subtree in `expr`.
 ///
@@ -181,7 +189,7 @@ pub enum ImplementPromqlError {
 pub fn implement_promql_for_asap_tier(
     metricsql: &str,
 ) -> Result<Vec<Rc<L4Node>>, ImplementPromqlError> {
-    let expr = parse_query_expr_canonical(metricsql)
+    let expr = parse_query_expr_canonical(metricsql, IMPLEMENT_PROMQL_ACCURACY)
         .map_err(|e| ImplementPromqlError::UnparseableMetricsql(e.to_string()))?;
 
     let mut roots: Vec<&QueryExpr> = Vec::new();
@@ -208,17 +216,16 @@ mod tests {
     }
 
     #[test]
-    fn bare_selector_implements_to_an_exact_sum_agg() {
-        // The PromQL parser models a bare selector as `Aggregate { Sum }`
-        // over the sample value -- same shape as
-        // asap_tier_analysis::bare_vector_selector_binds_to_exact_agg on
-        // the flat path (Capability::ExactAgg(Sum)), NOT the "no call
-        // node found" case (that's for something parse-failure-adjacent,
-        // not a bare selector).
+    fn bare_selector_has_no_aggregate_root_to_implement() {
+        // L1 adoption (design-target-architecture.md Part B), accepted
+        // behavior change -- see
+        // asap_tier_analysis::bare_selector_is_no_longer_asap_tier_answerable's
+        // comment: `lower_promql` doesn't implicitly wrap a bare selector
+        // in `Aggregate { Sum }` the way the retired local parser did, so
+        // there's no `Aggregate` node here at all to find a root at.
         let roots =
             implement_promql_for_asap_tier("http_requests_total").expect("parses and implements");
-        assert_eq!(roots.len(), 1);
-        assert!(matches!(roots[0].expr, SummaryExpr::SummaryAgg { .. }));
+        assert!(roots.is_empty(), "{roots:?}");
     }
 
     #[test]
@@ -321,17 +328,22 @@ mod tests {
     /// behavior shift.
     #[test]
     fn implement_frequency_as_agg_test() {
+        // Per this test's own prior instructions: the gap it used to
+        // document (under-realizing to `Logical` because `asap-plan` had
+        // no `Extension`/`Frequency` opinion) is now closed -- not via an
+        // `Extension` hook, but because L1 adoption
+        // (design-target-architecture.md Part B) makes `count_over_time`
+        // lower directly to `AggIntent::Count { accuracy: Epsilon(...) }`
+        // (a real, first-class, non-exact intent) rather than needing
+        // this deployment's `Frequency` extension wrapper at all --
+        // `asap-plan` realizes a non-exact `Count` as a real CMS-backed
+        // `SummaryAgg` + `SummaryEstimate` on its own.
         let roots = implement_promql_for_asap_tier("count_over_time(http_requests_total[5m])")
             .expect("parses and implements");
         assert_eq!(roots.len(), 1);
         assert!(
-            matches!(roots[0].expr, SummaryExpr::Logical(_)),
-            "EXPECTED (for now): asap-plan has no Extension/Frequency \
-             opinion, so this under-realizes to Logical -- see module doc. \
-             If this assertion starts failing because it's now realized as \
-             a SummaryAgg/SummaryEstimate, the gap has been closed \
-             upstream or by a local Extension hook -- update this test to \
-             assert the new, better behavior instead of reverting it. Got: {:?}",
+            matches!(roots[0].expr, SummaryExpr::SummaryEstimate { .. }),
+            "expected a realized SummaryEstimate, got: {:?}",
             roots[0].expr,
         );
     }
diff --git a/control_plane/src/lib.rs b/control_plane/src/lib.rs
index a00ceffa..f64cfd6b 100644
--- a/control_plane/src/lib.rs
+++ b/control_plane/src/lib.rs
@@ -43,8 +43,9 @@
 //! - `intent_algebra` — `AggIntent` + `QueryExpr` DAG (canonical L3 IR;
 //!   `relational` carries the L2 relational IR the parsers emit and
 //!   `lower` lowers it to the canonical L3 types).
-//! - `query_parser` — front-end parsers (Layer 1): `promql.rs` / `sql.rs`
-//!   emit the L2 relational `relational::QueryExpr` tree.
+//! - `query_parser` — L1 entry point: `parse_query_expr_canonical`/`parse_query`
+//!   call `asap_frontend_promql::lower_promql` directly (no local parser
+//!   since design-target-architecture.md Part B; SQL not yet adopted).
 //! - `physical` — L5 framework (allocator, planner, plan, sketch_catalog,
 //!   colored_dag, stage_split, topology).
 //! - `optimizer` — L4 rule engine + cost model traits/impls + baseline
diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs
index 1f34b256..e328b448 100644
--- a/control_plane/src/main.rs
+++ b/control_plane/src/main.rs
@@ -578,6 +578,14 @@ async fn main() {
 async fn handle_plan(State(st): State<AppState>, Json(spec): Json<QuerySpec>) -> impl IntoResponse {
     let wc = spec.workload.clone();
     let query_string = spec.query_string.clone();
+    // Captured before `spec` moves into `analyze` below -- L1 adoption
+    // (design-target-architecture.md Part B) needs a real AccuracyTarget
+    // for `parse_query_expr_canonical`; this is the one call site in the
+    // pipeline with an actual per-query accuracy value available, so
+    // thread it through rather than a flat deployment-wide default.
+    let accuracy = control_plane::types_v2::accuracy_target_from_legacy_accuracy_sla(
+        spec.accuracy_sla,
+    );
     let workload = match st.analyzer.analyze(spec) {
         Ok(w) => w,
         Err(e) => return (StatusCode::UNPROCESSABLE_ENTITY, e.to_string()).into_response(),
@@ -608,7 +616,7 @@ async fn handle_plan(State(st): State<AppState>, Json(spec): Json<QuerySpec>) ->
         let mut bound_physical: Option<control_plane::sketch_algebra::PhysicalExpr> = None;
         let mut plan_summary = None;
         if let Some(ref qs) = query_string {
-            match parse_query_expr_canonical(qs) {
+            match parse_query_expr_canonical(qs, accuracy) {
                 Err(e) => {
                     warn!(query = %qs, error = %e, "parse_query_expr_canonical failed; skipping algebra pipeline")
                 }
diff --git a/control_plane/src/pipeline.rs b/control_plane/src/pipeline.rs
index abeb2248..760d2c84 100644
--- a/control_plane/src/pipeline.rs
+++ b/control_plane/src/pipeline.rs
@@ -168,10 +168,19 @@ impl Analyzer {
         };
 
         // ── Step 1: parse query_string if provided ─────────────────────────
+        // Real `spec.accuracy` when the caller supplied one (L1 adoption,
+        // design-target-architecture.md Part B needs an explicit
+        // AccuracyTarget); the deployment-wide `Epsilon(0.01)` default
+        // otherwise, matching this same fallback's use elsewhere.
         let parsed = spec
             .query_string
             .as_deref()
-            .map(|q| query_parser::parse_query(q))
+            .map(|q| {
+                query_parser::parse_query(
+                    q,
+                    spec.accuracy.clone().unwrap_or(AccuracyTarget::Epsilon(0.01)),
+                )
+            })
             .transpose()
             .with_context(|| "failed to parse query_string")?;
 
@@ -543,6 +552,14 @@ mod tests {
     /// time_window, and quantiles — no explicit fields required.
     #[test]
     fn query_string_promql_populates_workload() {
+        // L1 adoption (design-target-architecture.md Part B), accepted
+        // behavior change: `sum by (host) (quantile_over_time(...))` no
+        // longer fuses into one shape -- it's genuinely two operations
+        // (sum the per-series quantiles, grouped by host), so the outer
+        // `Sum` now also contributes to this flat summary and flips
+        // `exact_required` (an outer exact fold over sketch-derived
+        // quantile values is real complexity the old fused behavior
+        // papered over, not something a sketch alone answers).
         let w = Analyzer::new()
             .analyze(qs_only(
                 "sum by (host) (quantile_over_time(0.99, latency[5m]))",
@@ -552,7 +569,7 @@ mod tests {
         assert_eq!(w.aggregations, vec![AggType::Quantile]);
         assert_eq!(w.time_window, Duration::from_secs(300));
         assert_eq!(w.quantiles, vec![0.99]);
-        assert!(!w.exact_required);
+        assert!(w.exact_required);
     }
 
     /// Explicit metric_name overrides the name derived from query_string.
diff --git a/control_plane/src/query_parser/mod.rs b/control_plane/src/query_parser/mod.rs
index d58c4cda..ceda82a0 100644
--- a/control_plane/src/query_parser/mod.rs
+++ b/control_plane/src/query_parser/mod.rs
@@ -1,5 +1,19 @@
 //! SP-1 query workload extraction — PromQL parser.
 //!
+//! L1 adoption (`control_plane/docs/design-target-architecture.md` Part
+//! B): the PromQL parsing + L1→L2→L3 lowering previously done by this
+//! crate's own `promql.rs` (retired) is now `asap_frontend_promql::lower_promql`
+//! directly — no local parser, no local L2 relational tree. This crate's
+//! own `intent_algebra::lower.rs` two heuristics (multi-agg fusion, the
+//! windowed-Count-as-Frequency trigger) do **not** run anymore; per
+//! explicit direction, this adopts whatever `AggIntent` classification
+//! ASAPController's `asap_l2::lower` produces as-is (e.g. a classic
+//! `by (le)` `histogram_quantile(...)` now correctly classifies as the
+//! exact `AggIntent::HistogramQuantile`, not the sketchable `Quantile`
+//! this deployment previously forced; grouped/windowed `count_over_time`
+//! becomes plain exact `Count`, not the `Frequency` extension) rather
+//! than reconciling it back to the old local behavior.
+//!
 //! # Entry points
 //!
 //! | Function | Returns | Use |
@@ -7,17 +21,9 @@
 //! | [`parse_query_expr_canonical`] | canonical `query_expr::QueryExpr` | Full algebra IR |
 //! | [`parse_query`] | `ParsedQuery` | Backward compat with existing analyzer |
 //!
-//! # Supported PromQL patterns (via `promql-parser` AST)
-//! - `quantile_over_time(φ, m{f}[w]) by (dims)`
-//! - `histogram_quantile(φ, rate(m{f}[w])) by (le)`
-//! - `avg/min/max/stddev/stdvar_over_time(m{f}[w]) by (dims)`
-//! - `sum/count_over_time(m{f}[w]) by (dims)`
-//! - `topk(k, *_over_time(…) by (dims))`
-//! - `count(*_over_time(…) by (dims))` — cardinality
-//! - `changes/resets(m{f}[w])`
-//! - Bare metric selector / binary op → `exact_required`
-
-pub mod promql;
+//! Both now take an explicit [`AccuracyTarget`] — `lower_promql` requires
+//! one (accuracy-driven parameter sizing happens as early as L1/L2 for
+//! some intents), where the old local pipeline took none.
 
 use std::collections::HashMap;
 use std::time::Duration;
@@ -26,6 +32,7 @@ use crate::intent_algebra::agg_intent::AggIntent;
 use crate::intent_algebra::query_expr::{Predicate, QueryExpr, Source};
 use crate::intent_algebra::{ColumnId, CompareOp, L3Expr, L3Scalar};
 use crate::types::AggType;
+use crate::types_v2::AccuracyTarget;
 
 // ── Output types (legacy — consumed by analyzer and planner) ──────────────────
 
@@ -77,40 +84,20 @@ pub enum QueryHint {
 
 // ── Public entry points ───────────────────────────────────────────────────────
 
-/// Parse a PromQL query string into the **legacy Layer-2**
-/// [`relational::QueryExpr`](crate::intent_algebra::relational::QueryExpr) IR.
-///
-/// The parser emits Layer-2 relational operators (`Aggregate { AggFunc }`,
-/// `Window`, `Filter`, …). The Layer-2 → Layer-3 sketch lowering
-/// and the conversion to the canonical IR both live inside
-/// [`intent_algebra::convert_root`](crate::intent_algebra::convert_root) —
-/// this function is just the parse front door.
-///
-/// Internal to the crate: the only caller is
-/// [`parse_query_expr_canonical`], which is the public canonical-IR entry.
-pub(crate) fn parse_query_expr(
-    query: &str,
-) -> anyhow::Result<crate::intent_algebra::relational::QueryExpr> {
-    promql::parse_promql_expr(query.trim())
-}
-
 /// Parse a PromQL query string into the **canonical** L3
 /// [`query_expr::QueryExpr`](crate::intent_algebra::query_expr::QueryExpr) IR.
 ///
-/// This is the single public algebra-IR entry point. It parses the query
-/// into the crate-internal legacy Layer-2 tree via [`parse_query_expr`],
-/// then runs that through
-/// [`intent_algebra::convert_root`](crate::intent_algebra::convert_root),
-/// which folds the Layer-2 → Layer-3 sketch lowering and the
-/// legacy → canonical conversion into one entry. The legacy IR is never
-/// observable to callers.
+/// This is the single public algebra-IR entry point — a direct call into
+/// `asap_frontend_promql::lower_promql`, which does the full L1 parse →
+/// L2 relational tree → L3 canonical conversion in one call. No local
+/// parser, no local L2 tree; `accuracy` is threaded onto every
+/// accuracy-bearing intent the same way ASAPController's own PromQL
+/// front end threads it.
 pub fn parse_query_expr_canonical(
     query: &str,
+    accuracy: AccuracyTarget,
 ) -> anyhow::Result<crate::intent_algebra::query_expr::QueryExpr> {
-    let legacy = parse_query_expr(query)?;
-    // `ConvertError` derives `thiserror::Error`, so `?` lifts it straight
-    // into `anyhow::Error`.
-    let canonical = crate::intent_algebra::convert_root(&legacy)?;
+    let canonical = asap_frontend_promql::lower_promql(query.trim(), accuracy)?;
     Ok(canonical)
 }
 
@@ -120,8 +107,8 @@ pub fn parse_query_expr_canonical(
 /// [`crate::analyzer::Analyzer`].  Internally it parses via
 /// [`parse_query_expr_canonical`] and extracts the flat summary by walking
 /// the canonical [`QueryExpr`] tree.
-pub fn parse_query(query: &str) -> anyhow::Result<ParsedQuery> {
-    let qe = parse_query_expr_canonical(query)?;
+pub fn parse_query(query: &str, accuracy: AccuracyTarget) -> anyhow::Result<ParsedQuery> {
+    let qe = parse_query_expr_canonical(query, accuracy)?;
     Ok(qe_to_parsed_query(&qe))
 }
 
@@ -165,6 +152,11 @@ fn root_scan_schema(qe: &QueryExpr) -> Option<&crate::intent_algebra::Schema> {
         QueryExpr::Scan { schema, .. } => Some(schema),
         QueryExpr::Filter { child, .. }
         | QueryExpr::Window { child, .. }
+        // `TimeRange`/`TimeShift` are `asap_l2::lower`'s range-vector-selector
+        // and offset/@ markers (L1 adoption, design-target-architecture.md
+        // Part B) -- pass-through wrappers over the same `Scan`.
+        | QueryExpr::TimeRange { child, .. }
+        | QueryExpr::TimeShift { child, .. }
         | QueryExpr::Aggregate { child, .. }
         | QueryExpr::Distinct { child, .. }
         | QueryExpr::Project { child, .. }
@@ -183,9 +175,13 @@ fn root_scan_schema(qe: &QueryExpr) -> Option<&crate::intent_algebra::Schema> {
             root_scan_schema(expr).or_else(|| root_scan_schema(child))
         }
         // `Ref` has no reachable `Scan` without a `LetBinding` scope, and
-        // the PromQL-surface superset (Scalar/EvalTime/VectorFromScalar/
-        // ScalarFromVector/Relabel/InfoJoin/Sample/TimeRange/TimeShift/
-        // WindowFunc) isn't constructed by this parser today.
+        // the remaining PromQL-surface superset (Scalar/EvalTime/
+        // VectorFromScalar/ScalarFromVector/Relabel/InfoJoin/Sample/
+        // WindowFunc) is real, new capability `lower_promql` adds but this
+        // crate's flat `ParsedQuery` extraction doesn't attempt to unpack
+        // yet -- accepted gap (design-target-architecture.md Part B): these
+        // shapes weren't reachable at all before this swap, so nothing
+        // regresses; `ParsedQuery` may come back incomplete for them.
         _ => None,
     }
 }
@@ -241,6 +237,18 @@ impl QeCollector {
                 }
                 self.visit(child, schema);
             }
+            // `TimeRange` is `asap_l2::lower`'s range-vector-selector marker
+            // (`m[5m]` in `quantile_over_time(φ, m[5m])`) -- the range-window
+            // duration this collector's `time_window` field wants, same as
+            // `Window::size` above. `TimeShift` (`offset`/`@`) is a pure
+            // pass-through, no window/label information of its own.
+            QueryExpr::TimeRange { range, child } => {
+                if self.time_window.is_none() {
+                    self.time_window = Some(*range);
+                }
+                self.visit(child, schema);
+            }
+            QueryExpr::TimeShift { child, .. } => self.visit(child, schema),
             QueryExpr::Aggregate {
                 reduction,
                 aggs,
@@ -477,120 +485,106 @@ pub(super) fn debs_hint(
 mod tests {
     use super::*;
 
+    const ACC: AccuracyTarget = AccuracyTarget::Epsilon(0.01);
+
     // Smoke tests for the parse entry point.
 
     #[test]
     fn promql_dispatched_correctly() {
         // `by` belongs to the aggregate operator, not the function call.
-        let pq = parse_query("sum by (host) (quantile_over_time(0.99, latency[5m]))").unwrap();
+        let pq = parse_query("sum by (host) (quantile_over_time(0.99, latency[5m]))", ACC).unwrap();
         assert!(pq.aggregations.contains(&AggType::Quantile));
         assert_eq!(pq.quantiles, vec![0.99]);
     }
 
     #[test]
     fn parse_query_expr_returns_expr() {
-        let pq =
-            parse_query("topk by (symbol) (10, count_over_time(financial_last_trade_price[5m]))")
-                .unwrap();
+        let pq = parse_query(
+            "topk by (symbol) (10, count_over_time(financial_last_trade_price[5m]))",
+            ACC,
+        )
+        .unwrap();
         // Should parse without error and extract the metric name.
         assert_eq!(pq.metric_name, "financial_last_trade_price");
     }
 
-    // ── Step γ7: canonical-IR entry point ────────────────────────────────────
+    // ── canonical-IR entry point ─────────────────────────────────────────────
 
     #[test]
-    fn canonical_promql_quantile_yields_window_over_aggregate() {
+    fn canonical_promql_quantile_yields_aggregate_over_time_range() {
         use crate::intent_algebra::query_expr::QueryExpr as CQueryExpr;
-        // `quantile_over_time` lowers to a legacy `WindowedAgg`, which
-        // `convert_root` maps to canonical `Window { child: Aggregate }`.
+        // `asap_l2::lower` models a range-vector selector (`m[5m]`) as
+        // `TimeRange`, not `Window` -- `Window` is reserved for real
+        // streaming/tumbling windows. `Aggregate` sits directly on top,
+        // no `Window` wrapper (see this module's own doc for why this
+        // differs from the retired local parser's shape).
         let expr = parse_query_expr_canonical(
             "quantile_over_time(0.99, http_request_duration{env=\"prod\"}[5m])",
+            ACC,
         )
         .unwrap();
         match expr {
-            CQueryExpr::Window { child, .. } => {
-                assert!(matches!(*child, CQueryExpr::Aggregate { .. }));
+            CQueryExpr::Aggregate { child, .. } => {
+                assert!(matches!(*child, CQueryExpr::TimeRange { .. }));
             }
-            other => panic!("expected canonical Window, got {other:?}"),
+            other => panic!("expected canonical Aggregate, got {other:?}"),
         }
     }
 
     #[test]
-    fn canonical_promql_avg_over_time_yields_window_over_aggregate() {
+    fn canonical_promql_avg_over_time_yields_aggregate_over_time_range() {
         use crate::intent_algebra::query_expr::QueryExpr as CQueryExpr;
-        // A bare `avg_over_time(m[w])` (no `by`) lowers to a legacy
-        // `WindowedAgg` over the implicit sample-value column, which
-        // `convert_root` maps to canonical `Window { child: Aggregate }`.
-        let expr = parse_query_expr_canonical("avg_over_time(cpu_seconds_total[10m])").unwrap();
+        // A bare `avg_over_time(m[w])` (no `by`) lowers to canonical
+        // `Aggregate { child: TimeRange { child: Scan } }`.
+        let expr = parse_query_expr_canonical("avg_over_time(cpu_seconds_total[10m])", ACC).unwrap();
         match expr {
-            CQueryExpr::Window { child, .. } => match *child {
-                CQueryExpr::Aggregate { child, .. } => {
+            CQueryExpr::Aggregate { child, .. } => match *child {
+                CQueryExpr::TimeRange { child, .. } => {
                     assert!(matches!(*child, CQueryExpr::Scan { .. }));
                 }
-                other => panic!("expected canonical Aggregate, got {other:?}"),
+                other => panic!("expected canonical TimeRange, got {other:?}"),
             },
-            other => panic!("expected canonical Window, got {other:?}"),
-        }
-    }
-
-    #[test]
-    fn legacy_entry_point_returns_raw_layer2() {
-        use crate::intent_algebra::relational::{AggFunc, QueryExpr as LQueryExpr};
-        // `parse_query_expr` is the crate-internal language-dispatch front
-        // door: it returns the raw legacy Layer-2 relational tree with no
-        // sketch lowering applied — the L2→L3 fusion now lives inside
-        // `convert_root`.
-        let layer2 =
-            parse_query_expr("quantile_over_time(0.99, http_request_duration{env=\"prod\"}[5m])")
-                .unwrap();
-        // Raw Layer 2: an `Aggregate { AggFunc::Quantile }` sitting
-        // *directly* over a `Window` — un-fused, un-lowered.
-        match layer2 {
-            LQueryExpr::Aggregate { aggs, input, .. } => {
-                assert!(matches!(
-                    aggs.as_slice(),
-                    [item] if matches!(item.func, AggFunc::Quantile(_))
-                ));
-                assert!(matches!(*input, LQueryExpr::Window { .. }));
-            }
-            other => panic!("expected raw Layer-2 Aggregate, got {other:?}"),
+            other => panic!("expected canonical Aggregate, got {other:?}"),
         }
     }
 }
 
 #[cfg(test)]
 mod doc_verify_all {
-    // These tests pin the design.md §6 worked examples against the
-    // canonical IR that `parse_query_expr_canonical` produces — the only
-    // algebra IR the parse path now emits. The legacy `WindowedAgg` /
-    // `SketchAgg` fusion the doc text once showed is folded by
-    // `convert_root` into the canonical `Window { Aggregate }` /
-    // `Aggregate { by: [], .. }` stacked forms.
+    // Pins the design.md §6 worked examples against the canonical IR
+    // `asap_frontend_promql::lower_promql` produces — the only algebra IR
+    // the parse path now emits.
     use super::parse_query_expr_canonical;
     use crate::intent_algebra::query_expr::QueryExpr;
     use crate::intent_algebra::{AggIntent, Reduction};
+    use crate::types_v2::AccuracyTarget;
+
+    const ACC: AccuracyTarget = AccuracyTarget::Epsilon(0.01);
 
     #[test]
     fn example4_promql_quantile() {
         let expr = parse_query_expr_canonical(
             "quantile_over_time(0.99, http_request_duration{env=\"prod\"}[5m])",
+            ACC,
         )
         .unwrap();
-        // Canonical fold of the legacy `WindowedAgg { Quantile }`:
-        // `Window { Aggregate { reduction: PerEntity, [Quantile] } }` — no
-        // explicit `by()`, and windowed with a single non-per-series intent,
-        // so there's no grouping concept at all (see #165's `Reduction`).
+        // `Aggregate { reduction: PerEntity, [Quantile], child: TimeRange }`
+        // — no explicit `by()`, and ranged over a single non-per-series
+        // intent, so there's no grouping concept at all (see #165's
+        // `Reduction`). No `Window` wrapper -- `asap_l2::lower` models the
+        // range-vector selector itself as `TimeRange`, not `Window`.
         match &expr {
-            QueryExpr::Window { child, .. } => match child.as_ref() {
-                QueryExpr::Aggregate {
-                    reduction, aggs, ..
-                } => {
-                    assert!(matches!(reduction, Reduction::PerEntity));
-                    assert!(matches!(aggs.as_slice(), [AggIntent::Quantile { .. }]));
-                }
-                other => panic!("expected Aggregate under Window, got {other:?}"),
-            },
-            other => panic!("expected Window, got {other:?}"),
+            QueryExpr::Aggregate {
+                reduction,
+                aggs,
+                child,
+                ..
+            } => {
+                assert!(matches!(reduction, Reduction::PerEntity));
+                assert!(matches!(aggs.as_slice(), [AggIntent::Quantile { .. }]));
+                assert!(matches!(child.as_ref(), QueryExpr::TimeRange { .. }));
+            }
+            other => panic!("expected Aggregate, got {other:?}"),
         }
     }
 
@@ -598,17 +592,36 @@ mod doc_verify_all {
     fn example5_promql_topk() {
         let expr = parse_query_expr_canonical(
             "topk by (service) (10, count_over_time(requests{env=\"prod\"}[1m]))",
+            ACC,
         )
         .unwrap();
-        // The legacy `TopK` folds to a canonical `Aggregate` carrying an
-        // `AggIntent::TopK`, over the `Window { Aggregate { by } } }` the
-        // grouped windowed frequency sketch lowers to — the group-by key
-        // folds straight into the inner `Aggregate.by: GroupKeys` (no
-        // `Partition` wrap; that node doesn't exist in the canonical IR).
+        // `topk(...)` ranking by `count_over_time(...)` is the heavy-hitter
+        // shape both this deployment and `asap_frontend_promql` route to a
+        // canonical `Aggregate` carrying `AggIntent::TopK`, over the
+        // `Window { Aggregate { by } } }` the grouped windowed count lowers
+        // to — regardless of what intent that INNER aggregate now carries
+        // (see this module's doc: no longer necessarily the `Frequency`
+        // extension), the outer `TopK` shape itself is unaffected.
         match &expr {
             QueryExpr::Aggregate { aggs, child, .. } => {
                 assert!(matches!(aggs.as_slice(), [AggIntent::TopK { k: 10, .. }]));
-                assert!(matches!(child.as_ref(), QueryExpr::Window { .. }));
+                // Inner reduction the TopK ranks by: `Aggregate { Count,
+                // child: TimeRange { child: Scan } }` -- same `TimeRange`
+                // shape as the other tests above, one level down. `Count`
+                // carries its own `AccuracyTarget` field (here `Epsilon(0.01)`,
+                // threaded from this call's `accuracy` argument) rather than
+                // being forced exact -- adopted as-is per this module's doc.
+                match child.as_ref() {
+                    QueryExpr::Aggregate {
+                        aggs: inner_aggs,
+                        child: inner_child,
+                        ..
+                    } => {
+                        assert!(matches!(inner_aggs.as_slice(), [AggIntent::Count { .. }]));
+                        assert!(matches!(inner_child.as_ref(), QueryExpr::TimeRange { .. }));
+                    }
+                    other => panic!("expected inner Aggregate, got {other:?}"),
+                }
             }
             other => panic!("expected Aggregate with TopK intent, got {other:?}"),
         }
diff --git a/control_plane/src/query_parser/promql.rs b/control_plane/src/query_parser/promql.rs
deleted file mode 100644
index 1273f411..00000000
--- a/control_plane/src/query_parser/promql.rs
+++ /dev/null
@@ -1,1187 +0,0 @@
-//! Layers 1→2 lowering: PromQL string → QueryExpr (relational plan).
-//!
-//! - **Layer 1**: the `promql-parser` crate parses the PromQL string into a
-//!   language-specific AST (`promql_parser::parser::Expr`).
-//! - **Layer 2**: the walk functions (`walk_qe`, `walk_call_qe`, `walk_aggregate_qe`)
-//!   interpret PromQL semantics (range vectors, aggregation operators, label matchers)
-//!   and emit relational operators (`Aggregate { AggFunc }` + `Window`).
-//!
-//! The output is a Layer 2 `QueryExpr` tree — the same relational operators that
-//! the SQL parser emits.  A shared lowering pass (`algebra::lower`) converts
-//! `Aggregate { AggFunc }` → `SketchAgg { AggIntent }` for both languages.
-//!
-//! # PromQL → AggFunc mapping (summary)
-//!
-//! | Expression | AggFunc |
-//! |---|---|
-//! | `quantile_over_time(φ, m[w])` | Quantile(φ) |
-//! | `histogram_quantile(φ, rate(m[w]))` | Quantile(φ)  (parser-level substitution; see step γ5) |
-//! | `avg_over_time(m[w])` | Avg |
-//! | `min_over_time(m[w])` | Min |
-//! | `max_over_time(m[w])` | Max |
-//! | `stddev/stdvar_over_time(m[w])` | StdDev / Variance |
-//! | `count_over_time(m[w])` | Count / CountDistinct (context) |
-//! | `sum_over_time(m[w])` | Sum |
-//! | `last_over_time / delta / deriv / predict_linear` | Delta / Sum (exact) |
-//! | `changes / resets` | Count |
-//! | `rate / irate / increase` | Rate / Increase |
-//! | `topk(k, …)` outer | TopK (structural — not AggFunc) |
-//! | `count(…over_time… by (d))` outer | CountDistinct |
-//! | `m{filters}` bare | Sum (exact) |
-//! | `m_a op m_b` binary | (BinaryOp — not an Aggregate) |
-
-use std::time::Duration;
-
-use anyhow::anyhow;
-use promql_parser::parser::{self, AggregateExpr, Call, Expr, LabelModifier, VectorSelector};
-
-use crate::intent_algebra::relational::{FilterOp, FilterVal, Predicate};
-
-// ── Walk context ──────────────────────────────────────────────────────────────
-
-/// PromQL `by(labels)` / `without(labels)` aggregation modifier, accumulated
-/// as we descend the AST. Control_plane-only walking state — `asap_l2`'s
-/// `relational::QueryExpr::Aggregate` has no separate `Partition` node to
-/// mirror this against (its `keys`/`without` fields live directly on
-/// `Aggregate`, see `intent_algebra::relational`'s module doc), so this
-/// folds straight into the nearest `Aggregate`/`Window` via
-/// [`fold_group_mod`] instead of wrapping a dedicated node.
-#[derive(Clone)]
-enum GroupMod {
-    By(Vec<String>),
-    Without(Vec<String>),
-}
-
-impl GroupMod {
-    fn keys(&self) -> &[String] {
-        match self {
-            GroupMod::By(k) | GroupMod::Without(k) => k,
-        }
-    }
-    fn is_empty(&self) -> bool {
-        self.keys().is_empty()
-    }
-}
-
-/// Context accumulated as we descend the AST.
-#[derive(Default, Clone)]
-struct WalkCtx {
-    /// GROUP BY / `without` clause from an outer Aggregate node.
-    partition: Option<GroupMod>,
-    /// Top-K k from an outer `topk` / `bottomk` operator.
-    topk: Option<u64>,
-    /// Whether the outer context is a `count()` aggregate (→ CountDistinct).
-    outer_count: bool,
-}
-
-// ── Helpers: MatrixSelector extraction ───────────────────────────────────────
-
-/// Extract `(metric_name, filters, window)` from a MatrixSelector argument at
-/// position `arg_idx` of a Call.
-fn extract_matrix_arg(
-    call: &Call,
-    arg_idx: usize,
-) -> anyhow::Result<(String, Vec<Predicate>, Duration)> {
-    let arg = call
-        .args
-        .args
-        .get(arg_idx)
-        .map(|b| b.as_ref())
-        .ok_or_else(|| anyhow!("missing arg {} in call to {}", arg_idx, call.func.name))?;
-    extract_inner_matrix(arg)
-}
-
-/// Walk into an expression until we find a MatrixSelector, then extract its info.
-fn extract_inner_matrix(expr: &Expr) -> anyhow::Result<(String, Vec<Predicate>, Duration)> {
-    match expr {
-        Expr::MatrixSelector(ms) => {
-            let (name, filters) = extract_vs_info(&ms.vs);
-            Ok((name, filters, ms.range))
-        }
-        Expr::Paren(p) => extract_inner_matrix(p.expr.as_ref()),
-        Expr::Call(c) => {
-            // rate/irate wraps a MatrixSelector.
-            extract_inner_matrix(c.args.args[0].as_ref())
-        }
-        other => Err(anyhow!(
-            "expected MatrixSelector, got {:?}",
-            std::mem::discriminant(other)
-        )),
-    }
-}
-
-// ── Helpers: VectorSelector info ─────────────────────────────────────────────
-
-fn extract_vs_info(vs: &VectorSelector) -> (String, Vec<Predicate>) {
-    // Metric name: prefer the explicit name field, fall back to __name__ matcher.
-    let name = vs.name.clone().unwrap_or_else(|| {
-        vs.matchers
-            .matchers
-            .iter()
-            .find(|m| m.name == "__name__")
-            .map(|m| m.value.clone())
-            .unwrap_or_default()
-    });
-
-    let filters = vs
-        .matchers
-        .matchers
-        .iter()
-        .filter(|m| m.name != "__name__")
-        .filter_map(matcher_to_predicate)
-        .collect();
-
-    (name, filters)
-}
-
-fn matcher_to_predicate(m: &promql_parser::label::Matcher) -> Option<Predicate> {
-    use promql_parser::label::MatchOp;
-    let (op, val) = match &m.op {
-        MatchOp::Equal => (FilterOp::Eq, FilterVal::Str(m.value.clone())),
-        MatchOp::NotEqual => (FilterOp::Ne, FilterVal::Str(m.value.clone())),
-        MatchOp::Re(re) => (
-            FilterOp::Regex(re.to_string()),
-            FilterVal::Str(m.value.clone()),
-        ),
-        MatchOp::NotRe(re) => (
-            FilterOp::NotRegex(re.to_string()),
-            FilterVal::Str(m.value.clone()),
-        ),
-    };
-    Some(Predicate {
-        col: m.name.clone(),
-        op,
-        val,
-    })
-}
-
-// ── Helpers: number extraction ────────────────────────────────────────────────
-
-fn extract_call_num_arg(call: &Call, idx: usize) -> anyhow::Result<f64> {
-    match call.args.args.get(idx).map(|b| b.as_ref()) {
-        Some(Expr::NumberLiteral(n)) => Ok(n.val),
-        Some(other) => Err(anyhow!(
-            "expected number at arg {} of {}, got {:?}",
-            idx,
-            call.func.name,
-            std::mem::discriminant(other)
-        )),
-        None => Err(anyhow!("missing arg {} in {}", idx, call.func.name)),
-    }
-}
-
-fn extract_number_param(param: &Option<Box<Expr>>) -> anyhow::Result<f64> {
-    match param {
-        Some(e) => match e.as_ref() {
-            Expr::NumberLiteral(n) => Ok(n.val),
-            other => Err(anyhow!(
-                "expected number param, got {:?}",
-                std::mem::discriminant(other)
-            )),
-        },
-        None => Err(anyhow!("missing required numeric parameter")),
-    }
-}
-
-// ── Helpers: GroupMod from LabelModifier ──────────────────────────────────────
-
-fn modifier_to_partition(modifier: &LabelModifier) -> GroupMod {
-    match modifier {
-        LabelModifier::Include(labels) => GroupMod::By(labels.labels.clone()),
-        LabelModifier::Exclude(labels) => GroupMod::Without(labels.labels.clone()),
-    }
-}
-
-// ── Direct QueryExpr emission ─────────────────────────────────────────────────
-//
-// `parse_promql_expr` walks the same PromQL AST but emits [`QueryExpr`] nodes
-// natively, preserving semantic nodes for the algebra optimizer:
-//
-// | PromQL pattern           | QueryExpr node                        |
-// |--------------------------|---------------------------------------|
-// | `histogram_quantile(φ…)` | Aggregate { Quantile(φ) } (step γ5)   |
-// | `m[5m:1m]` subquery      | PromQLSubquery { 5m, Some(1m) }      |
-// | `a op b` binary          | BinaryOp { VectorMatch }             |
-
-use crate::intent_algebra::relational::{
-    AggFunc, AggItem, BinaryOpKind, ColumnRef as QeColumnRef, GroupSide, L2SortKey, QueryExpr,
-    SourceSpec as QeSourceSpec, VectorGrouping, VectorMatch, VectorMatchKind,
-};
-use crate::intent_algebra::{
-    is_frequency_heavy_hitter, ArithOp, CompareOp, L2Expr, RankingMeasure,
-};
-use promql_parser::parser::{token::TokenType, BinaryExpr, VectorMatchCardinality};
-
-/// Parse a PromQL expression string directly into an optimised [`QueryExpr`].
-///
-/// This preserves `PromQLSubquery` and `BinaryOp` nodes natively;
-/// `histogram_quantile(φ, …)` is substituted into a plain
-/// `Aggregate { Quantile(φ) }` per Step γ5 of the relational migration.
-pub fn parse_promql_expr(query: &str) -> anyhow::Result<QueryExpr> {
-    let expr = parser::parse(query).map_err(|e| anyhow!("PromQL parse error: {e}"))?;
-    walk_qe(&expr, WalkCtx::default())
-}
-
-fn walk_qe(expr: &Expr, ctx: WalkCtx) -> anyhow::Result<QueryExpr> {
-    match expr {
-        Expr::Aggregate(agg) => walk_aggregate_qe(agg, ctx),
-        Expr::Call(call) => walk_call_qe(call, ctx),
-
-        // Binary op: map to QueryExpr::BinaryOp with VectorMatch.
-        Expr::Binary(bin) => walk_binary_qe(bin),
-
-        Expr::Paren(p) => walk_qe(p.expr.as_ref(), ctx),
-
-        // Subquery `expr[range:resolution]` → PromQLSubquery.
-        Expr::Subquery(sq) => {
-            let inner = walk_qe(sq.expr.as_ref(), ctx.clone())?;
-            Ok(QueryExpr::PromQLSubquery {
-                range: sq.range,
-                resolution: sq.step,
-                input: Box::new(inner),
-            })
-        }
-
-        // Bare vector selector → Source + Filter, with `Aggregate(Sum)`
-        // wrapping ONLY when the outer context is value-aggregating
-        // (sum / avg / min / max / etc.). PromQL's `count(metric)`
-        // operates on the result-set's LABEL-SETS — the inner
-        // selector is just "the things to count," not a value to sum.
-        // PromQL's `topk(k, metric)` operates on the SERIES — the
-        // inner selector is the population to rank, not a value to
-        // sum. Synthesizing `Aggregate(Sum)` underneath either outer
-        // would collect a redundant `ExactAgg(Sum)` candidate
-        // alongside the intended `CardinalityApprox` /
-        // `FrequencyTopk(*WithHeap)` one; the engine's
-        // "all candidates must succeed" semantic then surfaces a
-        // `CapabilityMiss` when no Sum policy is registered (e.g. an
-        // HLL-only or CMS-with-heap-only deploy).
-        Expr::VectorSelector(vs) => {
-            let (name, filters) = extract_vs_info(vs);
-            let source = QueryExpr::Source(QeSourceSpec::new(name));
-            let filtered = apply_qe_filters(source, filters);
-            if ctx.outer_count || ctx.topk.is_some() {
-                Ok(filtered)
-            } else {
-                Ok(QueryExpr::Aggregate {
-                    keys: vec![],
-                    without: false,
-                    aggs: vec![AggItem {
-                        alias: Some("value".into()),
-                        func: AggFunc::Sum,
-                        col: QeColumnRef::SampleValue,
-                    }],
-                    having: None,
-                    input: Box::new(filtered),
-                })
-            }
-        }
-
-        Expr::NumberLiteral(_) | Expr::StringLiteral(_) => Err(anyhow!(
-            "unexpected literal at top level of PromQL expression"
-        )),
-
-        #[allow(unreachable_patterns)]
-        _ => Err(anyhow!("unsupported PromQL expression type")),
-    }
-}
-
-/// Whether `expr` (a `topk`/`bottomk` argument) is `count_over_time(...)`,
-/// possibly parenthesized — the one PromQL shape ranked by
-/// `RankingMeasure::Frequency`, the only measure with a realised
-/// heavy-hitter sketch today (`agg_intent::is_frequency_heavy_hitter`).
-/// A bare `count(...)` doesn't qualify: it's a *cross-series* reduction
-/// (one value, not per-series), so ranking by it isn't a per-series
-/// heavy-hitter shape in the first place.
-fn is_count_over_time(expr: &Expr) -> bool {
-    match expr {
-        Expr::Paren(p) => is_count_over_time(p.expr.as_ref()),
-        Expr::Call(c) => c.func.name == "count_over_time",
-        _ => false,
-    }
-}
-
-fn walk_aggregate_qe(agg: &AggregateExpr, ctx: WalkCtx) -> anyhow::Result<QueryExpr> {
-    let partition = agg.modifier.as_ref().map(modifier_to_partition);
-    let op_name = format!("{}", agg.op);
-
-    match op_name.as_str() {
-        "topk" | "bottomk" => {
-            let k = extract_number_param(&agg.param)? as u64;
-            let descending = op_name == "topk";
-            // A ranking is the heavy-hitter `TopK` intent only when it
-            // takes the *top* k (`descending` — `bottomk` never
-            // qualifies) *and* ranks by a measure with a realised
-            // heavy-hitter sketch — today, unweighted frequency
-            // (`count_over_time(...)`) only. Every other measure
-            // (avg/quantile/rate/a bare selector/...) is
-            // `RankingMeasure::NonAdditive` and falls through to a
-            // generic `Sort + Limit` order-by-value below — matching
-            // ASAPController's `frontend-promql` design (issue #38: "the
-            // descending-plus-measure rule is shared with the L3
-            // canonicalize promotion so the two cannot drift"). Before
-            // this, `topk(k, <anything>)` unconditionally forced a
-            // `Count`-shaped inner regardless of what was actually being
-            // ranked — silently wrong for `topk(k, avg_over_time(...))`
-            // and friends.
-            let measure = if is_count_over_time(agg.expr.as_ref()) {
-                RankingMeasure::Frequency
-            } else {
-                RankingMeasure::NonAdditive
-            };
-            if is_frequency_heavy_hitter(descending, measure) {
-                let inner_ctx = WalkCtx {
-                    partition: partition.clone(),
-                    topk: Some(k),
-                    outer_count: false,
-                };
-                let inner = walk_qe(agg.expr.as_ref(), inner_ctx)?;
-                // Don't fold the group keys into a separate wrapper here —
-                // the inner Aggregate already has them (or will, once its
-                // own construction site folds `partition` in).
-                return Ok(QueryExpr::TopK {
-                    k,
-                    by: partition
-                        .as_ref()
-                        .map(|p| p.keys().iter().cloned().map(QeColumnRef::Named).collect())
-                        .unwrap_or_default(),
-                    input: Box::new(inner),
-                });
-            }
-            // Generic order-by-value + limit: `bottomk`, and any `topk`
-            // ranking by a non-count measure. `ctx.topk` stays `None` so
-            // `build_qe_aggregate` doesn't force a `Count`-shaped inner.
-            let inner_ctx = WalkCtx {
-                partition: partition.clone(),
-                topk: None,
-                outer_count: false,
-            };
-            let inner = walk_qe(agg.expr.as_ref(), inner_ctx)?;
-            let sorted = QueryExpr::Sort {
-                keys: vec![L2SortKey {
-                    expr: L2Expr::Column(QeColumnRef::SampleValue),
-                    ascending: !descending,
-                    nulls_first: false,
-                }],
-                partition_by: partition
-                    .as_ref()
-                    .map(|p| p.keys().iter().cloned().map(QeColumnRef::Named).collect())
-                    .unwrap_or_default(),
-                input: Box::new(inner),
-            };
-            let result = QueryExpr::Limit {
-                n: k,
-                offset: 0,
-                input: Box::new(sorted),
-            };
-            Ok(result)
-        }
-        "count" => {
-            let inner_ctx = WalkCtx {
-                partition: partition.clone(),
-                topk: None,
-                outer_count: true,
-            };
-            let inner = walk_qe(agg.expr.as_ref(), inner_ctx)?;
-            let result = QueryExpr::Aggregate {
-                keys: vec![],
-                without: false,
-                aggs: vec![AggItem {
-                    alias: Some("count".into()),
-                    func: AggFunc::CountDistinct,
-                    col: QeColumnRef::SampleValue,
-                }],
-                having: None,
-                input: Box::new(inner),
-            };
-            Ok(fold_group_mod(result, partition.as_ref()))
-        }
-        "sum" | "avg" | "min" | "max" | "group" => {
-            let inner_ctx = WalkCtx {
-                partition: partition.clone(),
-                topk: ctx.topk,
-                outer_count: false,
-            };
-            let inner = walk_qe(agg.expr.as_ref(), inner_ctx)?;
-            Ok(fold_group_mod(inner, partition.as_ref()))
-        }
-        "stddev" => {
-            let inner_ctx = WalkCtx {
-                partition: partition.clone(),
-                topk: None,
-                outer_count: false,
-            };
-            let inner = walk_qe(agg.expr.as_ref(), inner_ctx)?;
-            let result = QueryExpr::Aggregate {
-                keys: vec![],
-                without: false,
-                aggs: vec![AggItem {
-                    alias: Some("stddev".into()),
-                    func: AggFunc::StdDev { population: false },
-                    col: QeColumnRef::SampleValue,
-                }],
-                having: None,
-                input: Box::new(inner),
-            };
-            Ok(fold_group_mod(result, partition.as_ref()))
-        }
-        "stdvar" => {
-            let inner_ctx = WalkCtx {
-                partition: partition.clone(),
-                topk: None,
-                outer_count: false,
-            };
-            let inner = walk_qe(agg.expr.as_ref(), inner_ctx)?;
-            let result = QueryExpr::Aggregate {
-                keys: vec![],
-                without: false,
-                aggs: vec![AggItem {
-                    alias: Some("stdvar".into()),
-                    func: AggFunc::Variance { population: false },
-                    col: QeColumnRef::SampleValue,
-                }],
-                having: None,
-                input: Box::new(inner),
-            };
-            Ok(fold_group_mod(result, partition.as_ref()))
-        }
-        "quantile" => {
-            let phi = extract_number_param(&agg.param)?;
-            let inner_ctx = WalkCtx {
-                partition: partition.clone(),
-                topk: None,
-                outer_count: false,
-            };
-            let inner = walk_qe(agg.expr.as_ref(), inner_ctx)?;
-            let result = QueryExpr::Aggregate {
-                keys: vec![],
-                without: false,
-                aggs: vec![AggItem {
-                    alias: Some("quantile".into()),
-                    func: AggFunc::Quantile(phi),
-                    col: QeColumnRef::SampleValue,
-                }],
-                having: None,
-                input: Box::new(inner),
-            };
-            Ok(fold_group_mod(result, partition.as_ref()))
-        }
-        other => Err(anyhow!("unsupported PromQL aggregate operator: {other}")),
-    }
-}
-
-fn walk_call_qe(call: &Call, ctx: WalkCtx) -> anyhow::Result<QueryExpr> {
-    let name = call.func.name;
-    match name {
-        // histogram_quantile(φ, bucket_metric) → plain Aggregate { Quantile(φ) }.
-        //
-        // Per Step γ5 of the relational migration: at the PromQL parser level
-        // we substitute `histogram_quantile(φ, bucket_metric)` with the same
-        // shape that `quantile_over_time(φ, m[w])` produces — an `Aggregate`
-        // carrying a single `AggFunc::Quantile(φ)`. Downstream code (the
-        // L1→L3 lowerer, the optimizer, the physical planner) then sees a
-        // plain Quantile and routes via the existing `AggIntent::Quantile`
-        // path. Bucket-aware physical reduction is a physical-planner
-        // concern, not an IR variant. The legacy `QueryExpr::HistogramQuantile`
-        // variant has been retired.
-        //
-        // The inner `rate(...)` is the buckets argument; we use a fresh
-        // `WalkCtx::default()` because an outer `topk` context would otherwise
-        // rewrite the Quantile(φ) into a Count-frequency aggregate, which
-        // would be semantically wrong for the histogram-quantile reduction.
-        "histogram_quantile" => {
-            let phi = extract_call_num_arg(call, 0)?;
-            let rate_expr = call.args.args[1].as_ref();
-            let (source, filters, window) = extract_inner_matrix(rate_expr)?;
-            Ok(build_qe_aggregate(
-                source,
-                filters,
-                window,
-                AggFunc::Quantile(phi),
-                WalkCtx::default(),
-            ))
-        }
-        // All other function calls: map to AggFunc (Layer 2).
-        "quantile_over_time" => {
-            let phi = extract_call_num_arg(call, 0)?;
-            let (source, filters, window) = extract_matrix_arg(call, 1)?;
-            let func = AggFunc::Quantile(phi);
-            Ok(build_qe_aggregate(source, filters, window, func, ctx))
-        }
-        // `predict_linear(v[w], t)` — `t` (seconds into the future) is a
-        // scalar 2nd argument, so it doesn't fit `walk_call_to_op`'s
-        // `(call, ctx, window)` shape; special-cased here like
-        // `quantile_over_time`'s φ argument above.
-        "predict_linear" => {
-            let seconds = extract_call_num_arg(call, 1)?;
-            let (source, filters, window) = extract_matrix_arg(call, 0)?;
-            let func = AggFunc::PredictLinear { seconds };
-            Ok(build_qe_aggregate(source, filters, window, func, ctx))
-        }
-        _ => {
-            let (source, filters, window) = if call.func.name == "rate"
-                || call.func.name == "irate"
-                || call.func.name == "increase"
-            {
-                let arg = call
-                    .args
-                    .args
-                    .first()
-                    .map(|b| b.as_ref())
-                    .ok_or_else(|| anyhow!("rate/irate/increase requires a matrix arg"))?;
-                extract_inner_matrix(arg)?
-            } else {
-                extract_matrix_arg(call, 0)?
-            };
-            let func = walk_call_to_op(call, &ctx, window)?;
-            Ok(build_qe_aggregate(source, filters, window, func, ctx))
-        }
-    }
-}
-
-fn walk_binary_qe(bin: &BinaryExpr) -> anyhow::Result<QueryExpr> {
-    let lhs = walk_qe(bin.lhs.as_ref(), WalkCtx::default())?;
-    let rhs = walk_qe(bin.rhs.as_ref(), WalkCtx::default())?;
-
-    let op = promql_token_to_binop(bin.op);
-
-    let vector_match = bin.modifier.as_ref().map(|m| {
-        let (kind, labels) = match &m.matching {
-            Some(LabelModifier::Include(ls)) => (VectorMatchKind::On, ls.labels.clone()),
-            Some(LabelModifier::Exclude(ls)) => (VectorMatchKind::Ignoring, ls.labels.clone()),
-            None => (VectorMatchKind::On, vec![]),
-        };
-        let grouping = match &m.card {
-            VectorMatchCardinality::ManyToOne(ls) => Some(VectorGrouping {
-                side: GroupSide::Left,
-                labels: ls.labels.clone(),
-            }),
-            VectorMatchCardinality::OneToMany(ls) => Some(VectorGrouping {
-                side: GroupSide::Right,
-                labels: ls.labels.clone(),
-            }),
-            _ => None,
-        };
-        VectorMatch {
-            kind,
-            labels,
-            grouping,
-        }
-    });
-
-    Ok(QueryExpr::BinaryOp {
-        op,
-        lhs: Box::new(lhs),
-        rhs: Box::new(rhs),
-        vector_match,
-    })
-}
-
-fn promql_token_to_binop(tok: TokenType) -> BinaryOpKind {
-    use promql_parser::parser::token;
-    // token::T_* are u8 constants; TokenType wraps them as TokenType(u8).
-    let id = tok.id();
-    match id {
-        token::T_ADD => BinaryOpKind::Arith(ArithOp::Add),
-        token::T_SUB => BinaryOpKind::Arith(ArithOp::Sub),
-        token::T_MUL => BinaryOpKind::Arith(ArithOp::Mul),
-        token::T_DIV => BinaryOpKind::Arith(ArithOp::Div),
-        token::T_MOD => BinaryOpKind::Arith(ArithOp::Mod),
-        token::T_POW => BinaryOpKind::Pow,
-        token::T_EQLC => BinaryOpKind::Compare(CompareOp::Eq),
-        token::T_NEQ => BinaryOpKind::Compare(CompareOp::Ne),
-        token::T_LSS => BinaryOpKind::Compare(CompareOp::Lt),
-        token::T_LTE => BinaryOpKind::Compare(CompareOp::Le),
-        token::T_GTR => BinaryOpKind::Compare(CompareOp::Gt),
-        token::T_GTE => BinaryOpKind::Compare(CompareOp::Ge),
-        token::T_LAND => BinaryOpKind::And,
-        token::T_LOR => BinaryOpKind::Or,
-        token::T_LUNLESS => BinaryOpKind::Unless,
-        token::T_ATAN2 => BinaryOpKind::Atan2,
-        _ => BinaryOpKind::Arith(ArithOp::Add), // unknown — default to add
-    }
-}
-
-/// Map a PromQL function call to an [`AggFunc`] (Layer 2 relational operator).
-/// `window` is the range-vector's duration — only `Rate`/`Increase` carry it
-/// on the `AggFunc` itself (`asap_l2`'s design: "no separate Window node").
-/// Every other function still relies on the caller wrapping its `Aggregate`
-/// in an `L2::Window`.
-///
-/// `predict_linear` is handled separately in `walk_call_qe` (its 2nd,
-/// scalar argument doesn't fit this function's `(call, ctx, window)`
-/// shape, matching how `quantile_over_time`/`histogram_quantile`'s φ
-/// argument is already special-cased there).
-fn walk_call_to_op(call: &Call, ctx: &WalkCtx, window: Duration) -> anyhow::Result<AggFunc> {
-    let name = call.func.name;
-    match name {
-        "quantile_over_time" => {
-            let phi = extract_call_num_arg(call, 0)?;
-            Ok(AggFunc::Quantile(phi))
-        }
-        "avg_over_time" => Ok(AggFunc::Avg),
-        "min_over_time" => Ok(AggFunc::Min),
-        "max_over_time" => Ok(AggFunc::Max),
-        "stddev_over_time" => Ok(AggFunc::StdDev { population: false }),
-        "stdvar_over_time" => Ok(AggFunc::Variance { population: false }),
-        "count_over_time" => {
-            // Two cases, in priority order:
-            //  * Inside `count by (...) (count_over_time(...))` →
-            //    `CountDistinct` (HLL distinct counting; the outer
-            //    count of inner counts is cardinality).
-            //  * Otherwise → plain `Count`, still windowed here — the
-            //    `lower.rs::agg_func_to_intents` grouped-or-windowed
-            //    substitution recognizes the windowed shape and routes
-            //    to CMS / CountSketch (per-series sample-count
-            //    estimation), matching the pre-`asap_l2` behavior this
-            //    used to reach via a dedicated `AggFunc::Frequency`
-            //    variant that no longer exists. `topk(N, count_over_time
-            //    (...))`'s heavy-hitter fusion is handled entirely in
-            //    `walk_aggregate_qe`'s `topk`/`bottomk` arm now (it no
-            //    longer forces `Count` here via `ctx.topk`).
-            Ok(if ctx.outer_count {
-                AggFunc::CountDistinct
-            } else {
-                AggFunc::Count
-            })
-        }
-        "sum_over_time" => Ok(AggFunc::Sum),
-        "last_over_time" => Ok(AggFunc::LastOverTime),
-        "present_over_time" => Ok(AggFunc::PresentOverTime),
-        "absent_over_time" => Ok(AggFunc::AbsentOverTime),
-        // Each of these used to collapse onto `Delta` (delta/idelta/deriv)
-        // or `Count` (changes/resets) — a crude placeholder bucketing from
-        // before `asap_l2` gave every one of them its own dedicated
-        // `AggFunc`/`AggIntent`. All are archive-only (no ASAP-tier
-        // Bind* rule exists for any of them, same as before this fix) —
-        // the correctness gain is `capability_for` now correctly
-        // returning `None` (route to archive) instead of the wrong
-        // `Some(ExactAgg(Sum))` / `Some(CardinalityApprox))` these used
-        // to produce, which claimed the ASAP tier could answer a
-        // `deriv()`/`changes()` query it structurally cannot.
-        "delta" => Ok(AggFunc::Delta),
-        "idelta" => Ok(AggFunc::IDelta),
-        "deriv" => Ok(AggFunc::Deriv),
-        "changes" => Ok(AggFunc::Changes),
-        "resets" => Ok(AggFunc::Resets),
-        // `Rate`/`Increase` now map onto their own dedicated `AggIntent`s
-        // in `lower.rs` (`capability_for` already has a real, tested
-        // `Rate | Increase => ExactAgg(Increase)` arm — this activates
-        // it for the first time via the PromQL path; previously
-        // collapsed onto `AggIntent::Sum`, which happened to route to
-        // the same *kind* of exact-precompute capability but under the
-        // wrong classification).
-        "rate" | "irate" => Ok(AggFunc::Rate { window }),
-        "increase" => Ok(AggFunc::Increase { window }),
-        other => Err(anyhow!("unsupported PromQL function: {other}")),
-    }
-}
-
-/// Short lowercase label for an `AggFunc`, used as the `AggItem` alias.
-/// `AggFunc` is foreign (from `asap_l2`) — Rust's orphan rules forbid
-/// implementing `Display` for it here, unlike this repo's pre-merge own
-/// `AggFunc`, which had one.
-fn agg_func_label(f: &AggFunc) -> String {
-    match f {
-        AggFunc::Count => "count".into(),
-        AggFunc::Sum => "sum".into(),
-        AggFunc::Avg => "avg".into(),
-        AggFunc::Min => "min".into(),
-        AggFunc::Max => "max".into(),
-        AggFunc::StdDev { .. } => "stddev".into(),
-        AggFunc::Variance { .. } => "variance".into(),
-        AggFunc::Quantile(_) => "quantile".into(),
-        AggFunc::CountDistinct => "count_distinct".into(),
-        AggFunc::HeavyHitters { .. } => "heavy_hitters".into(),
-        AggFunc::Rate { .. } => "rate".into(),
-        AggFunc::Increase { .. } => "increase".into(),
-        AggFunc::Delta => "delta".into(),
-        other => format!("{other:?}").to_lowercase(),
-    }
-}
-
-/// Build a Layer 2 `QueryExpr`: `Aggregate { AggFunc, input: Window { ... } }`.
-fn build_qe_aggregate(
-    metric: String,
-    filters: Vec<Predicate>,
-    window: std::time::Duration,
-    func: AggFunc,
-    ctx: WalkCtx,
-) -> QueryExpr {
-    let source = QueryExpr::Source(QeSourceSpec::new(metric));
-    let filtered = apply_qe_filters(source, filters);
-    let windowed = QueryExpr::Window {
-        duration: window,
-        slide: None,
-        input: Box::new(filtered),
-    };
-    let actual_func = if ctx.topk.is_some() {
-        // Inside topk context, the aggregation is frequency-based.
-        AggFunc::Count
-    } else {
-        func
-    };
-    // Propagate partition keys into the Aggregate's GROUP BY so
-    // `lower.rs`'s `agg_func_to_intents` sees a grouped `Count` → the
-    // `Frequency` sketch trigger (not a bare, exact `Count`).
-    let (group_keys, without): (Vec<String>, bool) = match &ctx.partition {
-        Some(GroupMod::By(k)) => (k.clone(), false),
-        Some(GroupMod::Without(k)) => (k.clone(), true),
-        None => (Vec::new(), false),
-    };
-    let alias = agg_func_label(&actual_func);
-    QueryExpr::Aggregate {
-        keys: group_keys.into_iter().map(QeColumnRef::Named).collect(),
-        without,
-        aggs: vec![AggItem {
-            alias: Some(alias),
-            func: actual_func,
-            col: QeColumnRef::SampleValue,
-        }],
-        having: None,
-        input: Box::new(windowed),
-    }
-}
-
-fn apply_qe_filters(input: QueryExpr, filters: Vec<Predicate>) -> QueryExpr {
-    if filters.is_empty() {
-        return input;
-    }
-    use crate::intent_algebra::{L2Expr, L3Scalar};
-
-    let conjuncts: Vec<L2Expr> = filters
-        .iter()
-        .map(|p| {
-            let col = L2Expr::Column(QeColumnRef::Named(p.col.clone()));
-            let val = |v: &FilterVal| match v {
-                FilterVal::Str(s) => L2Expr::Literal(L3Scalar::Utf8(s.clone())),
-                FilterVal::Num(n) => L2Expr::Literal(L3Scalar::Float64(*n)),
-                FilterVal::Int(i) => L2Expr::Literal(L3Scalar::Int64(*i)),
-                FilterVal::Null => L2Expr::Literal(L3Scalar::Null),
-            };
-            match &p.op {
-                FilterOp::Eq => L2Expr::Compare {
-                    left: Box::new(col),
-                    op: CompareOp::Eq,
-                    right: Box::new(val(&p.val)),
-                },
-                FilterOp::Ne => L2Expr::Compare {
-                    left: Box::new(col),
-                    op: CompareOp::Ne,
-                    right: Box::new(val(&p.val)),
-                },
-                FilterOp::Lt => L2Expr::Compare {
-                    left: Box::new(col),
-                    op: CompareOp::Lt,
-                    right: Box::new(val(&p.val)),
-                },
-                FilterOp::Le => L2Expr::Compare {
-                    left: Box::new(col),
-                    op: CompareOp::Le,
-                    right: Box::new(val(&p.val)),
-                },
-                FilterOp::Gt => L2Expr::Compare {
-                    left: Box::new(col),
-                    op: CompareOp::Gt,
-                    right: Box::new(val(&p.val)),
-                },
-                FilterOp::Ge => L2Expr::Compare {
-                    left: Box::new(col),
-                    op: CompareOp::Ge,
-                    right: Box::new(val(&p.val)),
-                },
-                FilterOp::Regex(r) => L2Expr::Compare {
-                    left: Box::new(col),
-                    op: CompareOp::Regex,
-                    right: Box::new(L2Expr::Literal(L3Scalar::Utf8(r.clone()))),
-                },
-                FilterOp::NotRegex(r) => L2Expr::Compare {
-                    left: Box::new(col),
-                    op: CompareOp::NotRegex,
-                    right: Box::new(L2Expr::Literal(L3Scalar::Utf8(r.clone()))),
-                },
-                FilterOp::Like => L2Expr::Compare {
-                    left: Box::new(col),
-                    op: CompareOp::Like,
-                    right: Box::new(val(&p.val)),
-                },
-                FilterOp::NotLike => L2Expr::Compare {
-                    left: Box::new(col),
-                    op: CompareOp::NotLike,
-                    right: Box::new(val(&p.val)),
-                },
-                FilterOp::IsNull => L2Expr::IsNull(Box::new(col)),
-                FilterOp::IsNotNull => L2Expr::IsNotNull(Box::new(col)),
-            }
-        })
-        .collect();
-    let pred = if conjuncts.len() == 1 {
-        conjuncts.into_iter().next().unwrap()
-    } else {
-        L2Expr::BoolAnd(conjuncts)
-    };
-    QueryExpr::Filter {
-        pred,
-        input: Box::new(input),
-    }
-}
-
-/// Fold `group` into the nearest `Aggregate` inside `qe` — `asap_l2`'s
-/// `Aggregate` carries `keys`/`without` directly (no separate `Partition`
-/// node to wrap in; see `intent_algebra::relational`'s module doc).
-/// Mirrors `lower.rs`'s `fold_partition_keys`, one layer up (L2, not L3):
-/// handles the shapes the walker actually produces (a bare `Aggregate` or
-/// a `Window` wrapping one); anything else (`BinaryOp`, a bare `Source`)
-/// has no `Aggregate` to fold into and passes through unchanged — e.g.
-/// `sum by (host) (a or b)`, where the group modifier belongs to a
-/// `BinaryOp` composition, not a reducing aggregate.
-fn fold_group_mod(qe: QueryExpr, group: Option<&GroupMod>) -> QueryExpr {
-    let Some(group) = group else {
-        return qe;
-    };
-    if group.is_empty() {
-        return qe;
-    }
-    match qe {
-        QueryExpr::Aggregate {
-            aggs,
-            having,
-            input,
-            ..
-        } => QueryExpr::Aggregate {
-            keys: group
-                .keys()
-                .iter()
-                .cloned()
-                .map(QeColumnRef::Named)
-                .collect(),
-            without: matches!(group, GroupMod::Without(_)),
-            aggs,
-            having,
-            input,
-        },
-        QueryExpr::Window {
-            duration,
-            slide,
-            input,
-        } => QueryExpr::Window {
-            duration,
-            slide,
-            input: Box::new(fold_group_mod(*input, Some(group))),
-        },
-        other => other,
-    }
-}
-
-// ── Tests ─────────────────────────────────────────────────────────────────────
-
-#[cfg(test)]
-mod tests {
-    use crate::types::AggType;
-    use std::time::Duration;
-
-    fn pq(q: &str) -> super::super::ParsedQuery {
-        super::super::parse_query(q)
-            .unwrap_or_else(|e| panic!("parse_query failed: {e}\nquery={q:?}"))
-    }
-
-    // ── quantile_over_time ────────────────────────────────────────────────────
-
-    #[test]
-    fn quantile_over_time_basic() {
-        // PromQL: `by` is part of the aggregate operator, not the function call.
-        let pq = pq("sum by (host) (quantile_over_time(0.99, latency{service=\"web\"}[5m]))");
-        assert_eq!(pq.aggregations, vec![AggType::Quantile]);
-        assert_eq!(pq.quantiles, vec![0.99]);
-        assert_eq!(pq.group_by_labels, vec!["host"]);
-        assert_eq!(
-            pq.label_filters.get("service").map(String::as_str),
-            Some("web")
-        );
-        assert_eq!(pq.time_window, Duration::from_secs(300));
-    }
-
-    #[test]
-    fn quantile_over_time_debs_ema() {
-        // Dotted names are invalid PromQL; use underscores.
-        let pq = pq("sum by (symbol) (quantile_over_time(0.5, financial_last_trade_price[5m]))");
-        assert_eq!(pq.aggregations, vec![AggType::Quantile]);
-        assert_eq!(pq.quantiles, vec![0.5]);
-        assert_eq!(pq.group_by_labels, vec!["symbol"]);
-    }
-
-    // ── histogram_quantile ────────────────────────────────────────────────────
-
-    #[test]
-    fn histogram_quantile_via_rate() {
-        let pq = pq("histogram_quantile(0.95, rate(http_duration_seconds_bucket[5m]))");
-        assert_eq!(pq.aggregations, vec![AggType::Quantile]);
-        assert_eq!(pq.quantiles, vec![0.95]);
-    }
-
-    /// Step γ5 contract: at the PromQL parser level, `histogram_quantile(φ, …)`
-    /// is substituted into a plain `QueryExpr::Aggregate { aggs: [AggItem {
-    /// func: AggFunc::Quantile(φ), … }], … }` so downstream code sees a
-    /// single canonical Quantile intent (no `QueryExpr::HistogramQuantile`
-    /// wrapper). `ParsedQuery.quantiles` records the φ via the existing
-    /// multi-quantile machinery.
-    #[test]
-    fn histogram_quantile_lowers_to_plain_aggregate_quantile() {
-        use crate::intent_algebra::relational::{AggFunc, ColumnRef, QueryExpr};
-
-        let qe = super::parse_promql_expr(
-            r#"histogram_quantile(0.99, rate(http_requests_bucket{le="0.5"}[5m]))"#,
-        )
-        .expect("parse should succeed");
-
-        // The top of the tree must be a plain Aggregate with a single
-        // Quantile(0.99) AggItem — NOT a HistogramQuantile wrapper.
-        match &qe {
-            QueryExpr::Aggregate { aggs, .. } => {
-                assert_eq!(aggs.len(), 1, "expected single AggItem, got {aggs:?}");
-                let item = &aggs[0];
-                match item.func {
-                    AggFunc::Quantile(phi) => {
-                        assert!((phi - 0.99).abs() < 1e-9, "expected φ=0.99, got {phi}");
-                    }
-                    ref other => panic!("expected AggFunc::Quantile(0.99), got {other:?}"),
-                }
-                assert!(matches!(item.col, ColumnRef::SampleValue));
-            }
-            other => panic!("expected Aggregate, got {other:?}"),
-        }
-
-        // The flat ParsedQuery view exposes the φ via the existing
-        // multi-quantile machinery.
-        let pq = pq(r#"histogram_quantile(0.99, rate(http_requests_bucket{le="0.5"}[5m]))"#);
-        assert_eq!(pq.aggregations, vec![AggType::Quantile]);
-        assert_eq!(pq.quantiles, vec![0.99]);
-        assert_eq!(pq.label_filters.get("le").map(String::as_str), Some("0.5"),);
-    }
-
-    // ── avg_over_time ─────────────────────────────────────────────────────────
-
-    #[test]
-    fn avg_over_time_is_exact() {
-        // `AggIntent::Avg` has no ASAP-tier sketch substitute
-        // (`capability_for` returns `None` — needs a cross-policy
-        // Sum+Count join) and ASAPController's own `asap-plan` treats it
-        // the same way (`pass_through_intents_stay_logical`), so `avg`
-        // routes through the exact/archive path like `Sum`/`Count`,
-        // not the p50-quantile-sketch approximation this used to be.
-        let pq = pq("avg by (symbol) (avg_over_time(financial_last_trade_price[5m]))");
-        assert_eq!(pq.aggregations, Vec::<AggType>::new());
-        assert!(pq.quantiles.is_empty());
-        assert!(pq.exact_required);
-    }
-
-    // ── min/max_over_time ─────────────────────────────────────────────────────
-
-    #[test]
-    fn min_over_time_with_by_is_ddsketch() {
-        let pq = pq("min by (symbol) (min_over_time(financial_last_trade_price[5m]))");
-        assert_eq!(pq.aggregations, vec![AggType::Quantile]);
-        assert_eq!(pq.quantiles, vec![0.0]);
-    }
-
-    #[test]
-    fn max_over_time_with_by_is_ddsketch() {
-        let pq = pq("max by (symbol) (max_over_time(financial_last_trade_price[5m]))");
-        assert_eq!(pq.aggregations, vec![AggType::Quantile]);
-        assert_eq!(pq.quantiles, vec![1.0]);
-    }
-
-    // ── topk ──────────────────────────────────────────────────────────────────
-
-    #[test]
-    fn topk_count_over_time() {
-        let pq = pq("topk by (symbol) (10, count_over_time(financial_last_trade_price[5m]))");
-        assert_eq!(pq.aggregations, vec![AggType::Frequency]);
-        assert_eq!(pq.group_by_labels, vec!["symbol"]);
-    }
-
-    #[test]
-    fn topk_avg_over_time() {
-        // `topk` ranking by a non-count measure (`avg_over_time`, here)
-        // is not a heavy-hitter shape — `RankingMeasure::NonAdditive`,
-        // per `agg_intent::is_frequency_heavy_hitter` — so this becomes
-        // a generic `Sort + Limit` over `avg_over_time`'s own exact
-        // `Aggregate{Avg}`, not a forced `Count`/`Frequency` aggregate.
-        // Before the topk/rate precision fix this incorrectly asserted
-        // `[Frequency]`; `avg_over_time` itself is exact (see
-        // `avg_over_time_is_exact`), not a p50 quantile-sketch anymore.
-        let pq = pq("topk by (host) (5, avg_over_time(cpu[5m]))");
-        assert_eq!(pq.aggregations, Vec::<AggType>::new());
-        assert!(pq.exact_required);
-    }
-
-    #[test]
-    fn topk_avg_over_time_is_sort_limit_not_topk_node() {
-        // Structural check backing `topk_avg_over_time` above: the tree
-        // must be `Limit { Sort { Aggregate{Quantile} } }`, not
-        // `QueryExpr::TopK` — confirms the non-heavy-hitter path is
-        // really taken, not just that the flattened `ParsedQuery`
-        // happens to read the same.
-        use crate::intent_algebra::relational::QueryExpr;
-
-        let qe = super::parse_promql_expr("topk by (host) (5, avg_over_time(cpu[5m]))")
-            .expect("parse should succeed");
-        match &qe {
-            QueryExpr::Limit { n, input, .. } => {
-                assert_eq!(*n, 5);
-                assert!(
-                    matches!(input.as_ref(), QueryExpr::Sort { .. }),
-                    "expected Sort under Limit, got {input:?}"
-                );
-            }
-            other => panic!("expected Limit{{Sort{{...}}}}, got {other:?}"),
-        }
-    }
-
-    #[test]
-    fn bottomk_count_over_time_is_never_heavy_hitter() {
-        // `bottomk` never qualifies as the heavy-hitter `TopK` intent
-        // even when ranking by `count_over_time` — `descending` must
-        // also hold (`is_frequency_heavy_hitter`), and `bottomk` is
-        // ascending by definition. Must still lower to `Limit{Sort{...}}`.
-        use crate::intent_algebra::relational::QueryExpr;
-
-        let qe = super::parse_promql_expr("bottomk(5, count_over_time(http_requests_total[5m]))")
-            .expect("parse should succeed");
-        assert!(
-            !matches!(qe, QueryExpr::TopK { .. }),
-            "bottomk must never produce the heavy-hitter TopK node, got {qe:?}"
-        );
-        match &qe {
-            QueryExpr::Limit { n, input, .. } => {
-                assert_eq!(*n, 5);
-                match input.as_ref() {
-                    QueryExpr::Sort { keys, .. } => {
-                        assert!(keys[0].ascending, "bottomk must sort ascending");
-                    }
-                    other => panic!("expected Sort under Limit, got {other:?}"),
-                }
-            }
-            other => panic!("expected Limit{{Sort{{...}}}}, got {other:?}"),
-        }
-    }
-
-    #[test]
-    fn topk_count_over_time_is_topk_node() {
-        // The one real heavy-hitter shape: `topk` (descending) ranking
-        // by `count_over_time` (`RankingMeasure::Frequency`).
-        use crate::intent_algebra::relational::QueryExpr;
-
-        let qe = super::parse_promql_expr("topk(5, count_over_time(http_requests_total[5m]))")
-            .expect("parse should succeed");
-        assert!(
-            matches!(qe, QueryExpr::TopK { k: 5, .. }),
-            "expected QueryExpr::TopK{{k: 5, ..}}, got {qe:?}"
-        );
-    }
-
-    // ── count cardinality ─────────────────────────────────────────────────────
-
-    #[test]
-    fn count_count_over_time_is_hll() {
-        let pq = pq("count by (symbol) (count_over_time(financial_last_trade_price[5m]))");
-        assert_eq!(pq.aggregations, vec![AggType::Cardinality]);
-    }
-
-    // ── stddev_over_time ──────────────────────────────────────────────────────
-
-    #[test]
-    fn stddev_over_time_is_exact() {
-        // `AggIntent::StdDev` has no ASAP-tier sketch substitute
-        // (`capability_for` returns `None`, same as `Avg`), so
-        // `stddev_over_time` routes exact -- no more IQR-proxy
-        // `[q(0.25), q(0.75)]` approximation.
-        let pq = pq("avg by (host) (stddev_over_time(cpu[5m]))");
-        assert_eq!(pq.aggregations, Vec::<AggType>::new());
-        assert!(pq.quantiles.is_empty());
-        assert!(pq.exact_required);
-    }
-
-    // ── sum_over_time → exact ─────────────────────────────────────────────────
-
-    #[test]
-    fn sum_over_time_exact() {
-        let pq = pq("sum by (service) (sum_over_time(request_bytes[1h]))");
-        assert!(pq.exact_required);
-    }
-
-    // ── label filters ─────────────────────────────────────────────────────────
-
-    #[test]
-    fn label_eq_filter() {
-        let pq = pq(r#"sum by (service) (count_over_time(hits{env="prod"}[5m]))"#);
-        assert_eq!(
-            pq.label_filters.get("env").map(String::as_str),
-            Some("prod")
-        );
-    }
-
-    // ── duration parsing ──────────────────────────────────────────────────────
-
-    #[test]
-    fn duration_1h() {
-        let pq = pq("avg by (host) (avg_over_time(cpu[1h]))");
-        assert_eq!(pq.time_window, Duration::from_secs(3600));
-    }
-
-    // ── DEBS hints ────────────────────────────────────────────────────────────
-
-    #[test]
-    fn debs_price_stats_min() {
-        use super::super::QueryHint;
-        let pq = pq("min by (symbol) (min_over_time(financial_last_trade_price[5m]))");
-        assert!(matches!(pq.hint, Some(QueryHint::DebsPriceStats)));
-    }
-
-    #[test]
-    fn debs_cardinality() {
-        use super::super::QueryHint;
-        let pq = pq("count by (symbol) (count_over_time(financial_last_trade_price[5m]))");
-        assert!(matches!(pq.hint, Some(QueryHint::DebsCardinality)));
-    }
-
-    // ── Complex queries ───────────────────────────────────────────────────────
-
-    #[test]
-    fn complex_topk_count_over_time_multi_label() {
-        // topk absorbs CountSketch (R8); multiple label filters extracted
-        let pq = pq(
-            r#"topk by (service) (10, count_over_time(http_requests_total{status="500",env="prod"}[5m]))"#,
-        );
-        assert_eq!(pq.aggregations, vec![AggType::Frequency]);
-        assert_eq!(pq.group_by_labels, vec!["service"]);
-        assert_eq!(
-            pq.label_filters.get("status").map(String::as_str),
-            Some("500")
-        );
-        assert_eq!(
-            pq.label_filters.get("env").map(String::as_str),
-            Some("prod")
-        );
-        assert_eq!(pq.time_window, Duration::from_secs(300));
-    }
-
-    #[test]
-    fn complex_histogram_quantile_multi_label() {
-        // histogram_quantile wraps rate → DDSketch; two label selectors
-        let pq = pq(
-            r#"histogram_quantile(0.99, rate(request_duration_seconds_bucket{service="checkout",region="us-east"}[10m]))"#,
-        );
-        assert_eq!(pq.aggregations, vec![AggType::Quantile]);
-        assert_eq!(pq.quantiles, vec![0.99]);
-        assert_eq!(
-            pq.label_filters.get("service").map(String::as_str),
-            Some("checkout")
-        );
-        assert_eq!(
-            pq.label_filters.get("region").map(String::as_str),
-            Some("us-east")
-        );
-        assert_eq!(pq.time_window, Duration::from_secs(600));
-    }
-}
diff --git a/control_plane/src/sketch_algebra/tests.rs b/control_plane/src/sketch_algebra/tests.rs
index 4b060b30..b2a679d7 100644
--- a/control_plane/src/sketch_algebra/tests.rs
+++ b/control_plane/src/sketch_algebra/tests.rs
@@ -655,7 +655,7 @@ fn phase_b_pattern_archive_only_routes_to_archive() {
 /// → L3 canonical via `lower`); `bind_query_expr` is the
 /// L3→L4 bottom-up walk under the supplied accuracy target.
 fn pipeline_l1_to_l4(query: &str, accuracy: AccuracyTarget) -> PhysicalExpr {
-    let qe = crate::query_parser::parse_query_expr_canonical(query)
+    let qe = crate::query_parser::parse_query_expr_canonical(query, accuracy.clone())
         .unwrap_or_else(|e| panic!("parse {query}: {e}"));
     bind_query_expr(&qe, accuracy).unwrap_or_else(|e| panic!("bind {query}: {e}"))
 }
diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs
index 94d32a94..b3f40d0d 100644
--- a/data_plane/src/query_engines/asap_query_engine/engine.rs
+++ b/data_plane/src/query_engines/asap_query_engine/engine.rs
@@ -698,8 +698,15 @@ impl ASAPQueryEngine {
         // since there's no legacy answer left to diff a live-served
         // result against.
         let mut any_live_served = false;
+        // Resilience fix -- see the instant-query `execute(&str)` path's
+        // identical comment above `for candidate in &analysis.candidates`
+        // for the full rationale (design-target-architecture.md Part B).
+        let mut last_miss_detail: Option<String> = None;
 
         for candidate in &analysis.candidates {
+            if combined_result.is_some() {
+                continue;
+            }
             // Resolve candidate → {sids} via the sid catalog. Schema-
             // retirement #5: prefer `instances_matching` over the
             // policy-fp reverse index — it's the more general
@@ -729,14 +736,12 @@ impl ASAPQueryEngine {
             }
             sids.extend(idx.instances_matching(&candidate.metric_name, &candidate.group_by_keys));
             if sids.is_empty() {
-                return Err(crate::query_engines::EngineError::capability_miss(
-                    crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(),
-                    format!(
-                        "SketchStore has no policy for metric `{}` satisfying \
-                         capability {:?} — failing over to archive",
-                        candidate.metric_name, candidate.required_capability,
-                    ),
+                last_miss_detail = Some(format!(
+                    "SketchStore has no policy for metric `{}` satisfying \
+                     capability {:?} — failing over to archive",
+                    candidate.metric_name, candidate.required_capability,
                 ));
+                continue;
             }
 
             let required: crate::storage_engines::sketch_db::index::Capability =
@@ -760,14 +765,12 @@ impl ASAPQueryEngine {
                 }
             }
             if hit_sids.is_empty() {
-                return Err(crate::query_engines::EngineError::capability_miss(
-                    crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(),
-                    format!(
-                        "SketchStore has no sid satisfying capability {:?} for \
-                         metric `{}` — failing over to archive",
-                        candidate.required_capability, candidate.metric_name
-                    ),
+                last_miss_detail = Some(format!(
+                    "SketchStore has no sid satisfying capability {:?} for \
+                     metric `{}` — failing over to archive",
+                    candidate.required_capability, candidate.metric_name
                 ));
+                continue;
             }
 
             // ExactAgg capability → dispatch the per-(group_by_keys)
@@ -818,65 +821,58 @@ impl ASAPQueryEngine {
                             | crate::storage_engines::sketch_db::data::AggregationType::MultipleIncrease
                     );
                         if is_exact_sum_family && candidate.outer_fn == OuterFn::SumOverTime {
-                            return Err(crate::query_engines::EngineError::capability_miss(
-                                crate::storage_engines::types::StorageBackend::SketchStore
-                                    .data_source_id(),
-                                format!(
-                                    "SketchStore cannot answer `sum_over_time` over counter \
+                            last_miss_detail = Some(format!(
+                                "SketchStore cannot answer `sum_over_time` over counter \
                                  deltas for `{query}` (issue #301) — failing over to archive"
-                                ),
                             ));
+                            continue;
                         }
                         let use_rate_path = candidate.range_seconds > 0
                             && candidate.outer_fn == OuterFn::Rate
                             && is_exact_sum_family;
                         if use_rate_path {
-                            reducer
-                            .evaluate_exact_agg_rate(
+                            match reducer.evaluate_exact_agg_rate(
                                 &hit_sids,
                                 *agg_type,
                                 &candidate.group_by_keys,
                                 candidate.range_seconds,
                                 start_ms,
                                 end_ms,
-                            )
-                            .map_err(|e| {
-                                crate::query_engines::EngineError::capability_miss(
-                                    crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(),
-                                    format!(
+                            ) {
+                                Ok(r) => r,
+                                Err(e) => {
+                                    last_miss_detail = Some(format!(
                                         "SketchStore exact-agg rate reducer failed for `{query}` over \
                                          [{start_ms}, {end_ms}]: {e:?} — failing over to archive"
-                                    ),
-                                )
-                            })?
+                                    ));
+                                    continue;
+                                }
+                            }
                         } else {
-                            reducer
-                                .evaluate_exact_agg(
-                                    &hit_sids,
-                                    *agg_type,
-                                    &candidate.group_by_keys,
-                                    start_ms,
-                                    end_ms,
-                                    false,
-                                )
-                                .map_err(|e| {
-                                    crate::query_engines::EngineError::capability_miss(
-                                        crate::storage_engines::types::StorageBackend::SketchStore
-                                            .data_source_id(),
-                                        format!(
+                            match reducer.evaluate_exact_agg(
+                                &hit_sids,
+                                *agg_type,
+                                &candidate.group_by_keys,
+                                start_ms,
+                                end_ms,
+                                false,
+                            ) {
+                                Ok(r) => r,
+                                Err(e) => {
+                                    last_miss_detail = Some(format!(
                                         "SketchStore exact-agg reducer failed for `{query}` over \
                                          [{start_ms}, {end_ms}]: {e:?} — failing over to archive"
-                                    ),
-                                    )
-                                })?
+                                    ));
+                                    continue;
+                                }
+                            }
                         }
                     }
                     // P2-4 (typed dispatch): route off the analyzer's typed
                     // `required_capability` via `evaluate_for_capability`
                     // instead of round-tripping it through a function-name
                     // string the reducer re-parses.
-                    _ => reducer
-                        .evaluate_for_capability(
+                    _ => match reducer.evaluate_for_capability(
                             &candidate.required_capability,
                             &hit_sids,
                             &candidate.function_args,
@@ -889,17 +885,16 @@ impl ASAPQueryEngine {
                             effective_is_cumulative(candidate),
                             start_ms,
                             end_ms,
-                        )
-                        .map_err(|e| {
-                            crate::query_engines::EngineError::capability_miss(
-                                crate::storage_engines::types::StorageBackend::SketchStore
-                                    .data_source_id(),
-                                format!(
+                        ) {
+                            Ok(r) => r,
+                            Err(e) => {
+                                last_miss_detail = Some(format!(
                                     "SketchStore reducer failed for `{query}` over \
-                                 [{start_ms}, {end_ms}]: {e:?} — failing over to archive"
-                                ),
-                            )
-                        })?,
+                                     [{start_ms}, {end_ms}]: {e:?} — failing over to archive"
+                                ));
+                                continue;
+                            }
+                        },
                 },
             };
             // Apply the analyzer's typed outer-aggregation operator on
@@ -924,7 +919,9 @@ impl ASAPQueryEngine {
         let result = combined_result.ok_or_else(|| {
             crate::query_engines::EngineError::capability_miss(
                 crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(),
-                format!("SketchStore reducer produced no result for `{query}`"),
+                last_miss_detail.unwrap_or_else(|| {
+                    format!("SketchStore reducer produced no result for `{query}`")
+                }),
             )
         })?;
 
@@ -1445,7 +1442,29 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu
             let streaming_snap = self.streaming_config_snapshot();
             let policy_registry = streaming_snap.policy_registry();
 
+            // Resilience fix (design-target-architecture.md Part B,
+            // completing the analyzer-side fix in
+            // `asap_tier_analysis::analyze_promql_for_asap_tier`):
+            // `lower_promql` genuinely produces multiple independent
+            // candidates for composed queries (e.g. `max by (zone)
+            // (quantile_over_time(...))` -> an outer `ExactAgg(MinMax)` +
+            // an inner `QuantileApprox`), where the retired local parser
+            // fused these into one shape. This loop used to hard-fail the
+            // whole query the moment ANY candidate had no matching sid --
+            // fine when there was always exactly one candidate, wrong now
+            // that a later candidate may still answer the query. Capacity-
+            // miss points below `continue` to the next candidate instead
+            // of returning immediately; genuine reducer/decode errors
+            // (not capability misses) still fail hard, unchanged. Once
+            // one candidate succeeds, skip the rest (first-success-wins,
+            // not "last write wins" -- multi-candidate result *folding*
+            // remains explicitly deferred future work per this loop's own
+            // pre-existing comment above).
+            let mut last_miss_detail: Option<String> = None;
             for candidate in &analysis.candidates {
+                if combined_result.is_some() {
+                    continue;
+                }
                 if candidate.range_seconds > 0 {
                     any_range_candidate = true;
                 }
@@ -1482,17 +1501,15 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu
                         &self.control_plane_client,
                         &req,
                     );
-                    return Err(crate::query_engines::EngineError::capability_miss(
-                        crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(),
-                        format!(
-                            "SketchStore has no policy for metric `{}` \
-                             with group_by_keys ⊇ {:?} satisfying capability \
-                             {:?} — failing over to archive",
-                            candidate.metric_name,
-                            candidate.group_by_keys,
-                            candidate.required_capability,
-                        ),
+                    last_miss_detail = Some(format!(
+                        "SketchStore has no policy for metric `{}` \
+                         with group_by_keys ⊇ {:?} satisfying capability \
+                         {:?} — failing over to archive",
+                        candidate.metric_name,
+                        candidate.group_by_keys,
+                        candidate.required_capability,
                     ));
+                    continue;
                 }
 
                 // Verify each sid carries the analyzer's required
@@ -1589,15 +1606,12 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu
                             &self.control_plane_client,
                             &req,
                         );
-                        return Err(crate::query_engines::EngineError::capability_miss(
-                            crate::storage_engines::types::StorageBackend::SketchStore
-                                .data_source_id(),
-                            format!(
-                                "SketchStore has no sid satisfying capability \
-                                 {:?} for metric `{}` — failing over to archive",
-                                candidate.required_capability, candidate.metric_name
-                            ),
+                        last_miss_detail = Some(format!(
+                            "SketchStore has no sid satisfying capability \
+                             {:?} for metric `{}` — failing over to archive",
+                            candidate.required_capability, candidate.metric_name
                         ));
+                        continue;
                     }
                 }
 
@@ -1666,18 +1680,15 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu
                             &self.control_plane_client,
                             &req,
                         );
-                        return Err(crate::query_engines::EngineError::capability_miss(
-                            crate::storage_engines::types::StorageBackend::SketchStore
-                                .data_source_id(),
-                            format!(
-                                "SketchStore FrequencyEstimate sid for metric `{}` cannot \
-                                 answer the per-item selector `{}` (CMS/CountSketch return \
-                                 the per-window bucket TOTAL, not a string-keyed estimate) — \
-                                 failing over to archive rather than returning a misleading \
-                                 total",
-                                candidate.metric_name, candidate.spatial_filter_canonical
-                            ),
+                        last_miss_detail = Some(format!(
+                            "SketchStore FrequencyEstimate sid for metric `{}` cannot \
+                             answer the per-item selector `{}` (CMS/CountSketch return \
+                             the per-window bucket TOTAL, not a string-keyed estimate) — \
+                             failing over to archive rather than returning a misleading \
+                             total",
+                            candidate.metric_name, candidate.spatial_filter_canonical
                         ));
+                        continue;
                     }
                 }
 
@@ -1726,16 +1737,14 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu
                         &self.control_plane_client,
                         &req,
                     );
-                    return Err(crate::query_engines::EngineError::capability_miss(
-                        crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(),
-                        format!(
-                            "SketchStore cannot answer `sum_over_time` over counter \
-                             deltas for metric `{}` (issue #301: Σ-of-cumulative-samples \
-                             not reconstructable from per-window deltas) — failing over \
-                             to archive",
-                            candidate.metric_name
-                        ),
+                    last_miss_detail = Some(format!(
+                        "SketchStore cannot answer `sum_over_time` over counter \
+                         deltas for metric `{}` (issue #301: Σ-of-cumulative-samples \
+                         not reconstructable from per-window deltas) — failing over \
+                         to archive",
+                        candidate.metric_name
                     ));
+                    continue;
                 }
 
                 let use_rate_path = is_exact_sum_family && candidate.outer_fn == OuterFn::Rate;
@@ -1867,59 +1876,49 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu
                             name,
                         ),
                     ) => {
-                        return Err(crate::query_engines::EngineError::capability_miss(
-                            crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(),
-                            format!(
-                                "SketchStore reducer does not support function `{name}` \
-                                 — failing over to archive"
-                            ),
+                        last_miss_detail = Some(format!(
+                            "SketchStore reducer does not support function `{name}` \
+                             — failing over to archive"
                         ));
+                        continue;
                     }
                     Err(crate::storage_engines::sketch_db::query::ASAPTierError::UnsupportedCapability {
                         function,
                         capability}) => {
-                        return Err(crate::query_engines::EngineError::capability_miss(
-                            crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(),
-                            format!(
-                                "SketchStore reducer cannot answer `{function}` against \
-                                 capability {capability:?} — failing over to archive"
-                            ),
+                        last_miss_detail = Some(format!(
+                            "SketchStore reducer cannot answer `{function}` against \
+                             capability {capability:?} — failing over to archive"
                         ));
+                        continue;
                     }
                     Err(crate::storage_engines::sketch_db::query::ASAPTierError::DeserializeFailure {
                         sid,
                         encoding,
                         reason}) => {
-                        return Err(crate::query_engines::EngineError::capability_miss(
-                            crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(),
-                            format!(
-                                "SketchStore reducer failed to decode sketch for sid \
-                                 {sid} (encoding={encoding:?}): {reason} — failing over \
-                                 to archive"
-                            ),
+                        last_miss_detail = Some(format!(
+                            "SketchStore reducer failed to decode sketch for sid \
+                             {sid} (encoding={encoding:?}): {reason} — failing over \
+                             to archive"
                         ));
+                        continue;
                     }
                     Err(crate::storage_engines::sketch_db::query::ASAPTierError::NoData {
                         metric_name: m}) => {
-                        return Err(crate::query_engines::EngineError::capability_miss(
-                            crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(),
-                            format!(
-                                "SketchStore reducer found no samples for metric \
-                                 `{m}` in window — failing over to archive"
-                            ),
+                        last_miss_detail = Some(format!(
+                            "SketchStore reducer found no samples for metric \
+                             `{m}` in window — failing over to archive"
                         ));
+                        continue;
                     }
                     Err(crate::storage_engines::sketch_db::query::ASAPTierError::MissingHeap {
                         sid,
                         sketch_kind}) => {
-                        return Err(crate::query_engines::EngineError::capability_miss(
-                            crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(),
-                            format!(
-                                "SketchStore reducer cannot enumerate top-k for sid \
-                                 {sid} (sketch_kind={sketch_kind:?}, no heap) — \
-                                 failing over to archive"
-                            ),
+                        last_miss_detail = Some(format!(
+                            "SketchStore reducer cannot enumerate top-k for sid \
+                             {sid} (sketch_kind={sketch_kind:?}, no heap) — \
+                             failing over to archive"
                         ));
+                        continue;
                     }
                 }
                 };
@@ -2009,6 +2008,17 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu
                 }
                 return Ok(warm_qr);
             }
+            // Every candidate the analyzer produced was skipped above
+            // (resilience fix, design-target-architecture.md Part B) --
+            // none of them had a servable sid/reducer path. Surface the
+            // last-recorded miss reason rather than falling through to
+            // the unrelated no-sketch-index branch below.
+            return Err(crate::query_engines::EngineError::capability_miss(
+                crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(),
+                last_miss_detail.unwrap_or_else(|| {
+                    format!("SketchStore found no servable candidate for `{query}`")
+                }),
+            ));
         }
 
         // CRITICAL #4: no-sketch-index fallback. The legacy
@@ -2777,16 +2787,19 @@ mod asap_tier_classify_tests {
 
     #[tokio::test]
     async fn execute_bare_selector_falls_over_to_archive() {
-        // A bare vector selector lowers (via
-        // `control_plane::asap_tier_analysis::analyze_promql_for_asap_tier`)
-        // to an `ExactAgg(Sum)` candidate — the control plane no longer
-        // rejects it outright with `NoCallNodeFound`. But the
-        // `SketchStore` here holds only a DDSketch (quantile) policy, so
-        // the candidate's `ExactAgg(Sum)` capability finds no matching
-        // policy and the query still fails over to the archive engine
-        // via `CapabilityMiss` — just with a capability-mismatch detail
-        // rather than an analyzer-shape rejection. Either way the
-        // routing outcome (→ archive) is unchanged.
+        // L1 adoption (design-target-architecture.md Part B), accepted
+        // behavior change: `lower_promql` doesn't implicitly wrap a bare
+        // selector in `Aggregate { Sum }` the way the retired local
+        // parser did (see `control_plane`'s
+        // `asap_tier_analysis::bare_selector_is_no_longer_asap_tier_answerable`),
+        // so this rejects with `NoCallNodeFound` again -- a DIFFERENT
+        // reason than the (now-stale) comment this replaced expected, but
+        // the routing OUTCOME is unchanged either way: capability-miss,
+        // fails over to archive. The `SketchStore` here holding only a
+        // DDSketch (quantile) policy is now moot for this specific query
+        // (rejected before ever reaching policy lookup), kept for the
+        // fixture's own sake / in case the bare-selector shape changes
+        // again.
         let idx = Arc::new(SketchStore::new());
         idx.register(dd_meta(2, "http_latency_ms", &["zone"]));
         idx.append_sample(
@@ -2804,7 +2817,9 @@ mod asap_tier_classify_tests {
         match result {
             Err(EngineError::CapabilityMiss { detail, .. }) => {
                 assert!(
-                    detail.contains("ExactAgg(Sum)") || detail.contains("no policy"),
+                    detail.contains("ExactAgg(Sum)")
+                        || detail.contains("no policy")
+                        || detail.contains("NoCallNodeFound"),
                     "expected a capability-miss fall-over to archive: {detail}"
                 );
             }
@@ -3750,12 +3765,17 @@ mod asap_tier_classify_tests {
         let sot = analyze_promql_for_asap_tier("sum_over_time(http_requests_total[5m])");
         let sum_by_rate =
             analyze_promql_for_asap_tier("sum by (zone) (rate(http_requests_total[5m]))");
+        // L1 adoption (design-target-architecture.md Part B), accepted
+        // behavior change: a bare selector no longer lowers to an
+        // implicit `Aggregate { Sum }` (see control_plane's
+        // `asap_tier_analysis::bare_selector_is_no_longer_asap_tier_answerable`),
+        // so it's no longer part of this test's comparison set.
         let bare = analyze_promql_for_asap_tier("http_requests_total");
 
         assert!(rate.unsupported.is_none() && !rate.candidates.is_empty());
         assert!(sot.unsupported.is_none() && !sot.candidates.is_empty());
         assert!(sum_by_rate.unsupported.is_none() && !sum_by_rate.candidates.is_empty());
-        assert!(bare.unsupported.is_none() && !bare.candidates.is_empty());
+        assert!(bare.candidates.is_empty(), "{bare:?}");
 
         // Whichever query has a `rate(...)` call ANYWHERE in its tree
         // (bare `rate(...)` or composed `sum by (...) (rate(...))`) binds
@@ -3767,23 +3787,29 @@ mod asap_tier_classify_tests {
         // why a Sum-registered sid still answers it). This is a real,
         // intentional behavior change from the Phase 2 semantic retarget
         // (Rate/Increase used to collapse onto AggIntent::Sum) -- not a
-        // stale assertion left over from before it. `sot`/`bare` have no
-        // `rate(...)` anywhere and stay `ExactAgg(Sum)`.
+        // stale assertion left over from before it. `sot` has no
+        // `rate(...)` anywhere and stays `ExactAgg(Sum)`.
+        //
+        // L1 adoption: `sum_by_rate` is now TWO candidates (the outer
+        // `sum by (zone)` reduction + the inner `rate(...)`), not one
+        // fused shape -- find the `Increase` one rather than assuming
+        // index 0.
         assert_eq!(
             rate.candidates[0].required_capability,
             Capability::ExactAgg(AggregationType::Increase),
         );
-        assert_eq!(
-            rate.candidates[0].required_capability, sum_by_rate.candidates[0].required_capability,
-            "composed `sum by (...) (rate(...))` binds the same Rate \
-             AggIntent as bare `rate(...)`",
-        );
-        assert_eq!(
-            sot.candidates[0].required_capability,
-            bare.candidates[0].required_capability,
+        assert!(
+            sum_by_rate
+                .candidates
+                .iter()
+                .any(|c| c.required_capability == Capability::ExactAgg(AggregationType::Increase)),
+            "composed `sum by (...) (rate(...))` must include the same Rate \
+             AggIntent as bare `rate(...)`: {sum_by_rate:?}",
         );
 
-        // `outer_fn` carries the counter-function distinction (#301).
+        // `outer_fn` carries the counter-function distinction (#301) --
+        // shared trace context across every candidate from one analysis
+        // call, so index 0 is fine here regardless of candidate count.
         assert_eq!(rate.candidates[0].outer_fn, OuterFn::Rate);
         assert_eq!(sot.candidates[0].outer_fn, OuterFn::SumOverTime);
         assert_eq!(
@@ -3792,7 +3818,6 @@ mod asap_tier_classify_tests {
             "composed `sum by (...) (rate(...))` MUST flag OuterFn::Rate \
              even though the outer function name is `sum`"
         );
-        assert_eq!(bare.candidates[0].outer_fn, OuterFn::Plain);
     }
 
     /// `sum by (zone) (rate(http_requests_total[5m]))` end-to-end.
diff --git a/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs b/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs
index 035a254c..0dd5e6cc 100644
--- a/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs
+++ b/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs
@@ -97,7 +97,7 @@ pub fn lower_promql_to_l4node(
         return Err(LoweringSkip::RateShape);
     }
 
-    let qe = control_plane::query_parser::parse_query_expr_canonical(query)
+    let qe = control_plane::query_parser::parse_query_expr_canonical(query, accuracy.clone())
         .map_err(|e| LoweringSkip::ParseFailed(e.to_string()))?;
 
     let physical = control_plane::sketch_algebra::bind_query_expr(&qe, accuracy)
@@ -151,16 +151,21 @@ mod tests {
     }
 
     #[test]
-    fn bare_selector_realizes_to_a_summary_agg() {
-        // Mirrors `implement_promql_for_asap_tier`'s own
-        // `bare_selector_implements_to_an_exact_sum_agg` test -- a bare
-        // selector is `Aggregate { Sum }` over the sample value.
-        let node = lower_promql_to_l4node("http_requests_total", accuracy())
-            .expect("bare selector should realize");
+    fn bare_selector_is_not_realized() {
+        // L1 adoption (design-target-architecture.md Part B), accepted
+        // behavior change: `asap_frontend_promql::lower_promql` no longer
+        // wraps a bare selector in an implicit `Aggregate { Sum }` (see
+        // control_plane's
+        // `asap_tier_implement::bare_selector_has_no_aggregate_root_to_implement`
+        // and `asap_tier_analysis::bare_selector_is_no_longer_asap_tier_answerable`
+        // for the sibling fixes). With no `Aggregate` node anywhere in the
+        // tree, `implement_tree_in_with` has nothing to bind and the whole
+        // expression stays one opaque `Logical` blob, which this module
+        // surfaces as `NotRealized`.
+        let result = lower_promql_to_l4node("http_requests_total", accuracy());
         assert!(
-            matches!(node.expr, SummaryExpr::SummaryAgg { .. }),
-            "expected SummaryAgg, got {:?}",
-            node.expr
+            matches!(result, Err(LoweringSkip::NotRealized)),
+            "expected NotRealized, got {result:?}"
         );
     }