facett-pipeline/src/lib.rs
//! **facett-pipeline** — a dataflow / Spark **pipeline DAG** viewer + job
//! progress, brought up to the facett look: **theme-driven** (every colour from
//! the active [`Theme`]), **animated flowing edges** (a marching-ants sweep along
//! each edge showing dataflow direction), **per-stage status glow** (running =
//! pulse, done = steady accent, failed = red bloom via
//! [`effects::glow_rect`]), and **throughput labels** on edges.
//!
//! It uses shared layered layout and Falsett (`facett-flow`) for persistent
//! geometry, GPU rendering, CPU fallback and throughput packets.
//! A [`Facet`], and now **engine-agnostic**: the canonical input is the neutral
//! [`PipelineModel`] (see [`model`]), produced by an engine's
//! [`PipelineAdapter`] — so "show another engine's
//! pipeline" is *a new adapter, not a new viewer* (one-viewer-many-adapters, like
//! facett-graphview). knut ships the Spark/SDP adapter; build with
//! [`PipelineView::from_model`] / [`from_adapter`](PipelineView::from_adapter).
//! The future graphview render path + extra engine adapters are DEFERRED — see the
//! [`model`] design note.
//!
//! ## Back-compat
//! The original surface is unchanged — [`PipelineView::new`],
//! [`with_progress`](PipelineView::with_progress), [`counts`](PipelineView::counts),
//! [`Status`], [`JobProgress`] all behave as before. The look/animation is
//! **additive**: themed + animated **on by default**, switch off with
//! [`with_animation(false)`](PipelineView::with_animation); edge throughput labels
//! are opt-in via [`with_throughput`](PipelineView::with_throughput).
//!
//! ## FC contract (the canonical Elm split, FC-2 / FC-9)
//! - **[`PipelineState`]** — the complete observable, serializable, round-trippable
//! state: the rich per-stage node model, the edge topology, the job progress, the
//! animation flags/speed, the throughput labels, the name-keyed selection (FC-5),
//! the injected clock (FC-7), and the shared pan/zoom camera + its fitted flag.
//! [`PipelineView::state`] hands back a `&PipelineState`.
//! - **[`Msg`] + [`PipelineView::update`]** — the single mutation path (FC-2): the
//! same messages a node click / camera gesture / fit button produce, so a headless
//! driver ([`facett_core::harness`]) reaches them too.
//! - **[`PipelineView::view`]** — a **pure** paint (FC-9): reads `&self`, paints the
//! themed animated DAG, and *returns* the [`Msg`]s the interactions produced; the
//! [`impl_facet_via_elm!`](facett_core::impl_facet_via_elm) bridge applies them.
//! - **[`Effect::PreviewRequest`]** — side work as data (FC-8): click-selecting a
//! stage asks the host to fetch + preview that dataset. The host drains it via
//! [`take_preview_request`](PipelineView::take_preview_request), and it is also the
//! value [`update`](PipelineView::update) returns so a `drive` test can assert it.
//! - **Rich `state_json`** — the per-node/status/progress/nav introspection a driver
//! reads is derived (partly from the composed `DepGraphView`), so the macro is
//! invoked in **form 3** (`custom_state_json`) to publish those exact keys instead
//! of a plain `serde(state())`.
use egui::{Align2, Color32, FontId, Pos2, Rect, Sense, Stroke, Vec2};
use facett_core::effects::{easing, glow_rect};
use facett_core::layered::{Align, Layered, LayeredOpts, RowOrder, layered, place};
use facett_core::nav::Navigable;
use facett_core::theme;
use facett_flow::attr::{Rates, Sizes, Status as FlowStatus, StatusTable};
use facett_flow::{Animation, CpuStyle, Edge, FlowScene, Graph, Node, PaintOpts, View};
use serde::{Deserialize, Serialize};
pub mod model;
/// The ONE JSON encoding of [`PipelineModel`] (both directions) — what a host posts
/// to the `facet-wrapped` wasm bundle and what knut streams over SSE.
pub mod wire;
pub use model::{
PipelineAdapter, PipelineEdge, PipelineMetrics, PipelineModel, Stage, StageMetrics,
StageStatus, StatusTally, UnknownReason,
};
// The **promotion-ladder** adapter: a slot ladder (`test2 → test1 → lab → prod`,
// or `dev → qa → staging → prod` — the rungs are DATA, never names baked in here)
// → the neutral `PipelineModel`, so a promotion ladder renders on THIS reused
// viewer. Unlike `runa-adapter` it is NOT feature-gated: that flag exists to keep
// an optional *dependency* (`dwarves-runa`) out of a default build, and the
// ladder adapter pulls in nothing at all — its input is a small local type the
// caller populates, so facett never depends on the promotion engine.
pub mod promote_adapter;
pub use promote_adapter::{
PromoteLadderAdapter, Promotion, PromotionLadder, Rung, RungStatus, SlotRole,
};
// The `runa` engine adapter (feature `runa-adapter`): a dwarves forge pipeline
// (`dwarves_runa::Pipeline`) → the neutral `PipelineModel`, so the forge's typed
// `*.runa.ron` pipeline renders on THIS reused viewer — "a new adapter, not a new
// viewer". The only place facett-pipeline touches the `runa` edda leaf.
#[cfg(feature = "runa-adapter")]
pub mod runa_adapter;
#[cfg(feature = "runa-adapter")]
pub use runa_adapter::RunaPipelineAdapter;
/// The render-layer amber for [`Status::Unknown`] — the constellation's drift/attention
/// amber (matching `facett-releaseview`'s `BumpDrift`), deliberately **not** the dim
/// grey of `Idle` and **not** the red of `Failed`. See [`Status::theme_color`].
pub const UNKNOWN_AMBER: Color32 = Color32::from_rgb(240, 180, 70);
/// A dataset/stage run status (generic — map your engine's status to this).
///
/// `#[non_exhaustive]`: matches must carry a `_ =>` arm, so the next variant is
/// not a breaking change. See [`StageStatus`] for the `Idle`-vs-`Unknown` rule.
#[non_exhaustive]
#[derive(Clone, Copy, PartialEq, Eq, Default, Debug, Serialize, Deserialize)]
pub enum Status {
#[default]
Idle,
Running,
Done,
Failed,
/// **State could not be determined**, with the machine-readable reason. The
/// render twin of [`StageStatus::Unknown`]; converts both ways with it.
Unknown(model::UnknownReason),
}
impl Status {
/// A reasonable fixed colour (kept for back-compat / non-themed callers).
/// The themed render uses [`theme_color`](Status::theme_color) instead.
pub fn color(self) -> Color32 {
match self {
Status::Done => Color32::from_rgb(80, 180, 120),
Status::Running => Color32::from_rgb(80, 140, 220),
Status::Failed => Color32::from_rgb(220, 80, 80),
Status::Idle => Color32::from_gray(130),
Status::Unknown(_) => UNKNOWN_AMBER,
}
}
/// The status colour **derived from the active [`Theme`]** — done/running ride
/// the palette's `accent`/`point`, failed is a warm red, idle is dimmed text.
/// This is what the themed DAG paints with.
///
/// ## Why `Unknown` is amber and not grey
///
/// `Idle` is `th.text_dim` — a *recessive* neutral that reads "nothing to see
/// here, not started". If `Unknown` shared it, an environment nobody has ever
/// looked at would render exactly like one we positively know is idle, and the
/// whole reason `Unknown` exists would be lost on screen. So `Unknown` is
/// **amber** (`UNKNOWN_AMBER`): an *attention* hue that says "resolve me",
/// borrowed from `facett-releaseview`'s drift colour so the constellation reads
/// consistently — and distinct from red, because unknown is not failure.
///
/// Amber is never the only channel: the node paints a **dashed** border (a
/// solid box asserts a known state; a broken one does not) and a **`?`** corner
/// marker, so the distinction survives greyscale and colour-vision deficiency
/// — the same belt-and-braces the failed node's `⚠` marker uses.
pub fn theme_color(self, th: &facett_core::Theme) -> Color32 {
match self {
Status::Done => th.point,
Status::Running => th.accent,
Status::Failed => facett_core::look::RED,
Status::Idle => th.text_dim,
Status::Unknown(_) => UNKNOWN_AMBER,
}
}
/// A short, stable label. For `Unknown` it **carries the reason**
/// (`"unknown:unreachable"`); [`family`](Status::family) gives the bare
/// `"unknown"` for legend chips.
pub fn label(self) -> &'static str {
match self {
Status::Idle => "idle",
Status::Running => "running",
Status::Done => "done",
Status::Failed => "failed",
Status::Unknown(r) => StageStatus::Unknown(r).label(),
}
}
/// The status family without the reason (`"idle"`/`"running"`/`"done"`/
/// `"failed"`/`"unknown"`).
pub fn family(self) -> &'static str {
match self {
Status::Unknown(_) => "unknown",
other => other.label(),
}
}
/// The reason, when this is [`Unknown`](Status::Unknown); `None` otherwise.
pub fn unknown_reason(self) -> Option<model::UnknownReason> {
match self {
Status::Unknown(r) => Some(r),
_ => None,
}
}
/// `true` for [`Unknown`](Status::Unknown), whatever the reason.
pub fn is_unknown(self) -> bool {
matches!(self, Status::Unknown(_))
}
}
/// Overall job progress (Spark stages/tasks).
///
/// The original three fields (`fraction`/`tasks_done`/`tasks_total`) are
/// unchanged; the **live-run** fields below are additive (default `0`) so a
/// consumer can surface real Spark throughput — rows processed, bytes read,
/// wall-clock elapsed, and a derived rows/sec — not just a task bar. All are
/// reflected in [`PipelineView::state_json`] for headless assertion.
#[derive(Clone, Default, Debug, PartialEq, Serialize, Deserialize)]
pub struct JobProgress {
pub fraction: f32,
pub tasks_done: u64,
pub tasks_total: u64,
/// Rows processed so far across the run (additive; `0` = unknown).
pub rows: u64,
/// Input bytes read so far across the run (additive; `0` = unknown).
pub bytes_read: u64,
/// Wall-clock seconds elapsed since the run started (additive; `0.0` = n/a).
pub elapsed_s: f32,
/// Tasks in flight right now (Spark `num_inflight_tasks`; additive).
pub inflight: u64,
}
impl JobProgress {
/// Live row throughput in rows/sec, derived from `rows / elapsed_s`. Returns
/// `0.0` when elapsed is non-positive (no rate yet). Pure + testable.
pub fn rows_per_s(&self) -> f32 {
if self.elapsed_s > 0.0 {
self.rows as f32 / self.elapsed_s
} else {
0.0
}
}
}
/// One pipeline stage/dataset node: its name + status, plus the optional live
/// detail a real Spark run surfaces — a sublabel (e.g. the output type
/// `materialized_view`), rows processed, and — for a [`Status::Failed`] node —
/// the **error message** to surface. Built from a knut `Dataset` + `NodeStatus`
/// by the consumer; the component owns no knut dependency.
#[derive(Clone, Default, Debug, PartialEq, Serialize, Deserialize)]
pub struct PipelineNode {
pub name: String,
pub status: Status,
/// Secondary line under the name (e.g. the output type). `None` falls back
/// to the status label.
pub detail: Option<String>,
/// Rows this node has materialised/processed (additive; `None` = unknown).
pub rows: Option<u64>,
/// Error message to surface on a failed node (drawn + emitted in state).
pub error: Option<String>,
/// Live per-node throughput in rows/sec (additive; `None` = unknown). When set
/// it draws under the status sub-label and rides `state_json`. Streaming
/// metrics (GAP #3) will populate this; the field + render path exist now.
pub throughput_rows_s: Option<f32>,
/// Live streaming lag/backlog in seconds (additive; `None` = unknown). Hook
/// for GAP #3 (watermark/backlog); rendered + emitted when present.
pub lag_s: Option<f32>,
}
impl PipelineNode {
/// A node with just a name + status (matching the old `(name, Status)` pair).
pub fn new(name: impl Into<String>, status: Status) -> Self {
PipelineNode {
name: name.into(),
status,
detail: None,
rows: None,
error: None,
throughput_rows_s: None,
lag_s: None,
}
}
pub fn with_detail(mut self, d: impl Into<String>) -> Self {
self.detail = Some(d.into());
self
}
pub fn with_rows(mut self, rows: u64) -> Self {
self.rows = Some(rows);
self
}
pub fn with_error(mut self, e: impl Into<String>) -> Self {
self.error = Some(e.into());
self
}
/// Attach a live per-node throughput (rows/sec). Drawn under the status line.
pub fn with_throughput_rows_s(mut self, r: f32) -> Self {
self.throughput_rows_s = Some(r);
self
}
/// Attach a live streaming lag/backlog (seconds). Drawn under the status line.
pub fn with_lag_s(mut self, l: f32) -> Self {
self.lag_s = Some(l);
self
}
}
/// How a single edge animates this frame — the **pure** core of the flowing-edge
/// effect, so the phase math is unit-testable without a painter.
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct EdgeFlow {
/// Marching-ants phase in `[0,1)`: the fractional offset of the dash pattern
/// along the edge, advancing with time × speed in the `from → to` direction.
pub phase: f32,
}
/// The marching-ants **phase** of an edge at wall-clock `time` (seconds) for a
/// given `speed` (cycles/sec). Pure + deterministic — `phase ∈ [0,1)`, increasing
/// with time, wrapping each cycle. `speed <= 0` freezes the flow at phase 0.
pub fn edge_phase(time: f64, speed: f32) -> f32 {
if speed <= 0.0 {
return 0.0;
}
(time * speed as f64).rem_euclid(1.0) as f32
}
/// The per-stage glow **intensity** in `[0,1]` for `status` at `time`. Running
/// pulses (eased sine), done holds a steady soft glow, failed blooms strongly,
/// idle doesn't glow. Pure, so the pulse curve is testable.
///
/// `Unknown` holds a **steady 0.5** — brighter than done, dimmer than failed, and
/// emphatically **not** idle's `0.0`. An un-glowing box is how the eye skips a
/// node; an unobserved rung is exactly the one that must not be skipped. It does
/// not pulse (pulsing means *activity*, and we have observed no activity).
pub fn status_glow(status: Status, time: f64, speed: f32) -> f32 {
match status {
Status::Running => {
let raw = ((time * speed.max(0.1) as f64 * 2.0).sin() as f32) * 0.5 + 0.5;
0.25 + 0.75 * easing::ease_in_out_cubic(raw)
}
Status::Done => 0.35,
Status::Failed => 0.9,
Status::Idle => 0.0,
Status::Unknown(_) => 0.5,
}
}
/// **The complete observable state (FC-1 / FC-3)** of a [`PipelineView`], in one
/// serializable, round-trippable struct. Everything a headless driver reads back is
/// here — or purely derived from here (the composed `DepGraphView` layout is a
/// function of `nodes` + `edge_ends`). [`PipelineView::state`] hands back a
/// `&PipelineState`; a driver snapshots it after feeding a `Vec<Msg>`.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct PipelineState {
/// The rich per-node model (name, status, detail, rows, error). The legacy
/// `statuses` view is derived from this.
pub nodes: Vec<PipelineNode>,
/// Edge endpoints, in the same order as the composed DAG's edges
/// (`(from_idx, to_idx)`). Kept so the animated painter can resolve flow
/// direction + labels.
pub edge_ends: Vec<(usize, usize)>,
/// Optional overall job progress (task bar + live-run readout).
pub progress: Option<JobProgress>,
/// Animate the flowing edges + status glow (on by default).
pub animated: bool,
/// Flow speed (marching-ants cycles/sec; also drives the running-pulse).
pub speed: f32,
/// Optional per-edge throughput label (e.g. "12k rows/s"), indexed like
/// `edge_ends`. Empty/`None` entries draw no label.
pub throughput: Vec<Option<String>>,
/// Name of the click-selected node, for the detail panel. Keyed on the stable
/// node **name** (FC-5), not a list index — reordering nodes can't re-point it.
pub selected: Option<String>,
/// **Injected animation clock** (seconds; FC-7). The flowing-edge phase + the
/// status-glow pulse are a pure function of this, so two renders from the same
/// state + clock produce identical frames + AccessKit. `None` = "follow ambient
/// `ui.input(i.time)`" (real-time UX); a test/host pins it via
/// [`set_clock`](PipelineView::set_clock) / [`tick`](PipelineView::tick).
pub clock_s: Option<f64>,
/// **Shared pan/zoom camera** (`facett_core::nav::Navigable`, GAP #1). The DAG
/// is laid out in *scene* coords (the fixed grid math) and painted through
/// `nav.to_screen`, so wheel zoom-to-cursor, drag-pan, keyboard pan/zoom and
/// fit-to-view all reuse the exact model the OSM 2D viewer drives. Emitted in
/// `state_json["nav"]` so navigation is headless-assertable.
pub nav: Navigable,
/// Set once the camera has been auto-fit to the DAG for the first paint, so we
/// frame the graph on open (like the OSM viewer) without re-fitting every frame.
pub fitted: bool,
/// The camera flight under way (🎯 Focus / ⤢ Fit), UI time — never serialized.
#[serde(skip)]
pub flight: Option<facett_core::nav::NavFlight>,
}
/// A robot-/CLI-addressable control message (FC-2) — the named boundary a headless
/// driver (or a host) drives the pipeline through, the same effect a node click /
/// camera gesture / fit button produces. Applied by [`PipelineView::update`].
#[derive(Clone, Debug, PartialEq)]
pub enum Msg {
/// Click-select a stage by its stable node **name** (FC-5). Raises a one-shot
/// [`Effect::PreviewRequest`] so the host can fetch that dataset's preview.
SelectNode(String),
/// Pan the camera by a screen-pixel delta (drag / arrows / two-finger).
Pan(Vec2),
/// Zoom the camera by `factor` (>1 zooms in) around a screen `anchor` point.
Zoom { factor: f32, anchor: Pos2 },
/// Fit the camera to the whole DAG within the given viewport `rect`.
Fit(Rect),
/// Ask for a re-fit on the next paint (the `⤢ Fit` button — the rect is only
/// known at paint time, so we defer the actual [`Msg::Fit`]).
RequestRefit,
/// 🎯 Focus: fly stage `name` to the centre of the visible canvas at a readable zoom.
FocusNode { name: String, now: f64 },
/// ⤢ Fit as a flight: the whole DAG, centred in the visible canvas.
FitFly(f64),
/// Advance a camera flight to `now`.
FlyTick(f64),
/// Pin the injected animation clock to an absolute time (seconds; FC-7).
SetClock(f64),
/// Advance the injected animation clock by `dt` seconds (init from 0 if unset).
Tick(f64),
}
/// **Side work as data (FC-8).** The pipeline is a read-only viewer with one honest
/// side effect: click-selecting a stage asks the host to **preview that dataset**
/// (fetch its table). The host drains it via
/// [`take_preview_request`](PipelineView::take_preview_request); it is also returned
/// from [`update`](PipelineView::update) so a `drive` test can assert the request.
#[derive(Clone, Debug, PartialEq)]
pub enum Effect {
/// The host should fetch + preview this dataset (by stage name).
PreviewRequest(String),
}
/// The pipeline view: a DAG + optional job progress, themed + animated.
pub struct PipelineView {
pub title: String,
/// All observable state (FC-3), in one serializable struct.
state: PipelineState,
/// The composed dependency-graph layout — **derived** from
/// `state.nodes` + `state.edge_ends` at construction, so it is not part of the
/// serializable [`PipelineState`] (like the matrix grid derived from its rows).
/// Reused for the legacy `state_json["dag"]` column summary.
layout: Layered,
scene: FlowScene,
/// Flow edge indices corresponding to valid public edge entries.
flow_edges: Vec<Option<usize>>,
/// One-shot "the host should preview this dataset" request, set when a node is
/// click-selected (the concrete side of [`Effect::PreviewRequest`]). The host
/// (Korp) drains it via [`take_preview_request`](Self::take_preview_request) and
/// fetches the table — keeps the component source-agnostic (owns no data dep).
/// Transient host-comms mailbox, deliberately outside the observable Model.
preview_request: Option<String>,
/// The optional **flow-sim overlay** (feature `flowsim`): edge particle advection
/// / static-arrows driven by per-edge throughput. `None` = the classic
/// marching-ants dashes. Enabled via [`with_flowsim`](Self::with_flowsim); the
/// host drives its mode/speed through [`flowsim_mut`](Self::flowsim_mut).
#[cfg(feature = "flowsim")]
flowsim: Option<facett_flowsim::FlowSim>,
/// The canvas rect of the last paint. The camera lives in SCREEN space
/// ([`Navigable`]), so when the canvas moves (a rail opens, the layout above it
/// changes) the picture follows it instead of staying where the canvas used to be.
track: facett_core::nav::CanvasTrack,
}
const NODE_W: f32 = 160.0;
const NODE_H: f32 = 56.0;
const COL_GAP: f32 = 220.0;
const ROW_GAP: f32 = 90.0;
const LEFT_PAD: f32 = 30.0;
const TOP_PAD: f32 = 40.0;
impl PipelineView {
/// `nodes` = `(dataset name, status)`; `edges` = `(reads_idx, target_idx)`.
/// Back-compat entry point; for the rich model (rows/error/detail) use
/// [`from_nodes`](Self::from_nodes).
pub fn new(
title: impl Into<String>,
nodes: Vec<(String, Status)>,
edges: Vec<(usize, usize)>,
) -> Self {
let rich = nodes
.into_iter()
.map(|(name, s)| PipelineNode::new(name, s))
.collect();
Self::from_nodes(title, rich, edges)
}
/// Build from the **rich** node model — each [`PipelineNode`] carrying its
/// status plus optional detail/rows/error. This is what a real Spark run maps
/// to (a knut `Dataset` + `NodeStatus` + the live stream), so the failed
/// node's error and per-node rows reach the view + its `state_json`.
pub fn from_nodes(
title: impl Into<String>,
nodes: Vec<PipelineNode>,
edges: Vec<(usize, usize)>,
) -> Self {
let edge_ends = edges;
let directed: Vec<_> = edge_ends
.iter()
.filter(|&&(a, b)| a < nodes.len() && b < nodes.len())
.map(|&(a, b)| (b as u32, a as u32))
.collect();
let layout = layered(
nodes.len(),
&directed,
LayeredOpts {
align: Align::Sources,
rows: RowOrder::Input,
},
);
let positions = place(
&layout,
[COL_GAP, ROW_GAP],
[LEFT_PAD + NODE_W / 2.0, TOP_PAD + NODE_H / 2.0],
);
let mut flow_edges = Vec::with_capacity(edge_ends.len());
let mut graph_edges = Vec::new();
for &(a, b) in &edge_ends {
if a < nodes.len() && b < nodes.len() {
flow_edges.push(Some(graph_edges.len()));
graph_edges.push(Edge {
from: format!("p{b}"),
to: format!("p{a}"),
});
} else {
flow_edges.push(None);
}
}
let graph = Graph {
nodes: nodes
.iter()
.enumerate()
.map(|(i, n)| Node {
id: format!("p{i}"),
label: n.name.clone(),
position: positions[i],
group: "pipeline".into(),
kind: 0,
})
.collect(),
edges: graph_edges,
};
let mut scene = FlowScene::with_key(graph, 0).expect("pipeline layout graph");
scene
.set_sizes(Sizes {
nodes: vec![[NODE_W, NODE_H]; nodes.len()],
})
.expect("pipeline sizes");
scene
.set_status(StatusTable {
nodes: nodes
.iter()
.map(|n| match n.status {
Status::Done => Some(FlowStatus::Ok),
Status::Failed => Some(FlowStatus::Error),
Status::Unknown(_) => Some(FlowStatus::Slow),
_ => None,
})
.collect(),
edges: vec![None; directed.len()],
})
.expect("pipeline statuses");
scene
.set_rates(Rates {
edges: edge_ends
.iter()
.zip(&flow_edges)
.filter_map(|(&(a, _), i)| {
i.map(|_| {
nodes[a]
.throughput_rows_s
.map(|v| if v.is_finite() { v.max(0.0) } else { 0.0 })
})
})
.collect(),
})
.expect("pipeline rates");
scene.set_arrows(true);
Self {
title: title.into(),
state: PipelineState {
nodes,
edge_ends,
progress: None,
animated: true,
speed: 0.6,
throughput: Vec::new(),
selected: None,
clock_s: None,
nav: Navigable::default(),
fitted: false,
flight: None,
},
layout,
scene,
flow_edges,
preview_request: None,
#[cfg(feature = "flowsim")]
flowsim: None,
track: facett_core::nav::CanvasTrack::default(),
}
}
/// Build the view from the **engine-agnostic [`PipelineModel`]** — the neutral
/// shape any engine's [`PipelineAdapter`] produces. This is the new canonical
/// entry point: the viewer is now *renderer-independent of the source engine*.
/// Stages → rich nodes (status/kind/error/metrics), the model's id-based
/// [`PipelineEdge`]s are resolved to index pairs (unknown ids dropped), edge
/// labels become throughput labels, and pipeline metrics → the progress bar.
/// The legacy [`new`](Self::new) / [`from_nodes`](Self::from_nodes) builders
/// still work and are unchanged.
pub fn from_model(model: PipelineModel) -> Self {
// Stages → the rich PipelineNode model the painter already renders.
let nodes: Vec<PipelineNode> = model
.stages
.iter()
.map(|s| PipelineNode {
name: s.name.clone(),
status: s.status.into(),
detail: s.kind.clone(),
rows: s.metrics.rows,
error: s.error.clone(),
throughput_rows_s: s.metrics.throughput_rows_s,
lag_s: s.metrics.lag_s,
})
.collect();
// Resolve id-based edges → index pairs in `(reads_idx, target_idx)` form
// (the painter's existing edge convention: `from` reads `to`, dataflow
// runs to→from). The model's edge is `from → to` source→consumer, so the
// painter pair is `(to_idx, from_idx)`. Edges naming an unknown stage id
// are dropped (an external upstream outside the rendered graph).
let mut edges = Vec::with_capacity(model.edges.len());
let mut throughput = Vec::with_capacity(model.edges.len());
for e in &model.edges {
match (model.stage_index(&e.from), model.stage_index(&e.to)) {
(Some(from_i), Some(to_i)) => {
edges.push((to_i, from_i));
throughput.push(e.label.clone());
}
_ => continue,
}
}
let mut view = Self::from_nodes(model.title, nodes, edges);
if throughput.iter().any(|l| l.is_some()) {
view = view.with_throughput(throughput);
}
if let Some(m) = model.metrics {
view = view.with_progress(JobProgress {
fraction: m.fraction.unwrap_or_else(|| {
if m.tasks_total > 0 {
m.tasks_done as f32 / m.tasks_total as f32
} else {
0.0
}
}),
tasks_done: m.tasks_done,
tasks_total: m.tasks_total,
rows: m.rows,
bytes_read: m.bytes_read,
elapsed_s: m.elapsed_s,
inflight: m.inflight,
});
}
view
}
/// Build the view straight from any [`PipelineAdapter`] — `adapter →
/// normalized model → view`. The one call a host makes regardless of engine.
pub fn from_adapter(adapter: &impl PipelineAdapter) -> Self {
Self::from_model(adapter.to_pipeline_model())
}
/// Pre-select a node by name (drives the detail panel + `selection_json`).
/// Unknown names clear the selection. Returns `self` for chaining.
pub fn with_selected(mut self, name: &str) -> Self {
self.state.selected = self
.state
.nodes
.iter()
.find(|n| n.name == name)
.map(|n| n.name.clone());
self
}
/// **Pin the animation clock** (seconds; FC-7). Once set, the flowing edges +
/// status glow are a pure function of this value — identical state + clock ⇒
/// identical frame + AccessKit tree. Pass it from an injected/host clock in a
/// test to make animation reproducible. Routes through the FC-2 [`update`](Self::update).
pub fn set_clock(&mut self, seconds: f64) {
let _ = self.update(Msg::SetClock(seconds));
}
/// Advance the injected clock by `dt` seconds (initialising from 0 if unset).
/// The host-driven analogue of [`Msg::Tick`] — steps the animation
/// deterministically frame by frame. Routes through [`update`](Self::update).
pub fn tick(&mut self, dt: f64) {
let _ = self.update(Msg::Tick(dt));
}
/// Builder form of [`set_clock`](Self::set_clock).
pub fn with_clock(mut self, seconds: f64) -> Self {
self.state.clock_s = Some(seconds);
self
}
/// **FC-8 quiescence.** `true` when no stage is `Running` and no Spark task is
/// in flight — i.e. the view has settled and a caller need not wait. Readable
/// without any timing hack (also surfaced in `state_json` as `idle`).
pub fn is_idle(&self) -> bool {
let any_running = self.state.nodes.iter().any(|n| n.status == Status::Running);
let any_inflight = self.state.progress.as_ref().is_some_and(|p| p.inflight > 0);
!(any_running || any_inflight)
}
pub fn with_progress(mut self, p: JobProgress) -> Self {
self.state.progress = Some(p);
self
}
/// Turn the flowing-edge / status-glow animation on or off (default on). When
/// off it renders a still, themed DAG (good for screenshots / headless).
pub fn with_animation(mut self, on: bool) -> Self {
self.state.animated = on;
self
}
/// Set the flow speed (marching-ants cycles/sec). Clamped non-negative.
pub fn with_speed(mut self, speed: f32) -> Self {
self.state.speed = speed.max(0.0);
self
}
/// Turn the **flow-sim overlay** on/off (feature `flowsim`). When on, each
/// dataflow edge shows advected particles (density = volume, speed = throughput)
/// with a static-arrows fallback, replacing the marching-ants dashes. The per-edge
/// `flow`/`volume` are derived from the same `throughput_rows_s`/`rows` the DAG
/// already carries (see `paint_dag`), so no new signal is needed. Same builder
/// shape as [`with_animation`](Self::with_animation).
#[cfg(feature = "flowsim")]
#[deprecated(since = "0.3.0", note = "Flow now renders throughput packets directly")]
pub fn with_flowsim(mut self, on: bool) -> Self {
self.flowsim = on.then(facett_flowsim::FlowSim::default);
self
}
/// Mutable access to the flow-sim overlay (feature `flowsim`), so a host (korp /
/// plandag) can drive its mode/speed: `view.flowsim_mut().map(|f|
/// f.update(FlowMsg::SetMode(..)))`. `None` when flowsim is off. This is the
/// host-facing Msg pass-through seam (§6.A/§6.B).
#[cfg(feature = "flowsim")]
#[deprecated(since = "0.3.0", note = "Flow now renders throughput packets directly")]
pub fn flowsim_mut(&mut self) -> Option<&mut facett_flowsim::FlowSim> {
self.flowsim.as_mut()
}
/// Attach per-edge throughput labels (e.g. `"12k rows/s"`), indexed like the
/// `edges` passed to [`new`](Self::new). Shorter/longer vectors are tolerated.
pub fn with_throughput(mut self, labels: Vec<Option<String>>) -> Self {
self.state.throughput = labels;
self
}
/// (idle, running, done, failed).
///
/// **Lossy since [`Status::Unknown`] exists** — an undetermined node is in
/// none of these four buckets, so they need not sum to the node count. The
/// signature and the four numbers are unchanged for every pre-`Unknown`
/// caller; use [`tally`](Self::tally) when the total matters.
pub fn counts(&self) -> (usize, usize, usize, usize) {
self.tally().as_legacy_tuple()
}
/// The full five-way [`StatusTally`] (`idle`/`running`/`done`/`failed`/
/// `unknown`), summing to the node count.
pub fn tally(&self) -> StatusTally {
let mut t = StatusTally::default();
for n in &self.state.nodes {
t.add(n.status.into());
}
t
}
/// **FC-3** — read the complete observable state at any frame boundary.
pub fn state(&self) -> &PipelineState {
&self.state
}
/// **FC-2 / FC-8** — the single mutation path. Apply one [`Msg`]; return the
/// [`Effect`]s the host should run (only a node click raises one:
/// [`Effect::PreviewRequest`]). The canvas click / camera gesture / fit button
/// all route their [`Msg`] here too, so live == headless.
pub fn update(&mut self, msg: Msg) -> Vec<Effect> {
match msg {
Msg::SelectNode(name) => {
// Click always selects (the name comes from a real node box); a
// fresh selection also raises the one-shot host preview request.
self.state.selected = Some(name.clone());
self.preview_request = Some(name.clone());
return vec![Effect::PreviewRequest(name)];
}
Msg::Pan(delta) => {
self.state.flight = None;
self.state.nav.pan(delta);
}
Msg::Zoom { factor, anchor } => {
self.state.flight = None;
self.state.nav.zoom_to(factor, anchor);
}
Msg::FocusNode { name, now } => {
let at = self.state.nodes.iter().position(|n| n.name == name).and_then(|i| self.layout_scene().get(i).copied());
if let (Some(p), Some(seen)) = (at, self.track.seen()) {
let scale = self.state.nav.scale.max(1.0).min(self.state.nav.max_scale);
self.state.flight = Some(facett_core::nav::NavFlight::to(&self.state.nav, self.state.flight, p, scale, seen, now));
}
}
Msg::FitFly(now) => {
if let (Some(bb), Some(seen)) = (self.scene_bbox(), self.track.seen()) {
self.state.flight = Some(facett_core::nav::NavFlight::fit(&self.state.nav, self.state.flight, bb, seen, 0.08, now));
self.state.fitted = true;
}
}
Msg::FlyTick(now) => {
if let Some(f) = self.state.flight {
if !f.tick(&mut self.state.nav, now) {
self.state.flight = None;
}
}
}
Msg::Fit(rect) => {
if let Some(bb) = self.scene_bbox() {
self.state.nav.fit(bb, rect, 0.08);
self.state.fitted = true;
}
}
Msg::RequestRefit => self.state.fitted = false,
Msg::SetClock(s) => self.state.clock_s = Some(s),
Msg::Tick(dt) => self.state.clock_s = Some(self.state.clock_s.unwrap_or(0.0) + dt),
}
Vec::new()
}
/// The persistent geometry used by the shared Falsett renderer.
pub fn flow_scene(&self) -> &FlowScene {
&self.scene
}
/// **Test/host hook (additive).** The public, return-asserted view of the
/// private [`layout_scene`](Self::layout_scene) — the toposort-column scene
/// centres the DAG painter places each stage box at (before the camera maps
/// them to screen). Exposed so the pipeline call-chain matrix can assert the
/// *layout/columns* stage directly: finite, x monotone-non-decreasing with the
/// toposort column, rows separated, count == nodes. Calls the **same** private
/// fn the painter uses — additive, no behaviour change.
pub fn layout_scene_centers(&self) -> Vec<Pos2> {
self.layout_scene()
}
/// Node centres in **scene** coords (the fixed-grid toposort layout, origin at
/// 0,0). The painter maps these through [`Navigable::to_screen`] so the whole
/// DAG pans/zooms as one. Scene-space is stable across frames — independent of
/// the allocated `rect` — which is what makes the camera coherent.
fn layout_scene(&self) -> Vec<Pos2> {
self.scene
.graph()
.nodes
.iter()
.map(|n| Pos2::new(n.position[0], n.position[1]))
.collect()
}
/// The scene-space bounding box enclosing every node box (with a small margin),
/// used by fit-to-view. `None` when there are no nodes.
fn scene_bbox(&self) -> Option<Rect> {
let centers = self.layout_scene();
if centers.is_empty() {
return None;
}
let mut bb = Rect::from_center_size(centers[0], Vec2::new(NODE_W, NODE_H));
for c in ¢ers[1..] {
bb = bb.union(Rect::from_center_size(*c, Vec2::new(NODE_W, NODE_H)));
}
Some(bb.expand(NODE_W * 0.25))
}
/// Read this frame's input (wheel zoom-to-cursor, trackpad pinch + pan,
/// drag-pan, keyboard pan/zoom, `F`/Ctrl-0 fit) and **return** the camera
/// [`Msg`]s it produced — the **pure** (FC-9) sibling of the OSM 2D viewer's
/// `handle_input`. The [`update`](Self::update) bridge applies them, so the DAG
/// feels identical (NAV-2) without the view mutating the model.
fn nav_input_msgs(&self, ui: &mut egui::Ui, rect: Rect, resp: &egui::Response) -> Vec<Msg> {
facett_core::nav::gestures(ui, rect, resp)
.into_iter()
.map(|g| match g {
facett_core::nav::NavGesture::Pan(d) => Msg::Pan(d),
facett_core::nav::NavGesture::Zoom { factor, anchor } => {
Msg::Zoom { factor, anchor }
}
facett_core::nav::NavGesture::Fit => Msg::Fit(rect),
})
.collect()
}
/// Force-fit the camera to the DAG within `rect` (exposed for hosts + tests).
/// Routes through the FC-2 [`update`](Self::update) path.
pub fn fit_to(&mut self, rect: Rect) {
let _ = self.update(Msg::Fit(rect));
}
/// Wheel-zoom to a screen anchor by factor `k` (exposed for headless tests).
pub fn zoom_to_cursor(&mut self, k: f32, anchor: Pos2) {
let _ = self.update(Msg::Zoom { factor: k, anchor });
}
/// Pan the camera by a screen-pixel delta (exposed for headless tests).
pub fn pan_pixels(&mut self, dx: f32, dy: f32) {
let _ = self.update(Msg::Pan(Vec2::new(dx, dy)));
}
/// Read-only view of the camera (scale + offset) — the headless nav surface.
pub fn nav(&self) -> &Navigable {
&self.state.nav
}
/// **Restore** a camera (and its fitted state) onto a freshly-built view. A
/// host that rebuilds the `PipelineView` each frame from changing data (e.g.
/// Korp's live stream) calls this so pan/zoom **persists** across rebuilds —
/// otherwise the camera would snap back to a fresh fit every frame.
pub fn set_camera(&mut self, nav: Navigable, fitted: bool) {
self.state.nav = nav;
self.state.fitted = fitted;
}
/// **Where the DAG is on screen**: every stage box mapped through the camera (what
/// a robot compares with [`seen_rect`](Self::seen_rect)). `None` with no stages.
pub fn screen_box(&self) -> Option<Rect> {
let centers = self.layout_scene();
let mut b: Option<Rect> = None;
for c in centers {
let r = Rect::from_center_size(c, Vec2::new(NODE_W, NODE_H));
let r = Rect::from_two_pos(self.state.nav.to_screen(r.min), self.state.nav.to_screen(r.max));
b = Some(b.map_or(r, |b| b.union(r)));
}
b
}
/// The part of the last painted canvas the viewer saw (canvas ∩ clip).
pub fn seen_rect(&self) -> Option<Rect> {
self.track.seen()
}
/// Whether the camera has been fit to the DAG yet (host carries this across
/// rebuilds together with [`nav`](Self::nav) / [`set_camera`](Self::set_camera)).
pub fn is_fitted(&self) -> bool {
self.state.fitted
}
/// Count of node boxes whose scene rect (mapped to screen) intersects `rect` —
/// the honest "visible nodes" number after pan/zoom. The analogue of the OSM
/// viewer's `visible_point_count`, so a test can assert culling by camera.
pub fn visible_node_count(&self, rect: Rect) -> usize {
let centers = self.layout_scene();
centers
.iter()
.filter(|&&c| {
let screen_c = self.state.nav.to_screen(c);
let half = Vec2::new(NODE_W, NODE_H) * 0.5 * self.state.nav.scale;
let nrect = Rect::from_center_size(screen_c, half * 2.0);
rect.intersects(nrect)
})
.count()
}
/// The currently click-selected node name, if any (read each frame by a host
/// to drive a preview pane). Part of the discovery contract.
pub fn selected(&self) -> Option<&str> {
self.state.selected.as_deref()
}
/// **Drain** the one-shot preview request: returns the dataset name the user
/// just clicked (once), so the host can fetch a table preview. `None` until the
/// next click. Keeps the component free of any data-fetch dependency. This is
/// the concrete side of the [`Effect::PreviewRequest`] that
/// [`update`](Self::update) also returns.
pub fn take_preview_request(&mut self) -> Option<String> {
self.preview_request.take()
}
/// The themed + animated DAG paint, **navigable** (GAP #1) and **pure** (FC-9):
/// nodes are laid out in scene coords and mapped through
/// [`Navigable::to_screen`], so wheel zoom-to-cursor, drag-pan, keyboard
/// pan/zoom and fit-to-view move the whole graph as one. It reads `&self`,
/// paints, and *returns* the camera / selection [`Msg`]s the interactions
/// produced (the [`update`](Self::update) bridge applies them). The grid is
/// reused only as the *scene* layout; the camera frames it.
fn paint_dag(&self, ui: &mut egui::Ui, time: f64) -> Vec<Msg> {
let mut msgs: Vec<Msg> = Vec::new();
let n = self.state.nodes.len();
if n == 0 {
ui.label("no stages");
return msgs;
}
let th = theme(ui);
// Take the whole available area as the navigable canvas (the camera, not
// the layout, decides what's on screen now).
let avail = facett_core::nav::canvas_size(ui);
let canvas = Vec2::new(avail.x.max(120.0), avail.y.max(320.0));
let (rect, resp) = ui.allocate_exact_size(canvas, Sense::click_and_drag());
// First paint: frame the DAG, CENTRED in the part of the canvas the viewer sees
// (canvas ∩ clip — a canvas in a scroll area is partly off-screen). FC-9: the
// view is pure, so we *emit* the fit instead of mutating — the `update` bridge
// applies it (one-frame deferred vs. the old in-paint fit).
let (seen, moved) = self.track.frame(ui, rect);
let now = ui.input(|i| i.time);
if !self.state.fitted {
msgs.push(Msg::Fit(seen));
} else if moved != Vec2::ZERO {
// the camera is in screen space: a canvas that moved takes its picture along
msgs.push(Msg::Pan(moved));
}
// a 🎯 Focus / ⤢ Fit flight advances, repainting only while it flies
if self.state.flight.is_some() {
msgs.push(Msg::FlyTick(now));
ui.ctx().request_repaint();
}
// Route this frame's pan/zoom/keyboard/fit input into camera Msgs.
msgs.extend(self.nav_input_msgs(ui, rect, &resp));
let scale = self.state.nav.scale;
let painter = ui.painter_at(rect);
painter.rect_filled(rect, egui::CornerRadius::ZERO, th.bg);
// Scene centres → screen centres through the shared camera.
// When the flow-sim overlay is live (feature `flowsim` + mode ≠ Off) it draws
// the edge motion (particles / static arrows), so the marching-ants dashes
// stand down — only the base curve + arrowhead draw beneath the overlay.
#[cfg(feature = "flowsim")]
let flowsim_active = self
.flowsim
.as_ref()
.is_some_and(|f| f.mode() != facett_flowsim::FlowMode::Off);
#[cfg(not(feature = "flowsim"))]
let flowsim_active = false;
let view = View::from_navigable(&self.state.nav, rect);
let painted = facett_flow::paint(
ui,
ui.id().with("pipeline-flow"),
rect,
&self.scene,
&view,
&PaintOpts {
seconds: time * (self.state.speed as f64 / 0.6),
cpu: CpuStyle::Full {
max_edges: usize::MAX,
},
animation: if flowsim_active {
Animation::External
} else if self.state.animated {
Animation::Flow
} else {
Animation::None
},
..Default::default()
},
);
// ── edges: themed line + animated marching-ants overlay + throughput ──
for (ei, flow_edge) in self.flow_edges.iter().enumerate() {
if let (Some(index), Some(Some(label))) = (flow_edge, self.state.throughput.get(ei))
&& let Some(mid) = self.scene.edge_point(*index, 0.5)
{
painter.text(
view.screen(mid) - Vec2::new(0.0, 9.0 * scale),
Align2::CENTER_BOTTOM,
label,
FontId::monospace(10.0 * scale),
th.text_dim,
);
}
}
// ── nodes: themed box + per-status glow + rows + selection ring ──
// The painter draws the visual; per node we also ride a hit-testable,
// labelled AccessKit node keyed on the **stable node name** (FC-4 + FC-5)
// so a robot driver / screen reader can find & click each stage by label.
// The AccessKit/hit rect is the **screen-space** (post-camera) box, so
// clicks land on what the user sees regardless of pan/zoom.
let base_id = ui.id().with("pipeline-nodes");
// Reserve semantic slots before input-order capping so an active node
// cannot disappear from the accessibility tree in a crowded viewport.
let focused = ui.memory(|memory| memory.focused());
let priority: Vec<usize> = painted
.visible_nodes
.iter()
.filter_map(|shown| {
let key = &self.state.nodes[shown.node].name;
(self.state.selected.as_deref() == Some(key.as_str())
|| focused == Some(facett_core::a11y::stable_id(base_id, key)))
.then_some(shown.node)
})
.collect();
let mut semantic_slots = 2_000usize.saturating_sub(priority.len());
for visible in &painted.visible_nodes {
let node = &self.state.nodes[visible.node];
let status = node.status;
let c = visible.rect.center();
let nrect = visible.rect;
// Cull boxes fully outside the pane (cheap; keeps big DAGs fast).
if !rect.intersects(nrect.expand(40.0 * scale)) {
continue;
}
let sc = status.theme_color(&th);
// Per-node AccessKit node: role=Button, label "{name} · {status} · N rows",
// value = rows, selected = is-this-the-picked-node. Click emits a Msg.
let rows_txt = node
.rows
.map(|r| format!(" · {} rows", human_count(r)))
.unwrap_or_default();
let a11y_label = format!("{} · {}{}", node.name, status.label(), rows_txt);
let mut sem = facett_core::a11y::Semantics::button(a11y_label)
.selected(self.state.selected.as_deref() == Some(node.name.as_str()));
if let Some(r) = node.rows {
sem = sem.value(r as f64);
}
let semantic = priority.contains(&visible.node)
|| if semantic_slots > 0 {
semantic_slots -= 1;
true
} else {
false
};
let resp = if semantic {
facett_core::a11y::node(ui, base_id, &node.name, Sense::click(), nrect, sem)
} else {
ui.interact(nrect, base_id.with(&node.name), Sense::click())
};
if resp.clicked() {
// FC-9: emit the selection Msg instead of mutating `self`. 📌 Stay: the
// camera never moves; 🎯 Focus: the stage flies to the centre.
if facett_core::fly::NavMode::remembered(ui.ctx()) == facett_core::fly::NavMode::Focus {
msgs.push(Msg::FocusNode { name: node.name.clone(), now: ui.input(|i| i.time) });
}
msgs.push(Msg::SelectNode(node.name.clone()));
}
let is_selected = self.state.selected.as_deref() == Some(node.name.as_str());
// Status glow underlay (failed blooms, running pulses, done steady).
let intensity = if self.state.animated {
status_glow(status, time, self.state.speed)
} else {
// Static look: a soft steady glow for the non-idle states. Unknown
// sits between done and failed — it must draw the eye even with
// animation off (an unobserved node is not a quiet node).
match status {
Status::Idle => 0.0,
Status::Failed => 0.7,
Status::Unknown(_) => 0.5,
_ => 0.3,
}
};
if intensity > 0.0 {
let glow = if status == Status::Failed {
facett_core::look::RED
} else {
sc
};
glow_rect(
&painter,
nrect,
glow,
intensity,
((6.0 * scale).round() as u32).clamp(1, 12),
);
}
let cr = egui::CornerRadius::same((6.0 * scale).round().clamp(0.0, 255.0) as u8);
painter.rect_filled(nrect, cr, th.node_fill);
let border = if is_selected { 3.0 } else { 2.0 } * scale.clamp(0.4, 2.0);
if status.is_unknown() {
// A **dashed** outline for a state we could not determine: a solid
// box asserts a known state, a broken one visibly does not. This is
// the non-colour channel of the Unknown treatment (greyscale + CVD
// safe), paired with the amber `sc` and the `?` marker below.
painter.extend(dashed_rect(
nrect,
Stroke::new(border, sc),
5.0 * scale,
4.0 * scale,
));
} else {
painter.rect_stroke(nrect, cr, Stroke::new(border, sc), egui::StrokeKind::Inside);
}
// selection ring so the picked node reads clearly.
if is_selected {
painter.rect_stroke(
nrect.expand(3.0 * scale),
egui::CornerRadius::same((8.0 * scale).round().clamp(0.0, 255.0) as u8),
Stroke::new(1.0_f32, th.accent),
egui::StrokeKind::Outside,
);
}
painter.text(
c - Vec2::new(0.0, 12.0 * scale),
Align2::CENTER_CENTER,
&node.name,
FontId::proportional(15.0 * scale),
th.text,
);
// Status label, plus the rows count when the run reported it, plus live
// per-node throughput/lag when the stream surfaced them (GAP #2/#3 hook).
let mut sub = match node.rows {
Some(r) => format!("{} · {} rows", status.label(), human_count(r)),
None => status.label().to_string(),
};
if let Some(t) = node.throughput_rows_s {
sub.push_str(&format!(" · {:.0} rows/s", t));
}
if let Some(l) = node.lag_s {
sub.push_str(&format!(" · {:.1}s lag", l));
}
painter.text(
c + Vec2::new(0.0, 11.0 * scale),
Align2::CENTER_CENTER,
sub,
FontId::monospace(11.0 * scale),
sc,
);
// A failed node shows a ⚠ corner marker; the message goes to the detail panel.
if status == Status::Failed {
painter.text(
nrect.right_top() + Vec2::new(-9.0 * scale, 9.0 * scale),
Align2::CENTER_CENTER,
"⚠",
FontId::proportional(13.0 * scale),
facett_core::look::RED,
);
}
// An undetermined node shows a **?** corner marker in the same slot —
// the glyph channel that keeps Unknown ≠ Idle without relying on hue.
if status.is_unknown() {
painter.text(
nrect.right_top() + Vec2::new(-9.0 * scale, 9.0 * scale),
Align2::CENTER_CENTER,
"?",
FontId::proportional(14.0 * scale),
UNKNOWN_AMBER,
);
}
}
// ── flow-sim overlay (feature `flowsim`) ──
// Derive per-edge `flow`/`volume` from the SAME model the DAG draws from and
// sample Falsett's shared `edge_point` (the pure `plan_edges` maps them
// to speed/density; the overlay draws particles or the static-arrows
// fallback). No new signal, no second Bézier, no knut change.
#[cfg(feature = "flowsim")]
if flowsim_active {
let mut polylines: Vec<Vec<Pos2>> = Vec::with_capacity(self.state.edge_ends.len());
for edge in &self.flow_edges {
polylines.push(
edge.map(|i| {
(0..=16)
.filter_map(|j| self.scene.edge_point(i, j as f32 / 16.0))
.map(|p| view.screen(p))
.collect()
})
.unwrap_or_default(),
);
}
// ONE derivation of the per-edge signal, shared with `state_json` (LAW 5)
// so the introspection can never disagree with the pixels.
let (flow, volume) = self.edge_flow_volume();
if let Some(fs) = self.flowsim.as_ref() {
// ── THE GPU LANE (feature `flowsim-gpu`) ──────────────────────────
// Try the device first: `FlowSim::render_gpu` pushes an `egui_wgpu`
// paint callback that advects every particle in a compute pass and
// draws them as additive HDR points. It returns `false` — and the CPU
// painter below runs unchanged — whenever it cannot honestly claim the
// frame (not opted in, no renderer installed, sub-pixel pane, dead
// DAG, arrows mode). The whole gate lives in flowsim so korp and
// plandag get the same protocol without re-deriving it.
//
// This call is the wiring `GFX_V2` / `flowsim-spike.md` §9.6 recorded
// as missing: `GpuParticles::advect` was proven against `advect_cpu` on
// a real device, and NOTHING drove it — every frame in the tree came
// from the CPU painter.
#[cfg(feature = "flowsim-gpu")]
let on_gpu = fs.render_gpu(ui, rect, &polylines, &flow, &volume, time);
#[cfg(not(feature = "flowsim-gpu"))]
let on_gpu = false;
if !on_gpu {
// Stateless render: the immutable `view(&self)` re-supplies geometry
// each frame (mode/speed mutate only via `flowsim_mut().update(..)`).
fs.render(&painter, &polylines, &flow, &volume, time);
}
}
}
// Keep the animation ticking — only when following the ambient clock. With
// an injected clock (`clock_s`), the host steps frames, so we don't spin.
if (self.state.animated || flowsim_active) && self.state.clock_s.is_none() {
ui.ctx().request_repaint();
}
msgs
}
}
/// Dashed unknown-status border, retained as the non-colour status channel.
fn dashed_rect(rect: Rect, stroke: Stroke, dash: f32, gap: f32) -> Vec<egui::Shape> {
// A closed 5-point ring (last == first) so the four corners join cleanly.
let ring = [
rect.left_top(),
rect.right_top(),
rect.right_bottom(),
rect.left_bottom(),
rect.left_top(),
];
egui::Shape::dashed_line(&ring, stroke, dash.max(1.0), gap.max(1.0))
}
impl PipelineView {
/// The copyable text: the selected node's line (`name status rows err`) when a
/// node is selected, else the whole node list as a TSV rectangle.
pub fn copy_text(&self) -> Option<String> {
if self.state.nodes.is_empty() {
return None;
}
let line = |n: &PipelineNode| {
format!(
"{}\t{}\t{}\t{}",
n.name,
n.status.label(),
n.rows.map(|r| r.to_string()).unwrap_or_default(),
n.error.as_deref().unwrap_or(""),
)
};
if let Some(sel) = self
.state
.selected
.as_deref()
.and_then(|name| self.state.nodes.iter().find(|n| n.name == name))
{
return Some(line(sel));
}
let mut out = String::from("name\tstatus\trows\terror");
for n in &self.state.nodes {
out.push('\n');
out.push_str(&line(n));
}
Some(out)
}
/// **FC-9 render** — a **pure** function of `&self`: paints the status legend,
/// the optional progress readout, the themed animated navigable DAG, and the
/// selected-node detail panel, and *returns* the [`Msg`]s the interactions
/// produced. It MUST NOT mutate the [`PipelineState`]; the
/// [`impl_facet_via_elm!`](facett_core::impl_facet_via_elm) bridge applies the
/// returned messages through [`update`](Self::update).
pub fn view(&self, ui: &mut egui::Ui) -> Vec<Msg> {
facett_core::navgraph::declare_own_draw_order(ui); // atoms follow the data order (navgraph §9)
let mut msgs: Vec<Msg> = Vec::new();
let th = theme(ui);
let t = self.tally();
let (idle, running, done, failed) = t.as_legacy_tuple();
// Themed status legend. The `⤢ Fit` button emits a RequestRefit Msg (the
// rect isn't known until paint, so the actual fit is deferred one frame).
let refit = ui
.horizontal(|ui| {
let chip = |ui: &mut egui::Ui, n: usize, s: Status| {
ui.colored_label(s.theme_color(&th), format!("● {n} {}", s.family()));
};
chip(ui, done, Status::Done);
chip(ui, running, Status::Running);
chip(ui, failed, Status::Failed);
chip(ui, idle, Status::Idle);
// The unknown chip appears **only when there is something unknown**,
// so a pipeline that can never be unknown renders byte-identically
// to before this variant existed. `?` not `●`: even in the legend,
// undetermined must not look like just another status dot.
if t.unknown > 0 {
ui.colored_label(UNKNOWN_AMBER, format!("? {} unknown", t.unknown));
}
// Navigation affordance + hint (GAP #1): a fit button + the gestures,
// so the camera is discoverable and CLI/keyboard parity is advertised.
ui.separator();
let clicked = ui
.button("⤢ Fit")
.on_hover_text("Frame the whole DAG (F / Ctrl-0)")
.clicked();
ui.label(
egui::RichText::new(format!(
"zoom {:.2}× · wheel zoom · drag pan · arrows · F fit",
self.state.nav.scale
))
.small()
.color(th.text_dim),
);
clicked
})
.inner;
if refit {
// ⤢ Fit flies to the whole DAG, centred in what the viewer sees
msgs.push(match self.track.seen() {
Some(_) => Msg::FitFly(ui.input(|i| i.time)),
None => Msg::RequestRefit,
});
}
if let Some(p) = &self.state.progress {
ui.add(
egui::ProgressBar::new(p.fraction)
.text(format!("{} / {} tasks", p.tasks_done, p.tasks_total)),
);
// Live-run readout: throughput · rows · bytes · elapsed (only the
// parts the run actually reported).
let mut bits: Vec<String> = Vec::new();
if p.elapsed_s > 0.0 {
bits.push(format!("{:.0} rows/s", p.rows_per_s()));
}
if p.rows > 0 {
bits.push(format!("{} rows", human_count(p.rows)));
}
if p.bytes_read > 0 {
bits.push(format!("{} read", human_bytes(p.bytes_read)));
}
if p.inflight > 0 {
bits.push(format!("{} inflight", p.inflight));
}
if p.elapsed_s > 0.0 {
bits.push(format!("{:.1}s", p.elapsed_s));
}
if !bits.is_empty() {
ui.label(
egui::RichText::new(bits.join(" · "))
.monospace()
.color(th.text_dim),
);
}
}
ui.separator();
// FC-7: drive animation from the injected clock when pinned; otherwise
// follow egui's ambient wall-time for the live (non-test) UX.
let time = self.state.clock_s.unwrap_or_else(|| ui.input(|i| i.time));
msgs.extend(self.paint_dag(ui, time));
// ── selected-node detail panel: status, rows, and the failure error ──
if let Some(name) = self.state.selected.clone()
&& let Some(node) = self.state.nodes.iter().find(|n| n.name == name)
{
ui.separator();
ui.horizontal(|ui| {
ui.strong(&node.name);
ui.colored_label(node.status.theme_color(&th), node.status.label());
if let Some(d) = &node.detail {
ui.label(egui::RichText::new(d).color(th.text_dim));
}
if let Some(r) = node.rows {
ui.label(egui::RichText::new(format!("{} rows", human_count(r))).monospace());
}
if let Some(t) = node.throughput_rows_s {
ui.label(
egui::RichText::new(format!("{t:.0} rows/s"))
.monospace()
.color(th.accent),
);
}
if let Some(l) = node.lag_s {
ui.label(
egui::RichText::new(format!("{l:.1}s lag"))
.monospace()
.color(th.text_dim),
);
}
});
if let Some(err) = &node.error {
ui.colored_label(facett_core::look::RED, format!("⚠ {err}"));
}
// An undetermined node explains itself in prose here — the panel is
// where "why don't we know?" has to be answerable without a log dive.
if let Some(r) = node.status.unknown_reason() {
ui.colored_label(
UNKNOWN_AMBER,
format!("? state not determined — {}", r.describe()),
);
}
}
// ── render-lane emit: this Facet::ui path RAN ─────────────────────────
#[cfg(feature = "testmatrix")]
facett_core::testmatrix::emit(
"facett-pipeline::PipelineView::ui",
"ui_render",
!self.state.nodes.is_empty(),
&format!(
"nodes={} edges={}",
self.state.nodes.len(),
self.state.edge_ends.len()
),
);
msgs
}
/// The rich, headless-assertable `state_json` — kept byte-for-byte identical to
/// the pre-Elm surface (`facett-demo` + `mega_matrix` read these keys), so it is
/// supplied to [`impl_facet_via_elm!`](facett_core::impl_facet_via_elm) in
/// **form 3** (`custom_state_json`) rather than a plain `serde(state())`.
fn state_json_value(&self) -> serde_json::Value {
let t = self.tally();
let (idle, running, done, failed) = t.as_legacy_tuple();
// Per-node readable model: exactly what each box renders, so a headless
// test can assert a specific node's status / rows / error (LAW 6).
let nodes: Vec<serde_json::Value> = self
.state
.nodes
.iter()
.map(|n| {
serde_json::json!({
"name": n.name,
// Carries the reason for an undetermined node
// (`"unknown:unreachable"`), so a text/JSON reader can never
// mistake it for `"idle"`.
"status": n.status.label(),
// …and the reason again, split out for machine consumers that
// would rather not parse the label. `null` unless Unknown.
"unknown_reason": n.status.unknown_reason().map(|r| r.token()),
"detail": n.detail,
"rows": n.rows,
"error": n.error,
"throughput_rows_s": n.throughput_rows_s,
"lag_s": n.lag_s,
})
})
.collect();
// The undetermined node(s), pulled out beside `errors` — a driver checking
// "did anything fail?" must be able to ask "and did we fail to LOOK?"
// just as cheaply, or the second question never gets asked.
let unknowns: Vec<serde_json::Value> = self
.state
.nodes
.iter()
.filter_map(|n| {
n.status
.unknown_reason()
.map(|r| serde_json::json!({ "node": n.name, "reason": r.token(), "why": r.describe() }))
})
.collect();
// The error(s) surfaced on failed nodes — pulled out for easy assertion.
let errors: Vec<serde_json::Value> = self
.state
.nodes
.iter()
.filter(|n| n.status == Status::Failed)
.map(|n| serde_json::json!({ "node": n.name, "error": n.error }))
.collect();
let selected = self
.state
.selected
.as_deref()
.and_then(|name| self.state.nodes.iter().find(|n| n.name == name))
.map(|n| {
serde_json::json!({
"name": n.name,
"status": n.status.label(),
"unknown_reason": n.status.unknown_reason().map(|r| r.token()),
"detail": n.detail,
"rows": n.rows,
"error": n.error,
})
});
serde_json::json!({
"node_count": self.state.nodes.len(),
"edge_count": self.state.edge_ends.len(),
"nodes": nodes,
// The four legacy buckets are unchanged; `unknown` is ADDITIVE — and
// it is the reason the four no longer have to sum to `node_count`.
"statuses": { "idle": idle, "running": running, "done": done, "failed": failed, "unknown": t.unknown },
"errors": errors,
"unknowns": unknowns,
"selected": selected,
"progress": self.state.progress.as_ref().map(|p| serde_json::json!({
"fraction": p.fraction,
"tasks_done": p.tasks_done,
"tasks_total": p.tasks_total,
"rows": p.rows,
"bytes_read": p.bytes_read,
"elapsed_s": p.elapsed_s,
"inflight": p.inflight,
"rows_per_s": p.rows_per_s(),
})),
"animated": self.state.animated,
"speed": self.state.speed,
// FC-8: a single observable quiescence flag — true when nothing is
// running and no task is in flight. A caller reads `idle` instead of
// guessing with a sleep.
"idle": self.is_idle(),
// FC-7: the injected clock (null = following ambient time).
"clock_s": self.state.clock_s,
// GAP #1: the shared camera, so navigation is headless-assertable
// (LAW: components emit what they render). `visible` = node boxes
// currently intersecting the last-fit scene bbox at this zoom.
"nav": {
"scale": self.state.nav.scale,
"offset": self.state.nav.offset,
"fitted": self.state.fitted,
},
"dag": { "nodes": self.state.nodes.len(), "edges": self.state.edge_ends.len(), "columns": self.layout.cols, "selected": null },
// The flow-sim overlay's own introspection, nested (feature `flowsim`):
// mode / n_edges / n_flowing / total_particles / gpu. `null` when the
// overlay is off, so "the pipeline is dead" (`n_flowing == 0`) is
// distinguishable from "the overlay isn't running".
"flowsim": self.flowsim_state_json(),
})
}
/// The nested flow-sim state (`None` → JSON `null`) — see `state_json`. Derived
/// from [`edge_flow_volume`](Self::edge_flow_volume), the SAME signal `paint_dag`
/// renders with, so the JSON cannot drift from the picture.
fn flowsim_state_json(&self) -> serde_json::Value {
#[cfg(feature = "flowsim")]
{
let (flow, volume) = self.edge_flow_volume();
return self.flowsim.as_ref().map_or(serde_json::Value::Null, |f| {
f.state_json_for(&flow, &volume)
});
}
#[cfg(not(feature = "flowsim"))]
serde_json::Value::Null
}
/// **The ONE per-edge flow-sim signal derivation** (feature `flowsim`), read by
/// both `paint_dag` and `flowsim_state_json`.
///
/// `flow[i]` is the consumer stage's live throughput (rows/s, `0` = not flowing);
/// `volume[i]` is a **relative** size hint in `[0,1]`.
///
/// Why relative: `plan_edges` reads `volume` as an ABSOLUTE multiplier on
/// `density_per_edge`, so feeding it raw row counts pins every real edge at
/// `max_density` — `8 × 48 000` clamps to 64 and so does `8 × 120 000`, and
/// density then carries no information at all. Normalising against the busiest
/// edge restores the intended `~1.0` size hint, so a fat edge is visibly denser
/// than a thin one. A stage with no row count falls back to the unit hint.
#[cfg(feature = "flowsim")]
fn edge_flow_volume(&self) -> (Vec<f32>, Vec<f32>) {
let n = self.state.nodes.len();
let mut flow = Vec::with_capacity(self.state.edge_ends.len());
let mut rows = Vec::with_capacity(self.state.edge_ends.len());
for &(from_i, to_i) in &self.state.edge_ends {
if from_i >= n || to_i >= n {
flow.push(0.0);
rows.push(None);
continue;
}
let node = &self.state.nodes[from_i];
flow.push(node.throughput_rows_s.unwrap_or(0.0));
rows.push(node.rows);
}
let max_rows = rows.iter().flatten().copied().max().unwrap_or(0) as f32;
let volume = rows
.iter()
.map(|r| match r {
Some(v) if max_rows > 0.0 => *v as f32 / max_rows,
Some(_) => 0.0,
None => 1.0,
})
.collect();
(flow, volume)
}
/// The click-selected detail the host panel reads (mirrors `Facet::selection_json`).
fn selection_json_value(&self) -> serde_json::Value {
match self
.state
.selected
.as_deref()
.and_then(|name| self.state.nodes.iter().find(|n| n.name == name))
{
Some(n) => serde_json::json!({
"node": n.name,
"status": n.status.label(),
"unknown_reason": n.status.unknown_reason().map(|r| r.token()),
"error": n.error,
}),
None => serde_json::Value::Null,
}
}
}
// ── typed copy (§16) — read-only: selected node or the whole DAG as rows ───────
impl facett_core::clip::CopySource for PipelineView {
fn copy_kinds(&self) -> &[facett_core::clip::ClipKind] {
use facett_core::clip::ClipKind;
&[ClipKind::Rows, ClipKind::Text]
}
fn copy_payload(&self) -> Option<facett_core::clip::ClipPayload> {
self.copy_text().map(facett_core::clip::ClipPayload::Rows)
}
}
// ── FC-2 / FC-3 / FC-8 / FC-9: the canonical Elm split ────────────────────────
impl facett_core::Elm for PipelineView {
type Model = PipelineState;
type Msg = Msg;
type Effect = Effect;
fn title(&self) -> &str {
&self.title
}
fn state(&self) -> &PipelineState {
&self.state
}
fn update(&mut self, msg: Msg) -> Vec<Effect> {
PipelineView::update(self, msg)
}
fn view(&self, ui: &mut egui::Ui) -> Vec<Msg> {
PipelineView::view(self, ui)
}
}
// The bridge macro writes `impl Facet for PipelineView` from the `Elm` impl: `title`,
// the FC-9 `ui` loop (`for m in view(ui) { update(m) }`), plus the extra overrides
// below. **Form 3** (`custom_state_json`) because the pipeline publishes a RICHER
// `state_json` than a plain `serde(state())`: the per-node/status/progress/nav keys
// AND the composed `DepGraphView`'s own `dag` block that `facett-demo`'s
// synthetic_sdp + `mega_matrix` read are reproduced byte-for-byte here.
facett_core::impl_facet_via_elm!(PipelineView, custom_state_json, {
fn copy(&mut self) -> Option<String> {
use facett_core::clip::CopySource as _;
self.copy_payload().map(|p| p.as_text())
}
fn caps(&self) -> facett_core::FacetCaps {
// Reads the Theme, tolerates host resizing, a node is click-selectable, the
// DAG pans/zooms/fits via the shared `Navigable` camera (GAP #1), and the
// selection/graph is copyable.
facett_core::FacetCaps::NONE
.themeable()
.resizable()
.selectable()
.navigable()
.copyable()
}
fn state_json(&self) -> serde_json::Value {
self.state_json_value()
}
fn selection_json(&self) -> serde_json::Value {
self.selection_json_value()
}
});
/// Format a row/record count compactly (e.g. `12.0k`, `3.4M`). Pure + testable;
/// the analogue of [`human_bytes`] for counts rather than bytes.
pub fn human_count(n: u64) -> String {
if n < 1_000 {
return n.to_string();
}
let (v, unit) = if n < 1_000_000 {
(n as f64 / 1_000.0, "k")
} else if n < 1_000_000_000 {
(n as f64 / 1_000_000.0, "M")
} else {
(n as f64 / 1_000_000_000.0, "B")
};
format!("{v:.1}{unit}")
}
/// Format a byte count with binary units (`512 B`, `1.0 KiB`, `3.5 MiB`).
/// Mirrors knut's `human_bytes` so edge/throughput labels read the same.
pub fn human_bytes(n: u64) -> String {
const UNITS: [&str; 6] = ["B", "KiB", "MiB", "GiB", "TiB", "PiB"];
if n < 1024 {
return format!("{n} B");
}
let mut v = n as f64;
let mut u = 0;
while v >= 1024.0 && u < UNITS.len() - 1 {
v /= 1024.0;
u += 1;
}
format!("{v:.1} {}", UNITS[u])
}
#[cfg(test)]
mod tests {
use super::*;
/// ⚠ **The visual distinction.** `Unknown` must not paint the same neutral as
/// `Idle`, in **any** theme — if it did, "nobody has looked at this" would
/// render identically to "we looked, it hasn't started", and the variant would
/// be decoration. Point `Status::Unknown` at `th.text_dim` and this goes red.
///
/// It must also differ from `Failed` (unknown is not failure) and from
/// `Done`/`Running` (it is certainly not either of those).
#[test]
fn unknown_never_paints_as_idle_in_any_theme() {
for ctor in facett_core::Theme::ALL {
let th = ctor();
let u = Status::Unknown(model::UnknownReason::NotProbed);
assert_ne!(
u.theme_color(&th),
Status::Idle.theme_color(&th),
"[{}] unknown must not be the idle neutral",
th.name
);
assert_ne!(
u.theme_color(&th),
Status::Failed.theme_color(&th),
"[{}] unknown is not failure",
th.name
);
assert_ne!(
u.theme_color(&th),
Status::Done.theme_color(&th),
"[{}] unknown is not done",
th.name
);
assert_ne!(
u.theme_color(&th),
Status::Running.theme_color(&th),
"[{}] unknown is not running",
th.name
);
assert_eq!(
u.theme_color(&th),
UNKNOWN_AMBER,
"[{}] it is the constellation amber",
th.name
);
}
// The non-themed fallback palette obeys the same rule.
let u = Status::Unknown(model::UnknownReason::Stale);
assert_ne!(u.color(), Status::Idle.color());
assert_ne!(u.color(), Status::Failed.color());
assert_eq!(u.color(), UNKNOWN_AMBER);
// The reason does not change the hue — every not-knowing reads the same.
assert_eq!(
u.color(),
Status::Unknown(model::UnknownReason::NotProbed).color()
);
}
/// ⚠ **The non-colour channel.** Colour alone is not a distinction (greyscale,
/// CVD): an unknown node's border is DASHED — many short segments — where a
/// known node's is one solid rect. Make `dashed_rect` return a single shape
/// and this goes red.
#[test]
fn the_unknown_border_is_dashed_not_solid() {
let r = Rect::from_min_max(Pos2::new(0.0, 0.0), Pos2::new(120.0, 60.0));
let dashes = dashed_rect(r, Stroke::new(2.0_f32, UNKNOWN_AMBER), 5.0, 4.0);
assert!(
dashes.len() > 8,
"a dashed ring is many segments, got {}",
dashes.len()
);
// Degenerate dash/gap must not divide by zero or hang.
assert!(!dashed_rect(r, Stroke::new(2.0_f32, UNKNOWN_AMBER), 0.0, 0.0).is_empty());
assert!(
dashed_rect(
Rect::from_min_max(Pos2::ZERO, Pos2::ZERO),
Stroke::new(1.0_f32, UNKNOWN_AMBER),
5.0,
4.0
)
.is_empty()
);
}
/// ⚠ **The glow channel.** Idle is `0.0` — literally invisible emphasis, which
/// is how the eye skips a node. An unobserved node is the one node that must
/// not be skipped, so its glow is non-zero and steady (steady, because pulsing
/// would claim activity we have not observed).
#[test]
fn unknown_glows_steadily_and_never_like_idle() {
let u = Status::Unknown(model::UnknownReason::Unreachable);
assert_eq!(status_glow(u, 1.23, 0.6), 0.5);
assert_ne!(
status_glow(u, 1.23, 0.6),
status_glow(Status::Idle, 1.23, 0.6)
);
assert!(
status_glow(u, 1.23, 0.6) > 0.0,
"an unobserved node must draw the eye"
);
// Steady: the same at every clock, unlike Running.
for k in 0..50 {
assert_eq!(
status_glow(u, k as f64 * 0.37, 0.6),
0.5,
"unknown does not pulse"
);
}
}
/// ⚠ **The observable surface.** A headless driver reading `state_json` must
/// see `unknown` counted, the reason on the node, and — critically — a status
/// string that is not `"idle"`.
#[test]
fn unknown_rides_state_json_with_its_reason() {
let view = PipelineView::new(
"ladder",
vec![
("a".into(), Status::Done),
(
"b".into(),
Status::Unknown(model::UnknownReason::Unreachable),
),
("c".into(), Status::Unknown(model::UnknownReason::NotProbed)),
],
vec![(0, 1), (1, 2)],
);
assert_eq!(view.tally().unknown, 2);
assert_eq!(view.tally().idle, 0, "not one of them counts as idle");
assert_eq!(view.tally().total(), 3);
assert_eq!(
view.counts(),
(0, 0, 1, 0),
"the legacy tuple is unchanged in shape"
);
let j = facett_core::Facet::state_json(&view);
assert_eq!(j["statuses"]["unknown"], 2);
assert_eq!(j["statuses"]["idle"], 0);
assert_eq!(j["nodes"][1]["status"], "unknown:unreachable");
assert_eq!(j["nodes"][1]["unknown_reason"], "unreachable");
assert_eq!(j["nodes"][0]["unknown_reason"], serde_json::Value::Null);
assert_eq!(j["unknowns"].as_array().unwrap().len(), 2);
assert_eq!(j["unknowns"][0]["node"], "b");
assert_eq!(j["unknowns"][0]["reason"], "unreachable");
assert!(
j["unknowns"][0]["why"]
.as_str()
.unwrap()
.contains("no answer")
);
for n in j["nodes"].as_array().unwrap() {
assert_ne!(
n["status"], "idle",
"no undetermined node may report itself idle"
);
}
// Selection surfaces the reason too.
let sel = facett_core::Facet::selection_json(&view.with_selected("c"));
assert_eq!(sel["status"], "unknown:not-probed");
assert_eq!(sel["unknown_reason"], "not-probed");
}
/// ⚠ **End to end**: a promotion ladder through the real adapter, the real
/// view and the real `state_json`. The unprobed prod rung reaches the screen as
/// unknown-with-a-reason — not idle, not done.
#[test]
fn a_promotion_ladder_renders_its_unobserved_rung_as_unknown() {
use std::collections::HashMap;
let ladder = PromotionLadder::linear(
"yvra",
vec![
Rung::new("test1", SlotRole::Test),
Rung::new("lab", SlotRole::Preprod),
Rung::new("g11.se", SlotRole::Prod).guarded(true),
],
);
let observed = HashMap::from([
("test1".to_string(), RungStatus::Promoted),
("lab".to_string(), RungStatus::Promoting),
]);
let adapter = PromoteLadderAdapter::new(&ladder).with_observations(observed);
let view = PipelineView::from_adapter(&adapter);
let j = facett_core::Facet::state_json(&view);
assert_eq!(j["node_count"], 3);
assert_eq!(j["edge_count"], 2, "the promotion order became edges");
assert_eq!(j["statuses"]["unknown"], 1);
assert_eq!(
j["statuses"]["idle"], 0,
"the unprobed prod rung is NOT idle"
);
assert_eq!(
j["statuses"]["done"], 1,
"…and only the genuinely promoted rung is done"
);
assert_eq!(j["nodes"][2]["status"], "unknown:not-probed");
assert_eq!(j["nodes"][2]["detail"], "prod · guarded");
assert_eq!(j["unknowns"][0]["node"], "g11.se");
// And it paints amber, not grey.
let th = facett_core::Theme::default();
assert_eq!(view.state().nodes[2].status.theme_color(&th), UNKNOWN_AMBER);
assert_ne!(
view.state().nodes[2].status.theme_color(&th),
Status::Idle.theme_color(&th)
);
}
#[test]
fn typed_copy_is_selected_node_or_whole_dag_rows() {
use facett_core::clip::{ClipKind, ClipPayload, CopySource};
let v = PipelineView::new(
"p",
vec![
("src".into(), Status::Done),
("sink".into(), Status::Running),
],
vec![(0, 1)],
);
// No selection -> the whole DAG as a TSV rectangle (header + one row/node).
let p = v.copy_payload().expect("nodes copy");
assert_eq!(p.kind(), ClipKind::Rows);
let text = p.as_text();
assert!(
text.starts_with("name\tstatus\trows\terror"),
"header first: {text}"
);
assert!(text.contains("\nsrc\tdone\t\t"), "src row present: {text}");
// Selecting a node narrows the copy to that node's line.
let v = v.with_selected("sink");
let sel = v.copy_payload().unwrap();
assert_eq!(sel.as_text(), "sink\trunning\t\t");
assert!(matches!(sel, ClipPayload::Rows(_)));
}
#[test]
fn counts_and_state_json() {
let view = PipelineView::new(
"pipe",
vec![
("src".into(), Status::Done),
("mid".into(), Status::Running),
("sink".into(), Status::Idle),
],
vec![(0, 1), (1, 2)],
)
.with_progress(JobProgress {
fraction: 0.5,
tasks_done: 25,
tasks_total: 50,
..Default::default()
});
assert_eq!(view.counts(), (1, 1, 1, 0));
let j = facett_core::Facet::state_json(&view);
assert_eq!(j["statuses"]["done"], 1);
assert_eq!(j["progress"]["tasks_done"], 25);
assert_eq!(j["node_count"], 3);
assert_eq!(j["edge_count"], 2);
assert_eq!(j["dag"]["nodes"], 3);
assert_eq!(j["dag"]["edges"], 2);
// Additive theming/animation fields are present + default-on.
assert_eq!(j["animated"], true);
}
#[test]
fn edge_phase_wraps_and_increases() {
// speed 0 → frozen at 0.
assert_eq!(edge_phase(123.4, 0.0), 0.0);
assert_eq!(edge_phase(5.0, -1.0), 0.0, "non-positive speed freezes");
// phase is in [0,1) and advances with time.
let speed = 0.5;
let p0 = edge_phase(0.0, speed);
let p1 = edge_phase(0.4, speed);
assert!((0.0..1.0).contains(&p0) && (0.0..1.0).contains(&p1));
assert!(p1 > p0, "phase advances over the first cycle");
// wraps: one full cycle (2.0s at 0.5 c/s) returns to ~0.
let wrapped = edge_phase(2.0, speed);
assert!(
wrapped < 1e-3 || (1.0 - wrapped) < 1e-3,
"phase wraps each cycle, got {wrapped}"
);
// monotone-mod over many samples, always bounded.
for k in 0..500 {
let v = edge_phase(k as f64 * 0.013, 1.7);
assert!((0.0..1.0).contains(&v), "phase bounded, got {v}");
}
}
#[test]
fn status_glow_curve_matches_each_state() {
// idle never glows; done/failed are steady; running pulses within bounds.
assert_eq!(status_glow(Status::Idle, 1.23, 0.6), 0.0);
assert_eq!(status_glow(Status::Done, 1.23, 0.6), 0.35);
assert_eq!(status_glow(Status::Failed, 1.23, 0.6), 0.9);
let mut lo = f32::MAX;
let mut hi = f32::MIN;
for k in 0..400 {
let v = status_glow(Status::Running, k as f64 * 0.05, 0.6);
assert!((0.0..=1.0).contains(&v), "running glow bounded, got {v}");
lo = lo.min(v);
hi = hi.max(v);
}
assert!(
hi - lo > 0.2,
"running status actually pulses (range {lo}..{hi})"
);
}
#[test]
fn builders_are_back_compat_and_additive() {
// The original two-step construction still works untouched…
let v = PipelineView::new("p", vec![("a".into(), Status::Done)], vec![]);
assert_eq!(v.counts(), (0, 0, 1, 0));
// …and the new builders chain without breaking it.
let v = v
.with_animation(false)
.with_speed(2.0)
.with_throughput(vec![Some("9k rows/s".into())]);
assert!(!v.state.animated);
assert_eq!(v.state.speed, 2.0);
assert_eq!(v.state.throughput.len(), 1);
let j = facett_core::Facet::state_json(&v);
assert_eq!(j["animated"], false);
}
#[test]
fn renders_headlessly_under_a_theme() {
let mut view = PipelineView::new(
"etl",
vec![
("raw".into(), Status::Done),
("clean".into(), Status::Running),
("agg".into(), Status::Failed),
("out".into(), Status::Idle),
],
vec![(1, 0), (2, 1), (3, 2)],
)
.with_throughput(vec![Some("12k/s".into()), Some("8k/s".into()), None]);
let r =
facett_core::harness::render_themed(&mut view, facett_core::Theme::cyberpunk_neon());
assert!(r.drew(), "themed pipeline drew vertices");
assert_eq!(r.state["dag"]["nodes"], 4);
// `drew()` cannot see colour, so "under a theme" was unverified: deleting
// `set_theme` from `render_themed` leaves the two assertions above green.
// Compare the ink cyberpunk_neon paints against the default palette's.
let mut plain = PipelineView::new(
"etl",
vec![
("raw".into(), Status::Done),
("clean".into(), Status::Running),
("agg".into(), Status::Failed),
("out".into(), Status::Idle),
],
vec![(1, 0), (2, 1), (3, 2)],
)
.with_throughput(vec![Some("12k/s".into()), Some("8k/s".into()), None]);
let base = facett_core::harness::headless_render(&mut plain);
assert_ne!(
r.ink_signature(),
base.ink_signature(),
"cyberpunk_neon painted the DAG exactly like the default palette — the \
theme never reached the paint",
);
}
#[test]
fn edges_render_as_smooth_splines_not_straight_lines() {
// #34: the DAG viewer draws each PipelineEdge as a cubic-Bézier SPLINE (reusing
// the shared `facett_core::edges::spline_points` helper), not a straight segment.
// Drive `paint_flow_edge` with a diagonal chord and prove the label anchor it
// returns — a point ON the curve at mid arc-length — bows measurably OFF the
// straight line between the endpoints. A straight edge would return a point ON
// that line (perpendicular distance ~0). RED if the spline regresses to a line.
let src = Pos2::new(0.0, 0.0);
let dst = Pos2::new(400.0, 120.0);
let scene = FlowScene::new(Graph {
nodes: vec![
Node {
id: "a".into(),
label: "a".into(),
position: [src.x, src.y],
group: "g".into(),
kind: 0,
},
Node {
id: "b".into(),
label: "b".into(),
position: [dst.x, dst.y],
group: "g".into(),
kind: 0,
},
],
edges: vec![Edge {
from: "a".into(),
to: "b".into(),
}],
})
.unwrap();
let anchor = scene.edge_point(0, 0.25).unwrap();
// Perpendicular distance of the anchor from the infinite src→dst line: |rel × dir|.
let dir = (dst - src).normalized();
let rel = anchor - src;
let perp = (rel.x * dir.y - rel.y * dir.x).abs();
assert!(
perp > 3.0,
"the edge bows off the chord as a spline (perp dist {perp:.2} px)"
);
// …and the anchor stays within the chord span (a bow, not a runaway control point).
let along = rel.x * dir.x + rel.y * dir.y;
assert!(
along > 0.0 && along < (dst - src).length(),
"anchor sits between the endpoints (t={along:.1})"
);
}
#[test]
fn point_at_arclen_walks_the_polyline() {
// The dash-marcher's arc-length walk: clamps at both ends, hits the midpoint of a
// straight sampled poly-line at its geometric middle.
let pts = [
Pos2::new(0.0, 0.0),
Pos2::new(10.0, 0.0),
Pos2::new(20.0, 0.0),
];
assert_eq!(
facett_core::edges::point_at_arclen(&pts, -5.0),
pts[0],
"clamps to the start"
);
assert_eq!(
facett_core::edges::point_at_arclen(&pts, 100.0),
pts[2],
"clamps to the end"
);
let mid = facett_core::edges::point_at_arclen(&pts, 10.0);
assert!(
(mid - Pos2::new(10.0, 0.0)).length() < 1e-3,
"arc-length 10 → the middle ({mid:?})"
);
}
#[test]
fn spline_dag_renders_clean_with_curved_edges_and_labels() {
// The whole DAG (curved edges + throughput labels anchored on the curve +
// arrowheads) renders headlessly with vertices and no panic — the spline path is
// snapshot-clean (the deck-wide error-atom scan covers it in facett-demo).
let mut view = PipelineView::new(
"etl",
vec![
("raw".into(), Status::Done),
("clean".into(), Status::Running),
("agg".into(), Status::Failed),
],
vec![(1, 0), (2, 1)],
)
.with_throughput(vec![Some("12k/s".into()), Some("8k/s".into())]);
let r =
facett_core::harness::render_themed(&mut view, facett_core::Theme::cyberpunk_neon());
assert!(r.drew(), "the spline DAG drew vertices");
assert_eq!(r.state["dag"]["edges"], 2);
}
// ── GAP #1: navigation (inject a DAG → fit/zoom/pan → assert real output) ──
/// A wide fan-out DAG so columns/rows span far enough that fit + zoom actually
/// change what's visible (real-in / real-out, not a no-panic smoke test).
fn wide_view() -> PipelineView {
let mut nodes = vec![PipelineNode::new("root", Status::Done)];
let mut edges = Vec::new();
for k in 0..8 {
nodes.push(PipelineNode::new(format!("leaf{k}"), Status::Idle));
edges.push((k + 1, 0)); // leaf reads root
}
PipelineView::from_nodes("wide", nodes, edges)
}
#[test]
fn fit_frames_the_whole_dag_and_zoom_in_culls_nodes() {
let rect = Rect::from_min_size(Pos2::new(0.0, 0.0), Vec2::new(400.0, 300.0));
let mut v = wide_view();
// FIT: the camera frames every node → all 9 boxes are visible.
v.fit_to(rect);
let after_fit = v.visible_node_count(rect);
assert_eq!(
after_fit, 9,
"fit shows the whole 9-node DAG, got {after_fit}"
);
let fit_scale = v.nav().scale;
assert!(
fit_scale > 0.0 && fit_scale < 1.0,
"a 9-node DAG is shrunk to fit a 400×300 pane, scale {fit_scale}"
);
// ZOOM IN hard at a corner anchor → fewer node boxes remain on screen.
for _ in 0..10 {
v.zoom_to_cursor(1.5, rect.left_top());
}
assert!(
v.nav().scale > fit_scale,
"zoom-in raised the scale ({} > {fit_scale})",
v.nav().scale
);
let after_zoom = v.visible_node_count(rect);
assert!(
after_zoom < after_fit,
"zooming in culls nodes off-screen ({after_zoom} < {after_fit})"
);
}
#[test]
fn zoom_to_cursor_keeps_the_anchor_scene_point_fixed() {
// The slippy-map invariant: the scene point under the cursor stays put.
let mut v = wide_view();
v.fit_to(Rect::from_min_size(Pos2::ZERO, Vec2::new(400.0, 300.0)));
let anchor = Pos2::new(180.0, 140.0);
let inv = |nav: &Navigable, s: Pos2| {
Pos2::new(
(s.x - nav.offset[0]) / nav.scale,
(s.y - nav.offset[1]) / nav.scale,
)
};
let before = inv(v.nav(), anchor);
v.zoom_to_cursor(2.0, anchor);
let after = inv(v.nav(), anchor);
assert!(
(before - after).length() < 1e-2,
"scene point under the cursor must stay fixed"
);
}
#[test]
fn pan_moves_the_camera_and_state_json_reports_nav() {
let mut v = wide_view();
v.fit_to(Rect::from_min_size(Pos2::ZERO, Vec2::new(400.0, 300.0)));
let before = v.nav().offset;
v.pan_pixels(120.0, -40.0);
let after = v.nav().offset;
assert!(
(after[0] - before[0] - 120.0).abs() < 1e-3,
"pan moved the camera in x"
);
assert!(
(after[1] - before[1] + 40.0).abs() < 1e-3,
"pan moved the camera in y"
);
// The nav camera is emitted as readable state (LAW: emit what you render).
let j = facett_core::Facet::state_json(&v);
assert_eq!(j["nav"]["fitted"], true);
assert!((j["nav"]["scale"].as_f64().unwrap() - v.nav().scale as f64).abs() < 1e-6);
assert!((j["nav"]["offset"][0].as_f64().unwrap() - after[0] as f64).abs() < 1e-3);
// caps now advertise navigable so the test matrix discovers the surface.
assert!(
facett_core::Facet::caps(&v).navigable,
"PipelineView advertises the navigable cap"
);
}
#[test]
fn per_node_throughput_and_lag_ride_state_json() {
// GAP #2/#3 hook: a node's live throughput/lag render + serialize.
let node = PipelineNode::new("clean", Status::Running)
.with_rows(12_000)
.with_throughput_rows_s(3400.0)
.with_lag_s(2.5);
let v = PipelineView::from_nodes("p", vec![node], vec![]);
let j = facett_core::Facet::state_json(&v);
assert_eq!(j["nodes"][0]["throughput_rows_s"], 3400.0);
assert_eq!(j["nodes"][0]["lag_s"], 2.5);
}
// ── FC-2 → FC-3 / FC-8 as *headless* properties via `facett_core::harness` ────
// feed a `Vec<Msg>` through the pure state-transition driver (no egui, no
// GPU) and assert the resulting `PipelineState` + the emitted `Effect`s. ───
fn sample() -> PipelineView {
PipelineView::new(
"etl",
vec![
("src".into(), Status::Done),
("clean".into(), Status::Running),
("sink".into(), Status::Idle),
],
vec![(1, 0), (2, 1)],
)
}
#[test]
fn harness_snapshot_drives_select_and_camera() {
use facett_core::harness;
let mut v = sample();
// Selecting a stage by name is observable headlessly (FC-2 → FC-3).
let snap = harness::snapshot(&mut v, [Msg::SelectNode("clean".into())]);
assert_eq!(
snap.selected.as_deref(),
Some("clean"),
"selection is observable headlessly"
);
assert_eq!(
&snap,
v.state(),
"the snapshot is a clone of the live state"
);
// Camera Msgs move the shared nav model, no render needed.
let base = v.nav().offset;
let snap = harness::snapshot(&mut v, [Msg::Pan(Vec2::new(15.0, -7.0))]);
assert!(
(snap.nav.offset[0] - base[0] - 15.0).abs() < 1e-3,
"Pan Msg shifts the camera"
);
// Tick advances the injected clock; SetClock pins it.
let snap = harness::snapshot(&mut v, [Msg::SetClock(2.0), Msg::Tick(0.5)]);
assert_eq!(
snap.clock_s,
Some(2.5),
"SetClock then Tick pins+advances the clock"
);
}
#[test]
fn drive_reports_preview_effect_and_state_round_trips() {
use facett_core::harness;
let mut v = sample();
// FC-8: click-selecting a stage asks the host to preview that dataset.
let effects = harness::drive(&mut v, [Msg::SelectNode("sink".into())]);
assert_eq!(
effects,
vec![Effect::PreviewRequest("sink".into())],
"SelectNode emits a PreviewRequest Effect"
);
// …and the concrete one-shot mailbox the host drains carries the same name.
assert_eq!(
v.take_preview_request().as_deref(),
Some("sink"),
"the mailbox mirrors the Effect"
);
assert!(
v.take_preview_request().is_none(),
"the request is one-shot"
);
// Camera Msgs raise no Effect.
let effects = harness::drive(&mut v, [Msg::Pan(Vec2::new(1.0, 1.0)), Msg::RequestRefit]);
assert!(effects.is_empty(), "camera Msgs are pure (no Effect)");
// FC-3: the observable Model round-trips through serde.
let json = serde_json::to_value(v.state()).unwrap();
let back: PipelineState = serde_json::from_value(json).unwrap();
assert_eq!(&back, v.state(), "serde(state) -> state round-trips");
}
#[test]
fn refit_then_fit_reframes_via_msgs() {
let rect = Rect::from_min_size(Pos2::ZERO, Vec2::new(400.0, 300.0));
let mut v = wide_view();
let _ = v.update(Msg::Fit(rect));
assert!(v.is_fitted(), "Fit Msg marks the camera fitted");
let _ = v.update(Msg::RequestRefit);
assert!(
!v.is_fitted(),
"RequestRefit clears fitted so the next paint reframes"
);
}
}