mvp v6 phase C: AgentRole::Gateway + BackendClient on AppState

controller/src/backend_client.rs

@@ -94,6 +94,42 @@ impl BackendClient {
             ))
         }
     }
+
+    /// Phase C (MVP v6) variant of [`Self::push_streaming_config`]
+    /// that POSTs `application/json`. The typed L5
+    /// `emit_backend_config_json` emitter produces a `serde_json::Value`
+    /// rather than a YAML document, and the ASAPQuery-backend's
+    /// `/api/v1/streaming-config` endpoint accepts both content types
+    /// (PR #297 / Phase B documents the JSON shape). Same 2xx-or-error
+    /// contract as the YAML variant; same fire-and-forget semantics
+    /// at the call site.
+    pub async fn post_streaming_config_json(&self, json: String) -> Result<()> {
+        debug!(
+            endpoint = %self.endpoint,
+            json_bytes = json.len(),
+            "posting streaming-config JSON to ASAPQuery-backend"
+        );
+        let resp = self
+            .http
+            .post(&self.endpoint)
+            .header("content-type", "application/json")
+            .body(json)
+            .send()
+            .await
+            .context("failed to POST streaming-config JSON to backend")?;
+
+        let status = resp.status();
+        if status.is_success() {
+            Ok(())
+        } else {
+            let body = resp.text().await.unwrap_or_default();
+            Err(anyhow::anyhow!(
+                "backend returned {} for streaming-config JSON POST: {}",
+                status,
+                body
+            ))
+        }
+    }
 }
 
 /// Fire-and-forget convenience helper used by the replanner. Logs
@@ -187,4 +223,39 @@ mod tests {
         // Must not panic or propagate — fire-and-forget semantics.
         push_or_log(&client, "cpu_usage", "content".to_string()).await;
     }
+
+    /// Phase C: the JSON variant POSTs the body verbatim, returns
+    /// `Ok(())` on a 2xx, and surfaces non-2xx as `Err`. Mock backend
+    /// captures the body so we can verify it round-trips.
+    #[tokio::test]
+    async fn json_post_round_trips_body() {
+        let sink = SharedSink(StdArc::new(Mutex::new(Vec::new())));
+        let url = start_mock_backend(sink.clone(), axum::http::StatusCode::OK).await;
+
+        let client = BackendClient::new(url);
+        let json = r#"{"aggregations":[{"aggregationId":7,"metric":"latency"}]}"#.to_string();
+        client
+            .post_streaming_config_json(json.clone())
+            .await
+            .expect("json post ok");
+
+        let received = sink.0.lock().unwrap();
+        assert_eq!(received.len(), 1);
+        assert_eq!(received[0], json);
+    }
+
+    /// Phase C: non-2xx from the backend surfaces as an error so the
+    /// caller (handle_plan) can log + move on.
+    #[tokio::test]
+    async fn json_post_non_2xx_is_error() {
+        let sink = SharedSink(StdArc::new(Mutex::new(Vec::new())));
+        let url =
+            start_mock_backend(sink.clone(), axum::http::StatusCode::BAD_REQUEST).await;
+
+        let client = BackendClient::new(url);
+        let result = client.post_streaming_config_json("{}".to_string()).await;
+        assert!(result.is_err(), "expected error on 400, got {result:?}");
+        let msg = result.unwrap_err().to_string();
+        assert!(msg.contains("400"), "error msg should mention 400: {msg}");
+    }
 }

controller/src/main.rs

@@ -67,6 +67,16 @@ struct AppState {
     /// agents' `sketch-runtime::PushExporter`. Read by decision
     /// loops in the replanner.
     runtime_samples:   Arc<runtime_samples::RuntimeSamplesStore>,
+    /// Phase C (MVP v6): shared `BackendClient` for posting
+    /// `StreamingConfig` JSON / YAML to the ASAPQuery-backend's
+    /// `POST /api/v1/streaming-config` endpoint. Phase B had this
+    /// only on the `Replanner`, so the typed L5 stage_split path in
+    /// `handle_plan` could only `info!`-log the backend JSON it
+    /// emitted. Sharing via `Arc` lets `AppState` and `Replanner`
+    /// both push without owning a duplicate client. `None` when
+    /// `CONTROLLER_BACKEND_ENDPOINT` is unset, matching the
+    /// pre-existing fire-and-forget contract.
+    backend_client:    Option<Arc<backend_client::BackendClient>>,
 }
 
 // ── Entry point ───────────────────────────────────────────────────────────────
