From 0221482f159982764ced8c71155f6b4d3043c6db Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Wed, 11 Nov 2026 17:16:24 +0530 Subject: [PATCH] test: write #46's remaining fuzz properties and complex fixtures - Lane isolation, write by write: across every checkpoint a lane slot is written once and never changes, no lane-body node gets a top-level `items`, and each (node, lane) activates exactly once. - Validator soundness: graphs the validator refuses are forced past it and shown to misbehave as each refusal says they would. - Timing schedules over generated shapes: two different per-node latency schedules with the same failures give the same output. - The refinement-loop-with-a-scatter and lane-switch fixtures. The two engine bugs these surfaced are fixed in their own PRs. Closes #46 --- .../loop_scatter_and_lane_switch_tests.rs | 252 ++++++++++ .../tinyflows/tests/complex_graphs_tests.rs | 1 + .../tests/fuzz_lane_isolation_tests.rs | 340 +++++++++++++ crates/tinyflows/tests/fuzz_timing_tests.rs | 146 +++++- .../tests/fuzz_validator_soundness_tests.rs | 463 ++++++++++++++++++ 5 files changed, 1201 insertions(+), 1 deletion(-) create mode 100644 crates/tinyflows/tests/complex_graphs/loop_scatter_and_lane_switch_tests.rs create mode 100644 crates/tinyflows/tests/fuzz_lane_isolation_tests.rs create mode 100644 crates/tinyflows/tests/fuzz_validator_soundness_tests.rs diff --git a/crates/tinyflows/tests/complex_graphs/loop_scatter_and_lane_switch_tests.rs b/crates/tinyflows/tests/complex_graphs/loop_scatter_and_lane_switch_tests.rs new file mode 100644 index 00000000..bdef5936 --- /dev/null +++ b/crates/tinyflows/tests/complex_graphs/loop_scatter_and_lane_switch_tests.rs @@ -0,0 +1,252 @@ +/// A refinement loop whose body scatters, with `candidates` proposed per pass. +/// +/// `candidates` is the expression `propose` sets per pass; each candidate is +/// evaluated in its own lane, gathered, and folded into the loop state. +fn refinement_loop(candidates: &str, loop_config: Value) -> WorkflowGraph { + WorkflowGraph { + name: "refinement_loop_scatter".to_string(), + nodes: vec![ + node( + "t", + NodeKind::Trigger, + json!({ "recursion_limit": 200, "max_node_visits": 100 }), + ), + node("refine", NodeKind::Loop, loop_config), + node( + "propose", + NodeKind::Transform, + json!({ "set": { "candidates": candidates } }), + ), + node("scatter", NodeKind::Scatter, json!({ "path": "candidates" })), + node( + "evaluate", + NodeKind::ToolCall, + json!({ "slug": "refine.evaluate", "args": "=item" }), + ), + node( + "gather", + NodeKind::Gather, + json!({ "from": ["evaluate"], "poll_interval_ms": 1 }), + ), + node("out", NodeKind::OutputParser, Value::Null), + ], + edges: vec![ + edge("t", "refine"), + port_edge("refine", "body", "propose"), + edge("propose", "scatter"), + edge("scatter", "evaluate"), + edge("evaluate", "gather"), + edge("gather", "refine"), + port_edge("refine", "done", "out"), + ], + ..Default::default() + } +} + +fn gathered_candidates(output: &Value) -> Vec { + output["nodes"]["gather"]["items"] + .as_array() + .expect("gather items") + .iter() + .filter_map(|item| item["json"]["json"]["c"].as_i64()) + .collect() +} + +/// Fan out per candidate, gather, accumulate the best score, stop `until` it +/// passes. Three candidates on every pass: pass k proposes 10k, 10k+1, 10k+2, +/// and `until best >= 30` stops after the third. +#[tokio::test] +async fn a_refinement_loop_scatters_its_candidates_on_every_pass() { + let graph = refinement_loop( + "=[{c: (.nodes.refine.iteration * 10)}, {c: (.nodes.refine.iteration * 10 + 1)}, \ + {c: (.nodes.refine.iteration * 10 + 2)}]", + json!({ + "max_iterations": 5, + "state": { + "init": { "best": 0, "passes": 0 }, + "update": { + "best": "=[.items[].json.c] | max", + "passes": "=state.passes + 1", + }, + }, + "until": "=state.best >= 30", + }), + ); + let compiled = compile(&graph).expect("a scatter inside a loop body is accepted"); + let mut caps = mock_capabilities(); + caps.tools = Arc::new(SelectiveFailure); + let trace = Arc::new(Trace::default()); + let observer: Arc = trace.clone(); + let outcome = tokio::time::timeout( + GUARD, + run_with_observer(&compiled, json!({}), &caps, &observer), + ) + .await + .expect("refinement loop hung") + .expect("refinement loop runs"); + let output = &outcome.output; + + assert_eq!(output["nodes"]["refine"]["exit_reason"], "until"); + assert_eq!(output["nodes"]["refine"]["iteration"], 3); + assert_eq!( + output["nodes"]["refine"]["state"], + json!({ "best": 32, "passes": 3 }) + ); + assert_eq!( + gathered_candidates(output), + vec![30, 31, 32], + "the last gather holds exactly the last pass's lanes, in lane order" + ); + assert!(output["nodes"]["evaluate"].get("items").is_none()); + assert!( + output["nodes"]["out"]["items"].is_array(), + "the converged loop left through done" + ); + + let mut order = trace.0.lock().expect("trace mutex poisoned").clone(); + order.retain(|id| id != "t"); + let pass = ["propose", "scatter", "evaluate", "evaluate", "evaluate", "gather", "refine"]; + let mut expected = vec!["refine"]; + for _ in 0..3 { + expected.extend(pass); + } + expected.push("out"); + assert_eq!(order, expected, "observed activation order"); +} + +/// scatter → switch → { fast: tool(on_error route) → ok | recover, slow: tool } +/// → gather. Row 1 fails on the routed arm (recovered as data); row 3 fails on +/// the unhandled arm (a failed lane, left to `on_lane_error`). +fn lane_switch_graph(policy: &str) -> WorkflowGraph { + WorkflowGraph { + name: format!("lane_switch_{policy}"), + nodes: vec![ + node("t", NodeKind::Trigger, json!({ "recursion_limit": 100 })), + node("scatter", NodeKind::Scatter, json!({ "path": "rows" })), + node("kind", NodeKind::Switch, json!({ "expression": "=item.kind" })), + node( + "fast", + NodeKind::ToolCall, + json!({ "slug": "lane.fast", "args": "=item", "on_error": "route" }), + ), + node("fast_ok", NodeKind::Transform, json!({ "set": { "arm": "fast" } })), + node( + "fast_recover", + NodeKind::Transform, + json!({ "set": { "arm": "recovered" } }), + ), + node( + "slow", + NodeKind::ToolCall, + json!({ "slug": "lane.slow", "args": "=item" }), + ), + node( + "gather", + NodeKind::Gather, + json!({ + "from": ["fast_ok", "fast_recover", "slow"], + "on_lane_error": policy, + "poll_interval_ms": 1, + }), + ), + ], + edges: vec![ + edge("t", "scatter"), + edge("scatter", "kind"), + port_edge("kind", "fast", "fast"), + port_edge("kind", "slow", "slow"), + port_edge("fast", "main", "fast_ok"), + port_edge("fast", "error", "fast_recover"), + edge("fast_ok", "gather"), + edge("fast_recover", "gather"), + edge("slow", "gather"), + ], + ..Default::default() + } +} + +fn positions(order: &[String], ids: &[&str]) -> Vec { + order + .iter() + .enumerate() + .filter(|(_, id)| ids.contains(&id.as_str())) + .map(|(index, _)| index) + .collect() +} + +/// Super-steps complete in order, so every switch finishes before any arm and +/// every arm tool before its routed follow-up. +fn assert_lane_order(order: &[String]) { + let count = |id: &str| order.iter().filter(|seen| *seen == id).count(); + assert_eq!(count("kind"), 4, "one switch per lane: {order:?}"); + assert_eq!(count("fast"), 2, "{order:?}"); + assert_eq!(count("slow"), 2, "{order:?}"); + let scatter = positions(order, &["scatter"]); + let switches = positions(order, &["kind"]); + let arms = positions(order, &["fast", "slow"]); + assert!(scatter.iter().max() < switches.iter().min(), "{order:?}"); + assert!(switches.iter().max() < arms.iter().min(), "{order:?}"); + assert!(order.iter().any(|id| id == "gather"), "{order:?}"); +} + +/// A lane-local `switch` with an `on_error: route` arm, under every +/// `on_lane_error` policy, asserting final state and observer order. +#[tokio::test] +async fn a_lane_switch_with_a_routed_error_arm_under_every_lane_policy() { + let input = json!({ "rows": [ + { "id": 0, "kind": "fast" }, + { "id": 1, "kind": "fast", "fail": true }, + { "id": 2, "kind": "slow" }, + { "id": 3, "kind": "slow", "fail": true } + ] }); + let mut caps = mock_capabilities(); + caps.tools = Arc::new(SelectiveFailure); + + for policy in ["collect", "skip", "fail_fast"] { + let compiled = compile(&lane_switch_graph(policy)).expect("lane switch compiles"); + let trace = Arc::new(Trace::default()); + let observer: Arc = trace.clone(); + let result = tokio::time::timeout( + GUARD, + run_with_observer(&compiled, input.clone(), &caps, &observer), + ) + .await + .expect("lane switch hung"); + let order = trace.0.lock().expect("trace mutex poisoned").clone(); + assert_lane_order(&order); + + if policy == "fail_fast" { + let error = result.expect_err("fail_fast fails the run").to_string(); + assert!(error.contains("fail_fast"), "{error}"); + assert!(error.contains("scheduled lane failure"), "{error}"); + continue; + } + + let outcome = result.expect("lane switch runs"); + let nodes = &outcome.output["nodes"]; + for id in ["kind", "fast", "fast_ok", "fast_recover", "slow"] { + assert!( + nodes[id].get("items").is_none(), + "{policy}: a lane wrote {id}'s top-level slot" + ); + } + assert_eq!(order.last().map(String::as_str), Some("gather"), "{order:?}"); + let fast = positions(&order, &["fast"]); + let follow_ups = positions(&order, &["fast_ok", "fast_recover"]); + assert_eq!(follow_ups.len(), 2, "one routed follow-up per fast lane: {order:?}"); + assert!(fast.iter().max() < follow_ups.iter().min(), "{order:?}"); + + let items = nodes["gather"]["items"].as_array().expect("gather items"); + assert_eq!(items[0]["json"]["arm"], "fast", "{policy}: lane 0"); + assert_eq!(items[1]["json"]["arm"], "recovered", "{policy}: lane 1"); + assert_eq!(items[2]["json"]["json"]["id"], 2, "{policy}: lane 2"); + match policy { + "collect" => { + assert_eq!(items.len(), 4, "{policy}: {items:?}"); + assert_eq!(items[3]["json"]["failed"], true); + assert_eq!(items[3]["json"]["lane"], 3); + } + _ => assert_eq!(items.len(), 3, "{policy} drops the failed lane: {items:?}"), + } + } +} diff --git a/crates/tinyflows/tests/complex_graphs_tests.rs b/crates/tinyflows/tests/complex_graphs_tests.rs index 8573c27e..4df5bd03 100644 --- a/crates/tinyflows/tests/complex_graphs_tests.rs +++ b/crates/tinyflows/tests/complex_graphs_tests.rs @@ -79,3 +79,4 @@ fn child_transform() -> WorkflowGraph { include!("complex_graphs/composition_tests.rs"); include!("complex_graphs/concurrency_and_failures_tests.rs"); +include!("complex_graphs/loop_scatter_and_lane_switch_tests.rs"); diff --git a/crates/tinyflows/tests/fuzz_lane_isolation_tests.rs b/crates/tinyflows/tests/fuzz_lane_isolation_tests.rs new file mode 100644 index 00000000..75d2b247 --- /dev/null +++ b/crates/tinyflows/tests/fuzz_lane_isolation_tests.rs @@ -0,0 +1,340 @@ +#![cfg(feature = "mock")] +//! Lane isolation observed write by write, not only in the final state. +//! +//! `fuzz_scatter_tests.rs` checks what a scatter run *ends* with. A lane that +//! wrote its slot twice, or wrote the node's top-level slot and was later +//! overwritten, ends the same way. There is no write hook, so this watches the +//! two public observation points instead: +//! +//! - every super-step boundary's state, from an [`InMemoryCheckpointer`]: a +//! lane slot goes absent → present once and never changes after, and a +//! lane-body node's top-level `items` never appears at any boundary; +//! - every activation's `StepPhase::After` frame: each (node, lane) pair +//! activates exactly once, and no body node activates outside a lane. This +//! catches two writes inside one step, which a boundary diff cannot see. +//! +//! Only the paths that are supposed to be lane-local are generated: success, +//! `on_error: continue` and `on_error: route`. Cancellation, retry backoff and +//! `Reenter` (a gate or spawn in a lane body) write the top-level slot today +//! and are deliberately out of scope here. + +use std::collections::BTreeMap; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use async_trait::async_trait; +use proptest::prelude::*; +use serde_json::{Value, json}; +use tinyflows::caps::ToolInvoker; +use tinyflows::caps::mock::mock_capabilities; +use tinyflows::compiler::{CompiledWorkflow, compile}; +use tinyflows::engine::{ + CancellationToken, Checkpointer, InMemoryCheckpointer, run_intercepted, run_with_checkpointer, +}; +use tinyflows::interception::{StepAction, StepFrame, StepInterceptor, StepPhase}; +use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph}; +use tinyflows::observability::{NoopObserver, RunObserver}; + +const GUARD: Duration = Duration::from_secs(5); +const THREAD: &str = "lane-isolation"; + +fn node(id: &str, kind: NodeKind, config: Value) -> Node { + Node { + id: id.to_string(), + kind, + type_version: 1, + name: id.to_string(), + config, + ports: vec![], + position: None, + } +} + +fn edge(from: &str, port: &str, to: &str) -> Edge { + Edge { + from_node: from.to_string(), + from_port: port.to_string(), + to_node: to.to_string(), + to_port: "main".to_string(), + } +} + +/// How the lane body ends. +#[derive(Debug, Clone, Copy)] +enum Tail { + /// Transforms only. + Plain, + /// A tool call with `on_error: continue`. + Continue, + /// A tool call with `on_error: route` into `ok` / `recover` arms. + Route, +} + +/// trigger → scatter → `work_0..n` transforms → [tail] → gather. +/// +/// Returns the graph and the ids of every node that runs inside a lane. +fn graph(body_len: usize, lanes: Option, tail: Tail) -> (WorkflowGraph, Vec) { + let mut scatter = json!({ "path": "rows" }); + if let Some(lanes) = lanes { + scatter["lanes"] = json!(lanes); + } + let mut nodes = vec![ + node( + "trigger", + NodeKind::Trigger, + json!({ "recursion_limit": 500, "max_node_visits": 300 }), + ), + node("scatter", NodeKind::Scatter, scatter), + ]; + let mut edges = vec![edge("trigger", "main", "scatter")]; + let mut members = Vec::new(); + let mut previous = "scatter".to_string(); + for index in 0..body_len { + let id = format!("work_{index}"); + nodes.push(node( + &id, + NodeKind::Transform, + json!({ "set": { format!("stage_{index}"): index } }), + )); + edges.push(edge(&previous, "main", &id)); + members.push(id.clone()); + previous = id; + } + let call = |policy: &str| { + node( + "call", + NodeKind::ToolCall, + json!({ "slug": "lane.maybe_fail", "args": "=item", "on_error": policy }), + ) + }; + let from: Vec = match tail { + Tail::Plain => vec![previous], + Tail::Continue => { + nodes.push(call("continue")); + edges.push(edge(&previous, "main", "call")); + members.push("call".to_string()); + vec!["call".to_string()] + } + Tail::Route => { + nodes.push(call("route")); + nodes.push(node( + "ok", + NodeKind::Transform, + json!({ "set": { "arm": "ok" } }), + )); + nodes.push(node( + "recover", + NodeKind::Transform, + json!({ "set": { "arm": "recover" } }), + )); + edges.push(edge(&previous, "main", "call")); + edges.push(edge("call", "main", "ok")); + edges.push(edge("call", "error", "recover")); + members.extend(["call", "ok", "recover"].map(String::from)); + vec!["ok".to_string(), "recover".to_string()] + } + }; + nodes.push(node( + "gather", + NodeKind::Gather, + json!({ "from": from.clone(), "poll_interval_ms": 1 }), + )); + for pred in &from { + edges.push(edge(pred, "main", "gather")); + } + let graph = WorkflowGraph { + name: "generated_lane_writes".to_string(), + nodes, + edges, + ..Default::default() + }; + (graph, members) +} + +fn expected_lane_count(items: usize, requested: Option) -> usize { + if items == 0 { + return 0; + } + let requested = requested.unwrap_or(items).clamp(1, 256); + if requested >= items { + items + } else { + items.div_ceil(items.div_ceil(requested)) + } +} + +/// Fails any call whose input row carries `fail: true`. +struct MaybeFail; + +#[async_trait] +impl ToolInvoker for MaybeFail { + async fn invoke( + &self, + _slug: &str, + args: Value, + _conn: Option<&str>, + ) -> tinyflows::error::Result { + if args.get("fail").and_then(Value::as_bool) == Some(true) { + Err(tinyflows::error::EngineError::Capability( + "generated lane failure".to_string(), + )) + } else { + Ok(args) + } + } +} + +/// Records `(node id, lane id)` for every `After` frame. +#[derive(Default)] +struct AfterFrames(Mutex)>>); + +#[async_trait] +impl StepInterceptor for AfterFrames { + async fn intercept(&self, frame: StepFrame<'_>) -> StepAction { + if frame.phase == StepPhase::After { + self.0.lock().expect("frames mutex poisoned").push(( + frame.node.id.clone(), + frame.lane.map(|lane| lane.id.clone()), + )); + } + StepAction::Continue { state_patch: None } + } +} + +fn runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build runtime") +} + +/// The run state at every super-step boundary, in order. +fn boundary_states(compiled: &CompiledWorkflow, input: Value) -> Vec { + let mut caps = mock_capabilities(); + caps.tools = Arc::new(MaybeFail); + runtime().block_on(async { + let checkpointer = Arc::new(InMemoryCheckpointer::::new()); + tokio::time::timeout( + GUARD, + run_with_checkpointer(compiled, input, &caps, checkpointer.clone(), THREAD), + ) + .await + .expect("scatter run hung") + .expect("scatter run"); + checkpointer + .get_thread(THREAD) + .await + .expect("read checkpoints") + .into_iter() + .map(|checkpoint| checkpoint.state) + .collect() + }) +} + +/// Every `After` frame of a second, identical run. +fn after_frames(compiled: &CompiledWorkflow, input: Value) -> Vec<(String, Option)> { + let mut caps = mock_capabilities(); + caps.tools = Arc::new(MaybeFail); + let frames = Arc::new(AfterFrames::default()); + runtime().block_on(async { + tokio::time::timeout( + GUARD, + run_intercepted( + compiled, + input, + &caps, + &(Arc::new(NoopObserver) as Arc), + CancellationToken::new(), + frames.clone(), + ), + ) + .await + .expect("intercepted scatter run hung") + .expect("intercepted scatter run"); + }); + frames.0.lock().expect("frames mutex poisoned").clone() +} + +fn lane_map<'a>(state: &'a Value, id: &str) -> Option<&'a serde_json::Map> { + state["nodes"][id].get("lanes").and_then(Value::as_object) +} + +proptest! { + #![proptest_config(ProptestConfig { cases: 64, ..ProptestConfig::default() })] + + #[test] + fn every_lane_slot_is_written_once_and_never_the_top_level( + rows in prop::collection::vec((any::(), any::()), 1..33), + body_len in 1usize..4, + requested in prop::option::of(1usize..17), + tail in prop_oneof![Just(Tail::Plain), Just(Tail::Continue), Just(Tail::Route)], + ) { + let (graph, members) = graph(body_len, requested, tail); + let compiled = compile(&graph).expect("generated graph compiles"); + let rows: Vec = rows + .iter() + .map(|(value, fail)| json!({ "value": value, "fail": fail })) + .collect(); + let expected = expected_lane_count(rows.len(), requested); + let input = json!({ "rows": rows }); + + // Boundary by boundary: a lane slot appears once and is then frozen; + // the top-level slot never appears at all. + let states = boundary_states(&compiled, input.clone()); + prop_assert!(!states.is_empty(), "the checkpointer recorded no boundaries"); + let mut first_seen: BTreeMap<(String, String), Value> = BTreeMap::new(); + for (boundary, state) in states.iter().enumerate() { + for id in &members { + prop_assert!( + state["nodes"][id].get("items").is_none(), + "boundary {boundary}: a lane activation wrote {id}'s top-level slot: {}", + state["nodes"][id] + ); + for (lane, slot) in lane_map(state, id).into_iter().flatten() { + let key = (id.clone(), lane.clone()); + match first_seen.get(&key) { + Some(first) => prop_assert_eq!( + first, slot, + "boundary {}: lane slot {}.{} was rewritten", boundary, id, lane + ), + None => { + first_seen.insert(key, slot.clone()); + } + } + } + } + } + + let last = states.last().expect("at least one boundary"); + let count = |id: &str| lane_map(last, id).map_or(0, serde_json::Map::len); + for id in members.iter().filter(|id| !matches!(id.as_str(), "ok" | "recover")) { + prop_assert_eq!(count(id), expected, "wrong lane count at {}", id); + } + if matches!(tail, Tail::Route) { + prop_assert_eq!(count("ok") + count("recover"), expected, "a lane was lost or doubled across the route arms"); + if let (Some(ok), Some(recover)) = (lane_map(last, "ok"), lane_map(last, "recover")) { + prop_assert!( + ok.keys().all(|lane| !recover.contains_key(lane)), + "one lane took both route arms" + ); + } + } + + // Activation by activation: exactly one `After` frame per (node, lane), + // and never an unlaned activation of a lane-body node. + let frames = after_frames(&compiled, input); + let mut per_lane: BTreeMap<(String, String), usize> = BTreeMap::new(); + for (id, lane) in frames.into_iter().filter(|(id, _)| members.contains(id)) { + prop_assert!(lane.is_some(), "{} activated outside any lane", id); + if let Some(lane) = lane { + *per_lane.entry((id, lane)).or_default() += 1; + } + } + for ((id, lane), activations) in &per_lane { + prop_assert_eq!(*activations, 1, "{}.{} activated more than once", id, lane); + } + let first_seen_keys: Vec<_> = first_seen.keys().cloned().collect(); + let frame_keys: Vec<_> = per_lane.keys().cloned().collect(); + prop_assert_eq!(first_seen_keys, frame_keys, "lane slots and lane activations disagree"); + } +} diff --git a/crates/tinyflows/tests/fuzz_timing_tests.rs b/crates/tinyflows/tests/fuzz_timing_tests.rs index c837a4e0..a7c53169 100644 --- a/crates/tinyflows/tests/fuzz_timing_tests.rs +++ b/crates/tinyflows/tests/fuzz_timing_tests.rs @@ -4,18 +4,30 @@ //! The same graph is run under opposite per-node latencies. Some branches fail //! through `on_error: continue`; those error items are part of the deterministic //! result too. Completion order must never leak into merged state. +//! +//! The second half runs generated graph shapes under two independent per-node +//! latency schedules with one shared failure set, injected through a +//! `StepInterceptor`: the stable output, or the error, must not depend on the +//! schedule. + +mod support; +use std::collections::HashSet; use std::sync::Arc; use std::time::Duration; +use async_trait::async_trait; use proptest::prelude::*; use serde_json::{Value, json}; +use support::graphgen::{arb_shape, graph_of}; use tinyflows::caps::ToolInvoker; use tinyflows::caps::mock::mock_capabilities; use tinyflows::compiler::compile; -use tinyflows::engine::run; +use tinyflows::engine::{CancellationToken, run, run_intercepted}; use tinyflows::error::EngineError; +use tinyflows::interception::{StepAction, StepFrame, StepInterceptor, StepPhase}; use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph}; +use tinyflows::observability::{NoopObserver, RunObserver}; const GUARD: Duration = Duration::from_secs(5); @@ -149,3 +161,135 @@ proptest! { prop_assert_eq!(forward, backward, "branch timing leaked into final state"); } } + +/// Per-node latency and failure injection, applied before each activation. +/// +/// Latency is a count of `yield_now` calls, never wall-clock time: engine waits +/// use a global timer thread, so a millisecond schedule would not be +/// reproducible. A node fails only when it is in `failing` *and* carries +/// `on_error: continue`, so a same-named node inside a child workflow (whose +/// config was not rewritten) is never failed by accident. +struct Schedule { + delays: Vec, + failing: HashSet, +} + +#[async_trait] +impl StepInterceptor for Schedule { + async fn intercept(&self, frame: StepFrame<'_>) -> StepAction { + if frame.phase != StepPhase::Before { + return StepAction::Continue { state_patch: None }; + } + let id = frame.node.id.as_str(); + let delay = id + .strip_prefix('n') + .and_then(|index| index.parse::().ok()) + .map_or(0, |index| self.delays[index % self.delays.len()]); + for _ in 0..delay { + tokio::task::yield_now().await; + } + let handled = frame.node.config.get("on_error").and_then(Value::as_str) == Some("continue"); + if handled && self.failing.contains(id) { + return StepAction::Fail { + message: format!("scheduled failure at {id}"), + }; + } + StepAction::Continue { state_patch: None } + } +} + +/// Removes scheduler observations, as `fuzz_graph.rs` does, plus spawn +/// tickets: the task runner numbers them in start order, so they are opaque +/// handles that legitimately differ between latency schedules. +fn stable_output(mut output: Value) -> Value { + fn strip(value: &mut Value) { + match value { + Value::Object(map) => { + map.remove("_activation_step"); + map.remove("started_at_step"); + map.remove("polls"); + map.remove("ticket"); + map.values_mut().for_each(strip); + } + Value::Array(items) => items.iter_mut().for_each(strip), + _ => {} + } + } + strip(&mut output); + output +} + +/// Marks the leaf nodes selected by `failures` as `on_error: continue`, and +/// returns their ids. Only `output_parser` and `transform` leaves are marked: +/// a failure there is turned into an error item, so the run still completes +/// and its whole output, error items included, is compared. +fn mark_failures(graph: &mut WorkflowGraph, failures: u64) -> HashSet { + let mut failing = HashSet::new(); + for (index, node) in graph.nodes.iter_mut().enumerate() { + let leaf = matches!(node.kind, NodeKind::OutputParser | NodeKind::Transform); + if !leaf || failures & (1 << (index % 64)) == 0 { + continue; + } + if !node.config.is_object() { + node.config = json!({}); + } + node.config["on_error"] = json!("continue"); + failing.insert(node.id.clone()); + } + failing +} + +fn run_scheduled( + compiled: &tinyflows::compiler::CompiledWorkflow, + delays: Vec, + failing: HashSet, +) -> Result { + runtime().block_on(async { + let observer = Arc::new(NoopObserver) as Arc; + let schedule = Arc::new(Schedule { delays, failing }); + match tokio::time::timeout( + Duration::from_secs(20), + run_intercepted( + compiled, + json!({}), + &mock_capabilities(), + &observer, + CancellationToken::new(), + schedule, + ), + ) + .await + { + Err(_) => Err("HUNG: scheduled run did not terminate".to_string()), + Ok(Ok((outcome, _resumable))) => Ok(stable_output(outcome.output)), + Ok(Err(error)) => Err(error.to_string()), + } + }) +} + +proptest! { + #![proptest_config(ProptestConfig { cases: 48, ..ProptestConfig::default() })] + + /// Two different latency schedules, one failure set: same stable output, or + /// the same error. + #[test] + fn generated_shapes_are_deterministic_across_latency_schedules( + shape in arb_shape(2), + first in prop::collection::vec(0u64..10, 16), + second in prop::collection::vec(0u64..10, 16), + failures in any::(), + ) { + let mut graph = graph_of(&shape); + let failing = mark_failures(&mut graph, failures); + let Ok(compiled) = compile(&graph) else { + return Ok(()); + }; + let a = run_scheduled(&compiled, first, failing.clone()); + let b = run_scheduled(&compiled, second, failing); + prop_assert!( + !matches!(&a, Err(e) if e.starts_with("HUNG")), + "first schedule hung" + ); + prop_assert_eq!(a, b, "latency schedule leaked into the result"); + } +} diff --git a/crates/tinyflows/tests/fuzz_validator_soundness_tests.rs b/crates/tinyflows/tests/fuzz_validator_soundness_tests.rs new file mode 100644 index 00000000..eedc1715 --- /dev/null +++ b/crates/tinyflows/tests/fuzz_validator_soundness_tests.rs @@ -0,0 +1,463 @@ +#![cfg(feature = "mock")] +//! Validator soundness: a refusal must name a graph that really misbehaves. +//! +//! The engine does not re-validate. `CompiledWorkflow` is a plain struct with +//! one public field, so a refused graph can be forced past `compile` and run. +//! Each case here first proves the refusal is real (`compile` fails with the +//! expected error), then runs the forced graph and asserts the specific wrong +//! behaviour the refusal exists to prevent. A refusal whose forced run behaves +//! correctly would be superstition; these tests would then fail and say so. +//! +//! Only the top-level graph can be forced: `sub_workflow` and `spawn` compile +//! their children at run time. + +mod support; + +use std::panic::AssertUnwindSafe; +use std::time::Duration; + +use proptest::prelude::*; +use serde_json::{Value, json}; + +use support::graphgen::{Shape, arb_shape, graph_of}; +use tinyflows::caps::mock::mock_capabilities; +use tinyflows::compiler::{CompiledWorkflow, compile}; +use tinyflows::engine::run; +use tinyflows::error::ValidationError; +use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph}; +use tinyflows::validate::validate_all; + +const GUARD: Duration = Duration::from_secs(5); + +fn node(id: &str, kind: NodeKind, config: Value) -> Node { + Node { + id: id.to_string(), + kind, + type_version: 1, + name: id.to_string(), + config, + ports: vec![], + position: None, + } +} + +fn edge(from: &str, to: &str) -> Edge { + port_edge(from, "main", to) +} + +fn port_edge(from: &str, port: &str, to: &str) -> Edge { + Edge { + from_node: from.to_string(), + from_port: port.to_string(), + to_node: to.to_string(), + to_port: "main".to_string(), + } +} + +fn graph(name: &str, nodes: Vec, edges: Vec) -> WorkflowGraph { + WorkflowGraph { + name: name.to_string(), + nodes, + edges, + ..Default::default() + } +} + +/// What a forced run did. +#[derive(Debug)] +enum Forced { + Ok(Value), + Err(String), + Hung, + Panicked, +} + +/// Runs `graph` with no validation at all. +fn force(graph: &WorkflowGraph, input: Value) -> Forced { + let compiled = CompiledWorkflow { + graph: graph.clone(), + }; + let outcome = std::panic::catch_unwind(AssertUnwindSafe(|| { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build runtime") + .block_on(async { + tokio::time::timeout(GUARD, run(&compiled, input, &mock_capabilities())).await + }) + })); + match outcome { + Err(_) => Forced::Panicked, + Ok(Err(_)) => Forced::Hung, + Ok(Ok(Err(error))) => Forced::Err(error.to_string()), + Ok(Ok(Ok(outcome))) => Forced::Ok(outcome.output), + } +} + +/// Asserts `graph` is refused, and refused for the reason under test. +fn assert_refused(graph: &WorkflowGraph, reason: impl Fn(&ValidationError) -> bool) { + assert!(compile(graph).is_err(), "{} was not refused", graph.name); + let errors = validate_all(graph); + assert!( + errors.iter().any(reason), + "{} was refused for the wrong reason: {errors:?}", + graph.name + ); +} + +fn refused_at(node_id: &'static str) -> impl Fn(&ValidationError) -> bool { + move |error| matches!(error, ValidationError::InvalidNodeConfig { node, .. } if node == node_id) +} + +/// An off-cycle arm into a merge on the cycle: from the second pass the barrier +/// can never complete, so the loop stops dead and never leaves through `done`. +#[test] +fn a_forced_off_cycle_merge_stops_the_loop_dead() { + let g = graph( + "off_cycle_arm", + vec![ + node("t", NodeKind::Trigger, json!({ "recursion_limit": 100 })), + node("outside", NodeKind::OutputParser, Value::Null), + node( + "l", + NodeKind::Loop, + json!({ "max_iterations": 3, "on_exceeded": "continue" }), + ), + node("work", NodeKind::OutputParser, Value::Null), + node("join", NodeKind::Merge, Value::Null), + node("out", NodeKind::OutputParser, Value::Null), + ], + vec![ + edge("t", "l"), + edge("t", "outside"), + port_edge("l", "body", "work"), + edge("work", "join"), + edge("outside", "join"), + edge("join", "l"), + port_edge("l", "done", "out"), + ], + ); + assert_refused( + &g, + |e| matches!(e, ValidationError::IllegalCycle(id) if id == "join"), + ); + if let Forced::Ok(output) = force(&g, json!({})) { + assert!( + output["nodes"]["out"].is_null(), + "the loop completed despite the off-cycle barrier: {output}" + ); + assert!( + output["nodes"]["l"]["iteration"].as_u64().unwrap_or(0) < 3, + "the loop ran every pass despite the off-cycle barrier: {output}" + ); + } +} + +/// Condition branches labelled on `to_port` share `from_port: main`, which the +/// engine lowers as a fan-out: both branches run, whatever the condition says. +#[test] +fn a_forced_condition_on_to_port_runs_both_branches() { + let labelled = |to: &str, label: &str| Edge { + from_node: "if".to_string(), + from_port: "main".to_string(), + to_node: to.to_string(), + to_port: label.to_string(), + }; + let g = graph( + "condition_on_to_port", + vec![ + node("t", NodeKind::Trigger, Value::Null), + node("if", NodeKind::Condition, json!({ "field": "flag" })), + node("yes", NodeKind::OutputParser, Value::Null), + node("no", NodeKind::OutputParser, Value::Null), + ], + vec![ + edge("t", "if"), + labelled("yes", "true"), + labelled("no", "false"), + ], + ); + assert_refused( + &g, + |e| matches!(e, ValidationError::InvalidConditionRouting { node, .. } if node == "if"), + ); + match force(&g, json!({ "flag": true })) { + Forced::Ok(output) => { + assert!( + !output["nodes"]["yes"].is_null(), + "yes did not run: {output}" + ); + assert!( + !output["nodes"]["no"].is_null(), + "the untaken branch did not run, so this refusal may be superstitious: {output}" + ); + } + other => panic!("expected a (wrongly) successful fan-out, got {other:?}"), + } +} + +/// A gather nobody scatters into. The validator says it would wait out its poll +/// budget; by reading `gather.rs`, `expected_lanes` falls back to the arrival +/// count (0), so it releases at once with nothing. Either way the work upstream +/// of it is lost. +#[test] +fn a_forced_gather_without_a_scatter_loses_its_input() { + let g = graph( + "orphan_gather", + vec![ + node("t", NodeKind::Trigger, json!({ "recursion_limit": 100 })), + node( + "work", + NodeKind::Transform, + json!({ "set": { "done": true } }), + ), + node( + "gather", + NodeKind::Gather, + json!({ "from": ["work"], "poll_interval_ms": 1, "max_polls": 5 }), + ), + ], + vec![edge("t", "work"), edge("work", "gather")], + ); + assert_refused(&g, refused_at("gather")); + if let Forced::Ok(output) = force(&g, json!({})) { + assert_eq!( + output["nodes"]["work"]["items"].as_array().map(Vec::len), + Some(1), + "work should have run: {output}" + ); + assert_eq!( + output["nodes"]["gather"]["items"] + .as_array() + .map(Vec::len) + .unwrap_or(0), + 0, + "a gather with no scatter collected its input, so this refusal may be superstitious" + ); + } +} + +/// A scatter with no gather: the lanes run, but a lane never writes the node's +/// top-level slot, so their output is invisible to the rest of the run. +#[test] +fn a_forced_scatter_without_a_gather_strands_its_lanes() { + let g = graph( + "ungathered_scatter", + vec![ + node("t", NodeKind::Trigger, json!({ "recursion_limit": 100 })), + node("scatter", NodeKind::Scatter, json!({ "path": "rows" })), + node( + "work", + NodeKind::Transform, + json!({ "set": { "done": true } }), + ), + ], + vec![edge("t", "scatter"), edge("scatter", "work")], + ); + assert_refused(&g, refused_at("scatter")); + match force(&g, json!({ "rows": [1, 2, 3] })) { + Forced::Ok(output) => { + assert_eq!( + output["nodes"]["work"]["lanes"] + .as_object() + .map(serde_json::Map::len), + Some(3), + "every lane should have run: {output}" + ); + assert!( + output["nodes"]["work"].get("items").is_none(), + "the lane output became visible, so this refusal may be superstitious: {output}" + ); + } + other => panic!("expected lanes that run and are stranded, got {other:?}"), + } +} + +/// A scatter inside a lane is refused statically and backstopped at run time: +/// the inner scatter fails rather than mis-collecting. +#[test] +fn a_forced_nested_scatter_fails_inside_its_lane() { + let g = graph( + "nested_scatter", + vec![ + node("t", NodeKind::Trigger, json!({ "recursion_limit": 100 })), + node("outer", NodeKind::Scatter, json!({ "path": "rows" })), + node("inner", NodeKind::Scatter, json!({ "path": "parts" })), + node( + "work", + NodeKind::Transform, + json!({ "set": { "done": true } }), + ), + // `timeout_partial` so the run settles however many lanes arrive, + // leaving the failed inner scatter's lane slot readable in state. + node( + "gather", + NodeKind::Gather, + json!({ + "from": ["work"], + "release": "timeout_partial", + "poll_interval_ms": 1, + "max_polls": 5, + }), + ), + ], + vec![ + edge("t", "outer"), + edge("outer", "inner"), + edge("inner", "work"), + edge("work", "gather"), + ], + ); + assert_refused(&g, refused_at("inner")); + let rows = json!({ "rows": [{ "parts": [1, 2] }, { "parts": [3] }] }); + let reported = match force(&g, rows) { + Forced::Ok(output) => output.to_string(), + Forced::Err(error) => error, + other => panic!("expected the nested-scatter backstop, got {other:?}"), + }; + assert!( + reported.contains("nested scatter is not supported"), + "the nested scatter ran: {reported}" + ); +} + +/// A loop head inside a lane: re-entry is counted in the node's top-level slot, +/// which a lane never writes, so the counter never advances and the loop never +/// finishes its two passes per lane. +#[test] +fn a_forced_loop_in_a_lane_never_counts_its_passes() { + let g = graph( + "loop_in_lane", + vec![ + node( + "t", + NodeKind::Trigger, + json!({ "recursion_limit": 80, "max_node_visits": 40 }), + ), + node("scatter", NodeKind::Scatter, json!({ "path": "rows" })), + node( + "l", + NodeKind::Loop, + json!({ "max_iterations": 2, "on_exceeded": "continue" }), + ), + node("w", NodeKind::OutputParser, Value::Null), + node( + "gather", + NodeKind::Gather, + json!({ "from": ["l"], "poll_interval_ms": 1, "max_polls": 5 }), + ), + ], + vec![ + edge("t", "scatter"), + edge("scatter", "l"), + port_edge("l", "body", "w"), + edge("w", "l"), + port_edge("l", "done", "gather"), + ], + ); + assert_refused(&g, refused_at("l")); + if let Forced::Ok(output) = force(&g, json!({ "rows": [1, 2] })) { + let gathered = output["nodes"]["gather"]["items"].as_array().map(Vec::len); + assert_ne!( + gathered, + Some(2), + "both lanes looped and exited correctly, so this refusal may be superstitious: {output}" + ); + } +} + +/// An edge to a node that does not exist: the runtime graph refuses to build. +#[test] +fn a_forced_edge_to_an_unknown_node_fails_the_run() { + let g = graph( + "ghost_edge", + vec![ + node("t", NodeKind::Trigger, Value::Null), + node("a", NodeKind::OutputParser, Value::Null), + ], + vec![edge("t", "a"), edge("a", "ghost")], + ); + assert_refused(&g, |e| matches!(e, ValidationError::UnknownNode(_))); + assert!( + !matches!(force(&g, json!({})), Forced::Ok(_)), + "a graph with a dangling edge ran to completion" + ); +} + +/// No trigger: there is no entry point to run from. +#[test] +fn a_forced_graph_without_a_trigger_fails_the_run() { + let g = graph( + "no_trigger", + vec![node("a", NodeKind::OutputParser, Value::Null)], + vec![], + ); + assert_refused(&g, |e| matches!(e, ValidationError::MissingTrigger)); + assert!( + matches!(force(&g, json!({})), Forced::Err(_)), + "a graph with no trigger did not fail cleanly" + ); +} + +/// Rewrites the root condition's branch labels from `from_port` onto +/// `to_port`. Returns the id of the `true` arm's entry node, which a correct run +/// of a graphgen branch never reaches (its field is always absent). +fn label_on_to_port(graph: &mut WorkflowGraph, condition: &str) -> String { + let mut yes = String::new(); + for edge in graph.edges.iter_mut().filter(|e| e.from_node == condition) { + if edge.from_port == "true" { + yes = edge.to_node.clone(); + } + edge.to_port = std::mem::replace(&mut edge.from_port, "main".to_string()); + } + yes +} + +proptest! { + #![proptest_config(ProptestConfig { cases: 48, ..ProptestConfig::default() })] + + /// For every runnable generated branch, mislabelling its condition is + /// refused, and the forced run takes the arm a correct run never takes. + #[test] + fn a_mislabelled_generated_condition_runs_its_untaken_arm( + yes in arb_shape(1), + no in arb_shape(1), + ) { + let valid = graph_of(&Shape::Branch(Box::new(yes), Box::new(no))); + if compile(&valid).is_err() { + return Ok(()); + } + // The root branch is the trigger's only successor. + let condition = valid + .edges + .iter() + .find(|e| e.from_node == "trigger") + .map(|e| e.to_node.clone()) + .expect("trigger edge"); + let mut forced = valid.clone(); + let yes_entry = label_on_to_port(&mut forced, &condition); + prop_assert!(!yes_entry.is_empty(), "the root branch has no true arm"); + + let errors = validate_all(&forced); + prop_assert!(compile(&forced).is_err(), "the mislabelled graph was accepted"); + prop_assert!( + errors.iter().all(|e| matches!(e, ValidationError::InvalidConditionRouting { .. })), + "the mutation introduced another refusal: {:?}", errors + ); + + if let Forced::Ok(output) = force(&valid, json!({})) { + prop_assert!(output["nodes"][&yes_entry].is_null(), "baseline took the true arm"); + } + let result = force(&forced, json!({})); + prop_assert!( + !matches!(result, Forced::Hung | Forced::Panicked), + "the forced run did not settle: {:?}", result + ); + if let Forced::Ok(output) = result { + prop_assert!( + !output["nodes"][&yes_entry].is_null(), + "the forced run did not take the untaken arm {}: {}", yes_entry, output + ); + } + } +}