From 00f67fd172c42e9fda8bd7aa41da957b01b4164d Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Mon, 7 Sep 2026 10:11:37 -0600
Subject: [PATCH 1/8] feat(cost): estimate bytes at explicit physical
 boundaries

---
 .../asap-aware-mapping/src/boundary_cost.rs   | 249 +++++++++++++++
 crates/asap-aware-mapping/src/lib.rs          |   1 +
 .../src/physical_plan_cost_model.rs           |  32 ++
 .../asap-aware-mapping/tests/boundary_cost.rs | 284 ++++++++++++++++++
 crates/devtools/src/bin/dag_export.rs         | 166 ++++++++++
 .../analytical-resource-cost.md               |   7 +-
 .../developer_docs/physical-boundary-costs.md |  74 +++++
 tools/dag-viewer/test_render.py               |  23 ++
 8 files changed, 835 insertions(+), 1 deletion(-)
 create mode 100644 crates/asap-aware-mapping/src/boundary_cost.rs
 create mode 100644 crates/asap-aware-mapping/tests/boundary_cost.rs
 create mode 100644 docs/developer_docs/physical-boundary-costs.md

diff --git a/crates/asap-aware-mapping/src/boundary_cost.rs b/crates/asap-aware-mapping/src/boundary_cost.rs
new file mode 100644
index 00000000..11f297fb
--- /dev/null
+++ b/crates/asap-aware-mapping/src/boundary_cost.rs
@@ -0,0 +1,249 @@
+//! Byte estimates at deployment-declared physical boundaries.
+
+use crate::analytical_cost::{
+    estimate_physical_dag, AnalyticalCostError, EvidenceBackedPhysicalDag, ExecutionMultiplicity,
+    PhysicalDagNode,
+};
+use crate::physical_operator_statistics::{ComparisonScope, OperatorStatistics};
+use serde::{Deserialize, Serialize};
+use std::collections::{HashMap, HashSet};
+
+pub const BOUNDARY_MODEL_VERSION: &str = "physical-boundary-bytes-v1";
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
+pub enum BoundaryKind {
+    Network {
+        source_location: String,
+        destination_location: String,
+    },
+    Materialization {
+        medium: MaterializationMedium,
+    },
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum MaterializationMedium {
+    Memory,
+    Disk,
+    ObjectStore,
+}
+
+/// A physical action on a producer output, distinct from a logical DAG edge.
+/// With `consumer = None`, one action serves all consumers. With a consumer,
+/// it is a separate action per execution of that downstream physical node.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct PhysicalBoundary {
+    pub id: String,
+    pub consumer: Option<String>,
+    pub kind: BoundaryKind,
+    pub logical_bytes: u64,
+    /// Encoded payload per execution; compression is explicit evidence.
+    pub encoded_bytes: u64,
+    /// Copies actually transferred or written (e.g. broadcast replicas).
+    pub copies: u64,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct BoundaryNodeEvidence {
+    pub node: PhysicalDagNode,
+    pub statistics: OperatorStatistics,
+    pub boundaries: Vec<PhysicalBoundary>,
+}
+
+/// Complete supplementary physical binding captured atomically with the
+/// planner snapshot. An empty per-node list explicitly asserts no boundary.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct BoundaryProfile {
+    pub evidence_version: String,
+    pub observed_at_ms: u64,
+    pub valid_until_ms: u64,
+    pub calibration: BoundaryCalibration,
+    pub nodes: HashMap<String, BoundaryNodeEvidence>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct BoundaryCalibration {
+    pub version: String,
+    pub cost_per_network_byte: f64,
+    pub cost_per_materialization_byte: f64,
+}
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct BoundaryResources {
+    pub network_bytes: u64,
+    pub materialization_bytes: u64,
+}
+
+impl BoundaryResources {
+    pub fn terms(self) -> [(&'static str, u64); 2] {
+        [
+            ("network_bytes", self.network_bytes),
+            ("materialization_bytes", self.materialization_bytes),
+        ]
+    }
+    fn add(&mut self, other: Self) -> Result<(), AnalyticalCostError> {
+        self.network_bytes = self
+            .network_bytes
+            .checked_add(other.network_bytes)
+            .ok_or(AnalyticalCostError::Overflow)?;
+        self.materialization_bytes = self
+            .materialization_bytes
+            .checked_add(other.materialization_bytes)
+            .ok_or(AnalyticalCostError::Overflow)?;
+        Ok(())
+    }
+}
+
+impl BoundaryCalibration {
+    pub fn cost(&self, value: BoundaryResources) -> Result<f64, AnalyticalCostError> {
+        if self.version.trim().is_empty()
+            || [
+                self.cost_per_network_byte,
+                self.cost_per_materialization_byte,
+            ]
+            .iter()
+            .any(|value| !value.is_finite() || *value < 0.0)
+        {
+            return Err(invalid("invalid boundary calibration"));
+        }
+        let cost = value.network_bytes as f64 * self.cost_per_network_byte
+            + value.materialization_bytes as f64 * self.cost_per_materialization_byte;
+        if !cost.is_finite() {
+            return Err(AnalyticalCostError::Overflow);
+        }
+        Ok(cost)
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct BoundaryEstimate {
+    pub total: BoundaryResources,
+    pub per_node: HashMap<String, BoundaryResources>,
+    pub per_boundary: HashMap<String, BoundaryResources>,
+    pub model_version: String,
+    pub evidence_version: String,
+    pub calibration_version: String,
+    pub cost: f64,
+}
+
+fn invalid(reason: &'static str) -> AnalyticalCostError {
+    AnalyticalCostError::InvalidPhysicalDag(reason)
+}
+
+pub fn estimate_boundaries(
+    dag: &EvidenceBackedPhysicalDag,
+    scope: &ComparisonScope,
+    profile: &BoundaryProfile,
+    evidence_version: &str,
+) -> Result<BoundaryEstimate, AnalyticalCostError> {
+    estimate_physical_dag(&dag.nodes, &dag.root, scope, dag)?;
+    let evaluations = scope.validate()?;
+    if profile.evidence_version.trim().is_empty()
+        || profile.evidence_version != evidence_version
+        || profile.observed_at_ms > scope.planning_time.0
+        || profile.valid_until_ms <= scope.planning_time.0
+    {
+        return Err(AnalyticalCostError::MissingOrStale("boundary profile"));
+    }
+    let by_id: HashMap<_, _> = dag
+        .nodes
+        .iter()
+        .map(|node| (node.id.as_str(), node))
+        .collect();
+    let mut reachable = HashSet::new();
+    let mut pending = vec![dag.root.as_str()];
+    while let Some(id) = pending.pop() {
+        if reachable.insert(id) {
+            pending.extend(by_id[id].children.iter().map(String::as_str));
+        }
+    }
+    let mut total = BoundaryResources::default();
+    let mut per_node = HashMap::new();
+    let mut per_boundary = HashMap::new();
+    for id in &reachable {
+        let node = by_id[id];
+        let evidence = profile.nodes.get(*id).ok_or_else(|| {
+            AnalyticalCostError::MissingOperatorStatistics(format!("boundary:{id}"))
+        })?;
+        if evidence.node != *node || evidence.statistics != dag.evidence[*id].statistics {
+            return Err(invalid(
+                "boundary evidence differs from physical node snapshot",
+            ));
+        }
+        let mut local = BoundaryResources::default();
+        for boundary in &evidence.boundaries {
+            if boundary.id.trim().is_empty() || per_boundary.contains_key(&boundary.id) {
+                return Err(invalid("duplicate or blank physical boundary identity"));
+            }
+            if boundary.logical_bytes != evidence.statistics.output().bytes
+                || (boundary.logical_bytes == 0) != (boundary.encoded_bytes == 0)
+                || boundary.copies == 0
+            {
+                return Err(invalid(
+                    "boundary payload is incompatible with producer output",
+                ));
+            }
+            let execution = if let Some(consumer) = &boundary.consumer {
+                if !reachable.contains(consumer.as_str())
+                    || !by_id[consumer.as_str()]
+                        .children
+                        .iter()
+                        .any(|child| child == *id)
+                {
+                    return Err(invalid(
+                        "boundary consumer is not a reachable physical edge",
+                    ));
+                }
+                by_id[consumer.as_str()].execution
+            } else {
+                node.execution
+            };
+            let executions = match execution {
+                ExecutionMultiplicity::Once => 1,
+                ExecutionMultiplicity::PerEvaluation => evaluations,
+            };
+            let bytes = boundary
+                .encoded_bytes
+                .checked_mul(boundary.copies)
+                .and_then(|bytes| bytes.checked_mul(executions))
+                .ok_or(AnalyticalCostError::Overflow)?;
+            let mut term = BoundaryResources::default();
+            match &boundary.kind {
+                BoundaryKind::Network {
+                    source_location,
+                    destination_location,
+                } => {
+                    if source_location.trim().is_empty()
+                        || destination_location.trim().is_empty()
+                        || source_location.trim() == destination_location.trim()
+                    {
+                        return Err(invalid(
+                            "network boundary needs distinct non-empty locations",
+                        ));
+                    }
+                    term.network_bytes = bytes;
+                }
+                BoundaryKind::Materialization { .. } => term.materialization_bytes = bytes,
+            }
+            local.add(term)?;
+            per_boundary.insert(boundary.id.clone(), term);
+        }
+        total.add(local)?;
+        per_node.insert((*id).into(), local);
+    }
+    Ok(BoundaryEstimate {
+        total,
+        per_node,
+        per_boundary,
+        model_version: BOUNDARY_MODEL_VERSION.into(),
+        evidence_version: profile.evidence_version.clone(),
+        calibration_version: profile.calibration.version.clone(),
+        cost: profile.calibration.cost(total)?,
+    })
+}
diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs
index e682fe66..89a44acc 100644
--- a/crates/asap-aware-mapping/src/lib.rs
+++ b/crates/asap-aware-mapping/src/lib.rs
@@ -184,6 +184,7 @@
 pub mod accuracy;
 pub mod accuracy_reconciliation;
 pub mod analytical_cost;
+pub mod boundary_cost;
 pub mod cost_model;
 pub mod empirical_comparison;
 pub mod empirical_cost;
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 86025910..0ebcacf8 100644
--- a/crates/asap-aware-mapping/src/physical_plan_cost_model.rs
+++ b/crates/asap-aware-mapping/src/physical_plan_cost_model.rs
@@ -30,6 +30,7 @@ pub struct PhysicalEvidenceSnapshot {
     pub scope: ComparisonScope,
     pub cache_profile: CacheProfile,
     pub storage_io: Option<crate::storage_io::StorageIoProfile>,
+    pub boundaries: Option<crate::boundary_cost::BoundaryProfile>,
 }
 
 /// Deployment evidence needed to price one planner alternative.
@@ -71,6 +72,10 @@ pub struct PhysicalPlanComparison {
         crate::storage_io::StorageEstimate,
         crate::storage_io::StorageEstimate,
     )>,
+    pub boundaries: Option<(
+        crate::boundary_cost::BoundaryEstimate,
+        crate::boundary_cost::BoundaryEstimate,
+    )>,
 }
 
 /// Planner cost model that admits only complete, cheaper physical plans.
@@ -218,6 +223,26 @@ impl<'a> PhysicalPlanCostModel<'a> {
                 ))
             })
             .transpose()?;
