diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs
index c3d8228a..d2bd0e75 100644
--- a/control_plane/src/backend_client.rs
+++ b/control_plane/src/backend_client.rs
@@ -324,6 +324,41 @@ impl BackendClient {
             ))
         }
     }
+
+    /// POST an encoded `BackendPlan` (protobuf bytes) to the backend's
+    /// `POST /api/v1/backend-plan` endpoint (see
+    /// `control_plane/docs/design-backend-plan-wire-format.md`), sent
+    /// alongside the streaming-config/storage-routing push, not in place
+    /// of it (see `emit::backend_push`'s call site). Same
+    /// transient/permanent classification as the other typed POST
+    /// methods.
+    pub async fn post_backend_plan_typed(
+        &self,
+        bytes: Vec<u8>,
+    ) -> std::result::Result<(), BackendPostError> {
+        let url = derive_backend_plan_url(&self.endpoint);
+        debug!(
+            endpoint = %url,
+            plan_bytes = bytes.len(),
+            "posting BackendPlan to ASAPQuery-backend (typed)"
+        );
+        let resp = self
+            .http
+            .post(&url)
+            .header("content-type", "application/x-protobuf")
+            .body(bytes)
+            .send()
+            .await
+            .map_err(classify_reqwest_error)?;
+
+        let status = resp.status();
+        if status.is_success() {
+            Ok(())
+        } else {
+            let body = resp.text().await.unwrap_or_default();
+            Err(classify_http_status(status, body, "BackendPlan POST"))
+        }
+    }
 }
 
 /// Map a streaming-config endpoint URL to the sibling storage-routing
@@ -344,6 +379,21 @@ fn derive_storage_routing_url(endpoint: &str) -> String {
     endpoint.to_string()
 }
 
+/// Map a streaming-config endpoint URL to the sibling `backend-plan`
+/// endpoint, same rewrite convention as [`derive_storage_routing_url`].
+fn derive_backend_plan_url(endpoint: &str) -> String {
+    const STREAMING_PATH_DASH: &str = "/api/v1/streaming-config";
+    const STREAMING_PATH_UNDERSCORE: &str = "/api/v1/streaming_config";
+    const PLAN_PATH: &str = "/api/v1/backend-plan";
+    if let Some(stripped) = endpoint.strip_suffix(STREAMING_PATH_DASH) {
+        return format!("{stripped}{PLAN_PATH}");
+    }
+    if let Some(stripped) = endpoint.strip_suffix(STREAMING_PATH_UNDERSCORE) {
+        return format!("{stripped}{PLAN_PATH}");
+    }
+    endpoint.to_string()
+}
+
 /// Fire-and-forget convenience helper used by the replanner. Logs
 /// errors at WARN and never propagates them — the replanner should
 /// never fail an entire replan because the backend was temporarily
@@ -497,6 +547,73 @@ mod tests {
         assert_eq!(derive_storage_routing_url("http://x/foo"), "http://x/foo");
     }
 
+    #[test]
+    fn backend_plan_url_rewrites_streaming_path() {
+        assert_eq!(
+            derive_backend_plan_url("http://backend:8088/api/v1/streaming-config"),
+            "http://backend:8088/api/v1/backend-plan"
+        );
+        assert_eq!(
+            derive_backend_plan_url("http://backend:8088/api/v1/streaming_config"),
+            "http://backend:8088/api/v1/backend-plan"
+        );
+    }
+
+    #[test]
+    fn backend_plan_url_preserves_unknown_paths_for_tests() {
+        assert_eq!(
+            derive_backend_plan_url("http://127.0.0.1:1/api/v1/backend-plan"),
+            "http://127.0.0.1:1/api/v1/backend-plan"
+        );
+        assert_eq!(derive_backend_plan_url("http://x/foo"), "http://x/foo");
+    }
+
+    #[tokio::test]
+    async fn backend_plan_post_round_trips_bytes_via_url_rewrite() {
+        let hits: StdArc<Mutex<Vec<Vec<u8>>>> = StdArc::new(Mutex::new(Vec::new()));
+        let hits_for_route = hits.clone();
+        let app = Router::new()
+            .route(
+                "/api/v1/backend-plan",
+                post(move |body: axum::body::Bytes| {
+                    let hits = hits_for_route.clone();
+                    async move {
+                        hits.lock().unwrap().push(body.to_vec());
+                        axum::http::StatusCode::OK
+                    }
+                }),
+            );
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+        tokio::time::sleep(Duration::from_millis(50)).await;
+
+        let client = BackendClient::new(format!("http://{addr}/api/v1/streaming-config"));
+        let bytes = vec![1u8, 2, 3, 4];
+        client
+            .post_backend_plan_typed(bytes.clone())
+            .await
+            .expect("backend-plan post ok");
+
+        let received = hits.lock().unwrap();
+        assert_eq!(received.len(), 1);
+        assert_eq!(received[0], bytes);
+    }
+
+    #[tokio::test]
+    async fn backend_plan_post_404_is_transient() {
+        let sink = SharedSink(StdArc::new(Mutex::new(Vec::new())));
+        let url = start_mock_backend(sink.clone(), axum::http::StatusCode::NOT_FOUND).await;
+        let client = BackendClient::new(url);
+        let err = client
+            .post_backend_plan_typed(vec![1, 2, 3])
+            .await
+            .expect_err("404 should surface as Err");
+        assert!(err.is_transient(), "404 must classify as transient: {err}");
+    }
+
     /// Phase α: full happy path. A mock backend hosts the storage
     /// routing endpoint; the client POSTs the control-plane-emitted JSON
     /// and the body round-trips verbatim. Mirrors `json_post_round_trips_body`.
