diff --git a/control_plane/examples/compile_workload_artifact.rs b/control_plane/examples/compile_workload_artifact.rs
new file mode 100644
index 00000000..0cd9115d
--- /dev/null
+++ b/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(())
+}
diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
index b90cd013..1b8832fc 100644
--- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs
+++ b/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);
diff --git a/data_plane/src/drivers/query/fallback/mod.rs b/data_plane/src/drivers/query/fallback/mod.rs
index 5a15b454..2f4c50d5 100644
--- a/data_plane/src/drivers/query/fallback/mod.rs
+++ b/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");
     }
 }
 
diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs
index 717cb081..088cc0dd 100644
--- a/data_plane/src/drivers/query/servers/http.rs
+++ b/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;
diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs
index b139f741..0bd2908c 100644
--- a/data_plane/src/precompute_engine/output_sink.rs
+++ b/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),
diff --git a/data_plane/src/precompute_engine/series_router.rs b/data_plane/src/precompute_engine/series_router.rs
index f66b2073..475620b8 100644
--- a/data_plane/src/precompute_engine/series_router.rs
+++ b/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
diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs
index 4cff1b52..35702b9c 100644
--- a/data_plane/src/precompute_engine/worker.rs
+++ b/data_plane/src/precompute_engine/worker.rs
@@ -202,6 +202,7 @@ impl Worker {
     pub async fn run(mut self) {
         info!("Worker {} started", self.id);
 
+        let mut processing_error: Option<String> = None;
         while let Some(msg) = self.receiver.recv().await {
             match msg {
                 WorkerMessage::GroupSamples {
@@ -223,6 +224,7 @@ impl Worker {
                     .entered();
                     if let Err(e) = self.process_group_samples(sid, policy_fp, &group_key, samples)
                     {
+                        processing_error = Some(e.to_string());
                         warn!(
                             "Worker {} error processing sid={} (policy_fp={}, group={}): {}",
                             self.id, sid, policy_fp, group_key, e
@@ -246,6 +248,7 @@ impl Worker {
                     )
                     .entered();
                     if let Err(e) = self.process_samples_raw(&series_key, samples) {
+                        processing_error = Some(e.to_string());
                         warn!("Worker {} raw error for {}: {}", self.id, series_key, e);
                     }
                     debug!(
@@ -278,6 +281,7 @@ impl Worker {
                         timestamp_ms,
                         accumulator,
                     ) {
+                        processing_error = Some(e.to_string());
                         warn!(
                             "Worker {} accumulator input error for sid={} (policy_fp={}, group={}): {}",
                             self.id, sid, policy_fp, group_key, e
@@ -290,6 +294,7 @@ impl Worker {
                 }
                 WorkerMessage::Flush => {
                     if let Err(e) = self.flush_all() {
+                        processing_error = Some(e.to_string());
                         warn!("Worker {} flush error: {}", self.id, e);
                     }
                     // Evict orphaned GroupStates whose agg_id has been
@@ -298,6 +303,13 @@ impl Worker {
                     // closed their windows); empty ones are freed.
                     self.evict_orphaned_groups();
                 }
+                WorkerMessage::Drain(reply) => {
+                    let result = self.force_close_all().map_err(|e| e.to_string());
+                    if let Err(error) = &result {
+                        processing_error = Some(error.clone());
+                    }
+                    let _ = reply.send(processing_error.clone().map_or(Ok(()), Err));
+                }
                 WorkerMessage::Shutdown => {
                     info!("Worker {} shutting down", self.id);
                     if let Err(e) = self.flush_all() {
@@ -3307,6 +3319,114 @@ aggregations:
         assert_ne!(accumulator.total_increase, independent_increases);
     }
 
+    // Acknowledgement proves FIFO input processing and trailing-window publication.
+    #[tokio::test]
+    async fn finite_input_drain_publishes_before_acknowledging() {
+        let config = make_agg_config(
+            1,
+            "cpu",
+            AggregationType::SingleSubpopulation,
+            "Sum",
+            10,
+            0,
+            vec![],
+        );
+        let sink = Arc::new(CapturingOutputSink::new());
+        let mut worker = make_worker(
+            HashMap::from([(1, config)]),
+            sink.clone(),
+            false,
+            0,
+            LateDataPolicy::Drop,
+        );
+        let (tx, rx) = tokio::sync::mpsc::channel(8);
+        worker.receiver = rx;
+        let task = tokio::spawn(worker.run());
+        tx.send(WorkerMessage::GroupSamples {
+            sid: 1,
+            policy_fp: PolicyFingerprint(1),
+            group_key: "".into(),
+            samples: group_samples("cpu", vec![(1000, 2.0), (2000, 3.0)]),
+            ingest_received_at: std::time::Instant::now(),
+        })
+        .await
+        .unwrap();
+        let (ack, completed) = tokio::sync::oneshot::channel();
+        tx.send(WorkerMessage::Drain(ack)).await.unwrap();
+        completed.await.unwrap().unwrap();
+        let outputs = sink.drain();
+        assert_eq!(outputs.len(), 1);
+        assert_eq!(
+            outputs[0]
+                .1
+                .as_any()
+                .downcast_ref::<SumAccumulator>()
+                .unwrap()
+                .sum,
+            5.0
+        );
+        let (ack, completed) = tokio::sync::oneshot::channel();
+        tx.send(WorkerMessage::Drain(ack)).await.unwrap();
+        completed.await.unwrap().unwrap();
+        assert_eq!(sink.len(), 0);
+        tx.send(WorkerMessage::Shutdown).await.unwrap();
+        task.await.unwrap();
+    }
+
+    // A sink failure remains visible on repeated barriers after panes were consumed.
+    #[tokio::test]
+    async fn finite_input_drain_does_not_hide_sink_failure_on_retry() {
+        struct FailedSink;
+        impl OutputSink for FailedSink {
+            fn emit_batch(
+                &self,
+                _: Vec<(PrecomputedOutput, Box<dyn AggregateCore>)>,
+            ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+                Err("deliberate sink failure".into())
+            }
+        }
+        let config = make_agg_config(
+            1,
+            "cpu",
+            AggregationType::SingleSubpopulation,
+            "Sum",
+            10,
+            0,
+            vec![],
+        );
+        let mut worker = make_worker(
+            HashMap::from([(1, config)]),
+            Arc::new(CapturingOutputSink::new()),
+            false,
+            0,
+            LateDataPolicy::Drop,
+        );
+        worker.output_sink = Arc::new(FailedSink);
+        let (tx, rx) = tokio::sync::mpsc::channel(8);
+        worker.receiver = rx;
+        let task = tokio::spawn(worker.run());
+        tx.send(WorkerMessage::GroupSamples {
+            sid: 1,
+            policy_fp: PolicyFingerprint(1),
+            group_key: "".into(),
+            samples: group_samples("cpu", vec![(1000, 2.0)]),
+            ingest_received_at: std::time::Instant::now(),
+        })
+        .await
+        .unwrap();
+        for _ in 0..2 {
+            let (ack, completed) = tokio::sync::oneshot::channel();
+            tx.send(WorkerMessage::Drain(ack)).await.unwrap();
+            assert!(completed
+                .await
+                .unwrap()
+                .unwrap_err()
+                .contains("deliberate sink failure"));
+        }
+        tx.send(WorkerMessage::Shutdown).await.unwrap();
+        task.await.unwrap();
+    }
+
     #[test]
     fn shutdown_force_close_emits_trailing_sample_window() {
         // 10s tumbling window; make_worker uses grace=0, isolating the
diff --git a/data_plane/tests/o11y_remote_write_wire.rs b/data_plane/tests/o11y_remote_write_wire.rs
new file mode 100644
index 00000000..88a146b5
--- /dev/null
+++ b/data_plane/tests/o11y_remote_write_wire.rs
@@ -0,0 +1,29 @@
+//! Validate the replay adapter's wire format with the production decoding libraries.
+use data_plane::drivers::ingest::prometheus_remote_write::WriteRequest;
+use prost::Message;
+
+// The Python adapter encodes x{job="a"} 2 at 1.234 OpenMetrics seconds.
+#[test]
+fn replay_wire_preserves_labels_value_and_millisecond_timestamp() {
+    let hex =
+        "29a00a270a0d0a085f5f6e616d655f5f1201780a080a036a6f62120161120c09000000000000004010d209";
+    let bytes: Vec<_> = (0..hex.len())
+        .step_by(2)
+        .map(|i| u8::from_str_radix(&hex[i..i + 2], 16).unwrap())
+        .collect();
+    let protobuf = snap::raw::Decoder::new().decompress_vec(&bytes).unwrap();
+    let write = WriteRequest::decode(protobuf.as_slice()).unwrap();
+    assert_eq!(write.timeseries.len(), 1);
+    let series = &write.timeseries[0];
+    assert_eq!(
+        series
+            .labels
+            .iter()
+            .map(|l| (l.name.as_str(), l.value.as_str()))
+            .collect::<Vec<_>>(),
+        vec![("__name__", "x"), ("job", "a")]
+    );
+    assert_eq!(series.samples.len(), 1);
+    assert_eq!(series.samples[0].value, 2.0);
+    assert_eq!(series.samples[0].timestamp, 1234);
+}
diff --git a/tools/o11y-execution/README.md b/tools/o11y-execution/README.md
new file mode 100644
index 00000000..deade70f
--- /dev/null
+++ b/tools/o11y-execution/README.md
@@ -0,0 +1,80 @@
+# Real-workload execution replay
+
+Developer acceptance tooling, stacked on #524 and the deployment/cost-selection
+foundation through #505. This is an executable harness, not a published benefit
+result. It invokes the **control-plane compiler**, which calls the pinned Planner,
+then boots the production data plane with that compiler's atomic install request.
+No family override, candidate index, or benchmark-selected winner is accepted.
+
+## Inputs and prerequisites
+
+- Original timestamped o11ybench OpenMetrics data. The strict numeric subset
+  preserves names, labels and finite values; seconds become Remote Write
+  milliseconds without rounding. Unsupported lines fail before any writes.
+- A JSON corpus with `upstream_revision` and `queries`, preserving every upstream
+  occurrence: `{id, query, eval_timestamp_ms}` (additional provenance is retained).
+  Export the pinned upstream task queries, not the historical 27-query fixture.
+  Provide generator revision, command and seed separately; a file hash cannot
+  establish that provenance.
+- A canonical **version 2** planning snapshot registering exactly the corpus's
+  unique queries and containing valid complete-workload provider cost evidence.
+  Collect evidence using `workload_cost_manifest`; keep calibration inputs and
+  evaluation inputs distinct. Do not reuse the demo snapshot's declared costs.
+  The harness does not invent registrations, cost quotes or unsupported bindings.
+- A dedicated, empty Prometheus instance with its Remote Write receiver enabled,
+  configured for the input's timestamp range. This process is also the fallback
+  service; do not point the runner at a production/shared instance.
+
+The harness does not provision Prometheus or calibrate the cost provider. These
+are required run inputs, not completed end-to-end acceptance evidence.
+
+## Run
+
+```sh
+cargo build --locked -p control_plane --example compile_workload_artifact
+cargo build --locked -p data_plane --bin data_plane
+python3 tools/o11y-execution/replay.py \
+  --metrics /path/metrics.txt --queries /path/queries.json \
+  --snapshot /path/o11y-costed-snapshot.json \
+  --compiler target/debug/examples/compile_workload_artifact \
+  --data-plane target/debug/data_plane \
+  --exact-url http://127.0.0.1:9090 --output /path/new-run
+python3 -m unittest discover -s tools/o11y-execution -v
+```
+
+Use an unused backend port (`--port`, default 18089). The output directory must
+not exist. The runner validates all samples, compiles the supplied workload,
+records candidates/selected plan/costs, starts its own backend, sends identical
+Remote Write batches to both services, and queries every corpus occurrence at its
+original evaluation time. It stops only the backend process it started. Partial
+ingest is journaled without automatic retries: restart with fresh services and a
+new output directory after investigating a failure.
+
+`planning.json` contains the envelope, cost comparison, lifecycle estimates and
+install request. `installed.json` records runtime status. `queries.json` preserves
+every response, occurrence ID, timing and `warm`, `exact_fallback` or `failed`
+classification. Forwarded responses carry a backend-owned `x-asap-execution`
+header; an unmarked success is not counted as warm or fallback. `ingestion.json`
+records accepted batches; acceptance does not prove worker completion.
+
+The settle interval is recorded, not a completion barrier. The first traversal is
+called `first_pass`, not “cold cache”; later traversals are `repeat`. All failures
+and fallback responses remain in the denominator. A completion file means the
+replay finished, **not** that accuracy or benefit acceptance passed. The next
+stacked PR adds matched exact queries and measurement/reporting.
+
+
+## Finite-input completion
+
+After all Remote Write batches are accepted, the runner calls
+`POST /api/v1/precompute/drain` and saves `drain.json`. The receiver seals input
+before worker barriers are queued. Each worker publishes its trailing panes,
+then acknowledges completion; prior processing or sink failures remain failures
+on repeated drains. Queries begin only after a successful completion response.
+Subsequent Remote Write requests return HTTP 409 for that process. Start a new
+backend process for another input generation.
+
+This endpoint is for a finite replay, not a live ingestion watermark. Closing a
+trailing pane does not by itself prove its coverage matches every query window;
+unsupported or incomplete readouts must still follow exact fallback. This change
+adds no local raw-query storage or hybrid operator execution.
diff --git a/tools/o11y-execution/replay.py b/tools/o11y-execution/replay.py
new file mode 100644
index 00000000..a9b58f14
--- /dev/null
+++ b/tools/o11y-execution/replay.py
@@ -0,0 +1,262 @@
+#!/usr/bin/env python3
+"""Replay supplied o11ybench inputs through the production backend. No winner overrides."""
+import argparse
+from decimal import Decimal
+import hashlib
+import json
+import math
+from pathlib import Path
+import re
+import socket
+import struct
+import subprocess
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+
+
+def classify(response, headers=None):
+    if response.get("status") != "success":
+        return "failed"
+    if (headers or {}).get("x-asap-execution") == "exact_fallback":
+        return "exact_fallback"
+    sources = {x for x in response.get("infos", []) if isinstance(x, str) and x.startswith("data_source:")}
+    if sources == {"data_source: asap_query"}:
+        return "warm"
+    if sources == {"data_source: exact_fallback"}:
+        return "exact_fallback"
+    return "failed"
+
+
+def validate_workload(snapshot, corpus):
+    rows = corpus["queries"]
+    if not corpus.get("upstream_revision") or not rows:
+        raise ValueError("a versioned, nonempty upstream query corpus is required")
+    ids = [row["id"] for row in rows]
+    if len(set(ids)) != len(ids):
+        raise ValueError("query occurrence IDs must be unique")
+    for row in rows:
+        if not isinstance(row["eval_timestamp_ms"], int) or row["eval_timestamp_ms"] < 0:
+            raise ValueError("each query needs its original nonnegative evaluation timestamp")
+    workload = snapshot["query_workload"]
+    if workload.get("query_batch"):
+        raise ValueError("this runner currently accepts repeating workload registrations only")
+    registered = {row["query"] for row in workload["repeating_queries"]}
+    if registered != {row["query"] for row in rows}:
+        raise ValueError("snapshot registrations must match the complete unique corpus exactly")
+    return rows
+
+
+_SAMPLE = re.compile(r'([a-zA-Z_:][a-zA-Z0-9_:]*)(?:\{(.*)\})?\s+(\S+)\s+(\d+(?:\.\d+)?)')
+_LABEL = re.compile(r'([a-zA-Z_][a-zA-Z0-9_]*)="((?:[^"\\]|\\[\\"n])*)"')
+
+
+def parse_samples(lines):
+    """Strict OpenMetrics subset: seconds converted losslessly to Remote Write milliseconds."""
+    rows, seen, latest = [], {}, -1
+    for number, line in enumerate(lines, 1):
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+        match = _SAMPLE.fullmatch(line)
+        if not match:
+            raise ValueError(f"unsupported sample at line {number}")
+        metric, raw_labels, value, timestamp = match.groups()
+        labels = {"__name__": metric}
+        rest = raw_labels or ""
+        while rest:
+            label = _LABEL.match(rest)
+            if not label or label[1] in labels:
+                raise ValueError(f"invalid or duplicate label at line {number}")
+            labels[label[1]] = re.sub(r'\\([\\"n])', lambda m: '\n' if m[1] == 'n' else m[1], label[2])
+            rest = rest[label.end():]
+            if rest:
+                if not rest.startswith(",") or len(rest) == 1:
+                    raise ValueError(f"invalid label separator at line {number}")
+                rest = rest[1:]
+        millis = Decimal(timestamp) * 1000
+        if millis != millis.to_integral_value():
+            raise ValueError(f"submillisecond timestamp at line {number}")
+        value, timestamp = float(value), int(millis)
+        key = tuple(sorted(labels.items()))
+        if not math.isfinite(value) or timestamp > 2**63 - 1 or timestamp < latest or timestamp <= seen.get(key, -1):
+            raise ValueError(f"nonfinite, duplicate, or out-of-order sample at line {number}")
+        latest, seen[key] = timestamp, timestamp
+        rows.append((labels, value, timestamp))
+    if not rows:
+        raise ValueError("empty dataset")
+    return rows
+
+
+def varint(value):
+    result = bytearray()
+    while value > 127:
+        result.append((value & 127) | 128)
+        value >>= 7
+    result.append(value)
+    return bytes(result)
+
+
+def field(number, payload):
+    return varint(number * 8 + 2) + varint(len(payload)) + payload
+
+
+def encode_write(rows):
+    """Remote Write v1 protobuf in an uncompressed-literal raw Snappy block."""
+    series = {}
+    for labels, value, timestamp in rows:
+        series.setdefault(tuple(sorted(labels.items())), []).append((value, timestamp))
+    wire = bytearray()
+    for labels, samples in series.items():
+        ts = b"".join(field(1, field(1, k.encode()) + field(2, v.encode())) for k, v in labels)
+        ts += b"".join(field(2, b"\x09" + struct.pack("<d", v) + b"\x10" + varint(t)) for v, t in samples)
+        wire.extend(field(1, ts))
+    length = len(wire)
+    if not length:
+        raise ValueError("cannot encode empty batch")
+    n = length - 1
+    if n < 60:
+        literal = bytes([n << 2])
+    else:
+        size = (n.bit_length() + 7) // 8
+        literal = bytes([(59 + size) << 2]) + n.to_bytes(size, "little")
+    return varint(length) + literal + wire
+
+
+def request(url, data=None, headers=None):
+    start = time.perf_counter_ns()
+    try:
+        with urllib.request.urlopen(urllib.request.Request(url, data=data, headers=headers or {}), timeout=60) as response:
+            body, status, received = response.read(), response.status, dict(response.headers.items())
+        try:
+            body = json.loads(body)
+        except (ValueError, UnicodeDecodeError):
+            body = {"raw": body.decode(errors="replace")}
+        return {"http_status": status, "response": body, "headers": {k.lower(): v for k, v in received.items()},
+                "elapsed_ns": time.perf_counter_ns() - start}
+    except (OSError, urllib.error.URLError) as error:
+        return {"http_status": getattr(error, "code", None), "response": {"status": "error", "error": str(error)},
+                "headers": {}, "elapsed_ns": time.perf_counter_ns() - start}
+
+
+def write_json(path, value):
+    path.write_text(json.dumps(value, indent=2, allow_nan=False) + "\n")
+
+
+def ingest(rows, endpoints, output):
+    batches = []
+    for offset in range(0, len(rows), 5000):
+        payload = encode_write(rows[offset:offset + 5000])
+        batch = {"offset": offset, "sample_count": len(rows[offset:offset + 5000]),
+                 "payload_sha256": hashlib.sha256(payload).hexdigest(), "endpoints": {}}
+        batches.append(batch)
+        for endpoint in endpoints:
+            # Persist intent first: a timeout can follow partial acceptance. Never retry silently.
+            batch["endpoints"][endpoint] = {"status": "attempting"}
+            write_json(output / "ingestion.json", batches)
+            result = request(endpoint.rstrip("/") + "/api/v1/write", payload, {
+                "Content-Type": "application/x-protobuf", "Content-Encoding": "snappy",
+                "X-Prometheus-Remote-Write-Version": "0.1.0"})
+            batch["endpoints"][endpoint] = result
+            write_json(output / "ingestion.json", batches)
+            if not result["http_status"] or not 200 <= result["http_status"] < 300:
+                raise RuntimeError(f"ingestion failed at batch {offset}; inspect partial acceptance before retry")
+
+
+def replay(queries, backend, output, repetitions):
+    rows = []
+    for repeat in range(repetitions):
+        for query in queries:
+            params = urllib.parse.urlencode({"query": query["query"], "time": f'{query["eval_timestamp_ms"] / 1000:.3f}'})
+            answer = request(backend.rstrip("/") + "/api/v1/query?" + params)
+            route = classify(answer["response"], answer["headers"])
+            if answer["http_status"] != 200:
+                route = "failed"
+            rows.append({**query, "repetition": repeat, "phase": "first_pass" if repeat == 0 else "repeat",
+                         "execution": route, **answer})
+            write_json(output / "queries.json", rows)
+    return rows
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--metrics", type=Path, required=True)
+    parser.add_argument("--queries", type=Path, required=True)
+    parser.add_argument("--snapshot", type=Path, required=True)
+    parser.add_argument("--compiler", type=Path, required=True)
+    parser.add_argument("--data-plane", type=Path, required=True)
+    parser.add_argument("--exact-url", required=True, help="dedicated empty Prometheus with Remote Write receiver enabled")
+    parser.add_argument("--port", type=int, default=18089)
+    parser.add_argument("--settle-seconds", type=float, default=0, help="deprecated; completion uses explicit finite-input drain")
+    parser.add_argument("--repetitions", type=int, default=2)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    if args.repetitions < 1 or not 0 <= args.settle_seconds <= 60:
+        parser.error("positive repetitions and settle-seconds in [0, 60] required")
+    corpus = json.loads(args.queries.read_text())
+    queries = validate_workload(json.loads(args.snapshot.read_text()), corpus)
+    samples = parse_samples(args.metrics.read_text().splitlines())
+    with socket.socket() as probe:
+        probe.bind(("127.0.0.1", args.port))
+    args.output.mkdir(parents=True, exist_ok=False)
+    provenance = {"schema_version": 1, "upstream_revision": corpus["upstream_revision"],
+                  "inputs": {str(p.resolve()): hashlib.sha256(p.read_bytes()).hexdigest()
+                             for p in [args.metrics, args.queries, args.snapshot, args.compiler, args.data_plane]},
+                  "samples": len(samples), "timestamp_min_ms": samples[0][2], "timestamp_max_ms": samples[-1][2],
+                  "query_occurrences": len(queries), "configuration": {k: str(v) for k, v in vars(args).items()},
+                  "limitations": ["generator provenance must be supplied separately", "first pass is not a guaranteed cold cache", "finite-input drain closes trailing panes and permanently seals Remote Write; no live-ingestion claim"]}
+    write_json(args.output / "run.json", provenance)
+    with (args.output / "planning.stderr").open("w") as log:
+        compiled = subprocess.run([str(args.compiler.resolve()), str(args.snapshot.resolve())], check=True,
+                                  stdout=subprocess.PIPE, stderr=log, text=True)
+    plan = json.loads(compiled.stdout)
+    write_json(args.output / "planning.json", plan)
+    artifact = args.output / "install.json"
+    write_json(artifact, plan["install_request"])
+    backend = f"http://127.0.0.1:{args.port}"
+    command = [str(args.data_plane.resolve()), "--profile", "asapquery", "--physical-plan", str(artifact.resolve()),
+               "--prometheus-server", args.exact_url, "--forward-unsupported-queries", "--http-port", str(args.port),
+               "--output-dir", str((args.output / "backend").resolve()), "--precompute-allowed-lateness-ms", "0",
+               "--precompute-flush-interval-ms", "25"]
+    write_json(args.output / "command.json", command)
+    with (args.output / "backend.log").open("w") as log:
+        child = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT)
+        try:
+            for _ in range(120):
+                if child.poll() is not None:
+                    raise RuntimeError("data plane exited; see backend.log")
+                if request(backend + "/api/v1/health")["http_status"] == 200:
+                    break
+                time.sleep(0.25)
+            else:
+                raise RuntimeError("data plane readiness timeout")
+            installed = request(backend + "/api/v1/physical-plan/status")
+            write_json(args.output / "installed.json", installed)
+            if installed["http_status"] != 200:
+                raise RuntimeError("could not read installed plan status")
+            envelope = plan["envelope"]
+            if not any(p["plan_id"] == envelope["plan_id"] and p["plan_version"] == envelope["plan_version"]
+                       and p["phase"] == "active" for p in installed["response"].get("plans", [])):
+                raise RuntimeError("runtime has not activated the selected plan generation")
+            ingest(samples, [args.exact_url, backend], args.output)
+            drained = request(backend + "/api/v1/precompute/drain", b"")
+            write_json(args.output / "drain.json", drained)
+            if drained["http_status"] != 200 or drained["response"].get("complete") is not True:
+                raise RuntimeError("finite-input materialization drain failed; see drain.json")
+            results = replay(queries, backend, args.output, args.repetitions)
+            write_json(args.output / "store.json", request(backend + "/api/v1/store/metrics"))
+            write_json(args.output / "completion.json", {"complete": True,
+                       "execution_counts": {k: sum(r["execution"] == k for r in results) for k in ["warm", "exact_fallback", "failed"]},
+                       "benefit_claim": None})
+        finally:
+            child.terminate()
+            try:
+                child.wait(timeout=10)
+            except subprocess.TimeoutExpired:
+                child.kill()
+                child.wait()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tools/o11y-execution/test_replay.py b/tools/o11y-execution/test_replay.py
new file mode 100644
index 00000000..3f51f6da
--- /dev/null
+++ b/tools/o11y-execution/test_replay.py
@@ -0,0 +1,45 @@
+"""Behavioral tests for the real-workload replay boundary (not speedup evidence)."""
+import unittest
+
+from replay import classify, validate_workload, encode_write, parse_samples
+
+
+class ReplayTests(unittest.TestCase):
+    def test_binding_or_unattributed_success_is_not_execution(self):
+        """Only an actual successful, attributed response establishes a route."""
+        self.assertEqual(classify({"status": "bound"}), "failed")
+        self.assertEqual(classify({"status": "success"}), "failed")
+        for source, expected in [("asap_query", "warm"), ("exact_fallback", "exact_fallback")]:
+            response = {"status": "success", "infos": [f"data_source: {source}"]}
+            self.assertEqual(classify(response), expected)
+            response["status"] = "error"
+            self.assertEqual(classify(response), "failed")
+        self.assertEqual(classify({"status": "success", "infos": ["data_source: asap_query"]},
+                                  {"x-asap-execution": "exact_fallback"}), "exact_fallback")
+
+    def test_corpus_occurrences_preserved_but_all_unique_queries_registered(self):
+        """The harness cannot quietly replace or omit upstream queries."""
+        corpus = {"upstream_revision": "abc", "queries": [
+            {"id": "a", "query": "up", "eval_timestamp_ms": 1000},
+            {"id": "b", "query": "up", "eval_timestamp_ms": 1000}]}
+        snapshot = {"query_workload": {"repeating_queries": [{"query": "up"}]}}
+        self.assertEqual(len(validate_workload(snapshot, corpus)), 2)
+        snapshot["query_workload"]["repeating_queries"] = [{"query": "synthetic"}]
+        with self.assertRaises(ValueError):
+            validate_workload(snapshot, corpus)
+
+    def test_timestamped_metrics_keep_labels_values_and_time(self):
+        """A data adapter must not shift timestamps or lose histogram labels."""
+        rows = parse_samples(['# TYPE x histogram', 'x_bucket{le="1",job="a"} 2 1.234'])
+        self.assertEqual(rows, [({"__name__": "x_bucket", "le": "1", "job": "a"}, 2.0, 1234)])
+        self.assertTrue(encode_write(rows))
+
+    def test_bad_input_fails_before_any_ingest(self):
+        """No silent sample drops, duplicate samples, or time reordering."""
+        for lines in [["x NaN 1"], ["x 1"], ["x 1 2", "x 2 1"], ["x 1 1", "x 2 1"]]:
+            with self.assertRaises(ValueError):
+                parse_samples(lines)
+
+
+if __name__ == "__main__":
+    unittest.main()