feat(dag-viewer): plan through stage_pipeline in the live viewer; delete dag_export

.github/workflows/rust.yml

@@ -8,7 +8,7 @@ on:
       - 'Cargo.toml'
       - 'Cargo.lock'
       - '.github/workflows/rust.yml'
-      # The types tests validate the viewer's exported-kind contract.
+      # The devtools viewer_contract test validates the viewer's operator-kind contract.
       - 'tools/dag-viewer/**'
   pull_request:
     types: [opened, synchronize, reopened, ready_for_review]
@@ -17,7 +17,7 @@ on:
       - 'Cargo.toml'
       - 'Cargo.lock'
       - '.github/workflows/rust.yml'
-      # The types tests validate the viewer's exported-kind contract.
+      # The devtools viewer_contract test validates the viewer's operator-kind contract.
       - 'tools/dag-viewer/**'
   workflow_dispatch:
 

crates/asap-aware-mapping/src/cost_model.rs

@@ -67,7 +67,35 @@ use crate::replacement::{
 
 // ── Recurring-cost vocabulary for mixed exact/summary plans (issue #171) ──
 
-pub use asap_types::cost::CostUnit;
+/// The unit a cost is expressed in. Costs in different units are never
+/// compared or aggregated.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
+pub enum CostUnit {
+    /// A recurring cost expressed per second (`cost_units / s`); it requires
+    /// an explicit recurrence interval or rate.
+    CostUnitsPerSecond,
+    /// A finite-run or one-shot total at some horizon.
+    CostUnits,
+}
+
+impl CostUnit {
+    /// Stable machine-readable name.
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Self::CostUnitsPerSecond => "cost_units_per_second",
+            Self::CostUnits => "cost_units",
+        }
+    }
+}
+
+impl std::fmt::Display for CostUnit {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            CostUnit::CostUnitsPerSecond => write!(f, "cost units/s"),
+            CostUnit::CostUnits => write!(f, "cost units"),
+        }
+    }
+}
 
 /// Who produced a set of [`ExactCompositionCostInputs`], and under which
 /// model version — carried into every composed decision's explanation and