diff --git a/control_plane/src/backend_plan/from_stage_config.rs b/control_plane/src/backend_plan/from_stage_config.rs
new file mode 100644
index 00000000..d632f417
--- /dev/null
+++ b/control_plane/src/backend_plan/from_stage_config.rs
@@ -0,0 +1,472 @@
+//! Build a [`BackendPlan`] from the L5 emitter's [`BackendStageConfig`] —
+//! the same input `crate::emit::stage_config::emit_backend_streaming_config_json`
+//! consumes to produce the legacy JSON wire format. This is the
+//! `BackendPlan`-side sibling of that function; see this crate's design
+//! doc (`control_plane/docs/design-backend-plan-wire-format.md`) for why
+//! both exist side by side (dual-push, see `backend_client.rs`).
+//!
+//! **Fingerprint parity is the load-bearing invariant here.** A
+//! `Materialization`'s `fingerprint` must equal what `data_plane`
+//! independently computes for the equivalent policy today (via
+//! `AggregationConfig::from_yaml_data` + `PolicyFingerprint::from_config`
+//! on the legacy JSON/YAML `StreamingConfig` push) — `SketchStore` sid
+//! registration and any future cross-reference between the two wire
+//! formats depend on the two identity spaces staying unified. Rather
+//! than re-deriving the field mapping a second time (real drift risk),
+//! this module reuses `build_backend_aggregation_json` (the exact same
+//! JSON `AggregationConfig::from_yaml_data` parses on the receiving end)
+//! and computes the fingerprint from *that*.
+
+use std::collections::HashMap;
+
+use anyhow::{Context, Result};
+use asap_sketch::SummaryKind;
+use asap_types::{AggregationConfig, MonitorSpec, PolicyFingerprint, QueryLanguage};
+
+use crate::emit::monitor::{agg_id_for_metric, MonitorIntent};
+use crate::emit::stage_config::build_backend_aggregation_json;
+use crate::intent_algebra::{ColumnRef, Source, WindowKind};
+use crate::physical::colored_dag::emitter::{BackendAggregation, BackendStageConfig};
+use crate::sketch_algebra::capability::{Capability, SketchKindHandle};
+
+use super::{BackendPlan, Materialization, RoutingEntry, StorageBackend, WindowSpec};
+
+/// Build a `BackendPlan` from a planning cycle's `BackendStageConfig` +
+/// declared CDM monitors. `plan_id`/`generated_at_unix_ms` are
+/// observability-only (see `BackendPlan`'s own doc) — callers typically
+/// reuse whatever counter/clock they already thread through the legacy
+/// `StreamingConfig` emit path.
+pub fn from_stage_config(
+    cfg: &BackendStageConfig,
+    monitors: &[MonitorIntent],
+    plan_id: u64,
+    generated_at_unix_ms: u64,
+) -> Result<BackendPlan> {
+    let mut materializations = HashMap::with_capacity(cfg.aggregations.len());
+    let mut fingerprint_by_agg_id: HashMap<&str, PolicyFingerprint> =
+        HashMap::with_capacity(cfg.aggregations.len());
+
+    for agg in &cfg.aggregations {
+        let fingerprint = policy_fingerprint_for_aggregation(agg)
+            .with_context(|| format!("aggregation_id {:?}", agg.aggregation_id))?;
+        fingerprint_by_agg_id.insert(agg.aggregation_id.as_str(), fingerprint);
+
+        let (kind, params) = match &agg.agg_type_override {
+            Some(exact_type) => exact_kind_params_for_override(exact_type)?,
+            None => (agg.sketch_kind.clone(), agg.sketch_params.clone()),
+        };
+
+        materializations.insert(
+            fingerprint,
+            Materialization {
+                fingerprint,
+                source: Source::TimeSeries {
+                    metric: agg.metric_name.clone(),
+                },
+                window: WindowSpec {
+                    kind: WindowKind::Tumbling,
+                    size_ms: agg.window_secs.saturating_mul(1000),
+                    slide_ms: None,
+                },
+                group_by: agg.grouping.clone(),
+                rollup: Vec::new(),
+                kind,
+                params,
+                col: ColumnRef::SampleValue,
+                retention: None,
+            },
+        );
+    }
+
+    let mut routing = Vec::with_capacity(cfg.readouts.len());
+    for readout in &cfg.readouts {
+        let Some(&fingerprint) = fingerprint_by_agg_id.get(readout.aggregation_id.as_str()) else {
+            // Orphan readout (no matching aggregation in this cycle's
+            // config) — nothing to route. Same "tolerate, don't error"
+            // stance the legacy JSON emitter takes toward its own
+            // readouts list.
+            continue;
+        };
+        let Some(agg) = cfg
+            .aggregations
+            .iter()
+            .find(|a| a.aggregation_id == readout.aggregation_id)
+        else {
+            continue;
+        };
+        let satisfies = capability_for_readout(agg, &readout.op)?;
+        routing.push(RoutingEntry {
+            satisfies,
+            materialization: fingerprint,
+            storage_backend: StorageBackend::SketchStore,
+        });
+    }
+
+    let plan_monitors = monitors
+        .iter()
+        .map(|m| MonitorSpec {
+            agg_id: agg_id_for_metric(&m.metric),
+            functional: m.functional.as_str().to_string(),
+            key: m.key.clone(),
+            tau: m.tau,
+            epsilon: m.epsilon,
+            window_ms: m.window_ms,
+            d: 0,
+            w: 0,
+            mode: String::new(),
+        })
+        .collect();
+
+    Ok(BackendPlan {
+        plan_id,
+        generated_at_unix_ms,
+        materializations,
+        routing,
+        monitors: plan_monitors,
+    })
+}
+
+/// Derive the fingerprint data_plane will independently compute for this
+/// aggregation. Round-trips through `build_backend_aggregation_json` +
+/// `AggregationConfig::from_yaml_data` — the exact same JSON shape and
+/// parser the real `POST /api/v1/streaming-config` handler uses (which
+/// parses its body as YAML regardless of declared content-type, since
+/// JSON is valid YAML) — rather than re-deriving the field mapping here.
+fn policy_fingerprint_for_aggregation(agg: &BackendAggregation) -> Result<PolicyFingerprint> {
+    let json = build_backend_aggregation_json(agg);
+    let text = serde_json::to_string(&json).context("serialize synthesized aggregation JSON")?;
+    let yaml_value: serde_yaml::Value =
+        serde_yaml::from_str(&text).context("parse synthesized aggregation JSON as YAML")?;
+    let cfg = AggregationConfig::from_yaml_data(&yaml_value, None, QueryLanguage::promql)
+        .context("build AggregationConfig from synthesized aggregation JSON")?;
+    Ok(cfg.policy_fingerprint())
+}
+
+/// Option B (post-#287) exact-agg override: `s` is already the wire
+/// `aggregationType` string (e.g. `"Sum"`) — parse it via
+/// `AggregationType::FromStr` (same parser
+/// `AggregationConfig::from_yaml_data` uses) and carry it as the
+/// matching `SummaryKind`/`SummaryParams` exact-agg pair.
+fn exact_kind_params_for_override(exact_type: &str) -> Result<(SummaryKind, asap_sketch::SummaryParams)> {
+    use asap_sketch::SummaryParams;
+    match exact_type {
+        "Sum" => Ok((SummaryKind::Sum, SummaryParams::Sum)),
+        "Count" => Ok((SummaryKind::Count, SummaryParams::Count)),
+        "MinMax" => Ok((SummaryKind::MinMax, SummaryParams::MinMax)),
+        "Increase" => Ok((SummaryKind::Increase, SummaryParams::Increase)),
+        "Rate" => Ok((SummaryKind::Rate, SummaryParams::Rate)),
+        other => anyhow::bail!("unrecognized agg_type_override {other:?} — no SummaryKind mapping"),
+    }
+}
+
+/// Capability this readout satisfies, given the aggregation it reads
+/// from. Exact-agg overrides always report `Capability::ExactAgg`
+/// (mirrors the wire's `aggregationType` bypass — see
+/// `BackendAggregation::agg_type_override`'s doc); otherwise derive from
+/// the sketch family + readout op, matching
+/// `sketch_algebra::capability::Capability`'s variant-per-query-shape
+/// design.
+fn capability_for_readout(agg: &BackendAggregation, op: &asap_sketch::SketchQuery) -> Result<Capability> {
+    use asap_sketch::SketchQuery;
+    use asap_types::AggregationType;
+
+    if let Some(exact_type) = &agg.agg_type_override {
+        let agg_type: AggregationType = exact_type
+            .parse()
+            .map_err(|e: String| anyhow::anyhow!("agg_type_override {exact_type:?}: {e}"))?;
+        return Ok(Capability::ExactAgg(agg_type));
+    }
+
+    let handle = sketch_kind_handle(&agg.sketch_kind)?;
+    Ok(match op {
+        SketchQuery::Quantile { .. } => Capability::QuantileApprox(handle),
+        SketchQuery::Cardinality => Capability::CardinalityApprox,
+        SketchQuery::PointCount { .. } => Capability::FrequencyEstimate(handle),
+        SketchQuery::TopK { .. } => Capability::FrequencyTopk(handle),
+    })
+}
+
+/// Map a `SummaryKind` to the `SketchKindHandle` it identifies as. Only
+/// covers the families real `Bind*` rules actually produce for sketch
+/// aggregations — mirrors `emit::stage_config::sketch_kind_to_backend_type`'s
+/// exhaustive match (and its `unreachable!()` for non-sketch kinds).
+fn sketch_kind_handle(kind: &SummaryKind) -> Result<SketchKindHandle> {
+    Ok(match kind {
+        SummaryKind::DDSketch => SketchKindHandle::DDSketch,
+        SummaryKind::Kll => SketchKindHandle::Kll,
+        SummaryKind::Hll => SketchKindHandle::Hll,
+        SummaryKind::CountSketch => SketchKindHandle::CountSketch,
+        SummaryKind::Cms => SketchKindHandle::CountMin,
+        SummaryKind::CmsWithHeap => SketchKindHandle::CmsWithHeap,
+        SummaryKind::CountSketchWithHeap => SketchKindHandle::CountSketchWithHeap,
+        other => anyhow::bail!("no SketchKindHandle mapping for non-sketch SummaryKind {other:?}"),
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::physical::colored_dag::emitter::{AggregationInput, BackendReadout};
+    use asap_sketch::{SketchQuery, SummaryParams};
+    use asap_types::{AggregationType, KeyByLabelNames, WindowKind as AsapWindowKind};
+    use std::collections::HashMap as StdHashMap;
+
+    fn agg(
+        aggregation_id: &str,
+        metric_name: &str,
+        sketch_kind: SummaryKind,
+        sketch_params: SummaryParams,
+        grouping: Vec<String>,
+    ) -> BackendAggregation {
+        BackendAggregation {
+            aggregation_id: aggregation_id.to_string(),
+            metric_name: metric_name.to_string(),
+            sketch_kind,
+            sketch_params,
+            window_secs: 60,
+            spatial_filter: String::new(),
+            grouping,
+            item_label: None,
+            aggregation_input: AggregationInput::SketchEnvelope,
+            agg_type_override: None,
+        }
+    }
+
+    /// Independently construct the `AggregationConfig` a hand-written
+    /// (non-JSON-round-trip) reader would build for this fixture, so the
+    /// parity test doesn't just check the implementation against itself.
+    fn hand_built_config(agg: &BackendAggregation) -> AggregationConfig {
+        let parameters: StdHashMap<String, serde_json::Value> = match &agg.sketch_params {
+            SummaryParams::DDSketch { alpha } => {
+                StdHashMap::from([("alpha".to_string(), serde_json::json!(alpha))])
+            }
+            SummaryParams::Hll { precision } => {
+                StdHashMap::from([("precision".to_string(), serde_json::json!(precision))])
+            }
+            SummaryParams::CountSketchWithHeap { width, depth, .. } => StdHashMap::from([
+                ("w".to_string(), serde_json::json!(width)),
+                ("d".to_string(), serde_json::json!(depth)),
+                ("with_heap".to_string(), serde_json::json!(true)),
+            ]),
+            SummaryParams::Cms { width, depth } => StdHashMap::from([
+                ("w".to_string(), serde_json::json!(width)),
+                ("d".to_string(), serde_json::json!(depth)),
+            ]),
+            other => unreachable!("fixture doesn't exercise {other:?}"),
+        };
+        AggregationConfig::new(
+            match agg.sketch_kind {
+                SummaryKind::DDSketch => AggregationType::DDSketch,
+                SummaryKind::Kll => AggregationType::DatasketchesKLL,
+                SummaryKind::Hll => AggregationType::HLL,
+                SummaryKind::Cms => AggregationType::CountMinSketch,
+                SummaryKind::CmsWithHeap => AggregationType::CountMinSketchWithHeap,
+                SummaryKind::CountSketch => AggregationType::CountSketch,
+                SummaryKind::CountSketchWithHeap => AggregationType::CountSketchWithHeap,
+                _ => unreachable!("fixture only uses sketch-typed kinds"),
+            },
+            String::new(),
+            parameters,
+            KeyByLabelNames::new(agg.grouping.clone()),
+            KeyByLabelNames::empty(),
+            KeyByLabelNames::empty(),
+            String::new(),
+            agg.window_secs,
+            agg.window_secs,
+            AsapWindowKind::Tumbling,
+            agg.spatial_filter.clone(),
+            agg.metric_name.clone(),
+            None,
+            None,
+            None,
+        )
+    }
+
+    fn sample_cfg() -> BackendStageConfig {
+        BackendStageConfig {
+            aggregations: vec![
+                agg(
+                    "agg0",
+                    "http_latency_ms",
+                    SummaryKind::DDSketch,
+                    SummaryParams::DDSketch { alpha: 0.01 },
+                    Vec::new(),
+                ),
+                agg(
+                    "agg1",
+                    "http_requests_total",
+                    SummaryKind::Hll,
+                    SummaryParams::Hll { precision: 14 },
+                    vec!["zone".to_string()],
+                ),
+            ],
+            readouts: vec![
+                BackendReadout {
+                    aggregation_id: "agg0".into(),
+                    op: SketchQuery::Quantile { q: 0.99 },
+                },
+                BackendReadout {
+                    aggregation_id: "agg1".into(),
+                    op: SketchQuery::Cardinality,
+                },
+            ],
+        }
+    }
+
+    #[test]
+    fn every_aggregation_produces_exactly_one_materialization() {
+        let cfg = sample_cfg();
+        let plan = from_stage_config(&cfg, &[], 1, 0).expect("build plan");
+        assert_eq!(plan.materializations.len(), cfg.aggregations.len());
+    }
+
+    #[test]
+    fn fingerprint_matches_hand_built_aggregation_config() {
+        let cfg = sample_cfg();
+        let plan = from_stage_config(&cfg, &[], 1, 0).expect("build plan");
+
+        for agg in &cfg.aggregations {
+            let expected = hand_built_config(agg).policy_fingerprint();
+            let m = plan
+                .materializations
+                .get(&expected)
+                .unwrap_or_else(|| panic!("no materialization for expected fingerprint of {agg:?}"));
+            assert_eq!(m.fingerprint, expected);
+        }
+    }
+
+    #[test]
+    fn sketch_kind_and_params_pass_through_unchanged() {
+        let cfg = sample_cfg();
+        let plan = from_stage_config(&cfg, &[], 1, 0).expect("build plan");
+        let ddsketch = plan
+            .materializations
+            .values()
+            .find(|m| matches!(m.source, Source::TimeSeries { ref metric } if metric == "http_latency_ms"))
+            .expect("ddsketch materialization present");
+        assert_eq!(ddsketch.kind, SummaryKind::DDSketch);
+        assert_eq!(ddsketch.params, SummaryParams::DDSketch { alpha: 0.01 });
+
+        let hll = plan
+            .materializations
+            .values()
+            .find(|m| matches!(m.source, Source::TimeSeries { ref metric } if metric == "http_requests_total"))
+            .expect("hll materialization present");
+        assert_eq!(hll.kind, SummaryKind::Hll);
+        assert_eq!(hll.params, SummaryParams::Hll { precision: 14 });
+        assert_eq!(hll.group_by, vec!["zone".to_string()]);
+    }
+
+    #[test]
+    fn readouts_map_to_routing_entries_pointing_at_the_right_fingerprint() {
+        let cfg = sample_cfg();
+        let plan = from_stage_config(&cfg, &[], 1, 0).expect("build plan");
+        assert_eq!(plan.routing.len(), 2);
+
+        let ddsketch_fp = plan
+            .materializations
+            .iter()
+            .find(|(_, m)| matches!(m.source, Source::TimeSeries { ref metric } if metric == "http_latency_ms"))
+            .map(|(fp, _)| *fp)
+            .expect("ddsketch fingerprint");
+        let quantile_entry = plan
+            .routing
+            .iter()
+            .find(|r| r.materialization == ddsketch_fp)
+            .expect("routing entry for ddsketch");
+        assert_eq!(quantile_entry.satisfies, Capability::QuantileApprox(SketchKindHandle::DDSketch));
+
+        let hll_fp = plan
+            .materializations
+            .iter()
+            .find(|(_, m)| matches!(m.source, Source::TimeSeries { ref metric } if metric == "http_requests_total"))
+            .map(|(fp, _)| *fp)
+            .expect("hll fingerprint");
+        let cardinality_entry = plan
+            .routing
+            .iter()
+            .find(|r| r.materialization == hll_fp)
+            .expect("routing entry for hll");
+        assert_eq!(cardinality_entry.satisfies, Capability::CardinalityApprox);
+    }
+
+    #[test]
+    fn topk_and_point_count_readouts_map_to_frequency_capabilities() {
+        let cfg = BackendStageConfig {
+            aggregations: vec![
+                agg(
+                    "agg0",
+                    "endpoint_count",
+                    SummaryKind::CountSketchWithHeap,
+                    SummaryParams::CountSketchWithHeap {
+                        width: 2048,
+                        depth: 5,
+                        heap_size: 10,
+                    },
+                    Vec::new(),
+                ),
+                agg(
+                    "agg1",
+                    "endpoint_hits",
+                    SummaryKind::Cms,
+                    SummaryParams::Cms {
+                        width: 4096,
+                        depth: 4,
+                    },
+                    Vec::new(),
+                ),
+            ],
+            readouts: vec![
+                BackendReadout {
+                    aggregation_id: "agg0".into(),
+                    op: SketchQuery::TopK { k: 10 },
+                },
+                BackendReadout {
+                    aggregation_id: "agg1".into(),
+                    op: SketchQuery::PointCount {
+                        key: ColumnRef::Named("user_42".into()),
+                        value: None,
+                    },
+                },
+            ],
+        };
+        let plan = from_stage_config(&cfg, &[], 1, 0).expect("build plan");
+        let topk_entry = plan
+            .routing
+            .iter()
+            .find(|r| r.satisfies == Capability::FrequencyTopk(SketchKindHandle::CountSketchWithHeap));
+        assert!(topk_entry.is_some(), "expected a FrequencyTopk routing entry: {:?}", plan.routing);
+
+        let freq_entry = plan
+            .routing
+            .iter()
+            .find(|r| r.satisfies == Capability::FrequencyEstimate(SketchKindHandle::CountMin));
+        assert!(freq_entry.is_some(), "expected a FrequencyEstimate routing entry: {:?}", plan.routing);
+    }
+
+    #[test]
+    fn exact_agg_override_reports_exact_agg_capability() {
+        let mut a = agg(
+            "agg0",
+            "http_requests_total",
+            SummaryKind::Sum, // sentinel value, suppressed by the override
+            SummaryParams::Sum,
+            vec!["zone".to_string()],
+        );
+        a.agg_type_override = Some("Sum".to_string());
+        let cfg = BackendStageConfig {
+            aggregations: vec![a],
+            readouts: vec![BackendReadout {
+                aggregation_id: "agg0".into(),
+                op: SketchQuery::Cardinality, // op is irrelevant for override rows
+            }],
+        };
+        let plan = from_stage_config(&cfg, &[], 1, 0).expect("build plan");
+        let (_, m) = plan.materializations.iter().next().expect("one materialization");
+        assert!(m.kind.is_exact());
+        assert_eq!(m.kind, SummaryKind::Sum);
+
+        let entry = &plan.routing[0];
+        assert_eq!(entry.satisfies, Capability::ExactAgg(AggregationType::Sum));
+    }
+}
diff --git a/control_plane/src/backend_plan/mod.rs b/control_plane/src/backend_plan/mod.rs
index 6283137b..ffcba029 100644
--- a/control_plane/src/backend_plan/mod.rs
+++ b/control_plane/src/backend_plan/mod.rs
@@ -24,6 +24,9 @@ pub mod proto {
     include!(concat!(env!("OUT_DIR"), "/control_plane.backend_plan.v1.rs"));
 }
 
+mod from_stage_config;
+pub use from_stage_config::from_stage_config;
+
 use std::collections::HashMap;
 
 use asap_sketch::{SummaryKind, SummaryParams};
diff --git a/control_plane/src/emit/backend_push.rs b/control_plane/src/emit/backend_push.rs
index 7fcba9e1..9e8f7a2a 100644
--- a/control_plane/src/emit/backend_push.rs
+++ b/control_plane/src/emit/backend_push.rs
@@ -44,11 +44,12 @@ use std::collections::{BTreeMap, HashMap};
 // import is gated to keep the non-test build warning-free.
 #[cfg(test)]
 use std::future::Future;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
-use std::time::{Duration, Instant};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use tokio::sync::Mutex;
-use tracing::{info, warn};
+use tracing::{debug, info, warn};
 
 use crate::backend_client::{BackendClient, BackendPostError};
 use crate::emit::{emit_backend_storage_routing, emit_backend_streaming_config_json};
@@ -73,6 +74,19 @@ const RETRY_MAX_ATTEMPTS: u32 = 5;
 const RETRY_BASE_DELAY: Duration = Duration::from_millis(100);
 const RETRY_DELAY_CAP: Duration = Duration::from_millis(2700);
 
+/// Monotonic counter for `BackendPlan.plan_id` — observability only, not
+/// identity (see `BackendPlan`'s own doc). One process-wide sequence is
+/// enough; there's no existing streaming-config version counter to
+/// reuse for parity.
+static PLAN_ID_COUNTER: AtomicU64 = AtomicU64::new(1);
+
+fn now_unix_ms() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0)
+}
+
 /// Cheap process-wide jitter source. We don't have `rand` in the
 /// control plane's dependency set and don't want to add it for one
 /// call site — `Instant::elapsed` reads the monotonic clock which is