+        let boundaries = snapshot
+            .boundaries
+            .as_ref()
+            .map(|profile| {
+                Ok((
+                    crate::boundary_cost::estimate_boundaries(
+                        &raw,
+                        scope,
+                        profile,
+                        &snapshot.version,
+                    )?,
+                    crate::boundary_cost::estimate_boundaries(
+                        &replacement,
+                        scope,
+                        profile,
+                        &snapshot.version,
+                    )?,
+                ))
+            })
+            .transpose()?;
         let (mut raw_cost, mut candidate_cost) = match self.calibration.validate() {
             Ok(()) => (
                 Cost(resources.raw.calibrated_cost(&self.calibration)?),
@@ -248,6 +273,10 @@ impl<'a> PhysicalPlanCostModel<'a> {
             raw_cost.0 += raw.cost;
             candidate_cost.0 += candidate.cost;
         }
+        if let Some((raw, candidate)) = &boundaries {
+            raw_cost.0 += raw.cost;
+            candidate_cost.0 += candidate.cost;
+        }
         if !raw_cost.0.is_finite() || !candidate_cost.0.is_finite() {
             return Err(AnalyticalCostError::Overflow);
         }
@@ -256,6 +285,7 @@ impl<'a> PhysicalPlanCostModel<'a> {
             raw_cost,
             candidate_cost,
             storage_io,
+            boundaries,
         })
     }
 }
@@ -500,6 +530,7 @@ mod tests {
                 scope: scope(),
                 cache_profile: CacheProfile::no_cache(),
                 storage_io: self.storage_io.clone(),
+                boundaries: None,
             })
         }
 
@@ -862,6 +893,7 @@ mod tests {
                     scope: scope(),
                     cache_profile: CacheProfile::no_cache(),
                     storage_io: None,
+                    boundaries: None,
                 })
             }
 
diff --git a/crates/asap-aware-mapping/tests/boundary_cost.rs b/crates/asap-aware-mapping/tests/boundary_cost.rs
new file mode 100644
index 00000000..2ed6c0f0
--- /dev/null
+++ b/crates/asap-aware-mapping/tests/boundary_cost.rs
@@ -0,0 +1,284 @@
+use asap_aware_mapping::analytical_cost::{
+    EvidenceBackedPhysicalDag, ExecutionMultiplicity, PhysicalDagNode, PhysicalNodeEvidence,
+    PhysicalOperator,
+};
+use asap_aware_mapping::physical_operator_statistics::{
+    ComparisonScope, EdgeStatistics, OperatorStatistics, SourceCoverage, UnaryEdgeStatistics,
+};
+use asap_types::pre_asap::query_expr::Source;
+use asap_types::workload::{
+    DataArrival, DurationMs, QueryRecurrence, QueryTimeScope, TimeSelection, TimestampMs,
+};
+use std::collections::HashMap;
+
+fn fixture() -> (EvidenceBackedPhysicalDag, ComparisonScope) {
+    let coverage = SourceCoverage {
+        source: Source::Table {
+            table_ref: "events".into(),
+        },
+        source_snapshot_id: "snapshot".into(),
+        predicates: vec![],
+        info_matchers: vec![],
+    };
+    let scope = ComparisonScope {
+        data_arrival: DataArrival::AtRest,
+        planning_time: TimestampMs(100),
+        horizon: DurationMs(1000),
+        recurrence: QueryRecurrence::OneTime {
+            invocations: 3,
+            execute_at: None,
+        },
+        time_selection: TimeSelection {
+            scope: QueryTimeScope::Longitudinal,
+            lookback: Some(DurationMs(1000)),
+            as_of: Some(TimestampMs(100)),
+        },
+        sources: vec![coverage.clone()],
+    };
+    let edge = EdgeStatistics {
+        rows: 10,
+        bytes: 80,
+    };
+    let unary = UnaryEdgeStatistics {
+        input: edge,
+        output: edge,
+        promql: None,
+    };
+    let nodes = vec![
+        PhysicalDagNode {
+            id: "scan".into(),
+            operator: PhysicalOperator::Scan,
+            children: vec![],
+            source_coverage: Some(coverage),
+            output_buffer_bytes: 0,
+            retained_bytes: 0,
+            execution: ExecutionMultiplicity::PerEvaluation,
+        },
+        PhysicalDagNode {
+            id: "left".into(),
+            operator: PhysicalOperator::PassThrough,
+            children: vec!["scan".into()],
+            source_coverage: None,
+            output_buffer_bytes: 0,
+            retained_bytes: 0,
+            execution: ExecutionMultiplicity::PerEvaluation,
+        },
+        PhysicalDagNode {
+            id: "right".into(),
+            operator: PhysicalOperator::PassThrough,
+            children: vec!["scan".into()],
+            source_coverage: None,
+            output_buffer_bytes: 0,
+            retained_bytes: 0,
+            execution: ExecutionMultiplicity::PerEvaluation,
+        },
+        PhysicalDagNode {
+            id: "root".into(),
+            operator: PhysicalOperator::Concat,
+            children: vec!["left".into(), "right".into()],
+            source_coverage: None,
+            output_buffer_bytes: 0,
+            retained_bytes: 0,
+            execution: ExecutionMultiplicity::PerEvaluation,
+        },
+    ];
+    let evidence = HashMap::from([
+        (
+            "scan".into(),
+            PhysicalNodeEvidence {
+                physical_id: "scan".into(),
+                statistics: OperatorStatistics::Scan {
+                    edges: unary,
+                    source_read_bytes: 80,
+                },
+                output_buffer_bytes: 0,
+            },
+        ),
+        (
+            "left".into(),
+            PhysicalNodeEvidence {
+                physical_id: "left".into(),
+                statistics: OperatorStatistics::PassThrough { edges: unary },
+                output_buffer_bytes: 0,
+            },
+        ),
+        (
+            "right".into(),
+            PhysicalNodeEvidence {
+                physical_id: "right".into(),
+                statistics: OperatorStatistics::PassThrough { edges: unary },
+                output_buffer_bytes: 0,
+            },
+        ),
+        (
+            "root".into(),
+            PhysicalNodeEvidence {
+                physical_id: "root".into(),
+                statistics: OperatorStatistics::Concat {
+                    inputs: vec![edge, edge],
+                    output: EdgeStatistics {
+                        rows: 20,
+                        bytes: 160,
+                    },
+                    promql: None,
+                },
+                output_buffer_bytes: 0,
+            },
+        ),
+    ]);
+    (
+        EvidenceBackedPhysicalDag {
+            nodes,
+            root: "root".into(),
+            evidence,
+        },
+        scope,
+    )
+}
+
+use asap_aware_mapping::boundary_cost::*;
+
+fn profile(dag: &EvidenceBackedPhysicalDag) -> BoundaryProfile {
+    BoundaryProfile {
+        evidence_version: "evidence-v1".into(),
+        observed_at_ms: 90,
+        valid_until_ms: 200,
+        calibration: BoundaryCalibration {
+            version: "bytes-v1".into(),
+            cost_per_network_byte: 2.0,
+            cost_per_materialization_byte: 3.0,
+        },
+        nodes: dag
+            .nodes
+            .iter()
+            .map(|node| {
+                (
+                    node.id.clone(),
+                    BoundaryNodeEvidence {
+                        node: node.clone(),
+                        statistics: dag.evidence[&node.id].statistics.clone(),
+                        boundaries: vec![],
+                    },
+                )
+            })
+            .collect(),
+    }
+}
+
+fn transfer(id: &str, consumer: Option<&str>) -> PhysicalBoundary {
+    PhysicalBoundary {
+        id: id.into(),
+        consumer: consumer.map(str::to_owned),
+        kind: BoundaryKind::Network {
+            source_location: "edge".into(),
+            destination_location: "backend".into(),
+        },
+        logical_bytes: 80,
+        encoded_bytes: 40,
+        copies: 2,
+    }
+}
+
+// Ordinary in-memory edges contribute no traffic; shared transfers count once.
+#[test]
+fn memory_edges_are_free_and_shared_transfer_is_counted_once() {
+    let (dag, scope) = fixture();
+    let mut profile = profile(&dag);
+    assert_eq!(
+        estimate_boundaries(&dag, &scope, &profile, "evidence-v1")
+            .unwrap()
+            .total,
+        BoundaryResources::default()
+    );
+    profile
+        .nodes
+        .get_mut("scan")
+        .unwrap()
+        .boundaries
+        .push(transfer("shared", None));
+    let estimate = estimate_boundaries(&dag, &scope, &profile, "evidence-v1").unwrap();
+    assert_eq!(estimate.total.network_bytes, 240); // 40 encoded bytes × 2 replicas × 3 evaluations
+    assert_eq!(estimate.total.materialization_bytes, 0);
+    assert_eq!(estimate.cost, 480.0);
+    assert_eq!(estimate.per_node["scan"].network_bytes, 240);
+    assert_eq!(estimate.per_boundary["shared"].network_bytes, 240);
+}
+
+// A retained producer materializes once and transfers separately to each reader.
+#[test]
+fn materialization_once_and_transfers_per_consumer_have_distinct_multiplicity() {
+    let (mut dag, scope) = fixture();
+    dag.nodes[0].execution = ExecutionMultiplicity::Once;
+    dag.nodes[0].retained_bytes = 80;
+    let mut profile = profile(&dag);
+    let mut materialize = transfer("persist", None);
+    materialize.kind = BoundaryKind::Materialization {
+        medium: MaterializationMedium::Disk,
+    };
+    materialize.copies = 1;
+    profile.nodes.get_mut("scan").unwrap().boundaries = vec![
+        materialize,
+        transfer("left-wire", Some("left")),
+        transfer("right-wire", Some("right")),
+    ];
+    let estimate = estimate_boundaries(&dag, &scope, &profile, "evidence-v1").unwrap();
+    assert_eq!(estimate.total.materialization_bytes, 40);
+    assert_eq!(estimate.total.network_bytes, 480);
+    assert_eq!(estimate.cost, 1080.0);
+}
+
+// Invalid endpoint, byte evidence, duplicate identity, and overflow fail closed.
+#[test]
+fn incompatible_boundary_evidence_is_rejected() {
+    let (dag, scope) = fixture();
+    for case in 0..7 {
+        let mut profile = profile(&dag);
+        let mut boundary = transfer("wire", Some("left"));
+        match case {
+            0 => boundary.consumer = Some("root".into()),
+            1 => boundary.logical_bytes = 79,
+            2 => boundary.copies = 0,
+            3 => boundary.encoded_bytes = 0,
+            4 => {
+                boundary.kind = BoundaryKind::Network {
+                    source_location: "same".into(),
+                    destination_location: "same".into(),
+                }
+            }
+            5 => boundary.copies = u64::MAX,
+            6 => profile
+                .nodes
+                .get_mut("scan")
+                .unwrap()
+                .boundaries
+                .push(boundary.clone()),
+            _ => unreachable!(),
+        }
+        profile
+            .nodes
+            .get_mut("scan")
+            .unwrap()
+            .boundaries
+            .push(boundary);
+        assert!(
+            estimate_boundaries(&dag, &scope, &profile, "evidence-v1").is_err(),
+            "case {case}"
+        );
+    }
+}
+
+// Missing evidence and non-finite coefficients cannot be treated as zero.
+#[test]
+fn missing_stale_and_non_finite_evidence_is_rejected() {
+    let (dag, scope) = fixture();
+    let mut evidence = profile(&dag);
+    evidence.nodes.remove("left");
+    assert!(estimate_boundaries(&dag, &scope, &evidence, "evidence-v1").is_err());
+    evidence = profile(&dag);
+    evidence.valid_until_ms = 100;
+    assert!(estimate_boundaries(&dag, &scope, &evidence, "evidence-v1").is_err());
+    evidence = profile(&dag);
+    evidence.calibration.cost_per_network_byte = f64::INFINITY;
+    assert!(estimate_boundaries(&dag, &scope, &evidence, "evidence-v1").is_err());
+    assert!(estimate_boundaries(&dag, &scope, &profile(&dag), "different").is_err());
+}
diff --git a/crates/devtools/src/bin/dag_export.rs b/crates/devtools/src/bin/dag_export.rs
index 3eb51a9a..fe3a7f36 100644
--- a/crates/devtools/src/bin/dag_export.rs
+++ b/crates/devtools/src/bin/dag_export.rs
@@ -90,6 +90,8 @@ use asap_types::types::AccuracyTarget;
 struct PlannerCostDocument {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     storage_io: Option<asap_aware_mapping::storage_io::StorageIoProfile>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    boundaries: Option<asap_aware_mapping::boundary_cost::BoundaryProfile>,
     /// Immutable catalog/runtime evidence generation shared by this file.
     evidence_version: String,
     calibration: ResourceCalibration,
@@ -279,6 +281,7 @@ fn plan_values_match_inner(
 
 struct ExportPhysicalProvider<'a> {
     storage_io: Option<&'a asap_aware_mapping::storage_io::StorageIoProfile>,
+    boundaries: Option<&'a asap_aware_mapping::boundary_cost::BoundaryProfile>,
     evidence_version: &'a str,
     target: &'a TargetPhysicalEvidence,
     candidate: &'a CandidatePhysicalEvidence,
@@ -301,6 +304,7 @@ impl PlannerPhysicalPlanProvider for ExportPhysicalProvider<'_> {
             scope: self.target.scope.resolve()?,
             cache_profile: self.target.scope.cache_profile.clone(),
             storage_io: self.storage_io.cloned(),
+            boundaries: self.boundaries.cloned(),
         })
     }
 
