feat(o11y): execute real inputs through Planner-selected backend plans

control_plane/examples/compile_workload_artifact.rs

@@ -0,0 +1,40 @@
+//! Control-plane entry point: cost-select a workload and emit its atomic install request.
+use control_plane::physical::compiler::BackendLocalPlanningSnapshot;
+use serde_json::json;
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let path = std::env::args()
+        .nth(1)
+        .ok_or("usage: compile_workload_artifact SNAPSHOT.json")?;
+    let snapshot: BackendLocalPlanningSnapshot = serde_json::from_slice(&std::fs::read(path)?)?;
+    if snapshot.snapshot_version != 2 {
+        return Err(
+            "execution evaluation requires version 2 complete workload cost evidence".into(),
+        );
+    }
+    let start = std::time::Instant::now();
+    let plan = snapshot.compile()?;
+    let elapsed = start.elapsed().as_nanos();
+    let comparison = plan
+        .cost_comparison
+        .ok_or("missing complete-plan comparison")?;
+    println!(
+        "{}",
+        serde_json::to_string_pretty(&json!({
+            "schema_version": 1,
+            "planning_elapsed_ns": elapsed,
+            "envelope": plan.envelope,
+            "cost_comparison": comparison,
+            "lifecycle_estimates": plan.lifecycle_estimates,
+            "install_request": {
+                "precompute_plan": plan.precompute_plan,
+                "transmission_plan": plan.transmission_plan,
+                "backend_plan": plan.backend_plan.encode_to_vec(),
+                "query_plan": plan.query_plan,
+                "storage_routing": null,
+                "adaptation_evidence": []
+            }
+        }))?
+    );
+    Ok(())
+}

data_plane/src/drivers/ingest/prometheus_remote_write.rs

