test: add production process E2E tests and manual runner

control_plane/build.rs

@@ -12,7 +12,9 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     // with `sketch-bench/sketch-runtime/proto/feedback.proto`.
     tonic_build::configure()
         .build_server(true)
-        .build_client(false)
+        // Keep the generated client available for black-box process E2E tests
+        // and for downstream agents that share this crate's wire contract.
+        .build_client(true)
         .compile_protos(&["proto/feedback.proto"], &["proto/"])?;
     Ok(())
 }

control_plane/src/emit/backend_push.rs

@@ -646,7 +646,7 @@ mod tests {
     }
 
     /// Happy path on the first attempt: zero retries, Ok outcome,
-    /// attempts == 1. This protects the smoke-test invariant that the
+    /// attempts == 1. This protects the retry unit-test invariant that the
     /// fire-and-forget happy path is unchanged when the backend is up
     /// before the controller's first POST.
     #[tokio::test(start_paused = true)]

control_plane/src/emit/mod.rs

@@ -1256,7 +1256,7 @@ mod runtime_tests {
     // is threaded through the pre-pop QuerySpec → analyzer →
     // QueryWorkload.group_by_labels → collect_metric_to_grouping_labels
     // → the emitter's keep_keys list. Without this round-trip the
-    // smoke test's sid catalog stays empty-per-zone.
+    // end-to-end test's sid catalog stays empty-per-zone.
     #[test]
     fn workload_entry_grouping_labels_round_trip_through_emit_to_keep_keys() {
         let yaml = r#"

control_plane/src/emit/stage_config.rs

@@ -5301,7 +5301,7 @@ mod tests {
     // wire-attr tuple, minting one sid per unique tuple — defeating
     // the streaming-config contract and ballooning the schema endpoint
     // per-metric sid count (51 for `http_requests_total_latency_ms`
-    // in the smoke test).
+    // in the end-to-end acceptance test).
     //
     // We chose OTTL `transform` over `attributes/keep` because the
     // attributes processor has NO native allowlist action (only

control_plane/src/main.rs

@@ -439,7 +439,7 @@ async fn main() {
     // Loop through every `(metric, role)` pair the workload-registry
     // pre-pop loop populated and POST the typed cumulative
     // streaming-config + storage-routing to the backend. Without this,
-    // queries that never trigger `POST /api/v1/plan` (the smoke
+    // queries that never trigger `POST /api/v1/plan` (the acceptance
     // harness, bootstrap deployments) hit the data plane's static
     // startup config (DDSketch only) and `sum by (zone) (…)` returns
     // `ExactAgg(Sum) capability not satisfied`.
@@ -3460,7 +3460,7 @@ mod api_tests {
     //
     // The (metric, role) cache key is exercised in tandem by the live
     // mvp-workload.yaml pre-pop loop (the workload registry lists 3
-    // entries for `http_requests_total`) → see the MVP smoke-test
+    // entries for `http_requests_total`) → see the MVP acceptance-test
     // pipeline. This in-process test exercises the cumulative-merge
     // plumbing in isolation against the same emit path used by both
     // the pre-pop loop and per-request replans.

control_plane/src/query_parser/mod.rs

@@ -451,7 +451,7 @@ mod tests {
 
     const ACC: AccuracyTarget = AccuracyTarget::Epsilon(0.01);
 
-    // Smoke tests for the parse entry point.
+    // Focused unit contracts for the PromQL parse entry point.
 
     #[test]
     fn promql_dispatched_correctly() {

control_plane/src/workload.rs

@@ -1058,107 +1058,4 @@ mod tests {
         let distinct: std::collections::HashSet<_> = roles.iter().copied().collect();
         assert_eq!(distinct.len(), 2, "Sum + Count = 2 distinct roles");
     }
-
-    #[test]
-    fn live_mvp_workload_yaml_assigns_three_roles_to_http_requests_total() {
-        // B2 full restructure regression: the live
-        // `deploy/configs/mvp-workload.yaml` carries THREE entries for
-        // `http_requests_total` (entries 2/3/4 — sum/sum+rate/count).
-        // Pre-B2 these collapsed onto one workload-store key and
-        // dropped two of the three plans, so the `sum by (zone)`
-        // query returned `ExactAgg(Sum) capability not satisfied`.
-        //
-        // The fix is the `(metric, role)` key + the per-entry role
-        // classification via `derive_agg_role`. This test pins that
-        // the three entries classify to two distinct roles (`Sum` for
-        // entries 2 + 3, `Count` for entry 4) — the multi-row keyed
-        // store can persist them all simultaneously.
-        use std::path::PathBuf;
-        let mut path = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
-        path.pop();
-        path.push("deploy/configs/mvp-workload.yaml");
-        if !path.exists() {
-            // Live file not in this checkout; skip silently (matches
-            // the sibling override test below).
-            return;
-        }
-        let registry = WorkloadRegistry::load(path.to_str().unwrap());
-        let http_requests_entries: Vec<&WorkloadEntry> = registry
-            .entries()
-            .iter()
-            .filter(|e| e.metric_name == "http_requests_total")
-            .collect();
-        assert!(
-            http_requests_entries.len() >= 3,
-            "mvp-workload.yaml is expected to carry ≥3 entries for \
-             http_requests_total (sum, sum(rate), count); got {}",
-            http_requests_entries.len()
-        );
-        let roles: Vec<AggRole> = http_requests_entries
-            .iter()
-            .map(|e| derive_agg_role(e))
-            .collect();
-        // At least one Sum and at least one Count among the entries.
-        assert!(
-            roles.contains(&AggRole::Sum),
-            "expected ≥1 Sum-role entry among http_requests_total in \
-             mvp-workload.yaml; got {roles:?}"
-        );
-        assert!(
-            roles.contains(&AggRole::Count),
-            "expected ≥1 Count-role entry among http_requests_total in \
-             mvp-workload.yaml; got {roles:?}"
-        );
-    }
-
-    #[test]
-    fn live_mvp_workload_yaml_loads_with_overrides() {
-        // Smoke-test the live deploy file. Confirms entries 5–8 carry
-        // their `sketch_family_override` after deserialization (the
-        // original stitching gap was this field being silently ignored
-        // by `serde`'s unknown-field default behaviour).
-        use std::path::PathBuf;
-        let mut path = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
-        path.pop();
-        path.push("deploy/configs/mvp-workload.yaml");
-        if !path.exists() {
-            // Live file not in this checkout; skip silently.
-            return;
-        }
-        let registry = WorkloadRegistry::load(path.to_str().unwrap());
-        let by_name: std::collections::HashMap<&str, &WorkloadEntry> = registry
-            .entries()
-            .iter()
-            .map(|e| (e.metric_name.as_str(), e))
-            .collect();
-
-        assert_eq!(
-            by_name
-                .get("request_size_bytes")
-                .and_then(|e| e.sketch_family_override.clone()),
-            Some(SketchType::KLL),
-            "request_size_bytes must carry KLL override",
-        );
-        assert_eq!(
-            by_name
-                .get("unique_users_per_min")
-                .and_then(|e| e.sketch_family_override.clone()),
-            Some(SketchType::HLL),
-            "unique_users_per_min must carry HLL override",
-        );
-        assert_eq!(
-            by_name
-                .get("top_endpoint_qps")
-                .and_then(|e| e.sketch_family_override.clone()),
-            Some(SketchType::CountSketch),
-            "top_endpoint_qps must carry CountSketch override",
-        );
-        assert_eq!(
-            by_name
-                .get("endpoint_request_freq")
-                .and_then(|e| e.sketch_family_override.clone()),
-            Some(SketchType::CountMinSketch),
-            "endpoint_request_freq must carry CountMinSketch override",
-        );
-    }
 }

control_plane/tests/component_process_e2e.rs

@@ -0,0 +1,218 @@
+//! Black-box component E2E for the production control-plane binary.
+//!
+//! A simulated collector connects to the production OpAMP WebSocket, a real
+//! workload is planned through the public HTTP API, and the emitted collector
+//! YAML is received over OpAMP. The same child process then accepts a runtime
+//! sample over its production gRPC service and exposes the accepted record in
+//! Prometheus metrics.
+
+use futures_util::StreamExt;
+use prost::Message;
+use std::net::TcpListener;
+use std::process::{Child, Command, Stdio};
+use std::time::Duration;
+use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+
+use control_plane::opamp::opamp_proto::ServerToAgent;
+use control_plane::runtime_samples::feedback::{
+    runtime_samples_client::RuntimeSamplesClient, PushBatch, RuntimeRecord,
+};
+
+struct ChildGuard(Child);
+
+impl Drop for ChildGuard {
+    fn drop(&mut self) {
+        let _ = self.0.kill();
+        let _ = self.0.wait();
+    }
+}
+
+fn unused_addr() -> String {
+    let listener = TcpListener::bind("127.0.0.1:0").expect("reserve loopback port");
+    let addr = listener.local_addr().expect("read loopback address");
+    drop(listener);
+    addr.to_string()
+}
+
+async fn wait_until_ready(client: &reqwest::Client, url: &str, child: &mut Child) {
+    for _ in 0..100 {
+        if let Some(status) = child.try_wait().expect("inspect control-plane process") {
+            panic!("control-plane exited before readiness: {status}");
+        }
+        if client
+            .get(url)
+            .send()
+            .await
+            .is_ok_and(|response| response.status().is_success())
+        {
+            return;
+        }
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    panic!("control-plane did not become ready at {url}");
+}
+
+async fn connect_agent(
+    address: &str,
+    child: &mut Child,
+) -> tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>> {
+    for _ in 0..100 {
+        if let Some(status) = child.try_wait().expect("inspect control-plane process") {
+            panic!("control-plane exited before OpAMP connection: {status}");
+        }
+        let mut request = format!("ws://{address}/v1/opamp")
+            .into_client_request()
+            .expect("build OpAMP request");
+        request
+            .headers_mut()
+            .insert("X-Agent-ID", "process-e2e-agent".parse().unwrap());
+        request
+            .headers_mut()
+            .insert("X-Agent-Role", "agent".parse().unwrap());
+        if let Ok((stream, _)) = tokio_tungstenite::connect_async(request).await {
+            return stream;
+        }
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    panic!("production OpAMP listener did not accept a collector connection");
+}
+
+async fn push_runtime_sample(address: &str) {
+    let endpoint = format!("http://{address}");
+    let mut connected = None;
+    for _ in 0..100 {
+        match RuntimeSamplesClient::connect(endpoint.clone()).await {
+            Ok(client) => {
+                connected = Some(client);
+                break;
+            }
+            Err(_) => tokio::time::sleep(Duration::from_millis(50)).await,
+        }
+    }
+    let mut client = connected
+        .unwrap_or_else(|| panic!("could not connect to production runtime service {endpoint}"));
+    let response = client
+        .push(PushBatch {
+            records: vec![RuntimeRecord {
+                source: "process-e2e-agent".into(),
+                sketch: "ddsketch".into(),
+                impl_name: "rust".into(),
+                schema_version: 1,
+                payload_json: serde_json::json!({
+                    "schema_version": 1,
+                    "bench": {"throughput_items_per_sec": {"mean": 42000.0}}
+                })
+                .to_string(),
+            }],
+        })
+        .await
+        .expect("push runtime sample to production gRPC service")
+        .into_inner();
+    assert_eq!(response.accepted, 1);
+}
+
+#[tokio::test]
+async fn production_binary_plans_pushes_opamp_config_and_ingests_feedback() {
+    let api_addr = unused_addr();
+    let opamp_addr = unused_addr();
+    let grpc_addr = unused_addr();
+
+    let child = Command::new(env!("CARGO_BIN_EXE_control_plane"))
+        .current_dir(env!("CARGO_MANIFEST_DIR"))
+        .env("CONTROLLER_ADDR", &api_addr)
+        .env("CONTROLLER_OPAMP_ADDR", &opamp_addr)
+        .env("CONTROLLER_GRPC_ADDR", &grpc_addr)
+        .env(
+            "CONTROLLER_WORKLOADS",
+            "/definitely/missing/e2e-workloads.yaml",
+        )
+        .env("CONTROLLER_SKETCH_DEFAULTS", "sketch_params_default.yml")
+        .stdout(Stdio::null())
+        .stderr(Stdio::null())
+        .spawn()
+        .expect("start production control-plane binary");
+    let mut child = ChildGuard(child);
+
+    let client = reqwest::Client::new();
+    let base = format!("http://{api_addr}");
+    wait_until_ready(&client, &format!("{base}/api/v1/cost-model"), &mut child.0).await;
+    let mut agent = connect_agent(&opamp_addr, &mut child.0).await;
+
+    let response = client
+        .post(format!("{base}/api/v1/plan"))
+        .json(&serde_json::json!({
+            "metric_name": "component_process_e2e_latency_ms",
+            "aggregations": ["quantile"],
+            "time_window": "1m",
+            "accuracy_sla": 0.01
+        }))
+        .send()
+        .await
+        .expect("POST workload to production control plane");
+    assert!(
+        response.status().is_success(),
+        "plan status: {}",
+        response.status()
+    );
+    let body: serde_json::Value = response.json().await.expect("decode plan response");
+    assert_eq!(body["metric"], "component_process_e2e_latency_ms");
+    assert!(body["sketch_type"].as_str().is_some());
+    assert!(body["valid_until"].as_str().is_some());
+    assert_eq!(body["agents_notified"], 1);
+
+    let frame = tokio::time::timeout(Duration::from_secs(5), agent.next())
+        .await
+        .expect("timed out waiting for OpAMP configuration")
+        .expect("OpAMP connection closed")
+        .expect("read OpAMP frame");
+    let data = match frame {
+        tokio_tungstenite::tungstenite::Message::Binary(data) => data,
+        other => panic!("expected binary OpAMP frame, got {other:?}"),
+    };
+    let payload = if data.first() == Some(&0) {
+        &data[1..]
+    } else {
+        &data
+    };
+    let message = ServerToAgent::decode(payload).expect("decode OpAMP ServerToAgent");
+    let config = message
+        .remote_config
+        .and_then(|remote| remote.config)
+        .expect("OpAMP response contains remote config");
+    let yaml = String::from_utf8(
+        config
+            .config_map
+            .get("")
+            .expect("default OpAMP config file")
+            .body
+            .clone(),
+    )
+    .expect("collector config is UTF-8 YAML");
+    let planned_sketch = body["sketch_type"]
+        .as_str()
+        .expect("plan contains sketch type")
+        .to_ascii_lowercase();
+    assert!(
+        yaml.to_ascii_lowercase().contains(&planned_sketch)
+            && yaml.contains("otlp/backend")
+            && yaml.contains("service:"),
+        "OpAMP YAML does not implement the selected {planned_sketch} plan:\n{yaml}"
+    );
+
+    push_runtime_sample(&grpc_addr).await;
+    for _ in 0..50 {
+        let metrics = client
+            .get(format!("{base}/metrics"))
+            .send()
+            .await
+            .expect("GET production metrics")
+            .text()
+            .await
+            .expect("read production metrics");
+        if metrics.contains("asap_runtime_samples_records_stored_total 1") {
+            return;
+        }
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    panic!("runtime sample was accepted but never surfaced in /metrics");
+}