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

[zzylol]

Why

The dag-viewer's live path (server.py, generate-sample.sh, the editor's "Plan selected workload") still ran the legacy dag_export binary. That binary drives the legacy candidate search, not the stage pipeline on main (Stage 0 → Stage 3, run_stage1_to_stage3). Moving the viewer onto stage_pipeline leaves dag_export, asap_types::dag_export and asap_types::cost unused, so this PR deletes them.

What

• The viewer, server.py and generate-sample.sh plan through stage_pipeline and read only asap-stage-pipeline/v1 documents.
• stage_pipeline gains what the editor needs from dag_export:

dag_export capability	Now
ad-hoc PromQL (--promql, --epsilon, --data-ingestion-interval-ms)	stage_pipeline --promql … --epsilon … --interval-ms …
ad-hoc SQL over declared tables	new --sql … --table '<json>' (repeatable; no built-in tables)
mixed SQL + PromQL in one plan	--promql/--sql in any order, planned as one workload (run_planning_stages_over)
query names	--name sets workload.queries[].id (default q<n>)
pre-ASAP DAG	Stage 0 dag, still written when Pass 1 fails
post-ASAP DAG and per-node cost	the selected Stage 2 PhysicalASAPDAG with Stage 3 costs[id].per_node
baseline/benefit comparison	a Stage 3 candidate table: the selected candidate, the other costed ones, rejected and pruned candidates with reasons, and Stage 1 composition errors

