diff --git a/crates/devtools/Cargo.toml b/crates/devtools/Cargo.toml
index edf20aef..944858a3 100644
--- a/crates/devtools/Cargo.toml
+++ b/crates/devtools/Cargo.toml
@@ -16,7 +16,7 @@ asap-logical-optimizer = { path = "../logical-optimizer" }
 # plans with. A devtool, not a stage crate, so the #572 guards allow it.
 asap-executor = { path = "../executor" }
 
-# Used by the show_ir / dag_export / variant_coverage bins (catalog schemas,
+# Used by the show_*_ir / stage_pipeline / variant_coverage bins (catalog schemas,
 # async SQL path, JSON output) and by the topk_ir / canonical_examples
 # examples. Regular deps, not dev-deps: `[[bin]]` targets can't see dev-deps.
 asap-types = { path = "../types" }
diff --git a/crates/devtools/src/bin/dag_export.rs b/crates/devtools/src/bin/dag_export.rs
deleted file mode 100644
index f94a4078..00000000
--- a/crates/devtools/src/bin/dag_export.rs
+++ /dev/null
@@ -1,3188 +0,0 @@
-// cargo run -p asap-lower --bin dag_export -- \
-//     --sql "SELECT service, COUNT(*) FROM metrics GROUP BY service" --name q1 \
-//     --promql "topk(5, rate(http_requests_total[5m]))" --name q2
-//
-// Lowers each given SQL/PromQL query to pre-ASAP IR and prints a single
-// `asap_types::dag_export::WorkloadDAG` as JSON on stdout — the input format
-// for `tools/dag-viewer` (issue #133). Redirect to a file and load it there:
-//   cargo run -p asap-lower --bin dag_export -- --sql "..." --name q1 > /tmp/dag.json
-//
-// `--name` is optional; an unnamed query defaults to `q<n>` (1-indexed).
-//
-// `--epsilon <f64>` is optional and applies to every query in the run: it
-// lowers with `AccuracyTarget::Epsilon(<f64>)` instead of the default
-// `AccuracyTarget::Exact`. Without it, every `AggIntent` lowers exact and
-// `asap_logical_optimizer::ASAPStrategies` never has a genuine sketch
-// alternative to report — so no node ever picks up a `SketchApproximation`
-// note. Pass it to actually exercise that path, e.g.:
-//   cargo run -p asap-lower --bin dag_export -- \
-//       --epsilon 0.01 --sql "SELECT quantile(0.99, latency) FROM metrics" --name p99
-//
-// `--post-asap` needs one of two cost sources to rank with, and does
-// nothing without either (see `--default-cost` and `--planner-cost-json`
-// below).
-//
-// `--post-asap` is optional and off by default. When passed, this binary
-// additionally runs `asap_logical_optimizer::pass1::replacement::search_workload` (this
-// binary took no strategies of its own — `default_strategies()` already
-// includes `AvgToSumOverCountStrategy` as of #282) over every lowered query
-// and ranks each discovered `TargetSubDAGCandidates` via `candidate_selection::cost_sorted`. The
-// best-ranked
-// candidate per group feeds two additive outputs:
-//
-//   - one `asap_types::dag_export::TargetReplacement` per group on whichever
-//     query's `NamedDAG.replacements` contains that target node (matched
-//     by `DAGNode::hash` + structural equality, the same collision-safe
-//     pattern `annotate_with_explanations` below already uses for notes) —
-//     a small, self-contained "before -> after" pair per replacement site;
-//   - one merged `NamedDAG.post_dag`: a single flattened DAG per
-//     query with every winning candidate spliced directly into the query's
-//     own pre-ASAP shape in place, built via
-//     `asap_types::dag_export::export_post_asap`.
-//
-// Together these surface every one of the four concrete replacement kinds:
-// the sketch family `ASAPStrategies`/`HydraGroupingStrategy` bound,
-// the CSE share/recompute choice `SharedSubDAGStrategy` found, the
-// workload-aware roll-up `RollupStrategy` derived, and the `avg ->
-// sum/count` rewrite `AvgToSumOverCountStrategy` proposes. Without
-// `--post-asap`, every existing invocation of this binary produces
-// byte-identical output to before (`NamedDAG.replacements` is empty and
-// `post_dag` is `None`, both skipped from the JSON entirely in that
-// case). E.g.:
-//   cargo run -p asap-lower --bin dag_export -- \
-//       --post-asap --default-cost --epsilon 0.01 \
-//       --sql "SELECT quantile(0.95, latency) FROM metrics" --name q1
-//
-// `--default-cost` and `--planner-cost-json` are the two mutually exclusive
-// ways to give `--post-asap` a cost model, and they differ in what the
-// export is allowed to claim:
-//
-//   - `--planner-cost-json <doc>` supplies complete deployment-owned
-//     physical-plan evidence. It both ranks the candidates and is exported:
-//     every decision carries a calibrated `CostUnits` annotation.
-//   - `--default-cost` ranks with `asap_plan_selection::cost::cost_model::
-//     DefaultCostModel` — structural node counts, owning no deployment
-//     evidence. The structure of the result is real (which replacements the
-//     search found, which one won per group, what the merged post-ASAP DAG
-//     looks like); the numbers are not exported at all. Every decision's
-//     cost is `CostSource::Unavailable` with `value: None`, which the viewer
-//     renders as "Not estimated". Use it to see what ASAPPlanner does with a
-//     workload before there is a deployment to measure.
-//
-// Absent both, `--post-asap` exports the raw plan only.
-
-use std::cell::RefCell;
-use std::collections::HashMap;
-use std::rc::Rc;
-use std::time::Instant;
-
-use asap_logical_optimizer::pass1::replacement::{
-    default_strategies_with_evidence, is_logical_rewrite, search_workload, search_workload_with,
-    Replacement, ReplacementSubDAG,
-};
-use asap_logical_optimizer::{AccuracyEvidenceProvider, PropagationStats};
-use asap_plan_selection::candidate_selection::global_selection;
-use asap_plan_selection::cost::analytical_cost::{
-    cache_hit_ratios, AnalyticalCostError, EvidenceBackedPhysicalDAG as PhysicalDAG,
-    PhysicalNodeEvidence, ResourceCalibration, ANALYTICAL_COST_MODEL_VERSION,
-};
-use asap_plan_selection::cost::cost_model::DefaultCostModel;
-use asap_plan_selection::cost::cost_model::{Cost, CostModel};
-use asap_plan_selection::cost::physical_operator_statistics::ComparisonScope;
-use asap_plan_selection::cost::physical_plan_cost_model::{
-    PhysicalEvidenceSnapshot, PhysicalPlanCostModel, PlannerPhysicalPlanProvider,
-};
-use asap_plan_selection::cost::query_physical_lowering::PhysicalNodeRequest;
-use asap_types::cost::{BaselineRef, CostAnnotation, CostInput, CostSource, CostUnit};
-use asap_types::dag_export::{
-    self, DAGDecision, DAGNote, ExportDAG, NamedDAG, PostAsapSubstitution, TargetRejection,
-    TargetReplacement, TargetReplacementAfter, WorkloadDAG,
-};
-use asap_types::ir::cse::{structural_hash, HashCache};
-use asap_types::ir::properties::CompositionOperator;
-use asap_types::ir::schema::{DataType, Field, Schema};
-use asap_types::ir::schema::{FieldDataType, SketchStatistic};
-use asap_types::ir::OperatorNode;
-use asap_types::types::AccuracyTarget;
-use asap_types::workload::resources::CacheProfile;
-
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-#[serde(deny_unknown_fields)]
-struct PlannerCostDocument {
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    storage_io: Option<asap_plan_selection::cost::storage_io::StorageIoProfile>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    #[serde(rename = "boundaries")]
-    handoffs: Option<asap_plan_selection::cost::physical_handoff_cost::PhysicalHandoffProfile>,
-    /// Immutable catalog/runtime evidence generation shared by this file.
-    evidence_version: String,
-    calibration: ResourceCalibration,
-    targets: Vec<TargetPhysicalEvidence>,
-}
-
-fn parse_planner_cost_document(raw: &str) -> Result<PlannerCostDocument, String> {
-    let value: serde_json::Value = serde_json::from_str(raw)
-        .map_err(|error| format!("planner cost evidence is invalid JSON: {error}"))?;
-    if value.get("targets").is_none() {
-        return Err("the compact analytical-cost payload is no longer supported; migrate to complete per-operator physical DAG evidence".into());
-    }
-    let document: PlannerCostDocument = serde_json::from_value(value)
-        .map_err(|error| format!("planner cost evidence is invalid: {error}"))?;
-    if document.evidence_version.trim().is_empty() {
-        return Err("planner cost evidence_version must be non-empty".into());
-    }
-    Ok(document)
-}
-
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-#[serde(deny_unknown_fields)]
-struct TargetPhysicalEvidence {
-    target: Rc<OperatorNode>,
-    scope: ComparisonScopeEvidence,
-    candidates: Vec<CandidatePhysicalEvidence>,
-}
-
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-#[serde(deny_unknown_fields)]
-struct ComparisonScopeEvidence {
-    data_arrival: asap_types::workload::DataArrival,
-    planning_time_ms: u64,
-    horizon_ms: u64,
-    evaluation_count: u64,
-    time_scope: String,
-    lookback_ms: Option<u64>,
-    as_of_ms: Option<u64>,
-    sources: Vec<asap_plan_selection::cost::physical_operator_statistics::ScanSelection>,
-    #[serde(default = "CacheProfile::no_cache")]
-    cache_profile: CacheProfile,
-}
-
-impl ComparisonScopeEvidence {
-    fn resolve(&self) -> Result<ComparisonScope, AnalyticalCostError> {
-        use asap_types::workload::{
-            DurationMs, QueryRecurrence, QueryTimeScope, TimeSelection, TimestampMs,
-        };
-        let scope = match self.time_scope.as_str() {
-            "real_time" => QueryTimeScope::RealTime,
-            "longitudinal" => QueryTimeScope::Longitudinal,
-            "mixed" => QueryTimeScope::Mixed,
-            "unknown" => QueryTimeScope::Unknown,
-            _ => return Err(AnalyticalCostError::MissingComparisonScope("time_scope")),
-        };
-        let comparison = ComparisonScope {
-            data_arrival: self.data_arrival,
-            planning_time: TimestampMs(self.planning_time_ms),
-            horizon: DurationMs(self.horizon_ms),
-            recurrence: QueryRecurrence::OneTime {
-                invocations: self.evaluation_count,
-                execute_at: None,
-            },
-            time_selection: TimeSelection {
-                scope,
-                lookback: self.lookback_ms.map(DurationMs),
-                as_of: self.as_of_ms.map(TimestampMs),
-            },
-            sources: self.sources.clone(),
-        };
-        comparison.validate()?;
-        Ok(comparison)
-    }
-}
-
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-#[serde(deny_unknown_fields)]
-struct QueryNodePhysicalEvidence {
-    logical_node: OperatorNode,
-    operator: asap_plan_selection::cost::analytical_cost::PhysicalOperator,
-    occurrence: usize,
-    synthetic: bool,
-    evidence: PhysicalNodeEvidence,
-}
-
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
-enum CandidatePhysicalEvidence {
-    Rewrite {
-        plan: serde_json::Value,
-        query_nodes: Vec<QueryNodePhysicalEvidence>,
-    },
-    Summary {
-        plan: serde_json::Value,
-        query_nodes: Vec<QueryNodePhysicalEvidence>,
-        physical_dag: PhysicalDAG,
-    },
-}
-
-impl CandidatePhysicalEvidence {
-    fn plan(&self) -> &serde_json::Value {
-        match self {
-            Self::Rewrite { plan, .. } | Self::Summary { plan, .. } => plan,
-        }
-    }
-
-    fn query_nodes(&self) -> &[QueryNodePhysicalEvidence] {
-        match self {
-            Self::Rewrite { query_nodes, .. } | Self::Summary { query_nodes, .. } => query_nodes,
-        }
-    }
-
-    fn matches(&self, candidate: &ReplacementSubDAG) -> bool {
-        let actual = match (self, &candidate.replacement) {
-            (Self::Summary { .. }, Replacement::SubDAG(node)) if !is_logical_rewrite(node) => {
-                serde_json::to_value(dag_export::export(node))
-            }
-            (Self::Rewrite { .. }, Replacement::SubDAG(node)) if is_logical_rewrite(node) => {
-                serde_json::to_value(dag_export::export(node))
-            }
-            _ => return false,
-        };
-        actual.is_ok_and(|actual| plan_values_match(&actual, self.plan()))
-    }
-
-    fn summary_dag(&self) -> Option<&PhysicalDAG> {
-        match self {
-            Self::Summary { physical_dag, .. } => Some(physical_dag),
-            Self::Rewrite { .. } => None,
-        }
-    }
-}
-
-/// Compare complete exported-plan identity while tolerating the one-ULP
-/// decimal round trip that `serde_json::Value` can introduce for derived
-/// floating-point guarantees. Integer configuration and DAG identity stay
-/// exact; no guarantee field is dropped or otherwise normalized away.
-fn plan_values_match(actual: &serde_json::Value, expected: &serde_json::Value) -> bool {
-    plan_values_match_inner(actual, expected, false)
-}
-
-fn plan_values_match_inner(
-    actual: &serde_json::Value,
-    expected: &serde_json::Value,
-    inside_guarantee: bool,
-) -> bool {
-    use serde_json::Value;
-    match (actual, expected) {
-        (Value::Object(actual), Value::Object(expected)) => {
-            actual.len() == expected.len()
-                && actual.iter().all(|(key, value)| {
-                    expected.get(key).is_some_and(|other| {
-                        plan_values_match_inner(
-                            value,
-                            other,
-                            inside_guarantee || key == "guarantee",
-                        )
-                    })
-                })
-        }
-        (Value::Array(actual), Value::Array(expected)) => {
-            actual.len() == expected.len()
-                && actual
-                    .iter()
-                    .zip(expected)
-                    .all(|(value, other)| plan_values_match_inner(value, other, inside_guarantee))
-        }
-        (Value::Number(actual), Value::Number(expected)) => {
-            match (actual.as_i64(), expected.as_i64()) {
-                (Some(actual), Some(expected)) => actual == expected,
-                _ => match (actual.as_u64(), expected.as_u64()) {
-                    (Some(actual), Some(expected)) => actual == expected,
-                    _ => match (actual.as_f64(), expected.as_f64()) {
-                        (Some(actual), Some(expected)) => {
-                            actual == expected
-                                || (inside_guarantee
-                                    && actual.to_bits().abs_diff(expected.to_bits()) <= 1)
-                        }
-                        _ => false,
-                    },
-                },
-            }
-        }
-        _ => actual == expected,
-    }
-}
-
-struct ExportPhysicalProvider<'a> {
-    storage_io: Option<&'a asap_plan_selection::cost::storage_io::StorageIoProfile>,
-    handoffs: Option<&'a asap_plan_selection::cost::physical_handoff_cost::PhysicalHandoffProfile>,
-    evidence_version: &'a str,
-    target: &'a TargetPhysicalEvidence,
-    candidate: &'a CandidatePhysicalEvidence,
-    used_query_nodes: RefCell<std::collections::HashSet<usize>>,
-}
-
-impl ExportPhysicalProvider<'_> {
-    fn all_query_evidence_used(&self) -> bool {
-        self.used_query_nodes.borrow().len() == self.candidate.query_nodes().len()
-    }
-}
-
-impl PlannerPhysicalPlanProvider for ExportPhysicalProvider<'_> {
-    fn capture_evidence_snapshot(
-        &self,
-        _target: &asap_logical_optimizer::pass1::replacement::TargetSubDAG<'_>,
-    ) -> Result<PhysicalEvidenceSnapshot, AnalyticalCostError> {
-        Ok(PhysicalEvidenceSnapshot {
-            version: self.evidence_version.into(),
-            scope: self.target.scope.resolve()?,
-            cache_profile: self.target.scope.cache_profile.clone(),
-            storage_io: self.storage_io.cloned(),
-            handoffs: self.handoffs.cloned(),
-        })
-    }
-
-    fn query_node_evidence(
-        &self,
-        snapshot: &PhysicalEvidenceSnapshot,
-        request: PhysicalNodeRequest<'_>,
-    ) -> Result<PhysicalNodeEvidence, AnalyticalCostError> {
-        if snapshot.scope != self.target.scope.resolve()? {
-            return Err(AnalyticalCostError::ComparisonScopeMismatch(
-                "planner evidence snapshot",
-            ));
-        }
-        let mut matches = self
-            .candidate
-            .query_nodes()
-            .iter()
-            .enumerate()
-            .filter(|(_, entry)| {
-                entry.logical_node == *request.logical_node
-                    && entry.operator == request.operator
-                    && entry.occurrence == request.occurrence
-                    && entry.synthetic == request.synthetic
-            });
-        let (index, evidence) = matches.next().ok_or_else(|| {
-            AnalyticalCostError::MissingOperatorStatistics(format!(
-                "logical occurrence {}",
-                request.occurrence
-            ))
-        })?;
-        if matches.next().is_some() {
-            return Err(AnalyticalCostError::InvalidPhysicalDAG(
-                "duplicate query-node evidence key",
-            ));
-        }
-        self.used_query_nodes.borrow_mut().insert(index);
-        Ok(evidence.evidence.clone())
-    }
-
-    fn summary_physical_dag(
-        &self,
-        snapshot: &PhysicalEvidenceSnapshot,
-        _summary: &Rc<OperatorNode>,
-        _target: &asap_logical_optimizer::pass1::replacement::TargetSubDAG<'_>,
-    ) -> Result<PhysicalDAG, AnalyticalCostError> {
-        if snapshot.scope != self.target.scope.resolve()? {
-            return Err(AnalyticalCostError::ComparisonScopeMismatch(
-                "planner evidence snapshot",
-            ));
-        }
-        self.candidate
-            .summary_dag()
-            .cloned()
-            .ok_or(AnalyticalCostError::InvalidPhysicalDAG(
-                "summary candidate is missing its physical DAG",
-            ))
-    }
-}
-
-struct ExportPlannerCostModel<'a> {
-    document: &'a PlannerCostDocument,
-}
-
-impl ExportPlannerCostModel<'_> {
-    fn bound<'a>(
-        &'a self,
-        candidate: &ReplacementSubDAG,
-        target: &asap_logical_optimizer::pass1::replacement::TargetSubDAG<'_>,
-    ) -> Option<(ExportPhysicalProvider<'a>, &'a ResourceCalibration)> {
-        let mut targets = self
-            .document
-            .targets
-            .iter()
-            .filter(|entry| entry.target == *target.root);
-        let target_evidence = targets.next()?;
-        if targets.next().is_some() {
-            return None;
-        }
-        let mut candidates = target_evidence
-            .candidates
-            .iter()
-            .filter(|entry| entry.matches(candidate));
-        let candidate_evidence = candidates.next()?;
-        if candidates.next().is_some() {
-            return None;
-        }
-        Some((
-            ExportPhysicalProvider {
-                storage_io: self.document.storage_io.as_ref(),
-                handoffs: self.document.handoffs.as_ref(),
-                evidence_version: &self.document.evidence_version,
-                target: target_evidence,
-                candidate: candidate_evidence,
-                used_query_nodes: RefCell::new(std::collections::HashSet::new()),
-            },
-            &self.document.calibration,
-        ))
-    }
-
-    fn annotations(
-        &self,
-        candidate: &ReplacementSubDAG,
-        target: &Rc<OperatorNode>,
-    ) -> (CostAnnotation, CostAnnotation, CostAnnotation) {
-        let target = asap_logical_optimizer::pass1::replacement::TargetSubDAG::new(target);
-        let Some((provider, calibration)) = self.bound(candidate, &target) else {
-            return winner_cost_annotations();
-        };
-        let Ok(model) = PhysicalPlanCostModel::new(&provider, calibration.clone()) else {
-            return winner_cost_annotations();
-        };
-        let Ok(estimate) = model.estimate_candidate(candidate, &target) else {
-            return winner_cost_annotations();
-        };
-        if !provider.all_query_evidence_used() {
-            return winner_cost_annotations();
-        }
-        let mut version = format!("{}+{}", ANALYTICAL_COST_MODEL_VERSION, calibration.version);
-        if let Some((storage, _)) = &estimate.storage_io {
-            version.push_str(&format!(
-                "+{}+{}",
-                storage.model_version, storage.calibration_version
-            ));
-        }
-        if let Some((handoff, _)) = &estimate.handoffs {
-            version.push_str(&format!(
-                "+{}+{}",
-                handoff.model_version, handoff.calibration_version
-            ));
-        }
-        let scope = &provider.target.scope;
-        let Ok((result_hits, buffer_hits)) = cache_hit_ratios(
-            &scope.cache_profile,
-            scope.evaluation_count,
-            scope.data_arrival,
-        ) else {
-            return winner_cost_annotations();
-        };
-        let input = |name: &str, value: f64, unit: &str| CostInput {
-            name: name.into(),
-            value,
-            unit: Some(unit.into()),
-        };
-        let mut cache_inputs = vec![
-            input(
-                "evaluation_count",
-                scope.evaluation_count as f64,
-                "evaluations",
-            ),
-            input("result_cache_hit_ratio", result_hits, "ratio"),
-            input("buffer_cache_hit_ratio", buffer_hits, "ratio"),
-        ];
-        if let CacheProfile::Evidence(evidence) = &scope.cache_profile {
-            cache_inputs.extend([
-                input(
-                    "distinct_evaluations",
-                    evidence.distinct_evaluations as f64,
-                    "evaluations",
-                ),
-                input(
-                    "repeated_identical_evaluations",
-                    evidence.repeated_identical_evaluations as f64,
-                    "evaluations",
-                ),
-                input(
-                    "result_cache_working_set",
-                    evidence.result_cache.working_set_bytes as f64,
-                    "bytes",
-                ),
-                input(
-                    "result_cache_capacity",
-                    evidence.result_cache.capacity_bytes as f64,
-                    "bytes",
-                ),
-                input(
-                    "buffer_cache_working_set",
-                    evidence.buffer_cache.working_set_bytes as f64,
-                    "bytes",
-                ),
-                input(
-                    "buffer_cache_capacity",
-                    evidence.buffer_cache.capacity_bytes as f64,
-                    "bytes",
-                ),
-            ]);
-            if let Some(ratio) = evidence.result_invalidation_ratio {
-                cache_inputs.push(input("result_cache_invalidation_ratio", ratio, "ratio"));
-            }
-        }
-        let inputs = |resources: asap_plan_selection::cost::analytical_cost::ResourceEstimate| {
-            let mut inputs = vec![
-                CostInput {
-                    name: "estimated_cpu_ops".into(),
-                    value: resources.cpu_ops(),
-                    unit: Some("operations".into()),
-                },
-                CostInput {
-                    name: "estimated_peak_memory".into(),
-                    value: resources.peak_memory_bytes() as f64,
-                    unit: Some("bytes".into()),
-                },
-                CostInput {
-                    name: "estimated_scan".into(),
-                    value: resources.scan_bytes() as f64,
-                    unit: Some("bytes".into()),
-                },
-            ];
-            inputs.extend(cache_inputs.iter().cloned());
-            inputs
-        };
-        let storage_inputs = |storage: &asap_plan_selection::cost::storage_io::StorageEstimate| {
-            let mut terms: Vec<_> = storage
-                .total
-                .terms()
-                .into_iter()
-                .map(|(name, value)| CostInput {
-                    name: name.into(),
-                    value: value as f64,
-                    unit: Some("operations".into()),
-                })
-                .collect();
-            let mut ids: Vec<_> = storage.per_node.keys().collect();
-            ids.sort();
-            for id in ids {
-                terms.extend(
-                    storage.per_node[id]
-                        .terms()
-                        .into_iter()
-                        .map(|(name, value)| CostInput {
-                            name: format!("physical_node:{id}:{name}"),
-                            value: value as f64,
-                            unit: Some("operations".into()),
-                        }),
-                );
-            }
-            terms
-        };
-        let mut raw_inputs = inputs(estimate.resources.raw);
-        let mut candidate_inputs = inputs(estimate.resources.candidate);
-        if let Some((raw, candidate)) = &estimate.storage_io {
-            raw_inputs.extend(storage_inputs(raw));
-            candidate_inputs.extend(storage_inputs(candidate));
-        }
-        let handoff_inputs =
-            |estimate: &asap_plan_selection::cost::physical_handoff_cost::PhysicalHandoffEstimate| {
-                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_handoff),
-                ] {
-                    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.handoffs {
-            raw_inputs.extend(handoff_inputs(raw));
-            candidate_inputs.extend(handoff_inputs(candidate));
-        }
-        let baseline = CostAnnotation::modeled(
-            estimate.raw_cost.0,
-            CostUnit::CostUnits,
-            &version,
-            raw_inputs,
-        )
-        .with_evidence_version(&self.document.evidence_version)
-        .with_cache_profile(snapshot_cache_version(&provider));
-        let selected = CostAnnotation::modeled(
-            estimate.candidate_cost.0,
-            CostUnit::CostUnits,
-            &version,
-            candidate_inputs,
-        )
-        .with_baseline(BaselineRef::PreAsapRecomputation, estimate.raw_cost.0)
-        .with_evidence_version(&self.document.evidence_version)
-        .with_cache_profile(snapshot_cache_version(&provider));
-        let benefit = CostAnnotation {
-            value: selected.delta,
-            unit: CostUnit::CostUnits,
-            source: CostSource::Modeled,
-            baseline: Some(BaselineRef::PreAsapRecomputation),
-            delta: None,
-            benefit_ratio: selected.benefit_ratio,
-            model_version: Some(version),
-            evidence_version: Some(self.document.evidence_version.clone()),
-            cache_profile: Some(snapshot_cache_version(&provider).into()),
-            benchmark_id: None,
-            inputs: Vec::new(),
-        };
-        (baseline, selected, benefit)
-    }
-}
-
-fn snapshot_cache_version<'a>(provider: &'a ExportPhysicalProvider<'_>) -> &'a str {
-    provider.target.scope.cache_profile.version()
-}
-
-impl CostModel for ExportPlannerCostModel<'_> {
-    fn candidate_cost_covers_complete_plan(&self) -> bool {
-        true
-    }
-
-    fn candidate_cost(
-        &self,
-        candidate: &ReplacementSubDAG,
-        target: &asap_logical_optimizer::pass1::replacement::TargetSubDAG<'_>,
-    ) -> Option<Cost> {
-        let (provider, calibration) = self.bound(candidate, target)?;
-        let cost = PhysicalPlanCostModel::new(&provider, calibration.clone())
-            .ok()?
-            .candidate_cost(candidate, target)?;
-        provider.all_query_evidence_used().then_some(cost)
-    }
-
-    fn rank_candidates(
-        &self,
-        _intent: &asap_types::ir::operator::AggIntent,
-        candidates: &[asap_types::ir::schema::SketchAlgorithm],
-    ) -> Vec<asap_types::ir::schema::SketchAlgorithm> {
-        // Candidate generation must not reintroduce the legacy structural
-        // cost model before complete physical alternatives are compared.
-        candidates.to_vec()
-    }
-
-    fn estimate_cost(
-        &self,
-        candidate: &ReplacementSubDAG,
-        target: &asap_logical_optimizer::pass1::replacement::TargetSubDAG<'_>,
-    ) -> f64 {
-        self.candidate_cost(candidate, target)
-            .map_or(f64::NAN, |cost| cost.0)
-    }
-}
-
-/// Baseline/selected/benefit [`CostAnnotation`]s for one [`Winner`] — issue
-/// #286's "replacement-region baseline cost, selected cost, and benefit"
-/// granularity item, reused verbatim for [`TargetReplacement`] and for the
-/// [`DAGDecision`] carried by every node the winning candidate produced or
-/// carried.
-///
-/// `dag_export` has no deployment-owned physical evidence provider. It must
-/// therefore expose costs as unavailable instead of guessing operator
-/// statistics or falling back to structural node counts. Callers that have
-/// complete evidence use `PhysicalPlanCostModel` before export and may
-/// attach its dimensional comparison to these fields.
-#[allow(dead_code)]
-fn winner_cost_annotations() -> (CostAnnotation, CostAnnotation, CostAnnotation) {
-    (
-        CostAnnotation::unavailable(CostUnit::CostUnits),
-        CostAnnotation::unavailable(CostUnit::CostUnits),
-        CostAnnotation::unavailable(CostUnit::CostUnits),
-    )
-}
-
-use asap_devtools::{lower_promql_with_data_ingestion_interval, lower_sql, SqlCatalog};
-
-enum Lang {
-    Sql,
-    PromQl,
-}
-
-fn default_catalog() -> SqlCatalog {
-    SqlCatalog::new()
-        .with_table(
-            "metrics",
-            Schema::with_time_index(
-                vec![
-                    Field::plain("ts", DataType::Timestamp, false),
-                    Field::plain("service", DataType::Utf8, false),
-                    Field::plain("region", DataType::Utf8, false),
-                    Field::plain("latency", DataType::Float64, false),
-                    Field::plain("bytes", DataType::Int64, false),
-                ],
-                0,
-                vec![],
-            ),
-        )
-        .with_table(
-            "hosts",
-            Schema::new(vec![
-                Field::plain("service", DataType::Utf8, false),
-                Field::plain("region", DataType::Utf8, false),
-            ]),
-        )
-}
-
-fn catalog(custom: &[String]) -> SqlCatalog {
-    let mut catalog = default_catalog();
-    for raw in custom {
-        let value: serde_json::Value = serde_json::from_str(raw)
-            .unwrap_or_else(|error| panic!("--table-schema must be valid JSON: {error}"));
-        let name = value["name"]
-            .as_str()
-            .expect("--table-schema.name must be a string");
-        let columns = value["columns"]
-            .as_array()
-            .expect("--table-schema.columns must be an array");
-        let columns: Vec<Field> = columns
-            .iter()
-            .map(|column| {
-                let column_name = column["name"]
-                    .as_str()
-                    .expect("column.name must be a string");
-                let data_type = match column["type"]
-                    .as_str()
-                    .expect("column.type must be a string")
-                    .to_ascii_lowercase()
-                    .as_str()
-                {
-                    "timestamp" => DataType::Timestamp,
-                    "utf8" | "string" => DataType::Utf8,
-                    "float64" | "double" => DataType::Float64,
-                    "int64" | "bigint" => DataType::Int64,
-                    other => panic!("unsupported column type {other:?}"),
-                };
-                Field::plain(
-                    column_name,
-                    data_type,
-                    column["nullable"].as_bool().unwrap_or(true),
-                )
-            })
-            .collect();
-        let schema = match value.get("time_index").and_then(|index| index.as_u64()) {
-            Some(index) => Schema::with_time_index(columns, index as usize, vec![]),
-            None => Schema::new(columns),
-        };
-        catalog = catalog.with_table(name, schema);
-    }
-    catalog
-}
-
-/// Parses `--sql "<query>" --name "<label>"` / `--promql "<query>" --name
-/// "<label>"` pairs off argv, in the order given, plus one optional global
-/// `--epsilon <f64>` and one optional global `--post-asap` flag. `--name` is
-/// optional and applies to the immediately preceding `--sql`/`--promql`.
-struct ParsedArgs {
-    entries: Vec<(String, Lang, String)>,
-    accuracy: AccuracyTarget,
-    post_asap: bool,
-    progress: bool,
-    table_schemas: Vec<String>,
-    planner_cost: Option<PlannerCostDocument>,
-    /// `--default-cost`: rank the `--post-asap` search with
-    /// [`DefaultCostModel`] and export no cost at all. Mutually exclusive
-    /// with `planner_cost`, which both ranks *and* is exported — see this
-    /// file's top-of-file usage doc for why the two can't be combined.
-    default_cost: bool,
-    topk_margin: Option<TopKMarginEvidence>,
-    promql_ingestion_interval_ms: Option<u64>,
-}
-
-#[derive(Debug, Clone, serde::Deserialize)]
-struct TopKMarginEvidence {
-    selected_lower_bound: f64,
-    excluded_upper_bound: f64,
-    interval_failure_probability: f64,
-}
-
-impl TopKMarginEvidence {
-    fn validate(&self) -> Result<(), &'static str> {
-        if !self.selected_lower_bound.is_finite() || !self.excluded_upper_bound.is_finite() {
-            return Err("Top-K bounds must be finite");
-        }
-        if self.selected_lower_bound <= self.excluded_upper_bound {
-            return Err("selected_lower_bound must exceed excluded_upper_bound");
-        }
-        if !(self.interval_failure_probability.is_finite()
-            && self.interval_failure_probability >= 0.0
-            && self.interval_failure_probability < 1.0)
-        {
-            return Err("interval_failure_probability must be in [0, 1)");
-        }
-        Ok(())
-    }
-}
-
-impl AccuracyEvidenceProvider for TopKMarginEvidence {
-    fn propagation_stats(
-        &self,
-        op: &CompositionOperator,
-        _family: &FieldDataType,
-        _query: Option<&SketchStatistic>,
-    ) -> PropagationStats {
-        if matches!(op, CompositionOperator::TopKSelection) {
-            PropagationStats {
-                topk_selected_lower_bound: Some(self.selected_lower_bound),
-                topk_excluded_upper_bound: Some(self.excluded_upper_bound),
-                topk_interval_failure_probability: Some(self.interval_failure_probability),
-                ..Default::default()
-            }
-        } else {
-            PropagationStats::default()
-        }
-    }
-}
-
-fn parse_args() -> ParsedArgs {
-    parse_args_from(std::env::args().skip(1))
-}
-
-fn parse_args_from(argv: impl Iterator<Item = String>) -> ParsedArgs {
-    let mut entries: Vec<(String, Lang, String)> = Vec::new();
-    let mut pending: Option<(Lang, String)> = None;
-    let mut accuracy = AccuracyTarget::Exact;
-    let mut post_asap = false;
-    let mut progress = false;
-    let mut table_schemas = Vec::new();
-    let mut planner_cost_json = None;
-    let mut default_cost = false;
-    let mut topk_margin_json = None;
-    let mut promql_ingestion_interval_ms = None;
-    let mut args = argv;
-
-    fn flush(entries: &mut Vec<(String, Lang, String)>, pending: &mut Option<(Lang, String)>) {
-        if let Some((lang, query)) = pending.take() {
-            entries.push((format!("q{}", entries.len() + 1), lang, query));
-        }
-    }
-
-    while let Some(arg) = args.next() {
-        match arg.as_str() {
-            "--sql" | "--promql" => {
-                flush(&mut entries, &mut pending);
-                let query = args
-                    .next()
-                    .unwrap_or_else(|| panic!("{arg} requires a query string"));
-                let lang = if arg == "--sql" {
-                    Lang::Sql
-                } else {
-                    Lang::PromQl
-                };
-                pending = Some((lang, query));
-            }
-            "--name" => {
-                let name = args.next().expect("--name requires a value");
-                let (lang, query) = pending.take().expect("--name must follow --sql/--promql");
-                entries.push((name, lang, query));
-            }
-            "--epsilon" => {
-                let raw = args.next().expect("--epsilon requires a value");
-                let epsilon: f64 = raw
-                    .parse()
-                    .unwrap_or_else(|_| panic!("--epsilon must be a float, got {raw:?}"));
-                accuracy = AccuracyTarget::Epsilon(epsilon);
-            }
-            "--post-asap" => {
-                post_asap = true;
-            }
-            "--progress" => {
-                progress = true;
-            }
-            "--table-schema" => {
-                table_schemas.push(args.next().expect("--table-schema requires JSON"));
-            }
-            "--planner-cost-json" | "--analytical-cost-json" => {
-                planner_cost_json = Some(
-                    args.next()
-                        .expect("planner cost evidence requires a JSON document"),
-                );
-            }
-            "--default-cost" => {
-                default_cost = true;
-            }
-            "--topk-margin-json" => {
-                topk_margin_json = Some(
-                    args.next()
-                        .expect("--topk-margin-json requires a JSON object"),
-                );
-            }
-            "--data-ingestion-interval-ms" => {
-                promql_ingestion_interval_ms = Some(
-                    args.next()
-                        .expect("--data-ingestion-interval-ms requires a value")
-                        .parse()
-                        .expect("--data-ingestion-interval-ms must be an unsigned integer"),
-                );
-            }
-            other => panic!("unrecognized argument: {other}"),
-        }
-    }
-    flush(&mut entries, &mut pending);
-    // Rejected rather than given a precedence order: the two flags disagree
-    // about what the export may claim, not just about which model ranks, so
-    // silently preferring one would make the exported costs depend on an
-    // argument order the caller never stated.
-    if default_cost && planner_cost_json.is_some() {
-        panic!(
-            "--default-cost and --planner-cost-json are mutually exclusive: --default-cost ranks \
-             with structural node counts and exports no cost, --planner-cost-json exports \
-             deployment-owned costs"
-        );
-    }
-    let planner_cost = planner_cost_json
-        .map(|raw| parse_planner_cost_document(&raw).unwrap_or_else(|error| panic!("{error}")));
-    let topk_margin = topk_margin_json.map(|raw| {
-        let evidence: TopKMarginEvidence = serde_json::from_str(&raw)
-            .unwrap_or_else(|error| panic!("--topk-margin-json is invalid: {error}"));
-        evidence
-            .validate()
-            .unwrap_or_else(|error| panic!("invalid Top-K margin evidence: {error}"));
-        evidence
-    });
-    ParsedArgs {
-        entries,
-        accuracy,
-        post_asap,
-        progress,
-        table_schemas,
-        planner_cost,
-        default_cost,
-        topk_margin,
-        promql_ingestion_interval_ms,
-    }
-}
-
-/// Attach workload-wide replacement explanations to their exact dag nodes.
-/// `node_hash` is only a narrowing filter; `source_node == Some(target)` is
-/// the collision-safe identity check (every node an ordinary
-/// [`dag_export::export`] produces carries `Some`; the `None` arm is
-/// defensive only).
-fn annotate_with_explanations(
-    dag: &mut ExportDAG,
-    explanations: &[asap_logical_optimizer::ReplacementExplanation],
-    matched: &mut [bool],
-) {
-    for (i, explanation) in explanations.iter().enumerate() {
-        for node in dag.nodes.iter_mut() {
-            if node.hash == Some(explanation.node_hash)
-                && node.source_node.as_ref() == Some(&explanation.target)
-            {
-                node.notes.push(DAGNote {
-                    kind: format!("{:?}", explanation.kind),
-                    reason: explanation.reason.clone(),
-                });
-                matched[i] = true;
-            }
-        }
-    }
-}
-
-/// One `TargetSubDAGCandidates`'s best-ranked candidate, kept alongside its own `target`
-/// — the unit both [`PostAsapResults::replacements`] and
-/// [`PostAsapResults::post_dags`] are built from, so the two outputs can
-/// never disagree about which candidate won for a given target.
-#[allow(dead_code)]
-struct Winner<'a> {
-    target: &'a Rc<OperatorNode>,
-    candidate: &'a ReplacementSubDAG,
-    costs: (CostAnnotation, CostAnnotation, CostAnnotation),
-}
-
-/// Short explanation intended for a selected winner in node-level UI. The
-/// candidate's full rationale remains available in pre-ASAP applicability
-/// notes; repeating that exhaustive prose on every post-ASAP region node
-/// obscures the actual decision.
-#[allow(dead_code)]
-fn decision_rationale(winner: &Winner<'_>) -> String {
-    match winner.candidate.strategy {
-        "AvgToSumOverCountStrategy" => {
-            "Rewrites AVG into SUM and COUNT under the same grouping, then divides SUM by COUNT."
-                .to_string()
-        }
-        "SemanticEquivalentRewriteStrategy" => {
-            "Composes compatible nested aggregates using their declared algebraic intent while preserving the output schema."
-                .to_string()
-        }
-        "SharedSubDAGStrategy" => match winner.candidate.provenance {
-            asap_logical_optimizer::pass1::replacement::ReplacementProvenance::CseShare => {
-                "Builds the repeated sub-DAG once and shares it across consumers.".to_string()
-            }
-            asap_logical_optimizer::pass1::replacement::ReplacementProvenance::CseRecompute => {
-                "Recomputes the sub-DAG per consumer because that has the lower estimated cost."
-                    .to_string()
-            }
-            _ => "Chooses the lowest-cost handling of the repeated sub-DAG.".to_string(),
-        },
-        "RollupStrategy" => {
-            "Answers this aggregate from a compatible finer-grained aggregate.".to_string()
-        }
-        "TopKLimitReuseStrategy" => {
-            "Derives this smaller top-k from a compatible larger top-k result shared by the workload."
-                .to_string()
-        }
-        _ => {
-            let summary = winner
-                .candidate
-                .rationale
-                .split_once(" — ")
-                .map_or(winner.candidate.rationale.as_str(), |(summary, _)| summary);
-            summary
-                .split_once(" (asap_")
-                .map_or(summary, |(plain, _)| plain)
-                .to_string()
-        }
-    }
-}
-
-/// Find the index into `winners` (if any) whose `target` is structurally
-/// identical to `expr` — `by_hash` is only a narrowing filter (keyed by
-/// [`structural_hash`]); `expr == target` is the collision-safe identity
-/// check, the same two-step discipline [`annotate_with_explanations`] and
-/// this file's original single-target matching already use. Returns an
-/// index rather than a `&Winner` directly so a caller can both use the
-/// match and record it (e.g. `matched[i] = true`) without juggling a second
-/// way to name the same winner.
-#[allow(dead_code)]
-fn lookup_winner(
-    by_hash: &HashMap<u64, Vec<usize>>,
-    winners: &[Winner<'_>],
-    cache: &mut HashCache,
-    expr: &OperatorNode,
-) -> Option<usize> {
-    let hash = structural_hash(expr, cache);
-    by_hash
-        .get(&hash)?
-        .iter()
-        .copied()
-        .find(|&i| expr == winners[i].target.as_ref())
-}
-
-/// One `(decision.id, baseline_cost, selected_cost)` triple per *distinct*
-/// [`DAGDecision`] carried anywhere in `dag` — collapsing every node that
-/// shares one `decision.id` (a replacement region can span many nodes, all
-/// carrying an identical clone of the same decision) down to a single
-/// entry, so a caller summing these never counts one decision's cost once
-/// per node it happens to touch.
-fn decision_cost_entries(dag: &ExportDAG) -> Vec<(u32, CostAnnotation, CostAnnotation)> {
-    let mut seen = std::collections::HashSet::new();
-    let mut entries = Vec::new();
-    for node in &dag.nodes {
-        let Some(decision) = &node.decision else {
-            continue;
-        };
-        if !seen.insert(decision.id) {
-            continue;
-        }
-        let (Some(baseline), Some(selected)) = (&decision.baseline_cost, &decision.selected_cost)
-        else {
-            continue;
-        };
-        entries.push((decision.id, baseline.clone(), selected.clone()));
-    }
-    entries
-}
-
-/// Build a [`TargetReplacement`] for `winner`, matching this file's own
-/// per-target `before`/`after` construction.
-#[allow(dead_code)]
-fn target_replacement(
-    decision_id: u32,
-    target_pre_id: u32,
-    winner: &Winner<'_>,
-) -> TargetReplacement {
-    let strategy = winner.candidate.strategy.to_string();
-    let before = dag_export::export(winner.target);
-    let after = match &winner.candidate.replacement {
-        Replacement::SubDAG(node) if is_logical_rewrite(node) => {
-            TargetReplacementAfter::Rewrite(dag_export::export(node))
-        }
-        Replacement::SubDAG(node) => TargetReplacementAfter::Summary(dag_export::export(node)),
-        Replacement::ExactComposition(_) => {
-            unreachable!("composition candidates are materialized by GlobalSelection")
-        }
-    };
-    let (baseline_cost, selected_cost, benefit) = winner.costs.clone();
-    // Derived from `selected_cost` so the legacy scalar field and the
-    // structured annotation can never drift apart.
-    let cost = selected_cost.value.unwrap_or(f64::NAN);
-    TargetReplacement {
-        decision_id,
-        target_pre_id,
-        strategy,
-        rationale: decision_rationale(winner),
-        rank: 0,
-        cost,
-        before,
-        after,
-        baseline_cost: Some(baseline_cost),
-        selected_cost: Some(selected_cost),
-        benefit: Some(benefit),
-    }
-}
-
-/// Is `replacement` `retain_exact`'s conservative no-op fallback — the
-/// target itself, unbound, carrying only an exact "kept pre-ASAP" guarantee?
-/// `ASAPStrategies` emits it for an intent with no summary
-/// realization at all (`STDDEV_POP`, `AVG`, ... dispatch to
-/// `Realization::PassThrough`). It is "nothing to bind here", not a
-/// replacement decision. A logical rewrite (no guarantee yet) and any sub-DAG
-/// with an ASAP operator are real candidates.
-fn is_trivial_retain_exact(replacement: &Replacement) -> bool {
-    matches!(
-        replacement,
-        Replacement::SubDAG(node) if node.guarantee.is_some() && !node.contains_asap()
-    )
-}
-
-/// The two additive `--post-asap` outputs — see this file's top-of-file
-/// usage doc for what each is for.
-struct PostAsapResults {
-    /// One `(query_name, TargetReplacement)` pair per discovered replacement
-    /// site whose target node is found in that query's own exported DAG. A
-    /// target can in principle be reachable from more than one query's root
-    /// after CSE (a shared sub-DAG), in which case it yields one pair per
-    /// matching query, each with that query's own `target_pre_id`.
-    replacements: Vec<(String, TargetReplacement)>,
-    /// One merged, whole-query [`ExportDAG`] per query, built via
-    /// [`dag_export::export_post_asap`] — every winning candidate spliced
-    /// directly into that query's own pre-ASAP shape in place.
-    post_dags: Vec<(String, ExportDAG)>,
-    /// One `(query_name, TargetRejection)` per accuracy-illegal candidate
-    /// the search refused (`TargetSubDAGCandidates::rejected`, issue #172) whose target
-    /// node is found in that query's own exported DAG.
-    rejections: Vec<(String, TargetRejection)>,
-}
-
-fn raw_only_post_asap_results() -> PostAsapResults {
-    PostAsapResults {
-        replacements: Vec::new(),
-        post_dags: Vec::new(),
-        rejections: Vec::new(),
-    }
-}
-
-/// Assign collision-free, explicit identities to structurally equal nodes
-/// across a set of exported query graphs. The full canonical sub-DAG string
-/// is the equality key; the compact integer is what JSON consumers receive.
-/// Consequently the viewer never needs to guess identity from labels,
-/// hashes, or a client-side node signature.
-fn assign_workload_node_ids(dags: &mut [&mut ExportDAG]) {
-    fn key_for(id: u32, dag: &ExportDAG, memo: &mut HashMap<u32, String>) -> String {
-        if let Some(key) = memo.get(&id) {
-            return key.clone();
-        }
-        let node = &dag.nodes[id as usize];
-        let child_keys: Vec<_> = node
-            .children
-            .iter()
-            .map(|child| key_for(*child, dag, memo))
-            .collect();
-        let key = serde_json::to_string(&(node.kind, &node.detail, &node.schema, child_keys))
-            .expect("exported DAG node content is serializable");
-        memo.insert(id, key.clone());
-        key
-    }
-
-    let mut ids = HashMap::<String, u32>::new();
-    let mut next_id = 0_u32;
-    for dag in dags.iter_mut() {
-        let mut memo = HashMap::new();
-        let keys: Vec<_> = dag
-            .nodes
-            .iter()
-            .map(|node| key_for(node.id, dag, &mut memo))
-            .collect();
-        for (node, key) in dag.nodes.iter_mut().zip(keys) {
-            let id = *ids.entry(key).or_insert_with(|| {
-                let id = next_id;
-                next_id += 1;
-                id
-            });
-            node.workload_node_id = Some(id);
-        }
-    }
-}
-
-/// Run `asap_logical_optimizer::pass1::replacement::search_workload` (its own
-/// `default_strategies()` — which includes `AvgToSumOverCountStrategy` as of
-/// #282 — is exactly the strategy set this binary wants; no custom list
-/// needed) over every lowered query, rank each discovered `TargetSubDAGCandidates` via
-/// `candidate_selection::global_selection`, and build both `--post-asap` outputs from the
-/// exact same set of winning candidates (see [`Winner`]), so the flat
-/// `replacements` list and the merged `post_dag` can never disagree about
-/// which candidate won for a given target.
-#[allow(dead_code)]
-fn run_post_asap_with_progress(
-    lowered_queries: &[(String, String, Rc<OperatorNode>)],
-    progress: bool,
-    cost_model: &dyn CostModel,
-    export_model: Option<&ExportPlannerCostModel<'_>>,
-    evidence: Option<&dyn AccuracyEvidenceProvider>,
-) -> PostAsapResults {
-    let mapping_started = Instant::now();
-    if progress {
-        eprintln!("[3/4] ASAP-aware mapping is running…");
-    }
-    let roots: Vec<(String, Rc<OperatorNode>)> = lowered_queries
-        .iter()
-        .map(|(name, _, qe)| (name.clone(), Rc::clone(qe)))
-        .collect();
-    let strategies;
-    let space = if let Some(evidence) = evidence {
-        strategies = default_strategies_with_evidence(evidence);
-        search_workload_with(roots, &strategies)
-    } else {
-        search_workload(roots)
-    };
-    let selection = global_selection(&space, cost_model);
-
-    // A group's top candidate can be `retain_exact`'s own conservative
-    // fallback — the *whole target* itself, unbound, carrying only an exact
-    // "kept pre-ASAP" guarantee (see `is_trivial_retain_exact`) — e.g. for
-    // a multi-measure/`HAVING`-bearing aggregate, or (the case that actually
-    // surfaces this: `STDDEV_POP`/`AVG`/`VARIANCE` dispatch to
-    // `Realization::PassThrough` with no alternative at all, per
-    // `realizations_for_intent`'s own doc) an intent with no summary
-    // realization whatsoever. This isn't a replacement decision — it's
-    // `ASAPStrategies` saying "nothing to bind here" — the
-    // identical "no-op candidate" concept `explanation.rs`'s own
-    // `sketch_finding_reason` already excludes from being reported as a
-    // finding ("a candidate list containing only the trivial no-op
-    // realization... isn't an opportunity, it's just the target's existing
-    // shape reflected back"). Treating this candidate as "no winner" (same
-    // as an empty candidate list) also keeps `post_dag` honest: splicing
-    // the target in for itself would tag every node of an unchanged sub-DAG
-    // with a "replacement" decision.
-    let winners: Vec<Winner<'_>> = selection
-        .target_selections()
-        .filter_map(|group| {
-            let candidate = group.chosen?;
-            // A composition is a reference-bearing logical choice, not an
-            // independently exportable replacement. The CLI's default cost
-            // model has no composition statistics and therefore cannot
-            // select one; callers that provide such statistics must export
-            // `GlobalSelection::assemble_selected_dag` instead.
-            if matches!(candidate.replacement, Replacement::ExactComposition(_)) {
-                return None;
-            }
-            if is_trivial_retain_exact(&candidate.replacement) {
-                return None;
-            }
-            Some(Winner {
-                target: group.target,
-                candidate,
-                costs: export_model.map_or_else(winner_cost_annotations, |model| {
-                    model.annotations(candidate, group.target)
-                }),
-            })
-        })
-        .collect();
-
-    // `by_hash` only narrows the search; `lookup_winner`'s own structural
-    // equality check is the real decision — see that function's doc.
-    let mut by_hash_cache = HashCache::new();
-    let mut by_hash: HashMap<u64, Vec<usize>> = HashMap::new();
-    for (i, winner) in winners.iter().enumerate() {
-        let hash = structural_hash(winner.target, &mut by_hash_cache);
-        by_hash.entry(hash).or_default().push(i);
-    }
-    if progress {
-        eprintln!(
-            "[3/4] ASAP-aware mapping done in {:.2} ms",
-            mapping_started.elapsed().as_secs_f64() * 1000.0
-        );
-    }
-
-    // ---- Merged, whole-query `post_dag`, one per query ---------------
-    //
-    // Built *before* the flat `replacements` pass below, not after: a
-    // winner whose target only exists inside another winner's own
-    // `Replacement::Rewrite` output (e.g. the `sum`/`count` aggregates
-    // `AvgToSumOverCountStrategy`'s rewrite exposes, which
-    // `default_strategies()` — #282 — now discovers and independently
-    // sketch-ranks in the same search pass) can never appear in any query's
-    // *original*, pre-rewrite `dag` — there's nothing wrong with that
-    // winner, it's just nested. `post_dag` is where it's expected to
-    // surface instead (`export_post_asap`'s recursive `find_winner`
-    // threading walks straight through a rewritten sub-DAG and re-checks
-    // every node inside it too), so the flat-`replacements` pass below
-    // checks there before deciding a miss is a real anomaly worth a
-    // warning.
-    if progress {
-        eprintln!("[4/4] Post-ASAP DAG generation is running…");
-    }
-    let post_started = Instant::now();
-    let mut post_dag_cache = HashCache::new();
-    let mut find_winner = |expr: &Rc<OperatorNode>| -> Option<PostAsapSubstitution> {
-        let i = lookup_winner(&by_hash, &winners, &mut post_dag_cache, expr)?;
-        let winner = &winners[i];
-        let (baseline_cost, selected_cost, benefit) = winner.costs.clone();
-        // Derived from `selected_cost`; see `target_replacement`'s identical
-        // derivation.
-        let cost = selected_cost.value.unwrap_or(f64::NAN);
-        let decision = DAGDecision {
-            id: i as u32,
-            strategy: winner.candidate.strategy.to_string(),
-            rationale: decision_rationale(winner),
-            rank: 0,
-            cost,
-            role: "replacement_region",
-            baseline_cost: Some(baseline_cost),
-            selected_cost: Some(selected_cost),
-            benefit: Some(benefit),
-        };
-        Some(match &winners[i].candidate.replacement {
-            Replacement::SubDAG(rc) if is_logical_rewrite(rc) => PostAsapSubstitution::Rewrite {
-                replacement: Rc::clone(rc),
-                decision,
-            },
-            Replacement::SubDAG(rc) => PostAsapSubstitution::Summary {
-                replacement: Rc::clone(rc),
-                decision,
-            },
-            Replacement::ExactComposition(_) => {
-                unreachable!("composition winners are excluded above")
-            }
-        })
-    };
-    let post_dags: Vec<(String, ExportDAG)> = lowered_queries
-        .iter()
-        .map(|(name, _, qe)| {
-            (
-                name.clone(),
-                dag_export::export_post_asap(qe, &mut find_winner),
-            )
-        })
-        .collect();
-
-    // ---- Flat per-target `replacements`, one list per query -----------
-    //
-    // Independently re-export every query's own DAG for matching — a
-    // fresh `export` per query, not reused from `main`'s own already-built
-    // `NamedDAG`s, so this function stays self-contained and callable on
-    // its own (see this file's `#[cfg(test)]` module). Deliberately anchored
-    // to the *original* `dag` only (never `post_dag`) — `target_pre_id`
-    // is documented as an id into `NamedDAG.DAG.nodes`, so a nested
-    // secondary target (see above) never gets a flat entry of its own here:
-    // it's already visible, in place, inside its parent's own `after`
-    // sub-DAG and inside `post_dag` as a whole.
-    let mut lookup_cache = HashCache::new();
-    let mut replacements = Vec::new();
-    let mut rejections = Vec::new();
-    let mut matched = vec![false; winners.len()];
-    // Groups with accuracy-refused candidates (issue #172): matched to a
-    // query's DAG nodes the same hash-then-structural-equality way.
-    let rejected_groups: Vec<_> = space
-        .target_subdag_candidates()
-        .filter(|group| !group.rejected.is_empty())
-        .collect();
-    let mut rejected_by_hash: HashMap<u64, Vec<usize>> = HashMap::new();
-    for (i, group) in rejected_groups.iter().enumerate() {
-        let hash = structural_hash(&group.target, &mut by_hash_cache);
-        rejected_by_hash.entry(hash).or_default().push(i);
-    }
-    for (name, _, qe) in lowered_queries {
-        let dag = dag_export::export(qe);
-        for node in &dag.nodes {
-            let Some(source_node) = node.source_node.as_ref() else {
-                continue; // never true for a plain `export` — defensive only.
-            };
-            if let Some(i) = lookup_winner(&by_hash, &winners, &mut lookup_cache, source_node) {
-                replacements.push((
-                    name.clone(),
-                    target_replacement(i as u32, node.id, &winners[i]),
-                ));
-                matched[i] = true;
-            }
-            let hash = structural_hash(source_node, &mut lookup_cache);
-            for &i in rejected_by_hash.get(&hash).into_iter().flatten() {
-                let group = rejected_groups[i];
-                if *source_node != group.target {
-                    continue;
-                }
-                rejections.extend(group.rejected.iter().map(|rejected| {
-                    (
-                        name.clone(),
-                        TargetRejection {
-                            target_pre_id: node.id,
-                            strategy: rejected.strategy.to_string(),
-                            description: rejected.description.clone(),
-                            error: rejected.error.clone(),
-                        },
-                    )
-                }));
-            }
-        }
-    }
-
-    // A winner can legitimately stay unmatched here: `default_strategies()`
-    // (#282) means `AvgToSumOverCountStrategy`'s rewrite output (and
-    // similarly `RollupStrategy`'s) can expose a brand-new `sum`/`count`
-    // descendant that the *same* search pass then independently discovers
-    // and ranks — a real winner, but one with no node anywhere in any
-    // query's original, pre-rewrite `dag` to attach a flat entry to
-    // (`target_pre_id` is documented as an id into `DAG.nodes`
-    // specifically). This isn't a data loss: `export_post_asap` still
-    // splices that winner in, in place, inside `post_dag` — see this
-    // function's own construction of `post_dags` above, which walks
-    // straight through a rewritten sub-DAG and resolves every nested
-    // winner too, recursively. So an unmatched winner here is expected,
-    // not necessarily a bug, whenever it's downstream of some other
-    // winner's own `Replacement::Rewrite` — logged as an FYI rather than a
-    // warning, since telling the two cases apart precisely would mean
-    // reimplementing `search`'s own private descendant-discovery walk
-    // (`discover_new_descendant_targets` in `asap_logical_optimizer::pass1::replacement`,
-    // not exposed) a second time here just to double-check something
-    // `post_dag`'s own construction already handled correctly.
-    for (winner, matched) in winners.iter().zip(&matched) {
-        if !matched {
-            let strategy = winner.candidate.strategy;
-            eprintln!(
-                "dag_export: post-asap replacement ({strategy}) has no node in any query's \
-                 original DAG — expected for a winner exposed only inside another winner's \
-                 own rewrite output (e.g. a sum/count descendant of an avg rewrite); still \
-                 present in that query's post_dag"
-            );
-        }
-    }
-    if progress {
-        eprintln!(
-            "[4/4] Post-ASAP DAG generation done in {:.2} ms",
-            post_started.elapsed().as_secs_f64() * 1000.0
-        );
-    }
-
-    PostAsapResults {
-        replacements,
-        post_dags,
-        rejections,
-    }
-}
-
-#[cfg(test)]
-fn run_post_asap(lowered_queries: &[(String, String, Rc<OperatorNode>)]) -> PostAsapResults {
-    run_post_asap_with_progress(lowered_queries, false, &DefaultCostModel, None, None)
-}
-
-#[tokio::main]
-async fn main() {
-    let ParsedArgs {
-        entries,
-        accuracy,
-        post_asap,
-        progress,
-        table_schemas,
-        planner_cost,
-        default_cost,
-        topk_margin,
-        promql_ingestion_interval_ms,
-    } = parse_args();
-    let sql_catalog = catalog(&table_schemas);
-    let planner_started = Instant::now();
-    if entries.is_empty() {
-        eprintln!(
-            "usage: dag_export --sql \"<query>\" [--name <label>] [--epsilon <f64>] [--post-asap [--default-cost | --planner-cost-json <doc>]] ..."
-        );
-        std::process::exit(1);
-    }
-
-    let mut lowered_queries = Vec::new();
-    let lowering_started = Instant::now();
-    if progress {
-        eprintln!("[1/4] Parsing and lowering SQL/PromQL queries…");
-    }
-    for (name, lang, query) in entries {
-        let lowered = match lang {
-            Lang::Sql => lower_sql(&query, &sql_catalog, accuracy.clone())
-                .await
-                .map_err(|e| e.to_string()),
-            Lang::PromQl => lower_promql_with_data_ingestion_interval(
-                &query,
-                accuracy.clone(),
-                promql_ingestion_interval_ms
-                    .expect("--data-ingestion-interval-ms is required for PromQL queries"),
-            )
-            .map_err(|e| e.to_string()),
-        };
-        match lowered {
-            Ok(qe) => {
-                lowered_queries.push((name, query, qe));
-            }
-            Err(e) => eprintln!("skipping {name:?} — lowering failed: {e}"),
-        }
-    }
-    if progress {
-        eprintln!(
-            "[1/4] Parsing and lowering done in {:.2} ms",
-            lowering_started.elapsed().as_secs_f64() * 1000.0
-        );
-    }
-
-    let pre_started = Instant::now();
-    if progress {
-        eprintln!("[2/4] Pre-ASAP DAG generation is running…");
-    }
-    let explanations = asap_logical_optimizer::explain_replacements(
-        lowered_queries
-            .iter()
-            .map(|(name, _, qe)| (name.clone(), qe.clone()))
-            .collect(),
-    );
-    let mut matched = vec![false; explanations.len()];
-    let mut queries = Vec::new();
-    for (name, source, qe) in &lowered_queries {
-        let mut dag = dag_export::export(qe);
-        annotate_with_explanations(&mut dag, &explanations, &mut matched);
-        queries.push(NamedDAG {
-            name: name.clone(),
-            source: Some(source.clone()),
-            dag,
-            replacements: Vec::new(),
-            post_dag: None,
-            workload_cost: None,
-            rejections: Vec::new(),
-        });
-    }
-    for (explanation, matched) in explanations.iter().zip(matched) {
-        if !matched {
-            eprintln!(
-                "dag_export: explanation at {} ({:?}, node_hash={}) matched no DAGNode",
-                explanation.location, explanation.kind, explanation.node_hash
-            );
-        }
-    }
-    if progress {
-        eprintln!(
-            "[2/4] Pre-ASAP DAG generation done in {:.2} ms",
-            pre_started.elapsed().as_secs_f64() * 1000.0
-        );
-    }
-
-    if post_asap {
-        let accuracy_evidence = topk_margin
-            .as_ref()
-            .map(|evidence| evidence as &dyn AccuracyEvidenceProvider);
-        let results = if let Some(document) = planner_cost.as_ref() {
-            let model = ExportPlannerCostModel { document };
-            run_post_asap_with_progress(
-                &lowered_queries,
-                progress,
-                &model,
-                Some(&model),
-                accuracy_evidence,
-            )
-        } else if default_cost {
-            // Ranks with structural node counts, exports no cost:
-            // `export_model: None` makes every decision's annotations
-            // `winner_cost_annotations()` — `CostSource::Unavailable`,
-            // `value: None`. `DefaultCostModel`'s number decides which
-            // candidate wins and is then discarded, never serialized, so
-            // the export claims a decision but no cost for it.
-            //
-            // `--topk-margin-json` still applies: it is accuracy evidence,
-            // orthogonal to which cost model ranks.
-            run_post_asap_with_progress(
-                &lowered_queries,
-                progress,
-                &DefaultCostModel,
-                None,
-                accuracy_evidence,
-            )
-        } else {
-            eprintln!(
-                "dag_export: --post-asap needs a cost model; pass --planner-cost-json for \
-                 deployment-owned costs, or --default-cost to rank structurally and export \
-                 the decisions without costs. Exporting the raw plan only"
-            );
-            raw_only_post_asap_results()
-        };
-        for (query_name, replacement) in results.replacements {
-            if let Some(named) = queries.iter_mut().find(|q| q.name == query_name) {
-                named.replacements.push(replacement);
-            }
-        }
-        for (query_name, post_dag) in results.post_dags {
-            if let Some(named) = queries.iter_mut().find(|q| q.name == query_name) {
-                named.post_dag = Some(post_dag);
-            }
-        }
-        for (query_name, rejection) in results.rejections {
-            if let Some(named) = queries.iter_mut().find(|q| q.name == query_name) {
-                named.rejections.push(rejection);
-            }
-        }
-    }
-
-    {
-        let mut pre_dags: Vec<_> = queries.iter_mut().map(|query| &mut query.dag).collect();
-        assign_workload_node_ids(&mut pre_dags);
-    }
-    {
-        let mut post_dags: Vec<_> = queries
-            .iter_mut()
-            .filter_map(|query| query.post_dag.as_mut())
-            .collect();
-        assign_workload_node_ids(&mut post_dags);
-    }
-
-    // Whole selected-workload cost/benefit (issue #286) — per query, and
-    // for the whole selected workload. `decision.id` (== the winning
-    // `Winner`'s own index — see `run_post_asap_with_progress`) is already
-    // a collision-free dedup key for a decision shared across multiple
-    // nodes (a replacement region spans several nodes, all carrying the
-    // same `decision.id`) and across multiple queries (a CSE-shared target
-    // reachable from more than one query's root) alike, so it's reused
-    // directly as `sum_workload_costs`'s dedup key — no separate lookup
-    // needed.
-    let mut workload_entries = Vec::new();
-    for query in &mut queries {
-        let Some(post_dag) = &query.post_dag else {
-            continue;
-        };
-        let entries = decision_cost_entries(post_dag);
-        if entries.is_empty() {
-            continue;
-        }
-        match asap_types::cost::workload_cost_summary(
-            entries
-                .iter()
-                .map(|(id, baseline, selected)| (Some(*id), baseline, selected)),
-            "dag_export-workload-cost-v1",
-        ) {
-            Ok(summary) => query.workload_cost = Some(summary),
-            Err(mismatch) => eprintln!(
-                "dag_export: workload cost aggregation for {:?} skipped — {mismatch}",
-                query.name
-            ),
-        }
-        workload_entries.extend(entries);
-    }
-    let workload_cost = if workload_entries.is_empty() {
-        None
-    } else {
-        match asap_types::cost::workload_cost_summary(
-            workload_entries
-                .iter()
-                .map(|(id, baseline, selected)| (Some(*id), baseline, selected)),
-            "dag_export-workload-cost-v1",
-        ) {
-            Ok(summary) => Some(summary),
-            Err(mismatch) => {
-                eprintln!("dag_export: workload-wide cost aggregation skipped — {mismatch}");
-                None
-            }
-        }
-    };
-
-    let workload = WorkloadDAG {
-        queries,
-        workload_cost,
-    };
-    if progress {
-        eprintln!(
-            "Total planner time: {:.2} ms",
-            planner_started.elapsed().as_secs_f64() * 1000.0
-        );
-    }
-    println!("{}", serde_json::to_string_pretty(&workload).unwrap());
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    use asap_devtools::PromqlError;
-    use asap_plan_selection::cost::analytical_cost::{
-        ExecutionMultiplicity, PhysicalDAGNode, PhysicalOperator,
-    };
-    use asap_plan_selection::cost::physical_operator_statistics::{
-        EdgeStatistics, OperatorStatistics, ScanSelection, UnaryEdgeStatistics,
-    };
-    use asap_plan_selection::cost::query_physical_lowering::lower_query_physical_dag;
-    use asap_types::ir::operator::{Reduction, Source};
-    use asap_types::ir::schema::{DataType, Field, Schema};
-    use asap_types::ir::NonASAPOp;
-
-    fn lower_promql(
-        query: &str,
-        accuracy: AccuracyTarget,
-    ) -> Result<Rc<OperatorNode>, PromqlError> {
-        lower_promql_with_data_ingestion_interval(query, accuracy, 1_000)
-    }
-
-    fn non_topk_query() -> Rc<OperatorNode> {
-        let scan = OperatorNode::new_shared(asap_types::ir::Operator::NonASAP(NonASAPOp::Scan {
-            source: Source::Table {
-                table_ref: "events".into(),
-            },
-            predicates: vec![],
-            schema: Schema::new(vec![Field::plain("v", DataType::Int64, false)]),
-        }))
-        .expect("scan leaf derives its schema");
-        OperatorNode::new_shared(asap_types::ir::Operator::NonASAP(NonASAPOp::Aggregate {
-            reduction: Reduction::by(vec![]),
-            measures: vec![asap_types::ir::operator::AggIntent::Count {
-                accuracy: AccuracyTarget::Epsilon(0.1),
-            }],
-            output_names: vec![],
-            filters: vec![],
-            having: None,
-            child: scan,
-        }))
-        .expect("count aggregate derives its schema")
-    }
-
-    fn fixture_raw_dag(
-        query: &Rc<OperatorNode>,
-        candidate: &ReplacementSubDAG,
-        document: &PlannerCostDocument,
-    ) -> PhysicalDAG {
-        let model = ExportPlannerCostModel { document };
-        let root = Rc::clone(query);
-        let target = asap_logical_optimizer::pass1::replacement::TargetSubDAG::new(&root);
-        let (provider, _) = model.bound(candidate, &target).unwrap();
-        let snapshot = provider.capture_evidence_snapshot(&target).unwrap();
-        let evidence =
-            |request: PhysicalNodeRequest<'_>| provider.query_node_evidence(&snapshot, request);
-        lower_query_physical_dag(&root, &snapshot.scope, &evidence).unwrap()
-    }
-
-    // JSON evidence reaches calibrated ranking and structured annotation inputs.
-    #[test]
-    fn storage_requests_export_and_change_plan_selection() {
-        use asap_plan_selection::cost::storage_io::*;
-        let (query, candidate, mut document) = cost_fixture();
-        let raw = fixture_raw_dag(&query, &candidate, &document);
-        let candidate_dag = cheap_candidate_dag();
-        let root = Rc::clone(&query);
-        let target = asap_logical_optimizer::pass1::replacement::TargetSubDAG::new(&root);
-        assert!(ExportPlannerCostModel {
-            document: &document
-        }
-        .candidate_cost(&candidate, &target)
-        .is_some());
-        let mut profile = StorageIoProfile {
-            evidence_version: document.evidence_version.clone(),
-            observed_at_ms: 900,
-            valid_until_ms: 2000,
-            calibration: StorageCalibration {
-                version: "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![],
-                };
-                profile.nodes.insert(
-                    node.id.clone(),
-                    StorageNodeEvidence {
-                        node: node.clone(),
-                        statistics,
-                        accesses,
-                    },
-                );
-            }
-        }
-        document.storage_io = 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 == "object_get_operations")
-                .unwrap()
-                .value,
-            160.0
-        );
-        assert_eq!(
-            selected
-                .inputs
-                .iter()
-                .find(|term| term.name == "object_get_operations")
-                .unwrap()
-                .value,
-            10.0
-        );
-        assert!(selected
-            .inputs
-            .iter()
-            .any(|term| term.name == "physical_node:summary-read:object_get_operations"));
-        assert!(selected
-            .model_version
-            .as_ref()
-            .unwrap()
-            .contains(STORAGE_IO_MODEL_VERSION));
-        assert_eq!(selected.cache_profile.as_deref(), Some("no-cache-v1"));
-        for name in ["result_cache_hit_ratio", "buffer_cache_hit_ratio"] {
-            assert_eq!(
-                selected
-                    .inputs
-                    .iter()
-                    .find(|term| term.name == name)
-                    .unwrap()
-                    .value,
-                0.0
-            );
-        }
-        assert_eq!(
-            selected.evidence_version.as_deref(),
-            Some("test-evidence-v1")
-        );
-        // JSON evidence also supports an objective priced only by requests.
-        let mut requests_only = parsed.clone();
-        requests_only.calibration.cost_per_cpu_op = 0.0;
-        requests_only.calibration.cost_per_scan_byte = 0.0;
-        requests_only.calibration.cost_per_retained_byte = 0.0;
-        let requests_only =
-            parse_planner_cost_document(&serde_json::to_string(&requests_only).unwrap()).unwrap();
-        let model = ExportPlannerCostModel {
-            document: &requests_only,
-        };
-        assert_eq!(model.candidate_cost(&candidate, &target), Some(Cost(10.0)));
-        let (baseline, selected, _) = model.annotations(&candidate, &root);
-        assert_eq!(baseline.value, Some(160.0));
-        assert_eq!(selected.value, Some(10.0));
-        // Aggregate cache hit assumptions cannot locate cached extents or
-        // reconstruct independently rounded physical storage requests.
-        let mut cached = serde_json::to_value(&requests_only).unwrap();
-        cached["targets"][0]["scope"]["cache_profile"] = serde_json::json!({
-            "profile": "evidence", "version": "cached-with-storage-v1",
-            "distinct_evaluations": 1, "repeated_identical_evaluations": 9,
-            "result_cache": {"working_set_bytes": 100, "capacity_bytes": 100},
-            "buffer_cache": {"working_set_bytes": 100, "capacity_bytes": 100},
-            "result_invalidation_ratio": null,
-        });
-        let cached = parse_planner_cost_document(&cached.to_string()).unwrap();
-        let cached_model = ExportPlannerCostModel { document: &cached };
-        assert!(cached_model.candidate_cost(&candidate, &target).is_none());
-        assert!(cached_model
-            .annotations(&candidate, &root)
-            .0
-            .value
-            .is_none());
-        document
-            .storage_io
-            .as_mut()
-            .unwrap()
-            .nodes
-            .get_mut("summary-read")
-            .unwrap()
-            .accesses
-            .push(StorageAccess {
-                operation: StorageOperation::DiskWrite,
-                extent_bytes: vec![1_000_000],
-                bytes_per_request: 1,
-            });
-        assert!(ExportPlannerCostModel {
-            document: &document
-        }
-        .candidate_cost(&candidate, &target)
-        .is_none());
-        document
-            .storage_io
-            .as_mut()
-            .unwrap()
-            .nodes
-            .remove(&raw.root);
-        assert!(ExportPlannerCostModel {
-            document: &document
-        }
-        .annotations(&candidate, &root)
-        .0
-        .value
-        .is_none());
-    }
-
-    // Declared handoffs survive JSON and affect the selected physical plan.
-    #[test]
-    fn handoff_bytes_export_and_change_plan_selection() {
-        use asap_plan_selection::cost::physical_handoff_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::clone(&query);
-        let target = asap_logical_optimizer::pass1::replacement::TargetSubDAG::new(&root);
-        assert!(ExportPlannerCostModel {
-            document: &document
-        }
-        .candidate_cost(&candidate, &target)
-        .is_some());
-        let mut profile = PhysicalHandoffProfile {
-            evidence_version: document.evidence_version.clone(),
-            observed_at_ms: 900,
-            valid_until_ms: 2000,
-            calibration: PhysicalHandoffCalibration {
-                version: "bytes-v1".into(),
-                cost_per_network_byte: 1.0,
-                cost_per_materialization_byte: 1.0,
-            },
-            plans: vec![],
-        };
-        for dag in [&raw, &candidate_dag] {
-            profile.plans.push(PhysicalHandoffPlanEvidence {
-                root: dag.root.clone(),
-                nodes: dag
-                    .nodes
-                    .iter()
-                    .map(|node| {
-                        (
-                            node.id.clone(),
-                            PhysicalHandoffNodeEvidence {
-                                node: node.clone(),
-                                statistics: dag.evidence[&node.id].statistics.clone(),
-                                handoffs: vec![],
-                            },
-                        )
-                    })
-                    .collect(),
-            });
-        }
-        profile.plans[1]
-            .nodes
-            .get_mut("summary-read")
-            .unwrap()
-            .handoffs
-            .push(PhysicalHandoff {
-                id: "summary-transfer".into(),
-                consumer: None,
-                kind: PhysicalHandoffKind::Network {
-                    source_location: "edge".into(),
-                    destination_location: "backend".into(),
-                },
-                logical_bytes: 2400,
-                encoded_bytes: 1200,
-                copies: 1,
-            });
-        document.handoffs = Some(profile);
-        // Rust handoff names retain the deployed planner-cost JSON field names.
-        let wire = serde_json::to_value(&document).unwrap();
-        assert!(wire.get("handoffs").is_none());
-        assert!(wire["boundaries"]["plans"][1]["nodes"]["summary-read"]["boundaries"].is_array());
-        let parsed = parse_planner_cost_document(&wire.to_string()).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(PHYSICAL_HANDOFF_MODEL_VERSION));
-        assert_eq!(
-            selected.evidence_version.as_deref(),
-            Some("test-evidence-v1")
-        );
-        let mut handoff_only = parsed.clone();
-        handoff_only.calibration.cost_per_cpu_op = 0.0;
-        handoff_only.calibration.cost_per_scan_byte = 0.0;
-        handoff_only.calibration.cost_per_retained_byte = 0.0;
-        let (baseline, selected, _) = ExportPlannerCostModel {
-            document: &handoff_only,
-        }
-        .annotations(&candidate, &root);
-        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_plan_selection::cost::storage_io::*;
-            let mut joint = handoff_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(PHYSICAL_HANDOFF_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.handoffs = 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 handoffs 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(PHYSICAL_HANDOFF_MODEL_VERSION));
-        }
-
-        let mut ambiguous = parsed.clone();
-        let plans = &mut ambiguous.handoffs.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
-            .handoffs
-            .as_mut()
-            .unwrap()
-            .calibration
-            .cost_per_network_byte = 1000.0;
-        assert!(ExportPlannerCostModel {
-            document: &document
-        }
-        .candidate_cost(&candidate, &target)
-        .is_none());
-        document.handoffs.as_mut().unwrap().plans[0]
-            .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,
-            planning_time_ms: 1_000,
-            horizon_ms: 10_000,
-            evaluation_count: 10,
-            time_scope: "longitudinal".into(),
-            lookback_ms: Some(10_000),
-            as_of_ms: Some(1_000),
-            sources: vec![ScanSelection {
-                source: Source::Table {
-                    table_ref: "events".into(),
-                },
-                source_snapshot_id: "snapshot-1".into(),
-                predicates: vec![],
-                info_matchers: vec![],
-            }],
-            cache_profile: CacheProfile::no_cache(),
-        }
-    }
-
-    fn edge(rows: u64, bytes: u64) -> EdgeStatistics {
-        EdgeStatistics { rows, bytes }
-    }
-
-    fn query_evidence(query: &Rc<OperatorNode>) -> Vec<QueryNodePhysicalEvidence> {
-        let entries = RefCell::new(Vec::new());
-        let scope = test_scope().resolve().unwrap();
-        let provider = |request: PhysicalNodeRequest<'_>| {
-            let (statistics, output_buffer_bytes) = match request.operator {
-                PhysicalOperator::Scan => (
-                    OperatorStatistics::Scan {
-                        edges: UnaryEdgeStatistics {
-                            input: edge(1_000, 64_000),
-                            output: edge(1_000, 64_000),
-                            promql: None,
-                        },
-                        source_read_bytes: 64_000,
-                    },
-                    64_000,
-                ),
-                PhysicalOperator::HashAggregate {
-                    grouping_key_count: 0,
-                    ..
-                } => (
-                    OperatorStatistics::HashAggregate {
-                        edges: UnaryEdgeStatistics {
-                            input: edge(1_000, 64_000),
-                            output: edge(1, 8),
-                            promql: None,
-                        },
-                        group_count: 1,
-                        key_bytes: 0,
-                        accumulator_bytes_per_group: 8,
-                    },
-                    8,
-                ),
-                _ => return Err(AnalyticalCostError::UnsupportedQueryOperator),
-            };
-            let evidence = PhysicalNodeEvidence {
-                physical_id: format!("{:?}-{}", request.operator, request.occurrence),
-                statistics,
-                output_buffer_bytes,
-            };
-            entries.borrow_mut().push(QueryNodePhysicalEvidence {
-                logical_node: request.logical_node.clone(),
-                operator: request.operator,
-                occurrence: request.occurrence,
-                synthetic: request.synthetic,
-                evidence: evidence.clone(),
-            });
-            Ok(evidence)
-        };
-        lower_query_physical_dag(query, &scope, &provider).unwrap();
-        entries.into_inner()
-    }
-
-    fn cheap_candidate_dag() -> PhysicalDAG {
-        let coverage = test_scope().sources[0].clone();
-        let statistics = OperatorStatistics::Scan {
-            edges: UnaryEdgeStatistics {
-                input: edge(100, 2_400),
-                output: edge(100, 2_400),
-                promql: None,
-            },
-            source_read_bytes: 64,
-        };
-        PhysicalDAG {
-            nodes: vec![PhysicalDAGNode {
-                id: "summary-read".into(),
-                operator: PhysicalOperator::Scan,
-                children: vec![],
-                scan_selection: Some(coverage),
-                output_buffer_bytes: 2_400,
-                retained_bytes: 0,
-                execution: ExecutionMultiplicity::PerEvaluation,
-            }],
-            root: "summary-read".into(),
-            evidence: [(
-                "summary-read".into(),
-                PhysicalNodeEvidence {
-                    physical_id: "summary-read".into(),
-                    statistics,
-                    output_buffer_bytes: 2_400,
-                },
-            )]
-            .into_iter()
-            .collect(),
-        }
-    }
-
-    fn candidate_plan(candidate: &ReplacementSubDAG) -> serde_json::Value {
-        match &candidate.replacement {
-            Replacement::SubDAG(node) => serde_json::to_value(dag_export::export(node)).unwrap(),
-            Replacement::ExactComposition(_) => {
-                unreachable!("cost fixtures select directly materialized candidates")
-            }
-        }
-    }
-
-    fn cost_fixture() -> (Rc<OperatorNode>, ReplacementSubDAG, PlannerCostDocument) {
-        let query = non_topk_query();
-        let root = Rc::clone(&query);
-        let space = search_workload(vec![(String::from("q"), Rc::clone(&root))]);
-        let group = space
-            .target_subdag_candidates()
-            .find(|group| group.target == query)
-            .expect("aggregate memo group");
-        let candidate = group
-            .candidates
-            .iter()
-            .find(|candidate| !is_trivial_retain_exact(&candidate.replacement))
-            .expect("summary candidate")
-            .clone();
-        let plan = candidate_plan(&candidate);
-        let document = PlannerCostDocument {
-            storage_io: None,
-            handoffs: None,
-            evidence_version: "test-evidence-v1".into(),
-            calibration: ResourceCalibration {
-                cost_per_cpu_op: 1.0,
-                cost_per_scan_byte: 1.0,
-                cost_per_retained_byte: 1.0,
-                version: "test".into(),
-            },
-            targets: vec![TargetPhysicalEvidence {
-                target: query.clone(),
-                scope: test_scope(),
-                candidates: vec![match &candidate.replacement {
-                    Replacement::SubDAG(node) if !is_logical_rewrite(node) => {
-                        CandidatePhysicalEvidence::Summary {
-                            plan,
-                            query_nodes: query_evidence(&query),
-                            physical_dag: cheap_candidate_dag(),
-                        }
-                    }
-                    Replacement::SubDAG(_) => CandidatePhysicalEvidence::Rewrite {
-                        plan,
-                        query_nodes: query_evidence(&query),
-                    },
-                    Replacement::ExactComposition(_) => {
-                        unreachable!("cost fixtures select directly materialized candidates")
-                    }
-                }],
-            }],
-        };
-        (query, candidate, document)
-    }
-
-    #[test]
-    fn planner_json_costs_and_exports_a_generic_non_topk_winner() {
-        let (query, candidate, document) = cost_fixture();
-        let json = serde_json::to_string(&document).unwrap();
-        let parsed = parse_planner_cost_document(&json).unwrap();
-        assert_eq!(parsed.targets[0].target, query);
-        assert!(parsed.targets[0].candidates[0].matches(&candidate));
-        let model = ExportPlannerCostModel { document: &parsed };
-        let target_rc = Rc::clone(&query);
-        let target = asap_logical_optimizer::pass1::replacement::TargetSubDAG::new(&target_rc);
-        let (provider, calibration) = model.bound(&candidate, &target).expect("exact binding");
-        let estimate = PhysicalPlanCostModel::new(&provider, calibration.clone())
-            .unwrap()
-            .estimate_candidate(&candidate, &target)
-            .unwrap();
-        assert!(estimate.candidate_cost < estimate.raw_cost);
-        let (baseline, selected, benefit) = model.annotations(&candidate, &query);
-        assert!(baseline.value.is_some());
-        assert!(selected.value.is_some());
-        assert!(benefit.value.is_some());
-        for annotation in [&baseline, &selected, &benefit] {
-            assert_eq!(
-                annotation.evidence_version.as_deref(),
-                Some("test-evidence-v1")
-            );
-            assert_eq!(annotation.cache_profile.as_deref(), Some("no-cache-v1"));
-            assert!(annotation
-                .model_version
-                .as_deref()
-                .is_some_and(|version| version.starts_with(ANALYTICAL_COST_MODEL_VERSION)));
-        }
-        assert!(selected
-            .inputs
-            .iter()
-            .any(|input| input.name == "estimated_scan"));
-        assert!(!selected.inputs.iter().any(|input| input.name == "topk_k"));
-    }
-
-    #[test]
-    fn legacy_cache_json_defaults_to_named_no_cache_but_malformed_profiles_fail() {
-        // Existing evidence files keep their costs and acquire explicit provenance.
-        let (query, candidate, document) = cost_fixture();
-        let mut json = serde_json::to_value(&document).unwrap();
-        json["targets"][0]["scope"]
-            .as_object_mut()
-            .unwrap()
-            .remove("cache_profile");
-        let parsed = parse_planner_cost_document(&json.to_string()).unwrap();
-        let target = query;
-        let legacy = ExportPlannerCostModel { document: &parsed }.annotations(&candidate, &target);
-        let explicit = ExportPlannerCostModel {
-            document: &document,
-        }
-        .annotations(&candidate, &target);
-        assert_eq!(legacy, explicit);
-        assert_eq!(legacy.0.cache_profile.as_deref(), Some("no-cache-v1"));
-        let exported = serde_json::to_value(legacy.0).unwrap();
-        assert_eq!(exported["cache_profile"], "no-cache-v1");
-
-        json["targets"][0]["scope"]["cache_profile"] = serde_json::json!({"profile": "evidence"});
-        assert!(parse_planner_cost_document(&json.to_string()).is_err());
-        json["targets"][0]["scope"]["cache_profile"] = serde_json::Value::Null;
-        assert!(parse_planner_cost_document(&json.to_string()).is_err());
-    }
-
-    #[test]
-    fn cache_json_affects_ranking_and_exports_declared_evidence() {
-        // Identical repeats hit the result cache; distinct evaluations still execute.
-        let (query, candidate, document) = cost_fixture();
-        let target_rc = query;
-        let target = asap_logical_optimizer::pass1::replacement::TargetSubDAG::new(&target_rc);
-        let no_cache = ExportPlannerCostModel {
-            document: &document,
-        }
-        .annotations(&candidate, &target_rc);
-        let mut json = serde_json::to_value(&document).unwrap();
-        json["targets"][0]["scope"]["cache_profile"] = serde_json::json!({
-            "profile": "evidence",
-            "version": "warm-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 parsed = parse_planner_cost_document(&json.to_string()).unwrap();
-        let model = ExportPlannerCostModel { document: &parsed };
-        let (baseline, selected, _) = model.annotations(&candidate, &target_rc);
-        let value = |annotation: &CostAnnotation, name: &str| {
-            annotation
-                .inputs
-                .iter()
-                .find(|input| input.name == name)
-                .unwrap()
-                .value
-        };
-        assert_eq!(
-            value(&baseline, "estimated_cpu_ops"),
-            value(&no_cache.0, "estimated_cpu_ops") * 0.2
-        );
-        assert_eq!(
-            value(&baseline, "estimated_scan"),
-            value(&no_cache.0, "estimated_scan") * 0.1
-        );
-        assert_eq!(value(&baseline, "result_cache_hit_ratio"), 1.0);
-        assert_eq!(value(&baseline, "buffer_cache_hit_ratio"), 0.5);
-        assert_eq!(value(&baseline, "distinct_evaluations"), 2.0);
-        assert_eq!(value(&baseline, "repeated_identical_evaluations"), 8.0);
-        assert_eq!(value(&baseline, "result_cache_working_set"), 100.0);
-        assert_eq!(value(&baseline, "buffer_cache_capacity"), 500.0);
-        assert_eq!(baseline.cache_profile.as_deref(), Some("warm-cache-v1"));
-        assert!(selected.value.unwrap() < no_cache.1.value.unwrap());
-        assert_eq!(
-            model.candidate_cost(&candidate, &target).unwrap().0,
-            selected.value.unwrap()
-        );
-        let exported = serde_json::to_value(&selected).unwrap();
-        assert_eq!(exported["cache_profile"], "warm-cache-v1");
-        assert!(exported["inputs"]
-            .as_array()
-            .unwrap()
-            .iter()
-            .any(|input| { input["name"] == "buffer_cache_capacity" && input["unit"] == "bytes" }));
-
-        json["targets"][0]["scope"]["cache_profile"]["distinct_evaluations"] = 10.into();
-        json["targets"][0]["scope"]["cache_profile"]["repeated_identical_evaluations"] = 0.into();
-        json["targets"][0]["scope"]["cache_profile"]["buffer_cache"]["capacity_bytes"] = 0.into();
-        let distinct = parse_planner_cost_document(&json.to_string()).unwrap();
-        let distinct = ExportPlannerCostModel {
-            document: &distinct,
-        }
-        .annotations(&candidate, &target_rc);
-        assert_eq!(distinct.0.value, no_cache.0.value);
-        assert_eq!(
-            value(&distinct.0, "estimated_cpu_ops"),
-            value(&no_cache.0, "estimated_cpu_ops")
-        );
-        assert_eq!(
-            value(&distinct.0, "estimated_scan"),
-            value(&no_cache.0, "estimated_scan")
-        );
-
-        json["targets"][0]["scope"]["cache_profile"]["distinct_evaluations"] = 9.into();
-        let invalid = parse_planner_cost_document(&json.to_string()).unwrap();
-        let invalid = ExportPlannerCostModel { document: &invalid };
-        assert!(invalid.candidate_cost(&candidate, &target).is_none());
-        assert!(invalid
-            .annotations(&candidate, &target_rc)
-            .0
-            .value
-            .is_none());
-    }
-
-    #[test]
-    fn duplicate_target_candidate_and_query_evidence_each_fail_closed() {
-        let (query, candidate, document) = cost_fixture();
-        let target_rc = query;
-        let target = asap_logical_optimizer::pass1::replacement::TargetSubDAG::new(&target_rc);
-
-        let mut duplicate_target = document.clone();
-        duplicate_target
-            .targets
-            .push(duplicate_target.targets[0].clone());
-        assert!(ExportPlannerCostModel {
-            document: &duplicate_target
-        }
-        .candidate_cost(&candidate, &target)
-        .is_none());
-
-        let mut duplicate_candidate = document.clone();
-        let candidate_entry = duplicate_candidate.targets[0].candidates[0].clone();
-        duplicate_candidate.targets[0]
-            .candidates
-            .push(candidate_entry);
-        assert!(ExportPlannerCostModel {
-            document: &duplicate_candidate
-        }
-        .candidate_cost(&candidate, &target)
-        .is_none());
-
-        let mut duplicate_query = document;
-        let query_entry = duplicate_query.targets[0].candidates[0].query_nodes()[0].clone();
-        match &mut duplicate_query.targets[0].candidates[0] {
-            CandidatePhysicalEvidence::Rewrite { query_nodes, .. }
-            | CandidatePhysicalEvidence::Summary { query_nodes, .. } => {
-                query_nodes.push(query_entry)
-            }
-        }
-        assert!(ExportPlannerCostModel {
-            document: &duplicate_query
-        }
-        .candidate_cost(&candidate, &target)
-        .is_none());
-    }
-
-    #[test]
-    fn incomplete_or_unused_json_evidence_fails_closed() {
-        let (query, candidate, document) = cost_fixture();
-        let target_rc = query;
-        let target = asap_logical_optimizer::pass1::replacement::TargetSubDAG::new(&target_rc);
-
-        let mut missing = document.clone();
-        match &mut missing.targets[0].candidates[0] {
-            CandidatePhysicalEvidence::Rewrite { query_nodes, .. }
-            | CandidatePhysicalEvidence::Summary { query_nodes, .. } => {
-                query_nodes.pop();
-            }
-        }
-        let missing = parse_planner_cost_document(&serde_json::to_string(&missing).unwrap())
-            .expect("well-formed but incomplete evidence document");
-        assert!(ExportPlannerCostModel { document: &missing }
-            .candidate_cost(&candidate, &target)
-            .is_none());
-
-        let mut unused = document;
-        let mut extra = unused.targets[0].candidates[0].query_nodes()[0].clone();
-        extra.occurrence = usize::MAX;
-        match &mut unused.targets[0].candidates[0] {
-            CandidatePhysicalEvidence::Rewrite { query_nodes, .. }
-            | CandidatePhysicalEvidence::Summary { query_nodes, .. } => query_nodes.push(extra),
-        }
-        let unused = parse_planner_cost_document(&serde_json::to_string(&unused).unwrap())
-            .expect("well-formed document with unused evidence");
-        assert!(ExportPlannerCostModel { document: &unused }
-            .candidate_cost(&candidate, &target)
-            .is_none());
-    }
-
-    #[test]
-    fn invalid_candidate_dag_from_json_fails_closed() {
-        let (query, candidate, mut document) = cost_fixture();
-        let CandidatePhysicalEvidence::Summary { physical_dag, .. } =
-            &mut document.targets[0].candidates[0]
-        else {
-            panic!("fixture must use a summary candidate");
-        };
-        physical_dag.nodes.push(physical_dag.nodes[0].clone());
-        let document = parse_planner_cost_document(&serde_json::to_string(&document).unwrap())
-            .expect("invalid physical semantics are checked by the estimator");
-        let target_rc = query;
-        let target = asap_logical_optimizer::pass1::replacement::TargetSubDAG::new(&target_rc);
-        assert!(ExportPlannerCostModel {
-            document: &document
-        }
-        .candidate_cost(&candidate, &target)
-        .is_none());
-    }
-
-    #[test]
-    fn global_selection_uses_the_cheapest_complete_physical_candidate() {
-        let query = non_topk_query();
-        let root = Rc::clone(&query);
-        let space = search_workload(vec![(String::from("q"), Rc::clone(&root))]);
-        let group = space
-            .target_subdag_candidates()
-            .find(|group| group.target == query)
-            .expect("aggregate memo group");
-        let candidates: Vec<_> = group
-            .candidates
-            .iter()
-            .filter(|candidate| {
-                matches!(&candidate.replacement, Replacement::SubDAG(node) if node.contains_asap())
-            })
-            .take(2)
-            .collect();
-        assert_eq!(
-            candidates.len(),
-            2,
-            "fixture needs two physical alternatives"
-        );
-
-        let mut first_dag = cheap_candidate_dag();
-        let OperatorStatistics::Scan {
-            source_read_bytes, ..
-        } = &mut first_dag
-            .evidence
-            .get_mut("summary-read")
-            .unwrap()
-            .statistics
-        else {
-            panic!("summary read fixture must be a scan")
-        };
-        *source_read_bytes = 1_024;
-        let second_dag = cheap_candidate_dag();
-        let document = PlannerCostDocument {
-            storage_io: None,
-            handoffs: None,
-            evidence_version: "test-evidence-v1".into(),
-            calibration: ResourceCalibration {
-                cost_per_cpu_op: 1.0,
-                cost_per_scan_byte: 1.0,
-                cost_per_retained_byte: 1.0,
-                version: "test".into(),
-            },
-            targets: vec![TargetPhysicalEvidence {
-                target: query.clone(),
-                scope: test_scope(),
-                candidates: vec![
-                    CandidatePhysicalEvidence::Summary {
-                        plan: candidate_plan(candidates[0]),
-                        query_nodes: query_evidence(&query),
-                        physical_dag: first_dag,
-                    },
-                    CandidatePhysicalEvidence::Summary {
-                        plan: candidate_plan(candidates[1]),
-                        query_nodes: query_evidence(&query),
-                        physical_dag: second_dag,
-                    },
-                ],
-            }],
-        };
-        let document = parse_planner_cost_document(&serde_json::to_string(&document).unwrap())
-            .expect("complete physical evidence");
-        let model = ExportPlannerCostModel {
-            document: &document,
-        };
-        let selection = global_selection(&space, &model);
-        let chosen = selection
-            .target_selections()
-            .find(|selected| *selected.target == query)
-            .and_then(|selected| selected.chosen)
-            .expect("one complete physical candidate should win");
-        assert!(document.targets[0].candidates[1].matches(chosen));
-        assert!(!document.targets[0].candidates[0].matches(chosen));
-    }
-
-    #[test]
-    fn cost_annotations_fail_closed_without_physical_evidence() {
-        let (baseline, selected, benefit) = winner_cost_annotations();
-        for annotation in [baseline, selected, benefit] {
-            assert!(annotation.value.is_none());
-            assert_eq!(annotation.source, asap_types::cost::CostSource::Unavailable);
-            assert_eq!(annotation.unit, CostUnit::CostUnits);
-        }
-    }
-
-    #[test]
-    fn missing_physical_evidence_keeps_the_export_raw_only() {
-        let results = raw_only_post_asap_results();
-        assert!(results.replacements.is_empty());
-        assert!(results.post_dags.is_empty());
-    }
-
-    fn argv(args: &[&str]) -> impl Iterator<Item = String> {
-        args.iter()
-            .map(|arg| arg.to_string())
-            .collect::<Vec<_>>()
-            .into_iter()
-    }
-
-    /// `--default-cost` is a cost *source* for `--post-asap`, not a second
-    /// way to spell the evidence document: it parses on its own and leaves
-    /// `planner_cost` empty, which is what keeps every exported cost
-    /// `Unavailable`.
-    #[test]
-    fn default_cost_parses_as_a_standalone_post_asap_cost_source() {
-        let parsed = parse_args_from(argv(&[
-            "--post-asap",
-            "--default-cost",
-            "--sql",
-            "SELECT COUNT(*) FROM metrics",
-            "--name",
-            "q1",
-        ]));
-        assert!(parsed.post_asap);
-        assert!(parsed.default_cost);
-        assert!(parsed.planner_cost.is_none());
-        assert_eq!(parsed.entries.len(), 1);
-    }
-
-    /// Absent the flag, nothing changes: `--post-asap` alone still reaches
-    /// `main`'s raw-only branch.
-    #[test]
-    fn default_cost_is_off_unless_asked_for() {
-        let parsed = parse_args_from(argv(&["--post-asap", "--sql", "SELECT 1", "--name", "q1"]));
-        assert!(!parsed.default_cost);
-        assert!(parsed.planner_cost.is_none());
-    }
-
-    #[test]
-    #[should_panic(expected = "mutually exclusive")]
-    fn default_cost_and_planner_cost_evidence_cannot_be_combined() {
-        parse_args_from(argv(&[
-            "--post-asap",
-            "--default-cost",
-            "--planner-cost-json",
-            r#"{"evidence_version":"v1","calibration":{},"targets":[]}"#,
-        ]));
-    }
-
-    /// The rejection is on the flag pair itself, not on the document
-    /// parsing: an unparseable document paired with `--default-cost` must
-    /// still report the conflict rather than a JSON error, so the caller is
-    /// told which argument to drop.
-    #[test]
-    #[should_panic(expected = "mutually exclusive")]
-    fn the_flag_conflict_is_reported_before_the_document_is_parsed() {
-        parse_args_from(argv(&["--default-cost", "--planner-cost-json", "not json"]));
-    }
-
-    #[test]
-    fn compact_analytical_payload_has_an_explicit_migration_error() {
-        let error = parse_planner_cost_document(r#"{"inputs":{"group_count":10}}"#)
-            .expect_err("legacy payload must fail");
-        assert!(error.contains("compact analytical-cost payload"));
-    }
-
-    #[test]
-    fn evidence_version_is_required_and_non_empty() {
-        let (_, _, mut document) = cost_fixture();
-        document.evidence_version.clear();
-        let error = parse_planner_cost_document(&serde_json::to_string(&document).unwrap())
-            .expect_err("an unnamed mutable evidence snapshot must be rejected");
-        assert!(error.contains("evidence_version"));
-    }
-
-    #[test]
-    fn exact_candidate_selector_does_not_confuse_the_same_strategy() {
-        let selected_query = lower_promql("up", AccuracyTarget::Exact).unwrap();
-        let other_query = lower_promql("process_cpu_seconds_total", AccuracyTarget::Exact).unwrap();
-        let selected = ReplacementSubDAG {
-            replacement: Replacement::SubDAG(Rc::clone(&selected_query)),
-            strategy: "same-strategy",
-            provenance:
-                asap_logical_optimizer::pass1::replacement::ReplacementProvenance::LogicalRewrite,
-            rationale: String::new(),
-        };
-        let other = ReplacementSubDAG {
-            replacement: Replacement::SubDAG(other_query),
-            strategy: "same-strategy",
-            provenance:
-                asap_logical_optimizer::pass1::replacement::ReplacementProvenance::LogicalRewrite,
-            rationale: String::new(),
-        };
-        let selector = CandidatePhysicalEvidence::Rewrite {
-            plan: serde_json::to_value(dag_export::export(&selected_query)).unwrap(),
-            query_nodes: Vec::new(),
-        };
-        assert!(selector.matches(&selected));
-        assert!(!selector.matches(&other));
-    }
-
-    #[test]
-    fn candidate_identity_includes_accuracy_guarantees() {
-        fn alter_guarantee(value: &mut serde_json::Value) -> bool {
-            match value {
-                serde_json::Value::Object(fields) => {
-                    if fields.contains_key("guarantee") {
-                        fields.insert("guarantee".into(), serde_json::json!({"changed": true}));
-                        return true;
-                    }
-                    fields.values_mut().any(alter_guarantee)
-                }
-                serde_json::Value::Array(values) => values.iter_mut().any(alter_guarantee),
-                _ => false,
-            }
-        }
-
-        let (_, candidate, document) = cost_fixture();
-        let mut selector = document.targets[0].candidates[0].clone();
-        let plan = match &mut selector {
-            CandidatePhysicalEvidence::Rewrite { plan, .. }
-            | CandidatePhysicalEvidence::Summary { plan, .. } => plan,
-        };
-        assert!(alter_guarantee(plan), "fixture must contain a guarantee");
-        assert!(!selector.matches(&candidate));
-    }
-
-    #[test]
-    fn one_ulp_tolerance_is_limited_to_guarantee_values() {
-        let value = 0.1_f64;
-        let adjacent = f64::from_bits(value.to_bits() + 1);
-        assert!(plan_values_match(
-            &serde_json::json!({"guarantee": {"bound": {"value": value}}}),
-            &serde_json::json!({"guarantee": {"bound": {"value": adjacent}}}),
-        ));
-        assert!(!plan_values_match(
-            &serde_json::json!({"detail": {"literal": value}}),
-            &serde_json::json!({"detail": {"literal": adjacent}}),
-        ));
-    }
-
-    #[test]
-    fn workload_wide_explanations_annotate_cross_query_reuse() {
-        let query = "sum by (job) (rate(http_requests_total[5m]))";
-        let a = lower_promql(query, AccuracyTarget::Exact).unwrap();
-        let b = lower_promql(query, AccuracyTarget::Exact).unwrap();
-        let explanations =
-            asap_logical_optimizer::explain_replacements(vec![("a", a.clone()), ("b", b.clone())]);
-        assert!(explanations.iter().any(|e| {
-            e.kind == asap_logical_optimizer::ExplanationKind::CommonSubexpressionReuse
-        }));
-
-        let mut matched = vec![false; explanations.len()];
-        let mut dag_a = dag_export::export(&a);
-        let mut dag_b = dag_export::export(&b);
-        annotate_with_explanations(&mut dag_a, &explanations, &mut matched);
-        annotate_with_explanations(&mut dag_b, &explanations, &mut matched);
-        assert!(dag_a.nodes.iter().any(|n| !n.notes.is_empty()));
-        assert!(dag_b.nodes.iter().any(|n| !n.notes.is_empty()));
-    }
-
-    #[test]
-    fn hash_collision_without_structural_equality_does_not_attach_a_note() {
-        let target = lower_promql(
-            "quantile(0.99, rate(http_requests_total[5m]))",
-            AccuracyTarget::Epsilon(0.01),
-        )
-        .unwrap();
-        let explanations = asap_logical_optimizer::explain_replacements(vec![("target", target)]);
-        let explanation = explanations
-            .iter()
-            .find(|e| e.kind == asap_logical_optimizer::ExplanationKind::SketchApproximation)
-            .unwrap();
-
-        let unrelated =
-            lower_promql("sum(rate(other_metric[5m]))", AccuracyTarget::Epsilon(0.01)).unwrap();
-        let mut dag = dag_export::export(&unrelated);
-        for node in &mut dag.nodes {
-            node.hash = Some(explanation.node_hash);
-        }
-        let mut matched = vec![false; explanations.len()];
-        annotate_with_explanations(&mut dag, &explanations, &mut matched);
-        assert!(dag.nodes.iter().all(|n| n.notes.is_empty()));
-    }
-
-    /// The `--post-asap` code path, exercised directly (not through the CLI):
-    /// a workload with one sketch-approximable quantile aggregate and one
-    /// `avg` aggregate must produce at least one `TargetReplacement` with
-    /// `after: TargetReplacementAfter::Summary(..)` (the bound sketch) and
-    /// at least one with `after: TargetReplacementAfter::Rewrite(..)` (the
-    /// `avg -> sum/count` rewrite) — see [`run_post_asap`]. Also checks that
-    /// a non-empty `post_dag` comes back for every query, since that's the
-    /// other `--post-asap` output `main` wires up.
-    #[tokio::test]
-    async fn post_asap_run_produces_both_summary_and_rewrite_replacements() {
-        let cat = default_catalog();
-        let sketch_query = lower_sql(
-            "SELECT approx_percentile_cont(latency, 0.95) FROM metrics",
-            &cat,
-            AccuracyTarget::Epsilon(0.01),
-        )
-        .await
-        .unwrap();
-        let avg_query = lower_sql(
-            "SELECT service, AVG(latency) FROM metrics GROUP BY service",
-            &cat,
-            AccuracyTarget::Epsilon(0.01),
-        )
-        .await
-        .unwrap();
-
-        let lowered_queries = vec![
-            (
-                "sketch".to_string(),
-                "SELECT approx_percentile_cont(latency, 0.95) FROM metrics".to_string(),
-                sketch_query,
-            ),
-            (
-                "avg".to_string(),
-                "SELECT service, AVG(latency) FROM metrics GROUP BY service".to_string(),
-                avg_query,
-            ),
-        ];
-
-        let results = run_post_asap(&lowered_queries);
-
-        assert!(
-            results
-                .replacements
-                .iter()
-                .any(|(_, r)| matches!(r.after, TargetReplacementAfter::Summary(_))),
-            "expected at least one Summary replacement (the sketch-bound quantile aggregate): {:?}",
-            results
-                .replacements
-                .iter()
-                .map(|(name, r)| (name.clone(), r.strategy.clone()))
-                .collect::<Vec<_>>()
-        );
-        assert!(
-            results
-                .replacements
-                .iter()
-                .any(|(_, r)| matches!(r.after, TargetReplacementAfter::Rewrite(_))),
-            "expected at least one Rewrite replacement (the avg -> sum/count rewrite): {:?}",
-            results
-                .replacements
-                .iter()
-                .map(|(name, r)| (name.clone(), r.strategy.clone()))
-                .collect::<Vec<_>>()
-        );
-        // A logical rewrite must retain the semantic rewrite owner's label
-        // (not fall through to "Rollup", the other LogicalRewrite shape).
-        // Depending on which equivalent memo candidate wins, the selected
-        // rewrite may be the original AVG decomposition or a subsequently
-        // exposed nested-aggregate composition.
-        assert!(results.replacements.iter().any(|(_, r)| matches!(
-            r.strategy.as_str(),
-            "AvgToSumOverCountStrategy" | "SemanticEquivalentRewriteStrategy"
-        )));
-
-        assert_eq!(results.post_dags.len(), 2, "one post_dag per query");
-        for (name, dag) in &results.post_dags {
-            assert!(
-                !dag.nodes.is_empty(),
-                "post_dag for {name:?} must not be empty"
-            );
-        }
-        let avg_post = &results
-            .post_dags
-            .iter()
-            .find(|(name, _)| name == "avg")
-            .expect("avg post DAG")
-            .1;
-        assert_eq!(
-            avg_post
-                .nodes
-                .iter()
-                .filter(|node| node.kind == "Scan")
-                .count(),
-            1,
-            "AVG's SUM and COUNT branches must retain their shared input as one DAG node"
-        );
-        let decisions: Vec<_> = results
-            .post_dags
-            .iter()
-            .flat_map(|(_, dag)| dag.nodes.iter().filter_map(|node| node.decision.as_ref()))
-            .collect();
-        assert!(
-            !decisions.is_empty(),
-            "post_dag nodes must carry explicit strategy metadata"
-        );
-        for decision in decisions {
-            assert!(!decision.strategy.is_empty());
-            assert!(!decision.rationale.is_empty());
-        }
-    }
-
-    // TopK queries of different sizes share one exact ranked prefix and differ
-    // only in their final Limit.
-    #[test]
-    fn workload_node_ids_share_ranked_prefix_across_topk_sizes() {
-        let small = lower_promql(
-            "topk(5, rate(http_requests_total[5m]))",
-            AccuracyTarget::Epsilon(0.01),
-        )
-        .unwrap();
-        let large = lower_promql(
-            "topk(10, rate(http_requests_total[5m]))",
-            AccuracyTarget::Epsilon(0.01),
-        )
-        .unwrap();
-        let lowered = vec![
-            (
-                "q3".into(),
-                "topk(5, rate(http_requests_total[5m]))".into(),
-                small,
-            ),
-            (
-                "q4".into(),
-                "topk(10, rate(http_requests_total[5m]))".into(),
-                large,
-            ),
-        ];
-        let mut results = run_post_asap(&lowered);
-        let mut dag_refs: Vec<_> = results.post_dags.iter_mut().map(|(_, dag)| dag).collect();
-        assign_workload_node_ids(&mut dag_refs);
-
-        let q3 = &results
-            .post_dags
-            .iter()
-            .find(|(name, _)| name == "q3")
-            .unwrap()
-            .1;
-        let q4 = &results
-            .post_dags
-            .iter()
-            .find(|(name, _)| name == "q4")
-            .unwrap()
-            .1;
-        let q3_root = &q3.nodes[q3.root as usize];
-        let q4_root = &q4.nodes[q4.root as usize];
-        assert!(q3_root.label.contains("Limit(5)"));
-        assert!(q4_root.label.contains("Limit(10)"));
-        assert_ne!(q3_root.workload_node_id, q4_root.workload_node_id);
-        let q3_ranked = &q3.nodes[q3_root.children[0] as usize];
-        let q4_ranked = &q4.nodes[q4_root.children[0] as usize];
-        assert!(q3_ranked.label.contains("Sort"));
-        assert_eq!(q3_ranked.workload_node_id, q4_ranked.workload_node_id);
-    }
-
-    #[tokio::test]
-    async fn join_demo_explicitly_shares_metrics_scan_with_grouped_query() {
-        let cat = default_catalog();
-        let q1 = lower_sql(
-            "SELECT service, COUNT(*) FROM metrics GROUP BY service",
-            &cat,
-            AccuracyTarget::Exact,
-        )
-        .await
-        .unwrap();
-        let q6 = lower_sql(
-            "SELECT metrics.service, COUNT(*) FROM metrics JOIN hosts ON metrics.service = hosts.service GROUP BY metrics.service",
-            &cat,
-            AccuracyTarget::Exact,
-        )
-        .await
-        .unwrap();
-        let mut q1_dag = dag_export::export(&q1);
-        let mut q6_dag = dag_export::export(&q6);
-        assign_workload_node_ids(&mut [&mut q1_dag, &mut q6_dag]);
-        let q1_scan = q1_dag
-            .nodes
-            .iter()
-            .find(|node| node.label == "Scan(metrics)")
-            .unwrap();
-        let q6_scan = q6_dag
-            .nodes
-            .iter()
-            .find(|node| node.label == "Scan(metrics)")
-            .unwrap();
-        assert_eq!(q1_scan.workload_node_id, q6_scan.workload_node_id);
-        assert!(q6_dag.nodes.iter().any(|node| node.kind == "Join"));
-    }
-
-    /// The `--default-cost` contract, end to end on the code path `main`
-    /// takes for it (`DefaultCostModel` ranking, `export_model: None`): the
-    /// structure must be real — replacements found, a merged `post_dag`
-    /// per query — while every cost stays `Unavailable` with no value, so
-    /// the viewer shows "Not estimated" and the structural ranking number
-    /// never escapes as if it were a measured cost.
-    #[tokio::test]
-    async fn default_cost_exports_real_structure_and_no_costs() {
-        let cat = default_catalog();
-        let query = lower_sql(
-            "SELECT service, AVG(latency) FROM metrics GROUP BY service",
-            &cat,
-            AccuracyTarget::Epsilon(0.01),
-        )
-        .await
-        .unwrap();
-        let lowered = vec![(
-            "q1".to_string(),
-            "SELECT service, AVG(latency) FROM metrics GROUP BY service".to_string(),
-            query,
-        )];
-
-        let results = run_post_asap(&lowered);
-
-        assert!(
-            !results.replacements.is_empty(),
-            "the search itself must still run under --default-cost"
-        );
-        assert!(results
-            .post_dags
-            .iter()
-            .all(|(_, dag)| !dag.nodes.is_empty()));
-
-        for (_, replacement) in &results.replacements {
-            for annotation in [
-                replacement.baseline_cost.as_ref(),
-                replacement.selected_cost.as_ref(),
-                replacement.benefit.as_ref(),
-            ] {
-                let annotation = annotation.expect("every replacement carries all three costs");
-                assert!(
-                    annotation.value.is_none(),
-                    "--default-cost must not export a value: {annotation:?}"
-                );
-                assert_eq!(annotation.source, CostSource::Unavailable);
-            }
-            assert!(
-                replacement.cost.is_nan(),
-                "the legacy scalar must stay unusable, not fall back to a structural count"
-            );
-        }
-
-        // Absent a value, the per-query aggregation `main` runs degrades to
-        // an unavailable summary rather than a number or an error.
-        for (_, dag) in &results.post_dags {
-            for (_, baseline, selected) in decision_cost_entries(dag) {
-                assert_eq!(baseline.source, CostSource::Unavailable);
-                assert_eq!(selected.source, CostSource::Unavailable);
-            }
-        }
-    }
-
-    /// Regression test for a real stack overflow found via manual testing
-    /// against real corpus queries (a `STDDEV_POP` aggregate, which — like
-    /// `AVG` — dispatches to `Realization::PassThrough` with no
-    /// alternative strategy of its own, so its *only* candidate is
-    /// `retain_exact`'s conservative fallback: the *entire target* itself,
-    /// unbound, carrying only an exact "kept pre-ASAP" guarantee).
-    /// `run_post_asap` must not treat that as a real winner: under the old
-    /// IR, splicing it into `export_post_asap` recursed forever (the spliced
-    /// payload structurally *was* the enclosing target, so every fresh
-    /// `find_winner` call found the identical winner again). Filtering this
-    /// shape out of `winners` (same "no-op candidate" concept
-    /// `explanation.rs`'s own `sketch_finding_reason` already excludes from
-    /// being a finding) is what keeps this terminating and keeps the output
-    /// free of a fake replacement: this test asserts both that
-    /// `run_post_asap` returns at all and that it reports nothing.
-    #[tokio::test]
-    async fn post_asap_does_not_recurse_forever_on_a_trivial_retain_exact_winner() {
-        let cat = default_catalog();
-        let stddev_query = lower_sql(
-            "SELECT STDDEV_POP(latency) FROM metrics",
-            &cat,
-            AccuracyTarget::Epsilon(0.01),
-        )
-        .await
-        .unwrap();
-        let lowered_queries = vec![(
-            "stddev".to_string(),
-            "SELECT STDDEV_POP(latency) FROM metrics".to_string(),
-            stddev_query,
-        )];
-
-        let results = run_post_asap(&lowered_queries);
-
-        // A trivial retain_exact winner must be filtered before it ever
-        // becomes a flat `TargetReplacement` — there's no real replacement
-        // to report for a target with no alternative at all.
-        assert!(
-            results.replacements.is_empty(),
-            "a target whose only candidate is the trivial retain_exact fallback \
-             shouldn't produce a flat replacement entry: {:?}",
-            results
-                .replacements
-                .iter()
-                .map(|(name, r)| (name.clone(), r.strategy.clone()))
-                .collect::<Vec<_>>()
-        );
-        assert_eq!(results.post_dags.len(), 1);
-        assert!(!results.post_dags[0].1.nodes.is_empty());
-    }
-}
diff --git a/crates/devtools/src/bin/sketch_coverage.rs b/crates/devtools/src/bin/sketch_coverage.rs
index 9d39c54a..3a3e604c 100644
--- a/crates/devtools/src/bin/sketch_coverage.rs
+++ b/crates/devtools/src/bin/sketch_coverage.rs
@@ -16,8 +16,7 @@
 // `--epsilon <f64>` (default 0.01) sets the `AccuracyTarget` every query in
 // every corpus lowers with. Without an approximate target,
 // `ASAPStrategies` never has a genuine sketch alternative to
-// report — see `dag_export`'s own `--epsilon` doc comment for the same
-// point, made there per-query instead of per-run.
+// report.
 //
 // This is a workload-level count (`explain_replacements` runs
 // `search_workload` once per corpus, over every query in it together), not
diff --git a/docs/develop_docs/physical-handoff-costs.md b/docs/develop_docs/physical-handoff-costs.md
index 1ccf3a2a..e830b1d1 100644
--- a/docs/develop_docs/physical-handoff-costs.md
+++ b/docs/develop_docs/physical-handoff-costs.md
@@ -12,8 +12,7 @@ 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 `PhysicalHandoffProfile` in the
-immutable `PhysicalEvidenceSnapshot`. `dag_export --planner-cost-json` accepts
-the same profile in a top-level `boundaries` field. With no profile, these
+immutable `PhysicalEvidenceSnapshot`. With no profile, these
 dimensions remain unestimated and the existing resource objective is preserved.
 
 The profile's `plans` list binds handoffs to complete physical alternatives.
@@ -95,6 +94,5 @@ Verification:
 
 ```sh
 cargo test -p asap-plan-selection --test physical_handoff_cost
-cargo test -p asap-devtools --bin dag_export handoff_bytes_export_and_change_plan_selection
 python3 -m unittest discover -s tools/dag-viewer -p 'test_render.py'
 ```
diff --git a/docs/develop_docs/storage-operation-costs.md b/docs/develop_docs/storage-operation-costs.md
index e9f1f0e3..5d3cc715 100644
--- a/docs/develop_docs/storage-operation-costs.md
+++ b/docs/develop_docs/storage-operation-costs.md
@@ -1,8 +1,7 @@
 # Storage operation estimates
 
 The physical-plan ranking adapter accepts an optional `StorageIoProfile` in
-its immutable `PhysicalEvidenceSnapshot`. `dag_export --planner-cost-json`
-accepts the same profile in the document's top-level `storage_io` field.
+its immutable `PhysicalEvidenceSnapshot`.
 Omitting the profile preserves the existing CPU/memory/scan-byte objective;
 operation counts are unestimated, not inferred to be zero.
 
@@ -84,5 +83,4 @@ Verification:
 ```sh
 cargo test -p asap-plan-selection --test storage_io
 cargo test -p asap-plan-selection --lib storage_io
-cargo test -p asap-devtools --bin dag_export storage_requests_export_and_change_plan_selection
 ```
diff --git a/docs/user_guide_docs/run-a-query.md b/docs/user_guide_docs/run-a-query.md
index 6b883dd0..2efe403c 100644
--- a/docs/user_guide_docs/run-a-query.md
+++ b/docs/user_guide_docs/run-a-query.md
@@ -1,6 +1,6 @@
 # ASAPPlanner CLI user guide
 
-Use the `asap-devtools` commands to inspect query IR, export DAGs, and inspect
+Use the `asap-devtools` commands to inspect query IR, plan workloads, and inspect
 corpus coverage. These commands do not deploy or execute a physical plan.
 
 To develop an application using the Rust library, start with
@@ -17,8 +17,6 @@ tool on first use.
 | --- | --- | --- |
 | `show_pre_asap_ir --data-ingestion-interval-ms 1000 queries.txt` | File path, or stdin when omitted | Prints canonical Pre-ASAP IR |
 | `show_post_asap_ir --data-ingestion-interval-ms 1000 queries.txt` | Same query file format | Prints all sketch-strategy Post-ASAP candidates using a fixed approximate target, in cost-model order |
-| `dag_export --data-ingestion-interval-ms 1000 --promql "<query>"` | One PromQL expression | Exports a query DAG for inspection |
-| `dag_export --sql "<query>"` | One SQL expression using the tool's catalog | Exports a query DAG for inspection |
 | `analyze_corpora --corpora --data-ingestion-interval-ms 1000 --out-dir <dir>` | Repository PromQL corpora, output directory | Writes successful/error IR dumps and summary reports |
 | `analyze_corpora --sql-corpora --out-dir <dir>` | Repository SQL corpora, output directory | Writes SQL corpus reports |
 | `variant_coverage --data-ingestion-interval-ms 1000` | Repository corpora | Reports Pre-ASAP IR variant coverage |
@@ -118,20 +116,6 @@ certified sketch. The tool prints plans, not query results.
 
 ## More inspection commands
 
-### Export a query DAG
-
-Export the IR of SQL or PromQL queries as JSON:
-
-```sh
-cargo run -p asap-devtools --bin dag_export -- --sql "<SQL query>"
-```
-
-or:
-
-```sh
-cargo run -p asap-devtools --bin dag_export -- --data-ingestion-interval-ms 1000 --promql "<PromQL query>"
-```
-
 ### See how the planner plans a workload
 
 ```sh