crates/knut-pipelines/tests/facett_handoff_contract.rs
//! **The knut → facett handoff contract.**
//!
//! facett's `PipelineView` is a pure function of what knut's `ViewerState`
//! exposes: datasets → nodes, flows' `(read → target)` → edges, `node_status`
//! → per-node colour, the error event attributable to a failed node → the
//! surfaced error, and `ExecutionProgress` → the live progress readout.
//!
//! This test is the **knut side** of that contract — it has no facett dependency
//! (the model must stand alone), and asserts that across the same matrix the
//! facett connector crosses (pipeline shape × node status × progress), knut's
//! `ViewerState` already carries every input the visual consumes, with the
//! correct values. If this passes and facett's `knut_facett_matrix` passes, the
//! handoff is proven from both ends.
//!
//! LAW 1: every cell injects a REAL run (events + progress) and asserts real
//! output — node/edge structure, per-node status, the failed node's error text,
//! and the progress numbers — never a bare "didn't panic".
use knut_pipelines::{
DataflowGraph, Dataset, EventLevel, Flow, NodeStatus, OutputType, PipelineRunEvent,
ViewerState, ViewerUpdate,
};
// The progress dimension of the matrix is the SHARED fixture (`testkit`), not a
// local twin: korp's pipeline matrix crosses the same three states and used to
// hand-roll a byte-identical `apply_prog`. Owning it beside the types it builds
// means widening `StageProgress` fixes both matrices at once.
use knut_pipelines::testkit::{Prog, apply_prog};
#[derive(Clone, Copy, Debug)]
enum Shape {
Linear,
Diamond,
WideFanOut,
}
#[derive(Clone, Copy, Debug)]
enum Mix {
AllPending,
MidRunMixed,
AllOk,
OneFailed,
}
fn graph(shape: Shape) -> DataflowGraph {
let ds = |n: &str| Dataset::new(n, OutputType::MaterializedView);
match shape {
Shape::Linear => DataflowGraph::new()
.with_dataset(ds("a"))
.with_dataset(ds("b"))
.with_dataset(ds("c"))
.with_flow(Flow::new("f_a", "a", Vec::<String>::new()))
.with_flow(Flow::new("f_b", "b", ["a"]))
.with_flow(Flow::new("f_c", "c", ["b"])),
Shape::Diamond => DataflowGraph::new()
.with_dataset(ds("src"))
.with_dataset(ds("l"))
.with_dataset(ds("r"))
.with_dataset(ds("sink"))
.with_flow(Flow::new("f_src", "src", Vec::<String>::new()))
.with_flow(Flow::new("f_l", "l", ["src"]))
.with_flow(Flow::new("f_r", "r", ["src"]))
.with_flow(Flow::new("f_sink", "sink", ["l", "r"])),
Shape::WideFanOut => {
let mut g = DataflowGraph::new()
.with_dataset(ds("root"))
.with_flow(Flow::new("f_root", "root", Vec::<String>::new()));
for k in 0..4 {
let name = format!("leaf{k}");
g = g.with_dataset(ds(&name)).with_flow(Flow::new(
format!("f_leaf{k}"),
name,
["root"],
));
}
g
}
}
}
/// Distinct (read → target) edges the connector draws from the flows.
fn edge_count(g: &DataflowGraph) -> usize {
use std::collections::BTreeSet;
let names: BTreeSet<&str> = g.datasets.iter().map(|d| d.name.as_str()).collect();
let mut edges = BTreeSet::new();
for f in &g.flows {
if !names.contains(f.target.as_str()) {
continue;
}
for r in &f.reads {
if names.contains(r.as_str()) {
edges.insert((r.as_str(), f.target.as_str()));
}
}
}
edges.len()
}
fn apply_mix(vs: &mut ViewerState, names: &[String], mix: Mix) -> Option<(String, String)> {
match mix {
Mix::AllPending => None,
Mix::AllOk => {
for n in names {
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: format!("{n} materialized"),
element: Some(n.clone()),
}));
}
None
}
Mix::MidRunMixed => {
if let Some(n) = names.first() {
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: format!("{n} materialized"),
element: Some(n.clone()),
}));
}
if let Some(n) = names.get(1) {
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: format!("running into {n}"),
element: Some(n.clone()),
}));
}
None
}
Mix::OneFailed => {
let n = names.len();
for name in names.iter().take(n.saturating_sub(1)) {
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Info,
message: format!("{name} materialized"),
element: Some(name.clone()),
}));
}
let failed = names.last().cloned().unwrap();
let err = format!("AnalysisException: missing column in {failed}");
vs.apply(ViewerUpdate::Event(PipelineRunEvent {
timestamp: None,
level: EventLevel::Error,
message: err.clone(),
element: Some(failed.clone()),
}));
Some((failed, err))
}
}
}
/// The error event attributable to a failed dataset — the same lookup the facett
/// connector performs to surface a node's failure text.
fn error_for(vs: &ViewerState, dataset: &str) -> Option<String> {
vs.events
.iter()
.rev()
.find(|e| {
e.level == EventLevel::Error
&& (e.element.as_deref() == Some(dataset) || e.message.contains(dataset))
})
.map(|e| e.message.clone())
}
#[test]
fn knut_exposes_every_connector_input_across_the_matrix() {
let shapes = [Shape::Linear, Shape::Diamond, Shape::WideFanOut];
let mixes = [
Mix::AllPending,
Mix::MidRunMixed,
Mix::AllOk,
Mix::OneFailed,
];
let progs = Prog::ALL;
let mut cells = 0usize;
for &shape in &shapes {
for &mix in &mixes {
for &prog in &progs {
cells += 1;
let label = format!("{shape:?}/{mix:?}/{prog:?}");
let g = graph(shape);
let n_nodes = g.datasets.len();
let n_edges = edge_count(&g);
let names: Vec<String> = g.datasets.iter().map(|d| d.name.clone()).collect();
let mut vs = ViewerState::new(g);
let failed = apply_mix(&mut vs, &names, mix);
apply_prog(&mut vs, prog);
let s = vs.state_json();
// (1) structure the connector turns into nodes/edges. The
// distinct (read → target) pairs are exactly the edges the
// facett connector draws; per shape that's 2/4/4.
assert_eq!(s["dataset_count"], n_nodes, "[{label}] dataset count");
let expect_edges = match shape {
Shape::Linear => 2,
Shape::Diamond => 4,
Shape::WideFanOut => 4,
};
assert_eq!(n_edges, expect_edges, "[{label}] connector edge count");
assert!(
vs.topo_order().is_some(),
"[{label}] acyclic — the connector can lay it out in columns"
);
// (2) per-node status — exactly what colours each node.
match mix {
Mix::AllPending => {
assert_eq!(s["status_counts"]["idle"], n_nodes, "[{label}] all idle");
}
Mix::AllOk => {
assert_eq!(s["status_counts"]["done"], n_nodes, "[{label}] all done");
}
Mix::MidRunMixed => {
assert_eq!(s["status_counts"]["done"], 1, "[{label}] one done");
assert_eq!(s["status_counts"]["running"], 1, "[{label}] one running");
}
Mix::OneFailed => {
assert_eq!(s["status_counts"]["failed"], 1, "[{label}] one failed");
}
}
// (3) the failed node carries a recoverable error event.
if let Some((fname, ferr)) = &failed {
assert_eq!(
vs.status_of(fname),
NodeStatus::Failed,
"[{label}] node is Failed"
);
assert_eq!(
error_for(&vs, fname).as_deref(),
Some(ferr.as_str()),
"[{label}] error event attributable to the failed node, verbatim"
);
} else {
// no failure → no error-level event the connector would surface.
assert!(
vs.events.iter().all(|e| e.level != EventLevel::Error),
"[{label}] no error events when nothing failed"
);
}
// (4) progress numbers the connector reads.
match prog {
Prog::None => assert!(s["job_progress"].is_null(), "[{label}] no progress"),
Prog::Streaming => {
let p = &s["job_progress"];
assert_eq!(p["total_tasks"], 8, "[{label}] total tasks");
assert_eq!(p["completed_tasks"], 5, "[{label}] completed tasks");
assert_eq!(p["inflight_tasks"], 3, "[{label}] inflight");
assert_eq!(
p["bytes_read"],
(1024 * 1024 + 512 * 1024),
"[{label}] bytes"
);
let frac = p["fraction"].as_f64().unwrap();
assert!((frac - 0.625).abs() < 1e-4, "[{label}] fraction {frac}");
}
Prog::Complete => {
let p = &s["job_progress"];
assert_eq!(p["total_tasks"], 6, "[{label}] complete total");
assert_eq!(p["completed_tasks"], 6, "[{label}] complete done");
assert_eq!(p["all_done"], true, "[{label}] all stages done");
}
}
}
}
}
assert_eq!(cells, 3 * 4 * 3, "every contract cell ran");
eprintln!("knut→facett handoff contract: {cells} cells passed");
}