@@ -247,6 +257,28 @@ async fn main() {
         }
     }
 
+    // ── Phase C: shared BackendClient ─────────────────────────────────────────
+    // Built once at startup; shared between Replanner (existing path —
+    // pushes the StreamingConfig YAML on every successful replan) and
+    // AppState (Phase C — pushes the typed L5 backend JSON emitted by
+    // `emit_backend_config_json` from `handle_plan`). `None` when
+    // `CONTROLLER_BACKEND_ENDPOINT` is unset preserves the
+    // fire-and-forget "skip silently" contract from Phase B.
+    let backend_client_shared: Option<Arc<backend_client::BackendClient>> =
+        backend_endpoint.as_ref().map(|endpoint| {
+            info!(
+                endpoint = %endpoint,
+                "ASAPQuery-backend StreamingConfig push enabled"
+            );
+            Arc::new(backend_client::BackendClient::new(endpoint.clone()))
+        });
+    if backend_client_shared.is_none() {
+        info!(
+            "ASAPQuery-backend StreamingConfig push disabled \
+             (set CONTROLLER_BACKEND_ENDPOINT=<url> to enable)"
+        );
+    }
+
     // ── Replanner — closes the SP-8 feedback loop ─────────────────────────────
     let replanner = {
         let mut r = Replanner::new(
@@ -257,19 +289,8 @@ async fn main() {
             Arc::clone(&scraper),
             opamp_ep.clone(),
         );
-        if let Some(endpoint) = backend_endpoint.as_ref() {
-            info!(
-                endpoint = %endpoint,
-                "ASAPQuery-backend StreamingConfig push enabled"
-            );
-            r = r.with_backend_client(Arc::new(backend_client::BackendClient::new(
-                endpoint.clone(),
-            )));
-        } else {
-            info!(
-                "ASAPQuery-backend StreamingConfig push disabled \
-                 (set CONTROLLER_BACKEND_ENDPOINT=<url> to enable)"
-            );
+        if let Some(client) = backend_client_shared.as_ref() {
+            r = r.with_backend_client(Arc::clone(client));
         }
         Arc::new(r)
     };
@@ -297,6 +318,7 @@ async fn main() {
         opamp_endpoint:    opamp_ep,
         workload_registry: Arc::clone(&workload_registry),
         runtime_samples:   Arc::clone(&runtime_samples_store),
+        backend_client:    backend_client_shared,
     };
 
     // ── Background tasks ──────────────────────────────────────────────────────
@@ -462,28 +484,65 @@ async fn handle_plan(
                             }
                         }
                         crate::stage_split::StageConfig::Gateway(gw) => {
-                            // No `AgentRole::Gateway` exists today
-                            // (Phase C adds it). Log the YAML so the
-                            // demo overlay can pick it up via stdout
-                            // until Phase C wires the role.
+                            // Phase C: AgentRole::Gateway is now wired
+                            // through the OpAMP role-routing path, so
+                            // the gateway YAML is pushed to gateway-role
+                            // collectors the same way the edge YAML is
+                            // pushed to agent-role collectors above.
                             match config::emit_gateway_yaml(&gw, &st.opamp_endpoint) {
-                                Ok(yaml) => info!(
-                                    stage = "gateway", bytes = yaml.len(),
-                                    yaml = %yaml,
-                                    "[USE_TYPED_STAGE_SPLIT] gateway YAML emitted (push deferred to Phase C)"
-                                ),
+                                Ok(yaml) => {
+                                    let hash = short_hash(&yaml);
+                                    info!(
+                                        stage = "gateway", bytes = yaml.len(),
+                                        "[USE_TYPED_STAGE_SPLIT] pushing typed gateway YAML"
+                                    );
+                                    st.opamp.push_to_role(
+                                        AgentRole::Gateway,
+                                        RemoteConfig { config_hash: hash, yaml },
+                                    ).await;
+                                }
                                 Err(e) => warn!(error = %e, "emit_gateway_yaml failed"),
                             }
                         }
                         crate::stage_split::StageConfig::Backend(be) => {
+                            // Phase C: post the typed L5 streaming-config
+                            // JSON to ASAPQuery-backend via the shared
+                            // BackendClient when configured. Without a
+                            // configured endpoint this still no-ops
+                            // silently — same fire-and-forget contract
+                            // as the existing Replanner path.
                             match config::emit_backend_config_json(&be) {
-                                Ok(json_doc) => info!(
-                                    stage = "backend",
-                                    aggregations = be.aggregations.len(),
-                                    readouts = be.readouts.len(),
-                                    json = %json_doc,
-                                    "[USE_TYPED_STAGE_SPLIT] backend streaming-config JSON emitted (push deferred to Phase C)"
-                                ),
+                                Ok(json_doc) => {
+                                    info!(
+                                        stage = "backend",
+                                        aggregations = be.aggregations.len(),
+                                        readouts = be.readouts.len(),
+                                        "[USE_TYPED_STAGE_SPLIT] posting typed backend JSON"
+                                    );
+                                    if let Some(client) = st.backend_client.as_ref() {
+                                        let body = json_doc.to_string();
+                                        match client.post_streaming_config_json(body).await {
+                                            Ok(()) => info!(
+                                                stage = "backend",
+                                                endpoint = %client.endpoint(),
+                                                "[USE_TYPED_STAGE_SPLIT] typed backend JSON push succeeded"
+                                            ),
+                                            Err(e) => warn!(
+                                                stage = "backend",
+                                                endpoint = %client.endpoint(),
+                                                error = %e,
+                                                "[USE_TYPED_STAGE_SPLIT] typed backend JSON push failed; \
+                                                 next replan cycle will retry"
+                                            ),
+                                        }
+                                    } else {
+                                        info!(
+                                            stage = "backend",
+                                            "[USE_TYPED_STAGE_SPLIT] no backend client configured; \
+                                             skipping JSON push (set CONTROLLER_BACKEND_ENDPOINT to enable)"
+                                        );
+                                    }
+                                }
                                 Err(e) => warn!(error = %e, "emit_backend_config_json failed"),
                             }
                             // Mention stage_id so `match` arms aren't
@@ -794,6 +853,16 @@ fn short_hash(s: &str) -> String {
 /// No background tasks are started; OpAMP/scraper hold no real connections.
 #[cfg(test)]
 fn test_app() -> (AppState, axum::Router) {
+    test_app_with_backend(None)
+}
+
+/// Phase C test helper: build an `AppState` whose `backend_client` is
+/// optionally set to a real `BackendClient` pointed at a mock URL. The
+/// `None` arm is the legacy path used by every existing test;
+/// `Some(url)` is the new entry point for Phase C tests that exercise
+/// the typed L5 backend-JSON push.
+#[cfg(test)]
+fn test_app_with_backend(backend_url: Option<String>) -> (AppState, axum::Router) {
     let online_store   = init_online_store();
     let plan_store     = Arc::new(PlanStore::new());
     let workload_store = Arc::new(WorkloadStore::new());
@@ -812,6 +881,8 @@ fn test_app() -> (AppState, axum::Router) {
         Arc::clone(&scraper),
         "ws://ctrl:4320/v1/opamp",
     ));
+    let backend_client = backend_url
+        .map(|u| Arc::new(backend_client::BackendClient::new(u)));
     let state = AppState {
         analyzer:          Arc::new(Analyzer::new()),
         planner,
@@ -824,6 +895,7 @@ fn test_app() -> (AppState, axum::Router) {
         opamp_endpoint:    "ws://ctrl:4320/v1/opamp".into(),
         workload_registry: Arc::new(WorkloadRegistry::empty()),
         runtime_samples:   runtime_samples::RuntimeSamplesStore::new(64),
+        backend_client,
     };
     let router = axum::Router::new()
         .route("/api/v1/plan",                  axum::routing::post(handle_plan))
@@ -860,6 +932,31 @@ mod api_tests {
         })
     }
 
+    // ── Phase C: AppState.backend_client wiring ───────────────────────────────
+
+    /// Default-constructed AppState (no `CONTROLLER_BACKEND_ENDPOINT`)
+    /// must leave `backend_client` as `None` so the typed L5 backend
+    /// JSON push silently no-ops, matching the Phase B fire-and-forget
+    /// contract.
+    #[test]
+    fn app_state_backend_client_none_by_default() {
+        let (state, _router) = test_app();
+        assert!(state.backend_client.is_none(),
+            "backend_client should default to None when no endpoint is configured");
+    }
+
+    /// When constructed with a backend URL (the production path takes
+    /// it from `CONTROLLER_BACKEND_ENDPOINT`), the field is populated
+    /// and ready for the Phase C `handle_plan` push.
+    #[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()),
+        );
+        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");
+    }
+
     // ── POST /api/v1/plan ─────────────────────────────────────────────────────
 
     #[tokio::test]