diff --git a/crates/asap-aware-mapping/src/analytical_cost.rs b/crates/asap-aware-mapping/src/analytical_cost.rs
index 3c19960a9..ad921d23a 100644
--- a/crates/asap-aware-mapping/src/analytical_cost.rs
+++ b/crates/asap-aware-mapping/src/analytical_cost.rs
@@ -14,7 +14,7 @@ use serde::{Deserialize, Serialize};
 
 use crate::physical_operator_statistics::{
     validate_comparison_scopes, ComparisonScope, EdgeStatistics, OperatorStatistics,
-    OperatorStatisticsProvider, PromqlEdgeStatistics, PromqlValueKind, SourceCoverage,
+    OperatorStatisticsProvider, PromqlEdgeStatistics, PromqlValueKind, ScanSelection,
 };
 
 /// Version of the analytical formulas applied to evidenced physical plans.
@@ -383,9 +383,9 @@ pub struct PhysicalDAGNode {
     pub operator: PhysicalOperator,
     pub children: Vec<String>,
     /// Exact comparison-scope coverage consumed by a scan. Non-scan nodes
-    /// leave this empty. Reusing `SourceCoverage` prevents a physical plan
+    /// leave this empty. Reusing `ScanSelection` prevents a physical plan
     /// from naming a source independently of its snapshot and predicates.
-    pub source_coverage: Option<SourceCoverage>,
+    pub scan_selection: Option<ScanSelection>,
     /// Maximum transient edge buffer, distinct from logical `output_bytes`.
     pub output_buffer_bytes: u64,
     /// State that remains live after this node finishes (zero for ordinary
@@ -573,9 +573,10 @@ pub fn estimate_physical_dag_with_cache(
         let node_statistics = &resolved_statistics[id];
         match node.operator {
             PhysicalOperator::Scan => {
-                let coverage = node.source_coverage.as_ref().ok_or_else(|| {
-                    AnalyticalCostError::MissingScanSourceCoverage(node.id.clone())
-                })?;
+                let coverage = node
+                    .scan_selection
+                    .as_ref()
+                    .ok_or_else(|| AnalyticalCostError::MissingScanSelection(node.id.clone()))?;
                 if !scope.sources.contains(coverage) {
                     return Err(AnalyticalCostError::ScanOutsideComparisonScope(
                         node.id.clone(),
@@ -585,9 +586,9 @@ pub fn estimate_physical_dag_with_cache(
                     consumed_sources.push(coverage);
                 }
             }
-            _ if node.source_coverage.is_some() => {
+            _ if node.scan_selection.is_some() => {
                 return Err(AnalyticalCostError::InvalidPhysicalDAG(
-                    "only scan nodes may declare source coverage",
+                    "only scan nodes may declare scan selection",
                 ));
             }
             _ => {}
@@ -2199,9 +2200,9 @@ pub enum AnalyticalCostError {
     UnsupportedSummaryOperation(&'static str),
     #[error("required comparison-scope field {0} is missing")]
     MissingComparisonScope(&'static str),
-    #[error("scan node {0} does not declare source coverage")]
-    MissingScanSourceCoverage(String),
-    #[error("scan node {0} reads source coverage outside the comparison scope")]
+    #[error("scan node {0} does not declare scan selection")]
+    MissingScanSelection(String),
+    #[error("scan node {0} reads scan selection outside the comparison scope")]
     ScanOutsideComparisonScope(String),
     #[error("raw and candidate comparison scopes differ in {0}")]
     ComparisonScopeMismatch(&'static str),
@@ -2234,7 +2235,7 @@ mod tests {
     use crate::physical_operator_statistics::{
         validate_comparison_scopes, BinaryEdgeStatistics, ComparisonScope, EdgeStatistics,
         OperatorStatistics, PartitionStatistics, PromqlBinaryEdgeStatistics, PromqlEdgeStatistics,
-        PromqlUnaryEdgeStatistics, PromqlValueKind, SourceCoverage, UnaryEdgeStatistics,
+        PromqlUnaryEdgeStatistics, PromqlValueKind, ScanSelection, UnaryEdgeStatistics,
     };
 
     /// Analytical estimates reuse the shared dimensions while preserving exact
@@ -2475,7 +2476,7 @@ mod tests {
                 id: "scan".into(),
                 operator: PhysicalOperator::Scan,
                 children: vec![],
-                source_coverage: Some(comparison_scope().sources[0].clone()),
+                scan_selection: Some(comparison_scope().sources[0].clone()),
                 output_buffer_bytes: 8,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
@@ -2484,7 +2485,7 @@ mod tests {
                 id: "filter".into(),
                 operator: filter_operator(),
                 children: vec!["scan".into()],
-                source_coverage: None,
+                scan_selection: None,
                 output_buffer_bytes: 8,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
@@ -2814,7 +2815,7 @@ mod tests {
                 id: "scan".into(),
                 operator: PhysicalOperator::Scan,
                 children: vec![],
-                source_coverage: Some(coverage),
+                scan_selection: Some(coverage),
                 output_buffer_bytes: 10,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
@@ -2823,7 +2824,7 @@ mod tests {
                 id: "left".into(),
                 operator: filter_operator(),
                 children: vec!["scan".into()],
-                source_coverage: None,
+                scan_selection: None,
                 output_buffer_bytes: 4,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
@@ -2832,7 +2833,7 @@ mod tests {
                 id: "right".into(),
                 operator: filter_operator(),
                 children: vec!["scan".into()],
-                source_coverage: None,
+                scan_selection: None,
                 output_buffer_bytes: 4,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
@@ -2841,7 +2842,7 @@ mod tests {
                 id: "root".into(),
                 operator: PhysicalOperator::Concat,
                 children: vec!["left".into(), "right".into()],
-                source_coverage: None,
+                scan_selection: None,
                 output_buffer_bytes: 8,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
@@ -2905,7 +2906,7 @@ mod tests {
                 id: "scan".into(),
                 operator: PhysicalOperator::Scan,
                 children: vec![],
-                source_coverage: Some(coverage),
+                scan_selection: Some(coverage),
                 output_buffer_bytes: 10,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::Once,
@@ -2914,7 +2915,7 @@ mod tests {
                 id: "state".into(),
                 operator: aggregate_operator(),
                 children: vec!["scan".into()],
-                source_coverage: None,
+                scan_selection: None,
                 output_buffer_bytes: 16,
                 retained_bytes: 32,
                 execution: ExecutionMultiplicity::Once,
@@ -2926,7 +2927,7 @@ mod tests {
                     offset: 0,
                 },
                 children: vec!["state".into()],
-                source_coverage: None,
+                scan_selection: None,
                 output_buffer_bytes: 16,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
@@ -2987,7 +2988,7 @@ mod tests {
                 lookback: Some(DurationMs(300_000)),
                 as_of: Some(TimestampMs(1_000)),
             },
-            sources: vec![SourceCoverage {
+            sources: vec![ScanSelection {
                 source: Source::Table {
                     table_ref: "metrics".into(),
                 },
@@ -3059,7 +3060,7 @@ mod tests {
                 id: "scan".into(),
                 operator: PhysicalOperator::Scan,
                 children: vec![],
-                source_coverage: Some(coverage),
+                scan_selection: Some(coverage),
                 output_buffer_bytes: 10,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
@@ -3068,7 +3069,7 @@ mod tests {
                 id: "filter".into(),
                 operator: filter_operator(),
                 children: vec!["scan".into()],
-                source_coverage: None,
+                scan_selection: None,
                 output_buffer_bytes: 4,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
@@ -3134,7 +3135,7 @@ mod tests {
                 id: "scan".into(),
                 operator: PhysicalOperator::Scan,
                 children: vec![],
-                source_coverage: Some(coverage),
+                scan_selection: Some(coverage),
                 output_buffer_bytes: 10,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
@@ -3143,7 +3144,7 @@ mod tests {
                 id: "filter".into(),
                 operator: filter_operator(),
                 children: vec!["scan".into()],
-                source_coverage: None,
+                scan_selection: None,
                 output_buffer_bytes: 4,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
@@ -3197,7 +3198,7 @@ mod tests {
                 id: "scan".into(),
                 operator: PhysicalOperator::Scan,
                 children: vec![],
-                source_coverage: Some(coverage),
+                scan_selection: Some(coverage),
                 output_buffer_bytes: 10,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
@@ -3206,7 +3207,7 @@ mod tests {
                 id: "filter".into(),
                 operator: filter_operator(),
                 children: vec!["scan".into()],
-                source_coverage: None,
+                scan_selection: None,
                 output_buffer_bytes: 0,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
@@ -3245,7 +3246,7 @@ mod tests {
             id: "scan".into(),
             operator: PhysicalOperator::Scan,
             children: vec![],
-            source_coverage: Some(SourceCoverage {
+            scan_selection: Some(ScanSelection {
                 source: asap_types::pre_asap::query_expr::Source::Table {
                     table_ref: "other_metrics".into(),
                 },
@@ -3283,7 +3284,7 @@ mod tests {
             id: "scan".into(),
             operator: PhysicalOperator::Scan,
             children: vec![],
-            source_coverage: None,
+            scan_selection: None,
             output_buffer_bytes: 10,
             retained_bytes: 0,
             execution: ExecutionMultiplicity::PerEvaluation,
@@ -3302,9 +3303,7 @@ mod tests {
 
         assert_eq!(
             estimate_physical_dag(&nodes, "scan", &comparison_scope(), &provided),
-            Err(AnalyticalCostError::MissingScanSourceCoverage(
-                "scan".into()
-            ))
+            Err(AnalyticalCostError::MissingScanSelection("scan".into()))
         );
     }
 
@@ -3312,7 +3311,7 @@ mod tests {
     fn physical_dag_rejects_an_unconsumed_scope_source() {
         let mut scope = comparison_scope();
         let coverage = scope.sources[0].clone();
-        scope.sources.push(SourceCoverage {
+        scope.sources.push(ScanSelection {
             source: asap_types::pre_asap::query_expr::Source::Table {
                 table_ref: "auxiliary".into(),
             },
@@ -3324,7 +3323,7 @@ mod tests {
             id: "scan".into(),
             operator: PhysicalOperator::Scan,
             children: vec![],
-            source_coverage: Some(coverage),
+            scan_selection: Some(coverage),
             output_buffer_bytes: 10,
             retained_bytes: 0,
             execution: ExecutionMultiplicity::PerEvaluation,
@@ -3357,7 +3356,7 @@ mod tests {
                 id: "scan".into(),
                 operator: PhysicalOperator::Scan,
                 children: vec![],
-                source_coverage: Some(coverage),
+                scan_selection: Some(coverage),
                 output_buffer_bytes: 10,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
@@ -3366,7 +3365,7 @@ mod tests {
                 id: "aggregate".into(),
                 operator: aggregate_operator(),
                 children: vec!["scan".into()],
-                source_coverage: None,
+                scan_selection: None,
                 output_buffer_bytes: 16,
                 retained_bytes: 32,
                 execution: ExecutionMultiplicity::Once,
@@ -3411,7 +3410,7 @@ mod tests {
             id: "scan".into(),
             operator: PhysicalOperator::Scan,
             children: vec![],
-            source_coverage: Some(coverage),
+            scan_selection: Some(coverage),
             output_buffer_bytes: 10,
             retained_bytes: 0,
             execution: ExecutionMultiplicity::PerEvaluation,
@@ -3642,7 +3641,7 @@ mod tests {
                 id: "scan".into(),
                 operator: PhysicalOperator::Scan,
                 children: vec![],
-                source_coverage: Some(comparison_scope().sources[0].clone()),
+                scan_selection: Some(comparison_scope().sources[0].clone()),
                 output_buffer_bytes: 0,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
diff --git a/crates/asap-aware-mapping/src/physical_operator_statistics.rs b/crates/asap-aware-mapping/src/physical_operator_statistics.rs
index 9cf8045bb..7be1d292c 100644
--- a/crates/asap-aware-mapping/src/physical_operator_statistics.rs
+++ b/crates/asap-aware-mapping/src/physical_operator_statistics.rs
@@ -26,12 +26,12 @@ pub struct ComparisonScope {
     pub horizon: DurationMs,
     pub recurrence: QueryRecurrence,
     pub time_selection: TimeSelection,
-    pub sources: Vec<SourceCoverage>,
+    pub sources: Vec<ScanSelection>,
 }
 
 /// Exact source selection covered by a physical plan.
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-pub struct SourceCoverage {
+pub struct ScanSelection {
     pub source: Source,
     /// Provider-owned stable identifier for the physical source contents,
     /// such as a catalog snapshot, table version, or object generation.
@@ -53,7 +53,7 @@ impl ComparisonScope {
         query: &QueryWorkloadEntry,
         planning_time: TimestampMs,
         horizon: DurationMs,
-        sources: Vec<SourceCoverage>,
+        sources: Vec<ScanSelection>,
     ) -> Result<Self, AnalyticalCostError> {
         let scope = Self {
             data_arrival: data.arrival,
@@ -87,7 +87,7 @@ impl ComparisonScope {
             .any(|(index, source)| self.sources[..index].contains(source))
         {
             return Err(AnalyticalCostError::MissingComparisonScope(
-                "duplicate source coverage",
+                "duplicate scan selection",
             ));
         }
         if self
diff --git a/crates/asap-aware-mapping/src/physical_plan_cost_model.rs b/crates/asap-aware-mapping/src/physical_plan_cost_model.rs
index 307fb9a61..509e6a13a 100644
--- a/crates/asap-aware-mapping/src/physical_plan_cost_model.rs
+++ b/crates/asap-aware-mapping/src/physical_plan_cost_model.rs
@@ -367,7 +367,7 @@ mod tests {
 
     use crate::analytical_cost::{ExecutionMultiplicity, PhysicalDAGNode, PhysicalOperator};
     use crate::physical_operator_statistics::{
-        EdgeStatistics, OperatorStatistics, SourceCoverage, UnaryEdgeStatistics,
+        EdgeStatistics, OperatorStatistics, ScanSelection, UnaryEdgeStatistics,
     };
     use crate::replacement::ReplacementStrategy;
 
@@ -438,7 +438,7 @@ mod tests {
                 lookback: Some(DurationMs(10_000)),
                 as_of: Some(TimestampMs(1_000)),
             },
-            sources: vec![SourceCoverage {
+            sources: vec![ScanSelection {
                 source: Source::Table {
                     table_ref: "events".into(),
                 },
@@ -506,7 +506,7 @@ mod tests {
                         id: "candidate-scan".into(),
                         operator: PhysicalOperator::Scan,
                         children: vec![],
-                        source_coverage: Some(scope.sources[0].clone()),
+                        scan_selection: Some(scope.sources[0].clone()),
                         output_buffer_bytes: 8,
                         retained_bytes: 0,
                         execution: ExecutionMultiplicity::Once,
@@ -518,7 +518,7 @@ mod tests {
                             accumulator_count: 1,
                         },
                         children: vec!["candidate-scan".into()],
-                        source_coverage: None,
+                        scan_selection: None,
                         output_buffer_bytes: 8,
                         retained_bytes: 8,
                         execution: ExecutionMultiplicity::Once,
@@ -527,7 +527,7 @@ mod tests {
                         id: "candidate-read".into(),
                         operator: PhysicalOperator::PassThrough,
                         children: vec!["candidate-state".into()],
-                        source_coverage: None,
+                        scan_selection: None,
                         output_buffer_bytes: 8,
                         retained_bytes: 0,
                         execution: ExecutionMultiplicity::PerEvaluation,
@@ -1006,7 +1006,7 @@ mod tests {
             ) -> Result<PhysicalDAG, AnalyticalCostError> {
                 let mut dag = self.0.summary_physical_dag(snapshot, summary, target)?;
                 dag.nodes[0]
-                    .source_coverage
+                    .scan_selection
                     .as_mut()
                     .unwrap()
                     .source_snapshot_id = "other".into();
diff --git a/crates/asap-aware-mapping/src/query_physical_lowering.rs b/crates/asap-aware-mapping/src/query_physical_lowering.rs
index 1b8e5a751..65b7fab93 100644
--- a/crates/asap-aware-mapping/src/query_physical_lowering.rs
+++ b/crates/asap-aware-mapping/src/query_physical_lowering.rs
@@ -9,7 +9,7 @@ use crate::analytical_cost::{
     PromqlSeriesSampleKind, PromqlVectorCardinality,
 };
 use crate::physical_operator_statistics::{
-    ComparisonScope, EdgeStatistics, OperatorStatistics, SourceCoverage,
+    ComparisonScope, EdgeStatistics, OperatorStatistics, ScanSelection,
 };
 
 pub struct PhysicalNodeRequest<'a> {
@@ -18,7 +18,7 @@ pub struct PhysicalNodeRequest<'a> {
     pub occurrence: usize,
     pub synthetic: bool,
     pub children: &'a [String],
-    pub source_coverage: Option<&'a SourceCoverage>,
+    pub scan_selection: Option<&'a ScanSelection>,
 }
 
 pub trait PhysicalNodeEvidenceProvider {
@@ -78,7 +78,7 @@ pub fn lower_query_physical_dag(
             occurrence: usize,
             synthetic: bool,
             children: &[String],
-            source_coverage: Option<&SourceCoverage>,
+            scan_selection: Option<&ScanSelection>,
         ) -> Result<PhysicalNodeEvidence, AnalyticalCostError> {
             let evidence = self.provider.evidence(PhysicalNodeRequest {
                 logical_node: query,
@@ -86,7 +86,7 @@ pub fn lower_query_physical_dag(
                 occurrence,
                 synthetic,
                 children,
-                source_coverage,
+                scan_selection,
             })?;
             if evidence.physical_id.is_empty() {
                 return Err(AnalyticalCostError::InvalidPhysicalDAG(
@@ -101,14 +101,14 @@ pub fn lower_query_physical_dag(
             evidence: PhysicalNodeEvidence,
             operator: PhysicalOperator,
             children: Vec<String>,
-            source_coverage: Option<SourceCoverage>,
+            scan_selection: Option<ScanSelection>,
         ) -> Result<String, AnalyticalCostError> {
             let id = evidence.physical_id.clone();
             let node = PhysicalDAGNode {
                 id: id.clone(),
                 operator,
                 children,
-                source_coverage,
+                scan_selection,
                 output_buffer_bytes: evidence.output_buffer_bytes,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
@@ -837,7 +837,7 @@ fn validate_source_consumption(
     let consumed = nodes
         .iter()
         .filter(|node| matches!(node.operator, PhysicalOperator::Scan))
-        .filter_map(|node| node.source_coverage.as_ref())
+        .filter_map(|node| node.scan_selection.as_ref())
         .collect::<Vec<_>>();
     for coverage in &consumed {
         if !scope.sources.contains(coverage) {
@@ -959,7 +959,7 @@ fn bind_scan_coverage(
     source: &asap_types::pre_asap::Source,
     predicates: &[asap_types::pre_asap::Predicate],
     scope: &ComparisonScope,
-) -> Result<SourceCoverage, AnalyticalCostError> {
+) -> Result<ScanSelection, AnalyticalCostError> {
     let mut matches = scope.sources.iter().filter(|coverage| {
         coverage.source == *source
             && coverage.predicates == predicates
@@ -971,7 +971,7 @@ fn bind_scan_coverage(
         .ok_or_else(|| AnalyticalCostError::ScanOutsideComparisonScope(node_id.into()))?;
     if matches.any(|candidate| candidate != &coverage) {
         return Err(AnalyticalCostError::InvalidPhysicalDAG(
-            "scan source coverage is ambiguous",
+            "scan selection is ambiguous",
         ));
     }
     Ok(coverage)
@@ -981,7 +981,7 @@ fn bind_info_coverage(
     node_id: &str,
     selector: &[asap_types::pre_asap::InfoMatcher],
     scope: &ComparisonScope,
-) -> Result<SourceCoverage, AnalyticalCostError> {
+) -> Result<ScanSelection, AnalyticalCostError> {
     use asap_types::pre_asap::{CompareOpKind, Source};
 
     let mut metric: Option<&str> = None;
@@ -1009,7 +1009,7 @@ fn bind_info_coverage(
         .ok_or_else(|| AnalyticalCostError::ScanOutsideComparisonScope(node_id.into()))?;
     if matches.next().is_some() {
         return Err(AnalyticalCostError::InvalidPhysicalDAG(
-            "info source coverage is ambiguous",
+            "info scan selection is ambiguous",
         ));
     }
     Ok(coverage)
@@ -1383,7 +1383,7 @@ mod tests {
         }
     }
 
-    fn scope(sources: Vec<SourceCoverage>) -> ComparisonScope {
+    fn scope(sources: Vec<ScanSelection>) -> ComparisonScope {
         ComparisonScope {
             data_arrival: DataArrival::AtRest,
             planning_time: TimestampMs(1_000),
@@ -1404,8 +1404,8 @@ mod tests {
     fn coverage(
         source: asap_types::pre_asap::Source,
         predicates: Vec<asap_types::pre_asap::Predicate>,
-    ) -> SourceCoverage {
-        SourceCoverage {
+    ) -> ScanSelection {
+        ScanSelection {
             source,
             source_snapshot_id: "snapshot-1".into(),
             predicates,
@@ -1414,7 +1414,7 @@ mod tests {
     }
 
     #[test]
-    fn info_source_coverage_includes_symbolic_selector_matchers() {
+    fn info_scan_selection_includes_symbolic_selector_matchers() {
         use asap_types::pre_asap::{CompareOpKind, InfoMatcher, Source};
 
         let selector = vec![InfoMatcher {
@@ -1422,7 +1422,7 @@ mod tests {
             op: CompareOpKind::Eq,
             value: "prod".into(),
         }];
-        let info_coverage = SourceCoverage {
+        let info_coverage = ScanSelection {
             source: Source::TimeSeries {
                 metric: "target_info".into(),
             },
@@ -1611,10 +1611,7 @@ mod tests {
         ));
         let physical_scan = &dag.nodes[0];
         assert_eq!(physical_scan.id, "query-2-scan");
-        assert_eq!(
-            physical_scan.source_coverage,
-            Some(scope.sources[0].clone())
-        );
+        assert_eq!(physical_scan.scan_selection, Some(scope.sources[0].clone()));
         assert_eq!(physical_scan.output_buffer_bytes, 1_024);
         assert_ne!(
             physical_scan.output_buffer_bytes,
@@ -1661,13 +1658,13 @@ mod tests {
             left: Rc::clone(&shared),
             right: Rc::clone(&shared),
         });
-        let source_coverage = coverage(
+        let scan_selection = coverage(
             Source::Table {
                 table_ref: "dimensions".into(),
             },
             vec![],
         );
-        let independent_scope = scope(vec![source_coverage.clone()]);
+        let independent_scope = scope(vec![scan_selection.clone()]);
         let scan_statistics = scan_stats(edge(100, 800), 800);
         let join_statistics = OperatorStatistics::HashJoin {
             edges: BinaryEdgeStatistics {
@@ -1703,7 +1700,7 @@ mod tests {
                 statistics,
             })
         };
-        let shared_scope = scope(vec![source_coverage]);
+        let shared_scope = scope(vec![scan_selection]);
         let no_cache = crate::analytical_cost::CacheProfile::no_cache();
         let shared_dag = lower_query_physical_dag(&root, &shared_scope, &shared_provider).unwrap();
         assert_eq!(shared_dag.nodes.len(), 2);
@@ -1860,13 +1857,13 @@ mod tests {
             child: limit,
         });
 
-        let source_coverage = coverage(
+        let scan_selection = coverage(
             Source::Table {
                 table_ref: "events".into(),
             },
             vec![],
         );
-        let scope = scope(vec![source_coverage]);
+        let scope = scope(vec![scan_selection]);
         let scan_statistics = scan_stats(edge(1_000, 8_000), 8_000);
         let dedup_statistics = OperatorStatistics::HashDeduplicate {
             edges: unary_edges(edge(800, 3_200), edge(500, 2_000)),
@@ -2032,7 +2029,7 @@ mod tests {
         assert_eq!(
             duplicate_scope.validate(),
             Err(AnalyticalCostError::MissingComparisonScope(
-                "duplicate source coverage"
+                "duplicate scan selection"
             ))
         );
 
@@ -2142,7 +2139,7 @@ mod tests {
         assert_eq!(
             lower_query_physical_dag(&root, &ambiguous_scope, &scripted(&conflicting)),
             Err(AnalyticalCostError::InvalidPhysicalDAG(
-                "scan source coverage is ambiguous"
+                "scan selection is ambiguous"
             ))
         );
 
diff --git a/crates/asap-aware-mapping/src/storage_io.rs b/crates/asap-aware-mapping/src/storage_io.rs
index 646a3532c..125ec915b 100644
--- a/crates/asap-aware-mapping/src/storage_io.rs
+++ b/crates/asap-aware-mapping/src/storage_io.rs
@@ -121,7 +121,7 @@ pub fn estimate_storage_io(
     profile: &StorageIoProfile,
     evidence_version: &str,
 ) -> Result<StorageEstimate, AnalyticalCostError> {
-    // Also prove source coverage, edge consistency, execution legality and DAG
+    // Also prove scan selection, edge consistency, execution legality and DAG
     // identity before using supplementary deployment evidence.
     estimate_physical_dag(&dag.nodes, &dag.root, scope, dag)?;
     let evaluations = scope.validate()?;
diff --git a/crates/asap-aware-mapping/src/summary_maintenance_cost/estimator.rs b/crates/asap-aware-mapping/src/summary_maintenance_cost/estimator.rs
index 5bf42d2e4..eac94a2cb 100644
--- a/crates/asap-aware-mapping/src/summary_maintenance_cost/estimator.rs
+++ b/crates/asap-aware-mapping/src/summary_maintenance_cost/estimator.rs
@@ -86,14 +86,14 @@ pub(super) fn estimate_heterogeneous_summary(
         };
         let inputs = node_evidence.inputs.validate()?;
         validate_arrival_rate(scope.data_arrival, inputs.ingestion_rate_per_second)?;
-        match node_evidence.source_coverage_index {
+        match node_evidence.scan_selection_index {
             Some(index) => {
                 let declared =
                     scope
                         .sources
                         .get(index)
                         .ok_or(AnalyticalCostError::MissingComparisonScope(
-                            "summary source coverage",
+                            "summary scan selection",
                         ))?;
                 if !matches!(&child.expr, SummaryExpr::KeepPreAsap(_))
                     || inputs.initial_input_rows != raw.planning_time_input_rows
@@ -178,7 +178,7 @@ pub(super) fn estimate_heterogeneous_summary(
             let bootstrap_extra_rows = bootstrap
                 .checked_sub(inputs.initial_input_rows)
                 .ok_or(AnalyticalCostError::Overflow)?;
-            let source_scan_bytes = if node_evidence.source_coverage_index.is_some() {
+            let source_scan_bytes = if node_evidence.scan_selection_index.is_some() {
                 inputs
                     .initial_source_scan_bytes
                     .checked_add(
@@ -223,7 +223,7 @@ pub(super) fn estimate_heterogeneous_summary(
                 .checked_add(state_bytes)
                 .ok_or(AnalyticalCostError::Overflow)?;
         }
-        if let Some(source_index) = node_evidence.source_coverage_index {
+        if let Some(source_index) = node_evidence.scan_selection_index {
             if node_evidence.bootstrap_read_identity.is_empty() {
                 return Err(AnalyticalCostError::MissingOrStale(
                     "bootstrap_read_identity",
diff --git a/crates/asap-aware-mapping/src/summary_maintenance_cost/evidence.rs b/crates/asap-aware-mapping/src/summary_maintenance_cost/evidence.rs
index f396e3cd8..7c8607685 100644
--- a/crates/asap-aware-mapping/src/summary_maintenance_cost/evidence.rs
+++ b/crates/asap-aware-mapping/src/summary_maintenance_cost/evidence.rs
@@ -161,7 +161,7 @@ pub struct SummaryAggregateEvidence {
     /// Index into `ComparisonScope.sources` when this state bootstraps directly
     /// from storage. `None` means its input is an already-materialized child
     /// edge and therefore has no additional source read.
-    pub source_coverage_index: Option<usize>,
+    pub scan_selection_index: Option<usize>,
     /// Provider-owned identity of the physical bootstrap read. Equal source
     /// coverage alone does not prove two independent builds share I/O.
     pub bootstrap_read_identity: String,
diff --git a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs
index 331b7e1a5..7859db9b8 100644
--- a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs
+++ b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs
@@ -165,19 +165,19 @@ fn validate_physical_scope_coverage(
         .filter(|node| node.operator == PhysicalOperator::Scan)
     {
         let coverage = node
-            .source_coverage
+            .scan_selection
             .as_ref()
-            .ok_or_else(|| AnalyticalCostError::MissingScanSourceCoverage(node.id.clone()))?;
+            .ok_or_else(|| AnalyticalCostError::MissingScanSelection(node.id.clone()))?;
         let Some(index) = scope.sources.iter().position(|value| value == coverage) else {
             return Err(AnalyticalCostError::ComparisonScopeMismatch(
-                "physical source coverage",
+                "physical scan selection",
             ));
         };
         covered.insert(index);
     }
     if covered.len() != scope.sources.len() {
         return Err(AnalyticalCostError::ComparisonScopeMismatch(
-            "physical source coverage",
+            "physical scan selection",
         ));
     }
     Ok(())
@@ -939,7 +939,7 @@ mod tests {
             query,
             asap_types::workload::TimestampMs(planning_time_ms),
             asap_types::workload::DurationMs(horizon_ms),
-            vec![crate::physical_operator_statistics::SourceCoverage {
+            vec![crate::physical_operator_statistics::ScanSelection {
                 source: Source::TimeSeries {
                     metric: "metrics".into(),
                 },
@@ -1611,7 +1611,7 @@ mod tests {
                 id: "raw-concat".into(),
                 operator: PhysicalOperator::Concat,
                 children: vec!["raw-scan".into(), "raw-scan-2".into()],
-                source_coverage: None,
+                scan_selection: None,
                 output_buffer_bytes: 0,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::Once,
@@ -1693,7 +1693,7 @@ mod tests {
         let mut extra = streaming_scope();
         extra
             .sources
-            .push(crate::physical_operator_statistics::SourceCoverage {
+            .push(crate::physical_operator_statistics::ScanSelection {
                 source: Source::TimeSeries {
                     metric: "unused".into(),
                 },
@@ -1720,7 +1720,7 @@ mod tests {
         let mut info_scope = streaming_scope();
         info_scope
             .sources
-            .push(crate::physical_operator_statistics::SourceCoverage {
+            .push(crate::physical_operator_statistics::ScanSelection {
                 source: Source::TimeSeries {
                     metric: "target_info".into(),
                 },
@@ -2057,7 +2057,7 @@ mod tests {
                 physical_id: "left-state".into(),
                 input: test_edge(),
                 output: test_edge(),
-                source_coverage_index: Some(0),
+                scan_selection_index: Some(0),
                 bootstrap_read_identity: "left-bootstrap".into(),
                 inputs: streaming_inputs(),
                 insert_cpu_ops: streaming_cpu().insert_cpu_ops.unwrap(),
@@ -2084,7 +2084,7 @@ mod tests {
                 physical_id: "right-state".into(),
                 input: test_edge(),
                 output: test_edge(),
-                source_coverage_index: Some(0),
+                scan_selection_index: Some(0),
                 bootstrap_read_identity: "right-bootstrap".into(),
                 inputs: second_inputs,
                 insert_cpu_ops: second_cpu.insert_cpu_ops.unwrap(),
@@ -2150,7 +2150,7 @@ mod tests {
             64.0
         );
 
-        // Equal SourceCoverage does not imply that two independent state
+        // Equal ScanSelection does not imply that two independent state
         // builds share one physical read. Only a provider-owned read identity
         // permits scan de-duplication.
         let mut shared_read = model;
@@ -3264,7 +3264,7 @@ mod tests {
             id: "raw-scan".into(),
             operator: PhysicalOperator::Scan,
             children: vec![],
-            source_coverage: Some(scope.sources[0].clone()),
+            scan_selection: Some(scope.sources[0].clone()),
             output_buffer_bytes: 0,
             retained_bytes: 0,
             execution: ExecutionMultiplicity::Once,
@@ -3415,7 +3415,7 @@ mod tests {
                     physical_id: format!("agg-{node:p}"),
                     input: test_edge(),
                     output: test_edge(),
-                    source_coverage_index: source_root.then_some(0),
+                    scan_selection_index: source_root.then_some(0),
                     bootstrap_read_identity: if source_root {
                         "shared-bootstrap".into()
                     } else {
@@ -3637,7 +3637,7 @@ mod tests {
             &entry,
             asap_types::workload::TimestampMs(0),
             asap_types::workload::DurationMs(5_000),
-            vec![crate::physical_operator_statistics::SourceCoverage {
+            vec![crate::physical_operator_statistics::ScanSelection {
                 source: Source::TimeSeries {
                     metric: "metrics".into(),
                 },
diff --git a/crates/asap-aware-mapping/tests/physical_handoff_cost.rs b/crates/asap-aware-mapping/tests/physical_handoff_cost.rs
index 0885db71c..a8a9ed42d 100644
--- a/crates/asap-aware-mapping/tests/physical_handoff_cost.rs
+++ b/crates/asap-aware-mapping/tests/physical_handoff_cost.rs
@@ -3,7 +3,7 @@ use asap_aware_mapping::analytical_cost::{
     PhysicalOperator,
 };
 use asap_aware_mapping::physical_operator_statistics::{
-    ComparisonScope, EdgeStatistics, OperatorStatistics, SourceCoverage, UnaryEdgeStatistics,
+    ComparisonScope, EdgeStatistics, OperatorStatistics, ScanSelection, UnaryEdgeStatistics,
 };
 use asap_types::pre_asap::query_expr::Source;
 use asap_types::workload::{
@@ -12,7 +12,7 @@ use asap_types::workload::{
 use std::collections::HashMap;
 
 fn fixture() -> (EvidenceBackedPhysicalDAG, ComparisonScope) {
-    let coverage = SourceCoverage {
+    let coverage = ScanSelection {
         source: Source::Table {
             table_ref: "events".into(),
         },
@@ -49,7 +49,7 @@ fn fixture() -> (EvidenceBackedPhysicalDAG, ComparisonScope) {
             id: "scan".into(),
             operator: PhysicalOperator::Scan,
             children: vec![],
-            source_coverage: Some(coverage),
+            scan_selection: Some(coverage),
             output_buffer_bytes: 0,
             retained_bytes: 0,
             execution: ExecutionMultiplicity::PerEvaluation,
@@ -58,7 +58,7 @@ fn fixture() -> (EvidenceBackedPhysicalDAG, ComparisonScope) {
             id: "left".into(),
             operator: PhysicalOperator::PassThrough,
             children: vec!["scan".into()],
-            source_coverage: None,
+            scan_selection: None,
             output_buffer_bytes: 0,
             retained_bytes: 0,
             execution: ExecutionMultiplicity::PerEvaluation,
@@ -67,7 +67,7 @@ fn fixture() -> (EvidenceBackedPhysicalDAG, ComparisonScope) {
             id: "right".into(),
             operator: PhysicalOperator::PassThrough,
             children: vec!["scan".into()],
-            source_coverage: None,
+            scan_selection: None,
             output_buffer_bytes: 0,
             retained_bytes: 0,
             execution: ExecutionMultiplicity::PerEvaluation,
@@ -76,7 +76,7 @@ fn fixture() -> (EvidenceBackedPhysicalDAG, ComparisonScope) {
             id: "root".into(),
             operator: PhysicalOperator::Concat,
             children: vec!["left".into(), "right".into()],
-            source_coverage: None,
+            scan_selection: None,
             output_buffer_bytes: 0,
             retained_bytes: 0,
             execution: ExecutionMultiplicity::PerEvaluation,
diff --git a/crates/asap-aware-mapping/tests/storage_io.rs b/crates/asap-aware-mapping/tests/storage_io.rs
index a12118f09..e9290d20a 100644
--- a/crates/asap-aware-mapping/tests/storage_io.rs
+++ b/crates/asap-aware-mapping/tests/storage_io.rs
@@ -3,7 +3,7 @@ use asap_aware_mapping::analytical_cost::{
     PhysicalOperator,
 };
 use asap_aware_mapping::physical_operator_statistics::{
-    ComparisonScope, EdgeStatistics, OperatorStatistics, SourceCoverage, UnaryEdgeStatistics,
+    ComparisonScope, EdgeStatistics, OperatorStatistics, ScanSelection, UnaryEdgeStatistics,
 };
 use asap_types::pre_asap::query_expr::Source;
 use asap_types::workload::{
@@ -12,7 +12,7 @@ use asap_types::workload::{
 use std::collections::HashMap;
 
 fn fixture() -> (EvidenceBackedPhysicalDAG, ComparisonScope) {
-    let coverage = SourceCoverage {
+    let coverage = ScanSelection {
         source: Source::Table {
             table_ref: "events".into(),
         },
@@ -49,7 +49,7 @@ fn fixture() -> (EvidenceBackedPhysicalDAG, ComparisonScope) {
             id: "scan".into(),
             operator: PhysicalOperator::Scan,
             children: vec![],
-            source_coverage: Some(coverage),
+            scan_selection: Some(coverage),
             output_buffer_bytes: 0,
             retained_bytes: 0,
             execution: ExecutionMultiplicity::PerEvaluation,
@@ -58,7 +58,7 @@ fn fixture() -> (EvidenceBackedPhysicalDAG, ComparisonScope) {
             id: "left".into(),
             operator: PhysicalOperator::PassThrough,
             children: vec!["scan".into()],
-            source_coverage: None,
+            scan_selection: None,
             output_buffer_bytes: 0,
             retained_bytes: 0,
             execution: ExecutionMultiplicity::PerEvaluation,
@@ -67,7 +67,7 @@ fn fixture() -> (EvidenceBackedPhysicalDAG, ComparisonScope) {
             id: "right".into(),
             operator: PhysicalOperator::PassThrough,
             children: vec!["scan".into()],
-            source_coverage: None,
+            scan_selection: None,
             output_buffer_bytes: 0,
             retained_bytes: 0,
             execution: ExecutionMultiplicity::PerEvaluation,
@@ -76,7 +76,7 @@ fn fixture() -> (EvidenceBackedPhysicalDAG, ComparisonScope) {
             id: "root".into(),
             operator: PhysicalOperator::Concat,
             children: vec!["left".into(), "right".into()],
-            source_coverage: None,
+            scan_selection: None,
             output_buffer_bytes: 0,
             retained_bytes: 0,
             execution: ExecutionMultiplicity::PerEvaluation,
@@ -353,7 +353,7 @@ fn storage_node_identity_statistics_and_calibration_provenance_are_bound() {
         .get_mut("scan")
         .unwrap()
         .node
-        .source_coverage
+        .scan_selection
         .as_mut()
         .unwrap()
         .source_snapshot_id = "another-source".into();
diff --git a/crates/devtools/src/bin/dag_export.rs b/crates/devtools/src/bin/dag_export.rs
index b0615f017..2728dab12 100644
--- a/crates/devtools/src/bin/dag_export.rs
+++ b/crates/devtools/src/bin/dag_export.rs
@@ -152,7 +152,7 @@ struct ComparisonScopeEvidence {
     time_scope: String,
     lookback_ms: Option<u64>,
     as_of_ms: Option<u64>,
-    sources: Vec<asap_aware_mapping::physical_operator_statistics::SourceCoverage>,
+    sources: Vec<asap_aware_mapping::physical_operator_statistics::ScanSelection>,
     #[serde(default = "CacheProfile::no_cache")]
     cache_profile: CacheProfile,
 }
@@ -1700,7 +1700,7 @@ mod tests {
         ExecutionMultiplicity, PhysicalDAGNode, PhysicalOperator,
     };
     use asap_aware_mapping::physical_operator_statistics::{
-        EdgeStatistics, OperatorStatistics, SourceCoverage, UnaryEdgeStatistics,
+        EdgeStatistics, OperatorStatistics, ScanSelection, UnaryEdgeStatistics,
     };
     use asap_aware_mapping::query_physical_lowering::lower_query_physical_dag;
     use asap_devtools::PromqlError;
@@ -2178,7 +2178,7 @@ mod tests {
             time_scope: "longitudinal".into(),
             lookback_ms: Some(10_000),
             as_of_ms: Some(1_000),
-            sources: vec![SourceCoverage {
+            sources: vec![ScanSelection {
                 source: Source::Table {
                     table_ref: "events".into(),
                 },
@@ -2261,7 +2261,7 @@ mod tests {
                 id: "summary-read".into(),
                 operator: PhysicalOperator::Scan,
                 children: vec![],
-                source_coverage: Some(coverage),
+                scan_selection: Some(coverage),
                 output_buffer_bytes: 2_400,
                 retained_bytes: 0,
                 execution: ExecutionMultiplicity::PerEvaluation,
diff --git a/crates/types/src/ir/cse.rs b/crates/types/src/ir/cse.rs
index 821da2395..b9046a1ab 100644
--- a/crates/types/src/ir/cse.rs
+++ b/crates/types/src/ir/cse.rs
@@ -112,6 +112,7 @@ pub fn structural_hash(node: &OperatorNode, cache: &mut HashCache) -> u64 {
         &node.schema,
         &node.guarantee,
         node.timing,
+        &node.coverage,
     );
     serde_json::to_string(&own)
         .unwrap_or_default()
@@ -167,6 +168,7 @@ fn same_node(left: &OperatorNode, right: &OperatorNode, memo: &mut EqMemo) -> bo
         && left.result_kind == right.result_kind
         && left.schema == right.schema
         && left.timing == right.timing
+        && left.coverage == right.coverage
         && same_value(&left.guarantee, &right.guarantee)
         && same_value(&own_fields(left), &own_fields(right))
 }
@@ -254,6 +256,7 @@ fn intern_bottom_up(
         schema: node.schema.clone(),
         guarantee: node.guarantee.clone(),
         timing: node.timing,
+        coverage: node.coverage.clone(),
     };
     let interned = table.intern(rebuilt);
     visited.insert(Rc::as_ptr(node), (Rc::clone(node), Rc::clone(&interned)));
diff --git a/crates/types/src/ir/error.rs b/crates/types/src/ir/error.rs
index 4b6849965..a0f9e2271 100644
--- a/crates/types/src/ir/error.rs
+++ b/crates/types/src/ir/error.rs
@@ -16,4 +16,6 @@ pub enum SchemaDerivationError {
     EmptyConcat,
     #[error("invalid per-series sample column: {0}")]
     InvalidSampleColumn(String),
+    #[error("invalid summary coverage: {0}")]
+    Coverage(#[from] super::summary_coverage::CoverageError),
 }
diff --git a/crates/types/src/ir/flat.rs b/crates/types/src/ir/flat.rs
index 434a3406f..830a294db 100644
--- a/crates/types/src/ir/flat.rs
+++ b/crates/types/src/ir/flat.rs
@@ -15,6 +15,7 @@ use serde::{Deserialize, Serialize};
 
 use super::node::{Operator, OperatorNode, OperatorResultKind};
 use super::query::QueryRoot;
+use super::summary_coverage::SummaryCoverage;
 use crate::post_asap::execution_data_state::ExecutionTiming;
 use crate::post_asap::guarantee::ResultGuarantee;
 use crate::pre_asap::schema::Schema;
@@ -30,6 +31,8 @@ pub struct FlatNode {
     pub schema: Schema,
     pub guarantee: Option<ResultGuarantee>,
     pub timing: Option<ExecutionTiming>,
+    #[serde(default)]
+    pub coverage: Option<SummaryCoverage>,
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -77,6 +80,7 @@ fn visit(
         schema: node.schema.clone(),
         guarantee: node.guarantee.clone(),
         timing: node.timing,
+        coverage: node.coverage.clone(),
     });
     originals.push(Rc::clone(node));
     ids.insert(Rc::as_ptr(node), id);
diff --git a/crates/types/src/ir/mod.rs b/crates/types/src/ir/mod.rs
index e42ad375d..4bc2b9f26 100644
--- a/crates/types/src/ir/mod.rs
+++ b/crates/types/src/ir/mod.rs
@@ -19,3 +19,5 @@ pub use scalar::{ExprSemantics, Predicate, ProjectItem, ScalarExpr, SortKey};
 pub mod canonicalize;
 pub mod cse;
 pub mod flat;
+/// Semantic observation coverage, separate from field layout and physical timing.
+pub mod summary_coverage;
diff --git a/crates/types/src/ir/node.rs b/crates/types/src/ir/node.rs
index a2c6c789c..7534353b0 100644
--- a/crates/types/src/ir/node.rs
+++ b/crates/types/src/ir/node.rs
@@ -10,6 +10,7 @@ use serde::{Deserialize, Serialize};
 
 use super::asap::ASAPOp;
 use super::non_asap::NonASAPOp;
+use super::summary_coverage::{CoverageError, SummaryCoverage};
 use crate::ir::SchemaDerivationError;
 use crate::post_asap::execution_data_state::ExecutionTiming;
 use crate::post_asap::guarantee::ResultGuarantee;
@@ -99,6 +100,8 @@ pub struct OperatorNode {
     pub schema: Schema,
     pub guarantee: Option<ResultGuarantee>,
     pub timing: Option<ExecutionTiming>,
+    #[serde(default)]
+    pub coverage: Option<SummaryCoverage>,
 }
 
 impl OperatorNode {
@@ -121,6 +124,7 @@ impl OperatorNode {
             schema,
             guarantee: None,
             timing: None,
+            coverage: None,
         }
     }
 
@@ -141,6 +145,25 @@ impl OperatorNode {
         self
     }
 
+    /// Attach caller-established coverage. Required on summary nodes; see
+    /// [`Self::requires_coverage`].
+    pub fn with_coverage(
+        mut self,
+        coverage: SummaryCoverage,
+    ) -> Result<Self, SchemaDerivationError> {
+        coverage.validate()?;
+        if self.result_kind != OperatorResultKind::State {
+            return Err(CoverageError::NotState.into());
+        }
+        self.coverage = Some(coverage);
+        Ok(self)
+    }
+
+    /// Summary nodes whose state can be composed must declare coverage.
+    pub fn requires_coverage(&self) -> bool {
+        matches!(self.asap(), Some(ASAPOp::SummaryAgg { .. }))
+    }
+
     pub fn non_asap(&self) -> Option<&NonASAPOp> {
         match &self.operator {
             Operator::NonASAP(op) => Some(op),
@@ -290,6 +313,13 @@ impl OperatorNode {
                     "invalid time or identity column in schema".into(),
                 ));
             }
+            match &node.coverage {
+                Some(coverage) => {
+                    (*node.as_ref()).clone().with_coverage(coverage.clone())?;
+                }
+                None if node.requires_coverage() => return Err(CoverageError::Missing.into()),
+                None => {}
+            }
             node.operator.validate_inputs()?;
             if node.result_kind != node.operator.output_kind() {
                 return Err(SchemaDerivationError::InvalidScalarSignature(
diff --git a/crates/types/src/ir/summary_coverage.rs b/crates/types/src/ir/summary_coverage.rs
new file mode 100644
index 000000000..3f3798a88
--- /dev/null
+++ b/crates/types/src/ir/summary_coverage.rs
@@ -0,0 +1,127 @@
+//! Joint time/population coverage for summary composition, independent of schema.
+//! Equality predicates are a deliberately narrow proof vocabulary. Unsupported
+//! predicates cannot be declared disjoint merely by giving them different names.
+use crate::pre_asap::Source;
+use serde::{Deserialize, Serialize};
+use std::collections::BTreeMap;
+use std::ops::Range;
+use thiserror::Error;
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct SummaryCoverage {
+    /// Observation data source, as named by `Scan`: a table or a time series.
+    /// Region time bounds refer to its time column.
+    pub source: Source,
+    /// Union of joint regions; never the Cartesian product of independent bounds.
+    pub regions: Vec<CoverageRegion>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct CoverageRegion {
+    /// Half-open absolute bounds on the source's time column, in milliseconds.
+    /// Only a materialized state (for example an ingested pane) has absolute
+    /// bounds; a state in a logical plan that is not yet evaluated uses `None`,
+    /// as does a source without a time column. `None` means no time restriction.
+    pub time_ms: Option<Range<i64>>,
+    /// Conjunction of `field = 'text'` equality predicates (PromQL label
+    /// matchers, SQL text columns), keyed by field name. Only Utf8 equality is
+    /// represented, so values are strings; any other restriction is not
+    /// recorded here. Empty means unrestricted.
+    pub population: BTreeMap<String, String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Error)]
+pub enum CoverageError {
+    #[error("coverage interval must have start < end")]
+    InvalidInterval,
+    #[error("population dimension names cannot be empty")]
+    InvalidPopulation,
+    #[error("summary coverage sources differ")]
+    SourceMismatch,
+    #[error("coverage overlap is not proven absent")]
+    PossibleOverlap,
+    #[error("coverage merge requires at least one input")]
+    EmptyMerge,
+    #[error("summary coverage requires state output")]
+    NotState,
+    #[error("summary node requires coverage")]
+    Missing,
+}
+
+impl SummaryCoverage {
+    pub fn validate(&self) -> Result<(), CoverageError> {
+        for (index, region) in self.regions.iter().enumerate() {
+            if region.time_ms.as_ref().is_some_and(Range::is_empty) {
+                return Err(CoverageError::InvalidInterval);
+            }
+            if region.population.keys().any(String::is_empty) {
+                return Err(CoverageError::InvalidPopulation);
+            }
+            if self.regions[..index]
+                .iter()
+                .any(|other| region.may_overlap(other))
+            {
+                return Err(CoverageError::PossibleOverlap);
+            }
+        }
+        Ok(())
+    }
+
+    /// Every observation in a region is assumed to contribute once to the state.
+    /// Compose once-per-observation summaries only when their joint regions are
+    /// provably disjoint. Update/reduction compatibility, family merge capability
+    /// and accuracy are checked by `SummaryMerge`, not here.
+    pub fn merge_disjoint(inputs: &[Self]) -> Result<Self, CoverageError> {
+        let first = inputs.first().ok_or(CoverageError::EmptyMerge)?;
+        let mut merged = first.clone();
+        merged.regions.clear();
+        for input in inputs {
+            input.validate()?;
+            if input.source != first.source {
+                return Err(CoverageError::SourceMismatch);
+            }
+            merged.regions.extend(input.regions.iter().cloned());
+        }
+        merged.validate()?;
+        // Coalesce adjacent intervals only for identical population predicates.
+        merged.regions.sort_by(|a, b| {
+            a.population.cmp(&b.population).then(
+                a.time_ms
+                    .as_ref()
+                    .map(|t| t.start)
+                    .cmp(&b.time_ms.as_ref().map(|t| t.start)),
+            )
+        });
+        let mut normalized: Vec<CoverageRegion> = Vec::new();
+        for region in merged.regions {
+            if let Some(last) = normalized.last_mut() {
+                if let (Some(last_time), Some(time)) = (&mut last.time_ms, &region.time_ms) {
+                    if last.population == region.population && last_time.end == time.start {
+                        last_time.end = time.end;
+                        continue;
+                    }
+                }
+            }
+            normalized.push(region);
+        }
+        merged.regions = normalized;
+        Ok(merged)
+    }
+}
+impl CoverageRegion {
+    fn may_overlap(&self, other: &Self) -> bool {
+        let time_overlaps = match (&self.time_ms, &other.time_ms) {
+            (Some(a), Some(b)) => a.start < b.end && b.start < a.end,
+            _ => true,
+        };
+        time_overlaps
+            && !self.population.iter().any(|(dimension, value)| {
+                other
+                    .population
+                    .get(dimension)
+                    .is_some_and(|other| other != value)
+            })
+    }
+}
diff --git a/crates/types/tests/schema_rebuilding.rs b/crates/types/tests/schema_rebuilding.rs
index 1f1b57dbc..fc7a3f145 100644
--- a/crates/types/tests/schema_rebuilding.rs
+++ b/crates/types/tests/schema_rebuilding.rs
@@ -1,3 +1,4 @@
+use asap_types::ir::summary_coverage::{CoverageRegion, SummaryCoverage};
 use asap_types::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode};
 use asap_types::post_asap::{
     ExactKind, ExactParams, ExecutionTiming, GroupingStrategy, ResultGuarantee, SummaryUpdate,
@@ -7,6 +8,27 @@ use asap_types::pre_asap::{
 };
 use std::rc::Rc;
 
+fn coverage() -> SummaryCoverage {
+    SummaryCoverage {
+        source: Source::Table {
+            table_ref: "t".into(),
+        },
+        regions: vec![CoverageRegion {
+            time_ms: None,
+            population: Default::default(),
+        }],
+    }
+}
+/// Rewrites clear coverage; a rewriter must declare it again for summary nodes.
+fn redeclare(node: OperatorNode) -> Rc<OperatorNode> {
+    let node = Rc::new(node);
+    if !node.requires_coverage() {
+        return node;
+    }
+    assert!(node.validate_structure().is_err());
+    Rc::new((*node).clone().with_coverage(coverage()).unwrap())
+}
+
 fn scan(key_type: DataType, name: &str) -> Rc<OperatorNode> {
     OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan {
         source: Source::Table {
@@ -40,7 +62,12 @@ fn aggregate(child: Rc<OperatorNode>, asap: bool) -> Rc<OperatorNode> {
             having: None,
         })
     };
-    OperatorNode::new_shared(operator).unwrap()
+    let node = OperatorNode::new(operator).unwrap();
+    Rc::new(if asap {
+        node.with_coverage(coverage()).unwrap()
+    } else {
+        node
+    })
 }
 
 /// Rewrites follow changed input types and inherited names for either category.
@@ -50,7 +77,7 @@ fn rebuilding_rederives_schema_for_both_categories() {
         let original = aggregate(scan(DataType::Int64, "key"), asap);
         original.validate_structure().unwrap();
         let replacement = scan(DataType::Utf8, "new_key");
-        let rebuilt = Rc::new(original.map_children(|_| replacement.clone()).unwrap());
+        let rebuilt = redeclare(original.map_children(|_| replacement.clone()).unwrap());
         assert_eq!(rebuilt.schema, rebuilt.operator.output_schema().unwrap());
         rebuilt.validate_structure().unwrap();
     }
@@ -64,14 +91,14 @@ fn rebuilding_preserves_only_explicit_naming_overrides() {
         let mut schema = original.schema.clone();
         schema.fields[0].name = "alias".into();
         schema.fields[0].table = Some("result".into());
-        let original = Rc::new(
-            OperatorNode::with_schema(original.operator.clone(), schema)
-                .with_guarantee(Some(ResultGuarantee::exact("fixture")))
-                .with_timing(Some(ExecutionTiming::QueryTime)),
-        );
+        let mut renamed = OperatorNode::with_schema(original.operator.clone(), schema)
+            .with_guarantee(Some(ResultGuarantee::exact("fixture")))
+            .with_timing(Some(ExecutionTiming::QueryTime));
+        renamed.coverage = original.coverage.clone();
+        let original = Rc::new(renamed);
         original.validate_structure().unwrap();
         let replacement = scan(DataType::Utf8, "new_key");
-        let rebuilt = Rc::new(original.map_children(|_| replacement.clone()).unwrap());
+        let rebuilt = redeclare(original.map_children(|_| replacement.clone()).unwrap());
         assert_eq!(rebuilt.schema.fields[0].name, "alias");
         assert_eq!(rebuilt.schema.fields[0].table.as_deref(), Some("result"));
         assert_eq!(
@@ -110,7 +137,9 @@ fn validation_rejects_structural_overrides_for_both_categories() {
         schema.fields.pop();
         invalid.push(schema);
         for schema in invalid {
-            let forged = Rc::new(OperatorNode::with_schema(original.operator.clone(), schema));
+            let mut forged = OperatorNode::with_schema(original.operator.clone(), schema);
+            forged.coverage = original.coverage.clone();
+            let forged = Rc::new(forged);
             assert!(
                 forged.validate_structure().is_err(),
                 "accepted structural override: {:?}",
diff --git a/crates/types/tests/structure_contract.rs b/crates/types/tests/structure_contract.rs
index d4e164f71..d7c144f4e 100644
--- a/crates/types/tests/structure_contract.rs
+++ b/crates/types/tests/structure_contract.rs
@@ -13,6 +13,19 @@ fn scan() -> Rc<OperatorNode> {
     }))
     .unwrap()
 }
+/// Tabular coverage for a whole-table summary.
+fn whole_table() -> asap_types::ir::summary_coverage::SummaryCoverage {
+    use asap_types::ir::summary_coverage::{CoverageRegion, SummaryCoverage};
+    SummaryCoverage {
+        source: Source::Table {
+            table_ref: "t".into(),
+        },
+        regions: vec![CoverageRegion {
+            time_ms: None,
+            population: Default::default(),
+        }],
+    }
+}
 /// Resolved filters cannot hide invalid scalar types or out-of-scope columns.
 #[test]
 fn invalid_predicates_are_rejected() {
@@ -103,6 +116,8 @@ fn state_evaluations_and_passthrough_keep_their_contracts() {
             grouping: GroupingStrategy::default(),
             filter: None,
         }))
+        .unwrap()
+        .with_coverage(whole_table())
         .unwrap(),
     );
     state.validate_structure().unwrap();
@@ -173,15 +188,19 @@ fn shared_construction_derives_both_operator_categories() {
     use asap_types::pre_asap::{ColumnRef, FieldDataType, Reduction};
 
     let input = scan();
-    let state = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg {
-        child: input.clone(),
-        family: FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum),
-        input: SummaryUpdate::column(ColumnRef::Named("x".into())),
-        reduction: Reduction::by(vec![]),
-        grouping: GroupingStrategy::default(),
-        filter: None,
-    }))
-    .unwrap();
+    let state = Rc::new(
+        OperatorNode::new(Operator::ASAP(ASAPOp::SummaryAgg {
+            child: input.clone(),
+            family: FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum),
+            input: SummaryUpdate::column(ColumnRef::Named("x".into())),
+            reduction: Reduction::by(vec![]),
+            grouping: GroupingStrategy::default(),
+            filter: None,
+        }))
+        .unwrap()
+        .with_coverage(whole_table())
+        .unwrap(),
+    );
     assert_eq!(state.result_kind, OperatorResultKind::State);
     assert!(!state.schema.fields.last().unwrap().is_plain());
     assert!(state.guarantee.is_none());
diff --git a/crates/types/tests/summary_coverage.rs b/crates/types/tests/summary_coverage.rs
new file mode 100644
index 000000000..3b99e9bba
--- /dev/null
+++ b/crates/types/tests/summary_coverage.rs
@@ -0,0 +1,150 @@
+//! Coverage composition preserves gaps and rejects duplicate observations.
+use asap_types::{
+    ir::operator_properties::Reduction,
+    ir::summary_coverage::*,
+    post_asap::SummaryUpdate,
+    pre_asap::{ColumnRef, Source},
+};
+fn table(name: &str) -> Source {
+    Source::Table {
+        table_ref: name.into(),
+    }
+}
+fn coverage(start: i64, end: i64, population: &[(&str, &str)]) -> SummaryCoverage {
+    SummaryCoverage {
+        source: table("flows"),
+        regions: vec![CoverageRegion {
+            time_ms: Some(start..end),
+            population: population
+                .iter()
+                .map(|(k, v)| (k.to_string(), v.to_string()))
+                .collect(),
+        }],
+    }
+}
+/// Adjacent panes coalesce; gaps remain disconnected rather than becoming a hull.
+#[test]
+fn time_union_preserves_gaps() {
+    let merged =
+        SummaryCoverage::merge_disjoint(&[coverage(0, 1, &[]), coverage(1, 2, &[])]).unwrap();
+    assert_eq!(merged.regions[0].time_ms, Some(0..2));
+    assert_eq!(merged.regions.len(), 1);
+    let gapped =
+        SummaryCoverage::merge_disjoint(&[coverage(0, 1, &[]), coverage(2, 3, &[])]).unwrap();
+    assert_eq!(gapped.regions.len(), 2);
+}
+/// Population partitions can overlap in time without sharing observations.
+#[test]
+fn population_and_joint_union() {
+    let merged = SummaryCoverage::merge_disjoint(&[
+        coverage(0, 2, &[("region", "us")]),
+        coverage(0, 2, &[("region", "eu")]),
+    ])
+    .unwrap();
+    assert_eq!(merged.regions.len(), 2);
+    let joint = SummaryCoverage::merge_disjoint(&[
+        coverage(0, 1, &[("region", "us")]),
+        coverage(1, 2, &[("region", "eu")]),
+    ])
+    .unwrap();
+    assert_eq!(joint.regions.len(), 2);
+    let decoded: SummaryCoverage =
+        serde_json::from_str(&serde_json::to_string(&joint).unwrap()).unwrap();
+    assert_eq!(decoded, joint);
+}
+/// Intersecting predicates and windows cannot authorize once-per-observation merge.
+#[test]
+fn overlap_and_identity_fail_closed() {
+    assert_eq!(
+        SummaryCoverage::merge_disjoint(&[coverage(0, 2, &[]), coverage(1, 3, &[])]),
+        Err(CoverageError::PossibleOverlap)
+    );
+    assert_eq!(
+        SummaryCoverage::merge_disjoint(&[
+            coverage(0, 2, &[("region", "us")]),
+            coverage(0, 2, &[("tier", "premium")])
+        ]),
+        Err(CoverageError::PossibleOverlap)
+    );
+    assert_eq!(
+        SummaryCoverage::merge_disjoint(&[
+            coverage(0, 2, &[("region", "us")]),
+            coverage(0, 2, &[("region", "us")])
+        ]),
+        Err(CoverageError::PossibleOverlap)
+    );
+    let mut other = coverage(1, 2, &[]);
+    other.source = table("other-flows");
+    assert_eq!(
+        SummaryCoverage::merge_disjoint(&[coverage(0, 1, &[]), other]),
+        Err(CoverageError::SourceMismatch)
+    );
+    assert_eq!(
+        coverage(2, 1, &[]).validate(),
+        Err(CoverageError::InvalidInterval)
+    );
+}
+
+/// Coverage is logical state metadata, and input rewrites invalidate its proof.
+#[test]
+fn node_coverage_is_required_checked_and_cleared_by_rewrites() {
+    use asap_types::{
+        ir::{ASAPOp, NonASAPOp, Operator, OperatorNode},
+        post_asap::{SketchAlgorithm, SketchKind, SketchParams},
+        pre_asap::{DataType, Field, FieldDataType, Schema},
+    };
+    let raw = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan {
+        source: table("flows"),
+        predicates: vec![],
+        schema: Schema::new(vec![Field::plain("latency", DataType::Float64, false)]),
+    }))
+    .unwrap();
+    let declared = coverage(0, 1, &[]);
+    assert!((*raw).clone().with_coverage(declared.clone()).is_err());
+    let state = OperatorNode::new(Operator::ASAP(ASAPOp::SummaryAgg {
+        child: raw,
+        family: FieldDataType::Sketch(
+            SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }),
+            Default::default(),
+        ),
+        input: SummaryUpdate::column(ColumnRef::Named("latency".into())),
+        reduction: Reduction::by(vec![]),
+        grouping: Default::default(),
+        filter: None,
+    }))
+    .unwrap();
+    // Summary nodes cannot validate without coverage.
+    assert!(matches!(
+        std::rc::Rc::new(state.clone()).validate_structure(),
+        Err(asap_types::ir::SchemaDerivationError::Coverage(
+            CoverageError::Missing
+        ))
+    ));
+    let state = state.with_coverage(declared.clone()).unwrap();
+    std::rc::Rc::new(state.clone())
+        .validate_structure()
+        .unwrap();
+    let rebuilt = state.map_children(Clone::clone).unwrap();
+    assert!(rebuilt.coverage.is_none());
+}
+
+/// Sources without a time column declare no time bounds; such a region overlaps
+/// any region it is not population-disjoint from.
+#[test]
+fn regions_without_time_bounds() {
+    let mut tabular = coverage(0, 1, &[("region", "us")]);
+    tabular.regions[0].time_ms = None;
+    let mut other = coverage(0, 1, &[("region", "eu")]);
+    other.regions[0].time_ms = None;
+    assert_eq!(
+        SummaryCoverage::merge_disjoint(&[tabular.clone(), other])
+            .unwrap()
+            .regions
+            .len(),
+        2
+    );
+    assert_eq!(
+        SummaryCoverage::merge_disjoint(&[tabular, coverage(5, 6, &[("region", "us")])]),
+        Err(CoverageError::PossibleOverlap)
+    );
+}
diff --git a/docs/design_docs/architecture/physical-plan-integration.md b/docs/design_docs/architecture/physical-plan-integration.md
index 7fffc9fdf..057ba7c17 100644
--- a/docs/design_docs/architecture/physical-plan-integration.md
+++ b/docs/design_docs/architecture/physical-plan-integration.md
@@ -75,7 +75,7 @@ lose post-ASAP summary implementations, while aligning it directly with
 Physical lowering is complete only when it recursively lowers the entire
 selected candidate DAG. It must:
 
-1. preserve the semantics and source coverage of the logical candidate;
+1. preserve the semantics and scan selection of the logical candidate;
 2. select an explicit physical algorithm for every logical operation;
 3. carry algorithm configuration on the physical operator rather than in a
    generic statistics record;
diff --git a/docs/design_docs/architecture/planner-runtime-contract.md b/docs/design_docs/architecture/planner-runtime-contract.md
index 2d071e654..58e3f06b6 100644
--- a/docs/design_docs/architecture/planner-runtime-contract.md
+++ b/docs/design_docs/architecture/planner-runtime-contract.md
@@ -73,7 +73,7 @@ and rollback are not an end-to-end Planner protocol.
 2. A physical-plan provider maps those candidates to executor-feasible complete
    alternatives. Unsupported candidates are omitted or explicitly rejected.
 3. The provider binds a stable alternative identity and complete evidence:
-   source coverage, input/output edges, operation counts, update and bootstrap
+   scan selection, input/output edges, operation counts, update and bootstrap
    fanout, retained state, CPU, memory, I/O, and accuracy facts.
 4. ASAPPlanner keeps constructible candidates with missing evidence visible
    in `CandidateLogicalASAPDAGs` but does not certify unknown accuracy. The
diff --git a/docs/design_docs/proposals/asap-aware-mapping/analytical-resource-cost.md b/docs/design_docs/proposals/asap-aware-mapping/analytical-resource-cost.md
index 520501731..b742a8970 100644
--- a/docs/design_docs/proposals/asap-aware-mapping/analytical-resource-cost.md
+++ b/docs/design_docs/proposals/asap-aware-mapping/analytical-resource-cost.md
@@ -365,11 +365,11 @@ coverage set must equal the scope source set: a Scan query with an empty scope,
 or a source-free query with a non-empty scope, fails closed. Empty snapshot
 identifiers, invalid recurrence, or a zero horizon also fail closed.
 
-Every reachable physical `Scan` carries one exact `SourceCoverage` copied from
+Every reachable physical `Scan` carries one exact `ScanSelection` copied from
 this scope. That coverage includes the existing `Source`, its provider-owned
 snapshot ID, and canonical ordinary predicates or info-metric matchers. A scan with no coverage, or coverage
 not present in `ComparisonScope.sources`, makes the plan unavailable. Other
-operators cannot declare source coverage. This prevents a DAG over source B
+operators cannot declare scan selection. This prevents a DAG over source B
 from being estimated under source A's comparison scope.
 
 ## General DAG costing
@@ -554,7 +554,7 @@ counts, releases transient output after its last consumer, and keeps retained
 state live. Consequently a shared scan is charged once per execution and a
 fan-out's memory includes the outputs that really coexist.
 
-Each estimate independently requires the semantic set of source coverages on
+Each estimate independently requires the semantic set of scan selections on
 its reachable Scan nodes to equal `ComparisonScope.sources`. Multiple physical
 Scans may repeat one coverage, but no scope source may be omitted and no Scan
 may add another coverage. This invariant is enforced by the estimator itself,
@@ -596,7 +596,7 @@ It consumes the existing query and physical-operator enums; it does not
 introduce a parallel logical operator vocabulary. For every occurrence, the lowerer sends a
 `PhysicalNodeRequest` containing the logical node, selected existing
 `PhysicalOperator`, occurrence and synthetic-role metadata, already-lowered
-child physical IDs, and any source coverage to a
+child physical IDs, and any scan selection to a
 `PhysicalNodeEvidenceProvider`. The provider atomically returns its own stable
 `physical_id`, the authoritative `OperatorStatistics`, and explicit
 `output_buffer_bytes`; logical edge bytes are never substituted for an
@@ -604,14 +604,14 @@ allocation. Missing evidence makes the entire query unavailable. The returned
 `EvidenceBackedPhysicalDAG` snapshots this evidence so costing does not re-read a live
 catalog after lowering.
 
-Each lowered Scan is bound to exactly one `SourceCoverage` in the comparison
+Each lowered Scan is bound to exactly one `ScanSelection` in the comparison
 scope by the existing source and canonical predicate values. The bound value
 therefore also supplies the provider-owned snapshot ID. Zero matches fail as
 outside scope; multiple matching coverages fail as ambiguous rather than
 choosing an arbitrary snapshot. When a predicate-bearing logical Scan expands
 to Scan → Filter, the synthetic Scan has its own physical ID, statistics, and
 buffer evidence and carries that exact coverage; the Filter has separate
-evidence and no source coverage.
+evidence and no scan selection.
 `ComparisonScope.sources` is an order-independent set of semantic coverages;
 duplicates are invalid. After lowering, every reachable physical Scan must use
 a member of that set and every member must be used by at least one Scan.
@@ -837,7 +837,7 @@ its complete evidence and physical child identities also agree. The cost model
 holds owning `Rc` references for bound target and summary roots, so pointer
 keys cannot become stale and alias a later allocation.
 
-A `SummaryAgg` that reads storage declares `source_coverage_index = Some(i)`,
+A `SummaryAgg` that reads storage declares `scan_selection_index = Some(i)`,
 a non-empty bootstrap-read identity, and positive physical source bytes. An
 aggregate over an already-materialized summary edge declares `None`, an empty
 read identity, and zero source bytes. Its logical input rows and bytes remain
@@ -854,7 +854,7 @@ those evolving evaluations over the complete horizon. Marking its nodes
 rejected. Validation follows only nodes reachable from the physical root. If
 the raw algorithm intentionally reads the same semantic source more than once,
 each reachable scan carries the same evolved source statistics and is charged
-separately; equal source coverage does not deduplicate physical I/O.
+separately; equal scan selection does not deduplicate physical I/O.
 
 This raw-evolution contract currently supports exactly one distinct source
 coverage. A multi-source streaming target is unavailable until per-source
@@ -962,14 +962,14 @@ bound physical DAG for a `SummaryExpr` candidate. The deployment implements
 `PlannerPhysicalPlanProvider`: query-node evidence is consumed atomically by
 the generic query lowerer, while summary binding returns a complete
 `EvidenceBackedPhysicalDAG`, including embedded raw work, build/read operators, retained
-state, execution multiplicity, and source coverage. The adapter calls
+state, execution multiplicity, and scan selection. The adapter calls
 `estimate_physical_dag_comparison`; it never calls `DefaultCostModel` or a
 structural-node-count fallback for final cost.
 
 A candidate is exposed to global selection only when both complete DAGs are
 valid and its calibrated cost is strictly below the raw baseline. Missing or
 stale evidence, an unknown physical algorithm, invalid edges, incomplete
-source coverage, or a candidate that is not cheaper yields `None`. When no
+scan selection, or a candidate that is not cheaper yields `None`. When no
 candidate remains, `chosen = None` preserves the raw pre-ASAP target.
 
 Logical CSE share/recompute rewrites are not complete physical alternatives: