nordisk/facett

The pipeline DAG subsystem — guide

Class: guide · audience: a techie wiring an engine into facett-pipeline or driving either DAG viewer from a host / test. The what is in pipeline-design.md; the why in pipeline-reasoning.md. This file is the workflow.

f0 — zero to a driven DAG (the whole workflow)

Both viewers follow the same shape: build a model → build the view → drive it with Msgs → read state_json/state(). The only thing that differs is where the model comes from:

1. facett-pipeline: implement PipelineAdapter (or build a PipelineModel by hand) → PipelineView::from_adapter(&adapter) / from_model(model).
2. facett-plandag: build a PlanModel directly (or start from PlanDagView::local(), a working demo) → PlanDagView::new(title, model).

Either view is then a normal facett_core::Elm component: paint it inside any egui::Ui, or drive it headless through facett_core::harness with no window and no GPU.

f1 — writing a PipelineAdapter

use facett_pipeline::model::{PipelineAdapter, PipelineModel, Stage, StageStatus, PipelineEdge, PipelineMetrics};

struct MyEngineAdapter<'a> { run: &'a MyEngineRun }

impl PipelineAdapter for MyEngineAdapter<'_> {
    fn to_pipeline_model(&self) -> PipelineModel {
        let mut m = PipelineModel::new(&self.run.job_name);
        for stage in self.run.stages() {
            m = m.with_stage(
                Stage::new(&stage.name, my_status_to_stage_status(stage.status))
                    .with_kind(&stage.output_kind)
                    .with_rows(stage.rows_out),
            );
        }
        for edge in self.run.edges() {
            m = m.with_edge(PipelineEdge::new(&edge.from, &edge.to).with_label(edge.throughput_label()));
        }
        m.with_metrics(PipelineMetrics { tasks_done: self.run.tasks_done, tasks_total: self.run.tasks_total, ..Default::default() })
    }
}

Stage ids are the edge endpoints — use a stable id distinct from the display name if your engine has opaque node ids (Flink vertex ids, etc). Any edge naming an id not present in stages is silently dropped. The worked references are knut_pipelines::FacettPipelineAdapter (feature facett-adapter, see knut-pipelines), RunaPipelineAdapter (src/runa_adapter.rs) and PromoteLadderAdapter (src/promote_adapter.rs).

Build the view directly from the adapter:

let view = facett_pipeline::PipelineView::from_adapter(&MyEngineAdapter { run: &run });

The one rule you must get right: Idle vs Unknown

StageStatus::Idle means "we know it has not started." It is a fact you are asserting on the engine's behalf.

StageStatus::Unknown(reason) means "we could not determine this node's state." It is the absence of a fact.

If your probe timed out, your credentials were rejected, your sources disagreed, or you simply never asked — that is Unknown, not Idle. Reporting it as Idle paints a calm neutral grey box over something nobody has looked at: a fake green. (That exact pattern has been found repeatedly in this codebase — a DHCP request read as a lease, a handshake initiation read as a completed handshake.)

use facett_pipeline::{StageStatus, UnknownReason};

let status = match self.probe(&node) {
    Ok(Some(s))               => map_engine_status(s),
    Ok(None)                  => StageStatus::Unknown(UnknownReason::NotProbed),
    Err(e) if e.is_timeout()  => StageStatus::Unknown(UnknownReason::Unreachable),
    Err(e) if e.is_auth()     => StageStatus::Unknown(UnknownReason::NotAuthorised),
    Err(_)                    => StageStatus::Unknown(UnknownReason::ProbeFailed),
};

The reasons are NotProbed (default) · Unreachable · ProbeFailed · NotAuthorised · Ambiguous · Stale · NotApplicable. Unknown renders amber with a dashed border and a ? marker — never the neutral grey of Idle. status.label() gives "unknown:unreachable"; status.family() gives "unknown"; reason.token() / reason.describe() give the machine token and a human sentence.

Reading counts back: PipelineModel::status_counts() / PipelineView::counts() keep their legacy (idle, running, done, failed) shape and therefore drop the unknowns. Use status_tally() / tally() → StatusTally { idle, running, done, failed, unknown } whenever the numbers have to add up.

Both StageStatus and Status are #[non_exhaustive] — always write a _ => arm when you match one.

f1b — rendering a promotion ladder (PromoteLadderAdapter)

A promotion ladder is a DAG with per-node status, so it renders on this same viewer. You do not write an adapter for it — you fill in PromotionLadder and hand it over.

use std::collections::HashMap;
use facett_pipeline::{
    PipelineView, PromoteLadderAdapter, PromotionLadder, Promotion, Rung, RungStatus,
    SlotRole, UnknownReason,
};

// 1. The topology — rungs and the order they promote in. These names are YOURS.
let ladder = PromotionLadder::linear("acme", vec![
    Rung::new("dev",     SlotRole::Test),
    Rung::new("qa",      SlotRole::Test),
    Rung::new("staging", SlotRole::Preprod),
    Rung::new("prod",    SlotRole::Prod).guarded(true).with_release("v2.4.1"),
]);
// (or `PromotionLadder::new(..).with_rung(..).with_promotion(Promotion::new("a","b"))`
//  when the ladder branches — one rung may promote into several.)

