diff --git a/docs/rqe_optimizer_TODO.md b/docs/rqe_optimizer_TODO.md
index 348ab3c1..c8d92af0 100644
--- a/docs/rqe_optimizer_TODO.md
+++ b/docs/rqe_optimizer_TODO.md
@@ -5,12 +5,12 @@ implemented, what is next, and what is intentionally deferred.
 
 ## Done in v1
 
-- Sliding deployments use `(capability, configuration, metric, G, x, y)`,
+- Sliding deployments use `(capability, configuration, metric, spatial_filter, G, x, y)`,
   where `x` is window size and `y` is slide.
 - Workload facts are per metric: full label set, scrape interval, and
   cardinality per label set. The arrival rate is derived from them, and
   `validate_facts` reports bad facts up front.
-- Eligibility enforces capability, metric, grouping, alignment, and the RQE's
+- Eligibility enforces capability, metric, spatial filter, grouping, alignment, and the RQE's
   measured `accuracy_metric` constraint.
 - Candidate generation uses legal windows and subset gcds of query intervals,
   keeping only multiples of the scrape interval. A deployment may serve
diff --git a/rqe-optimizer/examples/small_problem.rs b/rqe-optimizer/examples/small_problem.rs
index 0bfec499..83b3737a 100644
--- a/rqe-optimizer/examples/small_problem.rs
+++ b/rqe-optimizer/examples/small_problem.rs
@@ -9,9 +9,9 @@
 //! no translation layer, since `rqe_optimizer::Deployment` embeds
 //! `AtomicCostEntry` directly.
 //!
-//! Every other number this example uses (RQE definitions, metric labels,
+//! Every other number this example uses (RAQE definitions, metric labels,
 //! cardinalities and scrape intervals) lives in the two tables below
-//! (`facts`, `rqes`) and nowhere else in this file.
+//! (`facts`, `raqes`) and nowhere else in this file.
 //!
 //! Run: `scripts/export_rqe_optimizer_costs.sh` once. Do not run this example
 //! with no mode flag on a large workload: eager mode retains every feasible
@@ -23,8 +23,10 @@
 //! `--print-first N` to display example mappings. `--milp` minimizes
 //! `w_cpu · CPU + w_mem · memory GiB` without enumerating mappings; set the
 //! weights with `--w-cpu X --w-mem Y` (default 1 and 0). Repeat
-//! `--latency-limit RQE_ID=SECONDS` to impose MILP latency bounds.
+//! `--latency-sla RAQE_ID=MS` to impose MILP latency SLAs.
 //! `--sample-mappings N` prints N feasible mappings and exits.
+//! `--allow-undeployable-families` also plans with measured families
+//! ASAPQuery can't deploy (dd, univmon-*, countsketch-heap).
 
 use std::collections::BTreeSet;
 use std::time::Instant;
@@ -34,16 +36,16 @@ use rqe_optimizer::candidates::{
     build_all_candidates, build_all_candidates_unpruned, eligible_deployments_for,
 };
 use rqe_optimizer::enumerate::{brute_force, for_each_mapping, for_each_mapping_while, unservable};
-use rqe_optimizer::milp::{minimize, MilpBounds, Objective};
+use rqe_optimizer::milp::{minimize, Objective};
 use rqe_optimizer::pareto::{pareto_front, ParetoFront};
 use rqe_optimizer::{
-    validate_facts, AccuracyDirection, AtomicCostTable, Capability, LabelSet, MetricFacts, Rqe,
+    validate_facts, AccuracyDirection, AtomicCostTable, Capability, LabelSet, MetricFacts, Raqe,
     WorkloadFacts,
 };
 
 /// Accuracy metric keys the real comparators actually report (checked
 /// against `aqpbm-core/src/accuracy/{aggregate,quantile,cardinality}.rs`).
-/// These three are lower-is-better errors; each RQE below pairs its metric with the matching
+/// These three are lower-is-better errors; each RAQE below pairs its metric with the matching
 /// [`AccuracyDirection`] so the comparison can't be applied the wrong way
 /// round.
 ///
