diff --git a/.github/workflows/mvp-ci.yml b/.github/workflows/mvp-ci.yml
index a2b9d4cb..6b8e5d8b 100644
--- a/.github/workflows/mvp-ci.yml
+++ b/.github/workflows/mvp-ci.yml
@@ -57,6 +57,10 @@ jobs:
           # post-step can package `target`; registry/git inputs are sufficient.
           cache-targets: false
 
+      - name: Test shared-workload comparison accounting
+        working-directory: ASAPQuery-backend
+        run: python3 -m unittest discover -s tools/shared-workload -p 'test*.py'
+
       - name: Check formatting
         working-directory: ASAPQuery-backend
         run: cargo fmt -p control_plane -p data_plane -- --check
diff --git a/control_plane/examples/compile_comparison_workload.rs b/control_plane/examples/compile_comparison_workload.rs
new file mode 100644
index 00000000..7c1fd6fd
--- /dev/null
+++ b/control_plane/examples/compile_comparison_workload.rs
@@ -0,0 +1,53 @@
+//! Compile reproducible comparison inputs without requiring a cost-optimality claim.
+use control_plane::physical::compiler::{BackendLocalPlanningSnapshot, PLANNER_REVISION};
+use serde_json::json;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let args: Vec<_> = std::env::args().skip(1).collect();
+    if args.len() != 2 {
+        return Err(
+            "usage: compile_comparison_workload prometheus|victoriametrics|clickhouse INPUT.json"
+                .into(),
+        );
+    }
+    let input = std::fs::read(&args[1])?;
+    let start = std::time::Instant::now();
+    let (install, selection) = match args[0].as_str() {
+        "clickhouse" => {
+            let workload = serde_json::from_slice(&input)?;
+            let (publication, trace) =
+                control_plane::clickhouse::compile_automatic_clickhouse_workload(&workload).await?;
+            (
+                serde_json::to_value(publication.install_request(None, Vec::new())?)?,
+                json!(trace),
+            )
+        }
+        engine @ ("prometheus" | "victoriametrics") => {
+            let snapshot: BackendLocalPlanningSnapshot = serde_json::from_slice(&input)?;
+            let plan = if engine == "victoriametrics" {
+                snapshot.compile_metricsql()?
+            } else {
+                snapshot.compile()?
+            };
+            let selection = json!({"cost_comparison": plan.cost_comparison, "lifecycle_estimates": plan.lifecycle_estimates});
+            (
+                json!({"summary_catalog": plan.summary_catalog, "collector_plans": plan.collector_plans,
+                    "precompute_plan": plan.precompute_plan, "transmission_plan": plan.transmission_plan,
+                    "query_plan": plan.query_plan, "storage_routing": null, "adaptation_evidence": []}),
+                selection,
+            )
+        }
+        _ => return Err("unknown comparison engine".into()),
+    };
+    serde_json::to_writer_pretty(
+        std::io::stdout(),
+        &json!({
+            "install": install, "selection": selection, "planner_revision": PLANNER_REVISION,
+            "backend_revision": env!("ASAPQUERY_BACKEND_REVISION"),
+            "planning_elapsed_ns": start.elapsed().as_nanos(),
+            "scope": "compile supplied planning evidence; measured execution does not establish cost-model optimality"
+        }),
+    )?;
+    Ok(())
+}
diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs
index c7f9be46..91dad725 100644
--- a/control_plane/src/clickhouse.rs
+++ b/control_plane/src/clickhouse.rs
@@ -39,37 +39,85 @@ pub async fn plan_clickhouse_sql(
     catalog: &SqlCatalog,
     accuracy: AccuracyTarget,
 ) -> Result<ClickHousePlannedQuery, ClickHousePlanningError> {
-    let canonical = lower_sql_dialect(sql, catalog, SqlDialect::ClickhouseSQL, accuracy.clone())
-        .await
-        .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?;
-    // SQL keeps relational parents such as Project and Filter above a
-    // summary-capable Aggregate. Use ASAPPlanner's recursive selector here;
-    // the PromQL deployment lowering retains its existing conservative rules.
+    plan_sql_cohort(&[(0, sql)], catalog, accuracy)
+        .await?
+        .pop()
+        .map(|(_, plan)| plan)
+        .ok_or_else(|| ClickHousePlanningError::Lower("SQL selection returned no root".into()))
+}
+
+async fn plan_sql_cohort(
+    queries: &[(usize, &str)],
+    catalog: &SqlCatalog,
+    accuracy: AccuracyTarget,
+) -> Result<Vec<(usize, ClickHousePlannedQuery)>, ClickHousePlanningError> {
+    let mut canonical = std::collections::BTreeMap::new();
+    for &(index, sql) in queries {
+        let root = lower_sql_dialect(sql, catalog, SqlDialect::ClickhouseSQL, accuracy.clone())
+            .await
+            .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?;
+        canonical.insert(index, root);
+    }
     let cost_model = ControlPlaneCostModel::new(accuracy.clone());
     let (selected, selection_trace) =
         crate::planner_selection::select_workload_with_accuracy_model_and_trace(
-            vec![(0, Rc::new(canonical.clone()))],
+            canonical
+                .iter()
+                .map(|(&index, root)| (index, Rc::new(root.clone())))
+                .collect(),
             accuracy,
             &cost_model,
             &asap_aware_mapping::NoAccuracyEvidence,
             &asap_aware_mapping::DefaultAccuracyModel,
         )?;
-    let selected = selected
+    if selected.len() != canonical.len() {
+        return Err(ClickHousePlanningError::Lower(
+            "SQL selection lost workload roots".into(),
+        ));
+    }
+    selected
         .into_iter()
-        .next()
-        .map(|(_, node)| node)
-        .ok_or_else(|| {
-            crate::planner_selection::SelectionError::Workload(
-                "SQL workload search returned no root".into(),
-            )
-        })?;
-    let physical = PhysicalExpr::committed(selected);
-    Ok(ClickHousePlannedQuery {
-        canonical_sql: canonical_sql_identity(&canonical),
-        canonical,
-        physical,
-        selection_trace,
-    })
+        .map(|(index, selected)| {
+            let canonical = canonical.remove(&index).ok_or_else(|| {
+                ClickHousePlanningError::Lower(
+                    "SQL selection returned an unknown or duplicate root".into(),
+                )
+            })?;
+            Ok((
+                index,
+                ClickHousePlannedQuery {
+                    canonical_sql: canonical_sql_identity(&canonical),
+                    canonical,
+                    physical: PhysicalExpr::committed(selected),
+                    selection_trace: selection_trace.clone(),
+                },
+            ))
+        })
+        .collect()
+}
+
+async fn plan_sql_workload(
+    queries: &[ClickHouseSqlWorkloadEntry],
+    catalog: &SqlCatalog,
+    accuracy: AccuracyTarget,
+) -> Result<Vec<(usize, ClickHousePlannedQuery)>, ClickHousePlanningError> {
+    // Physical bindings carry pane origins and cumulative evaluation semantics.
+    // Share only within the same runtime range; canonical SQL keeps source
+    // predicates, table identity, and aggregation semantics distinct inside it.
+    let mut cohorts = std::collections::BTreeMap::<_, Vec<_>>::new();
+    for (index, query) in queries.iter().enumerate() {
+        validate_sql_evaluation(query)?;
+        cohorts
+            .entry((query.start_ms, query.end_ms, query.cumulative))
+            .or_default()
+            .push((index, query.sql.as_str()));
+    }
+    let mut planned = Vec::with_capacity(queries.len());
+    for cohort in cohorts.into_values() {
+        planned.extend(plan_sql_cohort(&cohort, catalog, accuracy.clone()).await?);
+    }
+    planned.sort_by_key(|(index, _)| *index);
+    Ok(planned)
 }
 
 pub async fn canonicalize_clickhouse_sql(
@@ -247,12 +295,10 @@ pub async fn compile_automatic_clickhouse_workload(
     let mut installed_dags = std::collections::BTreeMap::new();
     let mut materializations = std::collections::BTreeMap::new();
     let mut selection_traces = std::collections::BTreeMap::new();
-    for query in &request.queries {
-        validate_sql_evaluation(query)?;
-        // Selection runs once. Compilation installs only SummaryAgg nodes
-        // actually visited in this selected DAG, never a scripted family.
-        let mut planned =
-            plan_clickhouse_sql(&query.sql, &catalog, request.accuracy.clone()).await?;
+    for (index, mut planned) in
+        plan_sql_workload(&request.queries, &catalog, request.accuracy.clone()).await?
+    {
+        let query = &request.queries[index];
         let selection_trace = std::mem::take(&mut planned.selection_trace);
         let template = planned.canonical_sql.clone();
         let (entry, installed) = compile_selected_sql(query, planned, |node, family| {
@@ -408,8 +454,10 @@ pub async fn compile_clickhouse_workload(
     let mut entries = std::collections::BTreeMap::new();
     let mut window_templates = std::collections::BTreeMap::<String, Vec<String>>::new();
     let mut installed_dags = std::collections::BTreeMap::new();
-    for query in &request.queries {
-        let planned = plan_clickhouse_sql(&query.sql, &catalog, request.accuracy.clone()).await?;
+    for (index, planned) in
+        plan_sql_workload(&request.queries, &catalog, request.accuracy.clone()).await?
+    {
+        let query = &request.queries[index];
         let template = planned.canonical_sql.clone();
         let (executable, installed) = compile_selected_sql(query, planned, |node, family| {
             bind_selected_node(node, family, query, request)
@@ -880,6 +928,98 @@ mod tests {
     use asap_types::{AggregationType, KeyByLabelNames, PrecomputeMaterialization, WindowKind};
     use planner_types::pre_asap::{Column, DataType, Schema};
 
+    // SQL publications must expose one selection over all compatible roots.
+    #[tokio::test]
+    async fn automatic_sql_selects_a_workload_cohort() {
+        let request = ClickHouseSqlAutomaticWorkload {
+            envelope: crate::physical::compiler::PlanEnvelope {
+                plan_id: 73, plan_version: 1, generated_at_unix_ms: 0,
+                activation_unix_ms: 0, expiry_unix_ms: None,
+                backend_compat: crate::physical::compiler::BACKEND_COMPAT.into(),
+                planner_revision: crate::physical::compiler::PLANNER_REVISION.into(),
+                capability_snapshot_id: "sql-cohort-test".into(),
+            },
+            tables: HashMap::from([("telemetry".into(), Schema::with_time_index(
+                vec![Column::new("timestamp_ms", DataType::Int64, false),
+                     Column::new("value", DataType::Float64, false)], 0, vec![]))]),
+            accuracy: AccuracyTarget::Exact,
+            queries: ["sum(value)", "sum(value) + 1 AS total"].into_iter().map(|expression| ClickHouseSqlWorkloadEntry {
+                sql: format!("SELECT {expression} FROM telemetry WHERE timestamp_ms >= 0 AND timestamp_ms < 2000"),
+                start_ms: 0, end_ms: 2000, cumulative: false,
+            }).collect(),
+        };
+        let (publication, traces) = compile_automatic_clickhouse_workload(&request)
+            .await
+            .unwrap();
+        assert_eq!(publication.query_plan.entries.len(), 2);
+        assert_eq!(publication.summary_catalog.materializations.len(), 1);
+        for trace in traces.values() {
+            assert_eq!(trace["roots"].as_array().unwrap().len(), 2);
+            assert_eq!(trace["roots"][0]["query_index"], 0);
+            assert_eq!(trace["roots"][1]["query_index"], 1);
+            assert_eq!(trace, traces.values().next().unwrap());
+        }
+        // Equal source ASTs cannot override incompatible runtime bindings.
+        let mut queries = request.queries;
+        queries[1].cumulative = true;
+        let separated = plan_sql_workload(
+            &queries,
+            &SqlCatalog {
+                tables: request.tables,
+            },
+            AccuracyTarget::Exact,
+        )
+        .await
+        .unwrap();
+        assert_eq!(separated.len(), 2);
+        for (index, plan) in separated {
+            assert_eq!(plan.selection_trace["roots"].as_array().unwrap().len(), 1);
+            assert_eq!(plan.selection_trace["roots"][0]["query_index"], index);
+        }
+    }
+
+    // Workload search can offer cross-query rollup without forcing its selection.
+    #[tokio::test]
+    async fn sql_cohort_exposes_rollup_candidates() {
+        let catalog = SqlCatalog {
+            tables: HashMap::from([(
+                "telemetry".into(),
+                Schema::with_time_index(
+                    vec![
+                        Column::new("timestamp_ms", DataType::Int64, false),
+                        Column::new("value", DataType::Float64, false),
+                        Column::new("job", DataType::Utf8, false),
+                    ],
+                    0,
+                    vec![],
+                ),
+            )]),
+        };
+        let grouped = "SELECT job, sum(value) AS total FROM telemetry WHERE timestamp_ms >= 0 AND timestamp_ms < 2000 GROUP BY job";
+        let total = "SELECT sum(value) AS total FROM telemetry WHERE timestamp_ms >= 0 AND timestamp_ms < 2000";
+        let has_rollup = |trace: &serde_json::Value| {
+            trace["groups"].as_array().unwrap().iter().any(|group| {
+                group["candidates"]
+                    .as_array()
+                    .unwrap()
+                    .iter()
+                    .any(|candidate| candidate["strategy"] == "RollupStrategy")
+            })
+        };
+        for sql in [grouped, total] {
+            let single = plan_clickhouse_sql(sql, &catalog, AccuracyTarget::Exact)
+                .await
+                .unwrap();
+            assert!(!has_rollup(&single.selection_trace));
+        }
+        let combined =
+            plan_sql_cohort(&[(0, grouped), (1, total)], &catalog, AccuracyTarget::Exact)
+                .await
+                .unwrap();
+        assert_eq!(combined.len(), 2);
+        assert!(has_rollup(&combined[0].1.selection_trace));
+    }
+
     // A dashboard refresh must reuse the installed identity without treating
     // value thresholds or window length as runtime parameters.
     #[tokio::test]
diff --git a/tools/shared-workload/BENEFITS.md b/tools/shared-workload/BENEFITS.md
new file mode 100644
index 00000000..fd260a10
--- /dev/null
+++ b/tools/shared-workload/BENEFITS.md
@@ -0,0 +1,112 @@
+# Native/ASAP execution comparison
+
+For developers running reproducible experiments. `experiment.py` compiles the
+supplied workload through the real control plane, starts three backend processes,
+verifies the installed generation is active, loads native and fallback services,
+materializes summaries, and invokes all six query endpoints. Each ASAP result is
+compared with its own native engine; MetricsQL is not assumed to equal PromQL.
+
+## Run the isolated functional smoke
+
+Build `control_plane`'s `compile_comparison_workload` example and `data_plane`.
+The workspace's sibling Collector and sketchlib checkouts must match the backend
+build requirements. Supply local immutable Docker image IDs or digests:
+
+```sh
+cargo build -p control_plane --example compile_comparison_workload
+cargo build -p data_plane --bin data_plane
+python3 tools/shared-workload/smoke.py \
+  --compiler "$PWD/target/debug/examples/compile_comparison_workload" \
+  --data-plane "$PWD/target/debug/data_plane" \
+  --prometheus-image "$PROMETHEUS_IMAGE" \
+  --clickhouse-image "$CLICKHOUSE_IMAGE" \
+  --victoriametrics-image "$VICTORIAMETRICS_IMAGE" \
+  --output /tmp/asap-comparison-smoke
+```
+
+Use `--docker 'sudo -n docker'` if your Docker access requires it. The smoke owns
+six temporary native/fallback containers, each limited to two CPUs and 2 GiB,
+and removes them on exit. ASAP processes run on the host: this smoke checks
+functionality, not equal-budget performance. It uses declared planning-cost
+inputs, explicitly marked as functional seeds, not empirical calibration.
+
+Two related sum queries exercise joint SQL selection, initial and incremental
+materialization, and reuse of the installed SQL plan at a later window. Success
+requires correct **warm** execution for every supported request in every engine,
+and correct **exact_fallback** execution for a third, uninstalled window. Expected
+values are checked independently so empty native/ASAP results cannot pass.
+
+## Run a supplied workload
+
+`smoke.py` retains `experiment.json`, planning inputs, and the query manifest as
+an editable input example. For another run, supply fresh, empty, isolated native
+and fallback instances; container URLs in a completed smoke are no longer live.
+
+```sh
+python3 tools/shared-workload/experiment.py \
+  --config experiment.json --output /tmp/asap-comparison-run
+```
+
+Configuration uses absolute paths for `compiler`, `data_plane`, `manifest`, each
+engine's `planning_input`, and each input batch's `openmetrics` and `jsonl` files.
+Prometheus and VictoriaMetrics inputs are `BackendLocalPlanningSnapshot` values;
+the MetricsQL compiler entry point selects the native frontend. SQL input is a
+`ClickHouseSqlAutomaticWorkload`, without forced families or materialization IDs.
+Version 2 snapshot cost-evidence requirements still apply.
+
+Each engine has distinct `native` and `fallback` endpoints with `url`, optional
+local `pid`, optional `storage_path`, and optional HTTP `headers`. ClickHouse also
+accepts `user` and `password`. The runner creates and drops its own fresh SQL
+database on both instances; `table` and `columns_sql` describe its input table.
+Do not put a database qualifier in workload SQL: the runner supplies its database.
+All provided planning inputs must match the batch schema and event times.
+
+`batches` is an ordered, non-overlapping initial batch followed by zero or more
+maintenance batches. SQL `start_ms`/`end_ms` specify contiguous input ranges aligned with the installed
+pane layout. SQL performs one complete backfill after loading all batches: the
+current backend rejects later repeated backfills after first-seen metadata appears.
+SQL source-load maintenance is measured, but continuous summary maintenance and
+its CPU break-even are explicitly unavailable. OpenMetrics timestamps are
+seconds. Only the final metrics batch drains and seals input. Validation should
+use finite samples with strictly increasing timestamps per series.
+
+Optional `relative_tolerance` and `absolute_tolerance` control numeric value
+comparison (defaults `1e-9` and `1e-12`). They are recorded in the report; they
+do not establish quantile rank-error or Top-K membership guarantees.
+
+The query manifest has `end_ms` and nonempty `queries`; each query provides
+`name`, `promql`, native `metricsql`, and `clickhouse_sql` without a FORMAT clause.
+An optional `evaluation_ms` overrides the manifest time. SQL accepts `{eval_ms}`
+and `{lookback_ms}` placeholders. SQL column metadata and result row multisets
+are compared; ordering-sensitive workloads need a separate order contract.
+
+For already prepared deployments, `benefits.py --endpoints ENDPOINTS.json
+--manifest MANIFEST.json --output OUTPUT` runs only comparison. Its endpoints
+map each engine to `native` and `asap` URLs (ClickHouse also needs `database`).
+It makes no setup or maintenance measurement claim.
+
+## Interpret the artifacts
+
+- `planning.json`, `install.json`, and `status.json`: actual selected and active
+  plans, source revisions, and selection traces.
+- `requests.jsonl`: independently retained native/ASAP responses, latency,
+  execution provenance, and process CPU deltas. A timeout does not skip another
+  endpoint. Every engine's mismatch contributes to a failing exit status.
+- `phases.json`: planning, startup/install, initial build, and incremental
+  maintenance, and SQL materialization. Metrics maintenance includes ingest and materialization; SQL materialization is one separately charged finite backfill. The final drain
+  is the metrics completion barrier; intermediate batches are not independent
+  steady-state maintenance measurements.
+- `report.json`: correctness, warm/hybrid/fallback counts, serial query latency,
+  phase CPU, full observation-interval CPU, process RSS, storage when available,
+  and conditional CPU break-even refresh count. Missing resource measurements
+  stay null. Native startup is included only if supplied by the provisioner.
+- `cleanup.json`: cleanup failures, if any.
+
+ASAP CPU includes its fallback service and compiler. Phase sums exclude gaps;
+observation-interval CPU includes idle/background work during other arms too.
+RSS samples and process lifetime high-water marks are not summary-state memory.
+CPU break-even assumes the same query mix and measured maintenance per dashboard refresh;
+failed comparisons, missing CPU, and non-positive net savings yield no estimate.
+Correct fallback can contribute to deployment-level timing, but only `warm`
+counts as summary-only acceleration. These finite serial runs do not measure
+sustained ingestion, concurrent throughput, or cost-model optimality.
diff --git a/tools/shared-workload/benefits.py b/tools/shared-workload/benefits.py
new file mode 100644
index 00000000..1d08a77f
--- /dev/null
+++ b/tools/shared-workload/benefits.py
@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+"""Compare three native/ASAP pairs without hiding failures or fallback execution."""
+import argparse
+import json
+import math
+from pathlib import Path
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'o11y-execution'))
+from compare import compare_results, distribution, process_snapshot, process_delta
+
+ENGINES = ('prometheus', 'clickhouse', 'victoriametrics')
+
+
+def http(url, data=None, headers=None, timeout=30):
+    start = time.perf_counter_ns()
+    try:
+        try:
+            stream = urllib.request.urlopen(urllib.request.Request(url, data=data, headers=headers or {}), timeout=timeout)
+        except urllib.error.HTTPError as error:
+            stream = error
+        with stream:
+            raw = stream.read().decode()
+            try:
+                body = json.loads(raw)
+            except ValueError:
+                body = raw
+            result = {'status': stream.code, 'headers': {k.lower(): v for k, v in stream.headers.items()}, 'body': body}
+    except (OSError, ValueError) as error:
+        result = {'status': None, 'headers': {}, 'error': str(error)}
+    return dict(result, elapsed_ns=time.perf_counter_ns() - start)
+
+
+def _query_endpoint(engine, endpoint, query, evaluation_ms, timeout):
+    endpoint = {'url': endpoint} if isinstance(endpoint, str) else endpoint
+    url = endpoint['url'].rstrip('/')
+    headers = endpoint.get('headers', {})
+    if engine == 'clickhouse':
+        sql = query['clickhouse_sql'].replace('{eval_ms}', str(evaluation_ms))
+        sql = sql.replace('{lookback_ms}', str(query.get('lookback_ms', 300000)))
+        # The workload supplies a SELECT without a FORMAT clause. JSON retains
+        # ClickHouse type metadata, including its integer-quoting contract.
+        return http(url + '/?' + urllib.parse.urlencode({'database': endpoint['database']}),
+                    (sql.rstrip().rstrip(';') + ' FORMAT JSON').encode(), headers, timeout)
+    expression = query['promql' if engine == 'prometheus' else 'metricsql']
+    return http(url + '/api/v1/query?' + urllib.parse.urlencode(
+        {'query': expression, 'time': evaluation_ms / 1000, **({'nocache': 1} if engine == 'victoriametrics' else {})}), headers=headers, timeout=timeout)
+
+
+def query_endpoint(engine, endpoint, query, evaluation_ms, timeout):
+    pids = endpoint.get('pids', []) if isinstance(endpoint, dict) else []
+    before = [process_snapshot(pid) for pid in pids]
+    result = _query_endpoint(engine, endpoint, query, evaluation_ms, timeout)
+    after = [process_snapshot(pid) for pid in pids]
+    deltas = [process_delta(a, b) for a, b in zip(before, after)]
+    result['process_resources'] = deltas
+    result['cpu_ns'] = sum(d['cpu_ns'] for d in deltas) if deltas and all(deltas) else None
+    return result
+
+
+def sql_compare(actual, expected, relative, absolute):
+    if actual.get('meta') != expected.get('meta'):
+        return {'equal': False, 'reason': 'SQL column names or types differ'}
+    a, b = actual.get('data'), expected.get('data')
+    if not isinstance(a, list) or not isinstance(b, list):
+        return {'equal': False, 'reason': 'missing SQL result rows'}
+    errors = []
+    def equal(x, y):
+        if isinstance(x, (int, float)) and not isinstance(x, bool) and isinstance(y, (int, float)) and not isinstance(y, bool):
+            if not math.isfinite(x) or not math.isfinite(y):
+                return False
+            errors.append(abs(x - y))
+            return math.isclose(x, y, rel_tol=relative, abs_tol=absolute)
+        if type(x) is not type(y):
+            return False
+        if isinstance(x, dict):
+            return x.keys() == y.keys() and all(equal(x[k], y[k]) for k in x)
+        if isinstance(x, list):
+            return len(x) == len(y) and all(equal(i, j) for i, j in zip(x, y))
+        return x == y
+    # SQL without ORDER BY promises a multiset. Preserve duplicate rows.
+    sort_key = lambda row: json.dumps(row, sort_keys=True)
+    matched = equal(sorted(a, key=sort_key), sorted(b, key=sort_key))
+    return {'equal': matched, 'max_absolute_error': max(errors, default=None),
+            'expected_rows': len(b), 'actual_rows': len(a)}
+
+
+def assess(engine, native, asap, relative=1e-9, absolute=1e-12):
+    comparison = {'equal': False, 'reason': 'endpoint failed'}
+    if native.get('status') == asap.get('status') == 200:
+        try:
+            comparison = (sql_compare(asap['body'], native['body'], relative, absolute) if engine == 'clickhouse'
+                          else compare_results(asap['body'], native['body'], relative, absolute))
+        except (KeyError, TypeError, ValueError, AttributeError) as error:
+            comparison = {'equal': False, 'reason': str(error)}
+    route = asap.get('headers', {}).get('x-asap-execution', 'unknown') if asap.get('status') == 200 else 'failed'
+    passed = comparison['equal'] and route != 'failed'
+    return {'native': native, 'asap': asap, 'comparison': comparison, 'passed': passed,
+            'execution': route, 'accelerated': passed and route == 'warm'}
+
+
+def compare_round(endpoints, query, evaluation_ms, timeout, relative=1e-9, absolute=1e-12, reverse=False):
+    results = {}
+    for engine in ENGINES:
+        responses = {}
+        for route in (('asap', 'native') if reverse else ('native', 'asap')):
+            start = time.perf_counter_ns()
+            try:
+                responses[route] = query_endpoint(engine, endpoints[engine][route], query, evaluation_ms, timeout)
+            except Exception as error:
+                responses[route] = {'status': None, 'headers': {}, 'error': str(error),
+                                    'elapsed_ns': time.perf_counter_ns() - start}
+        results[engine] = assess(engine, responses['native'], responses['asap'], relative, absolute)
+    return results
+
+
+def summarize(records):
+    summary = {}
+    for engine in ENGINES:
+        rows = [record['engines'][engine] for record in records]
+        passed = bool(rows) and all(r['passed'] for r in rows)
+        summary[engine] = {'requests': len(rows), 'passed': passed,
+                           'equal': sum(r['passed'] for r in rows),
+                           'accelerated': sum(r['accelerated'] for r in rows),
+                           'execution_counts': {k: sum(r['execution'] == k for r in rows)
+                                                for k in ('warm', 'hybrid', 'exact_fallback', 'failed', 'unknown')},
+                           'native_latency': distribution([r['native']['elapsed_ns'] for r in rows]),
+                           'asap_latency': distribution([r['asap']['elapsed_ns'] for r in rows]),
+                           'native_over_asap_latency_ratio': (sum(r['native']['elapsed_ns'] for r in rows) /
+                                                             sum(r['asap']['elapsed_ns'] for r in rows)) if passed else None}
+    return summary
+
+
+def run(endpoints, manifest, repetitions, timeout, relative=1e-9, absolute=1e-12, record_sink=None):
+    if repetitions < 1 or not manifest['queries']:
+        raise ValueError('nonempty workload and positive repetitions required')
+    if timeout <= 0 or not math.isfinite(timeout) or any(not math.isfinite(v) or v < 0 for v in (relative, absolute)):
+        raise ValueError('timeout and tolerances must be finite and valid')
+    records = []
+    for repeat in range(repetitions):
+        for index, query in enumerate(manifest['queries']):
+            evaluation_ms = query.get('evaluation_ms', manifest['end_ms'])
+            record = {'query_index': index, 'name': query.get('name', str(index)), 'repeat': repeat,
+                      'evaluation_ms': evaluation_ms,
+                      'engines': compare_round(endpoints, query, evaluation_ms, timeout, relative, absolute, bool(repeat % 2))}
+            records.append(record)
+            if record_sink:
+                record_sink(record)
+    summary = summarize(records)
+    return {'schema_version': 1, 'passed': all(v['passed'] for v in summary.values()),
+            'numeric_tolerance': {'relative': relative, 'absolute': absolute},
+            'timing_scope': 'serial alternating HTTP requests; fixed evaluation times; not a concurrency/throughput measurement',
+            'summary': summary, 'records': records}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    for name in ('endpoints', 'manifest', 'output'):
+        parser.add_argument('--' + name, type=Path, required=True)
+    parser.add_argument('--repetitions', type=int, default=10)
+    parser.add_argument('--timeout', type=float, default=30)
+    parser.add_argument('--relative-tolerance', type=float, default=1e-9)
+    parser.add_argument('--absolute-tolerance', type=float, default=1e-12)
+    args = parser.parse_args()
+    args.output.mkdir(parents=True, exist_ok=False)
+    with (args.output / 'requests.jsonl').open('w') as trace:
+        def record(row):
+            trace.write(json.dumps(row) + '\n')
+            trace.flush()
+        report = run(json.loads(args.endpoints.read_text()), json.loads(args.manifest.read_text()),
+                     args.repetitions, args.timeout, args.relative_tolerance, args.absolute_tolerance, record_sink=record)
+    (args.output / 'report.json').write_text(json.dumps(report, indent=2) + '\n')
+    raise SystemExit(0 if report['passed'] else 1)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/tools/shared-workload/costs.py b/tools/shared-workload/costs.py
new file mode 100644
index 00000000..5d14a9df
--- /dev/null
+++ b/tools/shared-workload/costs.py
@@ -0,0 +1,63 @@
+"""Cost accounting for finite, serial replay; missing measurements stay missing."""
+import math
+from benefits import ENGINES
+
+
+def complete_sum(values):
+    return sum(values) if values and all(v is not None and math.isfinite(v) and v >= 0 for v in values) else None
+
+
+def break_even(upfront, native_query, asap_query, incremental_maintenance):
+    if any(v is None for v in (upfront, native_query, asap_query, incremental_maintenance)):
+        return {'refreshes': None, 'reason': 'incomplete CPU evidence'}
+    saved = native_query - asap_query - incremental_maintenance
+    if saved <= 0:
+        return {'refreshes': None, 'reason': 'no positive net CPU saving per refresh'}
+    return {'refreshes': max(0, math.ceil(upfront / saved)), 'net_cpu_saved_per_refresh_ns': saved,
+            'assumption': 'same query mix and maintenance CPU per refresh as this finite replay; excludes future drift'}
+
+
+def summarize_costs(phases, records, lifetime=None):
+    result = {}
+    refresh_count = len({r['repeat'] for r in records})
+    for engine in ENGINES:
+        rows = [r['engines'][engine] for r in records]
+        eligible = bool(rows) and all(r['passed'] for r in rows)
+        arms = {}
+        for arm in ('native', 'asap'):
+            selected = [p for p in phases if p['engine'] == engine and
+                        (p['arm'] == arm or (arm == 'asap' and p['arm'] == 'fallback'))]
+            def phase_cpu(p):
+                return p.get('compiler_cpu_ns') if p['phase'] == 'planning' else p.get('cpu_ns')
+            upfront = [p for p in selected if p['phase'] != 'maintenance']
+            maintenance = [p for p in selected if p['phase'] == 'maintenance']
+            arms[arm] = {
+                'upfront_cpu_ns': complete_sum([phase_cpu(p) if p['complete'] else None for p in upfront]),
+                'maintenance_cpu_ns': complete_sum([phase_cpu(p) if p['complete'] else None for p in maintenance]) if maintenance else 0,
+                'maintenance_batches': len(maintenance),
+                'query_cpu_ns': complete_sum([r[arm].get('cpu_ns') for r in rows]),
+                'upfront_wall_ns': complete_sum([p['wall_ns'] if p['complete'] else None for p in upfront]),
+                'maintenance_wall_ns': complete_sum([p['wall_ns'] if p['complete'] else None for p in maintenance]) if maintenance else 0,
+                'query_wall_ns': complete_sum([r[arm]['elapsed_ns'] for r in rows]),
+            }
+            arms[arm]['accounted_phase_cpu_ns'] = complete_sum([arms[arm][k] for k in ('upfront_cpu_ns', 'maintenance_cpu_ns', 'query_cpu_ns')])
+            arms[arm]['accounted_serial_wall_ns'] = complete_sum([arms[arm][k] for k in ('upfront_wall_ns', 'maintenance_wall_ns', 'query_wall_ns')])
+        native, asap = arms['native'], arms['asap']
+        subtract = lambda a, b: a - b if a is not None and b is not None else None
+        per_refresh = lambda value: value / refresh_count if value is not None and refresh_count else None
+        amortization = break_even(subtract(asap['upfront_cpu_ns'], native['upfront_cpu_ns']),
+                                 per_refresh(native['query_cpu_ns']), per_refresh(asap['query_cpu_ns']),
+                                 per_refresh(subtract(asap['maintenance_cpu_ns'], native['maintenance_cpu_ns'])))
+        if not eligible:
+            amortization = {'refreshes': None, 'reason': 'failed or unequal query results'}
+        total = (lifetime or {}).get(engine)
+        def ratio(a, b):
+            return a / b if eligible and a is not None and b is not None and b > 0 else None
+        result[engine] = {'observed_refreshes': refresh_count,
+                          'native_over_asap_phase_cpu_ratio': ratio(native['accounted_phase_cpu_ns'], asap['accounted_phase_cpu_ns']),
+                          'native_over_asap_serial_total_time_ratio': ratio(native['accounted_serial_wall_ns'], asap['accounted_serial_wall_ns']),
+                          'native_over_asap_lifetime_cpu_ratio': ratio(total.get('native_cpu_ns'), total.get('asap_plus_fallback_and_planner_cpu_ns')) if total else None,
+                          'arms': arms, 'cpu_break_even': amortization, 'lifetime': total,
+                          'comparison_eligible': eligible,
+                          'scope': 'phase CPU includes whole backend and fallback processes; phase sums exclude gaps; lifetime includes gaps and background work; serial wall time is not CPU or concurrent throughput'}
+    return result
diff --git a/tools/shared-workload/experiment.py b/tools/shared-workload/experiment.py
new file mode 100644
index 00000000..75d011dd
--- /dev/null
+++ b/tools/shared-workload/experiment.py
@@ -0,0 +1,346 @@
+#!/usr/bin/env python3
+"""Compile, install, materialize, and compare three native/ASAP deployments.
+
+Native and fallback services must be separately provisioned, empty, isolated
+instances. This runner owns its three backend children and a fresh SQL database.
+"""
+import argparse
+import base64
+from contextlib import contextmanager
+import hashlib
+import itertools
+import json
+import os
+from pathlib import Path
+import resource
+import socket
+import subprocess
+import time
+import uuid
+
+import benefits
+from compare import process_snapshot, process_delta
+from replay import iter_samples, encode_write
+from process_lifecycle import stop
+
+
+def sha256(path):
+    digest = hashlib.sha256()
+    with Path(path).open('rb') as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b''):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def save(path, value):
+    path.write_text(json.dumps(value, indent=2, allow_nan=False) + '\n')
+
+
+def require(result):
+    if result.get('status') is None or not 200 <= result['status'] < 300:
+        raise RuntimeError(str(result))
+    return result.get('body')
+
+
+def post(url, value=None):
+    return require(benefits.http(url, json.dumps(value).encode() if value is not None else b'',
+                                 {'Content-Type': 'application/json'}))
+
+
+def port():
+    with socket.socket() as sock:
+        sock.bind(('127.0.0.1', 0))
+        return sock.getsockname()[1]
+
+
+class Measurements:
+    def __init__(self, path):
+        self.path, self.rows = path, []
+
+    @contextmanager
+    def phase(self, engine, arm, phase, pids):
+        before = [process_snapshot(pid) for pid in pids]
+        started = time.perf_counter_ns()
+        row = {'engine': engine, 'arm': arm, 'phase': phase, 'complete': False}
+        try:
+            yield row
+            row['complete'] = True
+        finally:
+            after = [process_snapshot(pid) for pid in pids]
+            deltas = [process_delta(a, b) for a, b in zip(before, after)]
+            row.update(wall_ns=time.perf_counter_ns() - started, processes=deltas,
+                       cpu_ns=sum(v['cpu_ns'] for v in deltas) if deltas and all(deltas) else None)
+            self.rows.append(row)
+            save(self.path, self.rows)
+
+
+def sql(endpoint, database, statement):
+    import urllib.parse
+    return require(benefits.http(endpoint['url'].rstrip('/') + '/?' + urllib.parse.urlencode({'database': database}),
+                                 statement.encode(), endpoint.get('headers'), timeout=120))
+
+
+def remote_write(endpoint, path):
+    with Path(path).open() as source:
+        rows = iter_samples(source)
+        while batch := list(itertools.islice(rows, 5000)):
+            require(benefits.http(endpoint['url'].rstrip('/') + '/api/v1/write', encode_write(batch), {
+                **endpoint.get('headers', {}), 'Content-Type': 'application/x-protobuf',
+                'Content-Encoding': 'snappy', 'X-Prometheus-Remote-Write-Version': '0.1.0'}, timeout=120))
+
+
+def load_sql(endpoint, database, table, path):
+    with Path(path).open() as source:
+        while batch := list(itertools.islice(source, 5000)):
+            sql(endpoint, database, f'INSERT INTO {table} FORMAT JSONEachRow\n' + ''.join(batch))
+
+
+def wait_ready(api, child):
+    deadline = time.monotonic() + 60
+    while time.monotonic() < deadline:
+        if child.poll() is not None:
+            raise RuntimeError('backend exited before readiness; inspect backend.log')
+        if benefits.http(api + '/api/v1/health', timeout=1)['status'] == 200:
+            return
+        time.sleep(.1)
+    raise TimeoutError('backend readiness timeout')
+
+
+def backfill(api, install, database, table, start_ms, end_ms):
+    for identity, definition in install['summary_catalog']['materializations'].items():
+        pane_ms = definition['window_layout']['pane_secs'] * 1000
+        if (start_ms - definition['pane_origin_ms']) % pane_ms or (end_ms - start_ms) % pane_ms:
+            raise ValueError('backfill input range must align with materialized panes')
+        post(api + '/api/v1/db/backfill', {'agg_id': int(identity), 'start_ms': start_ms, 'end_ms': end_ms,
+             'source': {'ClickHouse': {'database': database, 'table': table}}, 'windows_total': (end_ms - start_ms) // pane_ms})
+    deadline = time.monotonic() + 120
+    while time.monotonic() < deadline:
+        jobs = require(benefits.http(api + '/api/v1/db/backfill/jobs'))
+        statuses = [job['status'] for job in jobs['jobs']]
+        if 'failed' in statuses:
+            raise RuntimeError(str(jobs))
+        if not install['summary_catalog']['materializations'] or (statuses and all(s == 'complete' for s in statuses)):
+            return jobs
+        time.sleep(.1)
+    raise TimeoutError('backfill completion timeout')
+
+
+def validate(config):
+    urls, pids = [], []
+    if not config.get('batches'):
+        raise ValueError('at least one input batch is required')
+    for engine in benefits.ENGINES:
+        for arm in ('native', 'fallback'):
+            endpoint = config['engines'][engine][arm]
+            urls.append(endpoint['url'].rstrip('/'))
+            if endpoint.get('user'):
+                credential = endpoint['user'] + ':' + endpoint.get('password', '')
+                endpoint.setdefault('headers', {})['Authorization'] = 'Basic ' + base64.b64encode(credential.encode()).decode()
+            if endpoint.get('pid'):
+                pids.append(endpoint['pid'])
+    if len(urls) != len(set(urls)) or len(pids) != len(set(pids)):
+        raise ValueError('native and fallback arms require distinct isolated service URLs and PIDs')
+    if not config['table'].replace('_', '').isalnum() or config['table'][0].isdigit():
+        raise ValueError('table must be a simple SQL identifier')
+
+
+def run(config, output):
+    validate(config)
+    # Validate the complete metric stream before any external write.
+    def sample_lines():
+        for batch in config['batches']:
+            with Path(batch['openmetrics']).open() as source:
+                yield from source
+    sample_count = sum(1 for _ in iter_samples(sample_lines()))
+    output.mkdir(parents=True, exist_ok=False)
+    measurements = Measurements(output / 'phases.json')
+    children, logs, databases, deployments, endpoints = [], [], [], {}, {}
+    database = 'asap_comparison_' + uuid.uuid4().hex
+    artifacts = [config['compiler'], config['data_plane'], config['manifest']]
+    artifacts += [e['planning_input'] for e in config['engines'].values()]
+    artifacts += [b[k] for b in config['batches'] for k in ('openmetrics', 'jsonl')]
+    save(output / 'provenance.json', {'inputs': {str(p): sha256(p) for p in artifacts},
+         'database': database, 'sample_count': sample_count, 'scope': 'finite input replay; native service startup included only when supplied by provisioner; includes incremental materialization batches and fallback; no sustained-load or planner-optimality claim'})
+    for engine in benefits.ENGINES:
+        for arm in ('native', 'fallback'):
+            startup = config['engines'][engine][arm].get('startup_cost')
+            if startup:
+                measurements.rows.append(dict(startup, engine=engine, arm=arm, phase='native_service_startup', complete=True))
+    lifetime_before = {engine: {arm: process_snapshot(config['engines'][engine][arm].get('pid'))
+                       for arm in ('native', 'fallback')} for engine in benefits.ENGINES}
+    try:
+        for engine in benefits.ENGINES:
+            settings = config['engines'][engine]
+            folder = output / engine
+            folder.mkdir()
+            planning_before = resource.getrusage(resource.RUSAGE_CHILDREN)
+            with measurements.phase(engine, 'asap', 'planning', []) as phase:
+                compiled = subprocess.run([config['compiler'], engine, settings['planning_input']],
+                    capture_output=True, text=True, timeout=300)
+                (folder / 'planning.stderr').write_text(compiled.stderr)
+                compiled.check_returncode()
+                plan = json.loads(compiled.stdout)
+                after = resource.getrusage(resource.RUSAGE_CHILDREN)
+                phase['compiler_cpu_ns'] = round((after.ru_utime + after.ru_stime - planning_before.ru_utime - planning_before.ru_stime) * 1e9)
+                save(folder / 'planning.json', plan)
+            install = plan['install']
+            save(folder / 'install.json', install)
+            api_port, query_port = port(), port()
+            api = f'http://127.0.0.1:{api_port}'
+            command = [config['data_plane'], '--http-port', str(api_port), '--output-dir', str(folder / 'state'),
+                       '--precompute-allowed-lateness-ms', '0', '--precompute-flush-interval-ms', '25',
+                       '--persistence-delete-older-than-secs', '0']
+            fallback = settings['fallback']
+            if engine == 'clickhouse':
+                bootstrap = folder / 'bootstrap.yaml'
+                bootstrap.write_text('aggregations: []\n')
+                command += ['--streaming-config', str(bootstrap), '--clickhouse-http-port', str(query_port),
+                    '--clickhouse-url', fallback['url'], '--clickhouse-database', database,
+                    '--clickhouse-backfill-database', database, '--clickhouse-backfill-table', config['table'],
+                    '--enable-backfill-worker']
+                for key in ('user', 'password'):
+                    if fallback.get(key):
+                        command += ['--clickhouse-' + key, fallback[key]]
+                for arm in ('native', 'fallback'):
+                    endpoint = settings[arm]
+                    with measurements.phase(engine, arm, 'schema', [endpoint.get('pid')]):
+                        sql(endpoint, 'default', f'CREATE DATABASE {database}')
+                        databases.append(endpoint)
+                        sql(endpoint, database, f'CREATE TABLE {config["table"]} ({config["columns_sql"]}) ENGINE = MergeTree ORDER BY tuple()')
+            else:
+                command += ['--profile', 'asapquery', '--physical-plan', str(folder / 'install.json'),
+                            '--prometheus-server', fallback['url'], '--forward-unsupported-queries']
+                if engine == 'victoriametrics':
+                    command += ['--victoriametrics-http-port', str(query_port), '--victoriametrics-url', fallback['url']]
+            log = (folder / 'backend.log').open('w')
+            logs.append(log)
+            start = time.perf_counter_ns()
+            child = subprocess.Popen(command, stdout=log, stderr=log)
+            children.append(child)
+            wait_ready(api, child)
+            if engine == 'clickhouse':
+                post(api + '/api/v1/physical-plan', install)
+                envelope = install['precompute_plan']['envelope']
+                post(api + '/api/v1/physical-plan/activate', {k: envelope[k] for k in ('plan_id', 'plan_version')})
+            status = require(benefits.http(api + '/api/v1/physical-plan/status'))
+            save(folder / 'status.json', status)
+            envelope = install['precompute_plan']['envelope']
+            if not any(p['phase'] == 'active' and all(p[k] == envelope[k] for k in ('plan_id', 'plan_version')) for p in status['plans']):
+                raise RuntimeError('compiled generation was not activated')
+            startup = process_snapshot(child.pid)
+            measurements.rows.append({'engine': engine, 'arm': 'asap', 'phase': 'startup_install', 'complete': True,
+                 'wall_ns': time.perf_counter_ns() - start, 'cpu_ns': startup['cpu_ns'] if startup else None})
+            endpoints[engine] = {'native': dict(settings['native'], database=database, pids=[settings['native'].get('pid')]),
+                'asap': {'url': api if engine == 'prometheus' else f'http://127.0.0.1:{query_port}',
+                         'database': database, 'headers': fallback.get('headers', {}) if engine == 'clickhouse' else {},
+                         'pids': [child.pid, fallback.get('pid')]}}
+            deployments[engine] = {'api': api, 'install': install, 'pid': child.pid}
+        for index, batch in enumerate(config['batches']):
+            phase_name = 'build' if index == 0 else 'maintenance'
+            for engine in benefits.ENGINES:
+                settings, deployment = config['engines'][engine], deployments[engine]
+                with measurements.phase(engine, 'native', phase_name, [settings['native'].get('pid')]):
+                    if engine == 'clickhouse':
+                        load_sql(settings['native'], database, config['table'], batch['jsonl'])
+                    else:
+                        remote_write(settings['native'], batch['openmetrics'])
+                        if engine == 'victoriametrics':
+                            post(settings['native']['url'].rstrip('/') + '/internal/force_flush')
+                with measurements.phase(engine, 'asap', phase_name, [deployment['pid'], settings['fallback'].get('pid')]):
+                    if engine == 'clickhouse':
+                        load_sql(settings['fallback'], database, config['table'], batch['jsonl'])
+                    else:
+                        remote_write(settings['fallback'], batch['openmetrics'])
+                        if engine == 'victoriametrics':
+                            post(settings['fallback']['url'].rstrip('/') + '/internal/force_flush')
+                        remote_write({'url': deployment['api']}, batch['openmetrics'])
+                        if index == len(config['batches']) - 1:
+                            drained = post(deployment['api'] + '/api/v1/precompute/drain')
+                            if drained.get('complete') is not True:
+                                raise RuntimeError('finite materialization drain incomplete')
+                        else:
+                            # Final drain is the completion barrier; no mid-stream drain may seal input.
+                            time.sleep(.1)
+        # The current backfill admission boundary treats historical outputs as
+        # first-seen input. A later second job is rejected. Use one complete,
+        # non-overlapping finite backfill; do not weaken the live overlap guard.
+        deployment = deployments['clickhouse']
+        with measurements.phase('clickhouse', 'asap', 'materialization',
+                                [deployment['pid'], config['engines']['clickhouse']['fallback'].get('pid')]):
+            backfill(deployment['api'], deployment['install'], database, config['table'],
+                     config['batches'][0]['start_ms'], config['batches'][-1]['end_ms'])
+        manifest = json.loads(Path(config['manifest']).read_text())
+        with (output / 'requests.jsonl').open('w') as trace:
+            def record(row):
+                trace.write(json.dumps(row) + '\n')
+                trace.flush()
+            report = benefits.run(endpoints, manifest, config.get('repetitions', 10), config.get('timeout', 30),
+                                  config.get('relative_tolerance', 1e-9), config.get('absolute_tolerance', 1e-12), record_sink=record)
+        save(output / 'phases.json', measurements.rows)
+        from costs import summarize_costs
+        lifetime = {}
+        from costs import complete_sum
+        for engine in benefits.ENGINES:
+            settings = config['engines'][engine]
+            after = {arm: process_snapshot(settings[arm].get('pid')) for arm in ('native', 'fallback')}
+            delta = {arm: process_delta(lifetime_before[engine][arm], after[arm]) for arm in after}
+            backend = process_snapshot(deployments[engine]['pid'])
+            compiler_cpu = [p['compiler_cpu_ns'] for p in measurements.rows if p['engine'] == engine and p['phase'] == 'planning']
+            lifetime[engine] = {
+                'native_cpu_ns': (after['native']['cpu_ns'] if settings['native'].get('startup_cost') and after['native']
+                                  else delta['native']['cpu_ns'] if delta['native'] else None),
+                'asap_plus_fallback_and_planner_cpu_ns': complete_sum([
+                    backend['cpu_ns'] if backend else None,
+                    (after['fallback']['cpu_ns'] if settings['fallback'].get('startup_cost') and after['fallback']
+                     else delta['fallback']['cpu_ns'] if delta['fallback'] else None), *compiler_cpu]),
+                'final_processes': dict(after, backend=backend),
+                'backend_state_directory_bytes': sum(p.stat().st_size for p in (output / engine / 'state').rglob('*') if p.is_file()),
+                'native_storage_bytes': storage_bytes(settings['native'].get('storage_path')),
+                'fallback_storage_bytes': storage_bytes(settings['fallback'].get('storage_path')),
+            }
+        report['costs'] = summarize_costs(measurements.rows, report['records'], lifetime)
+        report['costs']['clickhouse']['cpu_break_even'] = {'refreshes': None, 'reason': 'SQL continuous summary maintenance is not measured; one finite backfill after loading all batches'}
+        save(output / 'report.json', report)
+        return report
+    except Exception as error:
+        save(output / 'failure.json', {'error': str(error), 'type': type(error).__name__})
+        raise
+    finally:
+        cleanup_errors = []
+        for child in reversed(children):
+            try:
+                stop(child)
+            except Exception as error:
+                cleanup_errors.append(str(error))
+        for log in logs:
+            log.close()
+        for endpoint in databases:
+            try:
+                sql(endpoint, 'default', f'DROP DATABASE {database}')
+            except Exception as error:
+                cleanup_errors.append(str(error))
+        save(output / 'cleanup.json', {'errors': cleanup_errors})
+        if cleanup_errors:
+            raise RuntimeError('experiment cleanup incomplete; inspect cleanup.json')
+
+
+def storage_bytes(path):
+    if path is None:
+        return None
+    try:
+        return sum(p.stat().st_size for p in Path(path).rglob('*') if p.is_file()) if Path(path).is_dir() else None
+    except OSError:
+        return None
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--config', type=Path, required=True)
+    parser.add_argument('--output', type=Path, required=True)
+    args = parser.parse_args()
+    report = run(json.loads(args.config.read_text()), args.output.resolve())
+    raise SystemExit(0 if report['passed'] else 1)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/tools/shared-workload/smoke.py b/tools/shared-workload/smoke.py
new file mode 100644
index 00000000..667293fe
--- /dev/null
+++ b/tools/shared-workload/smoke.py
@@ -0,0 +1,162 @@
+#!/usr/bin/env python3
+"""Provision isolated native/fallback containers and run the full comparison path."""
+import argparse
+import copy
+import json
+from pathlib import Path
+import shlex
+import subprocess
+import time
+import uuid
+
+import benefits
+import experiment
+
+
+def fixture(output, compiler, backend):
+    # Keep event times recent enough for native TSDB ingestion, and all SQL
+    # panes aligned. MetricsQL and PromQL are compared to their own references.
+    start = ((int(time.time()) - 60) // 4) * 4000 + 1000
+    query = 'sum_over_time(data[4s])'
+    queries = [{'name': 'temporal_sum', 'promql': query, 'metricsql': query,
+                'clickhouse_sql': f'SELECT sum(value) AS value FROM raw_samples WHERE ts_ms >= {start + 4000} AND ts_ms < {start + 8000}'}]
+    queries.append({'name': 'temporal_sum_plus_one', 'promql': query + ' + 1', 'metricsql': query + ' + 1',
+                    'clickhouse_sql': queries[0]['clickhouse_sql'].replace('sum(value)', 'sum(value) + 1')})
+    manifest = {'end_ms': start + 7000, 'queries': queries}
+    experiment.save(output / 'manifest.json', manifest)
+    batches = []
+    for index in range(2):
+        om, sql = output / f'batch-{index}.openmetrics', output / f'batch-{index}.jsonl'
+        samples = [(start + step * 1000, step + 1) for step in range(index * 4, (index + 1) * 4)]
+        om.write_text(''.join(f'data {value} {ts / 1000:.3f}\n' for ts, value in samples) + '# EOF\n')
+        sql.write_text(''.join(json.dumps({'ts_ms': ts, 'value': value}) + '\n' for ts, value in samples))
+        batches.append({'openmetrics': str(om), 'jsonl': str(sql), 'start_ms': start + index * 4000, 'end_ms': start + (index + 1) * 4000})
+    root = Path(__file__).resolve().parents[2]
+    template = json.loads((root / 'docs/examples/asapquery-planning-snapshot.json').read_text())
+    now = int(time.time() * 1000)
+    template['query_workload']['repeating_queries'] = [{
+        'query': query, 'demand': {'fixed_interval_at': {'interval': 4000, 'evaluation_phase': 0}},
+        'requirements': {'accuracy': 'implicit_exact', 'response_latency': 'unspecified'},
+        'predictability': {'predictable': {'known_at': None}},
+        'time_selection': {'scope': 'real_time', 'lookback': 4000, 'as_of': None}}]
+    extra = copy.deepcopy(template['query_workload']['repeating_queries'][0])
+    extra['query'] += ' + 1'
+    template['query_workload']['repeating_queries'].append(extra)
+    template['environment'].update(observed_at_unix_ms=now, activation_unix_ms=now, max_evidence_age_ms=600000)
+    implementation = template['implementation']
+    implementation.update(evidence_observed_at_unix_ms=now, evidence_valid_for_ms=600000,
+                          source_sample_interval_ms=1000, query_staleness_margin_ms=10000)
+    implementation['implementation_cost'].update(observed_at_unix_ms=now, valid_for_ms=600000)
+    # These declared costs are a functional-test seed, not measured calibration.
+    # The experiment reports observed execution costs independently of them.
+    template['implementation']['implementation_cost']['model_version'] = 'functional-smoke-declared-costs'
+    for engine in ('prometheus', 'victoriametrics'):
+        snapshot = copy.deepcopy(template)
+        # Canonical workload uses promql; compile_metricsql selects the native frontend.
+        snapshot['query_workload']['language'] = 'promql'
+        experiment.save(output / f'{engine}.json', snapshot)
+    # Read the pinned revision from Cargo.lock, the same source as build.rs.
+    import re
+    lock = (root / 'Cargo.lock').read_text()
+    revision = re.search(r'ASAPPlanner\?rev=[^#]+#([a-f0-9]{40})', lock).group(1)
+    sql = {'envelope': {'plan_id': 731, 'plan_version': 1, 'generated_at_unix_ms': now,
+                       'activation_unix_ms': now, 'expiry_unix_ms': None,
+                       'backend_compat': 'asap-query-backend.v1', 'planner_revision': revision,
+                       'capability_snapshot_id': 'comparison-functional-smoke'},
+           'tables': {'raw_samples': {'columns': [
+               {'name': 'ts_ms', 'dtype': 'int64', 'nullable': False, 'table': None},
+               {'name': 'value', 'dtype': 'float64', 'nullable': False, 'table': None}],
+               'time_index': 0, 'group_keys': []}}, 'accuracy': 'Exact',
+           'queries': [{'sql': q['clickhouse_sql'].replace(str(start + 4000), str(start)).replace(str(start + 8000), str(start + 4000)),
+                        'start_ms': start, 'end_ms': start + 4000, 'cumulative': False} for q in queries]}
+    experiment.save(output / 'clickhouse.json', sql)
+    # This query is deliberately absent from all three publications. Successful
+    # native fallback must remain distinct from the warm coverage above.
+    queries.append({'name': 'unplanned_window', 'promql': 'sum_over_time(data[3s])',
+                    'metricsql': 'sum_over_time(data[3s])',
+                    'clickhouse_sql': f'SELECT sum(value) AS value FROM raw_samples WHERE ts_ms >= {start + 5000} AND ts_ms < {start + 8000}'})
+    experiment.save(output / 'manifest.json', manifest)
+    return {'compiler': str(compiler), 'data_plane': str(backend), 'manifest': str(output / 'manifest.json'),
+            'table': 'raw_samples', 'columns_sql': 'ts_ms Int64, value Float64', 'batches': batches,
+            'repetitions': 5, 'engines': {e: {'planning_input': str(output / f'{e}.json')} for e in benefits.ENGINES}}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--docker', default='docker', help='Docker argv prefix, e.g. "sudo -n docker"')
+    parser.add_argument('--compiler', type=Path, required=True)
+    parser.add_argument('--data-plane', type=Path, required=True)
+    parser.add_argument('--output', type=Path, required=True)
+    for engine in benefits.ENGINES:
+        parser.add_argument('--' + engine + '-image', required=True, help='immutable local image ID or digest')
+    args = parser.parse_args()
+    output = args.output.resolve()
+    output.mkdir(parents=True, exist_ok=False)
+    config = fixture(output, args.compiler.resolve(), args.data_plane.resolve())
+    docker = shlex.split(args.docker)
+    names, inspections = [], []
+    prometheus_config = output / 'prometheus.yml'
+    prometheus_config.write_text('global:\n  scrape_interval: 1h\nscrape_configs: []\n')
+    try:
+        for engine in benefits.ENGINES:
+            image = getattr(args, engine + '_image')
+            # Resolve once; run this immutable ID, never a moving tag twice.
+            image_id = subprocess.check_output(docker + ['image', 'inspect', image, '--format', '{{.Id}}'], text=True).strip()
+            for arm in ('native', 'fallback'):
+                name = 'asap-benefits-' + uuid.uuid4().hex[:12]
+                internal = {'prometheus': 9090, 'clickhouse': 8123, 'victoriametrics': 8428}[engine]
+                command = docker + ['run', '-d', '--name', name, '--pull', 'never', '--cpus', '2', '--memory', '2g',
+                                    '-p', f'127.0.0.1::{internal}']
+                if engine == 'prometheus':
+                    command += ['-v', f'{prometheus_config}:/etc/prometheus/prometheus.yml:ro']
+                if engine == 'clickhouse':
+                    command += ['-e', 'CLICKHOUSE_USER=asap_test', '-e', 'CLICKHOUSE_PASSWORD=asap_test_local']
+                command += [image_id]
+                if engine == 'prometheus':
+                    command += ['--config.file=/etc/prometheus/prometheus.yml', '--web.enable-remote-write-receiver']
+                if engine == 'victoriametrics':
+                    command += ['-search.latencyOffset=0s']
+                started = time.perf_counter_ns()
+                subprocess.run(command, check=True, stdout=subprocess.DEVNULL)
+                names.append(name)
+                info = json.loads(subprocess.check_output(docker + ['inspect', name]))[0]
+                inspections.append(info)
+                mapped = info['NetworkSettings']['Ports'][f'{internal}/tcp'][0]['HostPort']
+                endpoint = {'url': f'http://127.0.0.1:{mapped}', 'pid': info['State']['Pid']}
+                if engine == 'clickhouse':
+                    endpoint.update(user='asap_test', password='asap_test_local')
+                config['engines'][engine][arm] = endpoint
+                path = '/ping' if engine == 'clickhouse' else ('/-/ready' if engine == 'prometheus' else '/health')
+                deadline = time.monotonic() + 60
+                while benefits.http(endpoint['url'] + path, timeout=1)['status'] != 200:
+                    if time.monotonic() >= deadline:
+                        raise TimeoutError(f'{engine} readiness timeout')
+                    time.sleep(.2)
+                process = experiment.process_snapshot(endpoint['pid'])
+                endpoint['startup_cost'] = {'wall_ns': time.perf_counter_ns() - started, 'cpu_ns': process['cpu_ns'] if process else None}
+        experiment.save(output / 'containers.json', inspections)
+        experiment.save(output / 'experiment.json', config)
+        report = experiment.run(config, output / 'run')
+        print(json.dumps({'passed': report['passed'], 'summary': report['summary']}, indent=2))
+        valid = report['passed']
+        for record in report['records']:
+            expected_route = 'exact_fallback' if record['name'] == 'unplanned_window' else 'warm'
+            expected_value = {'temporal_sum': 26, 'temporal_sum_plus_one': 27, 'unplanned_window': 21}[record['name']]
+            for engine, result in record['engines'].items():
+                valid &= result['execution'] == expected_route
+                for arm in ('native', 'asap'):
+                    body = result[arm]['body']
+                    values = [r['value'] for r in body['data']] if engine == 'clickhouse' else [float(r['value'][1]) for r in body['data']['result']]
+                    valid &= values == [expected_value]
+        experiment.save(output / 'acceptance.json', {'passed': bool(valid), 'expected': 'two warm queries and one unplanned fallback per engine, with known nonempty values'})
+        if not valid:
+            raise SystemExit('smoke requires correct values, warm supported queries, and explicit fallback for the unplanned query')
+    finally:
+        for name in reversed(names):
+            logs = subprocess.run(docker + ['logs', name], capture_output=True, text=True)
+            (output / f'{name}.log').write_text(logs.stdout + logs.stderr)
+            subprocess.run(docker + ['rm', '-f', name], stdout=subprocess.DEVNULL, check=False)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/tools/shared-workload/test_benefits.py b/tools/shared-workload/test_benefits.py
new file mode 100644
index 00000000..759163d5
--- /dev/null
+++ b/tools/shared-workload/test_benefits.py
@@ -0,0 +1,112 @@
+"""Endpoint failures and fallback must not manufacture a successful benchmark."""
+import unittest
+from unittest.mock import patch
+import benefits
+
+
+def response(value=1, route='warm'):
+    return {'status': 200, 'headers': {'x-asap-execution': route}, 'elapsed_ns': 10,
+            'body': {'status': 'success', 'data': {'resultType': 'vector', 'result':
+                     [{'metric': {}, 'value': [1, str(value)]}]}}}
+
+
+class ComparisonTests(unittest.TestCase):
+    def test_native_timeout_still_runs_all_six_endpoints(self):
+        calls = []
+        def request(engine, endpoint, query, evaluation_ms, timeout):
+            calls.append((engine, endpoint))
+            if engine == 'prometheus' and endpoint == 'native':
+                raise TimeoutError('native timed out')
+            return response()
+        with patch.object(benefits, 'query_endpoint', side_effect=request):
+            report = benefits.compare_round({e: {'native': 'native', 'asap': 'asap'} for e in benefits.ENGINES},
+                                            {'promql': 'm', 'metricsql': 'm', 'clickhouse_sql': 'SELECT 1'}, 1000, 1)
+        self.assertEqual(len(calls), 6)
+        self.assertFalse(report['prometheus']['passed'])
+        self.assertEqual(report['prometheus']['asap']['status'], 200)
+
+    def test_vm_mismatch_fails_even_when_other_engines_match(self):
+        def request(engine, endpoint, *_):
+            return response(99 if engine == 'victoriametrics' and endpoint == 'asap' else 1)
+        with patch.object(benefits, 'query_endpoint', side_effect=request):
+            report = benefits.compare_round({e: {'native': 'native', 'asap': 'asap'} for e in benefits.ENGINES},
+                                            {}, 1000, 1)
+        self.assertFalse(report['victoriametrics']['passed'])
+
+    def test_fallback_and_missing_provenance_are_not_acceleration(self):
+        for route in ['exact_fallback', 'unknown', 'hybrid']:
+            result = benefits.assess('prometheus', response(), response(route=route))
+            self.assertTrue(result['passed'])
+            self.assertFalse(result['accelerated'])
+
+    def test_numeric_tolerance_is_explicit(self):
+        self.assertFalse(benefits.assess('prometheus', response(100), response(101))['passed'])
+        accepted = benefits.assess('prometheus', response(100), response(101), relative=.02)
+        self.assertTrue(accepted['passed'])
+        self.assertEqual(accepted['comparison']['max_absolute_error'], 1)
+
+    def test_sql_metadata_and_values_are_checked(self):
+        native = dict(response(), body={'meta': [{'name': 'v', 'type': 'Float64'}], 'data': [{'v': 1}]})
+        asap = dict(native, body={'meta': [{'name': 'v', 'type': 'UInt64'}], 'data': [{'v': 1}]})
+        self.assertFalse(benefits.assess('clickhouse', native, asap)['passed'])
+
+
+
+class CostTests(unittest.TestCase):
+    def test_maintenance_can_eliminate_query_savings(self):
+        from costs import break_even
+        self.assertIsNone(break_even(100, 10, 5, 6)['refreshes'])
+        self.assertEqual(break_even(100, 10, 5, 1)['refreshes'], 25)
+
+    def test_missing_cpu_is_not_zero(self):
+        from costs import complete_sum, break_even
+        self.assertIsNone(complete_sum([1, None]))
+        self.assertIsNone(break_even(100, None, 5, 0)['refreshes'])
+
+class WorkflowTests(unittest.TestCase):
+    def test_backfill_splits_range_at_installed_pane_width(self):
+        import experiment
+        install = {'summary_catalog': {'materializations': {
+            '7': {'window_layout': {'pane_secs': 4}, 'pane_origin_ms': 1000}}}}
+        with patch.object(experiment, 'post') as post, patch.object(benefits, 'http', return_value={
+                'status': 200, 'body': {'jobs': [{'status': 'complete'}]}}):
+            experiment.backfill('http://backend', install, 'db', 'samples', 1000, 9000)
+        self.assertEqual(post.call_args.args[1]['windows_total'], 2)
+
+    def test_backfill_rejects_partial_pane_without_enqueuing(self):
+        import experiment
+        install = {'summary_catalog': {'materializations': {
+            '7': {'window_layout': {'pane_secs': 4}, 'pane_origin_ms': 1000}}}}
+        with patch.object(experiment, 'post') as post:
+            with self.assertRaises(ValueError):
+                experiment.backfill('http://backend', install, 'db', 'samples', 1000, 8000)
+            post.assert_not_called()
+
+    def test_native_and_fallback_cannot_be_the_same_instance(self):
+        import experiment
+        config = {'table': 'samples', 'batches': [{}], 'engines': {
+            e: {arm: {'url': 'http://same'} for arm in ('native', 'fallback')} for e in benefits.ENGINES}}
+        with self.assertRaises(ValueError):
+            experiment.validate(config)
+
+class RefreshAccountingTests(unittest.TestCase):
+    def test_break_even_counts_dashboard_refreshes_not_individual_queries(self):
+        from costs import summarize_costs
+        phases = []
+        for engine in benefits.ENGINES:
+            for arm, phase, cpu in [('native', 'build', 10), ('asap', 'build', 110), ('asap', 'maintenance', 5)]:
+                phases.append({'engine': engine, 'arm': arm, 'phase': phase, 'cpu_ns': cpu, 'wall_ns': cpu, 'complete': True})
+        records = [{'repeat': 0, 'engines': {engine: {
+            'passed': True, 'native': {'cpu_ns': 10, 'elapsed_ns': 10},
+            'asap': {'cpu_ns': 5, 'elapsed_ns': 5}} for engine in benefits.ENGINES}} for _ in range(2)]
+        result = summarize_costs(phases, records)
+        self.assertEqual(result['prometheus']['observed_refreshes'], 1)
+        self.assertEqual(result['prometheus']['cpu_break_even']['refreshes'], 20)
+        records[0]['engines']['victoriametrics']['passed'] = False
+        failed = summarize_costs(phases, records)['victoriametrics']
+        self.assertIsNone(failed['cpu_break_even']['refreshes'])
+        self.assertIsNone(failed['native_over_asap_phase_cpu_ratio'])
+
+
+if __name__ == '__main__':
+    unittest.main()