diff --git a/proptest-regressions/engine_merge_tests.txt b/proptest-regressions/engine_merge_tests.txt new file mode 100644 index 0000000..b7f1ffd --- /dev/null +++ b/proptest-regressions/engine_merge_tests.txt @@ -0,0 +1,7 @@ +# Seeds for failure cases proptest has generated in the past. It is +# automatically read and these particular cases re-run before any +# novel cases are generated. +# +# It is recommended to check this file in to source control so that +# everyone who runs the test benefits from these saved cases. +cc f78576800bd471035e4f5ddf75018f7f19417f718915c85213fdefb25ca72003 # shrinks to base = Null, replacement = Null, path = ["a"] diff --git a/src/engine.rs b/src/engine.rs index bc5321f..6783148 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -278,13 +278,68 @@ fn merge(base: &mut Value, update: Value) { *base = value.clone(); return; } - match (base, update) { - (Value::Object(base), Value::Object(update)) => { + match update { + Value::Object(update) => { + // Recurse even when this subtree does not exist yet. Assigning the + // incoming object wholesale would leave any nested `$replace` + // sentinel as literal state on its first write. + if !base.is_object() { + *base = Value::Object(Map::new()); + } + let base = base.as_object_mut().expect("object created above"); for (key, value) in update { merge(base.entry(key).or_insert(Value::Null), value); } } - (base, update) => *base = update, + update => *base = update, + } +} + +/// Cancels background tasks that a cancelled run started. +/// +/// A `spawn` hands task ownership to the host runner. Merely stopping graph +/// scheduling does not stop that work, so without this pass a cancellation +/// between `spawn` and `gate` leaks every issued ticket. The runner contract +/// makes cancellation a no-op for already-settled work, so every issued ticket +/// can be handed back without racing an extra poll against task completion. +async fn cancel_spawned_tasks( + workflow: &CompiledWorkflow, + state: &Value, + capabilities: &Capabilities, +) { + let Some(runner) = capabilities.tasks.as_ref() else { + return; + }; + for spawn in workflow + .graph + .nodes + .iter() + .filter(|node| node.kind == NodeKind::Spawn) + { + let Some(slot) = state.get("nodes").and_then(|nodes| nodes.get(&spawn.id)) else { + continue; + }; + let mut item_lists: Vec<&Vec> = slot + .get("items") + .and_then(Value::as_array) + .into_iter() + .collect(); + item_lists.extend( + slot.get("lanes") + .and_then(Value::as_object) + .into_iter() + .flat_map(|lanes| lanes.values()) + .filter_map(|lane| lane.get("items").and_then(Value::as_array)), + ); + for ticket in item_lists.into_iter().flatten().filter_map(|item| { + item.get("json") + .and_then(|json| json.get("ticket")) + .and_then(Value::as_str) + }) { + if let Err(error) = runner.cancel(ticket).await { + tracing::warn!(%ticket, %error, "failed to cancel spawned task"); + } + } } } @@ -1205,11 +1260,10 @@ fn build_graph( // Successors on the emitted port, needed only inside a lane: `Plain` // routing normally rides static edges, but a lane has to re-schedule // every successor as a `Send`, so it needs the target list explicitly. - let plain_targets: Vec = graph - .edges + let plain_targets_by_port = outgoing_by_port(graph, &node.id); + let plain_targets: Vec = plain_targets_by_port .iter() - .filter(|e| e.from_node == node.id) - .map(|e| e.to_node.clone()) + .flat_map(|(_, targets)| targets.iter().cloned()) .collect(); // Which successors end a lane. Routing to one of these is a plain // activation, so the lanes converge on it instead of each running their @@ -1240,6 +1294,7 @@ fn build_graph( let token = token.clone(); let routing = routing.clone(); let plain_targets = plain_targets.clone(); + let plain_targets_by_port = plain_targets_by_port.clone(); let gather_nodes = gather_nodes.clone(); // The resume value delivered to this node on a checkpointed resume, if // any. A bare `true` means "approve the interrupted gate"; a structured @@ -1302,7 +1357,11 @@ fn build_graph( if let Some(lane) = lane.as_ref() { let emitted = port.unwrap_or("main"); let targets: Vec = match &routing { - HandlerRouting::Plain => plain_targets.clone(), + HandlerRouting::Plain => plain_targets_by_port + .iter() + .find(|(port, _)| port == emitted) + .map(|(_, targets)| targets.clone()) + .unwrap_or_default(), HandlerRouting::FanOut(targets) => targets.clone(), HandlerRouting::PortCommand(groups) => groups .iter() @@ -1831,18 +1890,46 @@ fn build_graph( }; match on_error { // Turn the failure into data on the default port. - "continue" => Ok(emit( - items_update(&node.id, &[error_item(&node.id, &err)], None)?, - None, - &[error_item(&node.id, &err)], - )), + "continue" => { + let item = error_item(&node.id, &err); + let update = match lane.as_ref() { + Some(lane) => lane_items_update( + &node.id, + lane, + std::slice::from_ref(&item), + None, + "ok", + None, + )?, + None => items_update( + &node.id, + std::slice::from_ref(&item), + None, + )?, + }; + Ok(emit(update, None, std::slice::from_ref(&item))) + } // Turn the failure into data on the `error` port so the // graph can route it to a recovery sub-graph. - "route" => Ok(emit( - items_update(&node.id, &[error_item(&node.id, &err)], Some("error"))?, - Some("error"), - &[error_item(&node.id, &err)], - )), + "route" => { + let item = error_item(&node.id, &err); + let update = match lane.as_ref() { + Some(lane) => lane_items_update( + &node.id, + lane, + std::slice::from_ref(&item), + Some("error"), + "ok", + None, + )?, + None => items_update( + &node.id, + std::slice::from_ref(&item), + Some("error"), + )?, + }; + Ok(emit(update, Some("error"), std::slice::from_ref(&item))) + } // "stop" (default) and any unknown policy fail the run. // // Stash the structured error before handing @@ -1857,6 +1944,23 @@ fn build_graph( // may produce several. _ => { let message = err.to_string(); + // A lane failure belongs to its gather, whose + // `on_lane_error` policy decides whether to + // collect, skip, or fail the overall run. Record + // it in the disjoint lane slot and carry the lane + // to the gather instead of aborting here. + if let Some(lane) = lane.as_ref() { + let meta = json!({ "error": message }); + let update = lane_items_update( + &node.id, + lane, + &[], + None, + "failed", + Some(&meta), + )?; + return Ok(emit(update, None, &[])); + } let mut slot = terminal_error.lock().expect("terminal error mutex poisoned"); if slot.is_none() { @@ -2270,6 +2374,10 @@ async fn build_and_run( .map(|interrupt| interrupt.id.clone()) .collect(); + if token.is_cancelled() { + cancel_spawned_tasks(workflow, &execution.state, capabilities).await; + } + tracing::info!( steps = execution.steps, visited = execution.visited.len(), @@ -2958,6 +3066,10 @@ mod merge_tests { } } +#[cfg(test)] +#[path = "engine_merge_tests.rs"] +mod merge_property_tests; + #[cfg(test)] mod lane_context_tests { use super::*; diff --git a/src/engine_merge_tests.rs b/src/engine_merge_tests.rs new file mode 100644 index 0000000..818400b --- /dev/null +++ b/src/engine_merge_tests.rs @@ -0,0 +1,100 @@ +//! Property tests for the workflow-state reducer. + +use proptest::prelude::*; +use serde_json::{Map, Value, json}; + +use super::{merge, replace}; + +fn leaf() -> impl Strategy { + prop_oneof![ + Just(Value::Null), + any::().prop_map(Value::Bool), + any::().prop_map(|value| json!(value)), + ".{0,16}".prop_map(Value::String), + prop::collection::vec(any::(), 0..8).prop_map(|values| json!(values)), + ] +} + +fn arbitrary_json() -> impl Strategy { + leaf().prop_recursive(4, 96, 8, |inner| { + prop_oneof![ + prop::collection::vec(inner.clone(), 0..6).prop_map(Value::Array), + prop::collection::btree_map("[a-z]{1,5}", inner, 0..6) + .prop_map(|entries| Value::Object(entries.into_iter().collect())), + ] + }) +} + +/// Real engine updates have a stable object shape down to the leaf being +/// assigned. Generating that shape avoids meaningless type-conflict examples +/// (`scalar` followed by `object`) that the engine never emits. +fn compatible_update() -> impl Strategy { + (0usize..5, 0usize..5, leaf()).prop_map(|(node, field, value)| { + json!({ "nodes": { format!("n{node}"): { format!("k{field}"): value } } }) + }) +} + +fn folded(initial: Value, updates: impl IntoIterator) -> Value { + let mut state = initial; + for update in updates { + merge(&mut state, update); + } + state +} + +fn nested_update(path: &[String], value: Value) -> Value { + path.iter() + .rev() + .fold(replace(value), |child, key| json!({ key: child })) +} + +fn assign_at_path(root: &mut Value, path: &[String], value: Value) { + if path.is_empty() { + *root = value; + return; + } + if !root.is_object() { + *root = Value::Object(Map::new()); + } + let child = root + .as_object_mut() + .expect("object created above") + .entry(path[0].clone()) + .or_insert(Value::Null); + assign_at_path(child, &path[1..], value); +} + +proptest! { + #![proptest_config(ProptestConfig { cases: 256, ..ProptestConfig::default() })] + + #[test] + fn compatible_update_sequences_are_associative( + run_state in arbitrary_json(), + updates in prop::collection::vec(compatible_update(), 0..40), + split in 0usize..40, + ) { + let initial = json!({ "run": run_state, "nodes": {} }); + let split = split.min(updates.len()); + let sequential = folded(initial.clone(), updates.clone()); + + let left = folded(Value::Object(Map::new()), updates[..split].iter().cloned()); + let right = folded(Value::Object(Map::new()), updates[split..].iter().cloned()); + let grouped = folded(initial, [left, right]); + + prop_assert_eq!(sequential, grouped); + } + + #[test] + fn replace_round_trips_at_arbitrary_nesting( + base in arbitrary_json(), + replacement in arbitrary_json(), + path in prop::collection::vec("[a-z]{1,5}", 0..7), + ) { + let mut actual = base.clone(); + merge(&mut actual, nested_update(&path, replacement.clone())); + + let mut expected = base; + assign_at_path(&mut expected, &path, replacement); + prop_assert_eq!(actual, expected); + } +} diff --git a/tests/complex_graphs_tests.rs b/tests/complex_graphs_tests.rs new file mode 100644 index 0000000..205974f --- /dev/null +++ b/tests/complex_graphs_tests.rs @@ -0,0 +1,552 @@ +#![cfg(feature = "mock")] +//! Named, hand-built compositions whose ordering cannot be covered by shallow +//! generated graphs alone. + +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use serde_json::{Value, json}; +use tinyflows::caps::ToolInvoker; +use tinyflows::caps::mock::mock_capabilities; +use tinyflows::compiler::compile; +use tinyflows::engine::{run_resumable, run_with_observer}; +use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph}; +use tinyflows::observability::RunObserver; + +const GUARD: Duration = Duration::from_secs(10); + +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(), + } +} + +#[derive(Default)] +struct Trace(Mutex>); + +impl RunObserver for Trace { + fn on_step_finish(&self, step: &tinyflows::observability::ExecutionStep) { + self.0 + .lock() + .expect("trace mutex poisoned") + .push(step.node_id.clone()); + } +} + +fn child_transform() -> WorkflowGraph { + WorkflowGraph { + name: "lane_child".to_string(), + nodes: vec![ + node("child_trigger", NodeKind::Trigger, Value::Null), + node( + "child_agent", + NodeKind::Agent, + json!({ "prompt": "refine this candidate" }), + ), + node( + "child_tag", + NodeKind::Transform, + json!({ "set": { "child_complete": true } }), + ), + ], + edges: vec![ + edge("child_trigger", "child_agent"), + edge("child_agent", "child_tag"), + ], + ..Default::default() + } +} + +/// scatter → per-lane child workflow → gather → accumulator loop → scatter. +/// +/// This crosses both state namespaces (lane slots and child `nodes`) before a +/// repeated reducer write and then fans the accumulated result out again. +#[tokio::test] +async fn scatter_child_gather_loop_and_second_scatter_compose() { + let child = serde_json::to_value(child_transform()).expect("serialize child"); + let graph = WorkflowGraph { + name: "two_stage_refinement".to_string(), + nodes: vec![ + node( + "trigger", + NodeKind::Trigger, + json!({ "recursion_limit": 500, "max_node_visits": 200, "max_concurrency": 3 }), + ), + node( + "first_scatter", + NodeKind::Scatter, + json!({ "path": "rows" }), + ), + node("child", NodeKind::SubWorkflow, json!({ "workflow": child })), + node( + "first_gather", + NodeKind::Gather, + json!({ "from": ["child"], "release": "quorum", "n": 3, "poll_interval_ms": 1 }), + ), + node( + "refine_loop", + NodeKind::Loop, + json!({ + "max_iterations": 2, + "on_exceeded": "continue", + "emit": "both", + "state": { + "init": { "passes": 0 }, + "update": { "passes": "=state.passes + 1" }, + }, + }), + ), + node( + "loop_body", + NodeKind::Transform, + json!({ "set": { "refined": true } }), + ), + node("second_scatter", NodeKind::Scatter, Value::Null), + node( + "finalize", + NodeKind::Transform, + json!({ "set": { "final": true } }), + ), + node( + "second_gather", + NodeKind::Gather, + json!({ "from": ["finalize"], "poll_interval_ms": 1 }), + ), + ], + edges: vec![ + edge("trigger", "first_scatter"), + edge("first_scatter", "child"), + edge("child", "first_gather"), + edge("first_gather", "refine_loop"), + port_edge("refine_loop", "body", "loop_body"), + edge("loop_body", "refine_loop"), + port_edge("refine_loop", "done", "second_scatter"), + edge("second_scatter", "finalize"), + edge("finalize", "second_gather"), + ], + ..Default::default() + }; + + let compiled = compile(&graph).expect("complex graph compiles"); + let trace = Arc::new(Trace::default()); + let observer: Arc = trace.clone(); + let outcome = tokio::time::timeout( + GUARD, + run_with_observer( + &compiled, + json!({ "rows": [{"id": 0}, {"id": 1}, {"id": 2}] }), + &mock_capabilities(), + &observer, + ), + ) + .await + .expect("complex graph hung") + .expect("complex graph runs"); + + assert_eq!( + outcome.output["nodes"]["child"]["lanes"] + .as_object() + .map(serde_json::Map::len), + Some(3), + "one child workflow ran in each first-stage lane" + ); + assert_eq!(outcome.output["nodes"]["refine_loop"]["iteration"], 2); + assert_eq!( + outcome.output["nodes"]["refine_loop"]["state"], + json!({ "passes": 2 }), + "the accumulator was replaced cleanly on both passes" + ); + let final_items = outcome.output["nodes"]["second_gather"]["items"] + .as_array() + .expect("second gather output"); + assert_eq!(final_items.len(), 4, "three child results plus accumulator"); + assert!( + final_items.iter().all(|item| item["json"]["final"] == true), + "every second-stage lane ran the finalizer" + ); + + let order = trace.0.lock().expect("trace mutex poisoned").clone(); + let first_gather = order.iter().position(|id| id == "first_gather").unwrap(); + let second_scatter = order.iter().position(|id| id == "second_scatter").unwrap(); + assert!(first_gather < second_scatter, "observed order: {order:?}"); + assert_eq!( + order.iter().filter(|id| id.as_str() == "loop_body").count(), + 2, + "the loop body must activate once per pass: {order:?}" + ); +} + +/// A child starts and collects asynchronous work, then pauses for approval; +/// the resumed parent starts and collects a second asynchronous task. +#[tokio::test] +async fn nested_async_gate_and_approval_resume_across_a_subworkflow_boundary() { + let child = WorkflowGraph { + name: "async_child".to_string(), + nodes: vec![ + node("ct", NodeKind::Trigger, Value::Null), + node( + "cspawn", + NodeKind::Spawn, + json!({ "target": "tool", "slug": "child.lookup" }), + ), + node( + "cgate", + NodeKind::Gate, + json!({ "from": ["cspawn"], "poll_interval_ms": 1 }), + ), + node( + "approve", + NodeKind::OutputParser, + json!({ "requires_approval": true }), + ), + node("cdone", NodeKind::OutputParser, Value::Null), + ], + edges: vec![ + edge("ct", "cspawn"), + edge("cspawn", "cgate"), + edge("cgate", "approve"), + edge("approve", "cdone"), + ], + ..Default::default() + }; + let graph = WorkflowGraph { + name: "nested_async_approval".to_string(), + nodes: vec![ + node("t", NodeKind::Trigger, json!({ "recursion_limit": 100 })), + node( + "sub", + NodeKind::SubWorkflow, + json!({ "workflow": serde_json::to_value(child).unwrap() }), + ), + node( + "pspawn", + NodeKind::Spawn, + json!({ "target": "tool", "slug": "parent.publish" }), + ), + node( + "pgate", + NodeKind::Gate, + json!({ "from": ["pspawn"], "poll_interval_ms": 1 }), + ), + ], + edges: vec![ + edge("t", "sub"), + edge("sub", "pspawn"), + edge("pspawn", "pgate"), + ], + ..Default::default() + }; + + let compiled = compile(&graph).expect("compile"); + let caps = mock_capabilities(); + let resumable = tokio::time::timeout(GUARD, run_resumable(&compiled, json!({}), &caps)) + .await + .expect("initial run hung") + .expect("initial run"); + assert_eq!( + resumable.outcome().pending_approvals, + vec!["sub::approve".to_string()] + ); + assert!(resumable.outcome().output["nodes"]["pspawn"].is_null()); + + let done = tokio::time::timeout(GUARD, resumable.resume(vec!["sub::approve".to_string()])) + .await + .expect("resume hung") + .expect("resume"); + assert!(done.pending_approvals.is_empty()); + assert_eq!(done.output["nodes"]["pgate"]["arrived"], 1); + assert_eq!( + done.output["nodes"]["pgate"]["items"][0]["json"]["slug"], + "parent.publish" + ); +} + +struct ConcurrencyProbe { + in_flight: AtomicUsize, + peak: AtomicUsize, +} + +#[async_trait::async_trait] +impl ToolInvoker for ConcurrencyProbe { + async fn invoke( + &self, + _slug: &str, + args: Value, + _conn: Option<&str>, + ) -> tinyflows::error::Result { + let now = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1; + self.peak.fetch_max(now, Ordering::SeqCst); + for _ in 0..6 { + tokio::task::yield_now().await; + } + self.in_flight.fetch_sub(1, Ordering::SeqCst); + Ok(args) + } +} + +/// The maximum graph concurrency also bounds a 256-lane scatter, rather than +/// applying only to ordinary static fan-out branches. +#[tokio::test] +async fn a_wide_scatter_honours_the_global_admission_bound() { + let graph = WorkflowGraph { + name: "wide_bounded_scatter".to_string(), + nodes: vec![ + node( + "t", + NodeKind::Trigger, + json!({ "max_concurrency": 4, "recursion_limit": 500 }), + ), + node("scatter", NodeKind::Scatter, json!({ "path": "rows" })), + node( + "work", + NodeKind::ToolCall, + json!({ "slug": "lane.work", "args": "=item" }), + ), + node( + "gather", + NodeKind::Gather, + json!({ "from": ["work"], "poll_interval_ms": 1 }), + ), + ], + edges: vec![ + edge("t", "scatter"), + edge("scatter", "work"), + edge("work", "gather"), + ], + ..Default::default() + }; + let probe = Arc::new(ConcurrencyProbe { + in_flight: AtomicUsize::new(0), + peak: AtomicUsize::new(0), + }); + let mut caps = mock_capabilities(); + caps.tools = probe.clone(); + let compiled = compile(&graph).expect("compile"); + let rows: Vec = (0..256).map(|index| json!({ "index": index })).collect(); + + let outcome = tokio::time::timeout( + GUARD, + tinyflows::engine::run(&compiled, json!({ "rows": rows }), &caps), + ) + .await + .expect("wide scatter hung") + .expect("wide scatter runs"); + let peak = probe.peak.load(Ordering::SeqCst); + assert!(peak > 1, "lanes should overlap, observed peak {peak}"); + assert!(peak <= 4, "max_concurrency=4 admitted {peak} lanes at once"); + assert_eq!( + outcome.output["nodes"]["gather"]["items"] + .as_array() + .map(Vec::len), + Some(256) + ); +} + +struct SelectiveFailure; + +#[async_trait::async_trait] +impl ToolInvoker for SelectiveFailure { + 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( + "scheduled lane failure".to_string(), + )) + } else { + Ok(args) + } + } +} + +fn failing_lane_graph(policy: &str) -> WorkflowGraph { + WorkflowGraph { + name: format!("lane_errors_{policy}"), + nodes: vec![ + node("t", NodeKind::Trigger, json!({ "recursion_limit": 100 })), + node("scatter", NodeKind::Scatter, json!({ "path": "rows" })), + node( + "work", + NodeKind::ToolCall, + json!({ "slug": "lane.maybe_fail", "args": "=item" }), + ), + node( + "gather", + NodeKind::Gather, + json!({ + "from": ["work"], + "on_lane_error": policy, + "poll_interval_ms": 1, + }), + ), + ], + edges: vec![ + edge("t", "scatter"), + edge("scatter", "work"), + edge("work", "gather"), + ], + ..Default::default() + } +} + +/// One lane fails while its siblings succeed, under all three gather policies. +#[tokio::test] +async fn lane_failures_collect_skip_or_fail_fast_as_configured() { + let input = json!({ "rows": [ + { "id": 0 }, + { "id": 1, "fail": true }, + { "id": 2 } + ] }); + let mut caps = mock_capabilities(); + caps.tools = Arc::new(SelectiveFailure); + + let collect = compile(&failing_lane_graph("collect")).expect("compile collect"); + let collected = tinyflows::engine::run(&collect, input.clone(), &caps) + .await + .expect("collect run"); + let items = collected.output["nodes"]["gather"]["items"] + .as_array() + .expect("collected items"); + assert_eq!(items.len(), 3); + assert_eq!( + items + .iter() + .filter(|item| item["json"]["failed"] == true) + .count(), + 1 + ); + + let skip = compile(&failing_lane_graph("skip")).expect("compile skip"); + let skipped = tinyflows::engine::run(&skip, input.clone(), &caps) + .await + .expect("skip run"); + assert_eq!( + skipped.output["nodes"]["gather"]["items"] + .as_array() + .map(Vec::len), + Some(2) + ); + + let fail_fast = compile(&failing_lane_graph("fail_fast")).expect("compile fail_fast"); + let error = tinyflows::engine::run(&fail_fast, input, &caps) + .await + .expect_err("fail_fast must fail the run"); + assert!(error.to_string().contains("scheduled lane failure")); +} + +/// Handled lane errors stay in lane-local state for both recovery policies. +/// The routed form also proves that different lanes can take different ports +/// and still reconverge at one gather. +#[tokio::test] +async fn lane_error_continue_and_route_never_write_the_top_level_slot() { + let input = json!({ "rows": [ + { "id": 0 }, + { "id": 1, "fail": true }, + { "id": 2 } + ] }); + let mut caps = mock_capabilities(); + caps.tools = Arc::new(SelectiveFailure); + + let mut continued = failing_lane_graph("collect"); + continued.name = "lane_continue".to_string(); + continued + .nodes + .iter_mut() + .find(|node| node.id == "work") + .expect("work node") + .config["on_error"] = json!("continue"); + let outcome = tinyflows::engine::run( + &compile(&continued).expect("compile continue"), + input.clone(), + &caps, + ) + .await + .expect("continue run"); + assert!(outcome.output["nodes"]["work"].get("items").is_none()); + assert_eq!( + outcome.output["nodes"]["gather"]["items"] + .as_array() + .map(Vec::len), + Some(3) + ); + + let routed = WorkflowGraph { + name: "lane_error_route".to_string(), + nodes: vec![ + node("t", NodeKind::Trigger, json!({ "recursion_limit": 100 })), + node("scatter", NodeKind::Scatter, json!({ "path": "rows" })), + node( + "work", + NodeKind::ToolCall, + json!({ + "slug": "lane.maybe_fail", + "args": "=item", + "on_error": "route", + }), + ), + node( + "success", + NodeKind::Transform, + json!({ "set": { "route": "main" } }), + ), + node( + "recover", + NodeKind::Transform, + json!({ "set": { "route": "error" } }), + ), + node( + "gather", + NodeKind::Gather, + json!({ "from": ["success", "recover"], "poll_interval_ms": 1 }), + ), + ], + edges: vec![ + edge("t", "scatter"), + edge("scatter", "work"), + port_edge("work", "main", "success"), + port_edge("work", "error", "recover"), + edge("success", "gather"), + edge("recover", "gather"), + ], + ..Default::default() + }; + let outcome = tinyflows::engine::run(&compile(&routed).expect("compile route"), input, &caps) + .await + .expect("route run"); + assert!(outcome.output["nodes"]["work"].get("items").is_none()); + let items = outcome.output["nodes"]["gather"]["items"] + .as_array() + .expect("gather items"); + assert_eq!(items.len(), 3); + assert_eq!( + items + .iter() + .filter(|item| item["json"]["route"] == "error") + .count(), + 1, + "only the failing lane takes the error arm" + ); +} diff --git a/tests/fuzz_cancellation_tests.proptest-regressions b/tests/fuzz_cancellation_tests.proptest-regressions new file mode 100644 index 0000000..460c764 --- /dev/null +++ b/tests/fuzz_cancellation_tests.proptest-regressions @@ -0,0 +1,7 @@ +# Seeds for failure cases proptest has generated in the past. It is +# automatically read and these particular cases re-run before any +# novel cases are generated. +# +# It is recommended to check this file in to source control so that +# everyone who runs the test benefits from these saved cases. +cc 46d852fccafd1976a8b5a7c9e6001586d40df63b88767acb178e33c01ce5a762 # shrinks to tasks = 1, fraction = 0 diff --git a/tests/fuzz_cancellation_tests.rs b/tests/fuzz_cancellation_tests.rs new file mode 100644 index 0000000..8bfb118 --- /dev/null +++ b/tests/fuzz_cancellation_tests.rs @@ -0,0 +1,253 @@ +#![cfg(feature = "mock")] +//! Generated cancellation schedules for `spawn` / `gate` workflows. +//! +//! Every generated task remains running forever. The runner flips the workflow +//! cancellation token after an arbitrary start, which makes every issued +//! ticket uncollected. A correct run must settle promptly and ask the runner to +//! cancel each of those tickets. + +use std::collections::BTreeSet; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use proptest::prelude::*; +use serde_json::{Value, json}; +use tinyflows::caps::mock::mock_capabilities; +use tinyflows::caps::{TaskRunner, TaskSpec, TaskState}; +use tinyflows::compiler::compile; +use tinyflows::engine::{CancellationToken, run_cancellable}; +use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph}; + +const GUARD: Duration = Duration::from_secs(2); + +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 { + Edge { + from_node: from.to_string(), + from_port: "main".to_string(), + to_node: to.to_string(), + to_port: "main".to_string(), + } +} + +fn spawned_graph(tasks: usize) -> WorkflowGraph { + let mut nodes = vec![ + node( + "trigger", + NodeKind::Trigger, + json!({ "recursion_limit": 100 }), + ), + node("fanout", NodeKind::OutputParser, Value::Null), + ]; + let mut edges = vec![edge("trigger", "fanout")]; + let mut sources = Vec::new(); + for index in 0..tasks { + let id = format!("spawn_{index}"); + nodes.push(node( + &id, + NodeKind::Spawn, + json!({ "target": "tool", "slug": format!("task.{index}") }), + )); + edges.push(edge("fanout", &id)); + sources.push(id); + } + nodes.push(node( + "gate", + NodeKind::Gate, + json!({ + "from": sources, + "release": "all", + "poll_interval_ms": 1, + "max_polls": 1_000, + }), + )); + for index in 0..tasks { + edges.push(edge(&format!("spawn_{index}"), "gate")); + } + WorkflowGraph { + name: "generated_cancellation".to_string(), + nodes, + edges, + ..Default::default() + } +} + +fn scattered_spawn_graph() -> WorkflowGraph { + WorkflowGraph { + name: "generated_lane_cancellation".to_string(), + nodes: vec![ + node( + "trigger", + NodeKind::Trigger, + json!({ "recursion_limit": 100 }), + ), + node("scatter", NodeKind::Scatter, json!({ "path": "rows" })), + node( + "spawn", + NodeKind::Spawn, + json!({ "target": "tool", "slug": "lane.task" }), + ), + node( + "gather", + NodeKind::Gather, + json!({ "from": ["spawn"], "poll_interval_ms": 1 }), + ), + ], + edges: vec![ + edge("trigger", "scatter"), + edge("scatter", "spawn"), + edge("spawn", "gather"), + ], + ..Default::default() + } +} + +struct CancellingRunner { + token: CancellationToken, + cancel_after: usize, + started: Mutex>, + cancelled: Mutex>, +} + +impl CancellingRunner { + fn new(token: CancellationToken, cancel_after: usize) -> Arc { + Arc::new(Self { + token, + cancel_after, + started: Mutex::new(Vec::new()), + cancelled: Mutex::new(Vec::new()), + }) + } +} + +#[async_trait::async_trait] +impl TaskRunner for CancellingRunner { + async fn start(&self, _spec: TaskSpec) -> tinyflows::error::Result { + let mut started = self.started.lock().expect("started mutex poisoned"); + let ticket = format!("ticket-{}", started.len()); + started.push(ticket.clone()); + if started.len() >= self.cancel_after { + self.token.cancel(); + } + Ok(ticket) + } + + async fn poll(&self, _ticket: &str) -> tinyflows::error::Result { + Ok(TaskState::Running) + } + + async fn cancel(&self, ticket: &str) -> tinyflows::error::Result<()> { + self.cancelled + .lock() + .expect("cancelled mutex poisoned") + .push(ticket.to_string()); + Ok(()) + } +} + +fn runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build runtime") +} + +proptest! { + #![proptest_config(ProptestConfig { cases: 48, ..ProptestConfig::default() })] + + #[test] + fn cancellation_settles_and_cleans_up_every_uncollected_ticket( + tasks in 1usize..9, + fraction in 0usize..8, + ) { + let cancel_after = 1 + fraction % tasks; + let compiled = compile(&spawned_graph(tasks)).expect("generated graph compiles"); + let token = CancellationToken::new(); + let runner = CancellingRunner::new(token.clone(), cancel_after); + let mut caps = mock_capabilities(); + caps.tasks = Some(runner.clone()); + + let outcome = runtime().block_on(async { + tokio::time::timeout( + GUARD, + run_cancellable(&compiled, json!({}), &caps, token), + ) + .await + .expect("a cancelled run must settle promptly") + .expect("cooperative cancellation returns a partial outcome") + }); + + prop_assert!(outcome.cancelled); + let started: BTreeSet = runner + .started + .lock() + .expect("started mutex poisoned") + .iter() + .cloned() + .collect(); + let cancelled: BTreeSet = runner + .cancelled + .lock() + .expect("cancelled mutex poisoned") + .iter() + .cloned() + .collect(); + prop_assert_eq!( + cancelled, + started, + "every issued, perpetually-running ticket must be cancelled" + ); + } + + #[test] + fn cancellation_also_cleans_up_tickets_stored_in_lane_slots( + rows in 1usize..9, + fraction in 0usize..8, + ) { + let cancel_after = 1 + fraction % rows; + let compiled = compile(&scattered_spawn_graph()).expect("generated graph compiles"); + let token = CancellationToken::new(); + let runner = CancellingRunner::new(token.clone(), cancel_after); + let mut caps = mock_capabilities(); + caps.tasks = Some(runner.clone()); + let input_rows: Vec = (0..rows).map(|index| json!({ "index": index })).collect(); + + let outcome = runtime().block_on(async { + tokio::time::timeout( + GUARD, + run_cancellable(&compiled, json!({ "rows": input_rows }), &caps, token), + ) + .await + .expect("a cancelled lane run must settle promptly") + .expect("cooperative cancellation returns a partial outcome") + }); + prop_assert!(outcome.cancelled); + + let started: BTreeSet = runner + .started + .lock() + .expect("started mutex poisoned") + .iter() + .cloned() + .collect(); + let cancelled: BTreeSet = runner + .cancelled + .lock() + .expect("cancelled mutex poisoned") + .iter() + .cloned() + .collect(); + prop_assert_eq!(cancelled, started); + } +} diff --git a/tests/fuzz_scatter_tests.rs b/tests/fuzz_scatter_tests.rs new file mode 100644 index 0000000..40ff23f --- /dev/null +++ b/tests/fuzz_scatter_tests.rs @@ -0,0 +1,155 @@ +#![cfg(feature = "mock")] +//! Property tests for scatter lane isolation and ordered gathering. + +use std::time::Duration; + +use proptest::prelude::*; +use serde_json::{Value, json}; +use tinyflows::caps::mock::mock_capabilities; +use tinyflows::compiler::compile; +use tinyflows::engine::run; +use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph}; + +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 { + Edge { + from_node: from.to_string(), + from_port: "main".to_string(), + to_node: to.to_string(), + to_port: "main".to_string(), + } +} + +fn graph(body_len: usize, requested_lanes: Option) -> WorkflowGraph { + let mut scatter_config = json!({ "path": "rows" }); + if let Some(lanes) = requested_lanes { + scatter_config["lanes"] = json!(lanes); + } + let mut nodes = vec![ + node( + "trigger", + NodeKind::Trigger, + json!({ "recursion_limit": 500, "max_node_visits": 300 }), + ), + node("scatter", NodeKind::Scatter, scatter_config), + ]; + let mut edges = vec![edge("trigger", "scatter")]; + 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, &id)); + previous = id; + } + nodes.push(node( + "gather", + NodeKind::Gather, + json!({ "from": [previous.clone()], "poll_interval_ms": 1 }), + )); + edges.push(edge(&previous, "gather")); + WorkflowGraph { + name: "generated_lane_isolation".to_string(), + nodes, + edges, + ..Default::default() + } +} + +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)) + } +} + +fn runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build runtime") +} + +proptest! { + #![proptest_config(ProptestConfig { cases: 96, ..ProptestConfig::default() })] + + #[test] + fn every_lane_has_one_disjoint_slot_at_every_body_node( + values in prop::collection::vec(any::(), 0..65), + body_len in 1usize..6, + requested in prop::option::of(1usize..33), + ) { + let graph = graph(body_len, requested); + let compiled = compile(&graph).expect("generated graph compiles"); + let rows: Vec = values.iter().map(|value| json!({ "value": value })).collect(); + let outcome = runtime().block_on(async { + tokio::time::timeout( + GUARD, + run(&compiled, json!({ "rows": rows }), &mock_capabilities()), + ) + .await + .expect("scatter/gather run hung") + .expect("generated run") + }); + + let expected = expected_lane_count(values.len(), requested); + for index in 0..body_len { + let id = format!("work_{index}"); + let slot = &outcome.output["nodes"][&id]; + // An empty scatter carries one empty, ordinary activation through + // its body so the downstream gather gets a chance to release. + // Non-empty scatters must use lane slots exclusively. + if expected > 0 { + prop_assert!( + slot.get("items").is_none(), + "lane activation wrote {id}'s top-level items: {slot}" + ); + } + let actual = slot + .get("lanes") + .and_then(Value::as_object) + .map_or(0, serde_json::Map::len); + prop_assert_eq!(actual, expected, "wrong lane count at {}", id); + + if let Some(lanes) = slot.get("lanes").and_then(Value::as_object) { + let mut indices: Vec = lanes + .values() + .filter_map(|lane| lane.get("index").and_then(Value::as_u64)) + .collect(); + indices.sort_unstable(); + prop_assert_eq!(indices, (0..expected as u64).collect::>()); + } + } + + let gathered = outcome.output["nodes"]["gather"]["items"] + .as_array() + .expect("gather items"); + let actual: Vec = gathered + .iter() + .filter_map(|item| item["json"]["value"].as_i64()) + .collect(); + let expected_values: Vec = values.iter().map(|value| i64::from(*value)).collect(); + prop_assert_eq!(actual, expected_values, "gather changed input order or cardinality"); + } +} diff --git a/tests/fuzz_timing_tests.rs b/tests/fuzz_timing_tests.rs new file mode 100644 index 0000000..c837a4e --- /dev/null +++ b/tests/fuzz_timing_tests.rs @@ -0,0 +1,151 @@ +#![cfg(feature = "mock")] +//! Generated timing schedules for parallel branches. +//! +//! 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. + +use std::sync::Arc; +use std::time::Duration; + +use proptest::prelude::*; +use serde_json::{Value, json}; +use tinyflows::caps::ToolInvoker; +use tinyflows::caps::mock::mock_capabilities; +use tinyflows::compiler::compile; +use tinyflows::engine::run; +use tinyflows::error::EngineError; +use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph}; + +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 { + Edge { + from_node: from.to_string(), + from_port: "main".to_string(), + to_node: to.to_string(), + to_port: "main".to_string(), + } +} + +fn graph(branches: usize, max_concurrency: usize) -> WorkflowGraph { + let mut nodes = vec![ + node( + "trigger", + NodeKind::Trigger, + json!({ "max_concurrency": max_concurrency }), + ), + node("fanout", NodeKind::OutputParser, Value::Null), + node("merge", NodeKind::Merge, Value::Null), + ]; + let mut edges = vec![edge("trigger", "fanout")]; + for index in 0..branches { + let id = format!("branch_{index}"); + nodes.push(node( + &id, + NodeKind::ToolCall, + json!({ + "slug": format!("branch.{index}"), + "args": { "index": index }, + "on_error": "continue", + }), + )); + edges.push(edge("fanout", &id)); + edges.push(edge(&id, "merge")); + } + WorkflowGraph { + name: "generated_timing_schedule".to_string(), + nodes, + edges, + ..Default::default() + } +} + +struct ScheduledTools { + delays: Vec, + failures: u16, +} + +#[async_trait::async_trait] +impl ToolInvoker for ScheduledTools { + async fn invoke( + &self, + slug: &str, + args: Value, + _conn: Option<&str>, + ) -> tinyflows::error::Result { + let index = slug + .rsplit('.') + .next() + .and_then(|value| value.parse::().ok()) + .expect("generated branch slug"); + for _ in 0..self.delays.get(index).copied().unwrap_or(0) { + tokio::task::yield_now().await; + } + if self.failures & (1 << index) != 0 { + Err(EngineError::Capability(format!( + "scheduled failure {index}" + ))) + } else { + Ok(json!({ "branch": index, "args": args })) + } + } +} + +async fn run_schedule( + compiled: &tinyflows::compiler::CompiledWorkflow, + delays: Vec, + failures: u16, +) -> Value { + let mut caps = mock_capabilities(); + caps.tools = Arc::new(ScheduledTools { delays, failures }); + tokio::time::timeout(GUARD, run(compiled, json!({}), &caps)) + .await + .expect("parallel run hung") + .expect("on_error continue settles") + .output +} + +fn runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build runtime") +} + +proptest! { + #![proptest_config(ProptestConfig { cases: 96, ..ProptestConfig::default() })] + + #[test] + fn merged_state_is_identical_under_opposite_completion_orders( + branches in 2usize..9, + raw_delays in prop::collection::vec(0u64..12, 8), + failures in any::(), + raw_limit in 1usize..9, + ) { + let delays: Vec = raw_delays.into_iter().take(branches).collect(); + let reverse: Vec = delays.iter().copied().rev().collect(); + let compiled = compile(&graph(branches, raw_limit.min(branches))) + .expect("generated graph compiles"); + + let (forward, backward) = runtime().block_on(async { + tokio::join!( + run_schedule(&compiled, delays, failures), + run_schedule(&compiled, reverse, failures), + ) + }); + prop_assert_eq!(forward, backward, "branch timing leaked into final state"); + } +} diff --git a/tests/reference_workflows_tests.rs b/tests/reference_workflows_tests.rs new file mode 100644 index 0000000..c22354a --- /dev/null +++ b/tests/reference_workflows_tests.rs @@ -0,0 +1,107 @@ +#![cfg(feature = "mock")] +//! Wire-format reference workflows for the asynchronous and lane node pairs. +//! +//! These intentionally start as JSON, so a renamed discriminator or config key +//! fails before execution rather than being hidden by Rust constructors. + +use std::time::Duration; + +use serde_json::{Value, json}; +use tinyflows::caps::mock::mock_capabilities; +use tinyflows::compiler::compile; +use tinyflows::engine::run; +use tinyflows::model::WorkflowGraph; + +const GUARD: Duration = Duration::from_secs(5); + +async fn load_compile_run(document: Value, input: Value) -> tinyflows::engine::RunOutcome { + let graph: WorkflowGraph = serde_json::from_value(document).expect("reference JSON loads"); + let encoded = serde_json::to_value(&graph).expect("reference graph serializes"); + let decoded: WorkflowGraph = serde_json::from_value(encoded).expect("round trip loads"); + assert_eq!(decoded, graph, "published workflow shape must round-trip"); + let compiled = compile(&graph).expect("reference workflow compiles"); + tokio::time::timeout(GUARD, run(&compiled, input, &mock_capabilities())) + .await + .expect("reference workflow hung") + .expect("reference workflow runs") +} + +#[tokio::test] +async fn spawn_and_gate_reference_workflow() { + let outcome = load_compile_run( + json!({ + "schema_version": 1, + "name": "parallel background research", + "nodes": [ + { "id": "start", "kind": "trigger", "type_version": 1, + "name": "start", "config": { "recursion_limit": 100 } }, + { "id": "fanout", "kind": "output_parser", "type_version": 1, + "name": "fanout", "config": null }, + { "id": "search", "kind": "spawn", "type_version": 1, + "name": "search", "config": { + "target": "tool", "slug": "research.search", "args": { "q": "rust" } + } }, + { "id": "summarize", "kind": "spawn", "type_version": 1, + "name": "summarize", "config": { + "target": "http", "request": { "url": "https://example.test/summary" } + } }, + { "id": "ready", "kind": "gate", "type_version": 1, + "name": "ready", "config": { + "from": ["search", "summarize"], "release": "all", + "poll_interval_ms": 1 + } } + ], + "edges": [ + { "from_node": "start", "from_port": "main", "to_node": "fanout", "to_port": "main" }, + { "from_node": "fanout", "from_port": "main", "to_node": "search", "to_port": "main" }, + { "from_node": "fanout", "from_port": "main", "to_node": "summarize", "to_port": "main" }, + { "from_node": "search", "from_port": "main", "to_node": "ready", "to_port": "main" }, + { "from_node": "summarize", "from_port": "main", "to_node": "ready", "to_port": "main" } + ] + }), + json!({}), + ) + .await; + + let items = outcome.output["nodes"]["ready"]["items"] + .as_array() + .expect("gate items"); + assert_eq!(items.len(), 2); + assert_eq!(items[0]["json"]["spec"], "tool"); + assert_eq!(items[1]["json"]["spec"], "http"); +} + +#[tokio::test] +async fn scatter_and_gather_reference_workflow() { + let outcome = load_compile_run( + json!({ + "schema_version": 1, + "name": "parallel row enrichment", + "nodes": [ + { "id": "start", "kind": "trigger", "type_version": 1, + "name": "start", "config": { "max_concurrency": 2, "recursion_limit": 100 } }, + { "id": "rows", "kind": "scatter", "type_version": 1, + "name": "rows", "config": { "path": "rows" } }, + { "id": "enrich", "kind": "transform", "type_version": 1, + "name": "enrich", "config": { "set": { "enriched": true } } }, + { "id": "all_rows", "kind": "gather", "type_version": 1, + "name": "all_rows", "config": { + "from": ["enrich"], "release": "all", "poll_interval_ms": 1 + } } + ], + "edges": [ + { "from_node": "start", "from_port": "main", "to_node": "rows", "to_port": "main" }, + { "from_node": "rows", "from_port": "main", "to_node": "enrich", "to_port": "main" }, + { "from_node": "enrich", "from_port": "main", "to_node": "all_rows", "to_port": "main" } + ] + }), + json!({ "rows": [{"id": 1}, {"id": 2}, {"id": 3}] }), + ) + .await; + + let items = outcome.output["nodes"]["all_rows"]["items"] + .as_array() + .expect("gather items"); + assert_eq!(items.len(), 3); + assert!(items.iter().all(|item| item["json"]["enriched"] == true)); +}