@@ -81,7 +83,7 @@ fn load_cost_table() -> AtomicCostTable {
 fn facts() -> WorkloadFacts {
     let http_metric_facts = || MetricFacts {
         labels: label_set(&["service", "endpoint", "pod"]),
-        scrape_interval_secs: 15,
+        scrape_interval_ms: 15_000,
         cardinality: [
             (label_set(&["service"]), 5),
             (label_set(&["service", "endpoint"]), 50),
@@ -96,103 +98,119 @@ fn facts() -> WorkloadFacts {
     .into()
 }
 
-fn rqes() -> Vec<Rqe> {
+fn raqes() -> Vec<Raqe> {
     let se = label_set(&["service", "endpoint"]);
     let s = label_set(&["service"]);
     let requests = || REQUESTS.to_string();
     let duration = || DURATION.to_string();
     vec![
-        Rqe {
+        Raqe {
             id: "req_rate_1h".to_string(),
             capability: Capability::RateOrIncrease,
-            lookback_secs: 3_600,
-            interval_secs: 60,
+            lookback_ms: 3_600_000,
+            interval_ms: 60_000,
             metric: requests(),
+            spatial_filter: String::new(),
             grouping_labels: se.clone(),
             accuracy_metric: RATE_ERR.to_string(),
-            accuracy_tolerance: 0.1,
+            accuracy_sla: 0.1,
             accuracy_direction: AccuracyDirection::LowerIsBetter,
+            latency_sla_ms: None,
         },
-        Rqe {
+        Raqe {
             id: "req_rate_1d".to_string(),
             capability: Capability::RateOrIncrease,
-            lookback_secs: 86_400,
-            interval_secs: 60,
+            lookback_ms: 86_400_000,
+            interval_ms: 60_000,
             metric: requests(),
+            spatial_filter: String::new(),
             grouping_labels: se.clone(),
             accuracy_metric: RATE_ERR.to_string(),
-            accuracy_tolerance: 0.1,
+            accuracy_sla: 0.1,
             accuracy_direction: AccuracyDirection::LowerIsBetter,
+            latency_sla_ms: None,
         },
-        Rqe {
+        Raqe {
             id: "req_rate_5m_tick".to_string(),
             capability: Capability::RateOrIncrease,
-            lookback_secs: 3_600,
-            interval_secs: 300,
+            lookback_ms: 3_600_000,
+            interval_ms: 300_000,
             metric: requests(),
+            spatial_filter: String::new(),
             grouping_labels: se.clone(),
             accuracy_metric: RATE_ERR.to_string(),
-            accuracy_tolerance: 0.1,
+            accuracy_sla: 0.1,
             accuracy_direction: AccuracyDirection::LowerIsBetter,
+            latency_sla_ms: None,
         },
-        Rqe {
+        Raqe {
             id: "latency_p99_1h".to_string(),
             capability: Capability::Quantile,
-            lookback_secs: 3_600,
-            interval_secs: 60,
+            lookback_ms: 3_600_000,
+            interval_ms: 60_000,
             metric: duration(),
+            spatial_filter: String::new(),
             grouping_labels: se.clone(),
             accuracy_metric: RANK_ERR.to_string(),
-            accuracy_tolerance: 0.05,
+            accuracy_sla: 0.05,
             accuracy_direction: AccuracyDirection::LowerIsBetter,
+            latency_sla_ms: None,
         },
-        Rqe {
+        Raqe {
             id: "latency_p99_6h_tick".to_string(),
             capability: Capability::Quantile,
-            lookback_secs: 21_600,
-            interval_secs: 300,
+            lookback_ms: 21_600_000,
+            interval_ms: 300_000,
             metric: duration(),
+            spatial_filter: String::new(),
             grouping_labels: se.clone(),
             accuracy_metric: RANK_ERR.to_string(),
-            accuracy_tolerance: 0.05,
+            accuracy_sla: 0.05,
             accuracy_direction: AccuracyDirection::LowerIsBetter,
+            latency_sla_ms: None,
         },
-        Rqe {
+        Raqe {
             id: "latency_p99_1d".to_string(),
             capability: Capability::Quantile,
-            lookback_secs: 86_400,
-            interval_secs: 60,
+            lookback_ms: 86_400_000,
+            interval_ms: 60_000,
             metric: duration(),
+            spatial_filter: String::new(),
             grouping_labels: se.clone(),
             accuracy_metric: RANK_ERR.to_string(),
-            accuracy_tolerance: 0.05,
+            accuracy_sla: 0.05,
             accuracy_direction: AccuracyDirection::LowerIsBetter,
+            latency_sla_ms: None,
         },
-        Rqe {
+        Raqe {
             id: "distinct_services_1h".to_string(),
             capability: Capability::Cardinality,
-            lookback_secs: 3_600,
-            interval_secs: 60,
+            lookback_ms: 3_600_000,
+            interval_ms: 60_000,
             metric: requests(),
+            spatial_filter: String::new(),
             grouping_labels: s.clone(),
             accuracy_metric: CARDINALITY_ERR.to_string(),
-            accuracy_tolerance: 0.1,
+            accuracy_sla: 0.1,
             accuracy_direction: AccuracyDirection::LowerIsBetter,
+            latency_sla_ms: None,
         },
         // The higher-is-better case: 0.9 is a *floor* on precision@k, not a
         // ceiling on error. Served by cms-heap, whose measured precision is
         // 1.0 at both configs under the zipf workload (it was 0.0 at both
         // under uniform -- top-k is meaningless without real skew).
-        Rqe {
+        Raqe {
             id: "top_endpoints_1h".to_string(),
             capability: Capability::TopK,
-            lookback_secs: 3_600,
-            interval_secs: 60,
+            lookback_ms: 3_600_000,
+            interval_ms: 60_000,
             metric: requests(),
+            spatial_filter: String::new(),
             grouping_labels: se.clone(),
             accuracy_metric: TOPK_PRECISION.to_string(),
-            accuracy_tolerance: 0.9,
+            accuracy_sla: 0.9,
             accuracy_direction: AccuracyDirection::HigherIsBetter,
+            latency_sla_ms: None,
         },
     ]
 }
@@ -214,57 +232,65 @@ fn positive_integer_flag(name: &str, default: u64) -> u64 {
 fn print_mapping(
     mapping_number: u64,
     mapping: &[usize],
-    rqes: &[Rqe],
+    raqes: &[Raqe],
     deployments: &[rqe_optimizer::Deployment],
 ) {
     println!("mapping {mapping_number}:");
-    for (rqe, &deployment_index) in rqes.iter().zip(mapping) {
+    for (raqe, &deployment_index) in raqes.iter().zip(mapping) {
         let deployment = &deployments[deployment_index];
         println!(
-            "  {} -> {} {} (x={}s, y={}s)",
-            rqe.id,
+            "  {} -> {} {} (x={}ms, y={}ms)",
+            raqe.id,
             deployment.config.sketch,
             deployment.config.sketch_config,
-            deployment.window_secs,
-            deployment.slide_secs,
+            deployment.window_ms,
+            deployment.slide_ms,
         );
     }
 }
 
 fn print_candidate(candidate_number: usize, deployment: &rqe_optimizer::Deployment) {
     println!(
-        "candidate {candidate_number}: {} {} {} by {:?} (x={}s, y={}s)",
+        "candidate {candidate_number}: {} {} {} by {:?} (x={}ms, y={}ms)",
         deployment.config.sketch,
         deployment.config.sketch_config,
         deployment.metric,
         deployment.grouping_labels,
-        deployment.window_secs,
-        deployment.slide_secs,
+        deployment.window_ms,
+        deployment.slide_ms,
     );
 }
 
-fn latency_bounds(rqes: &[Rqe]) -> Vec<Option<f64>> {
+/// Only `--milp` enforces latency SLAs, so every other mode rejects the flag.
+/// `--candidates-only` and `--sample-mappings` run before the MILP branch.
+fn apply_latency_slas(raqes: &mut [Raqe]) {
     let args: Vec<_> = std::env::args().collect();
-    let mut bounds = vec![None; rqes.len()];
-    for pair in args.windows(2).filter(|pair| pair[0] == "--latency-limit") {
-        let (id, seconds) = pair[1]
+    let has = |flag: &str| args.iter().any(|arg| arg == flag);
+    assert!(
+        !has("--latency-sla")
+            || (has("--milp") && !has("--candidates-only") && !has("--sample-mappings")),
+        "--latency-sla is only enforced with --milp, without --candidates-only or --sample-mappings"
+    );
+    for pair in args.windows(2).filter(|pair| pair[0] == "--latency-sla") {
+        let (id, ms) = pair[1]
             .split_once('=')
-            .unwrap_or_else(|| panic!("--latency-limit expects RQE_ID=SECONDS"));
-        let seconds = seconds
+            .unwrap_or_else(|| panic!("--latency-sla expects RAQE_ID=MS"));
+        let ms = ms
             .parse::<f64>()
             .ok()
-            .filter(|value| *value > 0.0)
-            .unwrap_or_else(|| panic!("latency limit must be a positive number of seconds"));
-        let index = rqes
-            .iter()
-            .position(|rqe| rqe.id == id)
-            .unwrap_or_else(|| panic!("unknown RQE in --latency-limit: {id}"));
+            .filter(|value| value.is_finite() && *value > 0.0)
+            .unwrap_or_else(|| {
+                panic!("latency SLA must be a finite positive number of milliseconds")
+            });
+        let raqe = raqes
+            .iter_mut()
+            .find(|raqe| raqe.id == id)
+            .unwrap_or_else(|| panic!("unknown RAQE in --latency-sla: {id}"));
         assert!(
-            bounds[index].replace(seconds).is_none(),
-            "duplicate latency limit for {id}"
+            raqe.latency_sla_ms.replace(ms).is_none(),
+            "duplicate latency SLA for {id}"
         );
     }
-    bounds
 }
 
 fn weight_flag(name: &str, default: f64) -> f64 {
@@ -301,19 +327,22 @@ fn print_plan_cost(label: &str, plan_cost: &PlanCost) {
 
 fn main() {
     let candidates_only = std::env::args().any(|arg| arg == "--candidates-only");
-    let rqes = rqes();
+    let mut raqes = raqes();
+    apply_latency_slas(&mut raqes);
     let cost_table = load_cost_table();
     let facts = facts();
-    if let Err(problems) = validate_facts(&rqes, &facts) {
+    if let Err(problems) = validate_facts(&raqes, &facts) {
         panic!("invalid workload facts: {problems:#?}");
     }
 
-    let unpruned_count =
-        candidates_only.then(|| build_all_candidates_unpruned(&rqes, &cost_table, &facts).len());
-    let deployments = build_all_candidates(&rqes, &cost_table, &facts);
+    let allow_undeployable = std::env::args().any(|arg| arg == "--allow-undeployable-families");
+    let unpruned_count = candidates_only.then(|| {
+        build_all_candidates_unpruned(&raqes, &cost_table, &facts, allow_undeployable).len()
+    });
+    let deployments = build_all_candidates(&raqes, &cost_table, &facts, allow_undeployable);
     println!(
-        "{} RQEs, {} candidate deployments (from {} real cost-table rows)",
-        rqes.len(),
+        "{} RAQEs, {} candidate deployments (from {} real cost-table rows)",
+        raqes.len(),
         deployments.len(),
         cost_table.len(),
     );
@@ -325,9 +354,9 @@ fn main() {
             unpruned_count - deployments.len(),
         );
         let mut possible_mappings = Some(1_u64);
-        for rqe in &rqes {
-            let eligible = eligible_deployments_for(rqe, &deployments);
-            println!("  {}: {} eligible deployments", rqe.id, eligible.len());
+        for raqe in &raqes {
+            let eligible = eligible_deployments_for(raqe, &deployments);
+            println!("  {}: {} eligible deployments", raqe.id, eligible.len());
             possible_mappings = possible_mappings.and_then(|count| {
                 count.checked_mul(eligible.len().try_into().expect("usize fits in u64"))
             });
@@ -347,7 +376,7 @@ fn main() {
         return;
     }
 
-    let missing = unservable(&rqes, &deployments);
+    let missing = unservable(&raqes, &deployments);
     if !missing.is_empty() {
         println!("unservable (no eligible deployment): {missing:?}");
         return;
@@ -356,9 +385,9 @@ fn main() {
     let sample_mappings = positive_integer_flag("--sample-mappings", 0);
     if sample_mappings > 0 {
         let mut printed = 0;
-        let result = for_each_mapping_while(&rqes, &deployments, |mapping| {
+        let result = for_each_mapping_while(&raqes, &deployments, |mapping| {
             printed += 1;
-            print_mapping(printed, mapping, &rqes, &deployments);
+            print_mapping(printed, mapping, &raqes, &deployments);
             printed < sample_mappings
         });
         println!(
@@ -374,23 +403,20 @@ fn main() {
     }
 
     if std::env::args().any(|arg| arg == "--milp") {
-        let bounds = MilpBounds {
-            max_query_latency_secs: latency_bounds(&rqes),
-        };
         let objective = Objective::AUCCost {
             w_cpu: weight_flag("--w-cpu", 1.0),
             w_mem: weight_flag("--w-mem", 0.0),
         };
-        let solution = minimize(&rqes, &deployments, &facts, &bounds, objective)
+        let solution = minimize(&raqes, &deployments, &facts, objective)
             .expect("small_problem MILP should be feasible");
         println!(
             "MILP solution for {objective:?}: {:.3e}",
             objective.value(&solution.plan_cost)
         );
         print_plan_cost("  totals", &solution.plan_cost);
-        print_mapping(1, &solution.mapping, &rqes, &deployments);
-        for (rqe, latency) in rqes.iter().zip(&solution.plan_cost.query_latency_secs) {
-            println!("  {}: query_latency={latency:.3e} sec", rqe.id);
+        print_mapping(1, &solution.mapping, &raqes, &deployments);
+        for (raqe, latency) in raqes.iter().zip(&solution.plan_cost.query_latency_ms) {
+            println!("  {}: query_latency={latency:.3e} ms", raqe.id);
         }
         return;
     }
@@ -401,11 +427,11 @@ fn main() {
         let print_first = positive_integer_flag("--print-first", 0);
         let started = Instant::now();
         let mut processed = 0_u64;
-        let mapping_count = for_each_mapping(&rqes, &deployments, |mapping| {
-            front.consider(mapping, score(&rqes, &deployments, mapping, &facts));
+        let mapping_count = for_each_mapping(&raqes, &deployments, |mapping| {
+            front.consider(mapping, score(&raqes, &deployments, mapping, &facts));
             processed += 1;
             if processed <= print_first {
-                print_mapping(processed, mapping, &rqes, &deployments);
+                print_mapping(processed, mapping, &raqes, &deployments);
             }
             if processed.is_multiple_of(progress_every) {
                 let elapsed_secs = started.elapsed().as_secs_f64();
@@ -428,12 +454,12 @@ fn main() {
         "WARNING: eager mode will retain every feasible mapping. For larger workloads, use --milp, \
          --candidates-only, or --sample-mappings N instead."
     );
-    let mappings = brute_force(&rqes, &deployments);
+    let mappings = brute_force(&raqes, &deployments);
     println!("{} feasible full mappings", mappings.len());
 
     let plan_costs: Vec<_> = mappings
         .iter()
-        .map(|m| score(&rqes, &deployments, m, &facts))
+        .map(|m| score(&raqes, &deployments, m, &facts))
         .collect();
     let front = pareto_front(&plan_costs);
     println!("{} on the Pareto front\n", front.len());
@@ -447,9 +473,9 @@ fn main() {
             &format!("mapping {i}: {} deployments", distinct_deployments.len()),
             plan_cost,
         );
-        for (rqe, latency) in rqes.iter().zip(&plan_cost.query_latency_secs) {
-            let rqe_id = &rqe.id;
-            println!("    {rqe_id}: query_latency={latency:.3e} sec");
+        for (raqe, latency) in raqes.iter().zip(&plan_cost.query_latency_ms) {
+            let raqe_id = &raqe.id;
+            println!("    {raqe_id}: query_latency={latency:.3e} ms");
         }
     }
 }
diff --git a/rqe-optimizer/src/analytical_cost_model.rs b/rqe-optimizer/src/analytical_cost_model.rs
index 28161846..51f9ad7e 100644
--- a/rqe-optimizer/src/analytical_cost_model.rs
+++ b/rqe-optimizer/src/analytical_cost_model.rs
@@ -15,7 +15,7 @@
 // ponytail: a family with one fixed-size instance for all groups (HydraKLL,
 // sketch-bench#142) needs a per-family shape so it isn't multiplied by card(G).
 
-use crate::{Capability, Deployment, Mapping, Rqe, WorkloadFacts};
+use crate::{secs, Capability, Deployment, Mapping, Raqe, WorkloadFacts};
 use std::collections::{BTreeMap, BTreeSet};
 
 pub(crate) const BYTES_PER_GIB: f64 = (1u64 << 30) as f64;
@@ -27,7 +27,7 @@ pub const OUTPUT_BYTES_PER_VALUE: f64 = 8.0;
 pub const OUTPUT_BYTES_PER_TOPK_ENTRY: f64 = 16.0;
 /// Top-k entries per group: the heap size every top-k cost row was measured
 /// at (sketch-bench's `CMS_HEAP_TOP_K`).
-// ponytail: fixed k. A large k needs `k` on the RQE and a benchmark that
+// ponytail: fixed k. A large k needs `k` on the RAQE and a benchmark that
 // varies the heap, whose memory the export leaves out.
 pub const TOPK_ENTRIES: f64 = 32.0;
 
@@ -50,14 +50,14 @@ impl std::ops::AddAssign for PhaseCost {
 pub struct PlanCost {
     /// Paid once per active deployment.
     pub ingest: PhaseCost,
-    /// Paid per RQE; memory as if every query runs at once.
+    /// Paid per RAQE; memory as if every query runs at once.
     pub merge: PhaseCost,
-    /// Paid per RQE; memory as if every query runs at once.
+    /// Paid per RAQE; memory as if every query runs at once.
     pub query: PhaseCost,
     /// Paid once per active deployment, sized for its longest lookback.
     pub storage: PhaseCost,
-    /// Index-aligned with the input RQE slice; display IDs need not be unique.
-    pub query_latency_secs: Vec<f64>,
+    /// Index-aligned with the input RAQE slice; display IDs need not be unique.
+    pub query_latency_ms: Vec<f64>,
 }
 
 impl PlanCost {
@@ -88,9 +88,9 @@ fn open_window_count(deployment: &Deployment) -> f64 {
 }
 
 /// `L / x`: windows merged per query.
-fn merged_window_count(rqe: &Rqe, deployment: &Deployment) -> f64 {
+fn merged_window_count(raqe: &Raqe, deployment: &Deployment) -> f64 {
     deployment
-        .query_instance_count(rqe.lookback_secs)
+        .query_instance_count(raqe.lookback_ms)
         .expect("only eligible pairs are costed") as f64
 }
 
@@ -110,19 +110,19 @@ pub(crate) fn ingest(deployment: &Deployment, facts: &WorkloadFacts) -> PhaseCos
 /// CPU `card(G) · (L/x − 1) · c_mrg / T`; memory `card(G) · m`, one
 /// accumulator per group. A direct query (`L == x`) merges nothing and costs
 /// neither.
-pub(crate) fn merge(rqe: &Rqe, deployment: &Deployment, facts: &WorkloadFacts) -> PhaseCost {
+pub(crate) fn merge(raqe: &Raqe, deployment: &Deployment, facts: &WorkloadFacts) -> PhaseCost {
     let groups = group_count(deployment, facts);
-    let merges_per_group = merged_window_count(rqe, deployment) - 1.0;
+    let merges_per_group = merged_window_count(raqe, deployment) - 1.0;
     PhaseCost {
         cpu_secs_per_sec: groups * merges_per_group * deployment.config.merge_cpu_secs
-            / rqe.interval_secs as f64,
-        memory_bytes: groups * merge_memory_per_group(rqe, deployment),
+            / secs(raqe.interval_ms),
+        memory_bytes: groups * merge_memory_per_group(raqe, deployment),
     }
 }
 
 /// One accumulator if the query merges at all.
-pub(crate) fn merge_memory_per_group(rqe: &Rqe, deployment: &Deployment) -> f64 {
-    if merged_window_count(rqe, deployment) > 1.0 {
+pub(crate) fn merge_memory_per_group(raqe: &Raqe, deployment: &Deployment) -> f64 {
+    if merged_window_count(raqe, deployment) > 1.0 {
         deployment.config.mem_bytes_per_instance
     } else {
         0.0
@@ -130,43 +130,44 @@ pub(crate) fn merge_memory_per_group(rqe: &Rqe, deployment: &Deployment) -> f64
 }
 
 /// CPU `card(G) · c_qry / T`; memory `card(G) ·` output bytes per group.
-pub(crate) fn query(rqe: &Rqe, deployment: &Deployment, facts: &WorkloadFacts) -> PhaseCost {
+pub(crate) fn query(raqe: &Raqe, deployment: &Deployment, facts: &WorkloadFacts) -> PhaseCost {
     let groups = group_count(deployment, facts);
-    let output_bytes_per_group = match rqe.capability {
+    let output_bytes_per_group = match raqe.capability {
         Capability::TopK => TOPK_ENTRIES * OUTPUT_BYTES_PER_TOPK_ENTRY,
         _ => OUTPUT_BYTES_PER_VALUE,
     };
     PhaseCost {
-        cpu_secs_per_sec: groups * deployment.config.query_cpu_secs / rqe.interval_secs as f64,
+        cpu_secs_per_sec: groups * deployment.config.query_cpu_secs / secs(raqe.interval_ms),
         memory_bytes: groups * output_bytes_per_group,
     }
 }
 
-/// Closed windows kept to answer `rqe`: `card(G) · m · ((L − x)/y + 1)`.
-pub(crate) fn storage_bytes(rqe: &Rqe, deployment: &Deployment, facts: &WorkloadFacts) -> f64 {
+/// Closed windows kept to answer `raqe`: `card(G) · m · ((L − x)/y + 1)`.
+pub(crate) fn storage_bytes(raqe: &Raqe, deployment: &Deployment, facts: &WorkloadFacts) -> f64 {
     let closed_windows = deployment
-        .closed_instance_count(rqe.lookback_secs)
+        .closed_instance_count(raqe.lookback_ms)
         .expect("only eligible pairs are costed") as f64;
     group_count(deployment, facts) * deployment.config.mem_bytes_per_instance * closed_windows
 }
 
-/// Serial CPU time of one query: `card(G) · (c_qry + (L/x − 1) · c_mrg)`.
-pub(crate) fn query_latency_secs(rqe: &Rqe, deployment: &Deployment, facts: &WorkloadFacts) -> f64 {
-    let merges_per_group = merged_window_count(rqe, deployment) - 1.0;
-    group_count(deployment, facts)
+/// Serial CPU time of one query, in ms: `card(G) · (c_qry + (L/x − 1) · c_mrg)`.
+pub(crate) fn query_latency_ms(raqe: &Raqe, deployment: &Deployment, facts: &WorkloadFacts) -> f64 {
+    let merges_per_group = merged_window_count(raqe, deployment) - 1.0;
+    1000.0
+        * group_count(deployment, facts)
         * (deployment.config.query_cpu_secs + merges_per_group * deployment.config.merge_cpu_secs)
 }
 
 pub fn score(
-    rqes: &[Rqe],
+    raqes: &[Raqe],
     deployments: &[Deployment],
     mapping: &Mapping,
     facts: &WorkloadFacts,
 ) -> PlanCost {
     assert_eq!(
         mapping.len(),
-        rqes.len(),
-        "mapping must have one deployment per RQE"
+        raqes.len(),
+        "mapping must have one deployment per RAQE"
     );
     let active_deployments: BTreeSet<usize> = mapping.iter().copied().collect();
     let mut ingest_cost = PhaseCost::default();
@@ -177,18 +178,18 @@ pub fn score(
     let mut storage_bytes_by_deployment: BTreeMap<usize, f64> = BTreeMap::new();
     let mut merge_cost = PhaseCost::default();
     let mut query_cost = PhaseCost::default();
-    let query_latency_secs = rqes
+    let query_latency_ms = raqes
         .iter()
         .zip(mapping)
-        .map(|(rqe, &deployment_index)| {
+        .map(|(raqe, &deployment_index)| {
             let deployment = &deployments[deployment_index];
             let stored = storage_bytes_by_deployment
                 .entry(deployment_index)
                 .or_default();
-            *stored = stored.max(storage_bytes(rqe, deployment, facts));
-            merge_cost += merge(rqe, deployment, facts);
-            query_cost += query(rqe, deployment, facts);
-            query_latency_secs(rqe, deployment, facts)
+            *stored = stored.max(storage_bytes(raqe, deployment, facts));
+            merge_cost += merge(raqe, deployment, facts);
+            query_cost += query(raqe, deployment, facts);
+            query_latency_ms(raqe, deployment, facts)
         })
         .collect();
 
@@ -200,22 +201,27 @@ pub fn score(
             cpu_secs_per_sec: 0.0,
             memory_bytes: storage_bytes_by_deployment.values().sum(),
         },
-        query_latency_secs,
+        query_latency_ms,
     }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::test_support::{deployment, facts, rqe};
+    use crate::test_support::{deployment, facts, raqe};
 
     #[test]
     fn scores_each_phase() {
         // 4 groups and 7 samples/sec. Window 20 s sliding by 10 s: 2 open
         // windows. Lookback 60 s every 30 s: 3 windows merged, and
         // (60 − 20)/10 + 1 = 5 closed windows stored.
-        let deployment = deployment(10.0, 2.0, 3.0, 5.0, 20, 10);
-        let result = score(&[rqe(60, 30)], &[deployment], &vec![0], &facts(4, 7));
+        let deployment = deployment(10.0, 2.0, 3.0, 5.0, 20_000, 10_000);
+        let result = score(
+            &[raqe(60_000, 30_000)],
+            &[deployment],
+            &vec![0],
+            &facts(4, 7),
+        );
 
         assert_eq!(result.ingest.cpu_secs_per_sec, 28.0); // 7 × 2 × 2
         assert_eq!(result.ingest.memory_bytes, 80.0); // 4 × 10 × 2
@@ -225,29 +231,34 @@ mod tests {
         assert_eq!(result.query.memory_bytes, 2048.0); // 4 × 32 × 16
         assert_eq!(result.storage.cpu_secs_per_sec, 0.0);
         assert_eq!(result.storage.memory_bytes, 200.0); // 4 × 10 × 5
-        assert_eq!(result.query_latency_secs, vec![44.0]); // 4 × (5 + 2 × 3)
+        assert_eq!(result.query_latency_ms, vec![44_000.0]); // 4 × (5 + 2 × 3) s
         assert!((result.cpu_secs_per_sec() - (28.0 + 44.0 / 30.0)).abs() < 1e-12);
         assert_eq!(result.memory_bytes(), 80.0 + 40.0 + 2048.0 + 200.0);
     }
 
     #[test]
     fn a_direct_query_stores_one_closed_window_and_never_merges() {
-        let deployment = deployment(10.0, 1.0, 1.0, 1.0, 60, 60);
-        let result = score(&[rqe(60, 60)], &[deployment], &vec![0], &facts(1, 1));
+        let deployment = deployment(10.0, 1.0, 1.0, 1.0, 60_000, 60_000);
+        let result = score(
+            &[raqe(60_000, 60_000)],
+            &[deployment],
+            &vec![0],
+            &facts(1, 1),
+        );
         assert_eq!(result.storage.memory_bytes, 10.0);
         assert_eq!(result.merge, PhaseCost::default());
     }
 
     #[test]
     fn shared_deployment_stores_for_its_longest_lookback() {
-        let deployment = deployment(10.0, 1.0, 1.0, 1.0, 10, 10);
-        let rqes = [rqe(60, 10), rqe(600, 10)];
-        let result = score(&rqes, &[deployment], &vec![0, 0], &facts(2, 2));
+        let deployment = deployment(10.0, 1.0, 1.0, 1.0, 10_000, 10_000);
+        let raqes = [raqe(60_000, 10_000), raqe(600_000, 10_000)];
+        let result = score(&raqes, &[deployment], &vec![0, 0], &facts(2, 2));
         // 2 groups × 10 bytes × ((600 − 10)/10 + 1) closed windows.
         assert_eq!(result.storage.memory_bytes, 1200.0);
-        // Ingest is paid once; merge once per RQE.
+        // Ingest is paid once; merge once per RAQE.
         assert_eq!(result.ingest.cpu_secs_per_sec, 2.0);
-        // Two RQEs × 2 groups × 10 bytes.
+        // Two RAQEs × 2 groups × 10 bytes.
         assert_eq!(result.merge.memory_bytes, 2.0 * 2.0 * 10.0);
     }
 }
diff --git a/rqe-optimizer/src/autosketch.rs b/rqe-optimizer/src/autosketch.rs
index 74c0a464..72fb4e6d 100644
--- a/rqe-optimizer/src/autosketch.rs
+++ b/rqe-optimizer/src/autosketch.rs
@@ -5,23 +5,23 @@
 //! fixed CMS width/depth grid to each variant's measured parameter axes.
 //!
 //! AutoSketch plans one query at a time, never shares state across queries,
-//! and constrains accuracy only. So every RQE is searched independently,
+//! and constrains accuracy only. So every RAQE is searched independently,
 //! gets its own deployment, and latency is not considered.
 //!
 //! One extension over the paper: AutoSketch's compiler maps an operator to one
 //! sketch algorithm and tunes its parameters, while this search covers every
-//! variant serving the RQE's capability and keeps the cheapest, so it chooses
-//! from the same sketches as the MILP.
+//! deployable variant serving the RAQE's capability and keeps the cheapest,
+//! so it chooses from the same sketches as the MILP.
 
 use crate::candidates::gcd;
-use crate::{AtomicCostEntry, Deployment, Mapping, Rqe, Seconds};
+use crate::{AtomicCostEntry, Deployment, Mapping, Millis, Raqe};
 use std::collections::{BTreeMap, VecDeque};
 use std::time::Instant;
 
-/// One RQE's search. `selected` and `probes` index into the cost table.
+/// One RAQE's search. `selected` and `probes` index into the cost table.
 #[derive(Debug, Clone)]
-pub struct RqeSearch {
-    pub rqe_id: String,
+pub struct RaqeSearch {
+    pub raqe_id: String,
     /// Smallest feasible configuration found; `None` means unservable.
     pub selected: Option<usize>,
     /// Distinct configurations whose accuracy was evaluated, in order.
@@ -29,78 +29,90 @@ pub struct RqeSearch {
     pub search_wall_secs: f64,
 }
 
-/// A whole-workload plan: one dedicated deployment per RQE, so identical
-/// choices are still paid for once per RQE, and the identity mapping.
+/// A whole-workload plan: one dedicated deployment per RAQE, so identical
+/// choices are still paid for once per RAQE, and the identity mapping.
 #[derive(Debug, Clone)]
 pub struct AutoSketchPlan {
     pub deployments: Vec<Deployment>,
     pub mapping: Mapping,
-    pub searches: Vec<RqeSearch>,
+    pub searches: Vec<RaqeSearch>,
 }
 
 /// One sliding sketch per query window: `x = S`, `y = T`, or `gcd(S, T)`
 /// when `T` does not divide `S`, so the window still tiles into slides.
-pub fn window_adapter(rqe: &Rqe) -> (Seconds, Seconds) {
-    (rqe.lookback_secs, gcd(rqe.lookback_secs, rqe.interval_secs))
+pub fn window_adapter(raqe: &Raqe) -> (Millis, Millis) {
+    (raqe.lookback_ms, gcd(raqe.lookback_ms, raqe.interval_ms))
 }
 
-/// Search every RQE independently. `accuracy` is the benchmark oracle: the
-/// measured metric named by `rqe.accuracy_metric` for a configuration, or
-/// `None` when it was not measured. Returns the IDs of unservable RQEs.
+/// Search every RAQE independently. `accuracy` is the benchmark oracle: the
+/// measured metric named by `raqe.accuracy_metric` for a configuration, or
+/// `None` when it was not measured. Returns the IDs of unservable RAQEs.
+/// `allow_undeployable_families` as in
+/// [`crate::Capability::candidate_families`].
 pub fn plan(
-    rqes: &[Rqe],
+    raqes: &[Raqe],
     costs: &[AtomicCostEntry],
     seed: u64,
-    mut accuracy: impl FnMut(&Rqe, &AtomicCostEntry) -> Option<f64>,
+    allow_undeployable_families: bool,
+    mut accuracy: impl FnMut(&Raqe, &AtomicCostEntry) -> Option<f64>,
 ) -> Result<AutoSketchPlan, Vec<String>> {
-    let searches: Vec<_> = rqes
+    let searches: Vec<_> = raqes
         .iter()
-        .map(|rqe| search(rqe, costs, seed, &mut accuracy))
+        .map(|raqe| {
+            search(
+                raqe,
+                costs,
+                seed,
+                allow_undeployable_families,
+                &mut accuracy,
+            )
+        })
         .collect();
     let unservable: Vec<_> = searches
         .iter()
         .filter(|s| s.selected.is_none())
-        .map(|s| s.rqe_id.clone())
+        .map(|s| s.raqe_id.clone())
         .collect();
     if !unservable.is_empty() {
         return Err(unservable);
     }
-    let deployments = rqes
+    let deployments = raqes
         .iter()
         .zip(&searches)
-        .map(|(rqe, s)| {
-            let (window_secs, slide_secs) = window_adapter(rqe);
+        .map(|(raqe, s)| {
+            let (window_ms, slide_ms) = window_adapter(raqe);
             Deployment {
-                capability: rqe.capability,
-                metric: rqe.metric.clone(),
-                grouping_labels: rqe.grouping_labels.clone(),
-                config: costs[s.selected.expect("unservable RQEs returned above")].clone(),
-                window_secs,
-                slide_secs,
+                capability: raqe.capability,
+                metric: raqe.metric.clone(),
+                spatial_filter: raqe.spatial_filter.clone(),
+                grouping_labels: raqe.grouping_labels.clone(),
+                config: costs[s.selected.expect("unservable RAQEs returned above")].clone(),
+                window_ms,
+                slide_ms,
             }
         })
         .collect();
     Ok(AutoSketchPlan {
         deployments,
-        mapping: (0..rqes.len()).collect(),
+        mapping: (0..raqes.len()).collect(),
         searches,
     })
 }
 
-/// Algorithm 4 for one RQE: LHS seeds per sketch variant, then
+/// Algorithm 4 for one RAQE: LHS seeds per sketch variant, then
 /// feasibility-directed neighbor search with the paper's pruning and
 /// stopping rules. Minimizes memory per instance, then insert CPU.
 pub fn search(
-    rqe: &Rqe,
+    raqe: &Raqe,
     costs: &[AtomicCostEntry],
     seed: u64,
-    mut accuracy: impl FnMut(&Rqe, &AtomicCostEntry) -> Option<f64>,
-) -> RqeSearch {
+    allow_undeployable_families: bool,
+    mut accuracy: impl FnMut(&Raqe, &AtomicCostEntry) -> Option<f64>,
+) -> RaqeSearch {
     let started = Instant::now();
-    let grids: Vec<Grid> = rqe
+    let grids: Vec<Grid> = raqe
         .capability
-        .families()
-        .iter()
+        .candidate_families(allow_undeployable_families)
         .filter_map(|variant| Grid::new(variant, costs))
         .collect();
 
@@ -132,8 +144,8 @@ pub fn search(
             continue;
         }
         probes.push(index);
-        let feasible = accuracy(rqe, &costs[index])
-            .is_some_and(|value| value.is_finite() && rqe.accuracy_ok(value));
+        let feasible = accuracy(raqe, &costs[index])
+            .is_some_and(|value| value.is_finite() && raqe.accuracy_ok(value));
         evaluated.insert(index, feasible);
         if feasible && best.is_none_or(|b| cheaper(index, b)) {
             best = Some(index);
@@ -163,8 +175,8 @@ pub fn search(
         }
     }
 
-    RqeSearch {
-        rqe_id: rqe.id.clone(),
+    RaqeSearch {
+        raqe_id: raqe.id.clone(),
         selected: best,
         probes,
         search_wall_secs: started.elapsed().as_secs_f64(),
@@ -351,27 +363,29 @@ mod tests {
             .collect()
     }
 
-    fn rqe(id: &str, lookback: Seconds, interval: Seconds, tolerance: f64) -> Rqe {
-        Rqe {
+    fn raqe(id: &str, lookback: Millis, interval: Millis, tolerance: f64) -> Raqe {
+        Raqe {
             id: id.into(),
             capability: Capability::TopK,
-            lookback_secs: lookback,
-            interval_secs: interval,
+            lookback_ms: lookback,
+            interval_ms: interval,
             metric: METRIC.into(),
+            spatial_filter: String::new(),
             grouping_labels: LabelSet::new(),
             accuracy_metric: ERR.into(),
-            accuracy_tolerance: tolerance,
+            accuracy_sla: tolerance,
             accuracy_direction: AccuracyDirection::LowerIsBetter,
+            latency_sla_ms: None,
         }
     }
 
-    fn table_accuracy(rqe: &Rqe, config: &AtomicCostEntry) -> Option<f64> {
-        config.query_accuracy.get(&rqe.accuracy_metric).copied()
+    fn table_accuracy(raqe: &Raqe, config: &AtomicCostEntry) -> Option<f64> {
+        config.query_accuracy.get(&raqe.accuracy_metric).copied()
     }
 
-    fn cheapest_feasible(rqe: &Rqe, costs: &[AtomicCostEntry]) -> usize {
+    fn cheapest_feasible(raqe: &Raqe, costs: &[AtomicCostEntry]) -> usize {
         (0..costs.len())
-            .filter(|&i| rqe.accuracy_ok_for(&costs[i]))
+            .filter(|&i| raqe.accuracy_ok_for(&costs[i]))
             .min_by(|&a, &b| compare_resources(&costs[a], &costs[b]))
             .unwrap()
     }
@@ -381,8 +395,8 @@ mod tests {
         let costs = grid(&[1, 2, 3], &[100, 200, 400, 800]);
         // Needs rows * cols >= 1000: rows=3 cols=400 (1200 counters) is the
         // smallest; rows=2 cols=800 and rows=3 cols=800 also pass.
-        let r = rqe("r", 3_600, 60, 1.0 / 1_000.0);
-        let found = search(&r, &costs, 7, table_accuracy);
+        let r = raqe("r", 3_600_000, 60_000, 1.0 / 1_000.0);
+        let found = search(&r, &costs, 7, false, table_accuracy);
         assert_eq!(found.selected, Some(cheapest_feasible(&r, &costs)));
     }
 
@@ -391,8 +405,8 @@ mod tests {
         let rows: Vec<u64> = (1..=8).collect();
         let cols: Vec<u64> = (0..8).map(|i| 64 << i).collect();
         let costs = grid(&rows, &cols);
-        let r = rqe("r", 3_600, 60, 1.0 / 2_000.0);
-        let found = search(&r, &costs, 42, table_accuracy);
+        let r = raqe("r", 3_600_000, 60_000, 1.0 / 2_000.0);
+        let found = search(&r, &costs, 42, false, table_accuracy);
         assert!(found.selected.is_some());
         assert!(
             found.probes.len() < costs.len(),
@@ -406,15 +420,18 @@ mod tests {
     #[test]
     fn never_shares_identical_choices() {
         let costs = grid(&[2, 3], &[256, 512]);
-        let rqes = vec![rqe("a", 3_600, 60, 0.01), rqe("b", 3_600, 60, 0.01)];
-        let plan = plan(&rqes, &costs, 1, table_accuracy).unwrap();
+        let raqes = vec![
+            raqe("a", 3_600_000, 60_000, 0.01),
+            raqe("b", 3_600_000, 60_000, 0.01),
+        ];
+        let plan = plan(&raqes, &costs, 1, false, table_accuracy).unwrap();
         assert_eq!(plan.deployments.len(), 2);
         assert_eq!(plan.deployments[0], plan.deployments[1]);
         assert_eq!(plan.mapping, vec![0, 1]);
 
         let facts = facts(1, 10);
-        let both = score(&rqes, &plan.deployments, &plan.mapping, &facts);
-        let one = score(&rqes[..1], &plan.deployments[..1], &vec![0], &facts);
+        let both = score(&raqes, &plan.deployments, &plan.mapping, &facts);
+        let one = score(&raqes[..1], &plan.deployments[..1], &vec![0], &facts);
         assert_eq!(
             both.ingest,
             PhaseCost {
@@ -428,24 +445,27 @@ mod tests {
     fn adapted_window_is_eligible() {
         let costs = grid(&[2, 3], &[256, 512]);
         // T divides S, and T does not divide S (gcd slide).
-        let rqes = vec![rqe("tiled", 3_600, 60, 0.01), rqe("gcd", 3_600, 280, 0.01)];
-        let plan = plan(&rqes, &costs, 1, table_accuracy).unwrap();
-        assert_eq!(window_adapter(&rqes[0]), (3_600, 60));
-        assert_eq!(window_adapter(&rqes[1]), (3_600, 40));
-        for (r, d) in rqes.iter().zip(&plan.deployments) {
+        let raqes = vec![
+            raqe("tiled", 3_600_000, 60_000, 0.01),
+            raqe("gcd", 3_600_000, 280_000, 0.01),
+        ];
+        let plan = plan(&raqes, &costs, 1, false, table_accuracy).unwrap();
+        assert_eq!(window_adapter(&raqes[0]), (3_600_000, 60_000));
+        assert_eq!(window_adapter(&raqes[1]), (3_600_000, 40_000));
+        for (r, d) in raqes.iter().zip(&plan.deployments) {
             assert!(is_eligible(r, d), "{} not eligible", r.id);
         }
     }
 
     #[test]
-    fn reports_unservable_rqe() {
+    fn reports_unservable_raqe() {
         let costs = grid(&[2, 3], &[256, 512]);
-        let rqes = vec![
-            rqe("ok", 3_600, 60, 0.01),
-            rqe("too_strict", 3_600, 60, 1e-9),
+        let raqes = vec![
+            raqe("ok", 3_600_000, 60_000, 0.01),
+            raqe("too_strict", 3_600_000, 60_000, 1e-9),
         ];
         assert_eq!(
-            plan(&rqes, &costs, 1, table_accuracy).unwrap_err(),
+            plan(&raqes, &costs, 1, false, table_accuracy).unwrap_err(),
             vec!["too_strict".to_string()]
         );
     }
@@ -453,8 +473,8 @@ mod tests {
     #[test]
     fn missing_measurement_is_infeasible() {
         let costs = grid(&[2, 3], &[256, 512]);
-        let r = rqe("r", 3_600, 60, 0.01);
-        let found = search(&r, &costs, 1, |_, _| None);
+        let r = raqe("r", 3_600_000, 60_000, 0.01);
+        let found = search(&r, &costs, 1, false, |_, _| None);
         assert_eq!(found.selected, None);
         assert_eq!(found.probes.len(), costs.len());
     }
@@ -471,10 +491,10 @@ mod tests {
         }
     }
 
-    fn quantile_rqe(tolerance: f64) -> Rqe {
-        Rqe {
+    fn quantile_raqe(tolerance: f64) -> Raqe {
+        Raqe {
             capability: Capability::Quantile,
-            ..rqe("q", 3_600, 60, tolerance)
+            ..raqe("q", 3_600_000, 60_000, tolerance)
         }
     }
 
@@ -484,9 +504,9 @@ mod tests {
         // because the seed is measured. A feasible seed shrinks to k = 5 and
         // stops; an infeasible one grows to k = 6 and stops.
         let costs: Vec<_> = (1..=10).map(kll).collect();
-        let r = quantile_rqe(1.0 / 6.0);
+        let r = quantile_raqe(1.0 / 6.0);
         for seed in 0..20 {
-            let found = search(&r, &costs, seed, table_accuracy);
+            let found = search(&r, &costs, seed, false, table_accuracy);
             let ks: Vec<u64> = found.probes.iter().map(|&i| i as u64 + 1).collect();
             assert_eq!(found.selected, Some(5), "seed {seed}: {ks:?}");
             let s = ks[0];
@@ -501,13 +521,14 @@ mod tests {
 
     #[test]
     fn chooses_the_cheaper_variant() {
+        // countsketch-heap isn't deployable, so this needs the study flag.
         let mut costs = grid(&[2, 3], &[256, 512]);
         let mut cs = cms(2, 256);
         cs.sketch = "countsketch-heap-topk-fastpath-vector2d".into();
         cs.mem_bytes_per_instance = 100.0;
         costs.push(cs);
-        let r = rqe("r", 3_600, 60, 0.01);
-        let found = search(&r, &costs, 3, table_accuracy);
+        let r = raqe("r", 3_600_000, 60_000, 0.01);
+        let found = search(&r, &costs, 3, true, table_accuracy);
         assert_eq!(found.selected, Some(costs.len() - 1));
     }
 
@@ -521,11 +542,11 @@ mod tests {
                 c
             })
             .collect();
-        let r = Rqe {
+        let r = Raqe {
             accuracy_direction: AccuracyDirection::HigherIsBetter,
-            ..quantile_rqe(0.95)
+            ..quantile_raqe(0.95)
         };
-        let found = search(&r, &costs, 11, table_accuracy);
+        let found = search(&r, &costs, 11, false, table_accuracy);
         assert_eq!(found.selected, Some(4)); // k = 5: 0.95
     }
 
diff --git a/rqe-optimizer/src/candidates.rs b/rqe-optimizer/src/candidates.rs
index bf522f13..2a4db293 100644
--- a/rqe-optimizer/src/candidates.rs
+++ b/rqe-optimizer/src/candidates.rs
@@ -1,10 +1,10 @@
 //! Candidate generation and eligibility (§3).
 
 use crate::analytical_cost_model::merge_memory_per_group;
-use crate::{AtomicCostEntry, Capability, Deployment, LabelSet, Rqe, Seconds, WorkloadFacts};
+use crate::{AtomicCostEntry, Capability, Deployment, LabelSet, Millis, Raqe, WorkloadFacts};
 use std::collections::{BTreeMap, BTreeSet};
 
-pub fn gcd(a: Seconds, b: Seconds) -> Seconds {
+pub fn gcd(a: Millis, b: Millis) -> Millis {
     if b == 0 {
         a
     } else {
@@ -12,7 +12,7 @@ pub fn gcd(a: Seconds, b: Seconds) -> Seconds {
     }
 }
 
-fn divisors(value: Seconds) -> Vec<Seconds> {
+fn divisors(value: Millis) -> Vec<Millis> {
     let mut out = Vec::new();
     let mut factor = 1;
     while factor <= value / factor {
@@ -31,7 +31,7 @@ fn divisors(value: Seconds) -> Vec<Seconds> {
 /// The gcd of every non-empty subset of `values`, in one pass without listing
 /// subsets: `gcd(A ∪ {v}) = gcd(gcd(A), v)`, so each new value extends every
 /// gcd found so far.
-fn subset_gcds(values: impl IntoIterator<Item = Seconds>) -> BTreeSet<Seconds> {
+fn subset_gcds(values: impl IntoIterator<Item = Millis>) -> BTreeSet<Millis> {
     let mut gcds = BTreeSet::new();
     for value in values {
         let extended: Vec<_> = gcds.iter().map(|&found| gcd(found, value)).collect();
@@ -44,9 +44,10 @@ fn subset_gcds(values: impl IntoIterator<Item = Seconds>) -> BTreeSet<Seconds> {
 /// Windows and slides are multiples of the metric's scrape interval: anything
 /// finer only splits one scrape's samples.
 fn candidate_deployments(
-    group: &[&Rqe],
+    group: &[&Raqe],
     costs: &[AtomicCostEntry],
-    scrape_interval_secs: Seconds,
+    scrape_interval_ms: Millis,
+    allow_undeployable_families: bool,
 ) -> Vec<Deployment> {
     if group.is_empty() {
         return Vec::new();
@@ -54,33 +55,35 @@ fn candidate_deployments(
     let capability = group[0].capability;
     let windows = group
         .iter()
-        .flat_map(|r| divisors(r.lookback_secs))
-        .filter(|window_secs| window_secs.is_multiple_of(scrape_interval_secs))
+        .flat_map(|r| divisors(r.lookback_ms / scrape_interval_ms))
+        .map(|scrapes| scrapes * scrape_interval_ms)
         .collect::<BTreeSet<_>>();
     let mut deployments = Vec::new();
-    for window_secs in windows {
-        // gcd(x, T) is the largest slide that serves an RQE. For RQEs sharing a
+    for window_ms in windows {
+        // gcd(x, T) is the largest slide that serves an RAQE. For RAQEs sharing a
         // deployment, the gcd of theirs is strictly better than any finer
         // slide that serves them all. Any subset might share.
         let slides = subset_gcds(
             group
                 .iter()
-                .filter(|r| r.lookback_secs % window_secs == 0)
-                .map(|r| gcd(window_secs, r.interval_secs)),
+                .filter(|r| r.lookback_ms % window_ms == 0)
+                .map(|r| gcd(window_ms, r.interval_ms)),
         );
-        for config in costs
-            .iter()
-            .filter(|c| capability.families().contains(&c.sketch.as_str()))
-        {
-            for &slide_secs in &slides {
-                if slide_secs.is_multiple_of(scrape_interval_secs) {
+        for config in costs.iter().filter(|c| {
+            capability
+                .candidate_families(allow_undeployable_families)
+                .any(|family| family == c.sketch)
+        }) {
+            for &slide_ms in &slides {
+                if slide_ms.is_multiple_of(scrape_interval_ms) {
                     deployments.push(Deployment {
                         capability,
                         metric: group[0].metric.clone(),
+                        spatial_filter: group[0].spatial_filter.clone(),
                         grouping_labels: group[0].grouping_labels.clone(),
                         config: config.clone(),
-                        window_secs,
-                        slide_secs,
+                        window_ms,
+                        slide_ms,
                     });
                 }
             }
@@ -89,12 +92,17 @@ fn candidate_deployments(
     deployments
 }
 
+/// Configurations come from [`Capability::candidate_families`].
 pub fn build_all_candidates(
-    rqes: &[Rqe],
+    raqes: &[Raqe],
     costs: &[AtomicCostEntry],
     facts: &WorkloadFacts,
+    allow_undeployable_families: bool,
 ) -> Vec<Deployment> {
-    prune_dominated_candidates(rqes, build_all_candidates_unpruned(rqes, costs, facts))
+    prune_dominated_candidates(
+        raqes,
+        build_all_candidates_unpruned(raqes, costs, facts, allow_undeployable_families),
+    )
 }
 
 /// Generate the complete v1 candidate set before dominance pruning.
@@ -102,21 +110,32 @@ pub fn build_all_candidates(
 /// The public optimizer entry point is [`build_all_candidates`]. This helper
 /// exists so diagnostics can report exactly how much safe pruning removed.
 pub fn build_all_candidates_unpruned(
-    rqes: &[Rqe],
+    raqes: &[Raqe],
     costs: &[AtomicCostEntry],
     facts: &WorkloadFacts,
+    allow_undeployable_families: bool,
 ) -> Vec<Deployment> {
-    let mut groups: BTreeMap<(Capability, &str, &LabelSet), Vec<&Rqe>> = BTreeMap::new();
-    for rqe in rqes {
+    let mut groups: BTreeMap<(Capability, &str, &str, &LabelSet), Vec<&Raqe>> = BTreeMap::new();
+    for raqe in raqes {
         groups
-            .entry((rqe.capability, &rqe.metric, &rqe.grouping_labels))
+            .entry((
+                raqe.capability,
+                &raqe.metric,
+                &raqe.spatial_filter,
+                &raqe.grouping_labels,
+            ))
             .or_default()
-            .push(rqe);
+            .push(raqe);
     }
     groups
         .iter()
-        .flat_map(|(&(_, metric, _), group)| {
-            candidate_deployments(group, costs, facts[metric].scrape_interval_secs)
+        .flat_map(|(&(_, metric, _, _), group)| {
+            candidate_deployments(
+                group,
+                costs,
+                facts[metric].scrape_interval_ms,
+                allow_undeployable_families,
+            )
         })
         .collect()
 }
@@ -124,15 +143,20 @@ pub fn build_all_candidates_unpruned(
 /// Remove a candidate only when another candidate can replace it in every
 /// mapping without making any modeled objective worse.
 ///
-/// This comparison is deliberately local to a (capability, metric, grouping)
-/// group, where `card(G)`, arrival rate and query output size are common
+/// This comparison is deliberately local to a (capability, metric,
+/// spatial_filter, grouping) group, where `card(G)`, arrival rate and query output size are common
 /// multipliers. So it compares ingest CPU and memory per group, then latency
-/// (merge and query CPU), merge memory and stored memory for each RQE the
+/// (merge and query CPU), merge memory and stored memory for each RAQE the
 /// dominated candidate can serve.
-pub fn prune_dominated_candidates(rqes: &[Rqe], candidates: Vec<Deployment>) -> Vec<Deployment> {
+pub fn prune_dominated_candidates(raqes: &[Raqe], candidates: Vec<Deployment>) -> Vec<Deployment> {
     let eligibility: Vec<Vec<bool>> = candidates
         .iter()
-        .map(|candidate| rqes.iter().map(|rqe| is_eligible(rqe, candidate)).collect())
+        .map(|candidate| {
+            raqes
+                .iter()
+                .map(|raqe| is_eligible(raqe, candidate))
+                .collect()
+        })
         .collect();
 
     candidates
@@ -148,7 +172,7 @@ pub fn prune_dominated_candidates(rqes: &[Rqe], candidates: Vec<Deployment>) ->
                     &eligibility[other_index],
                     candidate,
                     &eligibility[candidate_index],
-                    rqes,
+                    raqes,
                 ) {
                     Dominance::StrictlyBetter => true,
                     // The earlier of two identical candidates wins, so a tie
@@ -168,17 +192,18 @@ enum Dominance {
     NotDominating,
 }
 
-/// Whether `replacement` can stand in for `original`: it serves every RQE
+/// Whether `replacement` can stand in for `original`: it serves every RAQE
 /// `original` serves and is no worse on any cost.
 fn dominance(
     replacement: &Deployment,
     replacement_coverage: &[bool],
     original: &Deployment,
     original_coverage: &[bool],
-    rqes: &[Rqe],
+    raqes: &[Raqe],
 ) -> Dominance {
     let same_group = replacement.capability == original.capability
         && replacement.metric == original.metric
+        && replacement.spatial_filter == original.spatial_filter
         && replacement.grouping_labels == original.grouping_labels;
     let covers_original = original_coverage
         .iter()
@@ -193,22 +218,22 @@ fn dominance(
         (ingest_cpu(replacement), ingest_cpu(original)),
         (ingest_memory(replacement), ingest_memory(original)),
     ];
-    for (rqe, _) in rqes
+    for (raqe, _) in raqes
         .iter()
         .zip(original_coverage)
         .filter(|(_, &serves)| serves)
     {
         costs.push((
-            query_latency(replacement, rqe),
-            query_latency(original, rqe),
+            query_latency(replacement, raqe),
+            query_latency(original, raqe),
         ));
         costs.push((
-            merge_memory_per_group(rqe, replacement),
-            merge_memory_per_group(rqe, original),
+            merge_memory_per_group(raqe, replacement),
+            merge_memory_per_group(raqe, original),
         ));
         costs.push((
-            stored_memory(replacement, rqe),
-            stored_memory(original, rqe),
+            stored_memory(replacement, raqe),
+            stored_memory(original, raqe),
         ));
     }
 
@@ -248,23 +273,23 @@ fn ingest_memory(deployment: &Deployment) -> f64 {
     open_window_count(deployment) * deployment.config.mem_bytes_per_instance
 }
 
-fn query_latency(deployment: &Deployment, rqe: &Rqe) -> f64 {
+fn query_latency(deployment: &Deployment, raqe: &Raqe) -> f64 {
     let instances = deployment
-        .query_instance_count(rqe.lookback_secs)
-        .expect("candidate coverage only contains exactly tiled RQEs");
+        .query_instance_count(raqe.lookback_ms)
+        .expect("candidate coverage only contains exactly tiled RAQEs");
     deployment.config.query_cpu_secs
         + instances.saturating_sub(1) as f64 * deployment.config.merge_cpu_secs
 }
 
-/// Closed-window bytes per group when serving `rqe`.
-fn stored_memory(deployment: &Deployment, rqe: &Rqe) -> f64 {
+/// Closed-window bytes per group when serving `raqe`.
+fn stored_memory(deployment: &Deployment, raqe: &Raqe) -> f64 {
     deployment
-        .closed_instance_count(rqe.lookback_secs)
-        .expect("candidate coverage only contains exactly tiled RQEs") as f64
+        .closed_instance_count(raqe.lookback_ms)
+        .expect("candidate coverage only contains exactly tiled RAQEs") as f64
         * deployment.config.mem_bytes_per_instance
 }
 
-pub fn eligible_deployments_for(r: &Rqe, deployments: &[Deployment]) -> Vec<usize> {
+pub fn eligible_deployments_for(r: &Raqe, deployments: &[Deployment]) -> Vec<usize> {
     deployments
         .iter()
         .enumerate()
@@ -273,15 +298,16 @@ pub fn eligible_deployments_for(r: &Rqe, deployments: &[Deployment]) -> Vec<usiz
         .collect()
 }
 
-pub fn is_eligible(r: &Rqe, d: &Deployment) -> bool {
+pub fn is_eligible(r: &Raqe, d: &Deployment) -> bool {
     r.capability == d.capability
         && r.metric == d.metric
+        && r.spatial_filter == d.spatial_filter
         && r.grouping_labels == d.grouping_labels
-        && d.window_secs != 0
-        && d.slide_secs != 0
-        && d.window_secs.is_multiple_of(d.slide_secs)
-        && r.lookback_secs.is_multiple_of(d.window_secs)
-        && r.interval_secs.is_multiple_of(d.slide_secs)
+        && d.window_ms != 0
+        && d.slide_ms != 0
+        && d.window_ms.is_multiple_of(d.slide_ms)
+        && r.lookback_ms.is_multiple_of(d.window_ms)
+        && r.interval_ms.is_multiple_of(d.slide_ms)
         && r.accuracy_ok_for(&d.config)
 }
 
@@ -291,17 +317,19 @@ mod tests {
     use crate::test_support::{facts, METRIC};
     use crate::{AccuracyDirection, LabelSet};
     use std::collections::BTreeMap;
-    fn rqe(id: &str, lookback: Seconds, interval: Seconds) -> Rqe {
-        Rqe {
+    fn raqe(id: &str, lookback: Millis, interval: Millis) -> Raqe {
+        Raqe {
             id: id.into(),
             capability: Capability::TopK,
-            lookback_secs: lookback,
-            interval_secs: interval,
+            lookback_ms: lookback,
+            interval_ms: interval,
             metric: METRIC.into(),
+            spatial_filter: String::new(),
             grouping_labels: LabelSet::new(),
             accuracy_metric: "err".into(),
-            accuracy_tolerance: 1.0,
+            accuracy_sla: 1.0,
             accuracy_direction: AccuracyDirection::LowerIsBetter,
+            latency_sla_ms: None,
         }
     }
     fn cost() -> AtomicCostEntry {
@@ -317,32 +345,50 @@ mod tests {
     }
     #[test]
     fn shared_slide_comes_from_subset_gcd_not_all_divisors() {
-        let rqes = vec![rqe("a", 60, 20), rqe("b", 60, 30)];
-        let candidates = build_all_candidates(&rqes, &[cost()], &facts(1, 1));
+        let raqes = vec![raqe("a", 60_000, 20_000), raqe("b", 60_000, 30_000)];
+        let candidates = build_all_candidates(&raqes, &[cost()], &facts(1, 1), false);
         let slides: BTreeSet<_> = candidates
             .iter()
-            .filter(|d| d.window_secs == 60)
-            .map(|d| d.slide_secs)
+            .filter(|d| d.window_ms == 60_000)
+            .map(|d| d.slide_ms)
             .collect();
-        assert_eq!(slides, BTreeSet::from([10, 20, 30]));
+        assert_eq!(slides, BTreeSet::from([10_000, 20_000, 30_000]));
     }
     #[test]
     fn windows_and_slides_are_multiples_of_the_scrape_interval() {
-        let rqes = vec![rqe("a", 60, 20), rqe("b", 60, 30)];
+        let raqes = vec![raqe("a", 60_000, 20_000), raqe("b", 60_000, 30_000)];
         let mut facts = facts(1, 1);
-        facts.get_mut(METRIC).unwrap().scrape_interval_secs = 15;
-        let candidates = build_all_candidates_unpruned(&rqes, &[cost()], &facts);
+        facts.get_mut(METRIC).unwrap().scrape_interval_ms = 15_000;
+        let candidates = build_all_candidates_unpruned(&raqes, &[cost()], &facts, false);
         assert!(!candidates.is_empty());
         // Without the filter, windows 1..60 and slides 10/20 would appear.
         assert!(candidates
             .iter()
-            .all(|d| d.window_secs % 15 == 0 && d.slide_secs % 15 == 0));
+            .all(|d| d.window_ms % 15_000 == 0 && d.slide_ms % 15_000 == 0));
     }
+    #[test]
+    fn different_spatial_filters_never_share_a_deployment() {
+        let unfiltered = raqe("a", 60_000, 60_000);
+        let filtered = Raqe {
+            spatial_filter: r#"{job="api"}"#.into(),
+            ..raqe("b", 60_000, 60_000)
+        };
+        let candidates = build_all_candidates(
+            &[unfiltered.clone(), filtered.clone()],
+            &[cost()],
+            &facts(1, 1),
+            false,
+        );
+        assert!(candidates
+            .iter()
+            .all(|d| is_eligible(&unfiltered, d) != is_eligible(&filtered, d)));
+    }
+
     #[test]
     fn a_min_query_is_only_offered_the_min_accumulator() {
-        let r = Rqe {
+        let r = Raqe {
             capability: Capability::Min,
-            ..rqe("r", 60, 60)
+            ..raqe("r", 60_000, 60_000)
         };
         let named = |sketch: &str| AtomicCostEntry {
             sketch: sketch.into(),
@@ -352,52 +398,74 @@ mod tests {
             &[r],
             &[named("exact-min"), named("exact-max")],
             &facts(1, 1),
+            false,
         );
         assert!(!candidates.is_empty());
         assert!(candidates.iter().all(|d| d.config.sketch == "exact-min"));
     }
 
+    #[test]
+    fn undeployable_families_are_candidates_only_when_allowed() {
+        let r = Raqe {
+            capability: Capability::Quantile,
+            ..raqe("r", 60_000, 60_000)
+        };
+        let named = |sketch: &str| AtomicCostEntry {
+            sketch: sketch.into(),
+            ..cost()
+        };
+        let costs = [named("kll-percall"), named("dd")];
+        let candidates =
+            build_all_candidates(std::slice::from_ref(&r), &costs, &facts(1, 1), false);
+        assert!(!candidates.is_empty());
+        assert!(candidates.iter().all(|d| d.config.sketch == "kll-percall"));
+        let all = build_all_candidates_unpruned(&[r], &costs, &facts(1, 1), true);
+        assert!(all.iter().any(|d| d.config.sketch == "dd"));
+    }
+
     #[test]
     fn eligibility_requires_exact_non_overlapping_tiling() {
-        let r = rqe("r", 600, 180);
+        let r = raqe("r", 600_000, 180_000);
         let d = Deployment {
             capability: Capability::TopK,
             metric: METRIC.into(),
+            spatial_filter: String::new(),
             grouping_labels: LabelSet::new(),
             config: cost(),
-            window_secs: 120,
-            slide_secs: 60,
+            window_ms: 120_000,
+            slide_ms: 60_000,
         };
         assert!(is_eligible(&r, &d));
         assert!(!is_eligible(
             &r,
             &Deployment {
-                window_secs: 128,
+                window_ms: 128_000,
                 ..d.clone()
             }
         ));
         assert!(!is_eligible(
             &r,
             &Deployment {
-                slide_secs: 70,
+                slide_ms: 70_000,
                 ..d
             }
         ));
     }
 
     #[test]
-    fn prunes_finer_slide_when_the_coarser_slide_serves_the_same_rqe() {
-        let r = rqe("r", 60, 60);
+    fn prunes_finer_slide_when_the_coarser_slide_serves_the_same_raqe() {
+        let r = raqe("r", 60_000, 60_000);
         let coarse = Deployment {
             capability: Capability::TopK,
             metric: METRIC.into(),
+            spatial_filter: String::new(),
             grouping_labels: LabelSet::new(),
             config: cost(),
-            window_secs: 60,
-            slide_secs: 60,
+            window_ms: 60_000,
+            slide_ms: 60_000,
         };
         let fine = Deployment {
-            slide_secs: 30,
+            slide_ms: 30_000,
             ..coarse.clone()
         };
 
@@ -408,19 +476,20 @@ mod tests {
 
     #[test]
     fn retains_candidate_with_lower_query_latency() {
-        let r = rqe("r", 60, 60);
+        let r = raqe("r", 60_000, 60_000);
         let cost = cost();
         let large_window = Deployment {
             capability: Capability::TopK,
             metric: METRIC.into(),
+            spatial_filter: String::new(),
             grouping_labels: LabelSet::new(),
             config: cost.clone(),
-            window_secs: 60,
-            slide_secs: 60,
+            window_ms: 60_000,
+            slide_ms: 60_000,
         };
         let small_window = Deployment {
-            window_secs: 30,
-            slide_secs: 30,
+            window_ms: 30_000,
+            slide_ms: 30_000,
             ..large_window.clone()
         };
 
@@ -432,18 +501,19 @@ mod tests {
 
     #[test]
     fn retains_candidate_with_lower_stored_memory() {
-        let r = rqe("r", 60, 60);
+        let r = raqe("r", 60_000, 60_000);
         // Stores (60 − 60) / 60 + 1 = 1 closed instance of 10 bytes.
         let whole_window = Deployment {
             capability: Capability::TopK,
             metric: METRIC.into(),
+            spatial_filter: String::new(),
             grouping_labels: LabelSet::new(),
             config: AtomicCostEntry {
                 mem_bytes_per_instance: 10.0,
                 ..cost()
             },
-            window_secs: 60,
-            slide_secs: 60,
+            window_ms: 60_000,
+            slide_ms: 60_000,
         };
         // Smaller, faster sketches, but (60 − 20) / 20 + 1 = 3 of them: 18 bytes.
         let panes = Deployment {
@@ -453,8 +523,8 @@ mod tests {
                 merge_cpu_secs: 0.1,
                 ..cost()
             },
-            window_secs: 20,
-            slide_secs: 20,
+            window_ms: 20_000,
+            slide_ms: 20_000,
             ..whole_window.clone()
         };
 
@@ -465,18 +535,19 @@ mod tests {
 
     #[test]
     fn retains_direct_query_candidate_that_never_merges() {
-        let r = rqe("r", 60, 60);
+        let r = raqe("r", 60_000, 60_000);
         // x == L: no merge, so no merge memory.
         let direct = Deployment {
             capability: Capability::TopK,
             metric: METRIC.into(),
+            spatial_filter: String::new(),
             grouping_labels: LabelSet::new(),
             config: AtomicCostEntry {
                 mem_bytes_per_instance: 10.0,
                 ..cost()
             },
-            window_secs: 60,
-            slide_secs: 60,
+            window_ms: 60_000,
+            slide_ms: 60_000,
         };
         // Smaller and cheaper on every other cost, but merges two windows,
         // holding a 4-byte accumulator per group.
@@ -487,8 +558,8 @@ mod tests {
                 merge_cpu_secs: 0.1,
                 ..cost()
             },
-            window_secs: 30,
-            slide_secs: 30,
+            window_ms: 30_000,
+            slide_ms: 30_000,
             ..direct.clone()
         };
 
diff --git a/rqe-optimizer/src/enumerate.rs b/rqe-optimizer/src/enumerate.rs
index 5f435b9e..abcaf7cd 100644
--- a/rqe-optimizer/src/enumerate.rs
+++ b/rqe-optimizer/src/enumerate.rs
@@ -1,32 +1,35 @@
 //! Brute-force search over full mappings (§4, §7 step 2).
 //!
-//! A mapping is one deployment choice per RQE (§4's only constraint is
-//! `Σ_D x_{i,D} = 1`; `y_D` is derived). No RQE is eligible outside its own
-//! `(capability, metric, grouping_labels)` group, so choices are independent and the search
-//! space is exactly the cartesian product of the per-RQE eligible lists --
+//! A mapping is one deployment choice per RAQE (§4's only constraint is
+//! `Σ_D x_{i,D} = 1`; `y_D` is derived). No RAQE is eligible outside its own
+//! `(capability, metric, spatial_filter, grouping_labels)` group, so choices are independent and the search
+//! space is exactly the cartesian product of the per-RAQE eligible lists --
 //! nothing to prune against. Sharing falls out when two choices land on the
 //! same index.
 //!
 //! An ILP would replace this module alone: `Mapping` already encodes
 //! `x_{i,D}` and `y_D`.
 
-use crate::{candidates::eligible_deployments_for, Deployment, Mapping, Rqe};
+use crate::{candidates::eligible_deployments_for, Deployment, Mapping, Raqe};
 
-/// RQEs nothing in `deployments` can serve (§3.4). Surfaced separately from
-/// `brute_force`: an unservable RQE is a modeling gap (tolerance too tight,
+/// RAQEs nothing in `deployments` can serve (§3.4). Surfaced separately from
+/// `brute_force`: an unservable RAQE is a modeling gap (tolerance too tight,
 /// no such config), not a normal empty result.
-pub fn unservable(rqes: &[Rqe], deployments: &[Deployment]) -> Vec<String> {
-    rqes.iter()
+pub fn unservable(raqes: &[Raqe], deployments: &[Deployment]) -> Vec<String> {
+    raqes
+        .iter()
         .filter(|r| eligible_deployments_for(r, deployments).is_empty())
         .map(|r| r.id.clone())
         .collect()
 }
 
-/// Every feasible full mapping. Empty if any RQE is unservable -- check
+/// Every feasible full mapping. Empty if any RAQE is unservable -- check
 /// [`unservable`] to tell the two cases apart.
-pub fn brute_force(rqes: &[Rqe], deployments: &[Deployment]) -> Vec<Mapping> {
+pub fn brute_force(raqes: &[Raqe], deployments: &[Deployment]) -> Vec<Mapping> {
     let mut mappings = Vec::new();
-    for_each_mapping(rqes, deployments, |mapping| mappings.push(mapping.to_vec()));
+    for_each_mapping(raqes, deployments, |mapping| {
+        mappings.push(mapping.to_vec())
+    });
     mappings
 }
 
@@ -34,11 +37,11 @@ pub fn brute_force(rqes: &[Rqe], deployments: &[Deployment]) -> Vec<Mapping> {
 /// product. The visitor receives a borrowed, index-aligned mapping that is
 /// valid only for the duration of the call. Returns the number visited.
 pub fn for_each_mapping(
-    rqes: &[Rqe],
+    raqes: &[Raqe],
     deployments: &[Deployment],
     mut visit: impl FnMut(&Mapping),
 ) -> u64 {
-    for_each_mapping_while(rqes, deployments, |mapping| {
+    for_each_mapping_while(raqes, deployments, |mapping| {
         visit(mapping);
         true
     })
@@ -57,11 +60,11 @@ pub struct EnumerationResult {
 /// and bounded inspection modes safe: they never need to enumerate the rest
 /// of the Cartesian product after collecting enough examples.
 pub fn for_each_mapping_while(
-    rqes: &[Rqe],
+    raqes: &[Raqe],
     deployments: &[Deployment],
     mut visit: impl FnMut(&Mapping) -> bool,
 ) -> EnumerationResult {
-    let choices: Vec<Vec<usize>> = rqes
+    let choices: Vec<Vec<usize>> = raqes
         .iter()
         .map(|r| eligible_deployments_for(r, deployments))
         .collect();
@@ -97,7 +100,7 @@ pub fn for_each_mapping_while(
     let mut visited = 0;
     let completed = visit_depth_first(
         &choices,
-        &mut Vec::with_capacity(rqes.len()),
+        &mut Vec::with_capacity(raqes.len()),
         &mut visit,
         &mut visited,
     );
@@ -126,55 +129,57 @@ mod tests {
         }
     }
 
-    fn build_all_candidates(rqes: &[Rqe], costs: &[AtomicCostEntry]) -> Vec<Deployment> {
-        candidates::build_all_candidates(rqes, costs, &facts(1, 1))
+    fn build_all_candidates(raqes: &[Raqe], costs: &[AtomicCostEntry]) -> Vec<Deployment> {
+        candidates::build_all_candidates(raqes, costs, &facts(1, 1), false)
     }
 
-    fn rqe(id: &str, interval: u64, lookback: u64) -> Rqe {
-        Rqe {
+    fn raqe(id: &str, interval: u64, lookback: u64) -> Raqe {
+        Raqe {
             id: id.to_string(),
             capability: Capability::TopK,
-            lookback_secs: lookback,
-            interval_secs: interval,
+            lookback_ms: lookback,
+            interval_ms: interval,
             metric: METRIC.into(),
+            spatial_filter: String::new(),
             grouping_labels: LabelSet::new(),
             accuracy_metric: "err".to_string(),
-            accuracy_tolerance: 1.0,
+            accuracy_sla: 1.0,
             accuracy_direction: crate::AccuracyDirection::LowerIsBetter,
+            latency_sla_ms: None,
         }
     }
 
     #[test]
-    fn two_rqes_can_share_one_deployment() {
-        let rqes = vec![rqe("a", 60, 3_600), rqe("b", 60, 3_600)];
+    fn two_raqes_can_share_one_deployment() {
+        let raqes = vec![raqe("a", 60_000, 3_600_000), raqe("b", 60_000, 3_600_000)];
         let table = vec![cost("cms-heap-topk-fastpath-vector2d")];
-        let deployments = build_all_candidates(&rqes, &table);
+        let deployments = build_all_candidates(&raqes, &table);
 
-        assert!(unservable(&rqes, &deployments).is_empty());
-        let mappings = brute_force(&rqes, &deployments);
+        assert!(unservable(&raqes, &deployments).is_empty());
+        let mappings = brute_force(&raqes, &deployments);
         assert!(
             mappings.iter().any(|m| m[0] == m[1]),
-            "expected at least one mapping where both identical RQEs land on the same deployment"
+            "expected at least one mapping where both identical RAQEs land on the same deployment"
         );
     }
 
     #[test]
-    fn unservable_rqe_yields_no_mappings() {
-        let rqes = vec![rqe("a", 60, 3_600)];
+    fn unservable_raqe_yields_no_mappings() {
+        let raqes = vec![raqe("a", 60_000, 3_600_000)];
         let table: Vec<AtomicCostEntry> = Vec::new(); // no configs at all
-        let deployments = build_all_candidates(&rqes, &table);
+        let deployments = build_all_candidates(&raqes, &table);
 
-        assert_eq!(unservable(&rqes, &deployments), vec!["a".to_string()]);
-        assert!(brute_force(&rqes, &deployments).is_empty());
+        assert_eq!(unservable(&raqes, &deployments), vec!["a".to_string()]);
+        assert!(brute_force(&raqes, &deployments).is_empty());
     }
 
     #[test]
     fn streaming_visits_the_same_mappings_as_eager_enumeration() {
-        let rqes = vec![rqe("a", 60, 3_600), rqe("b", 60, 3_600)];
-        let deployments = build_all_candidates(&rqes, &[cost("cms-heap-topk-fastpath-vector2d")]);
-        let eager = brute_force(&rqes, &deployments);
+        let raqes = vec![raqe("a", 60_000, 3_600_000), raqe("b", 60_000, 3_600_000)];
+        let deployments = build_all_candidates(&raqes, &[cost("cms-heap-topk-fastpath-vector2d")]);
+        let eager = brute_force(&raqes, &deployments);
         let mut streamed = Vec::new();
-        let count = for_each_mapping(&rqes, &deployments, |mapping| {
+        let count = for_each_mapping(&raqes, &deployments, |mapping| {
             streamed.push(mapping.clone())
         });
         assert_eq!(count as usize, eager.len());
@@ -183,10 +188,10 @@ mod tests {
 
     #[test]
     fn streaming_can_stop_after_a_fixed_number_of_mappings() {
-        let rqes = vec![rqe("a", 60, 3_600), rqe("b", 60, 3_600)];
-        let deployments = build_all_candidates(&rqes, &[cost("cms-heap-topk-fastpath-vector2d")]);
+        let raqes = vec![raqe("a", 60_000, 3_600_000), raqe("b", 60_000, 3_600_000)];
+        let deployments = build_all_candidates(&raqes, &[cost("cms-heap-topk-fastpath-vector2d")]);
         let mut sampled = Vec::new();
-        let result = for_each_mapping_while(&rqes, &deployments, |mapping| {
+        let result = for_each_mapping_while(&raqes, &deployments, |mapping| {
             sampled.push(mapping.clone());
             sampled.len() < 2
         });
diff --git a/rqe-optimizer/src/lib.rs b/rqe-optimizer/src/lib.rs
index 4fefad39..2561194b 100644
--- a/rqe-optimizer/src/lib.rs
+++ b/rqe-optimizer/src/lib.rs
@@ -1,9 +1,9 @@
-//! v1 brute-force solver for the RQE -> sketch-deployment mapping problem.
+//! v1 brute-force solver for the RAQE -> sketch-deployment mapping problem.
 //! `docs/rqe_sketch_deployment_v1.md` is the problem statement; doc comments
 //! cite its section numbers.
 //!
 //! Three independent stages, so an ILP can replace [`enumerate`] alone:
-//! [`candidates`] builds 𝒟 and per-RQE eligibility (§3), [`enumerate`]
+//! [`candidates`] builds 𝒟 and per-RAQE eligibility (§3), [`enumerate`]
 //! searches (§4), [`analytical_cost_model`] scores (§5).
 //!
 //! Every number the algorithm uses is caller-supplied, except the query
@@ -21,14 +21,19 @@ use std::collections::BTreeSet;
 
 pub use aqpbm_core::{AtomicCostEntry, AtomicCostTable};
 
-/// Whole seconds. Window alignment is expressed with exact `%` checks.
-pub type Seconds = u64;
+/// Whole milliseconds. Window alignment is expressed with exact `%` checks.
+pub type Millis = u64;
+
+/// Rates (CPU-sec/sec, samples/sec) stay per second.
+pub(crate) fn secs(ms: Millis) -> f64 {
+    ms as f64 / 1000.0
+}
 
 /// A group-by key, compared as a set (§3's `labels_i == labels_D` rule).
 pub type LabelSet = BTreeSet<String>;
 
-/// What an RQE's statistic needs (§1). `families()` is the only place
-/// capability scope is encoded; widening it means editing that match arm.
+/// What an RAQE's statistic needs (§1). A variant becomes a candidate only
+/// if it is in both `families()` and [`DEPLOYABLE_FAMILIES`].
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub enum Capability {
     SumOrCount,
@@ -41,6 +46,16 @@ pub enum Capability {
 }
 
 impl Capability {
+    pub const ALL: [Capability; 7] = [
+        Capability::SumOrCount,
+        Capability::Min,
+        Capability::Max,
+        Capability::RateOrIncrease,
+        Capability::Quantile,
+        Capability::Cardinality,
+        Capability::TopK,
+    ];
+
     /// `AtomicCostEntry.sketch` values serving this capability. These are
     /// sketch-bench *variant* strings, not algorithm names: fastpath and
     /// regularpath have different cost profiles, so `"cms"` would be
@@ -69,16 +84,41 @@ impl Capability {
             ],
         }
     }
+
+    /// The variants the MILP and AutoSketch plan with: the subset of
+    /// [`Capability::families`] in [`DEPLOYABLE_FAMILIES`], or all of them
+    /// when `allow_undeployable_families` (for studies; the plan can't be
+    /// deployed).
+    pub fn candidate_families(
+        self,
+        allow_undeployable_families: bool,
+    ) -> impl Iterator<Item = &'static str> {
+        self.families().iter().copied().filter(move |family| {
+            allow_undeployable_families || DEPLOYABLE_FAMILIES.contains(family)
+        })
+    }
 }
 
-/// Facts about one metric, given as input. RQEs and deployments on the same
+/// Variants ASAPQuery can deploy, each with the `AggregationType` its query
+/// engine runs. Both sides use `asap_sketchlib` for the sketches.
+pub const DEPLOYABLE_FAMILIES: &[&str] = &[
+    "exact-sum",                       // MultipleSum
+    "exact-min",                       // MultipleMinMax, sub_type "min"
+    "exact-max",                       // MultipleMinMax, sub_type "max"
+    "exact-increase",                  // MultipleIncrease
+    "kll-percall",                     // DatasketchesKLL
+    "hll",                             // HLL
+    "cms-heap-topk-fastpath-vector2d", // CountMinSketchWithHeap
+];
+
+/// Facts about one metric, given as input. RAQEs and deployments on the same
 /// metric read the same samples.
 #[derive(Debug, Clone)]
 pub struct MetricFacts {
     /// Every label the metric's series carry.
     pub labels: LabelSet,
     /// Each series yields one sample per scrape.
-    pub scrape_interval_secs: Seconds,
+    pub scrape_interval_ms: Millis,
     /// `card(X)`: distinct value combinations of each label set `X` in use.
     /// Must include `labels` itself, whose cardinality is the series count.
     pub cardinality: BTreeMap<LabelSet, u64>,
@@ -87,27 +127,47 @@ pub struct MetricFacts {
 impl MetricFacts {
     /// `λ`: samples/sec across all of the metric's series.
     pub fn arrival_rate_per_sec(&self) -> f64 {
-        self.cardinality[&self.labels] as f64 / self.scrape_interval_secs as f64
+        self.cardinality[&self.labels] as f64 / secs(self.scrape_interval_ms)
     }
 }
 
 /// [`MetricFacts`] keyed by metric name.
 pub type WorkloadFacts = BTreeMap<String, MetricFacts>;
 
-/// Every problem with `facts` for serving `rqes`. The rest of the crate
-/// indexes `facts` without checks, so call this first on caller input.
-pub fn validate_facts(rqes: &[Rqe], facts: &WorkloadFacts) -> Result<(), Vec<String>> {
+/// Every problem with `raqes` and the `facts` serving them. The rest of the
+/// crate indexes `facts` without checks, so call this first on caller input.
+pub fn validate_facts(raqes: &[Raqe], facts: &WorkloadFacts) -> Result<(), Vec<String>> {
     let mut problems = BTreeSet::new();
-    for rqe in rqes {
-        let (metric, grouping) = (&rqe.metric, &rqe.grouping_labels);
+    for raqe in raqes {
+        let id = &raqe.id;
+        if raqe.lookback_ms == 0 || raqe.interval_ms == 0 {
+            problems.insert(format!("{id}: lookback and interval must be nonzero"));
+        }
+        if !raqe.spatial_filter.is_empty() {
+            problems.insert(format!("{id}: spatial filters are not supported yet"));
+        }
+        if let Some(limit) = raqe.latency_sla_ms {
+            if !(limit.is_finite() && limit > 0.0) {
+                problems.insert(format!("{id}: latency SLA {limit} ms is not positive"));
+            }
+        }
+        let (metric, grouping) = (&raqe.metric, &raqe.grouping_labels);
         let Some(metric_facts) = facts.get(metric) else {
             problems.insert(format!("{metric}: no facts for this metric"));
             continue;
         };
         let all_labels = &metric_facts.labels;
         let cardinality = &metric_facts.cardinality;
-        if metric_facts.scrape_interval_secs == 0 {
+        let scrape_ms = metric_facts.scrape_interval_ms;
+        if scrape_ms == 0 {
             problems.insert(format!("{metric}: scrape interval is 0"));
+        } else if !(raqe.lookback_ms.is_multiple_of(scrape_ms)
+            && raqe.interval_ms.is_multiple_of(scrape_ms))
+        {
+            problems.insert(format!(
+                "{id}: lookback {} ms and interval {} ms must be multiples of the {scrape_ms} ms scrape interval",
+                raqe.lookback_ms, raqe.interval_ms
+            ));
         }
         if !grouping.is_subset(all_labels) {
             problems.insert(format!(
@@ -142,7 +202,7 @@ pub fn validate_facts(rqes: &[Rqe], facts: &WorkloadFacts) -> Result<(), Vec<Str
 /// (lower better); top-k reports `precision_at_k`/`recall_at_k` (higher
 /// better), so its tolerance is a floor.
 ///
-/// Named per RQE, never inferred from the metric name: an unanticipated name
+/// Named per RAQE, never inferred from the metric name: an unanticipated name
 /// would get a silent wrong default, and the failure mode is a hard
 /// constraint accepting exactly what it should reject.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -155,36 +215,46 @@ pub enum AccuracyDirection {
 
 /// One repeating query expression (§1).
 #[derive(Debug, Clone)]
-pub struct Rqe {
+pub struct Raqe {
     pub id: String,
     pub capability: Capability,
     /// `L_i`.
-    pub lookback_secs: Seconds,
-    /// `T_i`: how often this RQE is queried.
-    pub interval_secs: Seconds,
+    pub lookback_ms: Millis,
+    /// `T_i`: how often this RAQE is queried.
+    pub interval_ms: Millis,
     pub metric: String,
+    /// Series selector beyond the metric name, as canonical PromQL matcher
+    /// text; empty for none. Part of stream identity, compared as an opaque
+    /// string.
+    // ponytail: validate_facts rejects non-empty filters until facts carry
+    // per-filter cardinality; without it a filtered stream is costed as the
+    // whole metric.
+    pub spatial_filter: String,
     /// `G`: the query's group-by labels.
     pub grouping_labels: LabelSet,
     /// Which `AtomicCostEntry::query_accuracy` key to check. Comparators
-    /// name their metrics differently per capability, so the RQE picks.
+    /// name their metrics differently per capability, so the RAQE picks.
     pub accuracy_metric: String,
     /// `tol_i`: hard constraint. Read as a ceiling or a floor depending on
-    /// [`Rqe::accuracy_direction`].
-    pub accuracy_tolerance: f64,
+    /// [`Raqe::accuracy_direction`].
+    pub accuracy_sla: f64,
     pub accuracy_direction: AccuracyDirection,
+    /// Hard ceiling on modeled query latency; `None` for no limit. Only the
+    /// MILP enforces it.
+    pub latency_sla_ms: Option<f64>,
 }
 
-impl Rqe {
+impl Raqe {
     /// The one place comparison direction is decided -- both the
     /// candidate prefilter and eligibility route here so they can't drift.
     pub fn accuracy_ok(&self, value: f64) -> bool {
         match self.accuracy_direction {
-            AccuracyDirection::LowerIsBetter => value <= self.accuracy_tolerance,
-            AccuracyDirection::HigherIsBetter => value >= self.accuracy_tolerance,
+            AccuracyDirection::LowerIsBetter => value <= self.accuracy_sla,
+            AccuracyDirection::HigherIsBetter => value >= self.accuracy_sla,
         }
     }
 
-    /// Whether `config` measured this RQE's metric and clears it. Missing
+    /// Whether `config` measured this RAQE's metric and clears it. Missing
     /// is not passing.
     pub fn accuracy_ok_for(&self, config: &AtomicCostEntry) -> bool {
         config
@@ -201,43 +271,44 @@ impl Rqe {
 pub struct Deployment {
     pub capability: Capability,
     pub metric: String,
+    pub spatial_filter: String,
     /// `G`: one instance per group, per window.
     pub grouping_labels: LabelSet,
     pub config: AtomicCostEntry,
     /// `x`: materialized sketch-window size.
-    pub window_secs: Seconds,
+    pub window_ms: Millis,
     /// `y`: materialized sketch slide.
-    pub slide_secs: Seconds,
+    pub slide_ms: Millis,
 }
 
 impl Deployment {
     /// Number of active overlapping instances receiving each item.
     pub fn active_instance_count(&self) -> Option<u64> {
-        if self.slide_secs == 0 || !self.window_secs.is_multiple_of(self.slide_secs) {
+        if self.slide_ms == 0 || !self.window_ms.is_multiple_of(self.slide_ms) {
             return None;
         }
-        Some(self.window_secs / self.slide_secs)
+        Some(self.window_ms / self.slide_ms)
     }
 
     /// Number of non-overlapping instances merged for one query window.
-    pub fn query_instance_count(&self, lookback_secs: Seconds) -> Option<u64> {
-        if self.window_secs == 0 || !lookback_secs.is_multiple_of(self.window_secs) {
+    pub fn query_instance_count(&self, lookback_ms: Millis) -> Option<u64> {
+        if self.window_ms == 0 || !lookback_ms.is_multiple_of(self.window_ms) {
             return None;
         }
-        Some(lookback_secs / self.window_secs)
+        Some(lookback_ms / self.window_ms)
     }
 
     /// Closed instances kept to serve a lookback: those wholly inside it,
     /// which start in `[now − L, now − x]`, so `(L − x) / y + 1`.
-    pub fn closed_instance_count(&self, lookback_secs: Seconds) -> Option<u64> {
-        self.query_instance_count(lookback_secs)?;
+    pub fn closed_instance_count(&self, lookback_ms: Millis) -> Option<u64> {
+        self.query_instance_count(lookback_ms)?;
         self.active_instance_count()?;
-        Some((lookback_secs - self.window_secs) / self.slide_secs + 1)
+        Some((lookback_ms - self.window_ms) / self.slide_ms + 1)
     }
 }
 
 /// A full mapping: one deployment index (into the `deployments` slice passed
-/// to `enumerate::brute_force`) per RQE, aligned with the `rqes` slice's
+/// to `enumerate::brute_force`) per RAQE, aligned with the `raqes` slice's
 /// order. `z_{i,D} = 1 <=> mapping[i] == D`'s index; `u_D = 1 <=>` `D`'s
 /// index appears anywhere in `mapping` (§4) -- `y` is never stored
 /// separately, it's always derived from `mapping`.
@@ -259,23 +330,25 @@ pub(crate) mod test_support {
             METRIC.to_string(),
             MetricFacts {
                 labels: labels.clone(),
-                scrape_interval_secs: 1,
+                scrape_interval_ms: 1_000,
                 cardinality: BTreeMap::from([(LabelSet::new(), groups), (labels, series)]),
             },
         )])
     }
 
-    pub fn rqe(lookback_secs: Seconds, interval_secs: Seconds) -> Rqe {
-        Rqe {
+    pub fn raqe(lookback_ms: Millis, interval_ms: Millis) -> Raqe {
+        Raqe {
             id: "r".into(),
             capability: Capability::TopK,
-            lookback_secs,
-            interval_secs,
+            lookback_ms,
+            interval_ms,
             metric: METRIC.into(),
+            spatial_filter: String::new(),
             grouping_labels: LabelSet::new(),
             accuracy_metric: "err".into(),
-            accuracy_tolerance: 1.0,
+            accuracy_sla: 1.0,
             accuracy_direction: AccuracyDirection::LowerIsBetter,
+            latency_sla_ms: None,
         }
     }
 
@@ -285,12 +358,13 @@ pub(crate) mod test_support {
         insert: f64,
         merge: f64,
         query: f64,
-        window_secs: Seconds,
-        slide_secs: Seconds,
+        window_ms: Millis,
+        slide_ms: Millis,
     ) -> Deployment {
         Deployment {
             capability: Capability::TopK,
             metric: METRIC.into(),
+            spatial_filter: String::new(),
             grouping_labels: LabelSet::new(),
             config: AtomicCostEntry {
                 sketch: "cms-heap-topk-fastpath-vector2d".into(),
@@ -301,8 +375,8 @@ pub(crate) mod test_support {
                 query_cpu_secs: query,
                 query_accuracy: BTreeMap::from([("err".into(), 0.0)]),
             },
-            window_secs,
-            slide_secs,
+            window_ms,
+            slide_ms,
         }
     }
 }
@@ -310,26 +384,38 @@ pub(crate) mod test_support {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use test_support::{deployment, rqe, METRIC};
+    use test_support::{deployment, raqe, METRIC};
 
     fn labels(names: &[&str]) -> LabelSet {
         names.iter().map(|s| s.to_string()).collect()
     }
 
+    #[test]
+    fn every_deployable_family_serves_a_capability() {
+        for family in DEPLOYABLE_FAMILIES {
+            assert!(
+                Capability::ALL
+                    .iter()
+                    .any(|c| c.families().contains(family)),
+                "{family} serves no capability"
+            );
+        }
+    }
+
     #[test]
     fn closed_instances_lie_wholly_inside_the_lookback() {
         // 10-minute windows every minute, over an hour: starts in minutes
         // [now − 60, now − 10].
-        let sliding = deployment(1.0, 1.0, 1.0, 1.0, 600, 60);
-        assert_eq!(sliding.closed_instance_count(3_600), Some(51));
-        assert_eq!(sliding.closed_instance_count(600), Some(1));
-        assert_eq!(sliding.closed_instance_count(900), None); // not tiled by x
+        let sliding = deployment(1.0, 1.0, 1.0, 1.0, 600_000, 60_000);
+        assert_eq!(sliding.closed_instance_count(3_600_000), Some(51));
+        assert_eq!(sliding.closed_instance_count(600_000), Some(1));
+        assert_eq!(sliding.closed_instance_count(900_000), None); // not tiled by x
     }
 
     #[test]
     fn arrival_rate_is_series_over_scrape_interval() {
         let metric_facts = MetricFacts {
-            scrape_interval_secs: 15,
+            scrape_interval_ms: 15_000,
             ..test_support::facts(4, 600)[METRIC].clone()
         };
         assert_eq!(metric_facts.arrival_rate_per_sec(), 40.0);
@@ -341,16 +427,16 @@ mod tests {
             METRIC.to_string(),
             MetricFacts {
                 labels: labels(&["service", "endpoint"]),
-                scrape_interval_secs: 15,
+                scrape_interval_ms: 15_000,
                 cardinality: BTreeMap::from([
                     (labels(&["service"]), 5),
                     (labels(&["service", "endpoint"]), 50),
                 ]),
             },
         )]);
-        let r = Rqe {
+        let r = Raqe {
             grouping_labels: labels(&["service"]),
-            ..rqe(60, 60)
+            ..raqe(60_000, 60_000)
         };
         assert_eq!(validate_facts(&[r], &facts), Ok(()));
     }
@@ -361,28 +447,56 @@ mod tests {
             METRIC.to_string(),
             MetricFacts {
                 labels: labels(&["service"]),
-                scrape_interval_secs: 0,
+                scrape_interval_ms: 0,
                 cardinality: BTreeMap::from([(labels(&["service"]), 0)]),
             },
         )]);
-        let rqes = [
-            Rqe {
+        let raqes = [
+            Raqe {
                 metric: "missing".into(),
-                ..rqe(60, 60)
+                ..raqe(60_000, 60_000)
             },
-            Rqe {
+            Raqe {
                 grouping_labels: labels(&["pod"]),
-                ..rqe(60, 60)
+                ..raqe(60_000, 60_000)
+            },
+            Raqe {
+                id: "filtered".into(),
+                grouping_labels: labels(&["service"]),
+                spatial_filter: r#"{job="api"}"#.into(),
+                latency_sla_ms: Some(f64::NAN),
+                ..raqe(60_000, 60_000)
             },
         ];
-        let problems = validate_facts(&rqes, &facts).unwrap_err();
+        let problems = validate_facts(&raqes, &facts).unwrap_err();
         let has = |needle: &str| problems.iter().any(|p| p.contains(needle));
         assert!(has("missing: no facts"));
         assert!(has("scrape interval is 0"));
         assert!(has("not a subset"));
         assert!(has("no cardinality for {\"pod\"}"));
         assert!(has("cardinality of {\"service\"} is 0"));
-        assert_eq!(problems.len(), 5, "{problems:?}");
+        assert!(has("filtered: spatial filters are not supported"));
+        assert!(has("filtered: latency SLA NaN ms is not positive"));
+        assert_eq!(problems.len(), 7, "{problems:?}");
+    }
+
+    #[test]
+    fn rejects_zero_and_scrape_unaligned_times() {
+        // 1 s scrape. 3_600 looks like seconds left unconverted.
+        let raqes = [
+            Raqe {
+                id: "zero".into(),
+                ..raqe(0, 60_000)
+            },
+            Raqe {
+                id: "unaligned".into(),
+                ..raqe(3_600, 60_000)
+            },
+        ];
+        let problems = validate_facts(&raqes, &test_support::facts(1, 1)).unwrap_err();
+        assert_eq!(problems.len(), 2, "{problems:?}");
+        assert!(problems[0].starts_with("unaligned: lookback 3600 ms"));
+        assert!(problems[1].starts_with("zero: lookback and interval must be nonzero"));
     }
 
     #[test]
@@ -391,16 +505,16 @@ mod tests {
             METRIC.to_string(),
             MetricFacts {
                 labels: labels(&["service", "endpoint"]),
-                scrape_interval_secs: 15,
+                scrape_interval_ms: 15_000,
                 cardinality: BTreeMap::from([
                     (labels(&["service"]), 60),
                     (labels(&["service", "endpoint"]), 50),
                 ]),
             },
         )]);
-        let r = Rqe {
+        let r = Raqe {
             grouping_labels: labels(&["service"]),
-            ..rqe(60, 60)
+            ..raqe(60_000, 60_000)
         };
         let problems = validate_facts(&[r], &facts).unwrap_err();
         assert!(problems[0].contains("60 groups but only 50 series"));
diff --git a/rqe-optimizer/src/milp.rs b/rqe-optimizer/src/milp.rs
index d26abd06..5ebacdfe 100644
--- a/rqe-optimizer/src/milp.rs
+++ b/rqe-optimizer/src/milp.rs
@@ -1,18 +1,18 @@
-//! MILP solver for one scalarized RQE deployment optimization (§4).
+//! MILP solver for one scalarized RAQE deployment optimization (§4).
 //!
 //! The model is documented in `docs/rqe_sketch_deployment_v1.md`. It avoids
 //! enumerating the Cartesian product of eligible deployment choices.
 
 use crate::analytical_cost_model::{self, score, PhaseCost, PlanCost, BYTES_PER_GIB};
 use crate::candidates::eligible_deployments_for;
-use crate::{Deployment, Mapping, Rqe, WorkloadFacts};
+use crate::{Deployment, Mapping, Raqe, WorkloadFacts};
 use good_lp::{
     default_solver, variable, Expression, ProblemVariables, ResolutionError, Solution, SolverModel,
     Variable,
 };
 
 /// What a [`minimize`] solve minimizes. Eligibility, assignment and activation
-/// rows, and bounds are the same for every objective.
+/// rows, and latency SLAs are the same for every objective.
 #[derive(Debug, Clone, Copy, PartialEq)]
 pub enum Objective {
     /// `w_cpu · CPU + w_mem · memory`. CPU is mean CPU-sec/sec, the area under
@@ -45,43 +45,29 @@ impl Objective {
     }
 }
 
-/// Optional hard bounds for a solve. An empty latency vector means no RQE has
-/// a latency bound; otherwise it is index-aligned with `rqes`.
-#[derive(Debug, Clone, Default)]
-pub struct MilpBounds {
-    pub max_query_latency_secs: Vec<Option<f64>>,
-}
-
 #[derive(Debug, Clone)]
 pub struct MilpSolution {
     pub mapping: Mapping,
     pub plan_cost: PlanCost,
 }
 
-/// Minimize `objective` subject to optional latency bounds. `facts` must pass
+/// Minimize `objective` subject to each RAQE's latency SLA. `facts` must pass
 /// [`crate::validate_facts`]. Missing eligible deployments are a caller input
 /// error; use `enumerate::unservable` to report them before calling this
 /// function.
 pub fn minimize(
-    rqes: &[Rqe],
+    raqes: &[Raqe],
     deployments: &[Deployment],
     facts: &WorkloadFacts,
-    bounds: &MilpBounds,
     objective: Objective,
 ) -> Result<MilpSolution, ResolutionError> {
-    assert!(
-        bounds.max_query_latency_secs.is_empty()
-            || bounds.max_query_latency_secs.len() == rqes.len(),
-        "latency bounds must be empty or index-aligned with rqes"
-    );
-
-    let eligible: Vec<Vec<usize>> = rqes
+    let eligible: Vec<Vec<usize>> = raqes
         .iter()
-        .map(|rqe| eligible_deployments_for(rqe, deployments))
+        .map(|raqe| eligible_deployments_for(raqe, deployments))
         .collect();
     assert!(
         eligible.iter().all(|choices| !choices.is_empty()),
-        "cannot solve a workload with an unservable RQE"
+        "cannot solve a workload with an unservable RAQE"
     );
 
     let mut variables = ProblemVariables::new();
@@ -98,7 +84,7 @@ pub fn minimize(
                 .collect()
         })
         .collect();
-    // Weighted storage cost per deployment: the max over the RQEs it serves.
+    // Weighted storage cost per deployment: the max over the RAQEs it serves.
     let deployment_storage_cost: Vec<Variable> = deployments
         .iter()
         .map(|_| variables.add(variable().min(0)))
@@ -110,33 +96,33 @@ pub fn minimize(
         .iter()
         .map(|deployment| objective.weigh(analytical_cost_model::ingest(deployment, facts)))
         .collect();
-    let merge_and_query_cost = |rqe: &Rqe, deployment: &Deployment| {
-        objective.weigh(analytical_cost_model::merge(rqe, deployment, facts))
-            + objective.weigh(analytical_cost_model::query(rqe, deployment, facts))
+    let merge_and_query_cost = |raqe: &Raqe, deployment: &Deployment| {
+        objective.weigh(analytical_cost_model::merge(raqe, deployment, facts))
+            + objective.weigh(analytical_cost_model::query(raqe, deployment, facts))
     };
-    let storage_cost_for_rqe_and_deployment = |rqe: &Rqe, deployment: &Deployment| {
+    let storage_cost_for_raqe_and_deployment = |raqe: &Raqe, deployment: &Deployment| {
         objective.weigh(PhaseCost {
             cpu_secs_per_sec: 0.0,
-            memory_bytes: analytical_cost_model::storage_bytes(rqe, deployment, facts),
+            memory_bytes: analytical_cost_model::storage_bytes(raqe, deployment, facts),
         })
     };
 
     // Real plans cost ~1e-6 CPU-sec/sec and read in microseconds, below
     // HiGHS's absolute gap (1e-6) and feasibility tolerance (1e-7). So costs
-    // are divided by `reference`, a plan's cost without sharing (each RQE on
-    // its cheapest deployment), and per-RQE bounds become exclusions. The
+    // are divided by `reference`, a plan's cost without sharing (each RAQE on
+    // its cheapest deployment), and latency SLAs become exclusions. The
     // result is re-scored on real costs.
-    let reference: f64 = rqes
+    let reference: f64 = raqes
         .iter()
         .zip(&eligible)
-        .map(|(rqe, choices)| {
+        .map(|(raqe, choices)| {
             choices
                 .iter()
                 .map(|&deployment_index| {
                     let deployment = &deployments[deployment_index];
                     ingest_cost[deployment_index]
-                        + merge_and_query_cost(rqe, deployment)
-                        + storage_cost_for_rqe_and_deployment(rqe, deployment)
+                        + merge_and_query_cost(raqe, deployment)
+                        + storage_cost_for_raqe_and_deployment(raqe, deployment)
                 })
                 .fold(f64::INFINITY, f64::min)
         })
@@ -151,44 +137,39 @@ pub fn minimize(
     for (&cost, &active_variable) in ingest_cost.iter().zip(&active) {
         goal.add_mul(cost / reference, active_variable);
     }
-    for (rqe_index, choices) in assignments.iter().enumerate() {
+    for (raqe_index, choices) in assignments.iter().enumerate() {
         for &(deployment_index, assignment) in choices {
             goal.add_mul(
-                merge_and_query_cost(&rqes[rqe_index], &deployments[deployment_index]) / reference,
+                merge_and_query_cost(&raqes[raqe_index], &deployments[deployment_index])
+                    / reference,
                 assignment,
             );
         }
     }
 
     let mut model = variables.minimise(goal).using(default_solver);
-    for (rqe_index, choices) in assignments.iter().enumerate() {
+    for (raqe_index, choices) in assignments.iter().enumerate() {
         let assignment_sum: Expression = choices.iter().map(|(_, variable)| *variable).sum();
         model.add_constraint(assignment_sum.eq(1));
 
         for &(deployment_index, assignment) in choices {
-            let rqe = &rqes[rqe_index];
+            let raqe = &raqes[raqe_index];
             let deployment = &deployments[deployment_index];
             model.add_constraint((assignment - active[deployment_index]).leq(0));
             // Zero when `w_mem` is 0; such a row can never bind.
             let scaled_storage_cost =
-                storage_cost_for_rqe_and_deployment(rqe, deployment) / reference;
+                storage_cost_for_raqe_and_deployment(raqe, deployment) / reference;
             if scaled_storage_cost > 0.0 {
                 model.add_constraint(
                     (scaled_storage_cost * assignment - deployment_storage_cost[deployment_index])
                         .leq(0),
                 );
             }
-            // Each RQE takes exactly one deployment, so a per-RQE bound just
+            // Each RAQE takes exactly one deployment, so its latency SLA just
             // forbids the choices over it. Comparing in f64 here, not in the
             // solver, keeps µs latencies clear of its feasibility tolerance.
-            let latency = analytical_cost_model::query_latency_secs(rqe, deployment, facts);
-            if bounds
-                .max_query_latency_secs
-                .get(rqe_index)
-                .copied()
-                .flatten()
-                .is_some_and(|limit| latency > limit)
-            {
+            let latency_ms = analytical_cost_model::query_latency_ms(raqe, deployment, facts);
+            if raqe.latency_sla_ms.is_some_and(|limit| latency_ms > limit) {
                 model.add_constraint(Expression::from(assignment).leq(0));
             }
         }
@@ -213,10 +194,10 @@ pub fn minimize(
                 .find_map(|&(deployment_index, assignment)| {
                     (solved.value(assignment) > 0.5).then_some(deployment_index)
                 })
-                .expect("MILP assigns exactly one deployment to every RQE")
+                .expect("MILP assigns exactly one deployment to every RAQE")
         })
         .collect();
-    let plan_cost = score(rqes, deployments, &mapping, facts);
+    let plan_cost = score(raqes, deployments, &mapping, facts);
     Ok(MilpSolution { mapping, plan_cost })
 }
 
@@ -228,13 +209,13 @@ mod tests {
 
     /// Window and slide 60 s, merge 1 CPU-sec.
     fn deployment(insert_cpu_secs: f64, memory: f64, query_cpu_secs: f64) -> Deployment {
-        test_support::deployment(memory, insert_cpu_secs, 1.0, query_cpu_secs, 60, 60)
+        test_support::deployment(memory, insert_cpu_secs, 1.0, query_cpu_secs, 60_000, 60_000)
     }
 
-    fn rqe(id: &str, lookback_secs: u64) -> Rqe {
-        Rqe {
+    fn raqe(id: &str, lookback_ms: u64) -> Raqe {
+        Raqe {
             id: id.into(),
-            ..test_support::rqe(lookback_secs, 60)
+            ..test_support::raqe(lookback_ms, 60_000)
         }
     }
 
@@ -243,15 +224,18 @@ mod tests {
     }
 
     /// The MILP's plan scores as well as the best of every mapping.
-    fn assert_matches_brute_force(rqes: &[Rqe], deployments: &[Deployment], objective: Objective) {
+    fn assert_matches_brute_force(
+        raqes: &[Raqe],
+        deployments: &[Deployment],
+        objective: Objective,
+    ) {
         let facts = facts(1, 1);
-        let best = brute_force(rqes, deployments)
+        let best = brute_force(raqes, deployments)
             .iter()
-            .map(|mapping| objective.value(&score(rqes, deployments, mapping, &facts)))
+            .map(|mapping| objective.value(&score(raqes, deployments, mapping, &facts)))
             .min_by(f64::total_cmp)
             .expect("test workload is servable");
-        let milp = minimize(rqes, deployments, &facts, &MilpBounds::default(), objective)
-            .expect("feasible MILP");
+        let milp = minimize(raqes, deployments, &facts, objective).expect("feasible MILP");
         let got = objective.value(&milp.plan_cost);
         assert!(
             (got - best).abs() <= 1e-9 * best,
@@ -260,60 +244,49 @@ mod tests {
     }
 
     #[test]
-    fn minimizes_cpu_and_respects_a_latency_bound() {
-        let rqes = vec![rqe("r", 60)];
+    fn minimizes_cpu_and_respects_a_latency_sla() {
+        let raqes = vec![raqe("r", 60_000)];
         // CPU 1 + 3/60 with latency 3, or 2 + 1/60 with latency 1.
         let deployments = vec![deployment(1.0, 20.0, 3.0), deployment(2.0, 10.0, 1.0)];
-        let solve = |bounds: &MilpBounds| {
-            minimize(
-                &rqes,
-                &deployments,
-                &facts(1, 1),
-                bounds,
-                Objective::default(),
-            )
-            .expect("feasible MILP")
-            .mapping
+        let solve = |raqes: &[Raqe]| {
+            minimize(raqes, &deployments, &facts(1, 1), Objective::default())
+                .expect("feasible MILP")
+                .mapping
         };
 
-        assert_eq!(solve(&MilpBounds::default()), vec![0]);
-        let bounded = MilpBounds {
-            max_query_latency_secs: vec![Some(2.0)],
+        assert_eq!(solve(&raqes), vec![0]);
+        let bounded = Raqe {
+            latency_sla_ms: Some(2_000.0),
+            ..raqes[0].clone()
         };
-        assert_eq!(solve(&bounded), vec![1]);
+        assert_eq!(solve(&[bounded]), vec![1]);
     }
 
     #[test]
     fn matches_brute_force_with_shared_deployment_costs() {
-        let rqes = vec![rqe("frequent", 60), rqe("long", 600)];
+        let raqes = vec![raqe("frequent", 60_000), raqe("long", 600_000)];
         let deployments = vec![
             deployment(1.0, 0.5 * BYTES_PER_GIB, 400.0),
             deployment(3.0, 0.1 * BYTES_PER_GIB, 1.0),
             deployment(0.2, 2.0 * BYTES_PER_GIB, 1.0),
         ];
         for objective in [weights(1.0, 0.0), weights(0.0, 1.0), weights(1.0, 4.0)] {
-            assert_matches_brute_force(&rqes, &deployments, objective);
+            assert_matches_brute_force(&raqes, &deployments, objective);
         }
     }
 
     #[test]
     fn memory_weight_trades_cpu_for_memory() {
-        let rqes = vec![rqe("r", 60)];
+        let raqes = vec![raqe("r", 60_000)];
         // CPU-heavy and small, or CPU-light and twice the size.
         let deployments = vec![
             deployment(1.0, 1.0 * BYTES_PER_GIB, 0.0),
             deployment(0.05, 2.0 * BYTES_PER_GIB, 0.0),
         ];
         let solve = |objective| {
-            minimize(
-                &rqes,
-                &deployments,
-                &facts(1, 1),
-                &MilpBounds::default(),
-                objective,
-            )
-            .expect("feasible MILP")
-            .mapping
+            minimize(&raqes, &deployments, &facts(1, 1), objective)
+                .expect("feasible MILP")
+                .mapping
         };
 
         assert_eq!(solve(weights(1.0, 0.0)), vec![1]);
@@ -321,10 +294,10 @@ mod tests {
     }
 
     #[test]
-    fn tiny_magnitudes_match_brute_force_and_respect_latency_bounds() {
+    fn tiny_magnitudes_match_brute_force_and_respect_latency_slas() {
         // Real plans cost ~1e-6 CPU-sec/sec with µs latencies: below HiGHS's
         // absolute gap and feasibility tolerances unless the model is scaled.
-        let rqes = vec![rqe("frequent", 60), rqe("long", 600)];
+        let raqes = vec![raqe("frequent", 60_000), raqe("long", 600_000)];
         let tiny = 1e-9;
         let deployments = vec![
             deployment(1.0 * tiny, 0.5 * tiny * BYTES_PER_GIB, 400.0 * tiny),
@@ -338,27 +311,22 @@ mod tests {
         })
         .collect::<Vec<_>>();
         for objective in [weights(1.0, 0.0), weights(1.0, 4.0)] {
-            assert_matches_brute_force(&rqes, &deployments, objective);
+            assert_matches_brute_force(&raqes, &deployments, objective);
         }
 
-        // The cheap deployment's latency is 1.5e-7 s, over a 1e-7 s bound by
+        // The cheap deployment's latency is 1.5e-7 s, over a 1e-7 s SLA by
         // less than HiGHS's absolute feasibility tolerance.
-        let rqes = vec![rqe("r", 60)];
+        let raqes = vec![Raqe {
+            latency_sla_ms: Some(1e-4),
+            ..raqe("r", 60_000)
+        }];
         let deployments = vec![
             deployment(tiny, tiny, 1.5e-7),
             deployment(10.0 * tiny, tiny, 0.5e-7),
         ];
-        let milp = minimize(
-            &rqes,
-            &deployments,
-            &facts(1, 1),
-            &MilpBounds {
-                max_query_latency_secs: vec![Some(1e-7)],
-            },
-            Objective::default(),
-        )
-        .expect("feasible MILP");
+        let milp = minimize(&raqes, &deployments, &facts(1, 1), Objective::default())
+            .expect("feasible MILP");
         assert_eq!(milp.mapping, vec![1]);
-        assert!(milp.plan_cost.query_latency_secs[0] <= 1e-7);
+        assert!(milp.plan_cost.query_latency_ms[0] <= 1e-4);
     }
 }
diff --git a/rqe-optimizer/src/pareto.rs b/rqe-optimizer/src/pareto.rs
index 66d9e0e1..a3694c7d 100644
--- a/rqe-optimizer/src/pareto.rs
+++ b/rqe-optimizer/src/pareto.rs
@@ -1,4 +1,4 @@
-//! Non-dominated filtering for `(CPU, memory, per-RQE latency)`.
+//! Non-dominated filtering for `(CPU, memory, per-RAQE latency)`.
 
 use crate::{analytical_cost_model::PlanCost, Mapping};
 
@@ -17,7 +17,7 @@ fn dominates(a: &[f64], b: &[f64]) -> bool {
 
 fn pareto_vector(plan_cost: &PlanCost) -> Vec<f64> {
     let mut vector = vec![plan_cost.cpu_secs_per_sec(), plan_cost.memory_bytes()];
-    vector.extend(&plan_cost.query_latency_secs);
+    vector.extend(&plan_cost.query_latency_ms);
     vector
 }
 
@@ -87,7 +87,7 @@ mod tests {
             merge: PhaseCost::default(),
             query: PhaseCost::default(),
             storage: PhaseCost::default(),
-            query_latency_secs: vec![latency],
+            query_latency_ms: vec![latency],
         }
     }
 
diff --git a/scripts/export_rqe_optimizer_costs.sh b/scripts/export_rqe_optimizer_costs.sh
index e7c19e3f..bb21d618 100755
--- a/scripts/export_rqe_optimizer_costs.sh
+++ b/scripts/export_rqe_optimizer_costs.sh
@@ -2,8 +2,10 @@
 # Measure the small atomic-cost grid consumed by rqe-optimizer.
 #
 # This is intentionally separate from export_atomic_costs.sh: that script
-# serves ASAPQuery's pinned grid, while this one covers exactly the variants
-# named by rqe_optimizer::Capability::families().
+# serves ASAPQuery's pinned grid, while this one covers the variants named by
+# rqe_optimizer::Capability::families(). Only those also in
+# DEPLOYABLE_FAMILIES become candidates; the rest are measured for when the
+# engine gains them, and for research baselines.
 #
 # Usage:
 #   scripts/export_rqe_optimizer_costs.sh