• Deleted: crates/devtools/src/bin/dag_export.rs, crates/types/src/dag_export.rs, crates/types/src/cost.rs, tools/dag-viewer/post_asap_fixture.json, and the WorkloadDAG reader, multi-file merge and IR labeling in viewer.js/render.py. CostUnit moves to asap_aware_mapping::cost_model, with the same serde.
• Dropped (agreed): --planner-cost-json/--default-cost/--topk-margin-json (evidence-driven Stage 3 cost is Calibrate plan selection's costs from measured sketch costs #679/Stage 3: evidence-driven cost (storage I/O, hand-off bytes) #615), the pre-ASAP replacement badges, --progress, and the old export format.
• Absorbs chore(devtools): delete the dag_export bin #616, chore(types): delete the dag_export module #617 and the server/generate-sample/editor parts of feat(dag-viewer): redesign the viewer around the stage pipeline #574.

How

• Combination limit: stage_pipeline refuses a workload with more than --max-combinations (default 10^4) whole-workload candidates before Stage 1 enumerates any. It exits with status 3 and writes nothing; results are never truncated. To make the check possible, the library splits at Pass 1 (run_stage1_to_stage3_from_pass1, PerComputationCandidates::combinations). --max-displayed only caps the display.
• Input validation: --table validates time_index and rejects duplicate tables. Duplicate query names are rejected; the error says when the name is a default.
• server.py: /api/plan serializes requests, validates the editor payload (types, unique table and query names), runs stage_pipeline (honoring CARGO_TARGET_DIR), and maps the combination-limit refusal to HTTP 422. /api/example serves dag.example.json.
• Viewer: the new DOM-free stage-workload.js turns a stage document into per-query pre (Stage 0) / post (selected plan) DAGs. Shared nodes merge by stage node id, and standalone scalar queries come from scalar_roots.
• Sample: generate-sample.sh regenerates dag.example.json from the editor's default 3 SQL + 2 PromQL workload. Stage 3 selects among all 144 candidates.

Key code interfaces added / modified

crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs, per_computation_candidates.rs, cost_model.rs

// new: Stage 1 to Stage 3 after Pass 1, so a caller can check the count before enumerating
/// Enumerates, composes, implements and selects among every whole-workload
/// candidate of a Pass 1 inventory; use it to inspect the inventory first.
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>;
/// Runs Stage 1 to Stage 3 over the frontends' roots; `targets[i]` is the
/// accuracy requirement of `roots[i]`. Only a Pass 1 failure is an `Err`.
pub fn run_stage1_to_stage3<Id: Clone + std::fmt::Debug>(
    roots: Vec<(Id, QueryRoot)>,
    targets: &[Option<AccuracyTarget>],
    data: &DataWorkload,
    models: PlanningModels<'_>,
) -> Result<Stage1ToStage3Run<Id>, PerComputationCandidateError>;  // now runs Pass 1, then delegates

impl<Id> PerComputationCandidates<Id> {
    /// The number of whole-workload candidates: the size of the Cartesian
    /// product of the targets' candidates. Check it before enumerating.
    pub fn combinations(&self) -> usize;  // new: whole-workload candidates, saturating at usize::MAX
}

// moved here from asap_types::cost (was `pub use asap_types::cost::CostUnit`); same variants and serde
/// The unit a cost is in; different units never compare. The type stays, moved to this crate.
pub enum CostUnit {
    /// A recurring cost per second.
    CostUnitsPerSecond,   // unchanged
    /// A finite-run total.
    CostUnits,            // unchanged
}
/// Stable machine-readable name, e.g. `"cost_units"`.
impl CostUnit { pub fn as_str(self) -> &'static str; }

crates/types/src/lib.rs

// removed: pub mod cost;  (unused once dag_export is gone)
//   CostUnit, CostSource, BaselineRef, CostInput, CostAnnotation, UnitMismatch, WorkloadCostSummary,
//   benefit_ratio, total_cost, sum_workload_costs, workload_cost_summary
// removed: pub mod dag_export;  (the viewer no longer reads the WorkloadDAG format)
//   DAGNode, DAGNote, DAGDecision, EdgeCostAnnotation, ExportDAG, NamedDAG, WorkloadDAG, SummaryDAGNode,
//   SummaryDAG, TargetReplacement, TargetReplacementAfter, TargetRejection, PostAsapSubstitution,
//   pub fn export(root: &Rc<OperatorNode>) -> ExportDAG, export_summary, export_post_asap

crates/devtools/src/planning_stages.rs

/// Why `run_planning_stages` produced no run.
pub enum PlanningStagesError {                                      // new
    /// Pass 1 failed on the frontends' roots, so there are no candidates;
    /// `stage_pipeline` still writes a document carrying the error.
    Pass1(String),
    /// The workload has more whole-workload candidates than the caller's
    /// limit; refused before any is enumerated.
    TooManyCombinations { combinations: usize, limit: usize },     // checked after Pass 1
    /// The frontends' roots or a composed candidate failed validation.
    Validation(String),
}
/// The error's display text, for callers that still return `String`.
impl From<PlanningStagesError> for String;

/// Plans a lowered workload with the built-in models, selecting the cheapest
/// candidate; `max_combinations` refuses a larger workload (`None`: no limit).
pub fn run_planning_stages(
    workload: &PlanningWorkload,
    roots: Vec<QueryRoot>,
    max_combinations: Option<usize>,                                // new
) -> Result<PlanningStagesRun, PlanningStagesError>;               // was Result<_, String>

// new: plans roots lowered from any mix of languages as one workload
/// Like `run_planning_stages`, with explicit targets: query `i` is
/// `roots[i]` with accuracy target `targets[i]`.
pub fn run_planning_stages_over(
    roots: Vec<QueryRoot>,
    targets: &[AccuracyTarget],
    data: &DataWorkload,
    max_combinations: Option<usize>,
) -> Result<PlanningStagesRun, PlanningStagesError>;

// new: SQL lowering for `stage_pipeline --sql`
/// Lowers a SQL workload over the tables of `catalog`; `Err` describes a
/// query that does not lower.
pub fn lower_sql_roots(
    workload: &PlanningWorkload,
    catalog: &asap_frontend_sql::SqlCatalog,
) -> Result<Vec<QueryRoot>, String>;

crates/devtools/src/bin/stage_pipeline.rs, bin/dag_export.rs (removed)

/// The most whole-workload candidates a run enumerates unless
/// `--max-combinations` says otherwise.
const MAX_COMBINATIONS: usize = 10_000;  // new: default for --max-combinations; exceeding it exits with status 3
// --sql adds a SQL query to the workload; --name names the query just before it;
// --table declares a table the SQL queries read; --max-combinations caps candidate enumeration.
// new flags: --sql <query> (repeatable, mixes with --promql in workload order), --name <name> (default q<n>),
//            --table '{"name", "columns": [{"name", "type", "nullable"}], "time_index"}' (repeatable),
//            --max-combinations <n>
/// Builds the SQL catalog from the `--table` declarations; `Err` for a
/// malformed or duplicate table.
fn sql_catalog(tables: &[String]) -> Result<SqlCatalog, String>;  // new
/// The `asap-stage-pipeline/v1` document for a completed run, showing the
/// first `max_displayed` candidates plus the selected one.
fn stage_pipeline(queries: Vec<Value>, run: &PlanningStagesRun, max_displayed: usize) -> Value;
// was fn stage_pipeline(workload: &PlanningWorkload, max_displayed: usize) -> Result<Value, String>
/// The document for a run's result; a Pass 1 failure still yields one, while
/// too many combinations or a failed validation yield a `Failure`.
fn document_of(                                                    // new: Pass 1 failure still yields a document
    queries: Vec<Value>,
    roots: &[QueryRoot],
    result: Result<PlanningStagesRun, PlanningStagesError>,
    max_displayed: usize,
) -> Result<Value, Failure>;
// removed: bin/dag_export, with --post-asap, --progress, --planner-cost-json (alias --analytical-cost-json),
//          --default-cost, --topk-margin-json, --table-schema (now --table) and
//          --data-ingestion-interval-ms (now --interval-ms)

asap-stage-pipeline/v1 document: workload.queries[].language is promql or sql and id is the --name; on a Pass 1 failure the document still carries the frontends' DAG and the Pass 1 error (shown):

{
  // Identifies the document; the viewer refuses any other format.
  "format": "asap-stage-pipeline/v1",
  // The queries planned, in workload order, with their requirements.
  "workload": { "queries": [{
    "id": "q1",                                   // --name, default q<n>
    "language": "sql",                            // "sql" or "promql"
    "text": "SELECT service, COUNT(*) FROM metrics GROUP BY service",   // the query as given
    "requirements": { "accuracy": { "epsilon": 0.01, "delta": 0.01 } }  // "exact" without --epsilon
  }] },
  // Stage 0: the query-language frontends' DAG over all queries.
  "stage0_query_language_frontends": { "dag": {
    // Flat nodes; each has operator, schema, result_kind, timing, materialization and guarantee.
    "nodes": [],
    // One root per query, by node index, e.g. [{"Operator": 2}, {"Operator": 6}].
    "roots": []
  } },
  // Stage 1: how many logical candidates were enumerated, whether the document
  // shows fewer, the shown ones, and a Pass 1 error when there are none.
  "stage1_logical_asap_optimization": { "combinations": 0, "capped": false, "candidates": [], "error": "Pass 1: ..." },
  // Stage 2: physical candidates and the logical ones pruned, with reasons.
  "stage2_physical_asap_optimization": { "candidates": [], "pruned": [] },
  // Stage 3: each shown candidate's cost, the selected id (null if none) and rejections.
  "stage3_plan_selection": { "costs": {}, "selected": null, "rejected": [] }
}

tools/dag-viewer/server.py, render.py

# server.py
# The stage_pipeline binary the server runs, under CARGO_TARGET_DIR (default target/).
BINARY = Path(os.environ.get("CARGO_TARGET_DIR", REPO / "target")) / "debug" / "stage_pipeline"  # was dag_export
# Builds the stage_pipeline binary with cargo before serving.
def build_planner() -> None: ...                        # renamed from build_exporter
# stage_pipeline refused the workload: more candidate combinations than it enumerates.
class PlanRejected(ValueError): ...                     # new: combination-limit refusal -> HTTP 422
# The stage_pipeline arguments for a planner request; ValueError for a malformed request.
def plan_args(payload: object, out: str) -> list[str]: ...  # new
# Serializes planning requests: stage_pipeline is CPU- and memory-heavy.
PLAN_LOCK = threading.Lock()                            # new: requests run one at a time
# Plans a request with stage_pipeline; raises PlanRejected on exit status 3.
def plan(payload: object) -> dict: ...                  # new: POST /api/plan returns the stage document
# removed: def plan_workload(payload: dict) -> dict

# render.py
# The only document format the viewer reads.
STAGE_PIPELINE_FORMAT = "asap-stage-pipeline/v1"        # new
# Returns the document if it is a stage-pipeline document; ValueError otherwise.
def check_document(document: object) -> dict: ...       # new
# Reads and checks a stage-pipeline document from a file.
def load_document(path: Path) -> dict: ...              # was load_workload(paths: list[Path]) -> dict
# Bakes the document into one self-contained viewer HTML page.
def render(document: dict) -> str: ...                  # was render(workload: dict)
# removed: prepare_workload, _semantic_label and the WorkloadDAG label helpers

tools/dag-viewer/stage-workload.js (new), viewer.js

// stage-workload.js (new)
// The document format this module reads.
const STAGE_PIPELINE_FORMAT = 'asap-stage-pipeline/v1';
// Whether `doc` is a stage-pipeline document.
function isStagePipelineDocument(doc);
// The viewer's workload: one entry per query, each node labeled with its Stage 3 cost.
function stageWorkload(doc);          // per-query Stage 0 and selected-plan DAGs
// Selected, costlier, rejected, pruned and composition-error candidates, in that order.
function stageCandidateRows(doc);     // Stage 3 candidate table rows
// Table header: candidates enumerated, how many shown, and any Pass 1 or selection error.
function stageCandidateHeader(doc);

// viewer.js
// Shows a stage-pipeline document, replacing what is loaded, with every query selected.
function showDocument(doc);           // replaces loadWorkload(parsed)
// Builds one Pre/Post-ASAP lane (compound parent plus nodes and edges) from a stageWorkload DAG.
function laneElements(laneId, laneLabel, dag, query, stage);  // laneCost parameter removed
// removed: formatCostUnit, formatBaselineRef, compactCost, laneCostLabel, renderCostAnnotation,
//          renderDecisionCostBlock, nodeCostBadgeSuffix, computeSelectionWorkloadCost, translationsForNode

Usage, end to end (run at 9ead2a5a): the editor's "Plan selected workload" and the CLI take the same path. What the user writes is one SQL and one PromQL query, the SQL table, and the accuracy; server.py's /api/plan turns the editor's request into exactly these stage_pipeline arguments (plan_args).

# --table: the SQL frontend's catalog (no built-in tables any more); the non-null columns let COUNT(*) plan.
# --sql/--promql, each followed by --name: the queries in workload order, planned as ONE workload.
# --epsilon/--delta: every query's accuracy target (ε = 0.01, δ = 0.01).
# --max-combinations (default 10^4): refuse a workload with more Stage 1 candidates before enumerating any.
stage_pipeline \
  --table '{"name": "metrics", "columns": [{"name": "ts", "type": "timestamp", "nullable": false},
            {"name": "service", "type": "utf8", "nullable": false}], "time_index": 0}' \
  --sql "SELECT service, COUNT(*) FROM metrics GROUP BY service" --name svc_counts \
  --promql "quantile(0.99, rate(x[5m]))" --name p99 \
  --epsilon 0.01 --delta 0.01 --out run.json

