diff --git a/CHANGELOG.md b/CHANGELOG.md index 60439235..b49aac89 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,6 +39,35 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **New node kind: `void`, a terminal sink.** It accepts items on `main`, + discards them, and activates nothing — the branch ends there, on purpose. A + branch could always dead-end (a node with no outgoing edges terminates), but + an unwired port reads exactly like a forgotten one, so there was no way to + declare "this is a side effect and nothing waits on it". `spawn → void` is now + the explicit spelling of a ticket no `gate` will collect; the abandon + semantics are unchanged from leaving it unwired. It adds **no** concurrency: + work upstream still runs inline in its own super-step, and only the result is + dropped. Its slot is `{items: [], port: null, discarded: N}`, which keeps + "never activated" (no slot at all), "activated with nothing to drop" and + "dropped N items" distinguishable. + + Validation refuses a `void` with any outgoing edge — including the `error` + edge `on_error: "route"` would otherwise demand, which is reported directly + rather than as a `MissingErrorRoute` the next rule would then reject — and one + with no incoming edge, since a node with no effect and no input declares + nothing. + + The scatter-lane dead-end rule is relaxed accordingly: a lane branch ending in + a `void` is now legal, making it the one dead end a lane may have. The rule + exists to catch *accidental* invisibility (a lane activation never writes the + node's top-level slot), and a `void` is the author declaring it. A `scatter` + with no `gather` anywhere is still refused, void downstream or not. + + Not included, deliberately: no lint for a `spawn` with neither a `gate` nor a + `void` downstream. `validate_all` has no severity tier, and making it a hard + error would break the documented "fire-and-forget is legal" contract. A + possible future addition if a warning channel ever lands. + - **Configurable agents.** An `agent` node can now be given dynamic context, an explicit tool allow-list, a model and provider, a working directory, advisory limits, and arbitrary harness metadata — while the agent diff --git a/README.md b/README.md index 0f24437b..dc2ede6a 100644 --- a/README.md +++ b/README.md @@ -244,6 +244,7 @@ done | `merge` | Fan-in barrier that combines multiple inputs; waits for all wired predecessors. | | `split_out` | Fan-out that emits one item per element of a list. | | `transform` | Pure, expression-based data transform / field mapping over the run state. | +| `void` | Terminal sink: discards its input and runs nothing downstream — the explicit dead end. | See the [Node Catalog](../../wiki/Node-Catalog) wiki page for config keys and ports. diff --git a/src/catalog.rs b/src/catalog.rs index 0c985897..e99d8e3c 100644 --- a/src/catalog.rs +++ b/src/catalog.rs @@ -33,7 +33,7 @@ use group_03::*; /// The node kinds, in the canonical order used wherever the DSL is enumerated /// (matches [`NodeKind`](crate::model::NodeKind)'s serde discriminators). -pub const NODE_KINDS: [&str; 20] = [ +pub const NODE_KINDS: [&str; 21] = [ "trigger", "agent", "tool_call", @@ -54,6 +54,7 @@ pub const NODE_KINDS: [&str; 20] = [ "gate", "scatter", "gather", + "void", ]; /// One config field a node of a given kind reads at run time. @@ -204,6 +205,7 @@ pub fn contract_for(kind: &str) -> Option { "gate" => contract_gate(), "scatter" => contract_scatter(), "gather" => contract_gather(), + "void" => contract_void(), _ => return None, }; Some(with_fan_out_fields(c)) diff --git a/src/catalog/contracts/group_02.rs b/src/catalog/contracts/group_02.rs index 4f97d73c..64ecb3d7 100644 --- a/src/catalog/contracts/group_02.rs +++ b/src/catalog/contracts/group_02.rs @@ -371,8 +371,9 @@ pub(super) fn contract_spawn() -> NodeKindContract { the same, the concurrency is not. That is a silent performance cliff, so check \ the host wires a TaskRunner before relying on overlap." .to_string(), - "Fire-and-forget is legal — a spawn no gate ever collects simply runs. If that \ - is not what you meant, wire a `gate`." + "Fire-and-forget is legal — a spawn no gate ever collects simply runs. Wire the \ + spawn into a `void` to say that on purpose; wire it into a `gate` if you \ + actually wanted the results." .to_string(), ], } diff --git a/src/catalog/contracts/group_03.rs b/src/catalog/contracts/group_03.rs index 30fd9712..09795ca4 100644 --- a/src/catalog/contracts/group_03.rs +++ b/src/catalog/contracts/group_03.rs @@ -115,3 +115,49 @@ pub(super) fn contract_gather() -> NodeKindContract { ], } } + +pub(super) fn contract_void() -> NodeKindContract { + NodeKindContract { + kind: "void".to_string(), + summary: "Terminal sink: accepts items, discards them, and runs nothing downstream." + .to_string(), + description: "The branch ends here, on purpose. A branch could always dead-end — a \ + node with no outgoing edges terminates — but an unwired port reads \ + exactly like a forgotten one, so there was no way to SAY \"this is a \ + side effect and nothing waits on it\". This node is that sentence. It \ + adds no concurrency: the work upstream still runs inline in its own \ + super-step, and only the result is dropped. For work that should \ + actually overlap, use `spawn`." + .to_string(), + config_fields: vec![], + ports: PortSpec::new(&["main"], &[]), + example: json!({ + "id": "drop", "kind": "void", "name": "Fire and forget: audit log", + "config": {} + }), + notes: vec![ + "An outgoing edge is a validation error, and so is having no incoming edge. \ + `on_error: \"route\"` is refused for the same reason — an `error` edge is an \ + outgoing edge; use \"stop\" or \"continue\"." + .to_string(), + "It writes {items: [], port: null, discarded: N} into its slot. A node that never \ + ran has no slot at all, so \"never activated\", \"activated with nothing to \ + drop\" (discarded: 0) and \"dropped N items\" stay distinguishable." + .to_string(), + "`discarded` counts THIS activation, not a running total: inside a loop body the \ + last iteration's value is what survives, and inside a scatter lane it lands \ + under lanes. rather than at the top level." + .to_string(), + "spawn -> void is the explicit spelling of a ticket no `gate` will ever collect. \ + Same abandon semantics as leaving it unwired, but said out loud." + .to_string(), + "It is the one dead end a scatter lane may have. Every other lane branch must \ + reach the `gather`, because a stranded lane's output is invisible rather than \ + merely uncollected — a void makes that invisibility the contract." + .to_string(), + "There is no `reason` config. The node's `name` is where the human explanation \ + goes, and unlike a config key it is rendered in the graph." + .to_string(), + ], + } +} diff --git a/src/catalog_tests.rs b/src/catalog_tests.rs index 3ad73936..5bcec5e5 100644 --- a/src/catalog_tests.rs +++ b/src/catalog_tests.rs @@ -23,7 +23,7 @@ fn every_node_kind_has_a_contract() { } } } - assert_eq!(all_contracts().len(), 20); + assert_eq!(all_contracts().len(), 21); } #[test] @@ -48,8 +48,8 @@ fn unknown_kind_has_no_contract() { } #[test] -fn node_kinds_has_20_entries_including_the_async_and_lane_pairs() { - assert_eq!(NODE_KINDS.len(), 20); +fn node_kinds_has_21_entries_including_the_async_and_lane_pairs() { + assert_eq!(NODE_KINDS.len(), 21); assert!(NODE_KINDS.contains(&"shell")); assert!(NODE_KINDS.contains(&"memory")); assert!(NODE_KINDS.contains(&"dedup")); @@ -67,6 +67,29 @@ fn node_kinds_has_20_entries_including_the_async_and_lane_pairs() { assert_eq!(NODE_KINDS[17], "gate"); assert_eq!(NODE_KINDS[18], "scatter"); assert_eq!(NODE_KINDS[19], "gather"); + assert!(NODE_KINDS.contains(&"void")); + assert_eq!(NODE_KINDS[20], "void"); +} + +#[test] +fn void_contract_takes_no_config_and_declares_no_output_port() { + // The two claims an authoring tool acts on: there is nothing to configure, + // and there is nowhere to draw an edge to. Both are enforced by validation, + // so the contract must not suggest otherwise. + let c = contract_for("void").expect("void contract exists"); + assert!( + c.config_fields.is_empty(), + "void takes no config; the reason goes in the node's name" + ); + assert_eq!(c.ports, PortSpec::new(&["main"], &[])); + assert!( + c.notes.iter().any(|n| n.contains("outgoing edge")), + "the contract must say an outgoing edge is refused" + ); + assert!( + c.notes.iter().any(|n| n.contains("spawn -> void")), + "the contract must document the ungathered-ticket spelling" + ); } #[test] diff --git a/src/fan_out_contract_tests.rs b/src/fan_out_contract_tests.rs index cdca025e..f42e02c4 100644 --- a/src/fan_out_contract_tests.rs +++ b/src/fan_out_contract_tests.rs @@ -24,6 +24,7 @@ fn kinds_that_cannot_map_do_not_advertise_them() { "merge", "transform", "code", + "void", ] { let c = contract_for(kind).expect("contract"); assert!( diff --git a/src/model/node_kind.rs b/src/model/node_kind.rs index d6401d54..939e9672 100644 --- a/src/model/node_kind.rs +++ b/src/model/node_kind.rs @@ -98,6 +98,23 @@ pub enum NodeKind { /// and leave the stragglers, or settle for whatever arrived before its /// deadline. See [`crate::nodes::release`]. Gate, + /// Terminal sink: accepts items on `main`, discards them, and activates no + /// successors. The branch ends here, on purpose. + /// + /// Purely declarative. A branch could always dead-end — a node with no + /// outgoing edges is lowered straight to the engine's `END` sentinel — but + /// an unwired port reads exactly like a forgotten one, so there was no way + /// to *say* "this is a side effect and nothing waits on it". This kind is + /// that sentence. It adds no concurrency: the work upstream of it still + /// runs inline in its own super-step, and only the result is dropped. + /// + /// Abandon semantics, identical to an ungathered [`NodeKind::Spawn`] + /// ticket — nothing is drained or cancelled at run end. `spawn` → `void` is + /// the explicit spelling of a ticket no [`NodeKind::Gate`] will collect. + /// + /// [`crate::validate`] refuses a `void` that has any outgoing edge, or that + /// has no incoming edge at all. + Void, } /// How a [`NodeKind::Trigger`] node is fired. diff --git a/src/model/node_kind_tests.rs b/src/model/node_kind_tests.rs index 2772c852..bb4f34d4 100644 --- a/src/model/node_kind_tests.rs +++ b/src/model/node_kind_tests.rs @@ -29,6 +29,7 @@ fn node_kind_variants_use_snake_case() { assert_wire(&NodeKind::SubWorkflow, "sub_workflow"); assert_wire(&NodeKind::Memory, "memory"); assert_wire(&NodeKind::Dedup, "dedup"); + assert_wire(&NodeKind::Void, "void"); } #[test] diff --git a/src/nodes/control_flow/mod.rs b/src/nodes/control_flow/mod.rs index 139537f0..035de4e8 100644 --- a/src/nodes/control_flow/mod.rs +++ b/src/nodes/control_flow/mod.rs @@ -1,8 +1,8 @@ //! Native control-flow node executors: `condition`, `switch`, `merge`, -//! `split_out`, `transform`, `loop`, and `dedup`. Most of these are pure — they -//! route and reshape data within the engine and use no host capability. `dedup` -//! is the one exception: it filters items against durable `StateStore` state -//! (see [`dedup`] for why it lives here anyway). +//! `split_out`, `transform`, `loop`, `void`, and `dedup`. Most of these are +//! pure — they route and reshape data within the engine and use no host +//! capability. `dedup` is the one exception: it filters items against durable +//! `StateStore` state (see [`dedup`] for why it lives here anyway). //! //! One module per node kind so parallel work can edit them without conflicts. @@ -15,6 +15,7 @@ pub mod scatter; pub mod split_out; pub mod switch; pub mod transform; +pub mod void; pub use condition::ConditionNode; pub use dedup::DedupNode; @@ -25,3 +26,4 @@ pub use scatter::{MAX_LANES, ScatterNode}; pub use split_out::SplitOutNode; pub use switch::SwitchNode; pub use transform::TransformNode; +pub use void::VoidNode; diff --git a/src/nodes/control_flow/void.rs b/src/nodes/control_flow/void.rs new file mode 100644 index 00000000..e94383bb --- /dev/null +++ b/src/nodes/control_flow/void.rs @@ -0,0 +1,102 @@ +//! `void` — the terminal sink. +//! +//! A `void` node accepts items on `main`, discards them, and activates nothing. +//! It is the end of its branch, and that is the whole feature. +//! +//! # Why a node for something the graph could already do +//! +//! A branch could always dead-end: [`crate::engine`] lowers any node with no +//! outgoing edges straight to the state-graph's `END` sentinel, and `END` is +//! filtered out of routing, so it contributes nothing to the next super-step's +//! active set. Wiring nothing to a port has exactly the same effect. +//! +//! What was missing is the *statement*. A port left unwired reads identically +//! to a port someone forgot to wire, so an author could not declare "this +//! branch is a side effect; nothing downstream waits on it", and a reviewer +//! could not tell intent from an accident. A `void` says it in the graph, where +//! both the reader and [`crate::validate`] can see it. +//! +//! One place that ambiguity was resolved *against* the author: a branch inside +//! a [`scatter`](crate::model::NodeKind::Scatter) lane that dead-ends is a hard +//! validation error, because a lane activation never writes the node's +//! top-level slot and so a stranded lane branch produces a wrong answer rather +//! than a failure. `void` makes that invisibility the contract instead of the +//! accident, and is therefore the one dead end a lane may have. +//! +//! # What it is not +//! +//! It adds **no concurrency**. Everything upstream of a `void` still runs +//! inline in its own super-step; only the *result* is dropped. If you want work +//! to overlap, that is [`spawn`](crate::model::NodeKind::Spawn) and the +//! `TaskRunner` capability, not this. +//! +//! It performs **no drain and no cancellation** at run end. Abandoning a branch +//! here is exactly what an ungathered `spawn` ticket already does, which makes +//! `spawn` → `void` the explicit spelling of "no [`gate`] will ever collect +//! this, and I meant that". +//! +//! [`gate`]: crate::model::NodeKind::Gate +//! +//! # Configuration +//! +//! None. A `void` node's `name` is where the human reason goes ("Fire and +//! forget: audit log") — it is already required, and unlike a config key it is +//! rendered by [`crate::visualization`]. Config is ignored entirely, including +//! `={{ … }}` expressions, so this node can emit no binding diagnostics. +//! +//! # What it leaves behind +//! +//! `{ "items": [], "port": null, "discarded": }` in its run-state slot. +//! Emitting nothing would otherwise be indistinguishable from never having run, +//! since a node that never ran has no slot at all — so the three cases stay +//! separable: +//! +//! | slot | meaning | +//! |---|---| +//! | absent (`null`) | never activated | +//! | `discarded: 0` | activated, had nothing to drop | +//! | `discarded: 3` | activated, dropped three items | +//! +//! `discarded` counts **this activation's** input, not a running total. Two +//! consequences worth knowing: inside a `loop` body it is overwritten every +//! iteration, so the value that survives is the last one; and inside a scatter +//! lane it lands at `nodes..lanes..discarded` rather than at the top +//! level, because a lane activation deliberately never writes the top-level +//! slot. A cumulative counter was considered and rejected for exactly those two +//! reasons — it would silently mean something different in each context. + +use async_trait::async_trait; +use serde_json::json; + +use crate::error::Result; +use crate::nodes::{NodeContext, NodeExecutor, NodeOutput}; + +/// Stable `tracing` grep prefix for every log line this node emits. +const LOG_PREFIX: &str = "[void]"; + +/// Terminal sink: discards its input and activates no successors. +/// +/// See the [module docs](self) for why an explicit node beats an unwired port. +#[derive(Debug, Default, Clone)] +pub struct VoidNode; + +#[async_trait] +impl NodeExecutor for VoidNode { + async fn execute(&self, ctx: NodeContext<'_>) -> Result { + let discarded = ctx.input.len(); + tracing::debug!( + node = %ctx.node.id, + discarded, + "{LOG_PREFIX} discarding items; nothing downstream runs" + ); + // No port: emitting on one would imply a successor could match it, and + // `validate` guarantees there is none. The `discarded` count is the only + // trace the node leaves, and it is what separates "ran on nothing" from + // "never ran" (see the module docs). + Ok(NodeOutput::empty().with_meta(json!({ "discarded": discarded }))) + } +} + +#[cfg(test)] +#[path = "void_tests.rs"] +mod tests; diff --git a/src/nodes/control_flow/void_tests.rs b/src/nodes/control_flow/void_tests.rs new file mode 100644 index 00000000..e18ad958 --- /dev/null +++ b/src/nodes/control_flow/void_tests.rs @@ -0,0 +1,127 @@ +use super::*; +use crate::caps::Capabilities; +use crate::caps::mock::mock_capabilities; +use crate::data::Item; +use crate::model::{Node, NodeKind}; +use serde_json::{Value, json}; + +fn void_node(id: &str, config: Value) -> Node { + Node { + id: id.to_string(), + kind: NodeKind::Void, + type_version: 1, + name: id.to_string(), + config, + ports: Vec::new(), + position: None, + } +} + +async fn run_void(caps: &Capabilities, node: &Node, input: &[Item]) -> NodeOutput { + let run = Value::Null; + let ctx = NodeContext { + node, + input, + run: &run, + nodes: &Value::Null, + caps, + agents: &[], + observer: &crate::observability::NoopObserver, + token: crate::engine::CancellationToken::new(), + lane: None, + resume: None, + step: 0, + }; + VoidNode.execute(ctx).await.expect("execute") +} + +#[tokio::test] +async fn discards_every_input_item_and_emits_nothing() { + // Test 1 (spec): the branch ends here — no items, no port to route on, no + // control instruction that could put the node back on the active set. + let caps = mock_capabilities(); + let node = void_node("sink", Value::Null); + let input = vec![ + Item::new(json!({ "id": "a" })), + Item::new(json!({ "id": "b" })), + Item::new(json!({ "id": "c" })), + ]; + + let out = run_void(&caps, &node, &input).await; + + assert!(out.items.is_empty(), "void must emit no items"); + assert!(out.port.is_none(), "void must not route on a port"); + assert!(out.control.is_none(), "void must not re-enter or interrupt"); +} + +#[tokio::test] +async fn records_the_discarded_count_in_meta() { + // Test 2 (spec): the count is the node's only trace, so it must be exact. + let caps = mock_capabilities(); + let node = void_node("sink", Value::Null); + let input = vec![ + Item::new(json!({ "id": "a" })), + Item::new(json!({ "id": "b" })), + Item::new(json!({ "id": "c" })), + ]; + + let out = run_void(&caps, &node, &input).await; + + assert_eq!(out.meta, Some(json!({ "discarded": 3 }))); +} + +#[tokio::test] +async fn empty_input_still_reports_zero_discarded() { + // Test 3 (spec): "activated with nothing to drop" must stay distinguishable + // from "never activated", which shows up as an absent slot rather than a + // zero. Without the meta both look identical downstream. + let caps = mock_capabilities(); + let node = void_node("sink", Value::Null); + + let out = run_void(&caps, &node, &[]).await; + + assert_eq!(out.meta, Some(json!({ "discarded": 0 }))); + assert!(out.items.is_empty()); +} + +#[tokio::test] +async fn ignores_config_entirely_and_emits_no_diagnostics() { + // Test 4 (spec): the kind declares no config fields, and in particular it + // must not resolve `=` expressions — a void that emitted null-binding + // diagnostics would be noise about data it exists to throw away. + let caps = mock_capabilities(); + let node = void_node( + "sink", + json!({ + "reason": "fire and forget", + "key": "=item.does.not.exist", + "on_error": "stop", + }), + ); + let input = vec![Item::new(json!({ "id": "a" }))]; + + let out = run_void(&caps, &node, &input).await; + + assert!(out.items.is_empty()); + assert!( + out.diagnostics.is_empty(), + "void resolves no expressions, so it can raise no null-binding diagnostics" + ); + assert_eq!(out.meta, Some(json!({ "discarded": 1 }))); +} + +#[tokio::test] +async fn is_pure_across_repeated_activations() { + // Test 5 (spec): the node touches no capability and holds no state, so a + // second activation behaves exactly like the first. This is what lets a + // void sit inside a loop body without accumulating anything. + let caps = mock_capabilities(); + let node = void_node("sink", Value::Null); + let input = vec![Item::new(json!({ "id": "a" }))]; + + let first = run_void(&caps, &node, &input).await; + let second = run_void(&caps, &node, &input).await; + + assert_eq!(first.meta, second.meta); + assert_eq!(first.meta, Some(json!({ "discarded": 1 }))); +} diff --git a/src/nodes/execution.rs b/src/nodes/execution.rs index 830a24fc..c53618bd 100644 --- a/src/nodes/execution.rs +++ b/src/nodes/execution.rs @@ -332,6 +332,7 @@ pub(crate) fn executor_for(kind: &NodeKind) -> Box { NodeKind::SplitOut => Box::new(control_flow::SplitOutNode), NodeKind::Transform => Box::new(control_flow::TransformNode), NodeKind::Dedup => Box::new(control_flow::DedupNode), + NodeKind::Void => Box::new(control_flow::VoidNode), NodeKind::Scatter => Box::new(control_flow::ScatterNode), NodeKind::Gather => Box::new(control_flow::GatherNode), NodeKind::Spawn => Box::new(integration::SpawnNode), diff --git a/src/nodes/mod_tests.rs b/src/nodes/mod_tests.rs index b9731d4e..1b42805d 100644 --- a/src/nodes/mod_tests.rs +++ b/src/nodes/mod_tests.rs @@ -8,7 +8,7 @@ use serde_json::json; fn all_kinds() -> Vec { use NodeKind::{ Agent, Code, Condition, Dedup, HttpRequest, Loop, Memory, Merge, OutputParser, Shell, - SplitOut, SubWorkflow, Switch, ToolCall, Transform, Trigger, + SplitOut, SubWorkflow, Switch, ToolCall, Transform, Trigger, Void, }; vec![ Trigger, @@ -27,6 +27,7 @@ fn all_kinds() -> Vec { Memory, Dedup, Loop, + Void, ] } diff --git a/src/validate.rs b/src/validate.rs index d5613d0a..21b0925c 100644 --- a/src/validate.rs +++ b/src/validate.rs @@ -22,8 +22,9 @@ fn kind_name(kind: &NodeKind) -> String { /// Currently checks: unique node ids, exactly one trigger node, that every edge /// references existing nodes, no duplicate edges, per-node `on_error` policy /// sanity (a known value, and an `error` edge when the policy is `route`), -/// declared-input sanity (addressable, unique names; defaults that match their -/// declared type), and loop legality (see [`validate_loops`] — cycles are +/// `void` topology (a terminal sink may have no outgoing edge, and must have an +/// incoming one), declared-input sanity (addressable, unique names; defaults +/// that match their declared type), and loop legality (see [`validate_loops`] — cycles are /// permitted; only the ones that cannot iterate are refused). /// /// # Errors @@ -43,7 +44,8 @@ pub fn validate(graph: &WorkflowGraph) -> Result<(), ValidationError> { /// /// Returns an empty `Vec` for a valid graph. The checks are ordered /// deterministically (duplicate ids → trigger count → edge integrity → -/// `on_error` policy → per-kind config → condition routing → declared inputs), +/// `on_error` policy → per-kind config → `void` topology → condition routing → +/// declared inputs), /// and every error is self-contained (no check can panic on a graph that failed /// an earlier one), so accumulating is safe. The first element is identical to what /// [`validate`] returns, preserving the historical single-error contract. @@ -108,12 +110,27 @@ pub fn validate_all(graph: &WorkflowGraph) -> Vec { match on_error { "stop" | "continue" => {} "route" => { - let has_error_edge = graph - .edges - .iter() - .any(|e| e.from_node == node.id && e.from_port == "error"); - if !has_error_edge { - errors.push(ValidationError::MissingErrorRoute(node.id.clone())); + if node.kind == NodeKind::Void { + // An `error` edge is still an outgoing edge, which the + // `void` check below forbids. Caught here instead of + // letting `MissingErrorRoute` fire, or the author would be + // told to add an edge that the next rule then rejects — + // advice with no fixed point. + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: "`void` is a terminal sink, so on_error \"route\" has nowhere to \ + route to (an `error` edge is an outgoing edge); use \"stop\" or \ + \"continue\"" + .to_string(), + }); + } else { + let has_error_edge = graph + .edges + .iter() + .any(|e| e.from_node == node.id && e.from_port == "error"); + if !has_error_edge { + errors.push(ValidationError::MissingErrorRoute(node.id.clone())); + } } } other => { @@ -397,6 +414,47 @@ pub fn validate_all(graph: &WorkflowGraph) -> Vec { } } + // `void` node topology checks. The kind asserts exactly one thing — "the + // branch ends here, deliberately" — so the two ways to contradict it are + // refused rather than absorbed. An outgoing edge would be dead (a leaf + // lowers to the engine's `END` sentinel) or would make the node not a void; + // and a void nothing routes into declares nothing at all, since a node with + // no effect and no input is the one orphan that cannot be work in progress. + // There is no general orphan check in this crate, and adding one is out of + // scope; this rule is safe precisely because the kind is new, so no + // existing graph can trip it. + for node in &graph.nodes { + if node.kind != NodeKind::Void { + continue; + } + let mut outgoing: Vec<&str> = graph + .edges + .iter() + .filter(|e| e.from_node == node.id) + .map(|e| e.to_node.as_str()) + .collect(); + outgoing.sort_unstable(); + outgoing.dedup(); + if !outgoing.is_empty() { + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: format!( + "`void` is a terminal sink and may not have outgoing edges (found \ + {outgoing:?}); remove the edge, or use a different kind if the branch is \ + meant to continue" + ), + }); + } + if !graph.edges.iter().any(|e| e.to_node == node.id) { + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: "`void` has no incoming edge, so it can never run and declares nothing; \ + wire the branch it is meant to terminate, or delete it" + .to_string(), + }); + } + } + // A `condition` node's outgoing edges must emit on `from_port` "true" or // "false" — routing is keyed EXCLUSIVELY on `from_port` (see // `engine::outgoing_by_port` / `handler_routing`), so any other value diff --git a/src/validate/scatter.rs b/src/validate/scatter.rs index c53ee3ec..a770cb4c 100644 --- a/src/validate/scatter.rs +++ b/src/validate/scatter.rs @@ -22,6 +22,12 @@ pub(super) fn validate_scatter_regions(graph: &WorkflowGraph, errors: &mut Vec = graph + .nodes + .iter() + .filter(|n| n.kind == NodeKind::Void) + .map(|n| n.id.as_str()) + .collect(); // A gather with no scatter waits on lanes nobody will ever open. for gather in &gathers { @@ -103,10 +109,19 @@ pub(super) fn validate_scatter_regions(graph: &WorkflowGraph, errors: &mut Vec) -> WorkflowGraph include!("validate_tests/validate_tests_part_01_tests.rs"); include!("validate_tests/validate_tests_part_02_tests.rs"); include!("validate_tests/validate_tests_part_03_tests.rs"); +include!("validate_tests/validate_tests_part_04_tests.rs"); diff --git a/src/validate_tests/validate_tests_part_04_tests.rs b/src/validate_tests/validate_tests_part_04_tests.rs new file mode 100644 index 00000000..62b69f3e --- /dev/null +++ b/src/validate_tests/validate_tests_part_04_tests.rs @@ -0,0 +1,230 @@ +// `void` topology rules. The kind asserts exactly one thing — "the branch ends +// here, deliberately" — so these pin the two ways to contradict it, the +// `on_error` interaction, and the one place the rule had to be *relaxed* +// (a lane branch may end in a void) without opening the hole it was guarding. + +use serde_json::json; + +fn void_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(), + } +} + +/// `t -> v(void)` — the minimal legal void. +fn graph_with_void() -> WorkflowGraph { + WorkflowGraph { + nodes: vec![node("t", NodeKind::Trigger), node("v", NodeKind::Void)], + edges: vec![void_edge("t", "v")], + ..Default::default() + } +} + +#[test] +fn void_as_a_leaf_is_accepted() { + assert_eq!(validate(&graph_with_void()), Ok(())); +} + +#[test] +fn void_with_an_outgoing_edge_is_rejected() { + let mut graph = graph_with_void(); + graph.nodes.push(node("x", NodeKind::Transform)); + graph.edges.push(void_edge("v", "x")); + + let errors = validate_all(&graph); + let reason = errors + .iter() + .find_map(|e| match e { + ValidationError::InvalidNodeConfig { node, reason } if node == "v" => Some(reason), + _ => None, + }) + .expect("void with an outgoing edge should be rejected"); + assert!( + reason.contains("terminal sink") && reason.contains("\"x\""), + "the message should name the offending target: {reason}" + ); +} + +#[test] +fn void_with_several_outgoing_edges_names_them_deterministically() { + // The message lists targets, so it has to be stable across runs — `errors` + // is a Vec an author reads, not a set. + let mut graph = graph_with_void(); + graph.nodes.push(node("b", NodeKind::Transform)); + graph.nodes.push(node("a", NodeKind::Transform)); + graph.edges.push(void_edge("v", "b")); + graph.edges.push(void_edge("v", "a")); + + let errors = validate_all(&graph); + let reason = errors + .iter() + .find_map(|e| match e { + ValidationError::InvalidNodeConfig { node, reason } if node == "v" => Some(reason), + _ => None, + }) + .expect("rejected"); + assert!(reason.contains("[\"a\", \"b\"]"), "{reason}"); +} + +#[test] +fn void_with_no_incoming_edge_is_rejected() { + let graph = WorkflowGraph { + nodes: vec![node("t", NodeKind::Trigger), node("v", NodeKind::Void)], + edges: vec![], + ..Default::default() + }; + + let errors = validate_all(&graph); + assert!( + errors.iter().any(|e| matches!( + e, + ValidationError::InvalidNodeConfig { node, reason } + if node == "v" && reason.contains("no incoming edge") + )), + "an orphan void declares nothing: {errors:?}" + ); +} + +#[test] +fn void_with_on_error_route_is_rejected_without_demanding_an_error_edge() { + // The whole point of the special case: `MissingErrorRoute` would tell the + // author to add an edge the void rule then rejects. + let mut graph = graph_with_void(); + graph.nodes[1].config = json!({ "on_error": "route" }); + + let errors = validate_all(&graph); + assert!( + !errors + .iter() + .any(|e| matches!(e, ValidationError::MissingErrorRoute(_))), + "must not ask for an `error` edge it would then refuse: {errors:?}" + ); + assert!( + errors.iter().any(|e| matches!( + e, + ValidationError::InvalidNodeConfig { node, reason } + if node == "v" && reason.contains("nowhere to route to") + )), + "{errors:?}" + ); +} + +#[test] +fn void_accepts_on_error_stop_and_continue() { + for policy in ["stop", "continue"] { + let mut graph = graph_with_void(); + graph.nodes[1].config = json!({ "on_error": policy }); + assert_eq!(validate(&graph), Ok(()), "on_error {policy} should be legal"); + } +} + +#[test] +fn execution_is_still_rejected_on_void() { + // Pins void out of the mapping kinds: it consumes a batch, it does not map + // over one. + let mut graph = graph_with_void(); + graph.nodes[1].config = json!({ "execution": "per_item" }); + + let errors = validate_all(&graph); + assert!( + errors.iter().any(|e| matches!( + e, + ValidationError::InvalidNodeConfig { node, reason } + if node == "v" && reason.contains("void") + )), + "{errors:?}" + ); +} + +// --- scatter lane interaction --------------------------------------------- + +/// `t -> fan(scatter) -> work -> collect(gather)`, plus an optional side branch +/// hanging off `work` whose nodes are appended by the caller. +fn lane_graph(extra_nodes: Vec, extra_edges: Vec) -> WorkflowGraph { + let mut nodes = vec![ + node("t", NodeKind::Trigger), + node("fan", NodeKind::Scatter), + node("work", NodeKind::Transform), + node("collect", NodeKind::Gather), + ]; + nodes[3].config = json!({ "from": ["work"] }); + nodes.extend(extra_nodes); + + let mut edges = vec![ + void_edge("t", "fan"), + void_edge("fan", "work"), + void_edge("work", "collect"), + ]; + edges.extend(extra_edges); + + WorkflowGraph { + nodes, + edges, + ..Default::default() + } +} + +#[test] +fn a_lane_side_branch_may_end_in_void() { + // The primary use case: fire-and-forget inside a lane, which was previously + // impossible to express because every lane node had to reach the gather. + let graph = lane_graph( + vec![node("notify", NodeKind::Transform), node("v", NodeKind::Void)], + vec![void_edge("work", "notify"), void_edge("notify", "v")], + ); + assert_eq!(validate(&graph), Ok(())); +} + +#[test] +fn a_lane_side_branch_that_dead_ends_without_a_void_is_still_rejected() { + // Same shape, minus the void: still an accident, still refused. + let graph = lane_graph( + vec![node("notify", NodeKind::Transform)], + vec![void_edge("work", "notify")], + ); + + let errors = validate_all(&graph); + assert!( + errors.iter().any(|e| matches!( + e, + ValidationError::InvalidNodeConfig { node, reason } + if node == "notify" && reason.contains("no path onward to a `gather`") + )), + "{errors:?}" + ); +} + +#[test] +fn a_scatter_whose_only_path_ends_in_void_is_still_rejected() { + // The hole the relaxation must not open. A scatter with no gather anywhere + // is a plain fan-out wearing a scatter costume, and a void downstream does + // not make it one — `region_members` yields nothing, so the "no `gather` + // downstream" error still fires. + let graph = WorkflowGraph { + nodes: vec![ + node("t", NodeKind::Trigger), + node("fan", NodeKind::Scatter), + node("work", NodeKind::Transform), + node("v", NodeKind::Void), + ], + edges: vec![ + void_edge("t", "fan"), + void_edge("fan", "work"), + void_edge("work", "v"), + ], + ..Default::default() + }; + + let errors = validate_all(&graph); + assert!( + errors.iter().any(|e| matches!( + e, + ValidationError::InvalidNodeConfig { node, reason } + if node == "fan" && reason.contains("no `gather` downstream") + )), + "{errors:?}" + ); +} diff --git a/src/visualization.rs b/src/visualization.rs index d9e03aef..f7472bfc 100644 --- a/src/visualization.rs +++ b/src/visualization.rs @@ -335,6 +335,7 @@ fn kind_name(kind: &NodeKind) -> &'static str { NodeKind::Scatter => "scatter", NodeKind::Gather => "gather", NodeKind::Gate => "gate", + NodeKind::Void => "void", } } diff --git a/tests/smoke_all_nodes.rs b/tests/smoke_all_nodes.rs index cb0ac80e..ce85bc9b 100644 --- a/tests/smoke_all_nodes.rs +++ b/tests/smoke_all_nodes.rs @@ -7,6 +7,9 @@ //! asserts the run succeeds and the node produced a well-formed slot (an `items` //! array). A final big test chains many kinds together end to end. //! +//! `void` is the one exception to the non-empty-slot rule: it exists to emit +//! nothing, so it gets its own assertions rather than `smoke_single_node`'s. +//! //! Gated behind the `mock` feature, so plain `cargo test` skips it while //! `cargo test --all-features` runs it. @@ -372,3 +375,29 @@ async fn smoke_scatter_gather() { .expect("the gather should produce an items array"); assert_eq!(items.len(), 2, "two lanes, two collected results"); } + +#[tokio::test] +async fn smoke_void() { + // `smoke_single_node` asserts a non-empty `items` slot, which a void can + // never satisfy — emitting nothing is the whole contract. What must hold + // instead is that it ran at all (a slot exists), that the slot is empty, + // and that it counted what it dropped. + let graph = WorkflowGraph { + name: "smoke".to_string(), + nodes: vec![ + trigger("t", TriggerKind::Manual), + node("n", NodeKind::Void, Value::Null), + ], + edges: vec![edge("t", "main", "n")], + ..Default::default() + }; + let compiled = compile(&graph).expect("compile"); + let outcome = run(&compiled, json!({ "x": 1 }), &mock_capabilities()) + .await + .expect("run should succeed"); + + assert_eq!(outcome.output["nodes"]["n"]["items"], json!([])); + assert_eq!(outcome.output["nodes"]["n"]["discarded"], 1); + assert!(outcome.output["nodes"]["n"]["port"].is_null()); + assert!(outcome.pending_approvals.is_empty()); +} diff --git a/tests/void_node_tests.rs b/tests/void_node_tests.rs new file mode 100644 index 00000000..d68ed085 --- /dev/null +++ b/tests/void_node_tests.rs @@ -0,0 +1,332 @@ +#![cfg(feature = "mock")] +//! End-to-end tests for the `void` node kind — the terminal sink. +//! +//! The claim under test is that a void ends its branch *without* costing +//! anything anywhere else: the sibling arm of a fan-out still produces its +//! output, a merge downstream of the other arm still fires, and a loop whose +//! body has a void side branch still iterates to its bound. Those are the +//! failure modes worth guarding, because all three would show up as a hang or a +//! silently missing result rather than as an error. +//! +//! Every run is wrapped in a timeout: a void that somehow stranded a barrier +//! would hang rather than fail, and a hung test takes the suite with it. +//! +//! Gated behind the `mock` cargo feature so plain `cargo test` skips it. + +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::{Edge, Node, NodeKind, WorkflowGraph}; + +/// How long any single run in this file may take before it is called a hang. +const GUARD: Duration = Duration::from_secs(20); + +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 port_edge(from: &str, from_port: &str, to: &str) -> Edge { + Edge { + from_node: from.to_string(), + from_port: from_port.to_string(), + to_node: to.to_string(), + to_port: "main".to_string(), + } +} + +async fn run_graph(graph: &WorkflowGraph, input: Value) -> Value { + let compiled = compile(graph).expect("compile"); + let outcome = tokio::time::timeout(GUARD, run(&compiled, input, &mock_capabilities())) + .await + .expect("run should not hang") + .expect("run should succeed"); + assert!( + outcome.pending_approvals.is_empty(), + "no approvals expected: {:?}", + outcome.pending_approvals + ); + outcome.output +} + +#[tokio::test] +async fn fan_out_arm_into_void_does_not_block_the_other_arm() { + // t -> fan -> {sink(void), keep}. The void arm must neither hold up `keep` + // nor stop the run from completing. + let graph = WorkflowGraph { + name: "void_fan_out".to_string(), + nodes: vec![ + node("t", NodeKind::Trigger, Value::Null), + node("fan", NodeKind::Transform, json!({ "set": { "n": 1 } })), + node( + "keep", + NodeKind::Transform, + json!({ "set": { "tag": "kept" } }), + ), + node("sink", NodeKind::Void, Value::Null), + ], + edges: vec![edge("t", "fan"), edge("fan", "keep"), edge("fan", "sink")], + ..Default::default() + }; + + let out = run_graph(&graph, json!({ "seed": 1 })).await; + + assert_eq!(out["nodes"]["keep"]["items"][0]["json"]["tag"], "kept"); + assert_eq!(out["nodes"]["sink"]["items"], json!([])); + assert_eq!(out["nodes"]["sink"]["discarded"], 1); + assert!( + out["nodes"]["sink"]["port"].is_null(), + "a void routes on no port" + ); +} + +#[tokio::test] +async fn a_void_that_never_runs_leaves_no_slot_at_all() { + // The distinction the `discarded` counter exists to preserve: an untaken + // branch's void has no slot, which is what separates it from a void that + // ran and dropped nothing. + let graph = WorkflowGraph { + name: "void_untaken".to_string(), + nodes: vec![ + node("t", NodeKind::Trigger, Value::Null), + node("check", NodeKind::Condition, json!({ "field": "=false" })), + node( + "taken", + NodeKind::Transform, + json!({ "set": { "tag": "yes" } }), + ), + node("sink", NodeKind::Void, Value::Null), + ], + edges: vec![ + edge("t", "check"), + port_edge("check", "true", "sink"), + port_edge("check", "false", "taken"), + ], + ..Default::default() + }; + + let out = run_graph(&graph, json!({ "seed": 1 })).await; + + assert_eq!(out["nodes"]["taken"]["items"][0]["json"]["tag"], "yes"); + assert!( + out["nodes"]["sink"].is_null(), + "the void on the untaken branch never activated, so it has no slot" + ); +} + +#[tokio::test] +async fn void_arm_beside_a_merge_does_not_strand_the_barrier() { + // A void has no outgoing edge, so it can never be in anyone's `waiting` + // set. This pins that: `m` must still fire on both real predecessors even + // though a third branch off the same fan-out dead-ends in a void. + let graph = WorkflowGraph { + name: "void_merge".to_string(), + nodes: vec![ + node("t", NodeKind::Trigger, Value::Null), + node("fan", NodeKind::Transform, json!({ "set": { "n": 1 } })), + node("a", NodeKind::Transform, json!({ "set": { "arm": "a" } })), + node("b", NodeKind::Transform, json!({ "set": { "arm": "b" } })), + node("sink", NodeKind::Void, Value::Null), + node("m", NodeKind::Merge, Value::Null), + ], + edges: vec![ + edge("t", "fan"), + edge("fan", "a"), + edge("fan", "b"), + edge("fan", "sink"), + edge("a", "m"), + edge("b", "m"), + ], + ..Default::default() + }; + + let out = run_graph(&graph, json!({ "seed": 1 })).await; + + let merged = out["nodes"]["m"]["items"] + .as_array() + .expect("the merge should have released"); + let arms: Vec<&str> = merged + .iter() + .filter_map(|i| i["json"]["arm"].as_str()) + .collect(); + assert!( + arms.contains(&"a") && arms.contains(&"b"), + "the merge must see both real predecessors, got {arms:?}" + ); + assert_eq!(out["nodes"]["sink"]["discarded"], 1); +} + +#[tokio::test] +async fn loop_body_with_a_void_side_branch_runs_every_iteration() { + // The motivating case: a fire-and-forget side effect hanging off a loop + // body must not gate the loop. `work` closes the back-edge; `notify -> sink` + // is the detached arm. If the void participated in re-entry at all, this + // either hangs (caught by GUARD) or stops short of its iteration bound. + let graph = WorkflowGraph { + name: "void_loop".to_string(), + nodes: vec![ + node( + "t", + NodeKind::Trigger, + json!({ "recursion_limit": 200, "max_node_visits": 200 }), + ), + node( + "head", + NodeKind::Loop, + json!({ "max_iterations": 3, "on_exceeded": "continue" }), + ), + node( + "work", + NodeKind::Transform, + json!({ "set": { "worked": true } }), + ), + node( + "notify", + NodeKind::Transform, + json!({ "set": { "notified": true } }), + ), + node("sink", NodeKind::Void, Value::Null), + node( + "report", + NodeKind::Transform, + json!({ "set": { "done": true } }), + ), + ], + edges: vec![ + edge("t", "head"), + port_edge("head", "body", "work"), + port_edge("head", "body", "notify"), + edge("notify", "sink"), + edge("work", "head"), + port_edge("head", "done", "report"), + ], + ..Default::default() + }; + + let out = run_graph(&graph, json!({ "seed": 1 })).await; + + assert_eq!( + out["nodes"]["head"]["iteration"], 3, + "the loop must reach its bound with the void arm attached" + ); + assert_eq!(out["nodes"]["report"]["items"][0]["json"]["done"], true); + assert!( + !out["nodes"]["sink"].is_null(), + "the side branch must actually have run" + ); + assert_eq!( + out["nodes"]["sink"]["items"], + json!([]), + "and it must still have emitted nothing" + ); +} + +#[tokio::test] +async fn spawn_into_void_completes_without_a_gate() { + // `spawn -> void` is the explicit spelling of a ticket nothing will collect. + // It must behave exactly like leaving the spawn unwired: the run completes, + // and nothing waits on the task. + let graph = WorkflowGraph { + name: "void_spawn".to_string(), + nodes: vec![ + node("t", NodeKind::Trigger, Value::Null), + node( + "kick", + NodeKind::Spawn, + json!({ "target": "tool", "slug": "demo" }), + ), + node("sink", NodeKind::Void, Value::Null), + ], + edges: vec![edge("t", "kick"), edge("kick", "sink")], + ..Default::default() + }; + + let out = run_graph(&graph, json!({ "seed": 1 })).await; + + let tickets = out["nodes"]["kick"]["items"] + .as_array() + .expect("spawn should emit a ticket"); + assert_eq!(tickets.len(), 1, "one ticket per started task"); + assert_eq!(out["nodes"]["sink"]["discarded"], 1); +} + +#[tokio::test] +async fn scatter_lane_with_a_void_side_branch_gathers_all_lanes() { + // A void is the one dead end a lane may have. Every lane must still reach + // the gather, and the void's per-lane slot must land under `lanes` rather + // than at the top level. + let graph = WorkflowGraph { + name: "void_scatter".to_string(), + nodes: vec![ + node( + "t", + NodeKind::Trigger, + json!({ "recursion_limit": 400, "max_node_visits": 300 }), + ), + node("fan", NodeKind::Scatter, json!({ "path": "rows" })), + node( + "work", + NodeKind::Transform, + json!({ "set": { "seen": "=item.v" } }), + ), + node( + "notify", + NodeKind::Transform, + json!({ "set": { "notified": true } }), + ), + node("sink", NodeKind::Void, Value::Null), + node( + "collect", + NodeKind::Gather, + json!({ "from": ["work"], "poll_interval_ms": 1 }), + ), + ], + edges: vec![ + edge("t", "fan"), + edge("fan", "work"), + edge("work", "collect"), + edge("work", "notify"), + edge("notify", "sink"), + ], + ..Default::default() + }; + + let out = run_graph( + &graph, + json!({ "rows": [{ "v": 1 }, { "v": 2 }, { "v": 3 }] }), + ) + .await; + + let gathered = out["nodes"]["collect"]["items"] + .as_array() + .expect("the gather should have released"); + assert_eq!(gathered.len(), 3, "every lane must still be collected"); + + let lanes = out["nodes"]["sink"]["lanes"] + .as_object() + .expect("a lane activation writes under `lanes`, not the top-level slot"); + assert_eq!(lanes.len(), 3, "the void ran once per lane"); + for (lane, slot) in lanes { + assert_eq!(slot["items"], json!([]), "lane {lane} emitted nothing"); + } +} diff --git a/wiki/Node-Catalog.md b/wiki/Node-Catalog.md index e8307a46..09a51509 100644 --- a/wiki/Node-Catalog.md +++ b/wiki/Node-Catalog.md @@ -32,6 +32,36 @@ Native routing logic — no host capabilities required. | `merge` | Fan-in barrier combining multiple inputs | Waits for all wired inputs; config `mode` (e.g. `append`) | | `split_out` | Fan-out: one item per element of a list | Downstream runs per item; config `path` | | `transform` | Pure, expression-based field mapping | Config `set` (field → `=`-expression map) | +| `void` | Terminal sink: discards its input, runs nothing downstream | In `main`, no out ports; no config | + +### Fire-and-forget with `void` + +A branch could always dead-end — a node with no outgoing edges terminates — but +an unwired port reads exactly like a forgotten one. `void` is how you *say* that +a branch is a side effect nothing waits on, so validation and the next reader +can both tell intent from an accident. + +It adds no concurrency: work upstream of a `void` still runs inline in its own +super-step, and only the result is dropped. For work that should genuinely +overlap, use `spawn` and a `TaskRunner`. + +Three places it earns its keep: + +- **`spawn → void`** — the explicit spelling of a ticket no `gate` will collect. + Same abandon semantics as leaving the spawn unwired, said out loud. +- **A loop-body side branch** — the void arm never joins the back-edge, so it + cannot gate an iteration. +- **Inside a scatter lane** — it is the *one* dead end a lane may have. Every + other lane branch must reach the `gather`, because a stranded lane's output is + invisible rather than merely uncollected; a `void` makes that invisibility the + contract instead of the accident. + +Validation refuses a `void` with any outgoing edge (including an `error` edge +from `on_error: "route"`), and one with no incoming edge. Its slot is +`{items: [], port: null, discarded: N}` — a node that never ran has no slot at +all, so "never activated", "activated with nothing to drop" and "dropped N" stay +distinguishable. `discarded` counts *that* activation, so in a loop the last +iteration's value survives, and in a lane it lands under `lanes.`. ## Capability-backed nodes