@@ -390,6 +394,7 @@ impl ExportPlannerCostModel<'_> {
         Some((
             ExportPhysicalProvider {
                 storage_io: self.document.storage_io.as_ref(),
+                boundaries: self.document.boundaries.as_ref(),
                 evidence_version: &self.document.evidence_version,
                 target: target_evidence,
                 candidate: candidate_evidence,
@@ -424,6 +429,12 @@ impl ExportPlannerCostModel<'_> {
                 storage.model_version, storage.calibration_version
             ));
         }
+        if let Some((boundary, _)) = &estimate.boundaries {
+            version.push_str(&format!(
+                "+{}+{}",
+                boundary.model_version, boundary.calibration_version
+            ));
+        }
         let scope = &provider.target.scope;
         let Ok((result_hits, buffer_hits)) = cache_hit_ratios(
             &scope.cache_profile,
@@ -537,6 +548,40 @@ impl ExportPlannerCostModel<'_> {
             raw_inputs.extend(storage_inputs(raw));
             candidate_inputs.extend(storage_inputs(candidate));
         }
+        let boundary_inputs =
+            |estimate: &asap_aware_mapping::boundary_cost::BoundaryEstimate| {
+                let mut terms: Vec<_> = estimate
+                    .total
+                    .terms()
+                    .into_iter()
+                    .map(|(name, value)| CostInput {
+                        name: name.into(),
+                        value: value as f64,
+                        unit: Some("bytes".into()),
+                    })
+                    .collect();
+                for (prefix, entries) in [
+                    ("physical_node", &estimate.per_node),
+                    ("boundary", &estimate.per_boundary),
+                ] {
+                    let mut ids: Vec<_> = entries.keys().collect();
+                    ids.sort();
+                    for id in ids {
+                        terms.extend(entries[id].terms().into_iter().map(|(name, value)| {
+                            CostInput {
+                                name: format!("{prefix}:{id}:{name}"),
+                                value: value as f64,
+                                unit: Some("bytes".into()),
+                            }
+                        }));
+                    }
+                }
+                terms
+            };
+        if let Some((raw, candidate)) = &estimate.boundaries {
+            raw_inputs.extend(boundary_inputs(raw));
+            candidate_inputs.extend(boundary_inputs(candidate));
+        }
         let baseline = CostAnnotation::modeled(
             estimate.raw_cost.0,
             CostUnit::CostUnits,
@@ -1756,6 +1801,125 @@ mod tests {
         .is_none());
     }
 