Inside stage_pipeline (bin/stage_pipeline.rs), each value is used once:

// `roots`: Stage 0, the SQL and PromQL frontends' roots in workload order (svc_counts, p99);
// written to the document as Stage 0's DAG even when Pass 1 fails, and the input to every later stage.
let roots = /* lower_sql_roots for --sql, lower_promql_roots for --promql, interleaved */;
// `result`: Stage 1 → 3 over `roots` with each query's target and the data workload (`--interval-ms`):
//   Ok(run)                              → the document below
//   Err(Pass1(e))                        → a document with Stage 0 and Stage 1's `error`
//   Err(TooManyCombinations { 12, 10 })  → with --max-combinations 10: exit 3, nothing written
let result = run_planning_stages_over(roots.clone(), &targets, &data, Some(max_combinations));
// `described`: the queries as the document lists them (id from --name, language, text, accuracy).
// `document`: the asap-stage-pipeline/v1 JSON written to --out; the viewer reads only this format.
let document = document_of(described, &roots, result, max_displayed)?;

What the user gets in run.json (and in the editor):

format                                   asap-stage-pipeline/v1
workload.queries[].id                    svc_counts, p99
stage1_logical_asap_optimization         combinations 12, candidates L1..L12 (Q1 exact/Cms/CountSketch/UnivMon × Q2 exact/Kll/DDSketch)
stage3_plan_selection.selected           P1 (all exact), 2.1263 cpu_ms_per_workload_evaluation
stage3_plan_selection.rejected           P2..P12 valid, "costlier: 2.126 vs 2.126 …" up to "costlier: 7.046 vs 2.126 …"

