feat: enforce plan-scoped sampling, GOS, delta, and adaptation

control_plane/src/backend_client.rs

@@ -370,6 +370,7 @@ impl BackendClient {
         backend_plan: Vec<u8>,
         query_plan: &crate::query_plan::QueryPlan,
         storage_routing: Option<serde_json::Value>,
+        adaptation_evidence: &[crate::physical::compiler::RuntimeAdaptationEvidence],
     ) -> std::result::Result<(), BackendPostError> {
         let url = derive_physical_plan_url(&self.endpoint);
         let response = self
@@ -381,6 +382,7 @@ impl BackendClient {
             "backend_plan": backend_plan,
             "query_plan": query_plan,
             "storage_routing": storage_routing,
+            "adaptation_evidence": adaptation_evidence,
             }))
             .send()
             .await

control_plane/src/emit/backend_push.rs

@@ -255,6 +255,7 @@ async fn push_documents_coupled(
     let transmission_plan = match crate::physical::compiler::TransmissionPlan::build(
         precompute_plan.envelope.clone(),
         precompute_plan,
+        &Default::default(),
     ) {
         Ok(plan) => plan,
         Err(error) => {
@@ -271,6 +272,7 @@ async fn push_documents_coupled(
                 plan_bytes.clone(),
                 &query_plan,
                 Some(routing.clone()),
+                &[],
             )
             .await
         {

control_plane/src/main.rs

@@ -583,6 +583,8 @@ struct PhysicalPlanQueryRequest {
     accuracy: types_v2::AccuracyTarget,
     lifecycle: physical::compiler::LifecyclePlanningInput,
     window_implementations: Vec<physical::compiler::WindowImplementationCandidate>,
+    #[serde(default)]
+    runtime_policy: physical::compiler::RuntimeRulePolicy,
 }
 
 #[derive(Debug, Deserialize)]
@@ -593,6 +595,8 @@ struct CompileAndPublishPhysicalPlanRequest {
     capability_snapshot_id: String,
     #[serde(default)]
     evidence: HashMap<String, physical::compiler::TopKMembershipEvidence>,
+    #[serde(default)]
+    runtime_adaptation_evidence: Vec<physical::compiler::RuntimeAdaptationEvidence>,
     planner_revision: String,
     max_evidence_age_ms: u64,
     plan_version: u64,
@@ -624,10 +628,11 @@ async fn handle_compile_and_publish_physical_plan(
     State(st): State<AppState>,
     Json(request): Json<CompileAndPublishPhysicalPlanRequest>,
 ) -> impl IntoResponse {
-    let (bundle, collector_ids, apply_timeout) = match compile_physical_plan_request(request) {
-        Ok(compiled) => compiled,
-        Err(response) => return response.into_response(),
-    };
+    let (bundle, collector_ids, apply_timeout, adaptation_evidence) =
+        match compile_physical_plan_request(request) {
+            Ok(compiled) => compiled,
+            Err(response) => return response.into_response(),
+        };
 
     let Some(backend) = st.backend_client.as_ref() else {
         return (
@@ -654,6 +659,7 @@ async fn handle_compile_and_publish_physical_plan(
             bundle.backend_plan.encode_to_vec(),
             &bundle.query_plan,
             None,
+            &adaptation_evidence,
         )
         .await
     {
@@ -714,7 +720,15 @@ async fn handle_compile_and_publish_physical_plan(
 // Only the Send-safe compiled bundle crosses an await point.
 fn compile_physical_plan_request(
     request: CompileAndPublishPhysicalPlanRequest,
-) -> Result<(physical::compiler::PhysicalPlan, Vec<String>, Duration), (StatusCode, String)> {
+) -> Result<
+    (
+        physical::compiler::PhysicalPlan,
+        Vec<String>,
+        Duration,
+        Vec<physical::compiler::RuntimeAdaptationEvidence>,
+    ),
+    (StatusCode, String),
+> {
     if request.queries.is_empty() || request.collector_ids.is_empty() {
         return Err((
             StatusCode::UNPROCESSABLE_ENTITY,
@@ -780,6 +794,7 @@ fn compile_physical_plan_request(
             accuracy: query.accuracy,
             lifecycle: query.lifecycle,
             window_implementations: query.window_implementations,
+            runtime_policy: query.runtime_policy,
         });
     }
 
@@ -804,7 +819,12 @@ fn compile_physical_plan_request(
         Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string())),
     };
     let apply_timeout = Duration::from_millis(request.apply_timeout_ms);
-    Ok((bundle, request.collector_ids, apply_timeout))
+    Ok((
+        bundle,
+        request.collector_ids,
+        apply_timeout,
+        request.runtime_adaptation_evidence,
+    ))
 }
 
 // ── Handlers ──────────────────────────────────────────────────────────────────

control_plane/src/opamp/mod.rs

@@ -54,6 +54,7 @@ pub const PLAN_STATUS_MESSAGE: &str = "plan_status";
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(rename_all = "SCREAMING_SNAKE_CASE")]
 pub enum CollectorPlanStatusKind {
+    Staged,
     Applied,
     Failed,
 }
@@ -283,8 +284,8 @@ impl OpampServer {
     }
 
     /// Publish all per-target physical plans and require an exact semantic
-    /// APPLIED report from every Collector. Config hashes are deliberately not
-    /// accepted as plan activation evidence.
+    /// STAGED report from every Collector. Activation is synchronized later at
+    /// the envelope timestamp; config hashes are not accepted as evidence.
     pub async fn publish_collector_plans(
         &self,
         plans: &[crate::physical::compiler::CollectorPlan],
@@ -327,7 +328,7 @@ impl OpampServer {
                     plan_id: plan.envelope.plan_id,
                     plan_version: plan.envelope.plan_version,
                 })?;
-            if report.status != CollectorPlanStatusKind::Applied {
+            if report.status != CollectorPlanStatusKind::Staged {
                 return Err(CollectorPlanPublishError::Rejected {
                     collector_id: plan.collector_id.clone(),
                     plan_id: report.plan_id,
@@ -1044,6 +1045,7 @@ mod tests {
                 emit_every_ms: 60_000,
                 full_checkpoint_every_ms: None,
                 destination_ref: "asapquery-backend".into(),
+                runtime_policy: crate::physical::compiler::RuntimeRulePolicy::default(),
             }],
         }
     }
@@ -1065,7 +1067,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn typed_plan_publication_waits_for_capability_and_exact_applied_status() {
+    async fn typed_plan_publication_waits_for_capability_and_exact_staged_status() {
         let (srv, addr) = start_server().await;
         let mut agent_ws = connect_ws_client(addr, "edge-a", "agent").await;
 
@@ -1105,7 +1107,7 @@ mod tests {
         let status = serde_json::to_vec(&CollectorPlanStatus {
             plan_id: 42,
             plan_version: 1,
-            status: CollectorPlanStatusKind::Applied,
+            status: CollectorPlanStatusKind::Staged,
             error: None,
         })
         .unwrap();
@@ -1126,7 +1128,7 @@ mod tests {
         assert_eq!(reports.len(), 1);
         assert_eq!(reports[0].plan_id, 42);
         assert_eq!(reports[0].plan_version, 1);
-        assert_eq!(reports[0].status, CollectorPlanStatusKind::Applied);
+        assert_eq!(reports[0].status, CollectorPlanStatusKind::Staged);
     }
 
     #[tokio::test]