refactor: execute precompute DAGs with shared physical operators

control_plane/Cargo.toml

@@ -12,7 +12,6 @@ name = "control_plane"
 path = "src/main.rs"
 
 [dependencies]
-asap-physical-operators.workspace = true
 tokio       = { version = "1",    features = ["full"] }
 axum        = { version = "0.7",  features = ["ws"] }
 futures-util = "0.3"
@@ -33,6 +32,7 @@ parking_lot    = "0.12"
 prometheus     = { version = "0.13", default-features = false, features = ["process"] }
 tonic          = { version = "0.12", features = ["gzip"] }
 asap_types.workspace = true
+asap-physical-operators.workspace = true
 
 # Planner's IR, replacement search, and frontends share the workspace pin.
 planner-types.workspace = true

control_plane/src/backend_client.rs

@@ -1,22 +1,11 @@
-//! HTTP client that pushes a freshly-generated `StreamingConfig` YAML
-//! to the ASAPQuery-backend's `POST /api/v1/streaming-config` endpoint.
-//!
-//! This is the control-plane-side **producer** of the PR E phase 1 / phase 2
-//! hot-reload contract that landed in ASAPQuery-backend PRs #10 and #12.
-//! The replanner calls into this module immediately after generating a
-//! new plan so the backend's active `StreamingConfig` is updated without
-//! a restart and subsequent queries observe the new aggregation layout.
-//!
-//! The client is **fire-and-forget at the call site** — the replanner
-//! awaits the POST but doesn't block its own return on the outcome.
-//! Errors are logged at WARN; the control plane is expected to be tolerant
-//! of transient backend unavailability because the next replan cycle
-//! will try again with the latest plan.
+//! Publish and activate complete physical-plan generations on the backend.
 
 use std::time::Duration;
 
+#[cfg(test)]
 use anyhow::{Context, Result};
 use reqwest::Client;
+#[cfg(test)]
 use tracing::debug;
 #[cfg(test)]
 use tracing::warn;
@@ -92,7 +81,7 @@ fn classify_http_status(status: reqwest::StatusCode, body: String, what: &str) -
     }
 }
 
-/// Minimal HTTP client for ASAPQuery-backend's streaming-config endpoint.
+/// Minimal HTTP client for ASAPQuery-backend's physical-plan endpoint.
 /// Built once at control-plane startup from the
 /// `CONTROL_PLANE_BACKEND_ENDPOINT` environment variable and shared
 /// via `Arc` with the replanner.