The viewer then reads the document with no other input:

// `workload`: one entry per query: its Stage 0 (pre-ASAP) DAG and the selected P1's (post-ASAP) DAG,
// each node labeled with its Stage 3 per-node cost; drawn as the Pre/Post-ASAP lanes.
const workload = stageWorkload(doc);
// `rows`: the Stage 3 candidate table: P1 selected, then P2..P12 costlier, then rejected, pruned and
// composition-error candidates (none here); shown under the lanes.
const rows = stageCandidateRows(doc);

Before this PR

editor / server.py → dag_export --post-asap (legacy search) → WorkloadDAG JSON → viewer

After this PR

editor / server.py → stage_pipeline (Stage 0 → Stage 3) → asap-stage-pipeline/v1 → viewer
                     pre-ASAP = Stage 0, post-ASAP = selected plan + Stage 3 per-node cost

Verification

• GitHub CI on head 3cb19154: test and format-and-lint passed.
• Run locally:
  • cargo fmt --check and cargo clippy -D warnings clean.
  • cargo nextest run --workspace: 1556 passed, 50 skipped. Includes the 3^9 combination limit, duplicate tables and names, time_index checks, the Pass 1 failure document, and that examples/scalar-queries.json is current.
  • test_render.py: 27 OK, stage-document only, including scalar-query placement and the candidate table and header.
  • test_server.py: 11 OK, including mixed-language order, request validation, one request at a time, and the 422 refusal.
  • verify_metricsql_external_consumer.sh OK. The Example 1 fixture regenerates byte-identical.
• Screenshots: not attached.

Limitations and follow-ups

• Markdown docs that still mention dag_export/asap_types::cost (viewer README, run-a-query, several develop and design docs) are left for Update the docs after the G1–G11 consolidation groups land #677.
• The rest of the viewer redesign is left for G11: feat(dag-viewer): redesign the viewer around the stage pipeline #574's stage lanes, candidate ranking and examples manifest, and fuller SQL editor support (feat(dag-viewer): plan SQL queries over declared tables in the editor #618). Only the selected candidate is drawn.
• Stage 3 rejections and Stage 1 errors still name queries by position (q5), not by editor name.

Human review — do not complete with an agent

• The MVP boundary is correct.
• New conceptual layers or public interfaces are necessary.
• The before/after description matches the intended product behavior.
• Human reviewer:
• Decision and rationale: