test(monitor): live multi-edge coupling over gRPC (differentiated sample_p)

data_plane/tests/monitor_grpc.rs

@@ -156,3 +156,87 @@ async fn stays_quiet_below_tau() {
         "edge should have received at least the initial grant"
     );
 }
+
+fn register_edge(edge_id: &str) -> EdgeToCoord {
+    EdgeToCoord {
+        msg: Some(edge_to_coord::Msg::Reg(MonitorRegister {
+            edge_id: edge_id.into(),
+            agg_id: 1,
+            key: vec![],
+            epoch_window_ms: 60_000,
+            window_start_ms: 0,
+        })),
+    }
+}
+
+fn report_edge(edge_id: &str, value: f64, seq: u64, rate: f64) -> EdgeToCoord {
+    EdgeToCoord {
+        msg: Some(edge_to_coord::Msg::Report(MonitorReport {
+            edge_id: edge_id.into(),
+            agg_id: 1,
+            key: vec![],
+            window_start_ms: 0,
+            local_value: value,
+            round: 0,
+            seq,
+            rate,
+        })),
+    }
+}
+
+/// Live multi-edge coupling: two edges open real gRPC streams, report SKEWED
+/// rates, and the coordinator ships back differentiated `SlackGrant.sample_p`
+/// over the wire — the hot edge sampled harder (smaller p) than the quiet one,
+/// both at/above the ε-derived coupling floor. τ is large so only grants flow.
+#[tokio::test]
+async fn two_edges_get_differentiated_sample_p_over_grpc() {
+    use asap_otel_proto::monitor::v1::coord_to_edge;
+
+    async fn connect_edge(url: &str) -> (mpsc::Sender<EdgeToCoord>, Arc<Mutex<Vec<f64>>>) {
+        let mut client = MonitorServiceClient::connect(url.to_string()).await.unwrap();
+        let (tx, rx) = mpsc::channel(32);
+        let mut inbound = client
+            .monitor(ReceiverStream::new(rx))
+            .await
+            .unwrap()
+            .into_inner();
+        let ps = Arc::new(Mutex::new(Vec::<f64>::new()));
+        let psc = ps.clone();
+        tokio::spawn(async move {
+            while let Some(Ok(env)) = inbound.next().await {
+                if let Some(coord_to_edge::Msg::Grant(g)) = env.msg {
+                    psc.lock().unwrap().push(g.sample_p);
+                }
+            }
+        });
+        (tx, ps)
+    }
+
+    let (url, _alerts) = start_server(vec![sum_cfg(1_000_000.0)]).await;
+    let (tx_hot, p_hot) = connect_edge(&url).await;
+    let (tx_quiet, p_quiet) = connect_edge(&url).await;
+
+    tx_hot.send(register_edge("e1")).await.unwrap();
+    tx_quiet.send(register_edge("e2")).await.unwrap();
+    tokio::time::sleep(Duration::from_millis(60)).await;
+
+    // e1 hot (100k items/win), e2 quiet (1k/win); small local values stay far
+    // below τ so only grants (carrying sample_p) flow, never an alert.
+    for seq in 1..=4 {
+        tx_hot.send(report_edge("e1", 1.0, seq, 100_000.0)).await.unwrap();
+        tx_quiet.send(report_edge("e2", 1.0, seq, 1_000.0)).await.unwrap();
+        tokio::time::sleep(Duration::from_millis(40)).await;
+    }
+    tokio::time::sleep(Duration::from_millis(250)).await;
+
+    let hot = *p_hot.lock().unwrap().last().expect("hot edge received a grant");
+    let quiet = *p_quiet.lock().unwrap().last().expect("quiet edge received a grant");
+    println!("LIVE coupling over gRPC: hot(e1,rate=100k) sample_p={hot} | quiet(e2,rate=1k) sample_p={quiet}");
+
+    assert!(
+        hot < quiet,
+        "hot edge sample_p {hot} should be < quiet edge sample_p {quiet}"
+    );
+    assert!(hot > 0.0 && hot <= 1.0, "hot p out of range: {hot}");
+    assert!(quiet > 0.0 && quiet <= 1.0, "quiet p out of range: {quiet}");
+}