diff --git a/.github/workflows/mvp-ci.yml b/.github/workflows/mvp-ci.yml
index a2b9d4cb..a7b96003 100644
--- a/.github/workflows/mvp-ci.yml
+++ b/.github/workflows/mvp-ci.yml
@@ -16,7 +16,7 @@ jobs:
   backend:
     name: MVP backend
     runs-on: ubuntu-latest
-    timeout-minutes: 45
+    timeout-minutes: 60
     steps:
       - name: Checkout ASAPQuery-backend
         uses: actions/checkout@v4
@@ -78,3 +78,18 @@ jobs:
           CARGO_NET_GIT_FETCH_WITH_CLI: "true"
         working-directory: ASAPQuery-backend
         run: cargo test -p control_plane
+
+      # Compiling the data plane is not evidence that it runs: plan
+      # installation, maintenance execution, persistence recovery and query
+      # routing all live in these library tests, and none of them were gated
+      # before. `--lib` deliberately excludes `data_plane/tests/`, which needs
+      # live ClickHouse / Prometheus / VictoriaMetrics endpoints.
+      - name: Test shared types and data plane
+        env:
+          CARGO_NET_GIT_FETCH_WITH_CLI: "true"
+          # Several storage tests wait on background flusher / sealer threads.
+          # Hosted runners are small and noisy, so give those waits headroom;
+          # a passing wait still returns as soon as its condition holds.
+          ASAP_TEST_TIMEOUT_SCALE: "4"
+        working-directory: ASAPQuery-backend
+        run: cargo test -p asap_types -p data_plane --lib
diff --git a/README.md b/README.md
index 38196adb..283a0232 100644
--- a/README.md
+++ b/README.md
@@ -139,17 +139,20 @@ text and deployment hints:
   accuracy_sla: 0.99
   assign_to_role: agent
   grouping_labels: [region]
+  repeat_every: 30s
 ```
 
 `query_string` is the preferred source for metric, aggregation, filters,
 grouping and range-window semantics. Optional registry fields include
-`sketch_family_override`, `sample_p`, `distinct_keys_per_window`, `item_label`
-and `monitor`. `accuracy_sla` is the legacy success fraction: `1.0` requests
-exact results and `0.99` permits epsilon `0.01`.
-
-The startup registry does not currently carry dashboard recurrence. The legacy
-planning API accepts an evaluation cadence at `POST /api/v1/plan`
-(`CONTROLLER_ADDR`, default port `8080`):
+`sketch_family_override`, `sample_p`, `distinct_keys_per_window`, `item_label`,
+`monitor` and `repeat_every`. `accuracy_sla` is the legacy success fraction:
+`1.0` requests exact results and `0.99` permits epsilon `0.01`. An unknown key
+is rejected rather than ignored, and a registry file that exists but does not
+parse fails startup instead of degrading to an empty registry.
+
+The startup registry and the legacy planning API carry the same evaluation
+cadence, so a declaration is costed identically through either entry point.
+`POST /api/v1/plan` (`CONTROLLER_ADDR`, default port `8080`):
 
 ```bash
 curl -X POST http://localhost:8080/api/v1/plan \
diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs
index c7f9be46..12937c01 100644
--- a/control_plane/src/clickhouse.rs
+++ b/control_plane/src/clickhouse.rs
@@ -358,8 +358,15 @@ fn materialize_selected_sql(
     }
     let grouping = asap_types::GroupingProjection::new(columns);
     grouping.validate_table_group_codec()?;
