facett-pipeline/tests/robot_flowsim.rs
//! **The GLOWING SPLINE, proven on real pixels.** (`--features flowsim`)
//!
//! `facett-flowsim` advects particles along each DAG edge at a speed/density driven
//! by the stage's live throughput. This file renders REAL wgpu framebuffers through
//! `egui_kittest` and asserts on the pixels — not on `state_json`, not on "a frame
//! was drawn".
//!
//! Every claim here is PAIRED with its counterexample, because each one alone has a
//! trivially-green twin:
//!
//! | claim | the lie it catches |
//! |---|---|
//! | live DAG: overlay ON ≠ overlay OFF | the overlay never reached the pixels |
//! | dead DAG: overlay ON **==** overlay OFF, byte-identical | the overlay is DECORATION — it glows over a pipeline that has stopped |
//! | live DAG: frame(t₀) ≠ frame(t₁) | the particles are painted but frozen |
//! | dead DAG: frame(t₀) **==** frame(t₁) | a stopped pipeline is animating anyway |
//! | dead DAG frames are non-blank | the "static fallback" lost the base curve + arrowhead |
//!
//! The two DAGs differ ONLY in `throughput_rows_s`, and each is always compared
//! against a frame of ITSELF, so the stage labels are identical across every pair
//! and the diffed pixels are the overlay and nothing else.
//!
//! Run: `cargo test -p facett-pipeline --features flowsim --test robot_flowsim`
//! Pixel legs SKIP loudly without a wgpu backend.
#![cfg(feature = "flowsim")]
use egui_kittest::Harness;
use facett_core::Facet;
use facett_flowsim::{FlowMode, FlowMsg};
use facett_pipeline::{PipelineNode, PipelineView, Status};
/// A 4-stage ETL DAG. `live` decides only whether the stages report a throughput —
/// everything else (names, statuses, row counts, hence every drawn label) is
/// identical, so a live/dead pixel diff is the overlay alone.
fn dag(live: bool) -> Vec<PipelineNode> {
let tp = |n: &str, st: Status, rows: u64, r: f32| {
let node = PipelineNode::new(n, st).with_rows(rows);
if live { node.with_throughput_rows_s(r) } else { node }
};
vec![
tp("raw", Status::Done, 120_000, 500.0),
tp("clean", Status::Running, 48_000, 9_000.0),
tp("agg", Status::Running, 12_000, 2_000.0),
tp("out", Status::Running, 4_000, 300.0),
]
}
const EDGES: [(usize, usize); 3] = [(1, 0), (2, 1), (3, 2)];
/// Build the view with the overlay in `mode` and the clock pinned to `t` (FC-7 —
/// without a pinned clock two frames could never be compared).
fn view(live: bool, mode: FlowMode, t: f64) -> PipelineView {
let mut v = PipelineView::from_nodes("ETL", dag(live), EDGES.to_vec())
.with_animation(false) // kill the marching-ants + glow pulse: the overlay is the only motion
.with_clock(t)
.with_flowsim(true);
v.flowsim_mut().unwrap().update(FlowMsg::SetMode(mode));
v
}
/// A **branching** DAG (`ingest → {parse, decode} → join → sink`). The layout puts
/// the fan-out siblings on separate rows, so the edges run diagonally and are long
/// enough for the Bézier curvature — and the particles riding it — to actually read.
/// A straight left-to-right chain packs the nodes edge-to-edge and hides the spline.
fn branching_view(live: bool, mode: FlowMode, t: f64) -> PipelineView {
let tp = |n: &str, st: Status, rows: u64, r: f32| {
let node = PipelineNode::new(n, st).with_rows(rows);
if live { node.with_throughput_rows_s(r) } else { node }
};
let nodes = vec![
tp("ingest", Status::Done, 240_000, 400.0),
tp("parse", Status::Running, 220_000, 12_000.0),
tp("decode", Status::Running, 60_000, 1_500.0),
tp("join", Status::Running, 180_000, 6_000.0),
tp("sink", Status::Running, 30_000, 900.0),
];
// (consumer, source): parse+decode read ingest, join reads both, sink reads join.
let edges = vec![(1, 0), (2, 0), (3, 1), (3, 2), (4, 3)];
let mut v = PipelineView::from_nodes("stream", nodes, edges).with_animation(false).with_clock(t).with_flowsim(true);
v.flowsim_mut().unwrap().update(FlowMsg::SetMode(mode));
v
}
/// The same DAG with its edges REMOVED — the baseline that isolates the static edge
/// art (base curve + arrowheads) from the node boxes and the themed canvas.
fn edgeless_view() -> PipelineView {
PipelineView::from_nodes("ETL", dag(false), Vec::new()).with_animation(false).with_clock(0.0)
}
/// Render one real wgpu frame, or `None` with a loud note when there is no adapter.
fn frame(live: bool, mode: FlowMode, t: f64) -> Option<image::RgbaImage> {
render_with(move |ui| view(live, mode, t).ui(ui))
}
fn render_with(build: impl Fn(&mut egui::Ui) + 'static) -> Option<image::RgbaImage> {
let mut h = Harness::builder().with_size(egui::vec2(900.0, 620.0)).build_ui(move |ui| {
facett_core::set_theme(ui.ctx(), facett_core::Theme::sci_fi());
build(ui);
});
h.run_steps(3);
match h.render() {
Ok(img) => Some(img),
Err(e) => {
eprintln!("[robot] no wgpu backend ({e}) — flowsim pixel leg skipped");
None
}
}
}
/// Pixels differing by more than a hair of dithering noise.
fn diff_px(a: &image::RgbaImage, b: &image::RgbaImage) -> usize {
assert_eq!(a.dimensions(), b.dimensions());
a.pixels()
.zip(b.pixels())
.filter(|(p, q)| {
(0..3).map(|i| (p.0[i] as i32 - q.0[i] as i32).abs()).sum::<i32>() > 12
})
.count()
}
/// Pixels that are not the window background — the "is there a picture at all" measure.
fn painted_px(img: &image::RgbaImage) -> usize {
let bg = *img.get_pixel(2, img.height() - 3);
img.pixels()
.filter(|p| (0..3).map(|i| (p.0[i] as i32 - bg.0[i] as i32).abs()).sum::<i32>() > 24)
.count()
}
/// **THE VISUAL PROOF.** All five paired pixel claims in one wgpu session.
#[test]
fn particles_reach_the_pixels_and_a_dead_pipeline_stays_dark() {
let Some(live_on_t0) = frame(true, FlowMode::Particles, 0.0) else { return };
let live_off = frame(true, FlowMode::Off, 0.0).expect("second render");
let live_on_t1 = frame(true, FlowMode::Particles, 0.42).expect("third render");
let dead_on_t0 = frame(false, FlowMode::Particles, 0.0).expect("fourth render");
let dead_off = frame(false, FlowMode::Off, 0.0).expect("fifth render");
let dead_on_t1 = frame(false, FlowMode::Particles, 0.42).expect("sixth render");
// ── 1. The overlay REACHES the pixels on a live DAG. ──────────────────────
// The floor is DERIVED, not guessed: the plan seeds `n` particles and each is a
// ~3.4 px-radius two-ring dot, so a faithful render must light well over 15 px
// apiece. A magic constant here would drift silently as the plan changes.
let n_particles = view(true, FlowMode::Particles, 0.0).state_json()["flowsim"]["total_particles"]
.as_u64()
.unwrap() as usize;
assert!(n_particles > 0, "the fixture must plan particles at all");
let floor = n_particles * 15;
let overlay_px = diff_px(&live_on_t0, &live_off);
eprintln!(
"[robot] live DAG: overlay changed {overlay_px} px vs the same frame with it Off \
({n_particles} particles planned, floor {floor})"
);
assert!(
overlay_px > floor,
"the flow overlay must visibly reach the framebuffer — only {overlay_px} px differ \
from the Off frame, for {n_particles} planned particles"
);
// ── 2. THE ANCHOR: on a DEAD DAG the overlay adds NOTHING. ────────────────
// Byte-identical, not "close". If a stopped pipeline glowed like a running one
// the feature would be decoration, and claim 1 above would prove nothing.
let dead_overlay_px = diff_px(&dead_on_t0, &dead_off);
eprintln!("[robot] dead DAG: overlay changed {dead_overlay_px} px (must be 0)");
assert_eq!(
dead_overlay_px, 0,
"a pipeline with zero throughput must not glow — the overlay lit {dead_overlay_px} px \
over stages that are moving no rows"
);
// ── 3. The particles MOVE. ────────────────────────────────────────────────
// Calibrated against the overlay's OWN footprint from claim 1: if the dots were
// painted but frozen this is exactly 0, and if they merely jittered it would be a
// small fraction. Requiring a third of the footprint to change means the dots
// genuinely relocated along their edges.
let motion_px = diff_px(&live_on_t0, &live_on_t1);
eprintln!(
"[robot] live DAG: {motion_px} px changed between t=0.0 and t=0.42 \
({}% of the {overlay_px} px overlay footprint)",
100 * motion_px / overlay_px.max(1)
);
assert!(
motion_px * 3 > overlay_px,
"the particles must advance on screen between two clocks — only {motion_px} px moved, \
against an overlay footprint of {overlay_px} px"
);
// ── 4. …and a dead pipeline does NOT animate. ─────────────────────────────
let dead_motion_px = diff_px(&dead_on_t0, &dead_on_t1);
assert_eq!(
dead_motion_px, 0,
"a dead pipeline must be perfectly still — {dead_motion_px} px moved between clocks"
);
// ── 5. The STATIC FALLBACK survives: base curve + arrowhead still drawn. ──
// Without this, claims 2 and 4 would also pass for a view that rendered nothing
// whatsoever. Measured against the SAME DAG with its edges removed, so the diff
// is the edge art alone — the node boxes and the themed canvas cancel out.
// (A whole-frame "is it non-blank" count would be ~95 % here just from the canvas
// fill, and could not tell a missing arrowhead from a present one.)
let dead_content = painted_px(&dead_on_t0);
let total = (dead_on_t0.width() * dead_on_t0.height()) as usize;
let edgeless = render_with(|ui| {
edgeless_view().ui(ui);
})
.expect("edgeless render");
let edge_art_px = diff_px(&dead_on_t0, &edgeless);
eprintln!(
"[robot] dead DAG paints {dead_content}/{total} px overall, of which {edge_art_px} px \
is edge art absent from the edgeless DAG (base curve + arrowheads)"
);
assert!(
edge_art_px > 500,
"with zero throughput the STATIC fallback must still draw the base curve and the \
arrowhead — only {edge_art_px} px separate the dead DAG from one with no edges at all"
);
#[cfg(feature = "testmatrix")]
facett_core::testmatrix::emit(
"facett-pipeline.robot_flowsim",
"particles_reach_the_pixels",
overlay_px > floor && dead_overlay_px == 0 && motion_px * 3 > overlay_px && dead_motion_px == 0,
&format!("overlay={overlay_px}px motion={motion_px}px edge_art={edge_art_px}px dead_overlay=0 dead_motion=0"),
);
}
/// **See the frame.** Writes the live-flow and dead-pipeline pictures side by side
/// under `tests/snapshots/` so the glowing spline is something a human can look at.
#[test]
fn flowsim_png_blobs_live_and_dead() {
let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/snapshots");
std::fs::create_dir_all(&dir).unwrap();
let shot = |live: bool, mode: FlowMode| render_with(move |ui| branching_view(live, mode, 0.31).ui(ui));
let Some(live) = shot(true, FlowMode::Particles) else { return };
let dead = shot(false, FlowMode::Particles).expect("dead render");
let arrows = shot(true, FlowMode::Arrows).expect("arrows render");
let (lp, dp, ap) = (
dir.join("flowsim_live.png"),
dir.join("flowsim_dead.png"),
dir.join("flowsim_arrows.png"),
);
live.save(&lp).unwrap();
dead.save(&dp).unwrap();
arrows.save(&ap).unwrap();
for p in [&lp, &dp, &ap] {
let n = std::fs::metadata(p).unwrap().len();
assert!(n > 1000, "{} is a real frame ({n} bytes)", p.display());
eprintln!("[robot] wrote {} ({n} B)", p.display());
}
// The three looks must be three DIFFERENT pictures, or the mode switch is a no-op.
assert!(diff_px(&live, &dead) > 300, "particles vs dead pipeline must look different");
assert!(diff_px(&live, &arrows) > 300, "particles vs the arrows fallback must look different");
assert!(diff_px(&arrows, &dead) > 100, "the arrows fallback must draw something of its own");
}
/// **The host binding, at the introspection surface.** `state_json` must report the
/// SAME plan the pixels came from: a live DAG flows on every edge, a dead one on
/// none, and density tracks the relative row counts rather than collapsing.
///
/// The density half is the one that caught a real bug: the binding fed `plan_edges`
/// RAW row counts, and `density_per_edge × 48 000` clamps to `max_density` exactly
/// like `× 120 000` does — so every edge had identical density and the "density =
/// volume" claim was dead on arrival.
#[test]
fn state_json_reports_the_same_plan_the_pixels_came_from() {
let live = view(true, FlowMode::Particles, 0.0).state_json();
let dead = view(false, FlowMode::Particles, 0.0).state_json();
assert_eq!(live["flowsim"]["n_edges"], 3);
assert_eq!(live["flowsim"]["n_flowing"], 3, "every stage reports throughput");
assert_eq!(dead["flowsim"]["n_edges"], 3, "the edges still exist…");
assert_eq!(dead["flowsim"]["n_flowing"], 0, "…they just carry nothing");
assert_eq!(dead["flowsim"]["total_particles"], 0);
assert!(live["flowsim"]["total_particles"].as_u64().unwrap() > 0);
// Density must SPREAD across the edges (rows 48k / 12k / 4k against a 120k max).
let per_edge: Vec<u64> = live["flowsim"]["per_edge_particles"]
.as_array()
.unwrap()
.iter()
.map(|v| v.as_u64().unwrap())
.collect();
eprintln!("[robot] per-edge particles = {per_edge:?} (rows 48k / 12k / 4k of a 120k max)");
assert_eq!(per_edge.len(), 3);
assert!(
per_edge[0] > per_edge[2],
"the fat edge must be denser than the thin one — got {per_edge:?}. Equal counts mean \
every edge clamped to max_density and density carries no information."
);
assert!(per_edge.iter().all(|&c| c > 0), "every flowing edge shows at least one particle: {per_edge:?}");
// Off → the overlay reports itself off, but the edge count is still honest.
let off = view(true, FlowMode::Off, 0.0).state_json();
assert_eq!(off["flowsim"]["mode"], "off");
}