facett-pipeline/src/model.rs
//! The **engine-agnostic, normalized pipeline model** — the neutral shape every
//! engine's adapter produces and the [`PipelineView`](crate::PipelineView) renders.
//!
//! ## Why this exists (one viewer, many adapters)
//!
//! `facett-pipeline` used to be fed a Spark/SDP-typed model directly (knut's
//! `ViewerState` → `(name, Status)` nodes + `(from, to)` edges, mapped by the
//! consumer). That coupled "show a pipeline" to "show a *Spark* pipeline". This
//! module breaks that coupling: it defines a **renderer-independent**
//! [`PipelineModel`] (stages + dataflow edges + per-stage status/progress/metrics +
//! names/ids) plus a [`PipelineAdapter`] trait. An engine integration becomes
//! *"write a [`PipelineAdapter`]"* — Flink, Beam, skade-streaming are then "just
//! another impl", not a new viewer. It mirrors facett-graphview's
//! one-engine-many-adapters pattern.
//!
//! **Hard rule:** nothing here depends on Spark/SDP types. The model speaks
//! `String` ids, a neutral [`StageStatus`], and an open [`StageMetrics`] /
//! [`PipelineMetrics`] bag. The Spark adapter lives in **knut** (`knut-pipelines`,
//! behind its `facett-adapter` feature), mapping its typed `ViewerState` →
//! [`PipelineModel`].
//!
//! ## Shape
//!
//! * [`PipelineModel`] — the whole pipeline: `title`, `stages`, `edges`,
//! optional pipeline-level [`PipelineMetrics`].
//! * [`Stage`] — one node: stable `id` + display `name`, a [`StageStatus`], an
//! optional `kind` sublabel (e.g. the engine's output type), an optional
//! `error`, and per-stage [`StageMetrics`].
//! * [`StageStatus::Unknown`] + [`UnknownReason`] — *"we could not determine this
//! node's state, and here is which kind of not-knowing we hit"*. An adapter
//! whose probe came back empty maps onto this, **never** onto `Idle` (which
//! asserts the node has not started) and never onto `Done`. See the
//! [`StageStatus`] docs for why the difference is load-bearing.
//! * [`StatusTally`] — the five-way status count (`idle`/`running`/`done`/
//! `failed`/`unknown`) that always sums to the node count.
//! * [`PipelineEdge`] — a dataflow edge `from → to` between stage **ids**
//! (source → consumer), with optional `label` (e.g. a throughput annotation).
//! * [`StageMetrics`] / [`PipelineMetrics`] — neutral metric bags (rows, bytes,
//! throughput, lag, task counts, elapsed) every engine can populate the parts
//! of that it has.
//!
//! [`PipelineView::from_model`](crate::PipelineView::from_model) consumes a
//! [`PipelineModel`]; the legacy `new` / `from_nodes` builders remain for
//! back-compat and are now thin wrappers that build a model internally.
//!
//! ## DEFERRED (design note — seam left, not built)
//!
//! Two future directions are intentionally **not** built here; the trait seam +
//! the normalized model are exactly what makes them additive later:
//!
//! 1. **Render the DAG through `facett-graphview` (#17), with a status/metrics
//! OVERLAY.** Today [`PipelineView`](crate::PipelineView) paints its own themed
//! DAG (composing `facett-graph` for layout). The future render path is to lay
//! the [`PipelineModel`] out via `facett-graphview` (still spike-stage, #17) and
//! overlay each stage's [`StageStatus`] + [`StageMetrics`] (and pipeline-level
//! [`PipelineMetrics`]) on top. Because the model is renderer-independent, that
//! swap touches only the *view*, never the adapters: every engine adapter keeps
//! producing the same [`PipelineModel`]. Not started — graphview is a spike.
//!
//! 2. **More engine adapters: Flink, Beam, skade-streaming.** Each is "just
//! another `impl PipelineAdapter`" — map the engine's job/stage graph + live
//! metrics into [`PipelineModel`]. Streaming engines populate
//! [`StageMetrics::lag_s`] / `throughput_rows_s` (the fields already exist).
//! Only knut's Spark/SDP adapter ships now; the others are deferred, and the
//! trait is deliberately the *only* thing a new engine must satisfy.
use serde::{Deserialize, Serialize};
use crate::Status;
/// **Why a state could not be determined** — the machine-readable payload of
/// [`StageStatus::Unknown`] / [`Status::Unknown`].
///
/// This exists so "unknown" is never a bare shrug: a caller that cannot read a
/// node's state must say *which kind of not-knowing* it hit, and the view (and
/// any `state_json` consumer) can act on that without parsing prose. Every
/// variant is a fact about the **observation**, never about the node.
///
/// It is deliberately [`Copy`] + `'static` — no `String` — so [`StageStatus`] and
/// [`Status`] stay `Copy`, `Eq` and `Serialize`, exactly as they were before
/// `Unknown` existed. Every existing `.copied()` / `==` call site keeps working.
///
/// `#[non_exhaustive]`: new kinds of not-knowing will be found (they always are),
/// and adding one must not break a downstream `match`. Always keep a `_ =>` arm.
#[non_exhaustive]
#[derive(Clone, Copy, PartialEq, Eq, Default, Debug, Serialize, Deserialize)]
pub enum UnknownReason {
/// **Never looked.** No probe has been attempted for this node — the default,
/// and the one that matters most: it is what a freshly built ladder holds
/// before anything has been observed. Explicitly *not* "idle".
#[default]
NotProbed,
/// The probe was attempted and the target did not answer (timeout, refused,
/// DNS, no route). We asked; nothing came back.
Unreachable,
/// The probe reached the target and errored — a response arrived, but it was
/// not an answer to the question (5xx, malformed body, parse failure).
ProbeFailed,
/// We are not permitted to look: missing/expired credentials, RBAC denial, no
/// kubeconfig context. The state exists; we may not read it.
NotAuthorised,
/// Two or more sources disagree (e.g. the registry tag and the live pod image
/// differ). A single truthful answer cannot be picked, so we report neither.
Ambiguous,
/// The last observation succeeded but is too old to be trusted as current.
/// A stale green is still a green light over an unknown, so it is Unknown.
Stale,
/// The node's state is not applicable / not modelled by this source, and the
/// caller wants that recorded rather than papered over.
NotApplicable,
}
impl UnknownReason {
/// A short, stable, machine-parseable token (kebab-case, no spaces). Stable
/// across releases — `state_json` and text output both key on it.
pub fn token(self) -> &'static str {
match self {
UnknownReason::NotProbed => "not-probed",
UnknownReason::Unreachable => "unreachable",
UnknownReason::ProbeFailed => "probe-failed",
UnknownReason::NotAuthorised => "not-authorised",
UnknownReason::Ambiguous => "ambiguous",
UnknownReason::Stale => "stale",
UnknownReason::NotApplicable => "not-applicable",
// No `_` arm on purpose: `#[non_exhaustive]` does NOT relax
// exhaustiveness inside the defining crate, so adding a reason breaks
// this build until it is given a token — which is what we want.
}
}
/// A one-line human sentence for a tooltip / detail panel.
pub fn describe(self) -> &'static str {
match self {
UnknownReason::NotProbed => "never probed — no observation has been attempted",
UnknownReason::Unreachable => "probe sent, no answer (timeout / refused / no route)",
UnknownReason::ProbeFailed => "probe reached the target but errored",
UnknownReason::NotAuthorised => "not permitted to read this state",
UnknownReason::Ambiguous => "sources disagree — no single truthful answer",
UnknownReason::Stale => "last observation is too old to be trusted",
UnknownReason::NotApplicable => "this source does not model that state",
}
}
}
/// Neutral run status of a single pipeline stage. The engine adapter maps its own
/// status enum onto this — it carries no engine semantics. Identical in spirit to
/// the render-layer [`Status`], but lives in the model so the model needn't import
/// the view's enum to describe state. They convert freely (`From` both ways).
///
/// ## `Idle` is a fact; `Unknown` is the absence of one
///
/// For a **pipeline**, `Idle` is honest: a stage that has not run really is idle,
/// and the engine knows it has not run. For a **promotion ladder** (or any view
/// whose nodes are live environments rather than steps of one run) it is a lie:
/// *"we have never observed this environment"* and *"this environment is idle"*
/// are different facts, and rendering the first as the second is a **fake green** —
/// a neutral grey box that reads "fine, just not running" over something nobody
/// has looked at. That pattern has been found repeatedly in this codebase (a DHCP
/// *request* read as a lease; a handshake *initiation* read as a completed
/// handshake) and it ships a green light over something that is not working.
///
/// So an observer that could not determine a node's state must say
/// [`StageStatus::Unknown`] with an [`UnknownReason`] — never `Idle`, never `Done`.
/// `Unknown` renders **amber with a dashed border and a `?` marker**, not the
/// neutral dim grey of `Idle` (see [`Status::theme_color`]).
///
/// `Idle` remains `#[default]` and every existing variant is untouched: this is a
/// purely additive change. The enum is `#[non_exhaustive]` so the *next* variant
/// is not a breaking change either — downstream `match`es need a `_ =>` arm.
#[non_exhaustive]
#[derive(Clone, Copy, PartialEq, Eq, Default, Debug)]
pub enum StageStatus {
/// Not started / no signal yet — and we **know** it has not started. Use
/// [`Unknown`](StageStatus::Unknown) when you could not find out.
#[default]
Idle,
/// Currently executing.
Running,
/// Completed successfully.
Done,
/// Failed (see the stage's `error`).
Failed,
/// **State could not be determined**, with the machine-readable reason why.
/// Not a status of the node — a status of our knowledge of it.
Unknown(UnknownReason),
}
impl StageStatus {
/// [`Unknown`](StageStatus::Unknown) with the given reason — the spelling to
/// reach for when a probe could not answer.
pub fn unknown(reason: UnknownReason) -> Self {
StageStatus::Unknown(reason)
}
/// The reason, when this is [`Unknown`](StageStatus::Unknown); `None` otherwise.
pub fn unknown_reason(self) -> Option<UnknownReason> {
match self {
StageStatus::Unknown(r) => Some(r),
_ => None,
}
}
/// `true` for [`Unknown`](StageStatus::Unknown), whatever the reason.
pub fn is_unknown(self) -> bool {
matches!(self, StageStatus::Unknown(_))
}
/// A short, stable label for text/JSON. For `Unknown` it **carries the
/// reason** — `"unknown:unreachable"` — so a text/JSON reader never sees a
/// bare `"unknown"` it has to go hunting for. Use
/// [`family`](StageStatus::family) for the bare `"unknown"` (legend chips).
pub fn label(self) -> &'static str {
match self {
StageStatus::Idle => "idle",
StageStatus::Running => "running",
StageStatus::Done => "done",
StageStatus::Failed => "failed",
StageStatus::Unknown(UnknownReason::NotProbed) => "unknown:not-probed",
StageStatus::Unknown(UnknownReason::Unreachable) => "unknown:unreachable",
StageStatus::Unknown(UnknownReason::ProbeFailed) => "unknown:probe-failed",
StageStatus::Unknown(UnknownReason::NotAuthorised) => "unknown:not-authorised",
StageStatus::Unknown(UnknownReason::Ambiguous) => "unknown:ambiguous",
StageStatus::Unknown(UnknownReason::Stale) => "unknown:stale",
StageStatus::Unknown(UnknownReason::NotApplicable) => "unknown:not-applicable",
}
}
/// The status **family** without the reason (`"idle"`, `"running"`, `"done"`,
/// `"failed"`, `"unknown"`) — for a legend chip or a tally key, where the
/// reason belongs on the node, not the header.
pub fn family(self) -> &'static str {
match self {
StageStatus::Unknown(_) => "unknown",
other => other.label(),
}
}
}
impl From<StageStatus> for Status {
fn from(s: StageStatus) -> Self {
match s {
StageStatus::Idle => Status::Idle,
StageStatus::Running => Status::Running,
StageStatus::Done => Status::Done,
StageStatus::Failed => Status::Failed,
StageStatus::Unknown(r) => Status::Unknown(r),
}
}
}
impl From<Status> for StageStatus {
fn from(s: Status) -> Self {
match s {
Status::Idle => StageStatus::Idle,
Status::Running => StageStatus::Running,
Status::Done => StageStatus::Done,
Status::Failed => StageStatus::Failed,
Status::Unknown(r) => StageStatus::Unknown(r),
}
}
}
/// A full five-way status tally — the honest counterpart of the legacy
/// four-tuple [`PipelineModel::status_counts`] / [`PipelineView::counts`], which
/// predate `Unknown` and therefore cannot report it.
///
/// **Read this one when the total matters.** `idle + running + done + failed` no
/// longer equals the node count once anything is `Unknown`; that shortfall is
/// exactly the silent-drop this type exists to prevent.
#[derive(Clone, Copy, Default, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct StatusTally {
/// Known to have not started.
pub idle: usize,
/// Known to be executing.
pub running: usize,
/// Known to have succeeded.
pub done: usize,
/// Known to have failed.
pub failed: usize,
/// **State could not be determined** — any [`StageStatus::Unknown`] reason.
pub unknown: usize,
}
impl StatusTally {
/// Count one status into the tally.
pub fn add(&mut self, s: StageStatus) {
match s {
StageStatus::Idle => self.idle += 1,
StageStatus::Running => self.running += 1,
StageStatus::Done => self.done += 1,
StageStatus::Failed => self.failed += 1,
// Every not-determined reason folds into one bucket; the per-node
// reason stays on the node (`state_json`'s `unknown_reason`).
StageStatus::Unknown(_) => self.unknown += 1,
// No `_` arm: a future `StageStatus` variant must be *given* a bucket
// here, never silently swept into a known one. In-crate exhaustiveness
// still applies under `#[non_exhaustive]`, so the compiler enforces it.
}
}
/// Every node accounted for — the sum of all five buckets.
pub fn total(self) -> usize {
self.idle + self.running + self.done + self.failed + self.unknown
}
/// The legacy `(idle, running, done, failed)` shape. **Lossy** — it drops
/// `unknown` entirely; kept only so the pre-`Unknown` API is unchanged.
pub fn as_legacy_tuple(self) -> (usize, usize, usize, usize) {
(self.idle, self.running, self.done, self.failed)
}
}
/// Per-stage metrics — a neutral, open bag. Every field is optional so an adapter
/// fills only what its engine actually reports; the renderer surfaces the parts
/// that are present. Spark fills `rows`/`throughput_rows_s`; a streaming engine
/// (Flink/Beam) additionally fills `lag_s`; a batch engine may fill none.
#[derive(Clone, Default, Debug, PartialEq)]
pub struct StageMetrics {
/// Rows this stage has materialised/processed (`None` = unknown).
pub rows: Option<u64>,
/// Live throughput in rows/sec (`None` = unknown).
pub throughput_rows_s: Option<f32>,
/// Streaming lag / backlog in seconds (`None` = unknown).
pub lag_s: Option<f32>,
}
impl StageMetrics {
/// `true` when no metric is present (so the renderer can skip a metrics line).
pub fn is_empty(&self) -> bool {
self.rows.is_none() && self.throughput_rows_s.is_none() && self.lag_s.is_none()
}
}
/// One stage/dataset node in the normalized pipeline.
#[derive(Clone, Default, Debug)]
pub struct Stage {
/// Stable identity — what [`PipelineEdge`] endpoints reference. Often equal to
/// `name`, but kept distinct so an engine with opaque ids (Flink vertex ids)
/// can carry them without colliding on display labels.
pub id: String,
/// Human display name.
pub name: String,
/// Run status.
pub status: StageStatus,
/// Optional kind/sublabel (e.g. the engine's output type — `materialized_view`,
/// `table`, …). Rendered under the name; falls back to the status label.
pub kind: Option<String>,
/// Error message when [`StageStatus::Failed`] (surfaced by the view).
pub error: Option<String>,
/// Per-stage metrics (rows / throughput / lag).
pub metrics: StageMetrics,
}
impl Stage {
/// A stage whose id and name are the same string, with a status.
pub fn new(name: impl Into<String>, status: StageStatus) -> Self {
let name = name.into();
Stage {
id: name.clone(),
name,
status,
..Default::default()
}
}
/// Set a distinct stable id (when it differs from the display name).
pub fn with_id(mut self, id: impl Into<String>) -> Self {
self.id = id.into();
self
}
/// Set the kind/sublabel (e.g. the engine's output type).
pub fn with_kind(mut self, kind: impl Into<String>) -> Self {
self.kind = Some(kind.into());
self
}
/// Attach an error message (for a failed stage).
pub fn with_error(mut self, error: impl Into<String>) -> Self {
self.error = Some(error.into());
self
}
/// Replace the metrics bag.
pub fn with_metrics(mut self, metrics: StageMetrics) -> Self {
self.metrics = metrics;
self
}
/// Set the rows metric.
pub fn with_rows(mut self, rows: u64) -> Self {
self.metrics.rows = Some(rows);
self
}
}
/// A dataflow edge in the normalized pipeline: data flows `from → to` (source →
/// consumer). Endpoints are stage **ids** (resolved to indices by the view).
#[derive(Clone, Default, Debug, PartialEq, Eq)]
pub struct PipelineEdge {
/// Source stage id (the upstream / producer).
pub from: String,
/// Target stage id (the downstream / consumer).
pub to: String,
/// Optional edge label (e.g. a throughput annotation `"12k rows/s"`).
pub label: Option<String>,
}
impl PipelineEdge {
/// A bare `from → to` edge.
pub fn new(from: impl Into<String>, to: impl Into<String>) -> Self {
PipelineEdge {
from: from.into(),
to: to.into(),
label: None,
}
}
/// Attach a label (rendered at the edge midpoint).
pub fn with_label(mut self, label: impl Into<String>) -> Self {
self.label = Some(label.into());
self
}
}
/// Pipeline-level aggregate progress/metrics — the neutral analogue of a "job
/// progress" bar. Every field optional; an adapter fills what its engine reports.
#[derive(Clone, Default, Debug, PartialEq)]
pub struct PipelineMetrics {
/// Overall completed fraction in `[0,1]` (`None` = unknown).
pub fraction: Option<f32>,
/// Tasks/units completed so far.
pub tasks_done: u64,
/// Total tasks/units planned.
pub tasks_total: u64,
/// Tasks in flight right now.
pub inflight: u64,
/// Rows processed across the run.
pub rows: u64,
/// Input bytes read across the run.
pub bytes_read: u64,
/// Wall-clock seconds elapsed since the run started.
pub elapsed_s: f32,
}
impl PipelineMetrics {
/// `true` when nothing has been reported (so the view can skip the bar).
pub fn is_empty(&self) -> bool {
self.fraction.is_none()
&& self.tasks_total == 0
&& self.tasks_done == 0
&& self.rows == 0
&& self.bytes_read == 0
&& self.elapsed_s == 0.0
&& self.inflight == 0
}
}
/// The **normalized, engine-agnostic pipeline model**: a title, the stages
/// (nodes), the dataflow edges (by stage id), and optional pipeline-level
/// metrics. This is the single shape [`PipelineView`](crate::PipelineView)
/// renders, and the single shape every [`PipelineAdapter`] produces.
#[derive(Clone, Default, Debug)]
pub struct PipelineModel {
/// Pipeline title.
pub title: String,
/// The stage nodes, in a stable order.
pub stages: Vec<Stage>,
/// Dataflow edges between stage ids (source → consumer).
pub edges: Vec<PipelineEdge>,
/// Optional pipeline-level progress/metrics.
pub metrics: Option<PipelineMetrics>,
}
impl PipelineModel {
/// An empty model with a title.
pub fn new(title: impl Into<String>) -> Self {
PipelineModel {
title: title.into(),
..Default::default()
}
}
/// Builder: append a stage.
pub fn with_stage(mut self, stage: Stage) -> Self {
self.stages.push(stage);
self
}
/// Builder: append an edge.
pub fn with_edge(mut self, edge: PipelineEdge) -> Self {
self.edges.push(edge);
self
}
/// Builder: set the pipeline-level metrics.
pub fn with_metrics(mut self, metrics: PipelineMetrics) -> Self {
self.metrics = Some(metrics);
self
}
/// Index of the stage with this id, if present (used to resolve edges).
pub fn stage_index(&self, id: &str) -> Option<usize> {
self.stages.iter().position(|s| s.id == id)
}
/// `(idle, running, done, failed)` stage-status counts.
///
/// **Lossy since `Unknown` exists:** a stage whose state could not be
/// determined is in *none* of these four buckets, so the four numbers no
/// longer have to sum to `stages.len()`. That is deliberate — an unknown
/// stage must never be counted as idle. Use [`status_tally`](Self::status_tally)
/// when you need every stage accounted for. Signature and the four existing
/// numbers are unchanged, so every pre-`Unknown` caller still compiles and
/// still gets the same answer for a model with no unknowns.
pub fn status_counts(&self) -> (usize, usize, usize, usize) {
self.status_tally().as_legacy_tuple()
}
/// The full five-way [`StatusTally`] — the four legacy buckets **plus**
/// `unknown`, summing to `stages.len()`.
pub fn status_tally(&self) -> StatusTally {
let mut t = StatusTally::default();
for s in &self.stages {
t.add(s.status);
}
t
}
}
/// **The seam.** An adapter maps one engine's specifics → the normalized
/// [`PipelineModel`]. Implement this once per engine; the viewer never learns the
/// engine's types.
///
/// The Spark/SDP adapter lives in knut (`knut_pipelines::FacettPipelineAdapter`,
/// feature `facett-adapter`). Future engines — **Flink**, **Beam**,
/// **skade-streaming** — are "just another `impl PipelineAdapter`" (DEFERRED; see
/// the crate docs). Keep the trait the *only* thing a new engine must satisfy.
pub trait PipelineAdapter {
/// Produce the normalized model for the current engine state.
fn to_pipeline_model(&self) -> PipelineModel;
}
#[cfg(test)]
mod tests {
use super::*;
/// Every reason, so a new one cannot be added without passing through here.
const ALL_REASONS: [UnknownReason; 7] = [
UnknownReason::NotProbed,
UnknownReason::Unreachable,
UnknownReason::ProbeFailed,
UnknownReason::NotAuthorised,
UnknownReason::Ambiguous,
UnknownReason::Stale,
UnknownReason::NotApplicable,
];
#[test]
fn status_roundtrips_with_view_status() {
let mut all = vec![
StageStatus::Idle,
StageStatus::Running,
StageStatus::Done,
StageStatus::Failed,
];
all.extend(ALL_REASONS.map(StageStatus::Unknown));
for s in all {
let v: Status = s.into();
let back: StageStatus = v.into();
assert_eq!(s, back, "StageStatus <-> Status round-trips");
assert_eq!(s.label(), v.label(), "labels agree across the boundary");
assert_eq!(s.family(), v.family(), "families agree across the boundary");
}
}
/// ⚠ **Additive guarantee.** `Idle` is still the default and the four original
/// variants are untouched — nothing about a pre-`Unknown` model changed.
#[test]
fn idle_is_still_the_default_and_the_old_labels_are_unchanged() {
assert_eq!(StageStatus::default(), StageStatus::Idle);
assert_eq!(Status::default(), Status::Idle);
assert_eq!(StageStatus::Idle.label(), "idle");
assert_eq!(StageStatus::Running.label(), "running");
assert_eq!(StageStatus::Done.label(), "done");
assert_eq!(StageStatus::Failed.label(), "failed");
assert_eq!(Stage::default().status, StageStatus::Idle);
}
/// ⚠ **The core distinction.** `Unknown` is equal to neither `Idle` nor `Done`,
/// and two different reasons are different values — so nothing downstream can
/// collapse "never observed" onto "not started" by accident.
#[test]
fn unknown_is_neither_idle_nor_done_and_reasons_are_distinct() {
let u = StageStatus::Unknown(UnknownReason::NotProbed);
assert_ne!(u, StageStatus::Idle, "unobserved is NOT idle");
assert_ne!(u, StageStatus::Done, "unobserved is NOT done — that is the fake green");
assert_ne!(u, StageStatus::Unknown(UnknownReason::Unreachable), "reasons are part of the value");
assert!(u.is_unknown());
assert!(!StageStatus::Idle.is_unknown());
assert_eq!(u.unknown_reason(), Some(UnknownReason::NotProbed));
assert_eq!(StageStatus::Idle.unknown_reason(), None);
assert_eq!(StageStatus::unknown(UnknownReason::Stale), StageStatus::Unknown(UnknownReason::Stale));
}
/// ⚠ **`label()` carries the reason.** Revert it to a bare `"unknown"` and this
/// goes red: a text/JSON reader must never have to guess *why* we don't know.
#[test]
fn the_label_carries_the_reason_and_family_does_not() {
let expected = [
(UnknownReason::NotProbed, "unknown:not-probed", "not-probed"),
(UnknownReason::Unreachable, "unknown:unreachable", "unreachable"),
(UnknownReason::ProbeFailed, "unknown:probe-failed", "probe-failed"),
(UnknownReason::NotAuthorised, "unknown:not-authorised", "not-authorised"),
(UnknownReason::Ambiguous, "unknown:ambiguous", "ambiguous"),
(UnknownReason::Stale, "unknown:stale", "stale"),
(UnknownReason::NotApplicable, "unknown:not-applicable", "not-applicable"),
];
assert_eq!(expected.len(), ALL_REASONS.len(), "every reason is pinned here");
for (r, label, token) in expected {
let s = StageStatus::Unknown(r);
assert_eq!(s.label(), label, "the label names the reason");
assert_eq!(r.token(), token);
assert!(s.label().starts_with("unknown:"), "…and stays parseable as unknown");
assert_eq!(s.family(), "unknown", "the family is the bare word (legend chips)");
assert_ne!(s.label(), "idle", "and can never be mistaken for idle");
assert!(!r.describe().is_empty(), "every reason explains itself in prose");
}
// Tokens are unique — a machine reader can key on them.
let mut tokens: Vec<&str> = ALL_REASONS.iter().map(|r| r.token()).collect();
tokens.sort_unstable();
let n = tokens.len();
tokens.dedup();
assert_eq!(tokens.len(), n, "reason tokens are unique");
assert_eq!(UnknownReason::default(), UnknownReason::NotProbed);
}
/// ⚠ **The tally accounts for everything.** Fold `Unknown` into the `idle`
/// bucket (the silent-drop this exists to stop) and `idle` goes to 2 here.
#[test]
fn the_tally_counts_unknown_separately_and_totals_every_stage() {
let m = PipelineModel::new("p")
.with_stage(Stage::new("a", StageStatus::Done))
.with_stage(Stage::new("b", StageStatus::Idle))
.with_stage(Stage::new("c", StageStatus::Unknown(UnknownReason::Unreachable)))
.with_stage(Stage::new("d", StageStatus::Unknown(UnknownReason::NotProbed)));
let t = m.status_tally();
assert_eq!(t.unknown, 2, "both undetermined stages land in `unknown`");
assert_eq!(t.idle, 1, "and NOT in `idle` — only the genuinely idle stage is there");
assert_eq!(t.done, 1);
assert_eq!(t.total(), 4, "every stage is accounted for");
// The legacy tuple is unchanged in shape and deliberately lossy.
assert_eq!(m.status_counts(), (1, 0, 1, 0));
assert_eq!(t.as_legacy_tuple(), m.status_counts());
let (i, r, d, f) = m.status_counts();
assert_eq!(i + r + d + f, 2, "the legacy tuple drops the unknowns — by design");
}
#[test]
fn model_builders_and_lookups() {
let m = PipelineModel::new("p")
.with_stage(Stage::new("a", StageStatus::Done))
.with_stage(Stage::new("b", StageStatus::Running).with_kind("table").with_rows(10))
.with_edge(PipelineEdge::new("a", "b").with_label("9k/s"))
.with_metrics(PipelineMetrics { tasks_done: 3, tasks_total: 4, ..Default::default() });
assert_eq!(m.stages.len(), 2);
assert_eq!(m.edges.len(), 1);
assert_eq!(m.stage_index("b"), Some(1));
assert_eq!(m.stage_index("missing"), None);
assert_eq!(m.status_counts(), (0, 1, 1, 0));
assert_eq!(m.stages[1].metrics.rows, Some(10));
assert!(!m.metrics.as_ref().unwrap().is_empty());
}
#[test]
fn metrics_emptiness() {
assert!(StageMetrics::default().is_empty());
assert!(!StageMetrics { rows: Some(1), ..Default::default() }.is_empty());
assert!(PipelineMetrics::default().is_empty());
assert!(!PipelineMetrics { rows: 1, ..Default::default() }.is_empty());
}
/// A trivial in-test adapter proves the seam compiles + produces a model that
/// the view can consume (real-in/real-out, no engine types involved).
#[test]
fn adapter_trait_produces_a_model() {
struct Toy;
impl PipelineAdapter for Toy {
fn to_pipeline_model(&self) -> PipelineModel {
PipelineModel::new("toy")
.with_stage(Stage::new("x", StageStatus::Done))
.with_stage(Stage::new("y", StageStatus::Failed).with_error("boom"))
.with_edge(PipelineEdge::new("x", "y"))
}
}
let m = Toy.to_pipeline_model();
assert_eq!(m.status_counts(), (0, 0, 1, 1));
assert_eq!(m.stages[1].error.as_deref(), Some("boom"));
}
}