-    let (table, value, window, population, timestamp) =
-        clickhouse_materialization_leaf_contract(node, query.start_ms, query.end_ms)?;
+    let leaf = clickhouse_materialization_leaf_contract(node, query.start_ms, query.end_ms)?;
+    let ClickHouseMaterializationLeaf {
+        table,
+        value,
+        value_source_column,
+        window_secs: window,
+        population,
+        timestamp_column: timestamp,
+    } = leaf;
     let window_secs = window.ok_or("SQL materialization requires a bounded window")?;
     let aggregation = BackendAggregation {
         aggregation_id: String::new(),
@@ -382,6 +389,7 @@ fn materialize_selected_sql(
     config.value_projection = Some(value);
     config.table_timestamp_column = Some(timestamp);
     config.table_population = Some(population);
+    config.value_source_column = value_source_column;
     config.partitioning = Some(asap_types::sds::PopulationPartitioning::Grouped);
     config.pane_origin_ms = Some(
         i64::try_from(query.start_ms)
@@ -567,9 +575,15 @@ fn bind_selected_node(
     query: &ClickHouseSqlWorkloadEntry,
     request: &ClickHouseSqlWorkload,
 ) -> Result<MaterializationBinding, crate::query_plan::QueryPlanError> {
-    let (table_ref, value_column, source_window, spatial_filter, timestamp_column) =
-        clickhouse_materialization_leaf_contract(node, query.start_ms, query.end_ms)
-            .map_err(crate::query_plan::QueryPlanError::Invalid)?;
+    let ClickHouseMaterializationLeaf {
+        table: table_ref,
+        value: value_column,
+        window_secs: source_window,
+        population: spatial_filter,
+        timestamp_column,
+        ..
+    } = clickhouse_materialization_leaf_contract(node, query.start_ms, query.end_ms)
+        .map_err(crate::query_plan::QueryPlanError::Invalid)?;
     let expected = crate::physical::compiler::physical_materialization_family(family);
     let selected = select_materialization(
         &request.precompute_plan.materializations,
@@ -614,20 +628,25 @@ fn constant_int64(expr: &QueryExpr) -> Option<i64> {
     }
 }
 
+/// The table leaf a SQL summary materialization is admitted against.
+#[derive(Debug)]
+struct ClickHouseMaterializationLeaf {
+    table: String,
+    value: asap_types::sds::ValueProjectionIdentity,
+    /// Producer typing for a column projection: what the ingest path must know
+    /// to read the column safely (integer exactness, NULL skipping). `None`
+    /// for constant projections, which carry their own literal.
+    value_source_column: Option<planner_types::pre_asap::Column>,
+    window_secs: Option<u64>,
+    population: asap_types::table_population::TablePopulation,
+    timestamp_column: String,
+}
+
 fn clickhouse_materialization_leaf_contract(
     node: &planner_types::post_asap::SummaryNode,
     evaluation_start_ms: u64,
     evaluation_end_ms: u64,
-) -> Result<
-    (
-        String,
-        asap_types::sds::ValueProjectionIdentity,
-        Option<u64>,
-        asap_types::table_population::TablePopulation,
-        String,
-    ),
-    String,
-> {
+) -> Result<ClickHouseMaterializationLeaf, String> {
     use planner_types::{
         post_asap::SummaryExpr,
         pre_asap::{CompareOpKind, QueryExpr, ScalarValue, Source},
@@ -676,6 +695,7 @@ fn clickhouse_materialization_leaf_contract(
     };
     use asap_types::sds::ValueProjectionIdentity;
     use planner_types::post_asap::SummaryInputExpr;
+    let mut value_source_column = None;
     let value_projection = match &input.weight {
         SummaryInputExpr::Column(
             planner_types::pre_asap::ColumnRef::Wildcard
@@ -701,9 +721,24 @@ fn clickhouse_materialization_leaf_contract(
                 .iter()
                 .find(|column| column.name == *name)
                 .ok_or("SQL summary value projection is not a source column")?;
-            if column.nullable || column.dtype != planner_types::pre_asap::DataType::Float64 {
-                return Err("SQL value readout requires a non-null Float64 source until typed/null-aware ingest is available".into());
+            // Numeric source columns are admitted with their declared type and
+            // nullability, which the ingest reader honours: integers get an
+            // exactness guard on the way into f64 summary state, and NULL rows
+            // are skipped the way a SQL aggregate skips them. Non-numeric
+            // columns have no value semantics to summarise and stay refused.
+            if !matches!(
+                column.dtype,
+                planner_types::pre_asap::DataType::Float64
+                    | planner_types::pre_asap::DataType::Int64
+            ) {
+                return Err(format!(
+                    "SQL value readout requires a numeric source column; `{name}` is {:?}",
+                    column.dtype
+                ));
             }
+            let mut source_column = column.clone();
+            source_column.table = None;
+            value_source_column = Some(source_column);
             ValueProjectionIdentity::Column { name: name.clone() }
         }
         SummaryInputExpr::Constant(value) if value.is_finite() => {
@@ -827,18 +862,19 @@ fn clickhouse_materialization_leaf_contract(
     let window_secs = explicit_window.or(inferred_window).ok_or_else(|| {
         "SQL table summary requires a positive whole-second timestamp range".to_string()
     })?;
-    Ok((
-        table_ref.to_owned(),
-        value_projection,
-        Some(window_secs),
+    Ok(ClickHouseMaterializationLeaf {
+        table: table_ref.to_owned(),
+        value: value_projection,
+        value_source_column,
+        window_secs: Some(window_secs),
         population,
-        schema
+        timestamp_column: schema
             .time_index
             .and_then(|index| schema.columns.get(index))
             .ok_or("SQL summary source has no timestamp projection")?
             .name
             .clone(),
-    ))
+    })
 }
 
 fn select_materialization<'a>(
@@ -993,6 +1029,133 @@ mod tests {
         value
     }
 
+    /// A table leaf whose value column has the given producer typing.
+    fn typed_value_leaf(
+        dtype: planner_types::pre_asap::DataType,
+        nullable: bool,
+    ) -> planner_types::post_asap::SummaryNode {
+        use planner_types::post_asap::{
+            SummaryExpr, SummaryInputExpr, SummaryNode, SummarySchema, SummaryUpdate,
+        };
+        use planner_types::pre_asap::{
+            Column, ColumnRef, CompareOpKind, DataType, Predicate, QueryExpr, Reduction,
+            ScalarValue, Schema, Source,
+        };
+        let schema = Schema::with_time_index(
+            vec![
+                Column::new("timestamp_ms", DataType::Timestamp, false),
+                Column::new("value", dtype, nullable),
+            ],
+            0,
+            Vec::new(),
+        );
+        let bound = |op: CompareOpKind, at: i64| {
+            Predicate(std::rc::Rc::new(QueryExpr::Compare {
+                left: std::rc::Rc::new(QueryExpr::Column(0)),
+                op,
+                right: std::rc::Rc::new(QueryExpr::Literal(ScalarValue::Int64(at))),
+            }))
+        };
+        let scan = QueryExpr::Scan {
+            source: Source::Table {
+                table_ref: "telemetry".into(),
+            },
+            predicates: vec![
+                bound(CompareOpKind::Ge, 0),
+                bound(CompareOpKind::Lt, 60_000),
+            ],
+            schema,
+        };
+        let family = materialization(
+            AggregationType::Sum,
+            "value",
+            60,
+            60,
+            ("variant", serde_json::json!(1)),
+        )
+        .accumulator_spec()
+        .unwrap()
+        .family;
+        let summary_schema = SummarySchema {
+            fields: vec![],
+            time_index: None,
+        };
+        SummaryNode {
+            expr: SummaryExpr::SummaryAgg {
+                child: std::rc::Rc::new(SummaryNode {
+                    expr: SummaryExpr::KeepPreAsap(std::rc::Rc::new(scan)),
+                    schema: summary_schema.clone(),
+                    guarantee: Default::default(),
+                }),
+                family,
+                input: SummaryUpdate {
+                    item: None,
+                    weight: SummaryInputExpr::Column(ColumnRef::Named("value".into())),
+                    weight_domain: Default::default(),
+                },
+                reduction: Reduction::Reduce(vec![].into()),
+                grouping: Default::default(),
+            },
+            schema: summary_schema,
+            guarantee: Default::default(),
+        }
+    }
+
+    /// Numeric source columns are admitted with their producer typing, which
+    /// the ingest reader needs to read them safely. Before this the contract
+    /// took non-null `Float64` only, so an ordinary nullable or integer
+    /// ClickHouse column could not be automatically materialized at all.
+    #[test]
+    fn numeric_value_columns_are_admitted_with_their_producer_typing() {
+        use planner_types::pre_asap::DataType;
+        for (dtype, nullable) in [
+            (DataType::Float64, false),
+            (DataType::Float64, true),
+            (DataType::Int64, false),
+            (DataType::Int64, true),
+        ] {
+            let leaf = clickhouse_materialization_leaf_contract(
+                &typed_value_leaf(dtype.clone(), nullable),
+                0,
+                60_000,
+            )
+            .unwrap_or_else(|error| panic!("{dtype:?}/{nullable}: {error}"));
+            assert_eq!(
+                leaf.value,
+                asap_types::sds::ValueProjectionIdentity::Column {
+                    name: "value".into()
+                }
+            );
+            let column = leaf
+                .value_source_column
+                .expect("a column projection carries its producer typing");
+            assert_eq!(column.dtype, dtype);
+            assert_eq!(column.nullable, nullable);
+            // Typing is a read concern, not an identity one: the same column
+            // is the same policy however it is declared.
+            assert_eq!(column.table, None);
+        }
+    }
+
+    /// Non-numeric columns have no value semantics to summarise. Refusing them
+    /// protects the result; it is not a gap to be widened.
+    #[test]
+    fn non_numeric_value_columns_stay_refused() {
+        use planner_types::pre_asap::DataType;
+        for dtype in [DataType::Utf8, DataType::Bool, DataType::Timestamp] {
+            let error = clickhouse_materialization_leaf_contract(
+                &typed_value_leaf(dtype.clone(), false),
+                0,
+                60_000,
+            )
+            .unwrap_err();
+            assert!(
+                error.contains("numeric source column"),
+                "{dtype:?}: {error}"
+            );
+        }
+    }
+
     #[test]
     fn keyed_summary_input_is_not_replaced_by_its_unit_weight() {
         use planner_types::post_asap::{
@@ -1427,12 +1590,21 @@ mod tests {
                 })
                 .collect(),
         };
+        // An Int64 source is admitted and carries its declared type into the
+        // materialization, which is what lets the ingest reader widen the
+        // column explicitly and fail loudly on a value beyond the exact
+        // Float64 range instead of silently summarising a rounded one.
         integer_source.tables.get_mut("telemetry").unwrap().columns[1].dtype = DataType::Int64;
-        assert!(
-            compile_automatic_clickhouse_workload(&integer_source)
-                .await
-                .is_err(),
-            "an Int64 source may contain values beyond exact Float64 ingest range"
+        let (integer_plan, _) = compile_automatic_clickhouse_workload(&integer_source)
+            .await
+            .expect("an Int64 source is admitted with its producer typing");
+        assert_eq!(
+            integer_plan.precompute_plan.materializations[0]
+                .value_source_column
+                .as_ref()
+                .expect("column projections carry their producer typing")
+                .dtype,
+            DataType::Int64
         );
         integer_source.tables.get_mut("telemetry").unwrap().columns[1].dtype = DataType::Float64;
         integer_source.queries[0].sql = integer_source.queries[0].sql.replace(
diff --git a/control_plane/src/emit/mod.rs b/control_plane/src/emit/mod.rs
index 064eaa7b..e930a720 100644
--- a/control_plane/src/emit/mod.rs
+++ b/control_plane/src/emit/mod.rs
@@ -909,38 +909,11 @@ mod runtime_tests {
     /// Mimics the pre-population loop in `main()` — turns each
     /// `WorkloadEntry` into a `QueryWorkload` via the shared `Analyzer`.
     fn populate_store_from_registry(registry: &WorkloadRegistry, store: &WorkloadStore) {
-        use crate::pipeline::{Analyzer, QuerySpec};
+        use crate::pipeline::Analyzer;
         use crate::types;
-        use crate::types_v2;
         let analyzer = Analyzer::new();
         for entry in registry.entries() {
-            // Mirrors main.rs's QuerySpec construction post-B3/B4:
-            // thread grouping_labels into group_by_labels; let the
-            // parser drive time_window when query_string is present.
-            let spec = QuerySpec {
-                query_string: entry.query_string.clone(),
-                metric_name: entry.metric_name.clone(),
-                label_filters: Default::default(),
-                group_by_labels: entry.grouping_labels.clone(),
-                aggregations: vec!["quantile".into()],
-                time_window: if entry.query_string.is_some() {
-                    String::new()
-                } else {
-                    "5m".into()
-                },
-                repeat_every: None,
-                accuracy_sla: entry.accuracy_sla,
-                latency_sla: None,
-                sketch_type: entry.sketch_family_override.clone(),
-                workload: types::WorkloadCharacteristics::default(),
-                id: None,
-                language: None,
-                accuracy: None,
-                dollars: None,
-                deployment_model: None,
-                shape: types_v2::QueryShape::default(),
-                data: types_v2::DataShape::default(),
-            };
+            let spec = crate::workload::query_spec_for_entry(entry);
             if let Ok(wl) = analyzer.analyze(spec) {
                 let role = crate::workload::derive_agg_role(entry);
                 store.set(
diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs
index 0beeb2ed..88570956 100644
--- a/control_plane/src/main.rs
+++ b/control_plane/src/main.rs
@@ -284,7 +284,18 @@ async fn main() {
     // ── Declarative workload registry ────────────────────────────────────────
     let workloads_path =
         std::env::var("CONTROLLER_WORKLOADS").unwrap_or_else(|_| "workloads.yaml".into());
-    let workload_registry = Arc::new(WorkloadRegistry::load(&workloads_path));
+    // A registry file that exists but does not parse is a startup failure, not
+    // an empty registry: every later step (plan pre-population, agent config
+    // push, cost accounting) would otherwise look exactly like a deployment
+    // that declared no workloads at all.
+    let workload_registry = match WorkloadRegistry::try_load(&workloads_path) {
+        Ok(registry) => Arc::new(registry),
+        Err(error) => {
+            tracing::error!(path = %workloads_path, %error, "unusable workload registry");
+            eprintln!("unusable workload registry at {workloads_path}: {error}");
+            std::process::exit(1);
+        }
+    };
 
     // Pre-populate PlanStore from the registry so agents get a config immediately.
     //
@@ -302,56 +313,10 @@ async fn main() {
     {
         let analyzer = Analyzer::new();
         for entry in workload_registry.entries() {
-            // MVP blocker B4 — let the analyzer parse `time_window` from
-            // `query_string` (matrix-selector `[range]`) instead of
-            // forcing a hardcoded "5m" default that overrides whatever
-            // the user wrote. The analyzer falls back to its own 5m
-            // default when the PromQL has no matrix selector (e.g.
-            // `count(unique_users_per_min)`), so this is strictly an
-            // improvement for queries that DO carry an explicit range.
-            // Empty string here means "no override; trust the parsed
-            // value or the analyzer's fallback".
-            //
-            // MVP blocker B3 — thread the WorkloadEntry's declarative
-            // `grouping_labels` into `QuerySpec.group_by_labels`. The
-            // analyzer merges these with any `by (...)` keys the
-            // PromQL parser surfaces, populating `QueryWorkload.
-            // group_by_labels`, which `collect_metric_to_grouping_labels`
-            // then drops into `EdgeStageConfig.metric_to_grouping_labels`
-            // so the agent's `keep_keys(datapoint.attributes, [...])`
-            // OTTL processor strips wire attrs down to this list
-            // BEFORE sketching.
-            let spec = pipeline::QuerySpec {
-                query_string: entry.query_string.clone(),
-                metric_name: entry.metric_name.clone(),
-                label_filters: Default::default(),
-                group_by_labels: entry.grouping_labels.clone(),
-                aggregations: vec!["quantile".into()],
-                // Empty when the entry HAS a `query_string` (the parser
-                // surfaces the matrix-selector range or its own 5m
-                // fallback). For entries without a query_string we
-                // can't trust the parser, so fall back to the
-                // historical 5m default so the analyzer doesn't error
-                // out at Step 4.
-                time_window: if entry.query_string.is_some() {
-                    String::new()
-                } else {
-                    "5m".into()
-                },
-                repeat_every: None,
-                accuracy_sla: entry.accuracy_sla,
-                latency_sla: None,
-                sketch_type: entry.sketch_family_override.clone(),
-                workload: types::WorkloadCharacteristics::default(),
-                // design.md alignment: defaults preserve legacy behaviour.
-                id: None,
-                language: None,
-                accuracy: None,
-                dollars: None,
-                deployment_model: None,
-                shape: types_v2::QueryShape::default(),
-                data: types_v2::DataShape::default(),
-            };
+            // One conversion, shared with every other caller: see
+            // `control_plane::workload::query_spec_for_entry` for the B3/B4
+            // field notes and for why the cadence must come from the entry.
+            let spec = control_plane::workload::query_spec_for_entry(entry);
             // B2 full restructure — derive the AggRole for this entry
             // BEFORE store insertion so collisions on metric name don't
             // overwrite a prior role's entry. The pre-B2 loop wrote
@@ -1222,6 +1187,7 @@ async fn handle_plan(State(st): State<AppState>, Json(spec): Json<QuerySpec>) ->
             item_label: None,
             // Role derivation does not depend on monitoring.
             monitor: None,
+            repeat_every: None,
         };
         control_plane::workload::derive_agg_role(&entry)
     };
@@ -1561,6 +1527,7 @@ async fn handle_plan_auto(
                 distinct_keys_per_window: None,
                 item_label: None,
                 monitor,
+                repeat_every: None,
             };
             let role = derive_agg_role(&entry);
             // (1) Inject the monitor unconditionally — only needs the metric name;
@@ -3719,26 +3686,7 @@ mod api_tests {
 
         let analyzer = Analyzer::new();
         for entry in registry.entries() {
-            let spec = pipeline::QuerySpec {
-                query_string: entry.query_string.clone(),
-                metric_name: entry.metric_name.clone(),
-                label_filters: Default::default(),
-                group_by_labels: vec![],
-                aggregations: vec!["quantile".into()],
-                time_window: "5m".into(),
-                repeat_every: None,
-                accuracy_sla: entry.accuracy_sla,
-                latency_sla: None,
-                sketch_type: entry.sketch_family_override.clone(),
-                workload: types::WorkloadCharacteristics::default(),
-                id: None,
-                language: None,
-                accuracy: None,
-                dollars: None,
-                deployment_model: None,
-                shape: types_v2::QueryShape::default(),
-                data: types_v2::DataShape::default(),
-            };
+            let spec = control_plane::workload::query_spec_for_entry(entry);
             if let Ok(wl) = analyzer.analyze(spec) {
                 let wc = types::WorkloadCharacteristics::default();
                 let plan = state.planner.plan(&wl, Some(&wc));
diff --git a/control_plane/src/store/workload.rs b/control_plane/src/store/workload.rs
index 346eec71..0799e864 100644
--- a/control_plane/src/store/workload.rs
+++ b/control_plane/src/store/workload.rs
@@ -270,6 +270,7 @@ mod tests {
                 distinct_keys_per_window: None,
                 item_label: None,
                 monitor: None,
+                repeat_every: None,
             },
             WorkloadEntry {
                 metric_name: "http_requests_total".into(),
@@ -283,6 +284,7 @@ mod tests {
                 distinct_keys_per_window: None,
                 item_label: None,
                 monitor: None,
+                repeat_every: None,
             },
             WorkloadEntry {
                 metric_name: "http_requests_total".into(),
@@ -296,6 +298,7 @@ mod tests {
                 distinct_keys_per_window: None,
                 item_label: None,
                 monitor: None,
+                repeat_every: None,
             },
         ];
 
diff --git a/control_plane/src/workload.rs b/control_plane/src/workload.rs
index 29af1072..16f9808f 100644
--- a/control_plane/src/workload.rs
+++ b/control_plane/src/workload.rs
@@ -214,7 +214,14 @@ fn collect_agg_intents(expr: &planner_types::pre_asap::QueryExpr, out: &mut Vec<
 }
 
 /// A single workload entry from the workloads YAML file.
+///
+/// `deny_unknown_fields`: a misspelled or unsupported key is a planning input
+/// the controller cannot honour. Accepting it silently would let the operator
+/// believe a declared cadence / hint reached the planner when it never left the
+/// YAML, so the registry rejects the file instead (see
+/// [`WorkloadRegistry::try_load`]).
 #[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct WorkloadEntry {
     /// Metric name this workload targets (e.g. `http_request_duration_seconds`).
     pub metric_name: String,
@@ -344,6 +351,80 @@ pub struct WorkloadEntry {
     /// future, the edge `threshold:` block. `None` / missing ⇒ no monitor.
     #[serde(default)]
     pub monitor: Option<MonitorDecl>,
+
+    /// Optional evaluation cadence for this declared query class, in the same
+    /// `1h5m30s` spelling [`crate::pipeline::parse_duration`] accepts.
+    ///
+    /// Why this exists: the cadence is a **cost input**, not a scheduler knob.
+    /// [`crate::physical::deployment_cost::delta`] uses it as the batch-mode
+    /// flush-period proxy, so the same declaration costed through the HTTP
+    /// `POST /api/v1/plan` path (whose [`crate::pipeline::QuerySpec`] has
+    /// carried `repeat_every` all along) and through this YAML registry used
+    /// to reach the planner with *different* flush periods — the startup path
+    /// hardcoded `None`. Threaded into `QuerySpec::repeat_every` by the
+    /// registry pre-pop loop in `main`, which the analyzer parses into
+    /// [`crate::types::QueryWorkload::repeat_every`].
+    ///
+    /// `None` / missing ⇒ unchanged behaviour (the cost model falls back to
+    /// its window-derived flush rate).
+    #[serde(default)]
+    pub repeat_every: Option<String>,
+}
+
+/// Build the planning [`crate::pipeline::QuerySpec`] a declarative registry
+/// entry describes.
+///
+/// **Why this exists**: the YAML → planner conversion used to be an inline
+/// struct literal, hand-copied into the controller's startup pre-population
+/// loop and into the test helpers that claim to mimic it. Every field added to
+/// [`WorkloadEntry`] then had to be re-threaded in each copy, and
+/// `repeat_every` is the field that proves the cost: the HTTP
+/// `POST /api/v1/plan` path carried the declared cadence into
+/// [`crate::types::QueryWorkload::repeat_every`] while the startup path pinned
+/// `None`, so the same declaration was costed with two different batch-mode
+/// flush periods depending on which entry point registered it.
+///
+/// Field notes preserved from the original loop:
+/// * MVP blocker B4 — `time_window` is left EMPTY when the entry carries a
+///   `query_string` so the analyzer parses the matrix-selector `[range]`
+///   instead of a hardcoded `5m` overriding what the operator wrote. Entries
+///   without a query string keep the historical `5m` default, because the
+///   parser has nothing to read and the analyzer errors out without one.
+/// * MVP blocker B3 — `grouping_labels` is threaded into `group_by_labels`.
+///   The analyzer merges it with any `by (...)` keys the PromQL parser
+///   surfaces, which `collect_metric_to_grouping_labels` drops into
+///   `EdgeStageConfig::metric_to_grouping_labels` so the agent's
+///   `keep_keys(datapoint.attributes, [...])` OTTL processor strips wire attrs
+///   down to this list BEFORE sketching.
+/// * `sketch_family_override` is threaded into `QuerySpec::sketch_type`, which
+///   populates `QueryWorkload::sketch_type_override` — what `bind_workload_typed`
+///   reads to honour the MVP §46 HLL / CountSketch / CountMinSketch pins.
+pub fn query_spec_for_entry(entry: &WorkloadEntry) -> crate::pipeline::QuerySpec {
+    crate::pipeline::QuerySpec {
+        query_string: entry.query_string.clone(),
+        metric_name: entry.metric_name.clone(),
+        label_filters: Default::default(),
+        group_by_labels: entry.grouping_labels.clone(),
+        aggregations: vec!["quantile".into()],
+        time_window: if entry.query_string.is_some() {
+            String::new()
+        } else {
+            "5m".into()
+        },
+        repeat_every: entry.repeat_every.clone(),
+        accuracy_sla: entry.accuracy_sla,
+        latency_sla: None,
+        sketch_type: entry.sketch_family_override.clone(),
+        workload: crate::types::WorkloadCharacteristics::default(),
+        // design.md alignment: defaults preserve legacy behaviour.
+        id: None,
+        language: None,
+        accuracy: None,
+        dollars: None,
+        deployment_model: None,
+        shape: crate::types_v2::QueryShape::default(),
+        data: crate::types_v2::DataShape::default(),
+    }
 }
 
 /// User-facing continuous-monitoring declaration on a [`WorkloadEntry`]. τ/ε and
@@ -471,34 +552,47 @@ impl WorkloadRegistry {
     }
 
     /// Load from a YAML file. Returns an empty registry on any error.
+    ///
+    /// Prefer [`Self::try_load`] on the startup path: a registry that parses
+    /// into nothing is indistinguishable, at every later step, from an
+    /// operator who declared no workloads at all.
     pub fn load(path: &str) -> Self {
-        match std::fs::read_to_string(path) {
-            Ok(contents) => match serde_yaml::from_str::<Vec<WorkloadEntry>>(&contents) {
-                Ok(entries) => {
-                    info!(path, count = entries.len(), "loaded workload registry");
-                    Self {
-                        entries,
-                        runtime: Default::default(),
-                    }
-                }
-                Err(e) => {
-                    warn!(path, error = %e, "invalid workloads YAML; using empty registry");
-                    Self {
-                        entries: vec![],
-                        runtime: Default::default(),
-                    }
-                }
-            },
-            Err(_) => {
-                info!(path, "workloads file not found; using empty registry");
-                Self {
-                    entries: vec![],
-                    runtime: Default::default(),
-                }
+        match Self::try_load(path) {
+            Ok(registry) => registry,
+            Err(error) => {
+                warn!(path, error = %error, "invalid workloads YAML; using empty registry");
+                Self::empty()
             }
         }
     }
 
+    /// Load from a YAML file, reporting an unusable file instead of degrading
+    /// to an empty registry.
+    ///
+    /// A **missing** file stays non-fatal — the controller is expected to run
+    /// without a declarative registry (the process e2e tests boot it with
+    /// `CONTROLLER_WORKLOADS` pointing at a path that does not exist). A file
+    /// that exists but does not parse is fatal: it carries planning input the
+    /// operator wrote down, and silently planning *nothing* from it has the
+    /// same observable shape as a controller that planned everything.
+    pub fn try_load(path: &str) -> Result<Self, String> {
+        let contents = match std::fs::read_to_string(path) {
+            Ok(contents) => contents,
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+                info!(path, "workloads file not found; using empty registry");
+                return Ok(Self::empty());
+            }
+            Err(error) => return Err(format!("cannot read workload registry: {error}")),
+        };
+        let entries = serde_yaml::from_str::<Vec<WorkloadEntry>>(&contents)
+            .map_err(|error| format!("cannot parse workload registry: {error}"))?;
+        info!(path, count = entries.len(), "loaded workload registry");
+        Ok(Self {
+            entries,
+            runtime: Default::default(),
+        })
+    }
+
     /// Create an empty registry (no file).
     pub fn empty() -> Self {
         Self {
@@ -543,6 +637,97 @@ impl WorkloadRegistry {
 mod tests {
     use super::*;
 
+    /// The declared cadence is a planning cost input, so it has to survive the
+    /// YAML entry point exactly as it survives `POST /api/v1/plan`.
+    #[test]
+    fn yaml_cadence_reaches_the_planning_workload() {
+        let yaml = r#"
+- metric_name: http_requests_total
+  query_string: "sum by (zone) (rate(http_requests_total[5m]))"
+  accuracy_sla: 0.99
+  repeat_every: 30s
+- metric_name: http_errors_total
+  query_string: "sum by (zone) (rate(http_errors_total[5m]))"
+  accuracy_sla: 0.99
+"#;
+        let entries: Vec<WorkloadEntry> = serde_yaml::from_str(yaml).unwrap();
+        assert_eq!(entries[0].repeat_every.as_deref(), Some("30s"));
+        assert_eq!(entries[1].repeat_every, None);
+
+        let analyzer = crate::pipeline::Analyzer::new();
+        let declared = analyzer.analyze(query_spec_for_entry(&entries[0])).unwrap();
+        assert_eq!(
+            declared.repeat_every,
+            Some(std::time::Duration::from_secs(30))
+        );
+        // An entry that declares no cadence keeps the historical `None`, so the
+        // cost model falls back to its window-derived flush rate.
+        let undeclared = analyzer.analyze(query_spec_for_entry(&entries[1])).unwrap();
+        assert_eq!(undeclared.repeat_every, None);
+    }
+
+    /// A cadence the duration parser cannot read is a declaration error, not a
+    /// silently dropped field.
+    #[test]
+    fn unparsable_cadence_fails_the_entry() {
+        let entry = WorkloadEntry {
+            metric_name: "http_requests_total".into(),
+            query_string: Some("sum(http_requests_total)".into()),
+            accuracy_sla: 0.99,
+            assign_to_role: "agent".into(),
+            sketch_family_override: None,
+            target_path: None,
+            grouping_labels: vec![],
+            sample_p: 1.0,
+            distinct_keys_per_window: None,
+            item_label: None,
+            monitor: None,
+            repeat_every: Some("every 30 seconds".into()),
+        };
+        let error = crate::pipeline::Analyzer::new()
+            .analyze(query_spec_for_entry(&entry))
+            .expect_err("unparsable cadence must not plan");
+        assert!(
+            format!("{error:#}").contains("repeat_every"),
+            "unexpected error: {error:#}"
+        );
+    }
+
+    /// An unsupported key is planning input the controller cannot honour;
+    /// accepting the file would report a cadence / hint that never left the YAML.
+    #[test]
+    fn unknown_registry_key_is_rejected() {
+        let path = std::env::temp_dir().join(format!(
+            "asap_workload_unknown_key_{}.yaml",
+            std::process::id()
+        ));
+        std::fs::write(
+            &path,
+            "- metric_name: http_requests_total
+  accuracy_sla: 0.99
+  repeat_evry: 30s
+",
+        )
+        .unwrap();
+        let error = WorkloadRegistry::try_load(path.to_str().unwrap())
+            .expect_err("unknown key must not load");
+        assert!(error.contains("repeat_evry"), "unexpected error: {error}");
+        // The lenient wrapper still degrades, which is why startup uses `try_load`.
+        assert!(WorkloadRegistry::load(path.to_str().unwrap())
+            .entries()
+            .is_empty());
+        std::fs::remove_file(&path).unwrap();
+    }
+
+    /// Running without a declarative registry stays supported: the process e2e
+    /// tests boot the controller with `CONTROLLER_WORKLOADS` pointing nowhere.
+    #[test]
+    fn missing_registry_file_is_not_a_startup_failure() {
+        let registry = WorkloadRegistry::try_load("/definitely/missing/workloads.yaml")
+            .expect("a missing registry is not an error");
+        assert!(registry.entries().is_empty());
+    }
+
     #[test]
     fn load_empty_on_missing_file() {
         let reg = WorkloadRegistry::load("/nonexistent/workloads.yaml");
@@ -617,6 +802,7 @@ mod tests {
                 epsilon: 0.05,
                 window_secs: 30,
             }),
+            repeat_every: None,
         });
         let intents = reg.monitor_intents("dp:4319");
         assert_eq!(intents.len(), 1);
@@ -644,6 +830,7 @@ mod tests {
                 epsilon: 0.05,
                 window_secs: 30,
             }),
+            repeat_every: None,
         });
         let intents = reg.monitor_intents("dp:4319");
         assert_eq!(intents.len(), 1, "replaced, not duplicated");
@@ -685,6 +872,7 @@ mod tests {
                 distinct_keys_per_window: None,
                 item_label: None,
                 monitor: None,
+                repeat_every: None,
             },
             WorkloadEntry {
                 metric_name: "b".into(),
@@ -698,6 +886,7 @@ mod tests {
                 distinct_keys_per_window: None,
                 item_label: None,
                 monitor: None,
+                repeat_every: None,
             },
             WorkloadEntry {
                 metric_name: "c".into(),
@@ -711,6 +900,7 @@ mod tests {
                 distinct_keys_per_window: None,
                 item_label: None,
                 monitor: None,
+                repeat_every: None,
             },
         ]);
         assert_eq!(reg.for_role("agent").len(), 2);
@@ -821,6 +1011,7 @@ mod tests {
             distinct_keys_per_window: None,
             item_label: None,
             monitor: None,
+            repeat_every: None,
         }
     }
 
diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs
index b8a5f4e6..0aab0521 100644
--- a/crates/asap_types/src/aggregation_config.rs
+++ b/crates/asap_types/src/aggregation_config.rs
@@ -156,6 +156,27 @@ pub struct PrecomputeMaterialization {
         skip_serializing_if = "Option::is_none"
     )]
     pub table_population: Option<crate::table_population::TablePopulation>,
+    /// Producer typing for a SQL table value projection: the source column's
+    /// declared type and nullability.
+    ///
+    /// **Why this is separate from [`Self::value_projection`]**: the
+    /// projection is the materialization's *identity* — which column or
+    /// constant is summarised, and part of the policy fingerprint. This is how
+    /// the ingest path must *read* that column, which the identity does not
+    /// determine: an integer column needs an exactness guard on its way into
+    /// f64 summary state, and a nullable column needs SQL's "aggregates skip
+    /// NULL" rule reproduced at the reader rather than a decode failure on the
+    /// first NULL row. Two materializations over the same column are the same
+    /// policy either way, so this deliberately stays out of the fingerprint.
+    ///
+    /// `None` ⇒ PromQL-mode materializations and legacy SQL definitions, which
+    /// keep the pre-typed behaviour (read the column as it comes).
+    #[serde(
+        default,
+        alias = "valueSourceColumn",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub value_source_column: Option<planner_types::pre_asap::Column>,
 }
 
 /// Policy-match handles for both the key and value dimensions of a
@@ -317,6 +338,7 @@ impl PrecomputeMaterialization {
                 .map(|name| crate::sds::ValueProjectionIdentity::Column { name }),
             table_population: None,
             table_timestamp_column: None,
+            value_source_column: None,
         }
     }
 
diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
index 8a7ba865..b37d4c79 100644
--- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs
+++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
@@ -1051,6 +1051,7 @@ mod tests {
             derived_input: None,
             table_timestamp_column: None,
             partitioning: None,
+            value_source_column: None,
         };
         let policy_fp = aggregation.policy_fp_u64();
         let streaming = StreamingConfig::new(HashMap::from([(policy_fp, aggregation)]));
@@ -1187,6 +1188,7 @@ mod tests {
                 derived_input: None,
                 table_timestamp_column: None,
                 partitioning: None,
+                value_source_column: None,
             };
         let cms = config(
             AggregationType::CountMinSketchWithHeap,
diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs
index ce1bbcee..a6d63925 100644
--- a/data_plane/src/drivers/query/servers/http.rs
+++ b/data_plane/src/drivers/query/servers/http.rs
@@ -3727,6 +3727,7 @@ aggregations:
                 derived_input: None,
                 table_timestamp_column: None,
                 partitioning: None,
+                value_source_column: None,
             };
             // PR 5: streaming-config is keyed on the policy
             // fingerprint. Build a marker→fingerprint map so the test
diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs
index 47fb0428..3c8d7ab4 100644
--- a/data_plane/src/precompute_engine/maintenance_runtime.rs
+++ b/data_plane/src/precompute_engine/maintenance_runtime.rs
@@ -2587,7 +2587,8 @@ mod tests {
                 &BTreeMap::new()
             )
             .is_err());
-        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+        let deadline =
+            crate::tests::test_utilities::timing::deadline(std::time::Duration::from_secs(5));
         while !store.seal_finite_summary_input(&generation).unwrap() {
             assert!(std::time::Instant::now() < deadline);
             std::thread::sleep(std::time::Duration::from_millis(5));
@@ -3023,7 +3024,8 @@ mod tests {
         assert!(store
             .series_ids_for_policy(configs[2].policy_fingerprint())
             .is_empty());
-        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+        let deadline =
+            crate::tests::test_utilities::timing::deadline(std::time::Duration::from_secs(5));
         while !store.seal_finite_summary_input(&generation).unwrap() {
             assert!(std::time::Instant::now() < deadline);
             std::thread::sleep(std::time::Duration::from_millis(5));
diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs
index 538940d8..c2e4ffc4 100644
--- a/data_plane/src/precompute_engine/output_sink.rs
+++ b/data_plane/src/precompute_engine/output_sink.rs
@@ -430,6 +430,7 @@ mod tests {
             derived_input: None,
             table_timestamp_column: None,
             partitioning: None,
+            value_source_column: None,
         }
     }
 
diff --git a/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs b/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs
index fb169e89..d19aad0e 100644
--- a/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs
+++ b/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs
@@ -50,6 +50,9 @@ pub struct ClickHouseReader {
     http: reqwest::Client,
     population: Option<asap_types::table_population::TablePopulation>,
     value_projection: Option<asap_types::sds::ValueProjectionIdentity>,
+    /// Producer typing for the projected column, when the materialization
+    /// carries it. Drives [`ClickHouseReader::value_column_guards`].
+    value_source_column: Option<planner_types::pre_asap::Column>,
     output_metric: Option<String>,
     grouping_projection: Option<asap_types::GroupingProjection>,
 }
@@ -77,6 +80,7 @@ impl ClickHouseReader {
             http: reqwest::Client::new(),
             population: None,
             value_projection: None,
+            value_source_column: None,
             output_metric: None,
             grouping_projection: None,
         })
@@ -139,24 +143,65 @@ impl ClickHouseReader {
         let value = match &self.value_projection {
             Some(asap_types::sds::ValueProjectionIdentity::Constant {
                 value: planner_types::pre_asap::ScalarValue::Int64(_),
-            }) => "{projected_value:Int64}",
+            }) => "{projected_value:Int64}".to_string(),
             Some(asap_types::sds::ValueProjectionIdentity::Constant {
                 value: planner_types::pre_asap::ScalarValue::Float64(_),
-            }) => "{projected_value:Float64}",
-            _ => c.value_column.as_str(),
+            }) => "{projected_value:Float64}".to_string(),
+            // A typed column projection is read through its declared type:
+            // integers via an explicit widening, everything else as-is.
+            _ => match self
+                .value_source_column
+                .as_ref()
+                .map(|column| &column.dtype)
+            {
+                Some(planner_types::pre_asap::DataType::Int64) => {
+                    format!("toFloat64({})", c.value_column)
+                }
+                _ => c.value_column.clone(),
+            },
         };
         format!(
             "SELECT {labels} AS labels, {timestamp} AS timestamp_ms, {value} AS value \
-             FROM {database}.{table} WHERE {population} \
+             FROM {database}.{table} WHERE {population}{value_guards} \
              AND {timestamp} >= {{start_ms:Int64}} AND {timestamp} < {{end_ms:Int64}} \
              ORDER BY labels, timestamp_ms FORMAT JSONEachRow",
             labels = labels,
             timestamp = c.timestamp_ms_column,
             value = value,
+            value_guards = self.value_column_guards(),
             database = c.database,
             table = c.table,
         )
     }
+
+    /// Predicates the typed value column needs before its rows may enter
+    /// summary state.
+    ///
+    /// * **NULL**: a SQL aggregate skips NULL inputs, so the summary that
+    ///   stands in for `sum(col)` / `count(col)` / `quantile(col)` has to skip
+    ///   them too. (Without this the JSON decode fails on the first NULL row,
+    ///   which is why nullable columns used to be refused at planning time.)
+    /// * **Integer exactness**: summary state is f64. Beyond 2^53 an integer
+    ///   no longer round-trips, so instead of silently summarising a rounded
+    ///   value the read fails loudly on the offending row. The guard evaluates
+    ///   to NULL for NULL rows, which the NULL predicate has already excluded.
+    fn value_column_guards(&self) -> String {
+        let Some(column) = self.value_source_column.as_ref() else {
+            return String::new();
+        };
+        let name = &self.config.value_column;
+        let mut guards = String::new();
+        if column.nullable {
+            guards.push_str(&format!(" AND {name} IS NOT NULL"));
+        }
+        if column.dtype == planner_types::pre_asap::DataType::Int64 {
+            guards.push_str(&format!(
+                " AND throwIf(abs({name}) > 9007199254740992, \
+                 'ASAP ClickHouse ingest: integer value exceeds the exact Float64 range') = 0"
+            ));
+        }
+        guards
+    }
 }
 
 /// Resolve a typed table source using deployment-local connection settings.
@@ -199,6 +244,7 @@ pub fn clickhouse_reader_factory(config: ClickHouseReaderConfig) -> ReaderFactor
             let mut reader = ClickHouseReader::new(source_config)?;
             reader.population = Some(materialization.table_population.clone().unwrap_or_default());
             reader.value_projection = Some(materialization.effective_value_projection().clone());
+            reader.value_source_column = materialization.value_source_column.clone();
             reader.output_metric = Some(materialization.metric.clone());
             reader.grouping_projection = Some(materialization.grouping_labels.clone());
             Ok(Arc::new(reader) as Arc<dyn RawSampleReader>)
@@ -410,6 +456,81 @@ mod tests {
         assert!(!sql.contains("{metric:String}"));
     }
 
+    /// SQL aggregates skip NULL inputs, so the summary standing in for one has
+    /// to skip them too — and a NULL row would otherwise fail the row decode,
+    /// which is why nullable columns used to be refused at planning time.
+    #[test]
+    fn nullable_value_columns_skip_null_rows_the_way_sql_aggregates_do() {
+        let mut reader = ClickHouseReader::new(config("samples")).unwrap();
+        reader.value_projection = Some(asap_types::sds::ValueProjectionIdentity::Column {
+            name: "value".into(),
+        });
+        reader.value_source_column = Some(planner_types::pre_asap::Column::new(
+            "value",
+            planner_types::pre_asap::DataType::Float64,
+            true,
+        ));
+        let sql = reader.sql();
+        assert!(sql.contains("AND value IS NOT NULL"), "{sql}");
+        // A float column is read as it comes; only integers are widened.
+        assert!(sql.contains(" value AS value"), "{sql}");
+        assert!(!sql.contains("toFloat64"), "{sql}");
+    }
+
+    /// A non-null column needs no NULL predicate: the guard tracks the declared
+    /// nullability, it is not applied blindly.
+    #[test]
+    fn non_null_value_columns_are_read_without_extra_predicates() {
+        let mut reader = ClickHouseReader::new(config("samples")).unwrap();
+        reader.value_projection = Some(asap_types::sds::ValueProjectionIdentity::Column {
+            name: "value".into(),
+        });
+        reader.value_source_column = Some(planner_types::pre_asap::Column::new(
+            "value",
+            planner_types::pre_asap::DataType::Float64,
+            false,
+        ));
+        let sql = reader.sql();
+        assert!(!sql.contains("IS NOT NULL"), "{sql}");
+        assert!(!sql.contains("throwIf"), "{sql}");
+    }
+
+    /// Summary state is f64. An integer column is widened explicitly, and a
+    /// value that no longer round-trips fails the read instead of entering the
+    /// summary rounded.
+    #[test]
+    fn integer_value_columns_are_widened_under_an_exactness_guard() {
+        let mut reader = ClickHouseReader::new(config("samples")).unwrap();
+        reader.value_projection = Some(asap_types::sds::ValueProjectionIdentity::Column {
+            name: "value".into(),
+        });
+        reader.value_source_column = Some(planner_types::pre_asap::Column::new(
+            "value",
+            planner_types::pre_asap::DataType::Int64,
+            false,
+        ));
+        let sql = reader.sql();
+        assert!(sql.contains("toFloat64(value) AS value"), "{sql}");
+        assert!(
+            sql.contains("throwIf(abs(value) > 9007199254740992"),
+            "{sql}"
+        );
+        assert!(!sql.contains("IS NOT NULL"), "{sql}");
+    }
+
+    /// A legacy definition without producer typing keeps its previous read.
+    #[test]
+    fn untyped_column_projection_reads_exactly_as_before() {
+        let mut reader = ClickHouseReader::new(config("samples")).unwrap();
+        reader.value_projection = Some(asap_types::sds::ValueProjectionIdentity::Column {
+            name: "value".into(),
+        });
+        let sql = reader.sql();
+        assert!(sql.contains(" value AS value"), "{sql}");
+        assert!(!sql.contains("IS NOT NULL"), "{sql}");
+        assert!(!sql.contains("throwIf"), "{sql}");
+    }
+
     #[test]
     fn constant_projection_uses_a_typed_parameter_without_a_fake_column() {
         let mut reader = ClickHouseReader::new(config("samples")).unwrap();
diff --git a/data_plane/src/storage_engines/sketch_db/backfill/service.rs b/data_plane/src/storage_engines/sketch_db/backfill/service.rs
index 3fd4347a..3b10f79b 100644
--- a/data_plane/src/storage_engines/sketch_db/backfill/service.rs
+++ b/data_plane/src/storage_engines/sketch_db/backfill/service.rs
@@ -389,7 +389,9 @@ mod tests {
         target: BackfillStatus,
         timeout_ms: u64,
     ) -> BackfillStatus {
-        let deadline = std::time::Instant::now() + std::time::Duration::from_millis(timeout_ms);
+        let deadline = crate::tests::test_utilities::timing::deadline(
+            std::time::Duration::from_millis(timeout_ms),
+        );
         loop {
             if let Some(j) = registry.get(job_id) {
                 if j.status == target || j.status.is_terminal() {
diff --git a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs
index a26140f1..7e7ac4d6 100644
--- a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs
+++ b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs
@@ -800,7 +800,8 @@ mod tests {
                 )
                 .unwrap();
         }
-        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+        let deadline =
+            crate::tests::test_utilities::timing::deadline(std::time::Duration::from_secs(5));
         while !store.seal_finite_summary_input(&generation).unwrap() {
             assert!(std::time::Instant::now() < deadline);
             std::thread::sleep(std::time::Duration::from_millis(5));
@@ -953,7 +954,8 @@ mod tests {
                 },
             )
             .unwrap();
-        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+        let deadline =
+            crate::tests::test_utilities::timing::deadline(std::time::Duration::from_secs(5));
         while !store.seal_finite_summary_input(&generation).unwrap() {
             assert!(std::time::Instant::now() < deadline);
             std::thread::sleep(std::time::Duration::from_millis(5));
@@ -1067,7 +1069,16 @@ mod tests {
     }
 
     #[test]
-    fn one_sid_with_two_populations_cannot_publish_a_partial_global_summary() {
+    fn one_sid_with_two_populations_publishes_one_complete_global_summary() {
+        // Two logical populations share a physical sid. A global reduce over
+        // them is publishable exactly when EVERY population contributes the
+        // window — `read_complete_raw_maintenance_cohort` enumerates the whole
+        // durable population set and fails on the first one that is missing,
+        // so the reduce sees both or it sees nothing. The partial publication
+        // this test used to assert against is therefore unreachable here; what
+        // still has to hold is that the complete reduce publishes ONCE, with
+        // both populations' state in it, and that a later lifetime which
+        // repeats a logical population is still refused (second half).
         let mut fixture: serde_json::Value = serde_json::from_str(include_str!(
             "../../../../../docs/examples/asapquery-compatibility-demo-snapshot.json"
         ))
@@ -1123,7 +1134,9 @@ mod tests {
             output.population_labels = Some(population);
             output.catalog_generation = Some(Arc::clone(&generation));
             let mut sum = SumAccumulator::new();
-            sum.update(5.0);
+            // Distinct per-population values: a summary built from only one of
+            // them reads back as 5 or 7, never as the pair.
+            sum.update(if instance == "a" { 5.0 } else { 7.0 });
             store
                 .publish_admitted_summary_update(
                     &generation,
@@ -1138,7 +1151,8 @@ mod tests {
         assert!(store
             .complete_raw_maintenance_population(source.policy_fingerprint().into(), &generation)
             .is_err());
-        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+        let deadline =
+            crate::tests::test_utilities::timing::deadline(std::time::Duration::from_secs(5));
         while !store.seal_finite_summary_input(&generation).unwrap() {
             assert!(std::time::Instant::now() < deadline);
             std::thread::sleep(std::time::Duration::from_millis(5));
@@ -1168,7 +1182,7 @@ mod tests {
             .unwrap();
         assert!(!frozen.singleton_population_complete);
         // The read set is deterministic and all-or-nothing, independently of
-        // the later routing decision (which still rejects this global reduce).
+        // the routing decision that consumes it.
         let request = |name: &str| {
             (
                 700,
@@ -1208,22 +1222,86 @@ mod tests {
                 &[request("a"), request("absent")]
             )
             .is_err());
-        let parts_before = persistence.manifest.live_parts().len();
         let resolver = crate::drivers::ingest::series_resolver::SeriesIdResolver::new();
-        for _ in 0..2 {
-            assert!(
-                crate::precompute_engine::maintenance_runtime::execute_finite_maintenance(
-                    &store,
-                    &resolver,
-                    &plan.precompute_plan
+        crate::precompute_engine::maintenance_runtime::execute_finite_maintenance(
+            &store,
+            &resolver,
+            &plan.precompute_plan,
+        )
+        .unwrap();
+
+        // One target series, one output population (the global reduce erases
+        // the source grouping), one window.
+        let target_id = target.policy_fingerprint().into();
+        let target_sids = store.series_ids_for_policy(target.policy_fingerprint());
+        assert_eq!(target_sids.len(), 1, "one global output series");
+        let published = store
+            .completed_maintenance_coordinates(target_id, &generation)
+            .unwrap();
+        assert_eq!(published.len(), 1);
+        let groups = &published[&target_sids[0]];
+        assert_eq!(groups.len(), 1);
+        assert_eq!(groups[&BTreeMap::new()], BTreeSet::from([(0, 60_000)]));
+
+        // The reduce consumed BOTH populations: the cohort the publication
+        // path reads is the whole durable population set for this window, and
+        // it carries the two distinct per-population sums. (Published summary
+        // state is sketch-encoded, and the immutable read-back path decodes
+        // exact accumulators only — the reduced VALUE is asserted against this
+        // same cohort contract by
+        // `precompute_engine::maintenance_runtime::tests`.)
+        let cohort = store
+            .read_complete_raw_maintenance_cohort(
+                &generation,
+                &BTreeSet::from([source_id]),
+                (0, 60_000),
+            )
+            .unwrap();
+        assert_eq!(cohort.inputs().len(), 2);
+        let contributions: Vec<(String, f64)> = cohort
+            .inputs()
+            .iter()
+            .map(|input| {
+                (
+                    input.group["instance"].clone(),
+                    input.windows[&(0, 60_000)]
+                        .query_statistic(
+                            asap_types::Statistic::Sum,
+                            &None,
+                            &std::collections::HashMap::new(),
+                        )
+                        .unwrap(),
                 )
-                .is_err()
-            );
-        }
-        assert!(store
-            .series_ids_for_policy(target.policy_fingerprint())
-            .is_empty());
-        assert_eq!(persistence.manifest.live_parts().len(), parts_before);
+            })
+            .collect();
+        assert_eq!(
+            contributions,
+            vec![("a".to_string(), 5.0), ("b".to_string(), 7.0)]
+        );
+
+        // Idempotent: a re-run republishes nothing and writes no new part.
+        let parts_after_publication = persistence.manifest.live_parts().len();
+        crate::precompute_engine::maintenance_runtime::execute_finite_maintenance(
+            &store,
+            &resolver,
+            &plan.precompute_plan,
+        )
+        .unwrap();
+        assert_eq!(
+            persistence.manifest.live_parts().len(),
+            parts_after_publication
+        );
+        assert_eq!(
+            store
+                .completed_maintenance_coordinates(target_id, &generation)
+                .unwrap(),
+            published
+        );
+        assert_eq!(
+            store.series_ids_for_policy(target.policy_fingerprint()),
+            target_sids
+        );
+        // The published output is durable and committed, not left pending.
         let records = store
             .persistence_metadata
             .read()
@@ -1232,9 +1310,15 @@ mod tests {
             .unwrap()
             .load_strict()
             .unwrap();
-        assert!(records.iter().all(|record| record.summary_definition_id
-            != Some(target.policy_fingerprint().into())
-            && record.pending_immutable.is_none()));
+        let published_target_sids: BTreeSet<u64> = records
+            .iter()
+            .filter(|record| !record.removed && record.summary_definition_id == Some(target_id))
+            .map(|record| record.sid)
+            .collect();
+        assert_eq!(published_target_sids.len(), 1);
+        assert!(records
+            .iter()
+            .all(|record| record.pending_immutable.is_none()));
         persistence.shutdown();
 
         // A catalog transition deliberately leaves old-generation payload
@@ -1281,7 +1365,8 @@ mod tests {
                 |writer| writer.ingest_precompute_with_series_id(702, source, &output, &sum),
             )
             .unwrap();
-        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+        let deadline =
+            crate::tests::test_utilities::timing::deadline(std::time::Duration::from_secs(5));
         while !restarted
             .seal_finite_summary_input(&next_generation)
             .unwrap()
@@ -1320,18 +1405,27 @@ mod tests {
             .series_ids_for_policy(target.policy_fingerprint())
             .is_empty());
         assert_eq!(restored.manifest.live_parts().len(), parts_before);
-        assert!(restarted
+        let after_restart = restarted
             .persistence_metadata
             .read()
             .unwrap()
             .as_ref()
             .unwrap()
             .load_strict()
-            .unwrap()
+            .unwrap();
+        // The refused lifetime publishes nothing NEW: the durable target state
+        // is exactly what the earlier complete reduce committed.
+        assert_eq!(
+            after_restart
+                .iter()
+                .filter(|record| !record.removed && record.summary_definition_id == Some(target_id))
+                .map(|record| record.sid)
+                .collect::<BTreeSet<u64>>(),
+            published_target_sids
+        );
+        assert!(after_restart
             .iter()
-            .all(|record| record.summary_definition_id
-                != Some(target.policy_fingerprint().into())
-                && record.pending_immutable.is_none()));
+            .all(|record| record.pending_immutable.is_none()));
         restored.shutdown();
     }
 
diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs
index 88ae71d6..d77030e2 100644
--- a/data_plane/src/storage_engines/sketch_db/index/mod.rs
+++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs
@@ -4791,7 +4791,10 @@ mod tests {
     }
 
     fn wait_until<F: Fn() -> bool>(f: F, timeout: std::time::Duration) -> bool {
-        let deadline = std::time::Instant::now() + timeout;
+        // Scaled: these waits are on background flusher / sealer threads that
+        // compete with the test harness for cores. See
+        // `crate::tests::test_utilities::timing`.
+        let deadline = crate::tests::test_utilities::timing::deadline(timeout);
         while std::time::Instant::now() < deadline {
             if f() {
                 return true;
diff --git a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs
index 469db04a..dbe81fbb 100644
--- a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs
+++ b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs
@@ -280,6 +280,7 @@ mod tests {
             derived_input: None,
             table_timestamp_column: None,
             partitioning: None,
+            value_source_column: None,
         }
     }
 
diff --git a/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs b/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs
index 4c8307db..ee66370c 100644
--- a/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs
+++ b/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs
@@ -721,7 +721,7 @@ mod tests {
         let mut handle = FlusherHandle::start(cfg, manifest.clone(), source.clone()).unwrap();
 
         // Wait until memory is below the low-water mark.
-        let deadline = Instant::now() + Duration::from_secs(2);
+        let deadline = crate::tests::test_utilities::timing::deadline(Duration::from_secs(2));
         while source.approx_memory_bytes() > 600 && Instant::now() < deadline {
             thread::sleep(Duration::from_millis(20));
         }
@@ -755,7 +755,7 @@ mod tests {
         cfg.hot_window_ms = Some(1_000); // 1 second hot window
         let mut handle = FlusherHandle::start(cfg, manifest.clone(), source.clone()).unwrap();
 
-        let deadline = Instant::now() + Duration::from_secs(2);
+        let deadline = crate::tests::test_utilities::timing::deadline(Duration::from_secs(2));
         while source.approx_memory_bytes() > 0 && Instant::now() < deadline {
             thread::sleep(Duration::from_millis(20));
         }
@@ -788,7 +788,7 @@ mod tests {
         cfg.hot_window_ms = Some(120_000); // 120s hot window, like live
         let mut handle = FlusherHandle::start(cfg, manifest.clone(), source.clone()).unwrap();
 
-        let deadline = Instant::now() + Duration::from_secs(3);
+        let deadline = crate::tests::test_utilities::timing::deadline(Duration::from_secs(3));
         while manifest.live_parts().is_empty() && Instant::now() < deadline {
             thread::sleep(Duration::from_millis(20));
         }
@@ -809,7 +809,7 @@ mod tests {
         cfg.delete_older_than_ms = Some(0); // then immediately expire it
         let mut handle = FlusherHandle::start(cfg, manifest.clone(), source.clone()).unwrap();
 
-        let deadline = Instant::now() + Duration::from_secs(2);
+        let deadline = crate::tests::test_utilities::timing::deadline(Duration::from_secs(2));
         while !manifest.live_parts().is_empty() && Instant::now() < deadline {
             thread::sleep(Duration::from_millis(20));
         }
diff --git a/data_plane/src/tests/test_utilities/engine_factories.rs b/data_plane/src/tests/test_utilities/engine_factories.rs
index dd88cab6..b5daec83 100644
--- a/data_plane/src/tests/test_utilities/engine_factories.rs
+++ b/data_plane/src/tests/test_utilities/engine_factories.rs
@@ -113,6 +113,7 @@ pub fn create_engine_single_pop_with_aggregated(
         derived_input: None,
         table_timestamp_column: None,
         partitioning: None,
+        value_source_column: None,
     };
     let agg_id = agg_config.policy_fp_u64();
     aggregation_configs.insert(agg_id, agg_config);
@@ -212,6 +213,7 @@ pub fn create_engine_dual_input(
         derived_input: None,
         table_timestamp_column: None,
         partitioning: None,
+        value_source_column: None,
     };
     let value_id = value_agg_config.policy_fp_u64();
     aggregation_configs.insert(value_id, value_agg_config);
@@ -241,6 +243,7 @@ pub fn create_engine_dual_input(
         derived_input: None,
         table_timestamp_column: None,
         partitioning: None,
+        value_source_column: None,
     };
     let keys_id = keys_agg_config.policy_fp_u64();
     aggregation_configs.insert(keys_id, keys_agg_config);
@@ -335,6 +338,7 @@ pub fn create_engine_two_metrics(
         derived_input: None,
         table_timestamp_column: None,
         partitioning: None,
+        value_source_column: None,
     };
     let id_a = agg_config_a.policy_fp_u64();
     aggregation_configs.insert(id_a, agg_config_a);
@@ -363,6 +367,7 @@ pub fn create_engine_two_metrics(
         derived_input: None,
         table_timestamp_column: None,
         partitioning: None,
+        value_source_column: None,
     };
     let id_b = agg_config_b.policy_fp_u64();
     aggregation_configs.insert(id_b, agg_config_b);
@@ -467,6 +472,7 @@ pub fn create_engine_three_metrics(
             derived_input: None,
             table_timestamp_column: None,
             partitioning: None,
+            value_source_column: None,
         };
         let id = cfg.policy_fp_u64();
         ids.push(id);
@@ -548,6 +554,7 @@ pub fn create_engine_multi_timestamp(
         derived_input: None,
         table_timestamp_column: None,
         partitioning: None,
+        value_source_column: None,
     };
     let agg_id = agg_config.policy_fp_u64();
     aggregation_configs.insert(agg_id, agg_config);
@@ -621,6 +628,7 @@ pub fn create_engine_multi_timestamp_with_window(
         derived_input: None,
         table_timestamp_column: None,
         partitioning: None,
+        value_source_column: None,
     };
     let agg_id = agg_config.policy_fp_u64();
     aggregation_configs.insert(agg_id, agg_config);
diff --git a/data_plane/src/tests/test_utilities/mod.rs b/data_plane/src/tests/test_utilities/mod.rs
index 8515aa5f..2b419d48 100644
--- a/data_plane/src/tests/test_utilities/mod.rs
+++ b/data_plane/src/tests/test_utilities/mod.rs
@@ -7,5 +7,6 @@
 //! `StoreQueryPlan` legacy types.
 
 pub mod engine_factories;
+pub mod timing;
 
 pub use engine_factories::*;
diff --git a/data_plane/src/tests/test_utilities/timing.rs b/data_plane/src/tests/test_utilities/timing.rs
new file mode 100644
index 00000000..3b579ff4
--- /dev/null
+++ b/data_plane/src/tests/test_utilities/timing.rs
@@ -0,0 +1,64 @@
+//! Wall-clock budgets for tests that wait on background progress.
+//!
+//! Several storage / maintenance tests wait for work that a *background*
+//! thread performs — the persistence flusher sealing and evicting epochs, the
+//! finite-input sealer draining an admitted revision. A fixed five-second
+//! budget is generous when the test owns the machine and far too tight when
+//! `cargo test` runs one test per core: on a 64-way runner the waiter and the
+//! background thread it is waiting on compete for the same CPUs, and the
+//! budget expires while the work is merely queued. That is what made the
+//! suite's flush / evict / restart tests fail under the default parallel run
+//! and pass on a serial re-run, which in turn kept the data-plane library
+//! tests out of the default CI gate.
+//!
+//! Scaling the budget with the test concurrency keeps both properties: a
+//! passing test still returns as soon as its condition holds (these are
+//! polling waits, not sleeps), and a genuinely stuck condition still fails —
+//! later, but deterministically.
+
+use std::time::Duration;
+
+/// Scale a wait budget by the concurrency the test binary is running at.
+///
+/// Override with `ASAP_TEST_TIMEOUT_SCALE=<n>` to pin a multiplier (useful on
+/// a loaded shared machine, or to re-tighten the budget while debugging a
+/// hang). Otherwise the multiplier is derived from `RUST_TEST_THREADS` when
+/// the harness was told a thread count, and from the machine's parallelism
+/// otherwise, capped so the worst case stays bounded.
+pub fn scaled(budget: Duration) -> Duration {
+    budget * scale()
+}
+
+fn scale() -> u32 {
+    if let Some(explicit) = std::env::var("ASAP_TEST_TIMEOUT_SCALE")
+        .ok()
+        .and_then(|value| value.trim().parse::<u32>().ok())
+    {
+        return explicit.max(1);
+    }
+    let threads = std::env::var("RUST_TEST_THREADS")
+        .ok()
+        .and_then(|value| value.trim().parse::<usize>().ok())
+        .or_else(|| std::thread::available_parallelism().ok().map(|n| n.get()))
+        .unwrap_or(1);
+    ((threads / 8) as u32).clamp(1, 8)
+}
+
+/// A deadline `budget` from now, scaled by [`scaled`].
+pub fn deadline(budget: Duration) -> std::time::Instant {
+    std::time::Instant::now() + scaled(budget)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn explicit_scale_overrides_the_derived_one() {
+        // The env var is process-global; assert the parsing contract on the
+        // pure helper instead of mutating it under a parallel harness.
+        assert!(scale() >= 1);
+        assert!(scaled(Duration::from_secs(5)) >= Duration::from_secs(5));
+        assert!(scaled(Duration::from_secs(5)) <= Duration::from_secs(40));
+    }
+}
diff --git a/docs/developer_docs/query-engine/clickhouse-sql-support.md b/docs/developer_docs/query-engine/clickhouse-sql-support.md
index 7873028a..919bf94a 100644
--- a/docs/developer_docs/query-engine/clickhouse-sql-support.md
+++ b/docs/developer_docs/query-engine/clickhouse-sql-support.md
@@ -65,8 +65,15 @@ publishes the shared catalog and both execution plans through the normal atomic
 install/activate path.
 
 The initial automatic binder supports bounded, whole-second scalar reductions
-over a non-null Float64 value column or a finite numeric literal, plus typed
-table predicates. Row counts use the shared typed constant `1` projection. It uses the query's fixed
+over a numeric value column — `Float64` or `Int64`, nullable or not — or a
+finite numeric literal, plus typed table predicates. The column's declared type
+and nullability travel with the materialization (`value_source_column`) because
+the ingest reader needs them: a nullable column is read with `IS NOT NULL`, so
+the summary skips NULL inputs exactly as the SQL aggregate it stands in for
+does, and an `Int64` column is widened explicitly with a guard that fails the
+read on a value beyond the exact `Float64` range rather than summarising a
+rounded one. Typing is a read concern, not an identity one, so it stays out of
+the policy fingerprint. Row counts use the shared typed constant `1` projection. It uses the query's fixed
 window as the materialization duration; this is not a cost-optimized pane/layout
 search. The initial fixed-window policy retains two windows (a completed window
 and the next active window); it does not certify arbitrary historical or moving
@@ -79,9 +86,9 @@ SQL `count(*)` now installs an exact row-count materialization through the same
 catalog and readout DAG as other summaries. The selected Count intent uses the
 existing physical SUM accumulator over typed constant `1`. It includes rows with a NULL value column.
 Planner rejects nullable `count(value)` until per-aggregate null exclusion is
-represented; it must not silently become row count. Non-Float64 named sources
-are rejected because the current Float64 ingest path cannot preserve arbitrary
-Int64 values exactly. Producer `Project` subtrees are also rejected until their
+represented; it must not silently become row count. Non-numeric named sources
+are rejected: a string, boolean or timestamp column has no value semantics to
+summarise. Producer `Project` subtrees are also rejected until their
 computation is executed, rather than skipped while binding the original table.
 The real process test covers SUM and row count mixed with a ClickHouse exact
 branch, and deletes a source row after materialization to prove summary readout.