@@ -280,6 +294,30 @@ async fn push_documents_coupled(
     (streaming_ok, routing_ok, RETRY_MAX_ATTEMPTS)
 }
 
+/// Best-effort, single-attempt push of the encoded `BackendPlan` — no
+/// in-function retry loop, unlike [`push_documents_coupled`]. A dropped
+/// push just leaves `data_plane`'s serving-time lookup falling back to
+/// `SketchStore` reconstruction until the next replan cycle re-pushes,
+/// so the next cycle is itself the retry backstop — same contract
+/// [`push_or_log`] already establishes for the legacy YAML path. Logs at
+/// WARN on failure; never affects [`PushOutcome`], which real callers
+/// key legacy-path behavior on.
+async fn push_backend_plan_best_effort(client: &Arc<BackendClient>, bytes: Vec<u8>) {
+    match client.post_backend_plan_typed(bytes).await {
+        Ok(()) => {
+            debug!(stage = "backend", endpoint = %client.endpoint(), "BackendPlan push succeeded");
+        }
+        Err(e) => {
+            warn!(
+                stage = "backend",
+                endpoint = %client.endpoint(),
+                error = %e,
+                "BackendPlan push failed; next replan cycle will retry"
+            );
+        }
+    }
+}
+
 /// Update the cumulative cache with `be` for `(metric, role)` and
 /// POST the cumulative streaming-config + storage-routing JSON
 /// documents to the backend.