@@ -1154,9 +1182,9 @@ mod tests {
     #[test]
     fn recurring_formulas_require_the_shared_rate_unit() {
         let mut inputs = known_inputs();
-        let shared: asap_types::cost::CostUnit = inputs.unit;
+        let shared: CostUnit = inputs.unit;
         assert_eq!(shared.as_str(), "cost_units_per_second");
-        inputs.unit = asap_types::cost::CostUnit::CostUnits;
+        inputs.unit = CostUnit::CostUnits;
         assert_eq!(read_operation_plan_cost_rate(&inputs), None);
         assert_eq!(maintenance_operation_plan_cost_rate(&inputs), None);
         assert_eq!(raw_recompute_cost_rate(&inputs), None);

crates/asap-aware-mapping/src/explanation.rs

@@ -233,12 +233,10 @@ pub enum ExplanationKind {
 /// [`crate::replacement::ReplacementSubDAG::rationale`]).
 ///
 /// `node_hash` is [`structural_hash`](asap_types::ir::cse::structural_hash)
-/// of the `TargetSubDAG`'s own `target` sub-DAG — the same function, on the
-/// same `Rc<OperatorNode>` shape, that [`asap_types::dag_export::DAGNode::hash`]
-/// is computed with. A downstream consumer that independently exported the
-/// same node (e.g. via `asap_types::dag_export::export`) can match
-/// this explanation to a `DAGNode` by first comparing hashes and then
-/// confirming structural equality with [`ReplacementExplanation::target`].
+/// of the `TargetSubDAG`'s own `target` sub-DAG. A downstream consumer that
+/// hashed the same node can match this explanation by first comparing hashes
+/// and then confirming structural equality with
+/// [`ReplacementExplanation::target`].
 #[derive(Debug, Clone, PartialEq)]
 pub struct ReplacementExplanation {
     pub kind: ExplanationKind,
@@ -304,8 +302,7 @@ fn findings_from_candidate_logical_asap_dags(
     space: &CandidateLogicalASAPDAGs<String>,
 ) -> Vec<ReplacementExplanation> {
     let locations = collect_locations(&space.roots);
-    // One cache for the whole pass, mirroring `dag_export::export`'s own
-    // `HashCache` reuse — this is a bottom-up pass over every discovered
+    // One cache for the whole pass — this is a bottom-up pass over every discovered
     // group, so amortizing the cache across groups (rather than resetting it
     // per group) is real, not just a micro-optimization.
     let mut hash_cache = HashCache::new();
@@ -626,27 +623,19 @@ mod tests {
     }
 
     /// `node_hash` must be the literal `structural_hash` a downstream
-    /// consumer would compute over the *same* `OperatorNode` sub-DAG via
-    /// `asap_types::dag_export::export` — the whole point of carrying it is
-    /// that two independent exports of the same DAG agree, with no
-    /// string-matching against `location` required.
+    /// consumer would compute over the *same* `OperatorNode` sub-DAG, with
+    /// no string-matching against `location` required.
     #[test]
-    fn node_hash_matches_dag_export_hash_for_the_same_sub_dag() {
+    fn node_hash_is_the_structural_hash_of_the_sub_dag() {
         let q = agg(vec![2], default_quantile(0.99), metric_scan(&["job"]));
-        let dag = asap_types::dag_export::export(&q);
-        let expected_hash = dag.nodes[dag.root as usize].hash;
+        let expected_hash = structural_hash(&q, &mut HashCache::new());
 
         let findings = explain_replacements(vec![("dashboard_p99", q)]).unwrap();
         let sketch = findings
             .iter()
             .find(|f| f.kind == ExplanationKind::SketchApproximation)
             .expect("expected a sketch finding");
-        assert_eq!(
-            Some(sketch.node_hash),
-            expected_hash,
-            "ReplacementExplanation::node_hash must match dag_export's DAGNode::hash \
-             for the same OperatorNode sub_dag"
-        );
+        assert_eq!(sketch.node_hash, expected_hash);
     }
 
     #[test]

crates/asap-aware-mapping/src/pass/mod.rs

@@ -35,7 +35,8 @@ use crate::per_computation_candidates::PerComputationCandidateError;
 use crate::stage3_plan_selection::{Selection, SelectionError};
 
 pub use stage1_to_stage3::{
-    run_stage1_to_stage3, LogicalCandidate, PrunedCandidate, Stage1ToStage3Pass, Stage1ToStage3Run,
+    run_stage1_to_stage3, run_stage1_to_stage3_from_pass1, LogicalCandidate, PrunedCandidate,
+    Stage1ToStage3Pass, Stage1ToStage3Run,
 };
 
 static DEFAULT_COST_MODEL: DefaultCostModel = DefaultCostModel;

crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs

@@ -120,11 +120,26 @@ pub fn run_stage1_to_stage3<Id: Clone + std::fmt::Debug>(
     models: PlanningModels<'_>,
 ) -> Result<Stage1ToStage3Run<Id>, PerComputationCandidateError> {
     let per_computation = generate_per_computation_candidates(roots)?;
-    let combinations = per_computation
-        .targets
-        .iter()
-        .map(|target| target.choices.candidates.len())
-        .fold(1usize, usize::saturating_mul);
+    Ok(run_stage1_to_stage3_from_pass1(
+        per_computation,
+        targets,
+        data,
+        models,
+    ))
+}
+
+/// [`run_stage1_to_stage3`] after Pass 1: enumerates every whole-workload
+/// candidate of `per_computation` (the Pass 1 inventory of the frontends'
+/// roots), so a caller can read the inventory, e.g. its
+/// [`combinations`](PerComputationCandidates::combinations), before
+/// enumerating.
+pub fn run_stage1_to_stage3_from_pass1<Id: Clone + std::fmt::Debug>(
+    per_computation: PerComputationCandidates<Id>,
+    targets: &[Option<AccuracyTarget>],
+    data: &DataWorkload,
+    models: PlanningModels<'_>,
+) -> Stage1ToStage3Run<Id> {
+    let combinations = per_computation.combinations();
     let query_of_target = query_of_target(&per_computation);
     let mut logical = Vec::new();
     let mut physical = Vec::new();
@@ -171,14 +186,14 @@ pub fn run_stage1_to_stage3<Id: Clone + std::fmt::Debug>(
         }
     }
     let selection = select_plan(&physical, targets, data, models);
-    Ok(Stage1ToStage3Run {
+    Stage1ToStage3Run {
         stage1_pass1_per_computation: per_computation,
         combinations,
         stage1_logical_asap_optimization: logical,
         stage2_physical_asap_optimization: physical,
         stage2_pruned: pruned,
         stage3_plan_selection: selection,
-    })
+    }
 }
 
 /// `query_of_target[t]`: the index of the query root that first reads target

crates/asap-aware-mapping/src/per_computation_candidates.rs

@@ -205,6 +205,17 @@ pub struct PerComputationCandidates<Id> {
     pub targets: Vec<ComputationTarget>,
 }
 
+impl<Id> PerComputationCandidates<Id> {
+    /// The number of whole-workload candidates: the size of the Cartesian
+    /// product of the targets' candidates, saturating at `usize::MAX`.
+    pub fn combinations(&self) -> usize {
+        self.targets
+            .iter()
+            .map(|target| target.choices.candidates.len())
+            .fold(1usize, usize::saturating_mul)
+    }
+}
+
 #[derive(Debug, Error)]
 pub enum PerComputationCandidateError {
     #[error(transparent)]

crates/asap-aware-mapping/src/replacement.rs

@@ -11024,8 +11024,7 @@ mod tests {
         let Replacement::SubDAG(node) = &candidate.replacement else {
             unreachable!()
         };
-        let exported = asap_types::dag_export::export_summary(node);
-        assert!(exported.nodes[exported.root as usize]
+        assert!(node
             .guarantee
             .as_ref()
             .is_some_and(ResultGuarantee::has_unknown));

crates/devtools/Cargo.toml

@@ -12,7 +12,7 @@ asap-frontend-promql = { path = "../frontend-promql" }
 asap-frontend-sql = { path = "../frontend-sql" }
 asap-aware-mapping = { path = "../asap-aware-mapping" }
 
-# 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" }