// 2. What you actually OBSERVED. A rung you did not probe simply stays out.
let observed = HashMap::from([
    ("dev".to_string(),     RungStatus::Promoted),
    ("qa".to_string(),      RungStatus::Promoted),
    ("staging".to_string(), RungStatus::Promoting),
]);

let view = PipelineView::from_adapter(
    &PromoteLadderAdapter::new(&ladder)
        .with_observations(observed)
        // optional: say WHY the rest are unknown, when it was not "never asked"
        .unobserved_reason(UnknownReason::NotAuthorised),
);

prod is absent from the map, so it renders amber, dashed, ? with "unknown:not-authorised" — not a grey "idle" box. That is the whole point:

RungStatus::default() is Unknown(NotProbed) on purpose: the cheapest thing to write is the honest thing. Note this is the opposite default from RunaPipelineAdapter::with_status, where a missing stage is Idle — there the engine owns the run and knows it has not started; here the rung is a live environment that exists whether or not you looked.

Rung::role / guarded / release become the node's sublabel ("prod · guarded · v2.4.1"), readable in the detail panel and in state_json["nodes"][i]["detail"].

f2 — driving PipelineView

Builders (chainable, additive over the legacy new/from_nodes surface):

use facett_pipeline::{PipelineView, JobProgress};

let view = PipelineView::from_model(model)
    .with_animation(true)     // flowing edges + status glow (default on)
    .with_speed(0.6)          // flow animation speed
    .with_progress(JobProgress { fraction: 0.5, tasks_done: 25, tasks_total: 50, ..Default::default() });

Paint it inside any egui::Ui — view.ui(ui) via the Facet bridge, or view.view(ui) directly if you want the returned Vec<Msg> yourself.

Drive it headless (no window, no GPU) through facett_core::harness:

use facett_core::harness;
use facett_pipeline::Msg;

let mut v = PipelineView::from_model(model);
let snap = harness::snapshot(&mut v, [Msg::SelectNode("clean".into())]);
assert_eq!(snap.selected.as_deref(), Some("clean"));

let effects = harness::drive(&mut v, [Msg::SelectNode("sink".into())]);
// effects == [Effect::PreviewRequest("sink".into())]
let name = v.take_preview_request(); // the host's fetch-and-preview mailbox

Camera gestures are Msgs too, so pan/zoom/fit are equally headless-drivable: v.fit_to(rect), v.zoom_to_cursor(k, anchor), v.pan_pixels(dx, dy) are thin wrappers over update(Msg::Fit/…). Pin the animation clock for reproducible frames: v.set_clock(2.0) / v.tick(0.5), or drive Msg::SetClock/Msg::Tick directly.

Read back what's on screen via facett_core::Facet::state_json(&view) — nodes[], statuses, progress, animated/speed, idle, clock_s, nav, and the composed dag block. A host that rebuilds PipelineView every frame from live data should carry the camera across rebuilds with view.set_camera(old.nav().clone(), old.is_fitted()), otherwise pan/zoom snaps back to a fresh fit each time.

f3 — driving PlanDagView

Build it from a PlanModel (initial plan, optionally .with_final(..) for an AQE re-plan), or start from the working demo:

use facett_plandag::PlanDagView;

let view = PlanDagView::local();               // self-contained AQE demo plan
// or:
let view = PlanDagView::new("my query", my_plan_model);

Drive the controls the header exposes, headless:

use facett_core::harness;
use facett_plandag::{Msg, CostMetric, Version};

let mut v = PlanDagView::local();
let snap = harness::snapshot(&mut v, [
    Msg::SetMetric(CostMetric::Memory),
    Msg::SetVersion(Version::Final),
    Msg::ToggleGroup("1".into()),
]);
assert_eq!(snap.metric, CostMetric::Memory);
assert_eq!(snap.collapsed, vec!["1".to_string()]);

Msg::CollapseAll / ExpandAll toggle every WholeStageCodegen group in the active version at once; Msg::Select(id) / ClearSelection drive the same selection a node click produces — select an id from v.resolved().nodes (the resolved, not raw, ids). harness::drive always returns [] here (Effect is uninhabited — this view does no I/O).

Read the resolved graph as data — the exact thing the painter draws — via view.resolved() (Resolved { nodes, edges }, each node carrying its heat in [0,1]) or facett_core::Facet::state_json(&view) for the same graph plus version/has_aqe/collapsed/groups/selected.

Persist/restore UI state across a session with portable_state() / load_state(&json) — round-trips metric, version, collapsed, selected without touching the underlying PlanModel.

Look & feel

Both views inherit facett's look automatically: every colour comes from the active Theme (facett_core::theme(ui)), so the same DAG renders correctly under cyberpunk_neon, sci_fi, etc. Use facett_core::harness::render_themed(&mut view, Theme::cyberpunk_neon()) to snapshot a themed render headlessly (RenderReport::drew() asserts it actually painted).