@@ -105,7 +94,7 @@ pub struct BackendClient {
 impl BackendClient {
     /// Construct a client pointing at the backend's plan-push endpoint.
     /// `endpoint` should be the full URL, e.g.
-    /// `http://backend.svc:8088/api/v1/streaming-config`.
+    /// `http://backend.svc:8088/api/v1/physical-plan`.
     ///
     /// A 5-second timeout bounds the duration a slow or unreachable
     /// backend can stall the replanner — consistent with the symmetric
@@ -136,11 +125,12 @@ impl BackendClient {
         &self.endpoint
     }
 
-    /// POST the given `StreamingConfig` YAML to the backend. Returns
+    /// Test-only transport helper: POST the given YAML to the backend. Returns
     /// `Ok(())` on any 2xx status, otherwise an error carrying the
     /// status code and response body. The caller (typically
     /// [`Replanner::replan_metric`]) logs the error and moves on — the
     /// next replan cycle will retry with the latest plan.
+    #[cfg(test)]
     pub async fn push_streaming_config(&self, yaml: String) -> Result<()> {
         debug!(
             endpoint = %self.endpoint,
@@ -212,6 +202,7 @@ impl BackendClient {
     /// that don't need retry semantics keep their `anyhow::Result`
     /// shape. The retry layer in `emit::backend_push` uses this typed
     /// variant.
+    #[cfg(test)]
     pub async fn post_streaming_config_json_typed(
         &self,
         json: String,
@@ -245,7 +236,7 @@ impl BackendClient {
 
     #[cfg(test)]
     /// Post backend storage-routing JSON. Derive the URL by replacing the
-    /// `/api/v1/streaming-config` suffix with `/api/v1/storage_routing`; URLs
+    /// `/api/v1/physical-plan` suffix with `/api/v1/storage_routing`; URLs
     /// without that suffix are used verbatim.
     pub async fn post_storage_routing_json(&self, json: String) -> Result<()> {
         let url = derive_storage_routing_url(&self.endpoint);
@@ -364,25 +355,18 @@ impl BackendClient {
 }
 
 fn derive_physical_plan_url(endpoint: &str) -> String {
-    const DASH: &str = "/api/v1/streaming-config";
-    const UNDERSCORE: &str = "/api/v1/streaming_config";
-    const PHYSICAL: &str = "/api/v1/physical-plan";
-    endpoint
-        .strip_suffix(DASH)
-        .or_else(|| endpoint.strip_suffix(UNDERSCORE))
-        .map(|base| format!("{base}{PHYSICAL}"))
-        .unwrap_or_else(|| endpoint.to_string())
+    endpoint.to_string()
 }
 
-/// Map a streaming-config endpoint URL to the sibling storage-routing
+/// Map a physical-plan endpoint URL to the sibling storage-routing
 /// endpoint by rewriting the trailing path component. URLs that don't
-/// end with `/api/v1/streaming-config` (or `/api/v1/streaming_config` —
+/// end with `/api/v1/physical-plan` (or `/api/v1/physical_plan` —
 /// either spelling is supported) pass through unchanged so tests can
 /// inject a mock-server URL directly.
 #[cfg(test)]
 fn derive_storage_routing_url(endpoint: &str) -> String {
-    const STREAMING_PATH_DASH: &str = "/api/v1/streaming-config";
-    const STREAMING_PATH_UNDERSCORE: &str = "/api/v1/streaming_config";
+    const STREAMING_PATH_DASH: &str = "/api/v1/physical-plan";
+    const STREAMING_PATH_UNDERSCORE: &str = "/api/v1/physical_plan";
     const ROUTING_PATH: &str = "/api/v1/storage_routing";
     if let Some(stripped) = endpoint.strip_suffix(STREAMING_PATH_DASH) {
         return format!("{stripped}{ROUTING_PATH}");
@@ -429,7 +413,7 @@ mod tests {
     async fn start_mock_backend(sink: SharedSink, status: axum::http::StatusCode) -> String {
         let app = Router::new()
             .route(
-                "/api/v1/streaming-config",
+                "/api/v1/physical-plan",
                 post(
                     move |State(sink): State<SharedSink>, body: axum::body::Bytes| async move {
                         let yaml = String::from_utf8_lossy(&body).to_string();
@@ -445,7 +429,7 @@ mod tests {
             axum::serve(listener, app).await.unwrap();
         });
         tokio::time::sleep(Duration::from_millis(50)).await;
-        format!("http://{addr}/api/v1/streaming-config")
+        format!("http://{addr}/api/v1/physical-plan")
     }
 
     #[tokio::test]
@@ -481,7 +465,7 @@ mod tests {
     #[tokio::test]
     async fn push_or_log_swallows_errors() {
         // Point at an unreachable port so the request fails fast.
-        let client = BackendClient::new("http://127.0.0.1:1/api/v1/streaming-config");
+        let client = BackendClient::new("http://127.0.0.1:1/api/v1/physical-plan");
         // Must not panic or propagate — fire-and-forget semantics.
         push_or_log(&client, "cpu_usage", "content".to_string()).await;
     }
@@ -522,15 +506,15 @@ mod tests {
 
     /// storage-routing-URL derivation rewrites the path
     /// component when the configured endpoint ends in
-    /// `/api/v1/streaming-config`, leaving everything else untouched.
+    /// `/api/v1/physical-plan`, leaving everything else untouched.
     #[test]
     fn storage_routing_url_rewrites_streaming_path() {
         assert_eq!(
-            derive_storage_routing_url("http://backend:8088/api/v1/streaming-config"),
+            derive_storage_routing_url("http://backend:8088/api/v1/physical-plan"),
             "http://backend:8088/api/v1/storage_routing"
         );
         assert_eq!(
-            derive_storage_routing_url("http://backend:8088/api/v1/streaming_config"),
+            derive_storage_routing_url("http://backend:8088/api/v1/physical_plan"),
             "http://backend:8088/api/v1/storage_routing"
         );
     }
@@ -578,7 +562,7 @@ mod tests {
         let server = tokio::spawn(async move {
             axum::serve(listener, app).await.unwrap();
         });
-        let client = BackendClient::new(format!("http://{addr}/api/v1/streaming-config"));
+        let client = BackendClient::new(format!("http://{addr}/api/v1/physical-plan"));
         client
             .post_catalog_plan_typed(&publication, None, &[])
             .await
@@ -630,7 +614,7 @@ mod tests {
         tokio::time::sleep(Duration::from_millis(50)).await;
         // Return the streaming-config URL — the client will rewrite
         // the path before issuing the POST.
-        format!("http://{addr}/api/v1/streaming-config")
+        format!("http://{addr}/api/v1/physical-plan")
     }
 
     #[tokio::test]
@@ -702,7 +686,7 @@ mod tests {
     #[tokio::test]
     async fn typed_connection_refused_is_transient() {
         // Port 1 on loopback is reserved and refuses connections.
-        let client = BackendClient::new("http://127.0.0.1:1/api/v1/streaming-config");
+        let client = BackendClient::new("http://127.0.0.1:1/api/v1/physical-plan");
         let err = client
             .post_streaming_config_json_typed("{}".to_string())
             .await

control_plane/src/emit/monitor.rs

@@ -9,7 +9,7 @@
 //!      by `asapedgeprocessor.ThresholdConfig`). The edge derives the agg_id
 //!      from the metric name itself, so this block carries no agg_id.
 //!   2. [`streaming_config_monitor_entry`] — the `monitors[]` JSON object for the
-//!      backend `StreamingConfig` (consumed by `asap_types::MonitorSpec`), where
+//!      backend `InstalledPrecomputePlan` (consumed by `asap_types::MonitorSpec`), where
 //!      `agg_id` IS carried and MUST equal [`agg_id_for_metric`].
 //!
 //! The functions are pure so they can be unit-tested and called from whichever
@@ -88,7 +88,7 @@ pub fn edge_threshold_block(intent: &MonitorIntent) -> Value {
     Value::Mapping(m)
 }
 
-/// Render the backend `StreamingConfig.monitors[]` JSON entry for this intent,
+/// Render the backend `InstalledPrecomputePlan.monitors[]` JSON entry for this intent,
 /// stamping the cross-language `agg_id`.
 pub fn streaming_config_monitor_entry(intent: &MonitorIntent) -> serde_json::Value {
     serde_json::json!({

control_plane/src/main.rs

@@ -41,7 +41,7 @@ struct AppState {
     /// input identity; incoming telemetry cannot supply its own descriptors.
     active_summary_catalog:
         Arc<tokio::sync::Mutex<Option<Arc<asap_types::summary_catalog::SummaryCatalog>>>>,
-    /// Shared client for posting streaming configs from HTTP planning and replanning.
+    /// Shared client for publishing physical plans from HTTP planning and replanning.
     /// `None` when `CONTROLLER_BACKEND_ENDPOINT` is unset; pushes are then skipped.
     backend_client: Option<Arc<backend_client::BackendClient>>,
 }
@@ -69,13 +69,13 @@ async fn main() {
         backend_endpoint.as_ref().map(|endpoint| {
             info!(
                 endpoint = %endpoint,
-                "ASAPQuery-backend StreamingConfig push enabled"
+                "ASAPQuery-backend InstalledPrecomputePlan push enabled"
             );
             Arc::new(backend_client::BackendClient::new(endpoint.clone()))
         });
     if backend_client_shared.is_none() {
         info!(
-            "ASAPQuery-backend StreamingConfig push disabled \
+            "ASAPQuery-backend physical-plan publication disabled \
              (set CONTROLLER_BACKEND_ENDPOINT=<url> to enable)"
         );
     }
@@ -996,9 +996,9 @@ mod api_tests {
     #[test]
     fn app_state_backend_client_some_when_constructed_with_url() {
         let (state, _router) =
-            test_app_with_backend(Some("http://127.0.0.1:1/api/v1/streaming-config".into()));
+            test_app_with_backend(Some("http://127.0.0.1:1/api/v1/physical-plan".into()));
         let bc = state.backend_client.expect("backend_client must be Some");
-        assert_eq!(bc.endpoint(), "http://127.0.0.1:1/api/v1/streaming-config");
+        assert_eq!(bc.endpoint(), "http://127.0.0.1:1/api/v1/physical-plan");
     }
 
     // ── POST /api/v1/plan ─────────────────────────────────────────────────────

control_plane/src/physical/compiler.rs

@@ -7012,15 +7012,22 @@ pub(crate) mod tests {
             &query_plan,
         )
         .unwrap();
-        // V1 has one stored output per definition; arbitrary output IDs are
-        // rejected before writer/reader agreement is considered.
+        // Output identity is independent of definition identity, but changing
+        // only the writer must still invalidate every unchanged reader binding.
         let mut rebound_writer = bundle.precompute_plan.clone();
         rebound_writer.schemas[0]
             .stored_output_reference
             .stored_output_id = asap_types::sds::StoredOutputId(123);
-        assert!(rebound_writer
+        rebound_writer
             .validate_against_catalog(&bundle.summary_catalog)
-            .is_err());
+            .unwrap();
+        assert!(
+            asap_types::plan_publication::validate_stored_output_references(
+                &rebound_writer,
+                &query_plan,
+            )
+            .is_err()
+        );
     }
 
     #[test]

control_plane/src/physical/executable_binding.rs

@@ -2,6 +2,48 @@
 
 pub use asap_types::executable_plan::*;
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum OperatorExecution {
+    Maintenance,
+    Query,
+}
+
+/// Keep the backend's ownership decision exhaustive over Planner's physical IR.
+/// Adding a payload variant upstream must therefore choose an executor here.
+fn operator_execution(
+    node: &planner_types::post_asap::ExecutableDagNode,
+) -> Result<OperatorExecution, String> {
+    use planner_types::post_asap::{ExecutableOperatorPayload as Payload, ExecutionTiming};
+
+    let declared = match &node.payload {
+        Payload::Binary { timing, .. }
+        | Payload::Value { timing, .. }
+        | Payload::SummaryMerge { timing } => *timing,
+        Payload::MembershipFilter { .. } | Payload::SummaryEstimate { .. } => {
+            ExecutionTiming::QueryTime
+        }
+        Payload::SummaryAgg { .. }
+        | Payload::SummaryJoin { .. }
+        | Payload::SummarySubtract
+        | Payload::SummaryDelete { .. } => ExecutionTiming::IngestionTime,
+        // These operators can be placed on either side of the stored-state
+        // boundary. Planner's validated output state is authoritative.
+        Payload::Fallback { .. } | Payload::RelationalJoin { .. } => node.output_state.timing,
+    };
+    if declared != node.output_state.timing {
+        return Err(format!(
+            "post-ASAP node {:?} has operator timing {} but output state {}",
+            node.id,
+            declared.as_str(),
+            node.output_state
+        ));
+    }
+    Ok(match declared {
+        ExecutionTiming::IngestionTime => OperatorExecution::Maintenance,
+        ExecutionTiming::QueryTime => OperatorExecution::Query,
+    })
+}
+
 /// Assign backend phases to a selected semantic DAG without changing its nodes.
 pub fn install_selected_dag(
     query_id: String,
@@ -15,12 +57,21 @@ pub fn install_selected_dag(
     let mut nodes = std::collections::BTreeMap::new();
     let mut precompute_sinks = Vec::new();
     for node in &dag.nodes {
+        if let planner_types::post_asap::ExecutableOperatorPayload::SummaryAgg {
+            family,
+            input,
+            grouping,
+            ..
+        } = &node.payload
+        {
+            asap_physical_operators::capability::validate_summary_kernel(family, input, grouping)
+                .map_err(|reason| format!("post-ASAP node {:?}: {reason}", node.id))?;
+        }
+        let execution = operator_execution(node)?;
         let binding = if let Some(summary_definition) = materialization(node.id) {
             precompute_sinks.push(node.id);
             BackendNodeBinding::Materialization { summary_definition }
-        } else if node.output_state.timing
-            == planner_types::post_asap::ExecutionTiming::IngestionTime
-        {
+        } else if execution == OperatorExecution::Maintenance {
             BackendNodeBinding::MaintenanceInput
         } else {
             query_node(node.id).map_or(BackendNodeBinding::QueryInput, |query_node| {

control_plane/src/physical/post_asap/deployment_expr.rs

@@ -59,7 +59,7 @@ pub enum PhysicalExpr {
     /// Phase ε.1 Mode 2: no sketch processor at the edge — raw OTLP
     /// forwards to the backend, which builds the sketch at ingest. The
     /// `family` and `params` are the sketch the backend will build, so
-    /// the backend's `StreamingConfig` `aggregation_input` is `raw` for
+    /// the backend's `InstalledPrecomputePlan` `aggregation_input` is `raw` for
     /// this metric.
     RawAtEdgeSketchAtBackend {
         /// Sketch family the backend will build at ingest.

control_plane/src/physical/post_asap/tests.rs

@@ -579,7 +579,7 @@ fn pipeline_l1_to_l4(query: &str, accuracy: AccuracyTarget) -> PhysicalExpr {
 }
 
 /// `quantile_over_time.yaml` — the asap-planner-rs `quantile_over_time`
-/// fixture maps to a KLL or DDSketch StreamingConfig row. The control plane
+/// fixture maps to a KLL or DDSketch InstalledPrecomputePlan row. The control plane
 /// path: L1 PromQL parse → L3 `Aggregate{Quantile{0.99}}` over `Window` →
 /// L4 bind picks Kll (default) or DDSketch. Either is functionally
 /// equivalent — both are quantile sketches.

control_plane/src/workload.rs

@@ -501,7 +501,7 @@ impl WorkloadRegistry {
     /// Inject (or replace, keyed by `metric_name`) a runtime workload entry.
     /// Used by the autonomous-allocation apply path to register a synthesized
     /// monitor so the next replan/repost emits it into the backend
-    /// `StreamingConfig` (the coordinator then derives the ε-floor `p`). Shared
+    /// `InstalledPrecomputePlan` (the coordinator then derives the ε-floor `p`). Shared
     /// across registry clones via the `Arc<RwLock<…>>` overlay.
     pub fn insert_runtime(&self, entry: WorkloadEntry) {
         let mut rt = self

crates/asap_types/src/aggregation_config.rs

@@ -334,7 +334,7 @@ impl PrecomputeMaterialization {
     }
 
     /// `PolicyFingerprint::as_u64()` — the u64-form handle used by the
-    /// policy-fingerprint-keyed call sites (e.g. `StreamingConfig`'s
+    /// policy-fingerprint-keyed call sites (e.g. `InstalledPrecomputePlan`'s
     /// `HashMap<u64, PrecomputeMaterialization>` keys). **Always** equal to
     /// `self.policy_fingerprint().as_u64()`. The value is content-
     /// addressed identity, NOT a controller-allocated counter id.

crates/asap_types/src/executable_plan.rs

@@ -369,7 +369,7 @@ mod tests {
         fn send_sync<T: Send + Sync>() {}
         send_sync::<InstalledPostAsapDag>();
         let wire = serde_json::json!({
-            "schema_version": 1, "query_id": "q", "nodes": [], "edges": [], "root": 0
+            "schema_version": 2, "query_id": "q", "nodes": [], "edges": [], "root": 0
         });
         let document: OwnedPostAsapDag = serde_json::from_value(wire.clone()).unwrap();
         assert_eq!(serde_json::to_value(document).unwrap(), wire);