@@ -435,6 +473,26 @@ async fn push_cumulative_entries(
         }
     };
 
+    // BackendPlan (design-backend-plan-wire-format.md): built from the
+    // SAME `cumulative_be` snapshot as the legacy documents above, so all
+    // three describe one consistent generation of planning state. This
+    // is a dual-push, alongside (not instead of) the legacy
+    // streaming-config / storage-routing documents — a failure here must
+    // never affect `PushOutcome`, which existing callers key real
+    // behavior on.
+    let plan_bytes = match crate::backend_plan::from_stage_config(
+        &cumulative_be,
+        monitors,
+        PLAN_ID_COUNTER.fetch_add(1, Ordering::Relaxed),
+        now_unix_ms(),
+    ) {
+        Ok(plan) => Some(plan.encode_to_vec()),
+        Err(e) => {
+            warn!(error = %e, "backend_plan::from_stage_config failed; skipping BackendPlan push (legacy push unaffected)");
+            None
+        }
+    };
+
     // Storage-routing: the routing classifier (`build_routing_entry` in
     // `emit/stage_config.rs`) reads `cfg.aggregations` to derive shape
     // routing, so we MUST merge every role's aggregations for one metric
@@ -495,6 +553,13 @@ async fn push_cumulative_entries(
     let (streaming_ok, routing_ok, attempts) =
         push_documents_coupled(client, streaming_body, routing_body).await;
 
+    // Best-effort BackendPlan push — same backoff schedule as the legacy
+    // documents, but its own outcome never feeds into `PushOutcome` (see
+    // this function's doc above `plan_bytes`).
+    if let Some(bytes) = plan_bytes {
+        push_backend_plan_best_effort(client, bytes).await;
+    }
+
     if streaming_ok && routing_ok {
         info!(
             stage = "backend",
@@ -770,6 +835,7 @@ mod tests {
     struct DualMock {
         streaming_hits: StdArc<StdAtomicU32>,
         routing_hits: StdArc<StdAtomicU32>,
+        plan_hits: StdArc<StdAtomicU32>,
         streaming_status: axum::http::StatusCode,
         routing_status: axum::http::StatusCode,
     }
@@ -781,6 +847,7 @@ mod tests {
         let mock = DualMock {
             streaming_hits: StdArc::new(StdAtomicU32::new(0)),
             routing_hits: StdArc::new(StdAtomicU32::new(0)),
+            plan_hits: StdArc::new(StdAtomicU32::new(0)),
             streaming_status,
             routing_status,
         };
@@ -803,6 +870,15 @@ mod tests {
                     },
                 ),
             )
+            .route(
+                "/api/v1/backend-plan",
+                post(
+                    |State(m): State<DualMock>, _body: axum::body::Bytes| async move {
+                        m.plan_hits.fetch_add(1, StdOrdering::SeqCst);
+                        axum::http::StatusCode::OK
+                    },
+                ),
+            )
             .with_state(mock.clone());
         let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
         let addr = listener.local_addr().unwrap();
@@ -836,6 +912,70 @@ mod tests {
         assert_eq!(mock.routing_hits.load(StdOrdering::SeqCst), 1);
     }
 
+    /// The dual-push also fires a best-effort `POST /api/v1/backend-plan`,
+    /// alongside — not instead of — the legacy documents.
+    #[tokio::test]
+    async fn coupled_push_also_fires_backend_plan_push() {
+        let (url, mock) =
+            start_dual_mock(axum::http::StatusCode::OK, axum::http::StatusCode::OK).await;
+        let client = StdArc::new(BackendClient::new(url));
+        let cache = Mutex::new(HashMap::new());
+        let outcome = post_typed_backend_for_role(
+            Some(&client),
+            &cache,
+            "latency",
+            AggRole::Quantile,
+            make_be("latency", "q"),
+            &[],
+        )
+        .await;
+        assert_eq!(outcome, PushOutcome::BothApplied);
+        assert_eq!(mock.plan_hits.load(StdOrdering::SeqCst), 1);
+    }
+
+    /// A BackendPlan push failure (backend doesn't implement the
+    /// endpoint yet, or returns an error) must NOT affect `PushOutcome`
+    /// — nothing depends on the plan push succeeding in this phase.
+    #[tokio::test]
+    async fn backend_plan_push_failure_does_not_affect_push_outcome() {
+        // A mock that only serves the legacy endpoints (no
+        // `/api/v1/backend-plan` route) — the plan push 404s.
+        let app = Router::new()
+            .route(
+                "/api/v1/streaming-config",
+                post(|_body: axum::body::Bytes| async { axum::http::StatusCode::OK }),
+            )
+            .route(
+                "/api/v1/storage_routing",
+                post(|_body: axum::body::Bytes| async { axum::http::StatusCode::OK }),
+            );
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+        tokio::time::sleep(Duration::from_millis(50)).await;
+
+        let client = StdArc::new(BackendClient::new(format!(
+            "http://{addr}/api/v1/streaming-config"
+        )));
+        let cache = Mutex::new(HashMap::new());
+        let outcome = post_typed_backend_for_role(
+            Some(&client),
+            &cache,
+            "latency",
+            AggRole::Quantile,
+            make_be("latency", "q"),
+            &[],
+        )
+        .await;
+        assert_eq!(
+            outcome,
+            PushOutcome::BothApplied,
+            "legacy documents must still report success even though the plan push 404s"
+        );
+    }
+
     /// P2-3: streaming-config succeeds (200) but storage-routing always
     /// returns a PERMANENT 400. The coupled push surfaces
     /// `Desynced { streaming_ok: true, routing_ok: false }` rather than a
diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs
index 68ebcd98..ade65c60 100644
--- a/control_plane/src/emit/stage_config.rs
+++ b/control_plane/src/emit/stage_config.rs
@@ -2962,7 +2962,7 @@ fn build_gateway_merge_block(mp: &GatewayMergeProcessor) -> Value {
 /// retired the controller-allocated id; identity is content-addressed
 /// in the backend via `PolicyFingerprint(u64)` derived from the fields
 /// above.
-fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonValue {
+pub(crate) fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonValue {
     // Option B (post-PR-#287): when `agg_type_override` is set, use
     // it as the wire `aggregationType` and emit an empty
     // `parameters` object — bypasses the sketch_kind → backend type
diff --git a/control_plane/src/sketch_algebra/cost_model.rs b/control_plane/src/sketch_algebra/cost_model.rs
index 9fbd8955..26258f81 100644
--- a/control_plane/src/sketch_algebra/cost_model.rs
+++ b/control_plane/src/sketch_algebra/cost_model.rs
@@ -399,6 +399,15 @@ impl CostModel for ForcedFamilyCostModel {
 /// `size_params` then fall back to the accuracy-driven default, which
 /// won't match anything registered either way, so the outcome
 /// (`find_candidates` finds nothing) is unchanged.
+///
+/// **Fallback status (design-backend-plan-wire-format.md §5):**
+/// `l4_lowering.rs` prefers reading planning's decision directly off an
+/// installed `BackendPlan`'s materializations (no reconstruction needed
+/// there — `Materialization.kind`/`.params` already ARE the pair
+/// `observed` needs). This type's caller
+/// (`observed_family_for_metric`, the `SketchStore`-metadata
+/// reconstruction) is the fallback for deploys with no `BackendPlan`
+/// installed yet, or for metrics a partial/stale plan doesn't cover.
 pub struct ObservedFamilyCostModel {
     inner: ControlPlaneCostModel,
     observed: Option<(SummaryKind, SummaryParams)>,
diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs
index 0a6fbb6c..b7d2b284 100644
--- a/data_plane/src/drivers/query/servers/http.rs
+++ b/data_plane/src/drivers/query/servers/http.rs
@@ -166,6 +166,12 @@ pub struct HttpServer {
     /// Hot-reloadable `StreamingConfig` source. `None` when hot-reload
     /// is not wired up by the caller (unit tests, legacy binaries).
     hot_reload_config: Option<crate::storage_engines::types::HotReloadStreamingConfig>,
+    /// Hot-reloadable `BackendPlan` handle for `GET/POST
+    /// /api/v1/backend-plan` (see `control_plane/docs/design-backend-plan-wire-format.md`).
+    /// `None` when not wired up (unit tests, legacy binaries) — the
+    /// endpoints return `503`. Shared with `ASAPQueryEngine` so its
+    /// serving-time lookup sees the same installed plan.
+    hot_reload_backend_plan: Option<crate::storage_engines::types::HotReloadBackendPlan>,
     /// Per-metric storage-backend routing table consulted by the HTTP
     /// instant-query handler at request time. When `Some(..)` and the
     /// query parses, the handler extracts the metric name from the
@@ -228,6 +234,8 @@ struct AppState {
     adapter: Arc<dyn HttpProtocolAdapter>,
     fallback: Option<Arc<dyn crate::drivers::query::fallback::FallbackClient>>,
     hot_reload_config: Option<crate::storage_engines::types::HotReloadStreamingConfig>,
+    /// See [`HttpServer::hot_reload_backend_plan`].
+    hot_reload_backend_plan: Option<crate::storage_engines::types::HotReloadBackendPlan>,
     /// See [`HttpServer::backend_storage_routing`].
     backend_storage_routing: Option<crate::query_engines::routing::HotReloadBackendStorageRouting>,
     /// Backfill registry (sketch DB §10). See `HttpServer::backfill`.
@@ -257,6 +265,7 @@ impl HttpServer {
             query_router,
             sketch_index,
             hot_reload_config: None,
+            hot_reload_backend_plan: None,
             backend_storage_routing: None,
             backfill: None,
             data_retention_ms: None,
@@ -305,6 +314,20 @@ impl HttpServer {
         self
     }
 
+    /// Attach a `HotReloadBackendPlan` handle so the
+    /// `GET/POST /api/v1/backend-plan` endpoints can install and read
+    /// the control plane's typed `BackendPlan` push. Additive alongside
+    /// [`Self::with_hot_reload_config`] — without this handle the
+    /// endpoints return `503 Service Unavailable`, same contract as the
+    /// legacy streaming-config handle.
+    pub fn with_hot_reload_backend_plan(
+        mut self,
+        handle: crate::storage_engines::types::HotReloadBackendPlan,
+    ) -> Self {
+        self.hot_reload_backend_plan = Some(handle);
+        self
+    }
+
     /// Attach a per-metric storage-backend routing table. The table is
     /// wrapped in a hot-reload handle internally so the
     /// `POST /api/v1/storage_routing` endpoint (Phase α) can swap it
@@ -415,6 +438,7 @@ impl HttpServer {
             adapter: adapter.clone(),
             fallback: self.config.adapter_config.fallback.clone(),
             hot_reload_config: self.hot_reload_config.clone(),
+            hot_reload_backend_plan: self.hot_reload_backend_plan.clone(),
             backend_storage_routing: self.backend_storage_routing.clone(),
             backfill: self.backfill.clone(),
             data_retention_ms: self.data_retention_ms,
@@ -452,6 +476,13 @@ impl HttpServer {
                 "/api/v1/streaming-config",
                 get(handle_get_streaming_config).post(handle_post_streaming_config),
             )
+            // BackendPlan wire format (design-backend-plan-wire-format.md):
+            // sibling of streaming-config above, read by ASAPQueryEngine's
+            // serving-time lookup. POST body is raw protobuf bytes.
+            .route(
+                "/api/v1/backend-plan",
+                get(handle_get_backend_plan).post(handle_post_backend_plan),
+            )
             // Phase α (MVP): control-plane-pushed `BackendStorageRouting`
             // table. POST replaces the current table atomically; GET
             // returns a JSON snapshot for operator diagnostics.
@@ -507,6 +538,7 @@ impl HttpServer {
             adapter: adapter.clone(),
             fallback: self.config.adapter_config.fallback.clone(),
             hot_reload_config: self.hot_reload_config.clone(),
+            hot_reload_backend_plan: self.hot_reload_backend_plan.clone(),
             backend_storage_routing: self.backend_storage_routing.clone(),
             backfill: self.backfill.clone(),
             data_retention_ms: self.data_retention_ms,
@@ -526,6 +558,13 @@ impl HttpServer {
                 "/api/v1/streaming-config",
                 get(handle_get_streaming_config).post(handle_post_streaming_config),
             )
+            // BackendPlan wire format (design-backend-plan-wire-format.md):
+            // sibling of streaming-config above, read by ASAPQueryEngine's
+            // serving-time lookup. POST body is raw protobuf bytes.
+            .route(
+                "/api/v1/backend-plan",
+                get(handle_get_backend_plan).post(handle_post_backend_plan),
+            )
             // Phase α (MVP): control-plane-pushed `BackendStorageRouting`
             // table. POST replaces the current table atomically; GET
             // returns a JSON snapshot for operator diagnostics.
@@ -2209,6 +2248,34 @@ mod tests {
             .expect("Failed to start test server")
     }
 
+    async fn setup_test_server_with_backend_plan(
+        hot_reload: Option<crate::storage_engines::types::HotReloadBackendPlan>,
+    ) -> u16 {
+        let adapter_config = AdapterConfig::prometheus_promql(
+            "http://127.0.0.1:9999".to_string(),
+            false,
+        );
+        let config = HttpServerConfig {
+            port: 0,
+            handle_http_requests: true,
+            adapter_config,
+        };
+        let streaming_config = Arc::new(StreamingConfig::default());
+        let query_engine = Arc::new(ASAPQueryEngine::new(streaming_config.clone(), 15000));
+        let mut server = HttpServer::new(
+            config,
+            query_engine,
+            Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()),
+        );
+        if let Some(handle) = hot_reload {
+            server = server.with_hot_reload_backend_plan(handle);
+        }
+        server
+            .start_test_server()
+            .await
+            .expect("Failed to start test server")
+    }
+
     #[tokio::test]
     async fn test_get_endpoint_plus_symbol_decoding() {
         // Enable debug logging for this test
@@ -2421,6 +2488,105 @@ aggregations:
         assert_eq!(body["status"], "error");
     }
 
+    // ── BackendPlan hot-reload (design-backend-plan-wire-format.md) ─────
+
+    /// POST an encoded `BackendPlan` and verify the active state via GET
+    /// reflects the swap, and that the underlying hot-reload handle
+    /// (cloned into the server at setup) sees it too — mirroring
+    /// `test_streaming_config_hot_reload_round_trip`.
+    #[tokio::test]
+    async fn test_backend_plan_hot_reload_round_trip() {
+        use control_plane::backend_plan::BackendPlan;
+
+        let hot_reload =
+            crate::storage_engines::types::HotReloadBackendPlan::new(BackendPlan::default());
+        let server_port = setup_test_server_with_backend_plan(Some(hot_reload.clone())).await;
+        let client = Client::new();
+
+        let initial = client
+            .get(format!("http://127.0.0.1:{server_port}/api/v1/backend-plan"))
+            .send()
+            .await
+            .expect("GET failed");
+        assert!(initial.status().is_success());
+        let initial_body: serde_json::Value = initial.json().await.unwrap();
+        assert_eq!(initial_body["materialization_count"], 0);
+
+        let new_plan = BackendPlan {
+            plan_id: 7,
+            generated_at_unix_ms: 123,
+            ..Default::default()
+        };
+        let bytes = new_plan.encode_to_vec();
+
+        let post_resp = client
+            .post(format!("http://127.0.0.1:{server_port}/api/v1/backend-plan"))
+            .header("content-type", "application/x-protobuf")
+            .body(bytes)
+            .send()
+            .await
+            .expect("POST failed");
+        let post_status = post_resp.status();
+        let post_body: serde_json::Value = post_resp.json().await.unwrap();
+        assert!(
+            post_status.is_success(),
+            "POST returned {post_status}: {post_body}"
+        );
+        assert_eq!(post_body["status"], "success");
+        assert_eq!(post_body["plan_id"], 7);
+
+        let after = client
+            .get(format!("http://127.0.0.1:{server_port}/api/v1/backend-plan"))
+            .send()
+            .await
+            .expect("GET after swap failed");
+        let after_body: serde_json::Value = after.json().await.unwrap();
+        assert_eq!(after_body["plan_id"], 7);
+        assert_eq!(after_body["generated_at_unix_ms"], 123);
+
+        assert_eq!(hot_reload.snapshot().plan_id, 7);
+    }
+
+    #[tokio::test]
+    async fn test_backend_plan_hot_reload_missing_handle_503() {
+        let server_port = setup_test_server_with_backend_plan(None).await;
+        let client = Client::new();
+
+        let get_resp = client
+            .get(format!("http://127.0.0.1:{server_port}/api/v1/backend-plan"))
+            .send()
+            .await
+            .unwrap();
+        assert_eq!(get_resp.status(), reqwest::StatusCode::SERVICE_UNAVAILABLE);
+
+        let post_resp = client
+            .post(format!("http://127.0.0.1:{server_port}/api/v1/backend-plan"))
+            .body("anything")
+            .send()
+            .await
+            .unwrap();
+        assert_eq!(post_resp.status(), reqwest::StatusCode::SERVICE_UNAVAILABLE);
+    }
+
+    #[tokio::test]
+    async fn test_backend_plan_hot_reload_rejects_bad_bytes() {
+        use control_plane::backend_plan::BackendPlan;
+        let hot_reload =
+            crate::storage_engines::types::HotReloadBackendPlan::new(BackendPlan::default());
+        let server_port = setup_test_server_with_backend_plan(Some(hot_reload)).await;
+        let client = Client::new();
+
+        let resp = client
+            .post(format!("http://127.0.0.1:{server_port}/api/v1/backend-plan"))
+            .body(vec![0xFFu8, 0xFF, 0xFF])
+            .send()
+            .await
+            .unwrap();
+        assert_eq!(resp.status(), reqwest::StatusCode::BAD_REQUEST);
+        let body: serde_json::Value = resp.json().await.unwrap();
+        assert_eq!(body["status"], "error");
+    }
+
     /// Set up a test server wired with a hot-reload handle and a
     /// shared `SketchStore` (the sid catalog the new sid-level
     /// reconcile reads + writes). Returns `(port, sketch_index)` so
@@ -5380,6 +5546,75 @@ async fn handle_post_streaming_config(
     (StatusCode::OK, axum::Json(body)).into_response()
 }
 
+// ── BackendPlan hot-reload (design-backend-plan-wire-format.md) ─────────
+//
+// `GET /api/v1/backend-plan`  — return the currently installed plan as
+//                                JSON (debug / verification).
+// `POST /api/v1/backend-plan` — accept a protobuf body, decode, and
+//                                atomically swap via ArcSwap.
+//
+// Sits alongside `/api/v1/streaming-config`, not in place of it — the
+// swap here does NOT touch the sid catalog / SketchStore reconciliation;
+// that lifecycle management stays on the streaming-config path.
+
+async fn handle_get_backend_plan(State(state): State<AppState>) -> axum::response::Response {
+    use axum::http::StatusCode;
+    use axum::response::IntoResponse;
+
+    let Some(handle) = state.hot_reload_backend_plan else {
+        let body = serde_json::json!({
+            "status": "error",
+            "error": "hot-reload backend-plan handle not attached; backend was built without HttpServer::with_hot_reload_backend_plan"});
+        return (StatusCode::SERVICE_UNAVAILABLE, axum::Json(body)).into_response();
+    };
+    let snap = handle.snapshot();
+    let body = serde_json::json!({
+        "status": "success",
+        "plan_id": snap.plan_id,
+        "generated_at_unix_ms": snap.generated_at_unix_ms,
+        "materialization_count": snap.materializations.len(),
+        "routing_count": snap.routing.len(),
+        "monitor_count": snap.monitors.len()});
+    (StatusCode::OK, axum::Json(body)).into_response()
+}
+
+async fn handle_post_backend_plan(
+    State(state): State<AppState>,
+    body: axum::body::Bytes,
+) -> axum::response::Response {
+    use axum::http::StatusCode;
+    use axum::response::IntoResponse;
+
+    let Some(handle) = state.hot_reload_backend_plan else {
+        let body = serde_json::json!({
+            "status": "error",
+            "error": "hot-reload backend-plan handle not attached; backend was built without HttpServer::with_hot_reload_backend_plan"});
+        return (StatusCode::SERVICE_UNAVAILABLE, axum::Json(body)).into_response();
+    };
+
+    let new_plan = match control_plane::backend_plan::BackendPlan::decode(&body) {
+        Ok(p) => p,
+        Err(e) => {
+            let body = serde_json::json!({
+                "status": "error",
+                "error": format!("BackendPlan decode error: {e}")});
+            return (StatusCode::BAD_REQUEST, axum::Json(body)).into_response();
+        }
+    };
+
+    let materialization_count = new_plan.materializations.len();
+    let routing_count = new_plan.routing.len();
+    let plan_id = new_plan.plan_id;
+    handle.swap(new_plan);
+
+    let body = serde_json::json!({
+        "status": "success",
+        "plan_id": plan_id,
+        "materialization_count": materialization_count,
+        "routing_count": routing_count});
+    (StatusCode::OK, axum::Json(body)).into_response()
+}
+
 // ── Phase α: BackendStorageRouting hot-reload endpoints ────────────
 
 /// `GET /api/v1/storage_routing` — return a JSON snapshot of the
diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs
index 9be3dc2b..18291a1f 100644
--- a/data_plane/src/main.rs
+++ b/data_plane/src/main.rs
@@ -418,6 +418,18 @@ async fn main() -> Result<()> {
         None
     };
 
+    // BackendPlan wire format (design-backend-plan-wire-format.md):
+    // install an empty hot-reload handle so `GET/POST
+    // /api/v1/backend-plan` don't 503 before the control plane's first
+    // push lands — same "install empty, let the first push fill it in"
+    // pattern as `bootstrap_routing` below. Shared with both the query
+    // engine (serving-time cutover, Phase 4) and the HTTP server (the
+    // push target) so a POST is observable by the next query, same
+    // sharing contract as `hot_reload_config`.
+    let hot_reload_backend_plan = data_plane::storage_engines::types::HotReloadBackendPlan::new(
+        control_plane::backend_plan::BackendPlan::default(),
+    );
+
     // Setup query engine. ASAPQueryEngine shares the same
     // HotReloadStreamingConfig handle as the HTTP server, so a POST
     // to /api/v1/streaming-config is observable by the next query
@@ -434,7 +446,8 @@ async fn main() -> Result<()> {
         // drives the Phase 6 archive failover via
         // EngineError::CapabilityMiss when the ASAP tier is empty
         // / ghost / unknown.
-        .with_sketch_index(sketch_index.clone());
+        .with_sketch_index(sketch_index.clone())
+        .with_hot_reload_backend_plan(hot_reload_backend_plan.clone());
         if let Some(control_plane_endpoint) = args.control_plane_endpoint.as_ref() {
             info!(
                 "Capability-miss notifications enabled → {}",
@@ -714,6 +727,7 @@ async fn main() -> Result<()> {
     // `SketchStore` (already passed in below).
     let mut server = HttpServer::new(http_config, engine, sketch_index.clone())
         .with_hot_reload_config(hot_reload_config.clone())
+        .with_hot_reload_backend_plan(hot_reload_backend_plan.clone())
         .with_probe_cache(probe_cache.clone());
 
     // Per-metric storage-backend routing table (issue #46
diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs
index c0a3cc5e..f33c215d 100644
--- a/data_plane/src/query_engines/asap_query_engine/engine.rs
+++ b/data_plane/src/query_engines/asap_query_engine/engine.rs
@@ -56,6 +56,14 @@ pub struct ASAPQueryEngine {
     /// the rest of the routing matrix.
     archive_engine:
         Option<Arc<dyn crate::query_engines::routing::query_engine_routing::QueryEngine>>,
+    /// BackendPlan wire format (design-backend-plan-wire-format.md). When
+    /// `Some`, `l4_lowering.rs`'s serving-time family/params lookup
+    /// prefers reading the installed plan's materializations directly
+    /// over reconstructing from `SketchStore` metadata
+    /// (`ObservedFamilyCostModel`). `None` when not wired up (unit
+    /// tests, legacy callers), which falls back to `SketchStore`
+    /// reconstruction only.
+    hot_reload_backend_plan: Option<crate::storage_engines::types::HotReloadBackendPlan>,
 }
 
 impl ASAPQueryEngine {
@@ -86,9 +94,30 @@ impl ASAPQueryEngine {
             control_plane_client: None,
             sketch_index: None,
             archive_engine: None,
+            hot_reload_backend_plan: None,
         }
     }
 
+    /// Attach a `HotReloadBackendPlan` handle so serving-time family/params
+    /// lookups prefer the control plane's installed `BackendPlan` over
+    /// `SketchStore` reconstruction (see this struct's field doc).
+    /// Without this call, lookups fall back to `SketchStore`
+    /// reconstruction unconditionally.
+    pub fn with_hot_reload_backend_plan(
+        mut self,
+        handle: crate::storage_engines::types::HotReloadBackendPlan,
+    ) -> Self {
+        self.hot_reload_backend_plan = Some(handle);
+        self
+    }
+
+    /// Snapshot of the currently installed `BackendPlan`, if a hot-reload
+    /// handle is wired up. `None` otherwise — callers fall back to the
+    /// `SketchStore`-reconstruction path.
+    fn backend_plan_snapshot(&self) -> Option<Arc<control_plane::backend_plan::BackendPlan>> {
+        self.hot_reload_backend_plan.as_ref().map(|h| h.snapshot())
+    }
+
     /// Phase-5 hybrid-stitch builder — attach an archive engine the
     /// `QueryEngine` trait adapter will dispatch to when the ASAP-tier
     /// reducer reports a coverage narrower than the requested range.
@@ -423,9 +452,10 @@ impl ASAPQueryEngine {
             // and every other "can't safely serve this way" outcome —
             // none of these are answerable via the sketch tier anymore;
             // the caller fails over to archive.
+            let backend_plan_snap = self.backend_plan_snapshot();
             let live_served_result =
                 crate::query_engines::asap_query_engine::live_serve::try_serve_from_summary_executor(
-                    idx, query, start_ms, end_ms, false,
+                    idx, query, start_ms, end_ms, false, backend_plan_snap.as_deref(),
                 );
             let result = match live_served_result {
                 Some(result) => result,
@@ -1125,6 +1155,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu
                 // `live_serve_hll_global_count_merges_across_sids` for the
                 // former), and every other "can't safely serve this way"
                 // outcome — the caller fails over to archive.
+                let backend_plan_snap = self.backend_plan_snapshot();
                 let live_served_result =
                     crate::query_engines::asap_query_engine::live_serve::try_serve_from_summary_executor(
                         idx,
@@ -1132,6 +1163,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu
                         t0_ms,
                         now_ms,
                         effective_is_cumulative(candidate),
+                        backend_plan_snap.as_deref(),
                     );
                 let result = match live_served_result {
                     Some(r) => r,
diff --git a/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs b/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs
index 08371d1d..1898fa11 100644
--- a/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs
+++ b/data_plane/src/query_engines/asap_query_engine/l4_lowering.rs
@@ -192,6 +192,25 @@ fn observed_family_for_metric(index: &SketchStore, metric: &str) -> Option<(Summ
     None
 }
 
+/// Look up what family/params `plan` says is materialized for `metric` —
+/// the `BackendPlan`-sourced sibling of [`observed_family_for_metric`].
+/// Unlike that function, no reconstruction is needed:
+/// `Materialization.kind`/`.params` already ARE the pair this needs,
+/// straight off the wire the control plane pushed. Returns the first
+/// matching materialization found (mirrors
+/// `observed_family_for_metric`'s "first sketch-typed one found"
+/// semantics); `None` when the plan has no materialization for this
+/// metric.
+fn observed_family_for_metric_from_plan(
+    plan: &control_plane::backend_plan::BackendPlan,
+    metric: &str,
+) -> Option<(SummaryKind, SummaryParams)> {
+    plan.materializations.values().find_map(|m| {
+        matches!(&m.source, control_plane::intent_algebra::Source::TimeSeries { metric: mm } if mm == metric)
+            .then(|| (m.kind.clone(), m.params.clone()))
+    })
+}
+
 /// Lower a raw PromQL query string to the `L4Node` tree
 /// `asap_sketch::exec::execute`/`SummaryExecutor` needs — the actual
 /// serving cutover (`live_serve.rs`). Returns `Err` for any shape serving
@@ -202,6 +221,7 @@ pub fn lower_promql_to_l4node(
     index: &SketchStore,
     query: &str,
     accuracy: AccuracyTarget,
+    backend_plan: Option<&control_plane::backend_plan::BackendPlan>,
 ) -> Result<Rc<L4Node>, LoweringSkip> {
     // Reuse the SAME candidate analysis `engine.rs` already runs for the
     // legacy dispatch, rather than re-deriving rate detection via a
@@ -221,12 +241,22 @@ pub fn lower_promql_to_l4node(
         .map_err(|e| LoweringSkip::ParseFailed(e.to_string()))?;
 
     // Serving time must reproduce the REAL planning decision, not
-    // independently re-derive one -- see this module's docs. `observed`
-    // is `None` when this metric has nothing registered (or only an
-    // `ExactAgg` sid, which bypasses `CostModel` entirely), in which case
-    // `ObservedFamilyCostModel` transparently falls back to the same
-    // accuracy-driven behavior as before.
-    let observed = find_metric_in_query_expr(&qe).and_then(|metric| observed_family_for_metric(index, &metric));
+    // independently re-derive one -- see this module's docs. Prefer
+    // reading it straight off an installed `BackendPlan`'s
+    // materializations when one covers this metric --
+    // `Materialization.kind`/`.params` already ARE the
+    // `(SummaryKind, SummaryParams)` pair this needs, no
+    // `AggregationConfig` reconstruction required (design-backend-plan-wire-format.md
+    // §5). Otherwise fall back to the `SketchStore`-reconstruction path
+    // (`observed_family_for_metric`), which is `None` when this metric
+    // has nothing registered (or only an `ExactAgg` sid, which bypasses
+    // `CostModel` entirely) -- `ObservedFamilyCostModel` then falls back
+    // further to the accuracy-driven default.
+    let observed = find_metric_in_query_expr(&qe).and_then(|metric| {
+        backend_plan
+            .and_then(|plan| observed_family_for_metric_from_plan(plan, &metric))
+            .or_else(|| observed_family_for_metric(index, &metric))
+    });
     let cost_model = ObservedFamilyCostModel::new(accuracy, observed);
 
     let physical = bind_query_expr_with_cost_model(&qe, &cost_model)
@@ -264,7 +294,7 @@ mod tests {
     #[test]
     fn rate_query_is_skipped_before_binding() {
         let idx = empty_index();
-        let result = lower_promql_to_l4node(&idx, "rate(http_requests_total[5m])", accuracy());
+        let result = lower_promql_to_l4node(&idx, "rate(http_requests_total[5m])", accuracy(), None);
         assert!(
             matches!(result, Err(LoweringSkip::RateShape)),
             "expected RateShape, got {result:?}"
@@ -274,7 +304,7 @@ mod tests {
     #[test]
     fn irate_query_is_skipped_before_binding() {
         let idx = empty_index();
-        let result = lower_promql_to_l4node(&idx, "irate(http_requests_total[5m])", accuracy());
+        let result = lower_promql_to_l4node(&idx, "irate(http_requests_total[5m])", accuracy(), None);
         assert!(
             matches!(result, Err(LoweringSkip::RateShape)),
             "expected RateShape, got {result:?}"
@@ -284,7 +314,7 @@ mod tests {
     #[test]
     fn unparseable_query_is_skipped() {
         let idx = empty_index();
-        let result = lower_promql_to_l4node(&idx, "this is not promql (((", accuracy());
+        let result = lower_promql_to_l4node(&idx, "this is not promql (((", accuracy(), None);
         assert!(
             matches!(result, Err(LoweringSkip::ParseFailed(_))),
             "expected ParseFailed, got {result:?}"
@@ -304,7 +334,7 @@ mod tests {
         // expression stays one opaque `Logical` blob, which this module
         // surfaces as `NotRealized`.
         let idx = empty_index();
-        let result = lower_promql_to_l4node(&idx, "http_requests_total", accuracy());
+        let result = lower_promql_to_l4node(&idx, "http_requests_total", accuracy(), None);
         assert!(
             matches!(result, Err(LoweringSkip::NotRealized)),
             "expected NotRealized, got {result:?}"
@@ -326,7 +356,7 @@ mod tests {
         // before this module started consulting the `SketchStore`.
         let idx = empty_index();
         let node =
-            lower_promql_to_l4node(&idx, "count_over_time(http_requests_total[5m])", accuracy())
+            lower_promql_to_l4node(&idx, "count_over_time(http_requests_total[5m])", accuracy(), None)
                 .expect("Frequency intent must realize via bind_query_expr/ControlPlaneCostModel");
         assert!(
             !matches!(node.expr, SummaryExpr::Logical(_)),
@@ -347,10 +377,147 @@ mod tests {
             &idx,
             "topk(5, sum by (host) (rate(http_requests_total[5m])))",
             accuracy(),
+            None,
         );
         assert!(
             matches!(result, Err(LoweringSkip::NotRealized) | Err(LoweringSkip::RateShape)),
             "expected NotRealized or RateShape (both are valid skips for this shape), got {result:?}"
         );
     }
+
+    // ── BackendPlan-sourced family lookup (design-backend-plan-wire-format.md §5) ────
+
+    mod backend_plan_cutover {
+        use super::*;
+        use crate::storage_engines::sketch_db::index::{
+            AccuracyBound, Capability, SketchInstanceMetadata, SketchKindHandle,
+        };
+        use control_plane::backend_plan::{BackendPlan, Materialization, WindowSpec};
+        use control_plane::intent_algebra::{ColumnRef, Source, WindowKind};
+        use std::collections::HashMap;
+
+        fn register_kll(idx: &SketchStore, metric: &str) {
+            let cfg = SketchConfig::Kll { k: 200 };
+            idx.register(SketchInstanceMetadata {
+                sid: 1,
+                metric_name: metric.to_string(),
+                group_by_keys: Default::default(),
+                capability: Some(Capability::QuantileApprox(SketchKindHandle::Kll)),
+                agg_kind: AggKind::Sketch {
+                    kind: SketchKindHandle::Kll,
+                    config: cfg.clone(),
+                    spatial_filter_canonical: String::new(),
+                },
+                accuracy: Some(AccuracyBound::from_config(&cfg)),
+                first_seen_unix_ms: 0,
+                retired_at_ms: None,
+                expires_at_ms: None,
+                policy_fp: asap_types::PolicyFingerprint::UNSET,
+            });
+        }
+
+        fn plan_with_ddsketch_materialization(metric: &str) -> BackendPlan {
+            let fingerprint = asap_types::PolicyFingerprint(42);
+            let mut materializations = HashMap::new();
+            materializations.insert(
+                fingerprint,
+                Materialization {
+                    fingerprint,
+                    source: Source::TimeSeries {
+                        metric: metric.to_string(),
+                    },
+                    window: WindowSpec {
+                        kind: WindowKind::Tumbling,
+                        size_ms: 60_000,
+                        slide_ms: None,
+                    },
+                    group_by: Vec::new(),
+                    rollup: Vec::new(),
+                    kind: SummaryKind::DDSketch,
+                    params: SummaryParams::DDSketch { alpha: 0.01 },
+                    col: ColumnRef::SampleValue,
+                    retention: None,
+                },
+            );
+            BackendPlan {
+                plan_id: 1,
+                generated_at_unix_ms: 0,
+                materializations,
+                routing: Vec::new(),
+                monitors: Vec::new(),
+            }
+        }
+
+        /// Extract the bound `(SummaryKind, SummaryParams)` from the
+        /// `SummaryEstimate { summary_input: L4Node { expr: SummaryAgg {
+        /// summary, params, .. }, .. }, .. }` shape a bare
+        /// `quantile_over_time` query lowers to (confirmed by inspecting
+        /// the tree directly).
+        fn bound_family(node: &L4Node) -> (SummaryKind, SummaryParams) {
+            match &node.expr {
+                SummaryExpr::SummaryEstimate { summary_input, .. } => match &summary_input.expr {
+                    SummaryExpr::SummaryAgg { summary, params, .. } => {
+                        (summary.clone(), params.clone())
+                    }
+                    other => panic!("expected SummaryAgg, got {other:?}"),
+                },
+                other => panic!("expected SummaryEstimate, got {other:?}"),
+            }
+        }
+
+        #[test]
+        fn without_a_plan_sketchstore_reconstruction_wins() {
+            // Baseline: no `BackendPlan` -- `observed_family_for_metric`'s
+            // SketchStore reconstruction is the only source.
+            let idx = SketchStore::new();
+            register_kll(&idx, "m");
+            let node = lower_promql_to_l4node(&idx, "quantile_over_time(0.99, m[1m])", accuracy(), None)
+                .expect("should lower");
+            assert_eq!(bound_family(&node).0, SummaryKind::Kll);
+        }
+
+        #[test]
+        fn a_plan_materialization_wins_over_sketchstore_reconstruction() {
+            // `SketchStore` has Kll registered for `m` (what
+            // reconstruction alone would find), but the installed
+            // `BackendPlan` says DDSketch for
+            // the SAME metric. The plan must win -- serving time reads
+            // planning's real (plan-sourced) decision, not whatever
+            // `SketchStore` metadata happens to reconstruct to.
+            let idx = SketchStore::new();
+            register_kll(&idx, "m");
+            let plan = plan_with_ddsketch_materialization("m");
+            let node = lower_promql_to_l4node(
+                &idx,
+                "quantile_over_time(0.99, m[1m])",
+                accuracy(),
+                Some(&plan),
+            )
+            .expect("should lower");
+            assert_eq!(
+                bound_family(&node).0,
+                SummaryKind::DDSketch,
+                "BackendPlan's materialization must take priority over SketchStore reconstruction"
+            );
+        }
+
+        #[test]
+        fn plan_present_but_no_materialization_for_metric_falls_back_to_sketchstore() {
+            // The plan is installed but doesn't cover THIS metric --
+            // `observed_family_for_metric_from_plan` returns `None` for
+            // it, so the lookup must fall through to SketchStore
+            // reconstruction, not silently fail to observe anything.
+            let idx = SketchStore::new();
+            register_kll(&idx, "m");
+            let plan = plan_with_ddsketch_materialization("some_other_metric");
+            let node = lower_promql_to_l4node(
+                &idx,
+                "quantile_over_time(0.99, m[1m])",
+                accuracy(),
+                Some(&plan),
+            )
+            .expect("should lower");
+            assert_eq!(bound_family(&node).0, SummaryKind::Kll);
+        }
+    }
 }
diff --git a/data_plane/src/query_engines/asap_query_engine/l4_readout.rs b/data_plane/src/query_engines/asap_query_engine/l4_readout.rs
index da4b26ca..15fb3b1c 100644
--- a/data_plane/src/query_engines/asap_query_engine/l4_readout.rs
+++ b/data_plane/src/query_engines/asap_query_engine/l4_readout.rs
@@ -69,8 +69,9 @@ pub fn execute_l4_readout(
     t1_ms: u64,
     is_cumulative: bool,
     accuracy: AccuracyTarget,
+    backend_plan: Option<&control_plane::backend_plan::BackendPlan>,
 ) -> Result<L4ReadoutOutcome, LoweringSkip> {
-    let node = lower_promql_to_l4node(index, query, accuracy)?;
+    let node = lower_promql_to_l4node(index, query, accuracy, backend_plan)?;
 
     let ctx = QueryExecutionContext {
         index,
@@ -253,6 +254,7 @@ mod tests {
             2_000,
             true,
             accuracy(),
+            None,
         )
         .expect("should execute");
         assert_eq!(outcome.series.len(), 1);
@@ -277,7 +279,7 @@ mod tests {
         register_hll(&idx, 1, "svc-a", &["a", "b", "c"]);
         register_hll(&idx, 2, "svc-b", &["d", "e", "f"]);
         let outcome =
-            execute_l4_readout(&idx, "count(unique_users)", 1_000, 2_000, true, accuracy())
+            execute_l4_readout(&idx, "count(unique_users)", 1_000, 2_000, true, accuracy(), None)
                 .expect("should execute");
         assert_eq!(
             outcome.series.len(),
@@ -322,7 +324,7 @@ mod tests {
             (1_000, 2_000),
             Box::new(crate::precompute_engine::operators::SumAccumulator::with_sum(42.0)),
         );
-        let outcome = execute_l4_readout(&idx, "sum(bytes_total)", 1_000, 2_000, true, accuracy())
+        let outcome = execute_l4_readout(&idx, "sum(bytes_total)", 1_000, 2_000, true, accuracy(), None)
             .expect("should execute");
         // Window-end-only coverage: a single window (1_000, 2_000) is
         // keyed by its end (2_000) alone, so both bounds equal 2_000 --
diff --git a/data_plane/src/query_engines/asap_query_engine/live_serve.rs b/data_plane/src/query_engines/asap_query_engine/live_serve.rs
index 1bbf0489..360358d6 100644
--- a/data_plane/src/query_engines/asap_query_engine/live_serve.rs
+++ b/data_plane/src/query_engines/asap_query_engine/live_serve.rs
@@ -73,13 +73,21 @@ pub fn try_serve_from_summary_executor(
     t0_ms: u64,
     t1_ms: u64,
     is_cumulative: bool,
+    backend_plan: Option<&control_plane::backend_plan::BackendPlan>,
 ) -> Option<ASAPTierResult> {
     if !summary_executor_live_enabled() {
         return None;
     }
 
-    let outcome = match execute_l4_readout(index, query, t0_ms, t1_ms, is_cumulative, LIVE_ACCURACY)
-    {
+    let outcome = match execute_l4_readout(
+        index,
+        query,
+        t0_ms,
+        t1_ms,
+        is_cumulative,
+        LIVE_ACCURACY,
+        backend_plan,
+    ) {
         Ok(outcome) => outcome,
         Err(skip) => {
             tracing::debug!(
@@ -227,6 +235,7 @@ mod tests {
             1_000,
             2_000,
             true,
+            None,
         );
         assert!(result.is_none(), "flag explicitly off must never serve");
     }
@@ -244,6 +253,7 @@ mod tests {
             1_000,
             2_000,
             true,
+            None,
         );
         assert!(result.is_some(), "unset flag must default to serving");
     }
@@ -258,6 +268,7 @@ mod tests {
             1_000,
             2_000,
             true,
+            None,
         );
         let result = result.expect("unambiguous single-series quantile must serve");
         assert_eq!(result.series.len(), 1);
@@ -278,7 +289,7 @@ mod tests {
         register_hll(&idx, 1, "svc-a", &["a", "b", "c"]);
         register_hll(&idx, 2, "svc-b", &["d", "e", "f"]);
         let result =
-            try_serve_from_summary_executor(&idx, "count(unique_users)", 1_000, 2_000, true);
+            try_serve_from_summary_executor(&idx, "count(unique_users)", 1_000, 2_000, true, None);
         let result = result.expect(
             "global-merge shape is no longer ambiguous -- it must be served, not declined",
         );
@@ -301,7 +312,7 @@ mod tests {
         let _guard = set_live_env("1");
         let idx = SketchStore::new();
         let result =
-            try_serve_from_summary_executor(&idx, "rate(http_requests_total[5m])", 0, 1000, true);
+            try_serve_from_summary_executor(&idx, "rate(http_requests_total[5m])", 0, 1000, true, None);
         assert!(result.is_none());
     }
 }
diff --git a/data_plane/src/storage_engines/types/hot_reload_config.rs b/data_plane/src/storage_engines/types/hot_reload_config.rs
index b81c832d..66b56372 100644
--- a/data_plane/src/storage_engines/types/hot_reload_config.rs
+++ b/data_plane/src/storage_engines/types/hot_reload_config.rs
@@ -81,6 +81,97 @@ use arc_swap::ArcSwap;
 
 use crate::storage_engines::types::StreamingConfig;
 
+/// Hot-reloadable `BackendPlan` state — same `ArcSwap` shape as
+/// [`HotReloadStreamingConfig`], applied to
+/// `control_plane::backend_plan::BackendPlan` (see
+/// `control_plane/docs/design-backend-plan-wire-format.md`). Lives
+/// alongside [`HotReloadStreamingConfig`], not in place of it:
+/// `POST /api/v1/backend-plan` installs the latest plan here for
+/// `ASAPQueryEngine`'s serving-time lookup to read, while
+/// `POST /api/v1/streaming-config` still drives sid-catalog lifecycle
+/// (registration/retirement) on its own path.
+#[derive(Clone)]
+pub struct HotReloadBackendPlan {
+    inner: Arc<ArcSwap<control_plane::backend_plan::BackendPlan>>,
+}
+
+impl HotReloadBackendPlan {
+    pub fn new(initial: control_plane::backend_plan::BackendPlan) -> Self {
+        Self {
+            inner: Arc::new(ArcSwap::new(Arc::new(initial))),
+        }
+    }
+
+    pub fn from_arc(initial: Arc<control_plane::backend_plan::BackendPlan>) -> Self {
+        Self {
+            inner: Arc::new(ArcSwap::new(initial)),
+        }
+    }
+
+    pub fn snapshot(&self) -> Arc<control_plane::backend_plan::BackendPlan> {
+        self.inner.load_full()
+    }
+
+    pub fn swap(
+        &self,
+        new: control_plane::backend_plan::BackendPlan,
+    ) -> Arc<control_plane::backend_plan::BackendPlan> {
+        self.inner.swap(Arc::new(new))
+    }
+}
+
+impl Default for HotReloadBackendPlan {
+    fn default() -> Self {
+        Self::new(control_plane::backend_plan::BackendPlan::default())
+    }
+}
+
+impl std::fmt::Debug for HotReloadBackendPlan {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let snap = self.snapshot();
+        f.debug_struct("HotReloadBackendPlan")
+            .field("plan_id", &snap.plan_id)
+            .field("materializations", &snap.materializations.len())
+            .field("routing", &snap.routing.len())
+            .finish()
+    }
+}
+
+#[cfg(test)]
+mod hot_reload_backend_plan_tests {
+    use super::*;
+    use control_plane::backend_plan::BackendPlan;
+
+    fn plan(plan_id: u64) -> BackendPlan {
+        BackendPlan {
+            plan_id,
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn snapshot_reflects_initial_plan() {
+        let hr = HotReloadBackendPlan::new(plan(1));
+        assert_eq!(hr.snapshot().plan_id, 1);
+    }
+
+    #[test]
+    fn swap_replaces_plan_atomically() {
+        let hr = HotReloadBackendPlan::new(plan(1));
+        let old = hr.swap(plan(2));
+        assert_eq!(old.plan_id, 1, "swap returns the pre-swap snapshot");
+        assert_eq!(hr.snapshot().plan_id, 2);
+    }
+
+    #[test]
+    fn clones_share_underlying_swap() {
+        let hr = HotReloadBackendPlan::new(plan(1));
+        let hr_clone = hr.clone();
+        hr.swap(plan(2));
+        assert_eq!(hr_clone.snapshot().plan_id, 2);
+    }
+}
+
 /// Thin wrapper around `ArcSwap<StreamingConfig>` with ergonomic
 /// snapshot + swap helpers. Cloneable; clones share the same
 /// underlying `ArcSwap` so all holders see the same swaps.