diff --git a/Cargo.lock b/Cargo.lock
index e1d15191..50c54223 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -166,6 +166,7 @@ dependencies = [
  "chrono",
  "clap 4.6.1",
  "elastic_dsl_utilities",
+ "flate2",
  "indexmap 2.14.0",
  "pretty_assertions",
  "promql-parser",
diff --git a/asap-planner-rs/Cargo.toml b/asap-planner-rs/Cargo.toml
index 5e9bf33d..0c89320d 100644
--- a/asap-planner-rs/Cargo.toml
+++ b/asap-planner-rs/Cargo.toml
@@ -37,6 +37,7 @@ indexmap.workspace = true
 chrono.workspace = true
 promql-parser = "0.5.0"
 reqwest = { version = "0.11", features = ["blocking", "json"] }
+flate2 = "1.1"
 rqe-optimizer = { git = "https://github.com/ProjectASAP/sketch-bench", branch = "main" }
 
 [dev-dependencies]
diff --git a/asap-planner-rs/src/main.rs b/asap-planner-rs/src/main.rs
index 62d71e96..44b9f057 100644
--- a/asap-planner-rs/src/main.rs
+++ b/asap-planner-rs/src/main.rs
@@ -1,6 +1,6 @@
 use asap_planner::optimizer::{
-    parse_weight, plan_milp, plan_to_planner_output, reject_unwritable_queries, MilpConfig,
-    MilpInputs,
+    parse_weight, plan_milp, plan_to_planner_output, reject_unwritable_queries, saved_plan_output,
+    MilpConfig, MilpInputs, PlanSelector,
 };
 use asap_planner::{
     Controller, ControllerConfig, ElasticController, ElasticRuntimeOptions, RuntimeOptions,
@@ -59,10 +59,33 @@ struct Args {
 
     /// `milp` plans with sketch-bench's rqe-optimizer: PromQL or SQL with
     /// --input_config only, labels from its `metrics:` hints or tables'
-    /// `metadata_columns`.
+    /// `metadata_columns`. `saved` writes a PromQL plan sketch-bench's
+    /// `autosketch_vs_asap` already solved.
     #[arg(long, value_enum, default_value = "legacy")]
     planner: PlannerArg,
 
+    /// Saved only. sketch-bench `autosketch_vs_asap` plan file
+    /// (`<table>.plans.json.gz`).
+    #[arg(long, required_if_eq("planner", "saved"))]
+    plan: Option<PathBuf>,
+
+    /// Saved only. The eval workload table the plan was solved for; its
+    /// `e2e` block gives the PromQL queries and metric labels.
+    #[arg(long = "workload-table", required_if_eq("planner", "saved"))]
+    workload_table: Option<PathBuf>,
+
+    /// Saved only. The plan's method, e.g. `asap`.
+    #[arg(long, required_if_eq("planner", "saved"))]
+    method: Option<String>,
+
+    /// Saved only. The plan's objective weights, e.g. `cpu` or `fargate`.
+    #[arg(long, required_if_eq("planner", "saved"))]
+    weights: Option<String>,
+
+    /// Saved only. The plan at this latency SLA; without it, the cheapest.
+    #[arg(long = "sla-ms")]
+    sla_ms: Option<f64>,
+
     /// MILP only. YAML workload facts: per metric, positive `value_range` and
     /// `cardinality` per label set, including all labels (the series count),
     /// plus the `shape` of each grouping sketches may serve.
@@ -101,6 +124,7 @@ enum EngineArg {
 enum PlannerArg {
     Legacy,
     Milp,
+    Saved,
 }
 
 fn main() -> anyhow::Result<()> {
@@ -118,21 +142,38 @@ fn main() -> anyhow::Result<()> {
         EngineArg::Precompute => StreamingEngine::Precompute,
     };
 
-    if args.planner == PlannerArg::Milp {
-        run_milp(&args)?;
-        println!("Generated configs in {}", args.output_dir.display());
-        return Ok(());
+    if args.planner != PlannerArg::Saved {
+        anyhow::ensure!(
+            args.plan.is_none()
+                && args.workload_table.is_none()
+                && args.method.is_none()
+                && args.weights.is_none()
+                && args.sla_ms.is_none(),
+            "--plan, --workload-table, --method, --weights and --sla-ms require --planner saved"
+        );
     }
-    anyhow::ensure!(
-        args.workload_facts.is_none()
-            && args.atomic_costs.is_none()
-            && args.saturation_dir.is_none()
-            && args.w_cpu.is_none()
-            && args.w_mem.is_none(),
-        "--workload-facts, --atomic-costs, --saturation-dir, --w-cpu and --w-mem require \
-         --planner milp"
-    );
+    if args.planner != PlannerArg::Milp {
+        anyhow::ensure!(
+            args.workload_facts.is_none()
+                && args.atomic_costs.is_none()
+                && args.saturation_dir.is_none()
+                && args.w_cpu.is_none()
+                && args.w_mem.is_none(),
+            "--workload-facts, --atomic-costs, --saturation-dir, --w-cpu and --w-mem require \
+             --planner milp"
+        );
+    }
+    let output_dir = args.output_dir.clone();
+    match args.planner {
+        PlannerArg::Milp => run_milp(&args)?,
+        PlannerArg::Saved => run_saved(&args)?,
+        PlannerArg::Legacy => run_legacy(args, engine)?,
+    }
+    println!("Generated configs in {}", output_dir.display());
+    Ok(())
+}
 
+fn run_legacy(args: Args, engine: StreamingEngine) -> anyhow::Result<()> {
     match args.query_language {
         QueryLanguage::promql => {
             let scrape_interval_ms = args.data_ingestion_interval_ms.ok_or_else(|| {
@@ -210,47 +251,49 @@ fn main() -> anyhow::Result<()> {
             ElasticController::from_file(&config_path, opts)?.generate_to_dir(&args.output_dir)?;
         }
     }
-
-    println!("Generated configs in {}", args.output_dir.display());
     Ok(())
 }
 
-fn run_milp(args: &Args) -> anyhow::Result<()> {
-    anyhow::ensure!(
-        matches!(
-            args.query_language,
-            QueryLanguage::promql | QueryLanguage::sql
-        ),
-        "--planner milp supports only --query-language promql or sql"
-    );
+/// Checks the flags `--planner milp` and `--planner saved` share: labels from
+/// the config, no legacy-only flags. Returns the
+/// positive scrape interval, which per-series sample rates divide by.
+fn check_optimizer_flags(args: &Args, planner: &str) -> anyhow::Result<u64> {
     anyhow::ensure!(
         args.query_log.is_none(),
-        "--planner milp needs --input_config: query logs have no label hints"
+        "--planner {planner} doesn't take --query-log: query logs have no label hints"
     );
     anyhow::ensure!(
         args.prometheus_url.is_none(),
-        "--planner milp takes labels from the config, not --prometheus-url"
+        "--planner {planner} takes labels from the config, not --prometheus-url"
     );
     anyhow::ensure!(
         !args.enable_punting && args.range_duration_ms == 0 && args.step_ms == 0,
-        "--enable-punting, --range-duration-ms and --step-ms don't apply to --planner milp"
+        "--enable-punting, --range-duration-ms and --step-ms don't apply to --planner {planner}"
     );
     anyhow::ensure!(
         args.clickhouse_url.is_none() && args.clickhouse_database.is_none(),
-        "--clickhouse-url and --clickhouse-database don't apply to --planner milp"
+        "--clickhouse-url and --clickhouse-database don't apply to --planner {planner}"
     );
+    args.data_ingestion_interval_ms
+        .filter(|&ms| ms > 0)
+        .ok_or_else(|| {
+            anyhow::anyhow!("--planner {planner} requires a positive --data-ingestion-interval-ms")
+        })
+}
+
+fn run_milp(args: &Args) -> anyhow::Result<()> {
+    anyhow::ensure!(
+        matches!(
+            args.query_language,
+            QueryLanguage::promql | QueryLanguage::sql
+        ),
+        "--planner milp supports only --query-language promql or sql"
+    );
+    let scrape_interval_ms = check_optimizer_flags(args, "milp")?;
     let config_path = args
         .input_config
         .as_deref()
         .ok_or_else(|| anyhow::anyhow!("--planner milp requires --input_config"))?;
-    let scrape_interval_ms = args
-        .data_ingestion_interval_ms
-        .ok_or_else(|| anyhow::anyhow!("--planner milp requires --data-ingestion-interval-ms"))?;
-    // Per-series sample rates divide by it.
-    anyhow::ensure!(
-        scrape_interval_ms > 0,
-        "--data-ingestion-interval-ms must be positive"
-    );
     let yaml = std::fs::read_to_string(config_path)?;
     let (promql, sql): (ControllerConfig, SQLControllerConfig);
     let config = if args.query_language == QueryLanguage::sql {
@@ -287,3 +330,31 @@ fn run_milp(args: &Args) -> anyhow::Result<()> {
         .write_to_dir(&args.output_dir)?;
     Ok(())
 }
+
+fn run_saved(args: &Args) -> anyhow::Result<()> {
+    anyhow::ensure!(
+        args.query_language == QueryLanguage::promql,
+        "--planner saved supports only --query-language promql"
+    );
+    let scrape_interval_ms = check_optimizer_flags(args, "saved")?;
+    anyhow::ensure!(
+        args.input_config.is_none(),
+        "--planner saved reads its queries from --workload-table, not --input_config"
+    );
+    saved_plan_output(
+        args.workload_table
+            .as_deref()
+            .expect("clap requires --workload-table with --planner saved"),
+        args.plan
+            .as_deref()
+            .expect("clap requires --plan with --planner saved"),
+        &PlanSelector {
+            method: args.method.as_deref().expect("clap requires --method"),
+            weights: args.weights.as_deref().expect("clap requires --weights"),
+            sla_ms: args.sla_ms,
+        },
+        scrape_interval_ms,
+    )?
+    .write_to_dir(&args.output_dir)?;
+    Ok(())
+}
diff --git a/asap-planner-rs/src/optimizer/aqe_extractor.rs b/asap-planner-rs/src/optimizer/aqe_extractor.rs
index 0ef3c933..f7f444ff 100644
--- a/asap-planner-rs/src/optimizer/aqe_extractor.rs
+++ b/asap-planner-rs/src/optimizer/aqe_extractor.rs
@@ -246,7 +246,7 @@ fn normalized_f64_bits(value: f64) -> u64 {
 /// dropped — they contribute no AQE. Only arithmetic operators are split;
 /// comparison and set operators are left as-is (treated as opaque leaves).
 /// An `avg` leaf becomes its sum and count leaves (see `rewrite_avg`).
-fn decompose_to_leaves(query: &str) -> Vec<String> {
+pub(crate) fn decompose_to_leaves(query: &str) -> Vec<String> {
     let (lhs, rhs) = match parse_binary_arms(query) {
         Some(arms) => arms,
         None => return rewrite_avg(query),
diff --git a/asap-planner-rs/src/optimizer/milp.rs b/asap-planner-rs/src/optimizer/milp.rs
index 7480c543..655a92bd 100644
--- a/asap-planner-rs/src/optimizer/milp.rs
+++ b/asap-planner-rs/src/optimizer/milp.rs
@@ -334,6 +334,17 @@ pub fn build_milp_workload(
     config: MilpConfig,
     facts: &WorkloadFacts,
     scrape_interval_ms: u64,
+) -> Result<MilpWorkload, MilpError> {
+    let workload = workload_from_config(config, scrape_interval_ms)?;
+    validate_facts(&workload.raqes, facts).map_err(MilpError::InvalidInputs)?;
+    Ok(workload)
+}
+
+/// [`build_milp_workload`] without the facts check, for a plan solved
+/// elsewhere.
+pub fn workload_from_config(
+    config: MilpConfig,
+    scrape_interval_ms: u64,
 ) -> Result<MilpWorkload, MilpError> {
     let mut items = config.items(scrape_interval_ms)?;
     // Stable Raqe order and ids across runs.
@@ -377,10 +388,8 @@ pub fn build_milp_workload(
     tracing::debug!(
         items = items.len(),
         raqes = raqes.len(),
-        metrics_with_facts = facts.len(),
         "milp inputs: built raqes"
     );
-    validate_facts(&raqes, facts).map_err(MilpError::InvalidInputs)?;
     Ok(MilpWorkload {
         items,
         raqes,
diff --git a/asap-planner-rs/src/optimizer/milp_output.rs b/asap-planner-rs/src/optimizer/milp_output.rs
index dbe77f08..e45e2733 100644
--- a/asap-planner-rs/src/optimizer/milp_output.rs
+++ b/asap-planner-rs/src/optimizer/milp_output.rs
@@ -94,14 +94,36 @@ pub fn plan_to_planner_output(
     workload: &MilpWorkload,
     solution: &MilpSolution,
 ) -> Result<PlannerOutput, MilpOutputError> {
+    let deployments: Vec<Deployment> = solution
+        .deployments
+        .iter()
+        .map(|planned| planned.deployment.clone())
+        .collect();
+    let served_by: Vec<usize> = solution.raqes.iter().map(|r| r.deployment).collect();
+    deployments_to_planner_output(config, workload, &deployments, &served_by)
+}
+
+/// [`plan_to_planner_output`] for a plan given as its deployments and, per
+/// Raqe of `workload`, the index of the deployment serving it.
+pub fn deployments_to_planner_output(
+    config: MilpConfig,
+    workload: &MilpWorkload,
+    deployments: &[Deployment],
+    served_by: &[usize],
+) -> Result<PlannerOutput, MilpOutputError> {
+    assert_eq!(
+        served_by.len(),
+        workload.raqes.len(),
+        "one deployment per Raqe"
+    );
     reject_unwritable_queries(config)?;
     // sketch-bench lets a deployment serve a RAQE grouped by a subset of its
     // grouping (a roll-up, sketch-bench#190). The engine reads a deployment
     // only at its own grouping, and `aggregation_config` takes the labels
     // from the first RAQE served, so refuse a roll-up rather than write
     // configs at the wrong grouping.
-    for (raqe, planned) in workload.raqes.iter().zip(&solution.raqes) {
-        let deployment = &solution.deployments[planned.deployment].deployment;
+    for (raqe, &index) in workload.raqes.iter().zip(served_by) {
+        let deployment = &deployments[index];
         if raqe.grouping_labels != deployment.grouping_labels {
             return Err(MilpOutputError::RollupUnsupported {
                 query: raqe.id.clone(),
@@ -113,11 +135,11 @@ pub fn plan_to_planner_output(
 
     let item_of = |raqe: usize| &workload.items[workload.raqe_items[raqe]];
     // Each item once per deployment, however many occurrences it has.
-    let mut served: Vec<Vec<&OptimizerItem>> = vec![Vec::new(); solution.deployments.len()];
+    let mut served: Vec<Vec<&OptimizerItem>> = vec![Vec::new(); deployments.len()];
     let mut seen = HashSet::new();
-    for (raqe, planned) in solution.raqes.iter().enumerate() {
-        if seen.insert((planned.deployment, workload.raqe_items[raqe])) {
-            served[planned.deployment].push(item_of(raqe));
+    for (raqe, &index) in served_by.iter().enumerate() {
+        if seen.insert((index, workload.raqe_items[raqe])) {
+            served[index].push(item_of(raqe));
         }
     }
 
@@ -131,8 +153,8 @@ pub fn plan_to_planner_output(
 
     let mut aggregations: IndexMap<String, IntermediateAggConfig> = IndexMap::new();
     let mut deployment_keys = Vec::with_capacity(served.len());
-    for (planned, items) in solution.deployments.iter().zip(&served) {
-        let mut aggregation = aggregation_config(&planned.deployment, items)?;
+    for (deployment, items) in deployments.iter().zip(&served) {
+        let mut aggregation = aggregation_config(deployment, items)?;
         if let Some(&(table, column)) = sql_columns.get(&aggregation.metric) {
             aggregation.metric = table.to_string();
             aggregation.table_name = Some(table.to_string());
@@ -144,8 +166,8 @@ pub fn plan_to_planner_output(
     }
 
     let mut queries: IndexMap<String, QueryPlanEntry> = IndexMap::new();
-    for (raqe, planned) in solution.raqes.iter().enumerate() {
-        let key = &deployment_keys[planned.deployment];
+    for (raqe, &index) in served_by.iter().enumerate() {
+        let key = &deployment_keys[index];
         for query in &item_of(raqe).query_strings {
             match queries.entry(query.clone()) {
                 Entry::Vacant(entry) => {
diff --git a/asap-planner-rs/src/optimizer/mod.rs b/asap-planner-rs/src/optimizer/mod.rs
index b3321be1..9ba74f44 100644
--- a/asap-planner-rs/src/optimizer/mod.rs
+++ b/asap-planner-rs/src/optimizer/mod.rs
@@ -3,6 +3,7 @@ pub mod atomic_costs;
 pub mod error;
 pub mod milp;
 pub mod milp_output;
+pub mod saved_plan;
 pub mod solution;
 pub mod workload_facts;
 
@@ -10,9 +11,13 @@ pub use aqe_extractor::{extract_aqes, RQE};
 pub use atomic_costs::{load_flat_atomic_cost_table, AtomicCostEntry, AtomicCostTable};
 pub use error::OptimizerError;
 pub use milp::{
-    build_milp_workload, parse_weight, plan_milp, solve_milp, MilpConfig, MilpError, MilpInputs,
-    MilpPlan, MilpWorkload,
+    build_milp_workload, parse_weight, plan_milp, solve_milp, workload_from_config, MilpConfig,
+    MilpError, MilpInputs, MilpPlan, MilpWorkload,
 };
-pub use milp_output::{plan_to_planner_output, reject_unwritable_queries, MilpOutputError};
+pub use milp_output::{
+    deployments_to_planner_output, plan_to_planner_output, reject_unwritable_queries,
+    MilpOutputError,
+};
+pub use saved_plan::{saved_plan_output, PlanSelector};
 pub use solution::OptimizerItem;
 pub use workload_facts::{load_workload_facts, parse_workload_facts, WorkloadFactsError};
diff --git a/asap-planner-rs/src/optimizer/saved_plan.rs b/asap-planner-rs/src/optimizer/saved_plan.rs
new file mode 100644
index 00000000..4a75e352
--- /dev/null
+++ b/asap-planner-rs/src/optimizer/saved_plan.rs
@@ -0,0 +1,495 @@
+//! Writes a plan solved elsewhere, sketch-bench `autosketch_vs_asap`'s plan
+//! file, as the planner's streaming and inference YAML, without re-solving.
+//! The workload comes from the eval table's `e2e` block: its PromQL queries,
+//! the plan RQEs each reads, and the metrics' labels.
+
+use std::collections::{BTreeMap, HashMap};
+use std::io::Read;
+use std::path::Path;
+
+use anyhow::{bail, ensure, Context};
+use rqe_optimizer::{heap_capacity, AtomicCostEntry, Deployment, LabelSet, Raqe};
+use serde::Deserialize;
+use serde_json::{json, Value};
+
+use crate::config::input::ControllerConfig;
+use crate::planner_output::PlannerOutput;
+
+use super::aqe_extractor::decompose_to_leaves;
+use super::milp::{workload_from_config, MilpConfig};
+use super::milp_output::{deployments_to_planner_output, reject_unwritable_queries};
+
+#[derive(Deserialize)]
+struct Table {
+    workloads: Vec<TableWorkload>,
+}
+
+#[derive(Deserialize)]
+struct TableWorkload {
+    e2e: Option<E2e>,
+}
+
+#[derive(Deserialize)]
+struct E2e {
+    queries: Vec<E2eQuery>,
+    metrics: BTreeMap<String, E2eMetric>,
+}
+
+#[derive(Deserialize)]
+struct E2eQuery {
+    promql: String,
+    interval_secs: u64,
+    /// Plan RQE ids, one per leaf of `promql` in order.
+    rqes: Vec<String>,
+}
+
+#[derive(Deserialize)]
+struct E2eMetric {
+    labels: Vec<E2eLabel>,
+}
+
+#[derive(Deserialize)]
+struct E2eLabel {
+    name: String,
+}
+
+#[derive(Deserialize)]
+struct PlanFile {
+    plans: Vec<Plan>,
+}
+
+#[derive(Deserialize)]
+struct Plan {
+    method: String,
+    weights: String,
+    kind: String,
+    sla_ms: Option<f64>,
+    chosen: Vec<Chosen>,
+}
+
+#[derive(Deserialize)]
+struct Chosen {
+    rqe: String,
+    sketch: String,
+    params: Value,
+    window_secs: u64,
+    slide_secs: u64,
+    /// Index of the plan's deployment.
+    deployment: usize,
+    /// Present only when the deployment's grouping differs from the RQE's
+    /// (a roll-up), which the converter refuses.
+    deployment_grouping: Option<LabelSet>,
+}
+
+/// Which plan of the file to write.
+pub struct PlanSelector<'a> {
+    pub method: &'a str,
+    pub weights: &'a str,
+    /// `None` for the cheapest plan, else the plan at this latency SLA.
+    pub sla_ms: Option<f64>,
+}
+
+/// The controller config the table's `e2e` block describes: one query group
+/// per repeat interval.
+fn e2e_config(e2e: &E2e) -> anyhow::Result<ControllerConfig> {
+    let mut by_interval: BTreeMap<u64, Vec<&str>> = BTreeMap::new();
+    for q in &e2e.queries {
+        by_interval
+            .entry(q.interval_secs)
+            .or_default()
+            .push(&q.promql);
+    }
+    let config = json!({
+        "query_groups": by_interval.iter().map(|(secs, queries)| json!({
+            "queries": queries,
+            "repetition_delay_ms": secs * 1000,
+        })).collect::<Vec<_>>(),
+        "metrics": e2e.metrics.iter().map(|(metric, m)| json!({
+            "metric": metric,
+            "labels": m.labels.iter().map(|l| &l.name).collect::<Vec<_>>(),
+        })).collect::<Vec<_>>(),
+    });
+    Ok(serde_json::from_value(config)?)
+}
+
+fn read_json<T: serde::de::DeserializeOwned>(path: &Path) -> anyhow::Result<T> {
+    let bytes = std::fs::read(path).with_context(|| format!("reading {}", path.display()))?;
+    let text = if bytes.starts_with(&[0x1f, 0x8b]) {
+        let mut text = String::new();
+        flate2::read::GzDecoder::new(&bytes[..]).read_to_string(&mut text)?;
+        text
+    } else {
+        String::from_utf8(bytes)?
+    };
+    serde_json::from_str(&text).with_context(|| format!("parsing {}", path.display()))
+}
+
+/// Streaming and inference YAML for the selected plan of `plan_path` (JSON,
+/// gzipped or not) over the workload of `table_path`.
+pub fn saved_plan_output(
+    table_path: &Path,
+    plan_path: &Path,
+    selector: &PlanSelector,
+    scrape_interval_ms: u64,
+) -> anyhow::Result<PlannerOutput> {
+    let table: Table = read_json(table_path)?;
+    let [TableWorkload { e2e: Some(e2e) }] = table.workloads.as_slice() else {
+        bail!(
+            "{}: expected one workload with an `e2e` block",
+            table_path.display()
+        );
+    };
+    let plans: PlanFile = read_json(plan_path)?;
+    let kind = if selector.sla_ms.is_some() {
+        "sla"
+    } else {
+        "cheapest"
+    };
+    let matching: Vec<&Plan> = plans
+        .plans
+        .iter()
+        .filter(|p| {
+            p.method == selector.method
+                && p.weights == selector.weights
+                && p.kind == kind
+                && p.sla_ms == selector.sla_ms
+        })
+        .collect();
+    let [plan] = matching.as_slice() else {
+        bail!(
+            "{}: {} plans match method {:?}, weights {:?}, {kind} {:?}; expected one",
+            plan_path.display(),
+            matching.len(),
+            selector.method,
+            selector.weights,
+            selector.sla_ms,
+        );
+    };
+    let mut chosen: HashMap<&str, &Chosen> = HashMap::new();
+    for c in &plan.chosen {
+        ensure!(
+            chosen.insert(&c.rqe, c).is_none(),
+            "the plan picks RQE {:?} twice",
+            c.rqe
+        );
+    }
+    let promql = e2e_config(e2e)?;
+    let config = MilpConfig::Promql(&promql);
+    // Before the leaf split, which turns an avg into two leaves.
+    reject_unwritable_queries(config)?;
+
+    // Leaf query string and repeat interval -> the plan RQE it reads.
+    let mut rqe_of: HashMap<(String, u64), &str> = HashMap::new();
+    for q in &e2e.queries {
+        let leaves = decompose_to_leaves(&q.promql);
+        ensure!(
+            leaves.len() == q.rqes.len(),
+            "query {:?} has {} leaves but names {} RQEs",
+            q.promql,
+            leaves.len(),
+            q.rqes.len()
+        );
+        for (leaf, rqe) in leaves.into_iter().zip(&q.rqes) {
+            let key = (leaf, q.interval_secs * 1000);
+            if let Some(other) = rqe_of.insert(key.clone(), rqe) {
+                ensure!(
+                    other == rqe,
+                    "leaf {:?} every {} ms reads both {other:?} and {rqe:?}",
+                    key.0,
+                    key.1
+                );
+            }
+        }
+    }
+
+    let workload = workload_from_config(config, scrape_interval_ms)?;
+
+    // Plan deployment index -> its position in `deployments`, in order of
+    // the first Raqe each serves.
+    let mut position: HashMap<usize, usize> = HashMap::new();
+    let mut deployments: Vec<Deployment> = Vec::new();
+    let mut served_by = Vec::with_capacity(workload.raqes.len());
+    for (raqe, &item) in workload.raqes.iter().zip(&workload.raqe_items) {
+        let item = &workload.items[item];
+        // An item's leaves (say p90 and p50 over one window) read one
+        // deployment, so their picks must name the same sketch.
+        let mut index = None;
+        for leaf in &item.query_strings {
+            let rqe = rqe_of
+                .get(&(leaf.clone(), item.t_repeat_ms))
+                .with_context(|| format!("leaf {leaf:?} has no RQE in the table's e2e block"))?;
+            let pick = chosen
+                .get(rqe)
+                .with_context(|| format!("the plan doesn't serve RQE {rqe:?}"))?;
+            let deployment = Deployment {
+                capability: raqe.capability,
+                metric: raqe.metric.clone(),
+                spatial_filter: raqe.spatial_filter.clone(),
+                grouping_labels: pick
+                    .deployment_grouping
+                    .clone()
+                    .unwrap_or_else(|| raqe.grouping_labels.clone()),
+                config: cost_entry(pick)?,
+                window_ms: pick.window_secs * 1000,
+                slide_ms: pick.slide_secs * 1000,
+                key_tracker: None,
+            };
+            ensure!(
+                can_serve(raqe, &deployment),
+                "RQE {rqe:?}'s {} {}s window sliding every {}s can't serve {leaf:?} \
+                 ({} ms lookback every {} ms); does --plan match --workload-table?",
+                pick.sketch,
+                pick.window_secs,
+                pick.slide_secs,
+                raqe.lookback_ms,
+                raqe.interval_ms
+            );
+            let at = *position.entry(pick.deployment).or_insert_with(|| {
+                deployments.push(deployment.clone());
+                deployments.len() - 1
+            });
+            // One plan deployment is one sketch: every Raqe it serves must agree.
+            ensure!(
+                deployments[at] == deployment,
+                "plan deployment {} serves Raqes that need different sketches: {:?} and {deployment:?}",
+                pick.deployment,
+                deployments[at]
+            );
+            ensure!(
+                index.is_none_or(|index| index == at),
+                "query {:?} is served by two deployments",
+                item.query_strings
+            );
+            index = Some(at);
+        }
+        let index = index.expect("an item has a query string");
+        served_by.push(index);
+    }
+    Ok(deployments_to_planner_output(
+        config,
+        &workload,
+        &deployments,
+        &served_by,
+    )?)
+}
+
+/// The rules of sketch-bench's `is_eligible` that need no workload facts or
+/// accuracy: a pick that breaks them belongs to another workload table.
+/// Grouping is left to the converter's roll-up check.
+fn can_serve(r: &Raqe, d: &Deployment) -> bool {
+    let properties = d.properties();
+    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)
+        && (properties.mergeable_across_windows || r.lookback_ms == d.window_ms)
+        && !properties.needs_delta_set_key_tracker
+        && heap_capacity(&d.config).is_none_or(|heap| {
+            d.heap_needed(r.lookback_ms, r.topk_k())
+                .is_some_and(|need| heap >= need)
+        })
+}
+
+/// The sketch config the YAML is written from. Costs and accuracy don't
+/// reach the YAML, so they are zero.
+fn cost_entry(pick: &Chosen) -> anyhow::Result<AtomicCostEntry> {
+    Ok(serde_json::from_value(json!({
+        "sketch": pick.sketch,
+        "sketch_config": {"algorithm": pick.sketch, "params": pick.params},
+        "mem_bytes_per_instance": 0.0,
+        "insert_cpu_secs": 0.0,
+        "merge_cpu_secs": 0.0,
+        "query_cpu_secs": 0.0,
+        "query_accuracy": {},
+        "accuracy_metric": "",
+        "measured_at": {
+            "items_per_instance": 0,
+            "keys_per_instance": null,
+            "value_range": null,
+            "merge_operand_items": null,
+            "distribution": null,
+        },
+    }))?)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const TABLE: &str = r#"{"workloads": [{"e2e": {
+        "queries": [
+            {"promql": "sum by (job) (data)", "interval_secs": 1, "rqes": ["sum"]},
+            {"promql": "quantile_over_time(0.9, data[5m]) / quantile_over_time(0.5, data[5m])",
+             "interval_secs": 60, "rqes": ["p90", "p50"]},
+            {"promql": "quantile_over_time(0.5, data[5m])", "interval_secs": 60, "rqes": ["p50"]}
+        ],
+        "metrics": {"data": {"type": "gauge", "labels": [
+            {"name": "job", "cardinality": 10},
+            {"name": "label_0", "cardinality": 1000, "child_of": "job"}
+        ]}}
+    }}]}"#;
+
+    fn pick(rqe: &str, deployment: usize, sketch: &str, params: Value, window: u64) -> Value {
+        json!({"rqe": rqe, "deployment": deployment, "sketch": sketch, "params": params,
+               "window_secs": window, "slide_secs": 60.min(window)})
+    }
+
+    /// The cheapest `asap`/`cpu` plan: the ratio's legs share one sliding
+    /// DDSketch. The SLA plan differs only in alpha.
+    fn plans(p50: Value) -> Value {
+        let dd = |alpha: f64| json!({"alpha": alpha});
+        json!({"plans": [
+            {"method": "asap", "weights": "cpu", "kind": "cheapest", "chosen": [
+                pick("sum", 3, "exact-sum", json!({}), 1),
+                pick("p90", 0, "dd", dd(0.02), 300),
+                p50,
+            ]},
+            {"method": "asap", "weights": "cpu", "kind": "sla", "sla_ms": 100.0, "chosen": [
+                pick("sum", 3, "exact-sum", json!({}), 1),
+                pick("p90", 0, "dd", dd(0.01), 300),
+                pick("p50", 0, "dd", dd(0.01), 300),
+            ]},
+        ]})
+    }
+
+    fn run(plan: Value, sla_ms: Option<f64>) -> anyhow::Result<PlannerOutput> {
+        run_on(TABLE, plan.to_string().as_bytes(), sla_ms)
+    }
+
+    fn run_on(table: &str, plan: &[u8], sla_ms: Option<f64>) -> anyhow::Result<PlannerOutput> {
+        let dir = tempfile::tempdir().unwrap();
+        let (table_path, plan_path) = (dir.path().join("table.json"), dir.path().join("plan.json"));
+        std::fs::write(&table_path, table).unwrap();
+        std::fs::write(&plan_path, plan).unwrap();
+        let selector = PlanSelector {
+            method: "asap",
+            weights: "cpu",
+            sla_ms,
+        };
+        saved_plan_output(&table_path, &plan_path, &selector, 1000)
+    }
+
+    fn refusal(plan: Value, sla_ms: Option<f64>) -> String {
+        run(plan, sla_ms).err().expect("refused").to_string()
+    }
+
+    fn shared_p50() -> Value {
+        pick("p50", 0, "dd", json!({"alpha": 0.02}), 300)
+    }
+
+    #[test]
+    fn writes_one_aggregation_per_plan_deployment() {
+        let output = run(plans(shared_p50()), None).unwrap();
+        assert_eq!(output.streaming_aggregation_count(), 2);
+        // The ratio's two leaves and the sum; the standalone p50 is a leaf already.
+        assert_eq!(output.inference_query_count(), 3);
+        let yaml = output.to_streaming_yaml_string().unwrap();
+        assert!(yaml.contains("windowSizeMs: 300000"), "{yaml}");
+        assert!(yaml.contains("slideIntervalMs: 60000"), "{yaml}");
+        assert!(yaml.contains("windowType: sliding"), "{yaml}");
+        assert!(yaml.contains("alpha: 0.02"), "{yaml}");
+    }
+
+    #[test]
+    fn sla_ms_selects_the_plan_at_that_sla() {
+        let yaml = run(plans(shared_p50()), Some(100.0))
+            .unwrap()
+            .to_streaming_yaml_string()
+            .unwrap();
+        assert!(yaml.contains("alpha: 0.01"), "{yaml}");
+        let err = refusal(plans(shared_p50()), Some(300.0));
+        assert!(err.contains("0 plans match"), "{err}");
+    }
+
+    #[test]
+    fn refuses_a_roll_up() {
+        let mut plan = plans(shared_p50());
+        plan["plans"][0]["chosen"][0]["deployment_grouping"] = json!(["job", "label_0"]);
+        let err = refusal(plan, None);
+        assert!(err.contains("roll-up"), "{err}");
+    }
+
+    #[test]
+    fn refuses_a_pick_from_another_table() {
+        // A 7s window doesn't divide the 5m lookback.
+        let p50 = pick("p50", 0, "dd", json!({"alpha": 0.02}), 7);
+        let err = refusal(plans(p50), None);
+        assert!(err.contains("does --plan match --workload-table"), "{err}");
+    }
+
+    #[test]
+    fn refuses_an_avg_before_splitting_it() {
+        let table = TABLE.replace(
+            r#""sum by (job) (data)", "interval_secs": 1"#,
+            r#""avg by (job) (data)", "interval_secs": 1"#,
+        );
+        let err = run_on(&table, plans(shared_p50()).to_string().as_bytes(), None)
+            .err()
+            .expect("refused")
+            .to_string();
+        assert!(err.contains("avg"), "{err}");
+        assert!(!err.contains("leaves"), "{err}");
+    }
+
+    #[test]
+    fn reads_a_gzipped_plan() {
+        use std::io::Write;
+        let mut gz = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
+        gz.write_all(plans(shared_p50()).to_string().as_bytes())
+            .unwrap();
+        let output = run_on(TABLE, &gz.finish().unwrap(), None).unwrap();
+        assert_eq!(output.streaming_aggregation_count(), 2);
+    }
+
+    #[test]
+    fn refuses_a_query_naming_too_few_rqes() {
+        let table = TABLE.replace(r#""rqes": ["p90", "p50"]"#, r#""rqes": ["p90"]"#);
+        let err = run_on(&table, plans(shared_p50()).to_string().as_bytes(), None)
+            .err()
+            .expect("refused")
+            .to_string();
+        assert!(err.contains("has 2 leaves but names 1 RQEs"), "{err}");
+    }
+
+    #[test]
+    fn refuses_a_query_on_two_deployments() {
+        // The ratio's legs share one item, so they must share a deployment.
+        let p50 = pick("p50", 1, "dd", json!({"alpha": 0.02}), 300);
+        let err = refusal(plans(p50), None);
+        assert!(err.contains("served by two deployments"), "{err}");
+    }
+
+    #[test]
+    fn refuses_a_pick_needing_a_key_tracker() {
+        let p50 = pick("p50", 0, "hydra-kll", json!({}), 300);
+        let err = refusal(plans(p50), None);
+        assert!(err.contains("does --plan match --workload-table"), "{err}");
+    }
+
+    #[test]
+    fn refuses_an_rqe_picked_twice() {
+        let mut plan = plans(shared_p50());
+        plan["plans"][0]["chosen"]
+            .as_array_mut()
+            .unwrap()
+            .push(shared_p50());
+        let err = refusal(plan, None);
+        assert!(err.contains(r#"picks RQE "p50" twice"#), "{err}");
+    }
+
+    #[test]
+    fn refuses_one_deployment_with_two_sketches() {
+        let p50 = pick("p50", 0, "dd", json!({"alpha": 0.05}), 300);
+        let err = refusal(plans(p50), None);
+        assert!(err.contains("different sketches"), "{err}");
+    }
+
+    #[test]
+    fn refuses_a_plan_missing_an_rqe() {
+        let mut plan = plans(shared_p50());
+        plan["plans"][0]["chosen"].as_array_mut().unwrap().pop();
+        let err = refusal(plan, None);
+        assert!(err.contains(r#"doesn't serve RQE "p50""#), "{err}");
+    }
+}
diff --git a/asap-tools/experiments/config/config.yaml b/asap-tools/experiments/config/config.yaml
index dfd4abd9..8539b4ff 100644
--- a/asap-tools/experiments/config/config.yaml
+++ b/asap-tools/experiments/config/config.yaml
@@ -90,6 +90,9 @@ query_engine:
 controller:
   punting: ???  # REQUIRED: Enable query punting based on performance heuristics (should_be_performant check)
   auto_discover_labels: false  # Query Prometheus for metric labels instead of using config hints
+  # Skip the planner and deploy the streaming_config.yaml and inference_config.yaml
+  # already in this directory, e.g. from `asap-planner --planner saved`.
+  prebuilt_output_dir: null
 
 # Optional manual planner windowing override. Omit this block to preserve the
 # default tumbling-window behavior. For tumbling windows, omit
diff --git a/asap-tools/experiments/experiment_run_e2e.py b/asap-tools/experiments/experiment_run_e2e.py
index 9b288ad8..13273a12 100644
--- a/asap-tools/experiments/experiment_run_e2e.py
+++ b/asap-tools/experiments/experiment_run_e2e.py
@@ -316,20 +316,42 @@ def main(cfg: DictConfig):
                     "Prometheus label discovery is disabled."
                 )
 
-            controller_service.start(
-                controller_input_file=controller_input_config,
-                streaming_engine=args.streaming_engine,
-                controller_remote_output_dir=CONTROLLER_REMOTE_OUTPUT_DIR,
-                punting=args.controller_punting,
-                discovery_backend=discovery_backend,
-                data_ingestion_interval_ms=data_ingestion_interval_ms,
-            )
-            sync.rsync_controller_config_remote_to_local(
-                provider,
-                CONTROLLER_REMOTE_OUTPUT_DIR,
-                CONTROLLER_LOCAL_OUTPUT_DIR,
-                node_offset=args.node_offset,
-            )
+            if args.controller_prebuilt_output_dir:
+                print(
+                    f"Skipping the planner: using configs from "
+                    f"{args.controller_prebuilt_output_dir}"
+                )
+                config.copy_prebuilt_controller_output(
+                    args.controller_prebuilt_output_dir,
+                    CONTROLLER_LOCAL_OUTPUT_DIR,
+                    os.path.join(
+                        local_experiment_root_dir,
+                        "controller_client_configs",
+                        f"{experiment_mode}_controller_input.yaml",
+                    ),
+                    data_ingestion_interval_ms,
+                )
+                sync.rsync_streaming_configs(
+                    provider,
+                    CONTROLLER_LOCAL_OUTPUT_DIR,
+                    CONTROLLER_REMOTE_OUTPUT_DIR,
+                    node_offset=args.node_offset,
+                )
+            else:
+                controller_service.start(
+                    controller_input_file=controller_input_config,
+                    streaming_engine=args.streaming_engine,
+                    controller_remote_output_dir=CONTROLLER_REMOTE_OUTPUT_DIR,
+                    punting=args.controller_punting,
+                    discovery_backend=discovery_backend,
+                    data_ingestion_interval_ms=data_ingestion_interval_ms,
+                )
+                sync.rsync_controller_config_remote_to_local(
+                    provider,
+                    CONTROLLER_REMOTE_OUTPUT_DIR,
+                    CONTROLLER_LOCAL_OUTPUT_DIR,
+                    node_offset=args.node_offset,
+                )
 
         if (
             config.check_exporter_and_queries_exist(
diff --git a/asap-tools/experiments/experiment_utils/config.py b/asap-tools/experiments/experiment_utils/config.py
index a731c005..5f8ce768 100644
--- a/asap-tools/experiments/experiment_utils/config.py
+++ b/asap-tools/experiments/experiment_utils/config.py
@@ -5,10 +5,13 @@
 
 import math
 import os
+import re
+import shutil
 import copy
 import yaml
 from typing import Any, Dict, List, Tuple
 
+import hydra.utils
 from omegaconf import DictConfig, ListConfig, OmegaConf
 
 import constants
@@ -535,6 +538,65 @@ def get_prometheus_data_ingestion_interval_ms(prometheus_config):
         raise ValueError(f"Invalid scrape interval string: {s}")
 
 
+PLANNER_OUTPUT_FILES = ("streaming_config.yaml", "inference_config.yaml")
+
+
+def check_prebuilt_controller_output(src):
+    for name in PLANNER_OUTPUT_FILES:
+        path = os.path.join(src, name)
+        if not os.path.isfile(path):
+            raise ValueError(f"controller.prebuilt_output_dir has no {name}: {path}")
+
+
+def _is_planned(query, planned):
+    """Whether `query` is planned whole, or is arithmetic over planned leaves
+    (the planner splits binary queries into leaves)."""
+    # ponytail: textual check, not a PromQL parse; a leaf that is a substring
+    # of another is removed longest first.
+    for leaf in sorted(planned, key=len, reverse=True):
+        query = query.replace(leaf, "")
+    return re.fullmatch(r"[\s()+\-*/0-9.]*", query) is not None
+
+
+def copy_prebuilt_controller_output(
+    src, dest, controller_input_config, scrape_interval_ms
+):
+    """Replaces `dest` with the planner configs in `src`, after checking they
+    plan every query of `controller_input_config` at this scrape interval."""
+    check_prebuilt_controller_output(src)
+    src_real, dest_real = os.path.realpath(src), os.path.realpath(dest)
+    if os.path.commonpath([src_real, dest_real]) in (src_real, dest_real):
+        raise ValueError(
+            f"controller.prebuilt_output_dir {src} and the experiment's output "
+            f"dir {dest} must not contain one another"
+        )
+    with open(os.path.join(src, "inference_config.yaml")) as f:
+        planned = {q["query"] for q in yaml.safe_load(f)["queries"]}
+    with open(controller_input_config) as f:
+        groups = yaml.safe_load(f)["query_groups"]
+    unplanned = [q for g in groups for q in g["queries"] if not _is_planned(q, planned)]
+    if unplanned:
+        raise ValueError(
+            f"controller.prebuilt_output_dir doesn't plan these queries: {unplanned}"
+        )
+    with open(os.path.join(src, "streaming_config.yaml")) as f:
+        aggregations = yaml.safe_load(f)["aggregations"]
+    bad = [
+        a["aggregationId"]
+        for a in aggregations
+        if a["slideIntervalMs"] % scrape_interval_ms
+    ]
+    if bad:
+        raise ValueError(
+            f"controller.prebuilt_output_dir aggregations {bad} slide off the "
+            f"{scrape_interval_ms} ms scrape interval: planned for another one?"
+        )
+    shutil.rmtree(dest, ignore_errors=True)
+    os.makedirs(dest)
+    for name in PLANNER_OUTPUT_FILES:
+        shutil.copy2(os.path.join(src, name), dest)
+
+
 class Args:
     """Helper class to convert Hydra config to argparse-like namespace for backward compatibility."""
 
@@ -605,6 +667,12 @@ def __init__(self, cfg: DictConfig):
         # Controller configuration
         self.controller_punting = cfg.controller.punting
         self.controller_auto_discover_labels = cfg.controller.auto_discover_labels
+        self.controller_prebuilt_output_dir = cfg.controller.get("prebuilt_output_dir")
+        if self.controller_prebuilt_output_dir:
+            self.controller_prebuilt_output_dir = hydra.utils.to_absolute_path(
+                self.controller_prebuilt_output_dir
+            )
+            check_prebuilt_controller_output(self.controller_prebuilt_output_dir)
 
         # Aggregate cleanup configuration
         # Valid policies: "circular_buffer", "read_based", "no_cleanup"
diff --git a/asap-tools/experiments/tests/test_prebuilt_controller_output.py b/asap-tools/experiments/tests/test_prebuilt_controller_output.py
new file mode 100644
index 00000000..b140933c
--- /dev/null
+++ b/asap-tools/experiments/tests/test_prebuilt_controller_output.py
@@ -0,0 +1,87 @@
+"""Tests for running e2e experiments on prebuilt planner configs."""
+
+import os
+import tempfile
+import unittest
+
+import yaml
+
+from experiment_utils.config import copy_prebuilt_controller_output
+
+QUERY = "sum by (job) (data)"
+
+
+class CopyPrebuiltControllerOutputTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.src = os.path.join(self.tmp.name, "src")
+        self.dest = os.path.join(self.tmp.name, "dest")
+        self.client = os.path.join(self.tmp.name, "client.yaml")
+        os.makedirs(self.src)
+        self.write(
+            os.path.join(self.src, "inference_config.yaml"),
+            {"queries": [{"query": QUERY}]},
+        )
+        self.write(
+            os.path.join(self.src, "streaming_config.yaml"),
+            {"aggregations": [{"aggregationId": 1, "slideIntervalMs": 1000}]},
+        )
+        self.write(self.client, {"query_groups": [{"queries": [QUERY]}]})
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    @staticmethod
+    def write(path, content):
+        with open(path, "w") as f:
+            yaml.safe_dump(content, f)
+
+    def copy(self, src=None, dest=None, scrape_interval_ms=1000):
+        copy_prebuilt_controller_output(
+            src or self.src, dest or self.dest, self.client, scrape_interval_ms
+        )
+
+    def test_replaces_stale_output(self):
+        os.makedirs(self.dest)
+        with open(os.path.join(self.dest, "controller.log"), "w") as f:
+            f.write("stale")
+        self.copy()
+        self.assertEqual(
+            sorted(os.listdir(self.dest)),
+            ["inference_config.yaml", "streaming_config.yaml"],
+        )
+
+    def test_refuses_missing_config(self):
+        os.remove(os.path.join(self.src, "inference_config.yaml"))
+        with self.assertRaisesRegex(ValueError, "inference_config.yaml"):
+            self.copy()
+
+    def test_refuses_nested_dirs(self):
+        for src, dest in [
+            (self.src, self.src),
+            (self.src, os.path.dirname(self.src)),
+            (self.src, os.path.join(self.src, "out")),
+        ]:
+            with self.assertRaisesRegex(ValueError, "contain one another"):
+                self.copy(src, dest)
+        self.assertTrue(os.path.isfile(os.path.join(self.src, "streaming_config.yaml")))
+
+    def test_accepts_arithmetic_over_planned_leaves(self):
+        self.write(
+            self.client,
+            {"query_groups": [{"queries": [f"({QUERY}) / {QUERY} * 100"]}]},
+        )
+        self.copy()
+
+    def test_refuses_an_unplanned_query(self):
+        self.write(self.client, {"query_groups": [{"queries": [QUERY, "other"]}]})
+        with self.assertRaisesRegex(ValueError, "other"):
+            self.copy()
+
+    def test_refuses_another_scrape_interval(self):
+        with self.assertRaisesRegex(ValueError, "15000 ms scrape interval"):
+            self.copy(scrape_interval_ms=15000)
+
+
+if __name__ == "__main__":
+    unittest.main()