@@ -100,6 +100,7 @@ struct ReceiverInner {
 
 #[derive(Default)]
 struct DedupState {
+    input_closed: bool,
     values: HashMap<(u64, u64, String, i64), DedupValue>,
     expiry: VecDeque<(Instant, u64, u64, String, i64)>,
 }
@@ -112,6 +113,8 @@ enum DedupValue {
 
 #[derive(Debug, thiserror::Error)]
 pub enum RemoteWriteError {
+    #[error("finite input has been closed by the drain barrier")]
+    InputClosed,
     #[error("compressed request exceeds {0} bytes")]
     CompressedTooLarge(usize),
     #[error("decompressed request exceeds {0} bytes")]
@@ -171,6 +174,15 @@ impl PrometheusRemoteWriteReceiver {
         self.inner.stats.clone()
     }
 
+    /// Permanently seal this finite source before queuing worker barriers.
+    pub async fn drain(&self) -> Result<(), String> {
+        {
+            let mut state = self.inner.dedup.lock().map_err(|e| e.to_string())?;
+            state.input_closed = true;
+        }
+        self.inner.ingest.router.drain().await
+    }
+
     /// Decode, validate, deduplicate, and enqueue one whole v1 request.
     /// All validation and all queue reservations complete before any message
     /// becomes visible to a worker.
@@ -235,6 +247,9 @@ impl PrometheusRemoteWriteReceiver {
             .dedup
             .lock()
             .expect("remote write dedup poisoned");
+        if dedup.input_closed {
+            return Err(RemoteWriteError::InputClosed);
+        }
         dedup.evict_before(now.checked_sub(config.dedup_horizon).unwrap_or(now));
 
         // Validate conflicts both against committed history and inside this
@@ -664,6 +679,28 @@ mod tests {
         })
     }
 
+    // Closing finite input prevents writes racing behind the completion barrier.
+    #[tokio::test]
+    async fn finite_input_drain_seals_receiver_and_propagates_worker_failure() {
+        let (receiver, mut worker) = configured_receiver();
+        receiver.accept(&one_sample(1.0)).unwrap();
+        let handle = receiver.clone();
+        let drain = tokio::spawn(async move { handle.drain().await });
+        assert!(matches!(
+            worker.recv().await.unwrap(),
+            WorkerMessage::GroupSamples { .. }
+        ));
+        let WorkerMessage::Drain(reply) = worker.recv().await.unwrap() else {
+            panic!("expected barrier")
+        };
+        assert!(matches!(
+            receiver.accept(&one_sample(1.0)),
+            Err(RemoteWriteError::InputClosed)
+        ));
+        reply.send(Err("sink write failed".into())).unwrap();
+        assert_eq!(drain.await.unwrap().unwrap_err(), "sink write failed");
+    }
+
     #[test]
     fn rejects_writes_without_an_active_physical_plan() {
         let (sender, _worker) = mpsc::channel(1);

data_plane/src/drivers/query/fallback/mod.rs

@@ -19,7 +19,7 @@ pub enum FallbackResponse {
 
 impl IntoResponse for FallbackResponse {
     fn into_response(self) -> Response {
-        match self {
+        let mut response = match self {
             FallbackResponse::Json(value) => Json(value).into_response(),
             FallbackResponse::Text(text) => {
                 // Return plain text with appropriate content type
@@ -32,7 +32,28 @@ impl IntoResponse for FallbackResponse {
                 )
                     .into_response()
             }
-        }
+        };
+        response.headers_mut().insert(
+            "x-asap-execution",
+            axum::http::HeaderValue::from_static("exact_fallback"),
+        );
+        response
+    }
+}
+
+#[cfg(test)]
+mod execution_attribution_tests {
+    use super::*;
+
+    // An HTTP fallback must be distinguishable from warm execution, even if
+    // its upstream response carries misleading or absent source annotations.
+    #[test]
+    fn forwarded_response_has_backend_owned_execution_marker() {
+        let response = FallbackResponse::Json(serde_json::json!({
+            "status": "success", "infos": ["data_source: asap_query"]
+        }))
+        .into_response();
+        assert_eq!(response.headers()["x-asap-execution"], "exact_fallback");
     }
 }
 

data_plane/src/drivers/query/servers/http.rs

@@ -470,6 +470,7 @@ impl HttpServer {
                     .layer(DefaultBodyLimit::max(request_body_limit)),
             )
             .route("/api/v1/store/metrics", get(handle_store_metrics))
+            .route("/api/v1/precompute/drain", post(handle_precompute_drain))
             .route(
                 "/api/v1/streaming-config",
                 get(handle_get_streaming_config).post(handle_post_streaming_config),
@@ -5577,6 +5578,8 @@ async fn handle_prometheus_remote_write(
 
     match receiver.accept(&body) {
         Ok(()) => StatusCode::NO_CONTENT.into_response(),
+        Err(crate::drivers::ingest::prometheus_remote_write::RemoteWriteError::InputClosed) =>
+            (StatusCode::CONFLICT, "finite input is closed").into_response(),
         Err(
             crate::drivers::ingest::prometheus_remote_write::RemoteWriteError::CompressedTooLarge(
                 _,
@@ -5636,6 +5639,29 @@ async fn handle_health(State(state): State<AppState>) -> axum::response::Respons
     (StatusCode::OK, "ok").into_response()
 }
 
+/// Explicit end-of-input; this seals Remote Write for the lifetime of the process.
+async fn handle_precompute_drain(State(state): State<AppState>) -> Response {
+    let Some(receiver) = state.remote_write.as_ref() else {
+        return (StatusCode::NOT_FOUND, "Remote Write is disabled").into_response();
+    };
+    match receiver.drain().await {
+        Ok(()) => (
+            StatusCode::OK,
+            axum::Json(serde_json::json!({
+                "status": "success", "input_closed": true, "complete": true
+            })),
+        )
+            .into_response(),
+        Err(error) => (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            axum::Json(serde_json::json!({
+                "status": "error", "input_closed": true, "complete": false, "error": error
+            })),
+        )
+            .into_response(),
+    }
+}
+
 /// Return list of metrics currently in the store.
 async fn handle_store_metrics(State(state): State<AppState>) -> axum::response::Response {
     use axum::http::StatusCode;

data_plane/src/precompute_engine/output_sink.rs

@@ -100,11 +100,9 @@ impl SketchStoreSink {
         }
     }
 
-    /// Best-effort write to `SketchStore` for one PrecomputedOutput.
-    /// Logs and skips on missing agg_config or other transient
-    /// inconsistencies — a SketchStore miss is recoverable in
-    /// practice because the control plane will re-emit the agg_config
-    /// on its next reconcile pass.
+    /// Write one PrecomputedOutput; false reports a rejected output to the caller.
+    /// Missing configuration or incompatible state must surface as failure:
+    /// a finite-input completion barrier cannot acknowledge dropped outputs.
     ///
     /// PR-6 follow-up: resolves the source `AggregationConfig` via
     /// `PolicyRegistry::get(output.policy_fp)`. The legacy
@@ -160,8 +158,18 @@ impl OutputSink for SketchStoreSink {
             return Ok(());
         }
         let _span = debug_span!("sketch_index_insert", batch_size = outputs.len()).entered();
+        let mut failed = 0;
         for (output, accumulator) in &outputs {
-            self.append_to_index(output, accumulator.as_ref());
+            if !self.append_to_index(output, accumulator.as_ref()) {
+                failed += 1;
+            }
+        }
+        if failed > 0 {
+            return Err(format!(
+                "SketchStore rejected {failed} of {} completed outputs",
+                outputs.len()
+            )
+            .into());
         }
         Ok(())
     }
@@ -375,9 +383,9 @@ mod tests {
     }
 
     #[test]
-    fn sketch_index_sink_skips_unknown_agg_id_gracefully() {
+    fn sketch_index_sink_reports_unknown_policy_as_failure() {
         // Streaming config does NOT contain agg_id=99 — the sink
-        // skips it (warn log) rather than panicking.
+        // reports a recoverable error rather than acknowledging a lost write.
         let streaming = StreamingConfig::new(HashMap::new());
         let hot_reload = HotReloadStreamingConfig::new(streaming.clone());
         let sketch_index = Arc::new(SketchStore::new());
@@ -389,12 +397,13 @@ mod tests {
 
         let output = PrecomputedOutput::new(1000, 2000, None, asap_types::PolicyFingerprint(99));
         let acc: Box<dyn AggregateCore> = Box::new(SumAccumulator::with_sum(1.0));
-        sink.emit_batch(vec![(output, acc)]).expect("emit ok");
+        sink.emit_batch(vec![(output, acc)])
+            .expect_err("unpersisted output must not be acknowledged");
         assert_eq!(sketch_index.instance_count(), 0);
     }
 
     /// CQ-6 — a registry-miss (policy_fp not in the running streaming
-    /// config) is a silent write-skip; with an `IngestObservability`
+    /// config) reports a failed write; with an `IngestObservability`
     /// handle wired in, the `dropped_policy_miss` counter must tick.
     #[test]
     fn sink_increments_policy_miss_counter_on_registry_miss() {
@@ -412,7 +421,8 @@ mod tests {
         // policy_fp=42 is absent from the empty registry → registry miss.
         let output = PrecomputedOutput::new(1000, 2000, None, asap_types::PolicyFingerprint(42));
         let acc: Box<dyn AggregateCore> = Box::new(SumAccumulator::with_sum(1.0));
-        sink.emit_batch(vec![(output, acc)]).expect("emit ok");
+        sink.emit_batch(vec![(output, acc)])
+            .expect_err("unpersisted output must not be acknowledged");
 
         assert_eq!(
             sketch_index.instance_count(),
@@ -430,7 +440,8 @@ mod tests {
         // miss — it must not bump the counter.
         let unset = PrecomputedOutput::new(1000, 2000, None, asap_types::PolicyFingerprint::UNSET);
         let acc2: Box<dyn AggregateCore> = Box::new(SumAccumulator::with_sum(1.0));
-        sink.emit_batch(vec![(unset, acc2)]).expect("emit ok");
+        sink.emit_batch(vec![(unset, acc2)])
+            .expect_err("unpersisted output must not be acknowledged");
         assert_eq!(
             obs.dropped_policy_miss
                 .load(std::sync::atomic::Ordering::Relaxed),

data_plane/src/precompute_engine/series_router.rs

@@ -85,6 +85,8 @@ pub enum WorkerMessage {
     },
     /// Signal the worker to flush/check idle windows.
     Flush,
+    /// Finite-input barrier: acknowledge only after queued input and trailing panes reach the sink.
+    Drain(tokio::sync::oneshot::Sender<Result<(), String>>),
     /// Graceful shutdown.
     Shutdown,
 }
@@ -126,6 +128,7 @@ impl fmt::Debug for WorkerMessage {
                 .field("accumulator_type", &accumulator.type_name())
                 .finish(),
             Self::Flush => f.write_str("Flush"),
+            Self::Drain(_) => f.write_str("Drain"),
             Self::Shutdown => f.write_str("Shutdown"),
         }
     }
@@ -205,7 +208,7 @@ impl SeriesRouter {
                 WorkerMessage::GroupSamples { sid, .. }
                 | WorkerMessage::AccumulatorInput { sid, .. } => self.worker_for_sid(*sid),
                 WorkerMessage::RawSamples { series_key, .. } => self.worker_for(series_key),
-                WorkerMessage::Flush | WorkerMessage::Shutdown => 0,
+                WorkerMessage::Flush | WorkerMessage::Drain(_) | WorkerMessage::Shutdown => 0,
             };
             let permit = self.senders[worker_idx]
                 .clone()
@@ -244,6 +247,23 @@ impl SeriesRouter {
         Ok(())
     }
 
+    /// Caller must stop new input before invoking this finite-source barrier.
+    pub async fn drain(&self) -> Result<(), String> {
+        let mut replies = Vec::new();
+        for sender in &self.senders {
+            let (tx, rx) = tokio::sync::oneshot::channel();
+            sender
+                .send(WorkerMessage::Drain(tx))
+                .await
+                .map_err(|e| e.to_string())?;
+            replies.push(rx);
+        }
+        for reply in replies {
+            reply.await.map_err(|e| e.to_string())??;
+        }
+        Ok(())
+    }
+
     /// Determine which worker handles a given sid bucket.
     ///
     /// Hashes the sid alone — the legacy `(agg_id, group_key)` tuple folded