+    // Declared boundaries survive JSON and affect the selected physical plan.
+    #[test]
+    fn boundary_bytes_export_and_change_plan_selection() {
+        use asap_aware_mapping::boundary_cost::*;
+        let (query, candidate, mut document) = cost_fixture();
+        let raw = fixture_raw_dag(&query, &candidate, &document);
+        let candidate_dag = cheap_candidate_dag();
+        let root = Rc::new(query.clone());
+        let target = asap_aware_mapping::replacement::TargetSubDAG::new(&root);
+        assert!(ExportPlannerCostModel {
+            document: &document
+        }
+        .candidate_cost(&candidate, &target)
+        .is_some());
+        let mut profile = BoundaryProfile {
+            evidence_version: document.evidence_version.clone(),
+            observed_at_ms: 900,
+            valid_until_ms: 2000,
+            calibration: BoundaryCalibration {
+                version: "bytes-v1".into(),
+                cost_per_network_byte: 1.0,
+                cost_per_materialization_byte: 1.0,
+            },
+            nodes: std::collections::HashMap::new(),
+        };
+        for dag in [&raw, &candidate_dag] {
+            for node in &dag.nodes {
+                profile.nodes.insert(
+                    node.id.clone(),
+                    BoundaryNodeEvidence {
+                        node: node.clone(),
+                        statistics: dag.evidence[&node.id].statistics.clone(),
+                        boundaries: vec![],
+                    },
+                );
+            }
+        }
+        profile
+            .nodes
+            .get_mut("summary-read")
+            .unwrap()
+            .boundaries
+            .push(PhysicalBoundary {
+                id: "summary-transfer".into(),
+                consumer: None,
+                kind: BoundaryKind::Network {
+                    source_location: "edge".into(),
+                    destination_location: "backend".into(),
+                },
+                logical_bytes: 2400,
+                encoded_bytes: 1200,
+                copies: 1,
+            });
+        document.boundaries = Some(profile);
+        let parsed =
+            parse_planner_cost_document(&serde_json::to_string(&document).unwrap()).unwrap();
+        let model = ExportPlannerCostModel { document: &parsed };
+        let (baseline, selected, _) = model.annotations(&candidate, &root);
+        assert_eq!(
+            baseline
+                .inputs
+                .iter()
+                .find(|term| term.name == "network_bytes")
+                .unwrap()
+                .value,
+            0.0
+        );
+        assert_eq!(
+            selected
+                .inputs
+                .iter()
+                .find(|term| term.name == "network_bytes")
+                .unwrap()
+                .value,
+            12_000.0
+        );
+        assert!(selected
+            .inputs
+            .iter()
+            .any(|term| term.name == "physical_node:summary-read:network_bytes"));
+        assert!(selected
+            .inputs
+            .iter()
+            .any(|term| term.name == "boundary:summary-transfer:network_bytes"));
+        assert!(selected
+            .model_version
+            .as_ref()
+            .unwrap()
+            .contains(BOUNDARY_MODEL_VERSION));
+        assert_eq!(
+            selected.evidence_version.as_deref(),
+            Some("test-evidence-v1")
+        );
+        document
+            .boundaries
+            .as_mut()
+            .unwrap()
+            .calibration
+            .cost_per_network_byte = 1000.0;
+        assert!(ExportPlannerCostModel {
+            document: &document
+        }
+        .candidate_cost(&candidate, &target)
+        .is_none());
+        document
+            .boundaries
+            .as_mut()
+            .unwrap()
+            .nodes
+            .remove(&raw.root);
+        assert!(ExportPlannerCostModel {
+            document: &document
+        }
+        .annotations(&candidate, &root)
+        .0
+        .value
+        .is_none());
+    }
+
     fn test_scope() -> ComparisonScopeEvidence {
         ComparisonScopeEvidence {
             data_arrival: asap_types::workload::DataArrival::AtRest,
@@ -1908,6 +2072,7 @@ mod tests {
         };
         let document = PlannerCostDocument {
             storage_io: None,
+            boundaries: None,
             evidence_version: "test-evidence-v1".into(),
             calibration: ResourceCalibration {
                 cost_per_cpu_op: 1.0,
@@ -2221,6 +2386,7 @@ mod tests {
         let second_dag = cheap_candidate_dag();
         let document = PlannerCostDocument {
             storage_io: None,
+            boundaries: None,
             evidence_version: "test-evidence-v1".into(),
             calibration: ResourceCalibration {
                 cost_per_cpu_op: 1.0,
diff --git a/docs/design_docs/asap-aware-mapping/analytical-resource-cost.md b/docs/design_docs/asap-aware-mapping/analytical-resource-cost.md
index cc444a7a..a618a957 100644
--- a/docs/design_docs/asap-aware-mapping/analytical-resource-cost.md
+++ b/docs/design_docs/asap-aware-mapping/analytical-resource-cost.md
@@ -1174,6 +1174,11 @@ request counts using the optional storage profile. See
 [storage operation estimates](../../developer_docs/storage-operation-costs.md)
 for units, formulas, evidence requirements, and calibration.
 
+The physical-plan adapter supports explicit network and materialization
+boundaries through optional deployment evidence. See
+[physical boundary estimates](../../developer_docs/physical-boundary-costs.md)
+for kinds, execution multiplicity, evidence, and calibration.
+
 An estimate is unavailable when required evidence, a physical formula, or a
 finite horizon is missing. Structural node counts are never substituted.
 
@@ -1181,7 +1186,7 @@ The current scalar includes CPU, retained memory, and source/disk reads. A
 complete deployment model may additionally require:
 
 - source and spill writes;
-- network transfer;
+- latency and contention at network boundaries;
 - parallelism and contention;
 - allocator fragmentation;
 - wall-clock critical-path latency;
diff --git a/docs/developer_docs/physical-boundary-costs.md b/docs/developer_docs/physical-boundary-costs.md
new file mode 100644
index 00000000..d2ac3552
--- /dev/null
+++ b/docs/developer_docs/physical-boundary-costs.md
@@ -0,0 +1,74 @@
+# Physical boundary byte estimates
+
+The physical-plan adapter accepts an optional `BoundaryProfile` in the
+immutable `PhysicalEvidenceSnapshot`. `dag_export --planner-cost-json` accepts
+the same profile in a top-level `boundaries` field. With no profile, these
+dimensions remain unestimated and the existing resource objective is preserved.
+
+The profile is supplementary physical-plan binding: each entry supplies the
+complete physical node, its authoritative statistics, and an explicit list of
+boundary actions on its output. Every reachable node needs an entry; an empty
+list declares ordinary in-memory dataflow with no boundary traffic. The full
+node/statistics match prevents rebinding evidence to a different operator with
+the same ID. Additional entries may describe other candidate plans. Evidence
+must match the immutable snapshot version and be current at planning time:
+`observed_at_ms <= planning_time < valid_until_ms`.
+
+Supported boundaries are:
+
+| Boundary | Required evidence | Dimension |
+|---|---|---|
+| Network/exchange/deployment transfer | Distinct nonempty source and destination locations | Network bytes |
+| Materialization/persistence | Memory, disk, or object-store medium | Materialization bytes |
+
+Every boundary also declares its unique physical ID, output logical bytes,
+encoded bytes per execution, and positive copy count. Logical bytes must equal
+the producer's output statistic. Encoded bytes capture an explicit compression
+or serialization estimate; empty and nonempty payloads must agree with the
+logical edge. A remote persistence operation may declare both a network action
+and a materialization action with distinct IDs; these contribute to different
+dimensions and require separate calibration coefficients.
+
+```text
+boundary_bytes = encoded_bytes * copies * executions
+```
+
+For a shared boundary (`consumer: null`), execution multiplicity comes from
+the producer. For a per-consumer boundary, `consumer` must identify an actual,
+reachable parent of the producer; multiplicity comes from that consumer.
+`Once` means one execution; `PerEvaluation` uses the comparison scope's demand.
+Shared producers are traversed once regardless of fan-out. Duplicate boundary
+IDs fail closed instead of being ambiguously counted or silently dropped.
+
+Example: a retained producer materializes 40 encoded bytes once. Two consumers
+each receive two copies of those 40 bytes over three evaluations. The totals
+are 40 materialization bytes and 480 network bytes. Logical in-memory edges
+without a boundary action add no traffic. A retained producer can therefore
+have a once-only persistence action and repeated transfers to its readers.
+
+`BoundaryEstimate` keeps network and materialization totals separate and returns
+per-node and per-boundary terms with model, evidence, and calibration provenance.
+A `BoundaryCalibration` supplies finite, nonnegative cost coefficients in the
+same cost units as the base resource model, and a nonempty version. Both
+alternatives use that profile when ranking. This models byte work, not transfer
+latency, bandwidth contention, memory lifetime, or storage requests.
+
+Annotations expose totals in `bytes`, plus terms named
+`physical_node:<id>:<dimension>` and `boundary:<id>:<dimension>`. The existing
+viewer cost sidebar displays all terms and the combined formula/calibration
+version (`physical-boundary-bytes-v1`) with the evidence version. No network
+traffic is inferred from logical edges, operator buffers, or scan bytes.
+
+Unknown endpoints, mismatched payloads, absent node evidence, duplicate IDs,
+stale evidence, invalid coefficients, and integer overflow return typed errors;
+ranking/export report the comparison as unavailable. This extends the physical
+plan adapter; lifecycle-specific summary-maintenance costing and caching are
+separate follow-up integration points.
+
+Verification:
+
+```sh
+cargo test -p asap-aware-mapping --test boundary_cost
+cargo test -p asap-devtools --bin dag_export boundary_bytes_export_and_change_plan_selection
+python3 -m unittest discover -s tools/dag-viewer -p 'test_render.py'
+```
diff --git a/tools/dag-viewer/test_render.py b/tools/dag-viewer/test_render.py
index d5b28f9f..c3463a7a 100644
--- a/tools/dag-viewer/test_render.py
+++ b/tools/dag-viewer/test_render.py
@@ -56,6 +56,29 @@ def named_graph(name: str, source: str = "SELECT 1") -> dict:
 
 
 class LoadWorkloadTests(unittest.TestCase):
+    def test_boundary_terms_and_provenance_survive_standalone_export(self):
+        """The standalone viewer retains byte totals, physical terms, and provenance."""
+        graph = named_graph("boundary-example")
+        annotation = {
+            "value": 1080.0,
+            "unit": "CostUnits",
+            "source": "Modeled",
+            "model_version": "physical-boundary-bytes-v1+bytes-v1",
+            "evidence_version": "evidence-v1",
+            "inputs": [
+                {"name": "network_bytes", "value": 480, "unit": "bytes"},
+                {"name": "materialization_bytes", "value": 40, "unit": "bytes"},
+                {"name": "physical_node:scan:network_bytes", "value": 480, "unit": "bytes"},
+                {"name": "boundary:persist:materialization_bytes", "value": 40, "unit": "bytes"},
+            ],
+        }
+        graph["graph"]["nodes"][0]["selected_cost"] = annotation
+        html = render({"queries": [graph]})
+        for term in annotation["inputs"]:
+            self.assertIn(term["name"], html)
+        self.assertIn(annotation["model_version"], html)
+        self.assertIn(annotation["evidence_version"], html)
+
     def test_loads_summary_maintenance_export_as_a_lifecycle_plan(self):
         graph = named_graph("unused")["graph"]
         summary = {

From 63ac9f68f8962e24a877bbe8fcb4ee738dbcc9e3 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Mon, 7 Sep 2026 10:46:30 -0600
Subject: [PATCH 2/8] fix(cost): scope boundary evidence to physical
 alternatives

---
 .../asap-aware-mapping/src/boundary_cost.rs   |  40 ++++--
 .../src/physical_plan_cost_model.rs           | 118 ++++++++++++++++-
 .../asap-aware-mapping/tests/boundary_cost.rs | 121 +++++++++++++++---
 crates/devtools/src/bin/dag_export.rs         |  55 +++++---
 .../developer_docs/physical-boundary-costs.md |  18 ++-
 5 files changed, 299 insertions(+), 53 deletions(-)

diff --git a/crates/asap-aware-mapping/src/boundary_cost.rs b/crates/asap-aware-mapping/src/boundary_cost.rs
index 11f297fb..3f5a2e52 100644
--- a/crates/asap-aware-mapping/src/boundary_cost.rs
+++ b/crates/asap-aware-mapping/src/boundary_cost.rs
@@ -63,6 +63,15 @@ pub struct BoundaryProfile {
     pub observed_at_ms: u64,
     pub valid_until_ms: u64,
     pub calibration: BoundaryCalibration,
+    pub plans: Vec<BoundaryPlanEvidence>,
+}
+
+/// Boundaries belong to a complete alternative: a producer's consumers may
+/// differ between plans even when its physical identity is unchanged.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct BoundaryPlanEvidence {
+    pub root: String,
     pub nodes: HashMap<String, BoundaryNodeEvidence>,
 }
 
@@ -151,6 +160,28 @@ pub fn estimate_boundaries(
     {
         return Err(AnalyticalCostError::MissingOrStale("boundary profile"));
     }
+    let mut matching_plans = profile.plans.iter().filter(|plan| {
+        plan.root == dag.root
+            && plan.nodes.len() == dag.nodes.len()
+            && dag.nodes.iter().all(|node| {
+                plan.nodes.get(&node.id).is_some_and(|evidence| {
+                    evidence.node == *node
+                        && dag.evidence.get(&node.id).is_some_and(|physical| {
+                            physical.physical_id == node.id
+                                && physical.output_buffer_bytes == node.output_buffer_bytes
+                                && evidence.statistics == physical.statistics
+                        })
+                })
+            })
+    });
+    let plan = matching_plans
+        .next()
+        .ok_or(AnalyticalCostError::MissingOrStale(
+            "boundary physical plan",
+        ))?;
+    if matching_plans.next().is_some() {
+        return Err(invalid("ambiguous boundary physical plan"));
+    }
     let by_id: HashMap<_, _> = dag
         .nodes
         .iter()
@@ -168,14 +199,7 @@ pub fn estimate_boundaries(
     let mut per_boundary = HashMap::new();
     for id in &reachable {
         let node = by_id[id];
-        let evidence = profile.nodes.get(*id).ok_or_else(|| {
-            AnalyticalCostError::MissingOperatorStatistics(format!("boundary:{id}"))
-        })?;
-        if evidence.node != *node || evidence.statistics != dag.evidence[*id].statistics {
-            return Err(invalid(
-                "boundary evidence differs from physical node snapshot",
-            ));
-        }
+        let evidence = &plan.nodes[*id];
         let mut local = BoundaryResources::default();
         for boundary in &evidence.boundaries {
             if boundary.id.trim().is_empty() || per_boundary.contains_key(&boundary.id) {
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 0ebcacf8..ace1b34c 100644
--- a/crates/asap-aware-mapping/src/physical_plan_cost_model.rs
+++ b/crates/asap-aware-mapping/src/physical_plan_cost_model.rs
@@ -109,8 +109,7 @@ impl<'a> PhysicalPlanCostModel<'a> {
                 "resource_calibration.version",
             ));
         }
-        // A zero base objective may be supplemented by storage coefficients;
-        // that check needs the target's immutable evidence snapshot.
+        // Supplemental storage or boundary pricing may supply a zero-base objective.
         match calibration.validate() {
             Ok(()) | Err(AnalyticalCostError::ZeroCalibration) => {}
             Err(error) => return Err(error),
@@ -262,7 +261,13 @@ impl<'a> PhysicalPlanCostModel<'a> {
                     .into_iter()
                     .any(|coefficient| coefficient > 0.0)
                 });
-                if !has_storage_objective {
+                // Boundary estimation above validates evidence and pricing.
+                // At least one dimension must have a positive coefficient.
+                let has_boundary_objective = snapshot.boundaries.as_ref().is_some_and(|profile| {
+                    profile.calibration.cost_per_network_byte > 0.0
+                        || profile.calibration.cost_per_materialization_byte > 0.0
+                });
+                if !has_storage_objective && !has_boundary_objective {
                     return Err(AnalyticalCostError::ZeroCalibration);
                 }
                 (Cost(0.0), Cost(0.0))
@@ -435,6 +440,7 @@ mod tests {
         storage_io: Option<crate::storage_io::StorageIoProfile>,
         summary_available: bool,
         candidate_scan_bytes: u64,
+        boundaries: Option<crate::boundary_cost::BoundaryProfile>,
         snapshot_calls: Cell<u64>,
         raw_evidence_calls: Cell<u64>,
     }
@@ -445,6 +451,7 @@ mod tests {
                 storage_io: None,
                 summary_available,
                 candidate_scan_bytes,
+                boundaries: None,
                 snapshot_calls: Cell::new(0),
                 raw_evidence_calls: Cell::new(0),
             }
@@ -530,7 +537,7 @@ mod tests {
                 scope: scope(),
                 cache_profile: CacheProfile::no_cache(),
                 storage_io: self.storage_io.clone(),
-                boundaries: None,
+                boundaries: self.boundaries.clone(),
             })
         }
 
@@ -705,6 +712,109 @@ mod tests {
         assert!(estimate.storage_io.is_none());
     }
 
+    // Explicit byte pricing can rank complete plans without pricing CPU or scans.
+    #[test]
+    fn boundary_only_objective_ranks_complete_plans() {
+        use crate::boundary_cost::*;
+        let root = query();
+        let target = TargetSubDAG::new(&root);
+        let mut provider = TestProvider::new(true, 800);
+        let raw = PhysicalPlanCostModel::new(&provider, calibration())
+            .unwrap()
+            .target_evidence(&target)
+            .unwrap()
+            .1;
+        let candidate = provider.summary_dag(&scope());
+        provider.boundaries = Some(BoundaryProfile {
+            evidence_version: "test-snapshot-1".into(),
+            observed_at_ms: 900,
+            valid_until_ms: 2000,
+            calibration: BoundaryCalibration {
+                version: "network-only-v1".into(),
+                cost_per_network_byte: 1.0,
+                cost_per_materialization_byte: 0.0,
+            },
+            plans: [&raw, &candidate]
+                .into_iter()
+                .map(|dag| BoundaryPlanEvidence {
+                    root: dag.root.clone(),
+                    nodes: dag
+                        .nodes
+                        .iter()
+                        .map(|node| {
+                            let statistics = dag.evidence[&node.id].statistics.clone();
+                            let bytes = statistics.output().bytes;
+                            let boundaries = if matches!(node.operator, PhysicalOperator::Scan) {
+                                vec![PhysicalBoundary {
+                                    id: "scan-transfer".into(),
+                                    consumer: None,
+                                    kind: BoundaryKind::Network {
+                                        source_location: "edge".into(),
+                                        destination_location: "backend".into(),
+                                    },
+                                    logical_bytes: bytes,
+                                    encoded_bytes: bytes,
+                                    copies: 1,
+                                }]
+                            } else {
+                                vec![]
+                            };
+                            (
+                                node.id.clone(),
+                                BoundaryNodeEvidence {
+                                    node: node.clone(),
+                                    statistics,
+                                    boundaries,
+                                },
+                            )
+                        })
+                        .collect(),
+                })
+                .collect(),
+        });
+        let zero_base = ResourceCalibration {
+            cost_per_cpu_op: 0.0,
+            cost_per_scan_byte: 0.0,
+            cost_per_retained_byte: 0.0,
+            version: "boundary-only-v1".into(),
+        };
+        let space = crate::replacement::search_workload_with(
+            vec![("q", Rc::clone(&root))],
+            &crate::replacement::default_strategies(),
+        );
+        let model = PhysicalPlanCostModel::new(&provider, zero_base.clone()).unwrap();
+        let selected = space.global_selection(&model);
+        assert!(selected
+            .for_target(&space.roots[0].1)
+            .unwrap()
+            .chosen
+            .is_some());
+        drop(model);
+        for coefficient in [0.0, -1.0, f64::NAN] {
+            provider
+                .boundaries
+                .as_mut()
+                .unwrap()
+                .calibration
+                .cost_per_network_byte = coefficient;
+            let model = PhysicalPlanCostModel::new(&provider, zero_base.clone()).unwrap();
+            assert!(space
+                .global_selection(&model)
+                .for_target(&space.roots[0].1)
+                .unwrap()
+                .chosen
+                .is_none());
+        }
+        provider.boundaries = None;
+        let model = PhysicalPlanCostModel::new(&provider, zero_base).unwrap();
+        assert!(space
+            .global_selection(&model)
+            .for_target(&space.roots[0].1)
+            .unwrap()
+            .chosen
+            .is_none());
+    }
+
     #[test]
     fn global_selection_uses_complete_physical_comparison() {
         let root = query();
diff --git a/crates/asap-aware-mapping/tests/boundary_cost.rs b/crates/asap-aware-mapping/tests/boundary_cost.rs
index 2ed6c0f0..97136de3 100644
--- a/crates/asap-aware-mapping/tests/boundary_cost.rs
+++ b/crates/asap-aware-mapping/tests/boundary_cost.rs
@@ -148,20 +148,23 @@ fn profile(dag: &EvidenceBackedPhysicalDag) -> BoundaryProfile {
             cost_per_network_byte: 2.0,
             cost_per_materialization_byte: 3.0,
         },
-        nodes: dag
-            .nodes
-            .iter()
-            .map(|node| {
-                (
-                    node.id.clone(),
-                    BoundaryNodeEvidence {
-                        node: node.clone(),
-                        statistics: dag.evidence[&node.id].statistics.clone(),
-                        boundaries: vec![],
-                    },
-                )
-            })
-            .collect(),
+        plans: vec![BoundaryPlanEvidence {
+            root: dag.root.clone(),
+            nodes: dag
+                .nodes
+                .iter()
+                .map(|node| {
+                    (
+                        node.id.clone(),
+                        BoundaryNodeEvidence {
+                            node: node.clone(),
+                            statistics: dag.evidence[&node.id].statistics.clone(),
+                            boundaries: vec![],
+                        },
+                    )
+                })
+                .collect(),
+        }],
     }
 }
 
@@ -179,6 +182,86 @@ fn transfer(id: &str, consumer: Option<&str>) -> PhysicalBoundary {
     }
 }
 
+fn profiles_for_alternatives(dags: &[&EvidenceBackedPhysicalDag]) -> BoundaryProfile {
+    let mut combined = profile(dags[0]);
+    combined.plans.clear();
+    for dag in dags {
+        let mut alternative = profile(dag);
+        alternative.plans[0]
+            .nodes
+            .get_mut("scan")
+            .unwrap()
+            .boundaries = vec![transfer(&format!("wire-{}", dag.root), Some(&dag.root))];
+        combined.plans.extend(alternative.plans);
+    }
+    combined
+}
+
+// Alternative consumers may share a producer identity without sharing transfers.
+#[test]
+fn shared_producer_boundaries_are_scoped_to_each_alternative() {
+    let (dag, scope) = fixture();
+    let alternative = |root: &str| {
+        let mut value = dag.clone();
+        value.root = root.into();
+        value
+            .nodes
+            .retain(|node| node.id == "scan" || node.id == root);
+        value.evidence.retain(|id, _| id == "scan" || id == root);
+        value
+    };
+    let raw = alternative("left");
+    let candidate = alternative("right");
+    let profile = profiles_for_alternatives(&[&raw, &candidate]);
+    for plan in [&raw, &candidate] {
+        let estimate = estimate_boundaries(plan, &scope, &profile, "evidence-v1").unwrap();
+        assert_eq!(estimate.total.network_bytes, 240);
+        assert_eq!(estimate.per_boundary.len(), 1);
+        assert!(estimate
+            .per_boundary
+            .contains_key(&format!("wire-{}", plan.root)));
+    }
+}
+
+// A boundary binding must identify exactly one complete physical alternative.
+#[test]
+fn missing_ambiguous_and_mismatched_plan_bindings_are_rejected() {
+    let (dag, scope) = fixture();
+    for case in 0..6 {
+        let mut profile = profile(&dag);
+        match case {
+            0 => profile.plans.clear(),
+            1 => profile.plans.push(profile.plans[0].clone()),
+            2 => profile.plans[0].root = "left".into(),
+            3 => {
+                profile.plans[0].nodes.remove("left");
+            }
+            4 => {
+                profile.plans[0]
+                    .nodes
+                    .get_mut("scan")
+                    .unwrap()
+                    .node
+                    .execution = ExecutionMultiplicity::Once
+            }
+            5 => {
+                let OperatorStatistics::Scan {
+                    source_read_bytes, ..
+                } = &mut profile.plans[0].nodes.get_mut("scan").unwrap().statistics
+                else {
+                    unreachable!()
+                };
+                *source_read_bytes += 1;
+            }
+            _ => unreachable!(),
+        }
+        assert!(
+            estimate_boundaries(&dag, &scope, &profile, "evidence-v1").is_err(),
+            "case {case}"
+        );
+    }
+}
+
 // Ordinary in-memory edges contribute no traffic; shared transfers count once.
 #[test]
 fn memory_edges_are_free_and_shared_transfer_is_counted_once() {
@@ -190,7 +273,7 @@ fn memory_edges_are_free_and_shared_transfer_is_counted_once() {
             .total,
         BoundaryResources::default()
     );
-    profile
+    profile.plans[0]
         .nodes
         .get_mut("scan")
         .unwrap()
@@ -216,7 +299,7 @@ fn materialization_once_and_transfers_per_consumer_have_distinct_multiplicity()
         medium: MaterializationMedium::Disk,
     };
     materialize.copies = 1;
-    profile.nodes.get_mut("scan").unwrap().boundaries = vec![
+    profile.plans[0].nodes.get_mut("scan").unwrap().boundaries = vec![
         materialize,
         transfer("left-wire", Some("left")),
         transfer("right-wire", Some("right")),
@@ -246,7 +329,7 @@ fn incompatible_boundary_evidence_is_rejected() {
                 }
             }
             5 => boundary.copies = u64::MAX,
-            6 => profile
+            6 => profile.plans[0]
                 .nodes
                 .get_mut("scan")
                 .unwrap()
@@ -254,7 +337,7 @@ fn incompatible_boundary_evidence_is_rejected() {
                 .push(boundary.clone()),
             _ => unreachable!(),
         }
-        profile
+        profile.plans[0]
             .nodes
             .get_mut("scan")
             .unwrap()
@@ -272,7 +355,7 @@ fn incompatible_boundary_evidence_is_rejected() {
 fn missing_stale_and_non_finite_evidence_is_rejected() {
     let (dag, scope) = fixture();
     let mut evidence = profile(&dag);
-    evidence.nodes.remove("left");
+    evidence.plans[0].nodes.remove("left");
     assert!(estimate_boundaries(&dag, &scope, &evidence, "evidence-v1").is_err());
     evidence = profile(&dag);
     evidence.valid_until_ms = 100;
diff --git a/crates/devtools/src/bin/dag_export.rs b/crates/devtools/src/bin/dag_export.rs
index fe3a7f36..d7d44dac 100644
--- a/crates/devtools/src/bin/dag_export.rs
+++ b/crates/devtools/src/bin/dag_export.rs
@@ -1824,21 +1824,28 @@ mod tests {
                 cost_per_network_byte: 1.0,
                 cost_per_materialization_byte: 1.0,
             },
-            nodes: std::collections::HashMap::new(),
+            plans: vec![],
         };
         for dag in [&raw, &candidate_dag] {
-            for node in &dag.nodes {
-                profile.nodes.insert(
-                    node.id.clone(),
-                    BoundaryNodeEvidence {
-                        node: node.clone(),
-                        statistics: dag.evidence[&node.id].statistics.clone(),
-                        boundaries: vec![],
-                    },
-                );
-            }
+            profile.plans.push(BoundaryPlanEvidence {
+                root: dag.root.clone(),
+                nodes: dag
+                    .nodes
+                    .iter()
+                    .map(|node| {
+                        (
+                            node.id.clone(),
+                            BoundaryNodeEvidence {
+                                node: node.clone(),
+                                statistics: dag.evidence[&node.id].statistics.clone(),
+                                boundaries: vec![],
+                            },
+                        )
+                    })
+                    .collect(),
+            });
         }
-        profile
+        profile.plans[1]
             .nodes
             .get_mut("summary-read")
             .unwrap()
@@ -1894,6 +1901,25 @@ mod tests {
             selected.evidence_version.as_deref(),
             Some("test-evidence-v1")
         );
+        let mut boundary_only = parsed.clone();
+        boundary_only.calibration.cost_per_cpu_op = 0.0;
+        boundary_only.calibration.cost_per_scan_byte = 0.0;
+        boundary_only.calibration.cost_per_retained_byte = 0.0;
+        let (baseline, selected, _) = ExportPlannerCostModel {
+            document: &boundary_only,
+        }
+        .annotations(&candidate, &root);
+        assert_eq!(baseline.value, Some(0.0));
+        assert_eq!(selected.value, Some(12_000.0));
+
+        let mut ambiguous = parsed.clone();
+        let plans = &mut ambiguous.boundaries.as_mut().unwrap().plans;
+        plans.push(plans[0].clone());
+        let model = ExportPlannerCostModel {
+            document: &ambiguous,
+        };
+        assert!(model.candidate_cost(&candidate, &target).is_none());
+        assert!(model.annotations(&candidate, &root).0.value.is_none());
         document
             .boundaries
             .as_mut()
@@ -1905,10 +1931,7 @@ mod tests {
         }
         .candidate_cost(&candidate, &target)
         .is_none());
-        document
-            .boundaries
-            .as_mut()
-            .unwrap()
+        document.boundaries.as_mut().unwrap().plans[0]
             .nodes
             .remove(&raw.root);
         assert!(ExportPlannerCostModel {
diff --git a/docs/developer_docs/physical-boundary-costs.md b/docs/developer_docs/physical-boundary-costs.md
index d2ac3552..b20ba902 100644
--- a/docs/developer_docs/physical-boundary-costs.md
+++ b/docs/developer_docs/physical-boundary-costs.md
@@ -5,12 +5,15 @@ immutable `PhysicalEvidenceSnapshot`. `dag_export --planner-cost-json` accepts
 the same profile in a top-level `boundaries` field. With no profile, these
 dimensions remain unestimated and the existing resource objective is preserved.
 
-The profile is supplementary physical-plan binding: each entry supplies the
-complete physical node, its authoritative statistics, and an explicit list of
-boundary actions on its output. Every reachable node needs an entry; an empty
-list declares ordinary in-memory dataflow with no boundary traffic. The full
-node/statistics match prevents rebinding evidence to a different operator with
-the same ID. Additional entries may describe other candidate plans. Evidence
+The profile's `plans` list binds boundaries to complete physical alternatives.
+Each plan supplies its `root` and a `nodes` map containing every physical node,
+its authoritative statistics, and an explicit list of boundary actions on its
+output. An empty boundary list declares ordinary in-memory dataflow with no
+boundary traffic. Exactly one plan must match the root, complete node set, and
+node/statistics snapshot; missing or ambiguous matches fail closed. This lets
+alternatives reuse a producer identity while declaring different transfers to
+their respective consumers. All alternatives share the profile's immutable
+evidence generation and calibration. Evidence
 must match the immutable snapshot version and be current at planning time:
 `observed_at_ms <= planning_time < valid_until_ms`.
 
@@ -52,6 +55,9 @@ A `BoundaryCalibration` supplies finite, nonnegative cost coefficients in the
 same cost units as the base resource model, and a nonempty version. Both
 alternatives use that profile when ranking. This models byte work, not transfer
 latency, bandwidth contention, memory lifetime, or storage requests.
+Base CPU, scan, and retained-memory coefficients may all be zero when at least
+one boundary coefficient is positive. An absent boundary profile or an entirely
+zero objective remains unavailable for ranking.
 
 Annotations expose totals in `bytes`, plus terms named
 `physical_node:<id>:<dimension>` and `boundary:<id>:<dimension>`. The existing

From 11d53b30aa78b652f1c7fc978a3069003a877b6a Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Tue, 8 Sep 2026 12:54:58 -0600
Subject: [PATCH 3/8] fix(cost): require calibration provenance for boundary
 ranking

---
 .../src/physical_plan_cost_model.rs           | 22 +++++++++
 .../asap-aware-mapping/tests/boundary_cost.rs | 47 +++++++++++++++++++
 .../developer_docs/physical-boundary-costs.md |  2 +
 3 files changed, 71 insertions(+)

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 ace1b34c..f56bdd2f 100644
--- a/crates/asap-aware-mapping/src/physical_plan_cost_model.rs
+++ b/crates/asap-aware-mapping/src/physical_plan_cost_model.rs
@@ -712,6 +712,28 @@ mod tests {
         assert!(estimate.storage_io.is_none());
     }
 
+    // Combined objectives must identify the base calibration even when its
+    // coefficients are zero and boundaries supply the entire objective.
+    #[test]
+    fn blank_base_calibration_version_is_rejected() {
+        let provider = TestProvider::new(true, 800);
+        for version in ["", " \t\n"] {
+            for boundary_only in [false, true] {
+                let mut calibration = calibration();
+                calibration.version = version.into();
+                if boundary_only {
+                    calibration.cost_per_cpu_op = 0.0;
+                    calibration.cost_per_scan_byte = 0.0;
+                    calibration.cost_per_retained_byte = 0.0;
+                }
+                assert!(
+                    PhysicalPlanCostModel::new(&provider, calibration).is_err(),
+                    "blank base provenance accepted (boundary_only={boundary_only})"
+                );
+            }
+        }
+    }
+
     // Explicit byte pricing can rank complete plans without pricing CPU or scans.
     #[test]
     fn boundary_only_objective_ranks_complete_plans() {
diff --git a/crates/asap-aware-mapping/tests/boundary_cost.rs b/crates/asap-aware-mapping/tests/boundary_cost.rs
index 97136de3..8e48d068 100644
--- a/crates/asap-aware-mapping/tests/boundary_cost.rs
+++ b/crates/asap-aware-mapping/tests/boundary_cost.rs
@@ -365,3 +365,50 @@ fn missing_stale_and_non_finite_evidence_is_rejected() {
     assert!(estimate_boundaries(&dag, &scope, &evidence, "evidence-v1").is_err());
     assert!(estimate_boundaries(&dag, &scope, &profile(&dag), "different").is_err());
 }
+
+// Individual actions may fit while accumulation across actions or nodes overflows.
+#[test]
+fn boundary_accumulation_and_calibrated_cost_overflow_are_rejected() {
+    use asap_aware_mapping::analytical_cost::AnalyticalCostError;
+    let (mut dag, mut scope) = fixture();
+    scope.recurrence = QueryRecurrence::OneTime {
+        invocations: 1,
+        execute_at: None,
+    };
+    for node in &mut dag.nodes {
+        node.execution = ExecutionMultiplicity::Once;
+    }
+    for different_nodes in [false, true] {
+        let mut evidence = profile(&dag);
+        for (id, node) in [
+            ("first", "scan"),
+            ("second", if different_nodes { "left" } else { "scan" }),
+        ] {
+            let mut boundary = transfer(id, None);
+            boundary.copies = 1;
+            boundary.encoded_bytes = u64::MAX / 2 + 1;
+            evidence.plans[0]
+                .nodes
+                .get_mut(node)
+                .unwrap()
+                .boundaries
+                .push(boundary);
+        }
+        assert!(matches!(
+            estimate_boundaries(&dag, &scope, &evidence, "evidence-v1"),
+            Err(AnalyticalCostError::Overflow)
+        ));
+    }
+    let mut evidence = profile(&dag);
+    evidence.plans[0]
+        .nodes
+        .get_mut("scan")
+        .unwrap()
+        .boundaries
+        .push(transfer("wire", None));
+    evidence.calibration.cost_per_network_byte = f64::MAX;
+    assert!(matches!(
+        estimate_boundaries(&dag, &scope, &evidence, "evidence-v1"),
+        Err(AnalyticalCostError::Overflow)
+    ));
+}
diff --git a/docs/developer_docs/physical-boundary-costs.md b/docs/developer_docs/physical-boundary-costs.md
index b20ba902..7250281f 100644
--- a/docs/developer_docs/physical-boundary-costs.md
+++ b/docs/developer_docs/physical-boundary-costs.md
@@ -58,6 +58,8 @@ latency, bandwidth contention, memory lifetime, or storage requests.
 Base CPU, scan, and retained-memory coefficients may all be zero when at least
 one boundary coefficient is positive. An absent boundary profile or an entirely
 zero objective remains unavailable for ranking.
+Both the base and boundary calibration versions must be nonempty, including
+when the base coefficients are all zero; combined annotations identify both.
 
 Annotations expose totals in `bytes`, plus terms named
 `physical_node:<id>:<dimension>` and `boundary:<id>:<dimension>`. The existing

From 2874c798994431d2caf0399fd14506a0d851f032 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Tue, 8 Sep 2026 13:10:45 -0600
Subject: [PATCH 4/8] refactor(types): share physical boundary resource
 vocabulary

---
 .../asap-aware-mapping/src/boundary_cost.rs   |  55 +-------
 .../asap-aware-mapping/tests/boundary_cost.rs |  28 ++++
 crates/types/src/resources.rs                 |   2 +
 crates/types/src/resources/boundary.rs        | 126 ++++++++++++++++++
 .../developer_docs/physical-boundary-costs.md |  10 ++
 5 files changed, 173 insertions(+), 48 deletions(-)
 create mode 100644 crates/types/src/resources/boundary.rs

diff --git a/crates/asap-aware-mapping/src/boundary_cost.rs b/crates/asap-aware-mapping/src/boundary_cost.rs
index 3f5a2e52..2474e72e 100644
--- a/crates/asap-aware-mapping/src/boundary_cost.rs
+++ b/crates/asap-aware-mapping/src/boundary_cost.rs
@@ -5,31 +5,12 @@ use crate::analytical_cost::{
     PhysicalDagNode,
 };
 use crate::physical_operator_statistics::{ComparisonScope, OperatorStatistics};
+pub use asap_types::resources::{BoundaryKind, BoundaryResources, MaterializationMedium};
 use serde::{Deserialize, Serialize};
 use std::collections::{HashMap, HashSet};
 
 pub const BOUNDARY_MODEL_VERSION: &str = "physical-boundary-bytes-v1";
 
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
-pub enum BoundaryKind {
-    Network {
-        source_location: String,
-        destination_location: String,
-    },
-    Materialization {
-        medium: MaterializationMedium,
-    },
-}
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-pub enum MaterializationMedium {
-    Memory,
-    Disk,
-    ObjectStore,
-}
-
 /// A physical action on a producer output, distinct from a logical DAG edge.
 /// With `consumer = None`, one action serves all consumers. With a consumer,
 /// it is a separate action per execution of that downstream physical node.
@@ -83,32 +64,6 @@ pub struct BoundaryCalibration {
     pub cost_per_materialization_byte: f64,
 }
 
-#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
-pub struct BoundaryResources {
-    pub network_bytes: u64,
-    pub materialization_bytes: u64,
-}
-
-impl BoundaryResources {
-    pub fn terms(self) -> [(&'static str, u64); 2] {
-        [
-            ("network_bytes", self.network_bytes),
-            ("materialization_bytes", self.materialization_bytes),
-        ]
-    }
-    fn add(&mut self, other: Self) -> Result<(), AnalyticalCostError> {
-        self.network_bytes = self
-            .network_bytes
-            .checked_add(other.network_bytes)
-            .ok_or(AnalyticalCostError::Overflow)?;
-        self.materialization_bytes = self
-            .materialization_bytes
-            .checked_add(other.materialization_bytes)
-            .ok_or(AnalyticalCostError::Overflow)?;
-        Ok(())
-    }
-}
-
 impl BoundaryCalibration {
     pub fn cost(&self, value: BoundaryResources) -> Result<f64, AnalyticalCostError> {
         if self.version.trim().is_empty()
@@ -255,10 +210,14 @@ pub fn estimate_boundaries(
                 }
                 BoundaryKind::Materialization { .. } => term.materialization_bytes = bytes,
             }
-            local.add(term)?;
+            local = local
+                .checked_add(term)
+                .ok_or(AnalyticalCostError::Overflow)?;
             per_boundary.insert(boundary.id.clone(), term);
         }
-        total.add(local)?;
+        total = total
+            .checked_add(local)
+            .ok_or(AnalyticalCostError::Overflow)?;
         per_node.insert((*id).into(), local);
     }
     Ok(BoundaryEstimate {
diff --git a/crates/asap-aware-mapping/tests/boundary_cost.rs b/crates/asap-aware-mapping/tests/boundary_cost.rs
index 8e48d068..3da56535 100644
--- a/crates/asap-aware-mapping/tests/boundary_cost.rs
+++ b/crates/asap-aware-mapping/tests/boundary_cost.rs
@@ -138,6 +138,34 @@ fn fixture() -> (EvidenceBackedPhysicalDag, ComparisonScope) {
 
 use asap_aware_mapping::boundary_cost::*;
 
+// Legacy mapping imports and the shared resource API are the very same Rust types.
+#[test]
+fn mapping_resource_reexports_are_wire_compatible_shared_types() {
+    let shared = asap_types::resources::BoundaryResources {
+        network_bytes: 480,
+        materialization_bytes: 40,
+    };
+    let legacy: BoundaryResources = shared;
+    assert_eq!(
+        serde_json::to_value(legacy).unwrap(),
+        serde_json::json!({"network_bytes": 480, "materialization_bytes": 40})
+    );
+    let shared_kind = asap_types::resources::BoundaryKind::Materialization {
+        medium: asap_types::resources::MaterializationMedium::Disk,
+    };
+    let mut boundary = transfer("persist", None);
+    boundary.kind = shared_kind;
+    let json = serde_json::to_value(&boundary).unwrap();
+    assert_eq!(
+        json["kind"],
+        serde_json::json!({"kind": "materialization", "medium": "disk"})
+    );
+    assert_eq!(
+        serde_json::from_value::<PhysicalBoundary>(json).unwrap(),
+        boundary
+    );
+}
+
 fn profile(dag: &EvidenceBackedPhysicalDag) -> BoundaryProfile {
     BoundaryProfile {
         evidence_version: "evidence-v1".into(),
diff --git a/crates/types/src/resources.rs b/crates/types/src/resources.rs
index d2107a7f..c37e761b 100644
--- a/crates/types/src/resources.rs
+++ b/crates/types/src/resources.rs
@@ -17,6 +17,8 @@ pub use storage::StorageResources;
 pub use cpu::{MeasuredCpu, ModeledCpu};
 pub use measurement::Measurement;
 pub use physical::{MeasuredResources, PhysicalResources};
+pub mod boundary;
+pub use boundary::{BoundaryKind, BoundaryResources, MaterializationMedium};
 
 #[cfg(test)]
 mod tests {
diff --git a/crates/types/src/resources/boundary.rs b/crates/types/src/resources/boundary.rs
new file mode 100644
index 00000000..266a040d
--- /dev/null
+++ b/crates/types/src/resources/boundary.rs
@@ -0,0 +1,126 @@
+//! Physical transfer and materialization dimensions, independent of planner policy.
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
+pub enum BoundaryKind {
+    Network {
+        source_location: String,
+        destination_location: String,
+    },
+    Materialization {
+        medium: MaterializationMedium,
+    },
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum MaterializationMedium {
+    Memory,
+    Disk,
+    ObjectStore,
+}
+
+/// Byte work at explicitly declared physical actions, not storage occupancy.
+/// Network traffic and materialization writes remain separate dimensions.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct BoundaryResources {
+    pub network_bytes: u64,
+    pub materialization_bytes: u64,
+}
+
+impl BoundaryResources {
+    pub fn terms(self) -> [(&'static str, u64); 2] {
+        [
+            ("network_bytes", self.network_bytes),
+            ("materialization_bytes", self.materialization_bytes),
+        ]
+    }
+
+    /// Add independent dimensions without wrapping or partially mutating either input.
+    pub fn checked_add(self, other: Self) -> Option<Self> {
+        Some(Self {
+            network_bytes: self.network_bytes.checked_add(other.network_bytes)?,
+            materialization_bytes: self
+                .materialization_bytes
+                .checked_add(other.materialization_bytes)?,
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// The shared vocabulary retains the original flat counters and tagged kind JSON.
+    #[test]
+    fn boundary_json_format_is_unchanged() {
+        let bytes = BoundaryResources {
+            network_bytes: 480,
+            materialization_bytes: 40,
+        };
+        let json = serde_json::json!({"network_bytes": 480, "materialization_bytes": 40});
+        assert_eq!(serde_json::to_value(bytes).unwrap(), json);
+        assert_eq!(
+            serde_json::from_value::<BoundaryResources>(json).unwrap(),
+            bytes
+        );
+        let network = BoundaryKind::Network {
+            source_location: "edge".into(),
+            destination_location: "backend".into(),
+        };
+        assert_eq!(
+            serde_json::to_value(network).unwrap(),
+            serde_json::json!({
+                "kind": "network", "source_location": "edge", "destination_location": "backend"
+            })
+        );
+        for (medium, name) in [
+            (MaterializationMedium::Memory, "memory"),
+            (MaterializationMedium::Disk, "disk"),
+            (MaterializationMedium::ObjectStore, "object_store"),
+        ] {
+            let kind = BoundaryKind::Materialization { medium };
+            let json = serde_json::json!({"kind": "materialization", "medium": name});
+            assert_eq!(serde_json::to_value(&kind).unwrap(), json);
+            assert_eq!(serde_json::from_value::<BoundaryKind>(json).unwrap(), kind);
+        }
+    }
+
+    /// Either dimension overflowing returns None without changing the original counters.
+    #[test]
+    fn checked_add_preserves_dimensions_and_rejects_overflow() {
+        let first = BoundaryResources {
+            network_bytes: 2,
+            materialization_bytes: 3,
+        };
+        let second = BoundaryResources {
+            network_bytes: 5,
+            materialization_bytes: 7,
+        };
+        assert_eq!(
+            first.checked_add(second),
+            Some(BoundaryResources {
+                network_bytes: 7,
+                materialization_bytes: 10,
+            })
+        );
+        for overflowing in [
+            BoundaryResources {
+                network_bytes: u64::MAX,
+                materialization_bytes: 0,
+            },
+            BoundaryResources {
+                network_bytes: 0,
+                materialization_bytes: u64::MAX,
+            },
+        ] {
+            assert!(first.checked_add(overflowing).is_none());
+        }
+        assert_eq!(
+            first.terms(),
+            [("network_bytes", 2), ("materialization_bytes", 3)]
+        );
+    }
+}
diff --git a/docs/developer_docs/physical-boundary-costs.md b/docs/developer_docs/physical-boundary-costs.md
index 7250281f..03ad49de 100644
--- a/docs/developer_docs/physical-boundary-costs.md
+++ b/docs/developer_docs/physical-boundary-costs.md
@@ -1,5 +1,15 @@
 # Physical boundary byte estimates
 
+`asap_types::resources` owns the canonical `BoundaryResources`, `BoundaryKind`,
+and `MaterializationMedium` definitions in `resources/boundary.rs`. The mapping
+crate re-exports those same types from `boundary_cost` for import compatibility;
+all estimator and export consumers therefore use shared definitions, not copies.
+Their existing JSON format is unchanged. The shared byte counters support checked
+addition without depending on planner errors. Snapshot binding, validation,
+calibration, and ranking remain in the mapping crate. Boundary traffic/write work
+is distinct from CPU work, scanned bytes, and stored byte occupancy; it is not
+collapsed into the generic CPU/byte resource container.
+
 The physical-plan adapter accepts an optional `BoundaryProfile` in the
 immutable `PhysicalEvidenceSnapshot`. `dag_export --planner-cost-json` accepts
 the same profile in a top-level `boundaries` field. With no profile, these

From 43fabf6bc8fdcba2a9aaba2271ca451b59d9cee4 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Tue, 8 Sep 2026 13:19:19 -0600
Subject: [PATCH 5/8] refactor(resources): split CPU measurement and physical
 types into modules

---
 crates/types/src/resources.rs | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/crates/types/src/resources.rs b/crates/types/src/resources.rs
index c37e761b..67734816 100644
--- a/crates/types/src/resources.rs
+++ b/crates/types/src/resources.rs
@@ -6,20 +6,19 @@
 //! Cache assumptions share this schema namespace but are not additive resource
 //! consumption; their numerical interpretation belongs to an estimator.
 
+pub mod boundary;
 pub mod cache;
 pub mod cpu;
 pub mod measurement;
 pub mod physical;
 pub mod storage;
+pub use boundary::{BoundaryKind, BoundaryResources, MaterializationMedium};
 pub use cache::{CacheCapacityEvidence, CacheEvidence, CacheProfile};
 pub use storage::StorageResources;
 
 pub use cpu::{MeasuredCpu, ModeledCpu};
 pub use measurement::Measurement;
 pub use physical::{MeasuredResources, PhysicalResources};
-pub mod boundary;
-pub use boundary::{BoundaryKind, BoundaryResources, MaterializationMedium};
-
 #[cfg(test)]
 mod tests {
     use super::*;

From 64d63809abbb41a18f6583cdf81a8c20ae22905a Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Tue, 8 Sep 2026 13:23:38 -0600
Subject: [PATCH 6/8] docs(resources): clarify network and materialization
 boundary scope

---
 crates/types/src/resources/boundary.rs | 17 ++++++++++++++---
 1 file changed, 14 insertions(+), 3 deletions(-)

diff --git a/crates/types/src/resources/boundary.rs b/crates/types/src/resources/boundary.rs
index 266a040d..e1972af5 100644
--- a/crates/types/src/resources/boundary.rs
+++ b/crates/types/src/resources/boundary.rs
@@ -1,17 +1,26 @@
 //! Physical transfer and materialization dimensions, independent of planner policy.
+//!
+//! A boundary is either a network transfer between execution locations or a
+//! materialization of an intermediate result in memory, on disk, or in an
+//! object store. This module defines their kinds, materialization media, and
+//! separate byte-work counters; it is not limited to network resources.
+//!
+//! Counters describe bytes transferred or materialized, not retained memory,
+//! allocated disk space, or storage request counts. Estimation, evidence
+//! validation, and calibration belong to the cost model, not this schema.
 
 use serde::{Deserialize, Serialize};
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
 pub enum BoundaryKind {
+    /// Transfer encoded data between execution locations.
     Network {
         source_location: String,
         destination_location: String,
     },
-    Materialization {
-        medium: MaterializationMedium,
-    },
+    /// Write an intermediate result to the specified medium for later use.
+    Materialization { medium: MaterializationMedium },
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -26,7 +35,9 @@ pub enum MaterializationMedium {
 /// Network traffic and materialization writes remain separate dimensions.
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
 pub struct BoundaryResources {
+    /// Bytes transferred across explicitly declared network boundaries.
     pub network_bytes: u64,
+    /// Bytes written across materialization boundaries, regardless of medium.
     pub materialization_bytes: u64,
 }
 

From d859bb8af2d6a80dd2c6c739dd9951487bf090de Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Tue, 8 Sep 2026 13:27:46 -0600
Subject: [PATCH 7/8] fix(cost): reject boundary profiles without cache-aware
 execution evidence

---
 .../src/physical_plan_cost_model.rs           |  9 +++++
 crates/devtools/src/bin/dag_export.rs         | 36 +++++++++++++++++++
 .../developer_docs/physical-boundary-costs.md |  7 ++++
 3 files changed, 52 insertions(+)

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 f56bdd2f..ad70cdc3 100644
--- a/crates/asap-aware-mapping/src/physical_plan_cost_model.rs
+++ b/crates/asap-aware-mapping/src/physical_plan_cost_model.rs
@@ -171,6 +171,15 @@ impl<'a> PhysicalPlanCostModel<'a> {
             ));
         }
         let scope = &snapshot.scope;
+        if snapshot.boundaries.is_some()
+            && matches!(snapshot.cache_profile, CacheProfile::Evidence(_))
+        {
+            // Scope-based boundary multiplicity does not describe which actions
+            // cache hits skip; do not mix pre-cache byte work with discounted CPU.
+            return Err(AnalyticalCostError::MissingOrStale(
+                "cache-aware boundary execution evidence",
+            ));
+        }
         let evidence = QueryEvidence {
             provider: self.provider,
             snapshot: &snapshot,
diff --git a/crates/devtools/src/bin/dag_export.rs b/crates/devtools/src/bin/dag_export.rs
index d7d44dac..c234895e 100644
--- a/crates/devtools/src/bin/dag_export.rs
+++ b/crates/devtools/src/bin/dag_export.rs
@@ -1912,6 +1912,42 @@ mod tests {
         assert_eq!(baseline.value, Some(0.0));
         assert_eq!(selected.value, Some(12_000.0));
 
+        // Explicit boundaries have no post-cache execution-count evidence.
+        // A nontrivial cache profile must fail closed instead of combining
+        // discounted CPU with pre-cache transfer multiplicity.
+        let mut combined = serde_json::to_value(&parsed).unwrap();
+        combined["targets"][0]["scope"]["cache_profile"] = serde_json::json!({
+            "profile": "evidence", "version": "combined-cache-v1",
+            "distinct_evaluations": 2, "repeated_identical_evaluations": 8,
+            "result_cache": {"working_set_bytes": 100, "capacity_bytes": 100},
+            "buffer_cache": {"working_set_bytes": 1000, "capacity_bytes": 500},
+            "result_invalidation_ratio": null
+        });
+        let combined = parse_planner_cost_document(&combined.to_string()).unwrap();
+        let combined_model = ExportPlannerCostModel {
+            document: &combined,
+        };
+        assert!(combined_model.candidate_cost(&candidate, &target).is_none());
+        assert!(combined_model
+            .annotations(&candidate, &root)
+            .0
+            .value
+            .is_none());
+        let (uncached_raw, uncached_selected, _) =
+            ExportPlannerCostModel { document: &parsed }.annotations(&candidate, &root);
+        for annotation in [&uncached_raw, &uncached_selected] {
+            assert_eq!(annotation.cache_profile.as_deref(), Some("no-cache-v1"));
+            assert_eq!(
+                annotation.evidence_version.as_deref(),
+                Some("test-evidence-v1")
+            );
+            assert!(annotation
+                .model_version
+                .as_ref()
+                .unwrap()
+                .contains(BOUNDARY_MODEL_VERSION));
+        }
+
         let mut ambiguous = parsed.clone();
         let plans = &mut ambiguous.boundaries.as_mut().unwrap().plans;
         plans.push(plans[0].clone());
diff --git a/docs/developer_docs/physical-boundary-costs.md b/docs/developer_docs/physical-boundary-costs.md
index 03ad49de..4b0d2965 100644
--- a/docs/developer_docs/physical-boundary-costs.md
+++ b/docs/developer_docs/physical-boundary-costs.md
@@ -52,6 +52,13 @@ reachable parent of the producer; multiplicity comes from that consumer.
 `Once` means one execution; `PerEvaluation` uses the comparison scope's demand.
 Shared producers are traversed once regardless of fan-out. Duplicate boundary
 IDs fail closed instead of being ambiguously counted or silently dropped.
+Boundary profiles currently require `CacheProfile::NoCache`. Cache evidence does
+not identify which physical transfers or materializations are skipped on a hit,
+and boundary execution counts derive from the comparison scope rather than a
+post-cache schedule. Combining a boundary profile with `CacheProfile::Evidence`
+therefore fails closed until cache-aware boundary execution evidence exists.
+Supported exports retain the no-cache profile provenance alongside boundary
+model/calibration versions.
 
 Example: a retained producer materializes 40 encoded bytes once. Two consumers
 each receive two copies of those 40 bytes over three evaluations. The totals

From 6da1673b8c3b98f3135d914afa76cc8d70674007 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Tue, 8 Sep 2026 13:33:19 -0600
Subject: [PATCH 8/8] test(cost): verify joint storage and boundary objective
 integration

---
 crates/devtools/src/bin/dag_export.rs | 81 +++++++++++++++++++++++++++
 1 file changed, 81 insertions(+)

diff --git a/crates/devtools/src/bin/dag_export.rs b/crates/devtools/src/bin/dag_export.rs
index c234895e..8c95fab6 100644
--- a/crates/devtools/src/bin/dag_export.rs
+++ b/crates/devtools/src/bin/dag_export.rs
@@ -1912,6 +1912,87 @@ mod tests {
         assert_eq!(baseline.value, Some(0.0));
         assert_eq!(selected.value, Some(12_000.0));
 
+        // Independent supplemental objectives add once, and either one can
+        // price a zero-base plan while retaining both sets of export evidence.
+        {
+            use asap_aware_mapping::storage_io::*;
+            let mut joint = boundary_only.clone();
+            let mut storage = StorageIoProfile {
+                evidence_version: joint.evidence_version.clone(),
+                observed_at_ms: 900,
+                valid_until_ms: 2000,
+                calibration: StorageCalibration {
+                    version: "joint-requests-v1".into(),
+                    cost_per_disk_read: 1.0,
+                    cost_per_disk_write: 1.0,
+                    cost_per_object_get: 1.0,
+                    cost_per_object_put: 1.0,
+                },
+                nodes: std::collections::HashMap::new(),
+            };
+            for dag in [&raw, &candidate_dag] {
+                for node in &dag.nodes {
+                    let statistics = dag.evidence[&node.id].statistics.clone();
+                    let accesses = match &statistics {
+                        OperatorStatistics::Scan {
+                            source_read_bytes, ..
+                        } => vec![StorageAccess {
+                            operation: StorageOperation::ObjectGet,
+                            extent_bytes: vec![*source_read_bytes],
+                            bytes_per_request: 4096,
+                        }],
+                        _ => vec![],
+                    };
+                    storage.nodes.insert(
+                        node.id.clone(),
+                        StorageNodeEvidence {
+                            node: node.clone(),
+                            statistics,
+                            accesses,
+                        },
+                    );
+                }
+            }
+            joint.storage_io = Some(storage);
+            let joint =
+                parse_planner_cost_document(&serde_json::to_string(&joint).unwrap()).unwrap();
+            let (raw_cost, candidate_cost, _) =
+                ExportPlannerCostModel { document: &joint }.annotations(&candidate, &root);
+            assert_eq!(raw_cost.value, Some(160.0));
+            assert_eq!(candidate_cost.value, Some(12_010.0));
+            for annotation in [&raw_cost, &candidate_cost] {
+                let version = annotation.model_version.as_ref().unwrap();
+                assert!(version.contains(STORAGE_IO_MODEL_VERSION));
+                assert!(version.contains(BOUNDARY_MODEL_VERSION));
+                assert!(version.contains("joint-requests-v1"));
+                assert_eq!(annotation.cache_profile.as_deref(), Some("no-cache-v1"));
+                assert!(annotation
+                    .inputs
+                    .iter()
+                    .any(|term| term.name == "object_get_operations"));
+                assert!(annotation
+                    .inputs
+                    .iter()
+                    .any(|term| term.name == "network_bytes"));
+            }
+            let mut storage_only = joint.clone();
+            storage_only.boundaries = None;
+            assert_eq!(
+                ExportPlannerCostModel {
+                    document: &storage_only
+                }
+                .candidate_cost(&candidate, &target),
+                Some(Cost(10.0))
+            );
+            let mut neither = storage_only;
+            neither.storage_io = None;
+            assert!(ExportPlannerCostModel { document: &neither }
+                .annotations(&candidate, &root)
+                .0
+                .value
+                .is_none());
+        }
+
         // Explicit boundaries have no post-cache execution-count evidence.
         // A nontrivial cache profile must fail closed instead of combining
         // discounted CPU with pre-cache transfer multiplicity.