diff --git a/src/flowrep/base_models.py b/src/flowrep/base_models.py
index ff43130b..f833d97c 100644
--- a/src/flowrep/base_models.py
+++ b/src/flowrep/base_models.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import abc
 import inspect
 import keyword
 from collections.abc import Hashable
@@ -94,7 +95,7 @@ def ensure_class_signature_from_init(cls: type, context: str) -> None:
 Labels = UniqueList[Label]
 
 
-class NodeRecipe(pydantic.BaseModel):
+class NodeRecipe(pydantic.BaseModel, abc.ABC):
     type: RecipeElementType
     inputs: Labels
     outputs: Labels
@@ -131,6 +132,7 @@ def _check_inputs_with_defaults_subset_of_inputs(self) -> Self:
     def validate_internal_data_completeness(self):
         return self
 
+    @abc.abstractmethod
     def __call__(self, *args, **kwargs):
         raise NotImplementedError(
             f"{self.__class__.__name__} is not a callable recipe type"
diff --git a/src/flowrep/prospective/for_recipe.py b/src/flowrep/prospective/for_recipe.py
index 6be6c24e..277bad0e 100644
--- a/src/flowrep/prospective/for_recipe.py
+++ b/src/flowrep/prospective/for_recipe.py
@@ -179,3 +179,9 @@ def validate_internal_data_completeness(self):
             self.prospective_nodes, self.input_edges
         )
         return self
+
+    def __call__(self, *args, **kwargs):
+        from flowrep import wfms
+
+        data = wfms._run_for(self, **wfms.variadic_to_inputs(self, *args, **kwargs))
+        return wfms.data_to_return(data)
diff --git a/src/flowrep/prospective/if_recipe.py b/src/flowrep/prospective/if_recipe.py
index a56070cc..e42f21c6 100644
--- a/src/flowrep/prospective/if_recipe.py
+++ b/src/flowrep/prospective/if_recipe.py
@@ -116,3 +116,9 @@ def validate_internal_data_completeness(self):
             self.prospective_nodes, self.input_edges
         )
         return self
+
+    def __call__(self, *args, **kwargs):
+        from flowrep import wfms
+
+        data = wfms._run_if(self, **wfms.variadic_to_inputs(self, *args, **kwargs))
+        return wfms.data_to_return(data)
diff --git a/src/flowrep/prospective/try_recipe.py b/src/flowrep/prospective/try_recipe.py
index 7988d791..ac6298f1 100644
--- a/src/flowrep/prospective/try_recipe.py
+++ b/src/flowrep/prospective/try_recipe.py
@@ -103,3 +103,9 @@ def validate_internal_data_completeness(self):
             self.prospective_nodes, self.input_edges
         )
         return self
+
+    def __call__(self, *args, **kwargs):
+        from flowrep import wfms
+
+        data = wfms._run_try(self, **wfms.variadic_to_inputs(self, *args, **kwargs))
+        return wfms.data_to_return(data)
diff --git a/src/flowrep/prospective/while_recipe.py b/src/flowrep/prospective/while_recipe.py
index 8717604b..15a01195 100644
--- a/src/flowrep/prospective/while_recipe.py
+++ b/src/flowrep/prospective/while_recipe.py
@@ -143,3 +143,9 @@ def validate_internal_data_completeness(self):
             self.prospective_nodes, self.input_edges
         )
         return self
+
+    def __call__(self, *args, **kwargs):
+        from flowrep import wfms
+
+        data = wfms._run_while(self, **wfms.variadic_to_inputs(self, *args, **kwargs))
+        return wfms.data_to_return(data)
diff --git a/src/flowrep/prospective/workflow_recipe.py b/src/flowrep/prospective/workflow_recipe.py
index e1a58549..7eacd5f2 100644
--- a/src/flowrep/prospective/workflow_recipe.py
+++ b/src/flowrep/prospective/workflow_recipe.py
@@ -87,10 +87,12 @@ def validate_internal_data_completeness(self):
 
     def __call__(self, *args, **kwargs):
         if self.reference is None:
-            raise ValueError(
-                f"{self.__class__.__name__} recipes are only callable when they are "
-                f"attached to an underlying python definiton in their reference field."
+            from flowrep import wfms
+
+            data = wfms._run_workflow(
+                self, **wfms.variadic_to_inputs(self, *args, **kwargs)
             )
+            return wfms.data_to_return(data)
         func = retrieve.import_from_string(self.reference.info.fully_qualified_name)
         return func(*args, **kwargs)
 
diff --git a/src/flowrep/wfms.py b/src/flowrep/wfms.py
index 6ebaf90d..882b0dc4 100644
--- a/src/flowrep/wfms.py
+++ b/src/flowrep/wfms.py
@@ -30,6 +30,10 @@
 from flowrep.retrospective import datastructures
 
 
+def _unsupported_recipe(recipe: Any) -> TypeError:
+    return TypeError(f"Unsupported recipe type: {type(recipe).__name__}")
+
+
 def run_recipe(
     recipe: union_types.RecipeDiscrimination, **kwargs: Any
 ) -> datastructures.NodeData:
@@ -37,7 +41,12 @@ def run_recipe(
     Execute a flowrep recipe, returning a populated :class:`LiveNode`.
 
     All inputs are passed as keyword arguments matching the recipe's input port names.
+    Inputs backed by a python default may be omitted; anything else must be supplied.
     """
+    if not isinstance(recipe, base_models.NodeRecipe):
+        # Guard before binding, which needs the recipe's input labels
+        raise _unsupported_recipe(recipe)
+    kwargs = variadic_to_inputs(recipe, **kwargs)
     match recipe:
         case atomic_recipe.AtomicRecipe():
             return _run_atomic(recipe, **kwargs)
@@ -54,7 +63,7 @@ def run_recipe(
         case while_recipe.WhileRecipe():
             return _run_while(recipe, **kwargs)
         case _:
-            raise TypeError(f"Unsupported recipe type: {type(recipe).__name__}")
+            raise _unsupported_recipe(recipe)
 
 
 # ---------------------------------------------------------------------------
@@ -205,6 +214,25 @@ def _populate_workflow_outputs(
 # ---------------------------------------------------------------------------
 
 
+def _iterated_value(
+    node: datastructures.ForEachData, for_port: str, body_port: str
+) -> Collection:
+    """
+    The value to scatter over one iterated port.
+
+    A for-node cannot fall back on a default here the way an atomic child can -- there
+    is nothing to iterate -- so an unfilled port is reported directly, rather than
+    leaking a bare ``NotData`` into ``itertools.product``.
+    """
+    value = node.input_ports[for_port].value
+    if isinstance(value, datastructures.NotData):
+        raise ValueError(
+            f"Iterated input '{for_port}' (body port '{body_port}') has no value to "
+            f"iterate over"
+        )
+    return cast(Collection, value)
+
+
 def _run_for(
     recipe: for_recipe.ForEachRecipe, **kwargs: Any
 ) -> datastructures.ForEachData:
@@ -244,15 +272,14 @@ def _run_for(
 
     # Build iteration axes
     nested_iters = [
-        cast(Collection, node.input_ports[body_to_for[p]].value)
-        for p in recipe.nested_ports
+        _iterated_value(node, body_to_for[p], p) for p in recipe.nested_ports
     ]
     zipped_iters = [
-        cast(Collection, node.input_ports[body_to_for[p]].value)
-        for p in recipe.zipped_ports
+        _iterated_value(node, body_to_for[p], p) for p in recipe.zipped_ports
     ]
-    # Note that we simply cast iterated input values to the form we expect, and let the
-    # user pay the price if runtime data is non-compliant.
+    # Note that beyond insisting the value arrived at all, we simply cast iterated
+    # input values to the form we expect, and let the user pay the price if runtime
+    # data is non-compliant.
 
     nested_combos = list(itertools.product(*nested_iters)) if nested_iters else [()]
     if zipped_iters:
@@ -504,3 +531,65 @@ def _populate_prospective_outputs(
                     source.port
                 ].value
                 break
+
+
+def variadic_to_inputs(recipe: base_models.NodeRecipe, /, *args, **kwargs):
+    """
+    Bind ``*args`` and ``**kwargs`` onto ``recipe.inputs``, as a helper for
+    ``NodeRecipe.__call__`` implementations and the generic recipe runner.
+
+    Every input must be filled except those in ``recipe.inputs_with_defaults``, which
+    only recipes backed by an underlying python function have any of. Nothing else can
+    supply a value after the fact, so an unfilled input is simply a missing one --
+    which is exactly the invariant
+    :func:`subgraph_validation.validate_nodes_are_fully_sourced` already holds children
+    to, so validated recipes bind their own children by construction.
+
+    Binding failures raise :class:`TypeError`, mirroring python's own behaviour for
+    bad call signatures. (Deliberately not :class:`ValueError`: recipes catch
+    exceptions by type, and a try-recipe handling ``ValueError`` must not be able to
+    swallow its caller's mistake and quietly return partial data.)
+    """
+    who = f"{type(recipe).__name__}()"
+    if len(args) > len(recipe.inputs):
+        raise TypeError(
+            f"One of your {who} calls takes {len(recipe.inputs)} inputs but "
+            f"{len(args)} positional arguments were given -- its inputs are "
+            f"{recipe.inputs}"
+        )
+    inputs = {}
+    for label, val in zip(recipe.inputs, args, strict=False):
+        inputs[label] = val
+    for label, val in kwargs.items():
+        if label in inputs:
+            raise TypeError(
+                f"One of your {who} calls got multiple values for input '{label}' -- "
+                f"as a positional arg ({inputs[label]}) and as a kwarg ({val})"
+            )
+        if label in recipe.inputs:
+            inputs[label] = val
+        else:
+            raise TypeError(
+                f"One of your {who} calls got an unexpected input '{label}' -- its "
+                f"inputs are {recipe.inputs}"
+            )
+    missing = [
+        label
+        for label in recipe.inputs
+        if label not in inputs and label not in recipe.inputs_with_defaults
+    ]
+    if missing:
+        raise TypeError(
+            f"One of your {who} calls is missing {len(missing)} required "
+            f"input: {missing}"
+        )
+    return inputs
+
+
+def data_to_return(data: datastructures.NodeData):
+    """A helper for ``NodeRecipe.__call__`` implementations"""
+    returns = tuple(p.value for p in data.output_ports.values())
+    if len(returns) == 1:
+        return returns[0]
+    else:
+        return returns
diff --git a/tests/integration/parsers/test_parsing_composite_workflow.py b/tests/integration/parsers/test_parsing_composite_workflow.py
index d2d04eb0..f04f8cf5 100644
--- a/tests/integration/parsers/test_parsing_composite_workflow.py
+++ b/tests/integration/parsers/test_parsing_composite_workflow.py
@@ -1,12 +1,19 @@
+import dataclasses
 import inspect
 import unittest
 
 from pyiron_snippets import versions
 
-from flowrep import std
+from flowrep import std, wfms
 from flowrep.compiler import source
 from flowrep.parsers import atomic_parser, workflow_parser
-from flowrep.prospective import workflow_recipe
+from flowrep.prospective import (
+    for_recipe,
+    if_recipe,
+    try_recipe,
+    while_recipe,
+    workflow_recipe,
+)
 
 from flowrep_static import library, makers
 
@@ -264,11 +271,7 @@ def full_composite(x, /, y, *, bound):
         },
         "output_edges": {"result": "identity_0.x"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_composite_workflow",
-                "qualname": "full_composite",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(full_composite)),
             "inputs_with_defaults": [],
             "restricted_input_kinds": {
                 "x": "POSITIONAL_ONLY",
@@ -316,5 +319,111 @@ def test_roundtrip_back_to_python(self):
         )
 
 
+# =====================================================================
+# Calling the recipes instead of writing the python
+# =====================================================================
+
+# The static recipe above already spells out one recipe of every flow-control type,
+# nested inside each other. Pull them back out and call them directly.
+
+_try_flow = full_composite_node.nodes["try_0"]
+_while_flow = _try_flow.try_node.recipe.nodes["while_0"]
+_for_flow = _while_flow.case.body.recipe.nodes["for_each_0"]
+_for_body = _for_flow.body_node.recipe  # A reference-free workflow
+_if_flow = _for_body.nodes["if_0"]
+
+
+@workflow_parser.workflow
+def try_by_recipe_call(x, /, y, *, bound):
+    """:func:`full_composite` with its ``try`` block replaced by a recipe call."""
+    a = std.add(x, y)
+    b, z = _try_flow(a, y, bound)
+    result = std.identity(z)
+    return result
+
+
+@workflow_parser.workflow
+def all_flows_by_recipe_call(x, /, y, *, bound):
+    """One call to every recipe type, chained so that each feeds the next.
+
+    Not equivalent to any of the functions above -- the point is only that a single
+    workflow exercises all five ``__call__`` implementations at once.
+    """
+    a = std.add(x, y)
+    b, z = _try_flow(a, y, bound)
+    n = _while_flow(b, bound, y)
+    rs = library.my_range(y)
+    acc = _for_flow(y, n, rs)
+    s = my_sum(acc)
+    v = _if_flow(s, y, n)
+    w = _for_body(v, y, n)
+    result = std.add(z, w)
+    return result
+
+
+class TestCallingCompositeRecipes(unittest.TestCase):
+    """
+    Integration test that recipes are callable: a workflow whose nodes are invoked as
+    recipe objects must give the same answer run as plain python and run by the WfMS.
+    """
+
+    _CASES = [(1, 2, 10), (3, 1, 8)]
+
+    def _via_wfms(self, func, x, y, bound):
+        data = wfms.run_recipe(func.flowrep_recipe, x=x, y=y, bound=bound)
+        return data.output_ports["result"].value
+
+    def test_try_call_matches_python_syntax(self):
+        """Calling the try-recipe stands in for writing the ``try`` block by hand."""
+        for x, y, bound in self._CASES:
+            with self.subTest(x=x, y=y, bound=bound):
+                self.assertEqual(
+                    try_by_recipe_call(x, y, bound=bound),
+                    full_composite(x, y, bound=bound),
+                )
+
+    def test_try_call_matches_wfms(self):
+        for x, y, bound in self._CASES:
+            with self.subTest(x=x, y=y, bound=bound):
+                self.assertEqual(
+                    try_by_recipe_call(x, y, bound=bound),
+                    self._via_wfms(try_by_recipe_call, x, y, bound),
+                )
+
+    def test_all_flows_call_matches_wfms(self):
+        for x, y, bound in self._CASES:
+            with self.subTest(x=x, y=y, bound=bound):
+                self.assertEqual(
+                    all_flows_by_recipe_call(x, y, bound=bound),
+                    self._via_wfms(all_flows_by_recipe_call, x, y, bound),
+                )
+
+    def test_underfilled_call_is_not_swallowed_by_the_handler(self):
+        """``bound`` is consumed by the while-condition, three levels inside a try
+        that handles ValueError. A missing input must not be mistaken for a domain
+        error and quietly answered with the except branch."""
+        with self.assertRaises(TypeError) as ctx:
+            _try_flow(3, 2)
+        self.assertEqual(
+            str(ctx.exception),
+            "One of your TryRecipe() calls is missing 1 required input: ['bound']",
+        )
+
+    def test_every_flow_control_type_is_called(self):
+        """Guard the premise of :func:`all_flows_by_recipe_call`: if a refactor of the
+        static recipe above changes what gets pulled out, the test above could quietly
+        stop covering a recipe type."""
+        self.assertIsInstance(_try_flow, try_recipe.TryRecipe)
+        self.assertIsInstance(_while_flow, while_recipe.WhileRecipe)
+        self.assertIsInstance(_for_flow, for_recipe.ForEachRecipe)
+        self.assertIsInstance(_if_flow, if_recipe.IfRecipe)
+        self.assertIsInstance(_for_body, workflow_recipe.WorkflowRecipe)
+        self.assertIsNone(
+            _for_body.reference,
+            msg="A referenced workflow would defer to its python function instead of "
+            "running its own graph",
+        )
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/tests/integration/parsers/test_parsing_for_nodes.py b/tests/integration/parsers/test_parsing_for_nodes.py
index 5388d78e..ece4a652 100644
--- a/tests/integration/parsers/test_parsing_for_nodes.py
+++ b/tests/integration/parsers/test_parsing_for_nodes.py
@@ -1,6 +1,9 @@
+import dataclasses
 import inspect
 import unittest
 
+from pyiron_snippets import versions
+
 from flowrep import std
 from flowrep.parsers import atomic_parser, workflow_parser
 from flowrep.prospective import for_recipe, workflow_recipe
@@ -77,11 +80,7 @@ def single_iteration(ns):
             "vecs": "for_each_0.vecs",
         },
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_for_nodes",
-                "qualname": "single_iteration",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(single_iteration)),
             "inputs_with_defaults": [],
         },
     }
@@ -176,11 +175,9 @@ def zipped_broadcast_and_transferred(a, bs, cs, ds):
             "sums": "for_each_0.sums",
         },
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_for_nodes",
-                "qualname": "zipped_broadcast_and_transferred",
-                "version": None,
-            },
+            "info": dataclasses.asdict(
+                versions.VersionInfo.of(zipped_broadcast_and_transferred)
+            ),
             "inputs_with_defaults": [],
         },
     }
@@ -284,11 +281,7 @@ def nested(ns):
         "edges": {},
         "output_edges": {"sq_sums": "for_each_0.sq_sums"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_for_nodes",
-                "qualname": "nested",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(nested)),
             "inputs_with_defaults": [],
         },
     }
@@ -410,11 +403,9 @@ def nested_with_passed_input(ns, range_offset, square_offset):
         "edges": {},
         "output_edges": {"sq_sums": "for_each_0.sq_sums"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_for_nodes",
-                "qualname": "nested_with_passed_input",
-                "version": None,
-            },
+            "info": dataclasses.asdict(
+                versions.VersionInfo.of(nested_with_passed_input)
+            ),
             "inputs_with_defaults": [],
         },
     }
diff --git a/tests/integration/parsers/test_parsing_if_nodes.py b/tests/integration/parsers/test_parsing_if_nodes.py
index 1d9f3b58..857de5dd 100644
--- a/tests/integration/parsers/test_parsing_if_nodes.py
+++ b/tests/integration/parsers/test_parsing_if_nodes.py
@@ -1,5 +1,8 @@
+import dataclasses
 import unittest
 
+from pyiron_snippets import versions
+
 from flowrep import std
 from flowrep.parsers import workflow_parser
 from flowrep.prospective import workflow_recipe
@@ -78,11 +81,7 @@ def simple_if_else(x, y):
         "edges": {},
         "output_edges": {"z": "if_0.z"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_if_nodes",
-                "qualname": "simple_if_else",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(simple_if_else)),
             "inputs_with_defaults": [],
         },
     }
@@ -189,11 +188,7 @@ def if_elif_else(x, y, flag):
         "edges": {},
         "output_edges": {"z": "if_0.z"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_if_nodes",
-                "qualname": "if_elif_else",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(if_elif_else)),
             "inputs_with_defaults": [],
         },
     }
@@ -275,11 +270,7 @@ def if_with_context(a, b):
         "edges": {"if_0.x": "add_0.added", "identity_0.x": "if_0.y"},
         "output_edges": {"z": "identity_0.x"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_if_nodes",
-                "qualname": "if_with_context",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(if_with_context)),
             "inputs_with_defaults": [],
         },
     }
@@ -380,11 +371,7 @@ def multi_output_if(x, y):
         "edges": {},
         "output_edges": {"a": "if_0.a", "b": "if_0.b"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_if_nodes",
-                "qualname": "multi_output_if",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(multi_output_if)),
             "inputs_with_defaults": [],
         },
     }
@@ -448,11 +435,7 @@ def if_no_else(x, y):
         "edges": {},
         "output_edges": {"z": "if_0.z"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_if_nodes",
-                "qualname": "if_no_else",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(if_no_else)),
             "inputs_with_defaults": [],
         },
     }
diff --git a/tests/integration/parsers/test_parsing_try_nodes.py b/tests/integration/parsers/test_parsing_try_nodes.py
index ed85d12f..0f731de4 100644
--- a/tests/integration/parsers/test_parsing_try_nodes.py
+++ b/tests/integration/parsers/test_parsing_try_nodes.py
@@ -1,3 +1,4 @@
+import dataclasses
 import unittest
 
 from pyiron_snippets import versions
@@ -80,11 +81,7 @@ def simple_try_except(x, y):
         "edges": {},
         "output_edges": {"z": "try_0.z"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_try_nodes",
-                "qualname": "simple_try_except",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(simple_try_except)),
             "inputs_with_defaults": [],
         },
     }
@@ -185,11 +182,7 @@ def try_multi_except(x, y):
         "edges": {},
         "output_edges": {"z": "try_0.z"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_try_nodes",
-                "qualname": "try_multi_except",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(try_multi_except)),
             "inputs_with_defaults": [],
         },
     }
@@ -273,11 +266,7 @@ def try_with_context(a, b):
         },
         "output_edges": {"z": "identity_0.x"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_try_nodes",
-                "qualname": "try_with_context",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(try_with_context)),
             "inputs_with_defaults": [],
         },
     }
@@ -372,11 +361,7 @@ def multi_output_try(x, y):
         "edges": {},
         "output_edges": {"a": "try_0.a", "b": "try_0.b"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_try_nodes",
-                "qualname": "multi_output_try",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(multi_output_try)),
             "inputs_with_defaults": [],
         },
     }
@@ -456,11 +441,7 @@ def try_tuple_exceptions(x, y):
         "edges": {},
         "output_edges": {"z": "try_0.z"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_try_nodes",
-                "qualname": "try_tuple_exceptions",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(try_tuple_exceptions)),
             "inputs_with_defaults": [],
         },
     }
diff --git a/tests/integration/parsers/test_parsing_while_nodes.py b/tests/integration/parsers/test_parsing_while_nodes.py
index 8c303ed6..cea8d116 100644
--- a/tests/integration/parsers/test_parsing_while_nodes.py
+++ b/tests/integration/parsers/test_parsing_while_nodes.py
@@ -1,5 +1,8 @@
+import dataclasses
 import unittest
 
+from pyiron_snippets import versions
+
 from flowrep import std
 from flowrep.parsers import workflow_parser
 from flowrep.prospective import while_recipe, workflow_recipe
@@ -75,11 +78,7 @@ def simple_while(a=10, b=20, c=40):
         },
         "output_edges": {"y": "identity_1.x"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_while_nodes",
-                "qualname": "simple_while",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(simple_while)),
             "inputs_with_defaults": ["a", "b", "c"],
         },
     }
@@ -191,11 +190,7 @@ def nested_while(x, m, n, a):
         "edges": {"while_0.y": "identity_0.x"},
         "output_edges": {"x": "while_0.x", "y": "while_0.y"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_while_nodes",
-                "qualname": "nested_while",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(nested_while)),
             "inputs_with_defaults": [],
         },
     }
@@ -276,11 +271,7 @@ def multi_reassign(x, y, bound):
             "y": "while_0.y",
         },
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_while_nodes",
-                "qualname": "multi_reassign",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(multi_reassign)),
             "inputs_with_defaults": [],
         },
     }
@@ -368,11 +359,7 @@ def sequential_whiles(x, y, m, n):
             "x": "while_1.x",
         },
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_while_nodes",
-                "qualname": "sequential_whiles",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(sequential_whiles)),
             "inputs_with_defaults": [],
         },
     }
@@ -444,11 +431,7 @@ def chained_body(x, a, b, bound):
         "edges": {},
         "output_edges": {"x": "while_0.x"},
         "reference": {
-            "info": {
-                "module": "integration.parsers.test_parsing_while_nodes",
-                "qualname": "chained_body",
-                "version": None,
-            },
+            "info": dataclasses.asdict(versions.VersionInfo.of(chained_body)),
             "inputs_with_defaults": [],
         },
     }
diff --git a/tests/unit/prospective/test_for_recipe.py b/tests/unit/prospective/test_for_recipe.py
index f3a41728..61695928 100644
--- a/tests/unit/prospective/test_for_recipe.py
+++ b/tests/unit/prospective/test_for_recipe.py
@@ -2,7 +2,7 @@
 
 import pydantic
 
-from flowrep import base_models, edge_models, subgraph_validation
+from flowrep import base_models, edge_models, std, subgraph_validation
 from flowrep.prospective import (
     atomic_recipe,
     for_recipe,
@@ -13,6 +13,28 @@
 from flowrep_static import makers
 
 
+def _make_runnable_for_node() -> for_recipe.ForEachRecipe:
+    """Negate each element of ``xs``, collecting the results into ``ys``."""
+    return for_recipe.ForEachRecipe(
+        inputs=["xs"],
+        outputs=["ys"],
+        body_node=helper_models.LabeledRecipe(
+            label="body", recipe=std.neg.flowrep_recipe
+        ),
+        input_edges={
+            edge_models.TargetHandle(node="body", port="a"): edge_models.InputSource(
+                port="xs"
+            ),
+        },
+        output_edges={
+            edge_models.OutputTarget(port="ys"): edge_models.SourceHandle(
+                node="body", port="negative"
+            ),
+        },
+        nested_ports=["a"],
+    )
+
+
 class TestForEachRecipeBasic(unittest.TestCase):
     def test_schema_generation(self):
         """model_json_schema() fails if forward refs aren't resolved."""
@@ -118,22 +140,14 @@ def test_valid_for_node_with_both_nested_and_zipped(self):
         self.assertEqual(for_node.nested_ports, ["a"])
         self.assertEqual(for_node.zipped_ports, ["b", "c"])
 
-    def test_call_raises(self):
-        recipe = for_recipe.ForEachRecipe(
-            inputs=["x"],
-            outputs=[],
-            body_node=makers.make_labeled_atomic(
-                "body",
-                inputs=["item"],
-                outputs=["result"],
-                inputs_with_defaults=["item"],
-            ),
-            input_edges={},
-            output_edges={},
-            nested_ports=["item"],
-        )
-        with self.assertRaises(NotImplementedError):
-            recipe(42)
+    def test_call(self):
+        """Calling a for-recipe maps its body over the iterated input."""
+        recipe = _make_runnable_for_node()
+        self.assertListEqual(recipe([1, 2, 3]), [-1, -2, -3])
+
+    def test_call_with_keywords(self):
+        recipe = _make_runnable_for_node()
+        self.assertListEqual(recipe(xs=[1, 2, 3]), [-1, -2, -3])
 
 
 class TestForEachRecipeLoopPortValidation(unittest.TestCase):
diff --git a/tests/unit/prospective/test_if_recipe.py b/tests/unit/prospective/test_if_recipe.py
index 834dd0a8..ec11afa3 100644
--- a/tests/unit/prospective/test_if_recipe.py
+++ b/tests/unit/prospective/test_if_recipe.py
@@ -2,7 +2,7 @@
 
 import pydantic
 
-from flowrep import base_models, edge_models, subgraph_validation
+from flowrep import base_models, edge_models, std, subgraph_validation
 from flowrep.prospective import (
     atomic_recipe,
     helper_models,
@@ -10,7 +10,7 @@
     workflow_recipe,
 )
 
-from flowrep_static import makers
+from flowrep_static import library, makers
 
 
 def _make_conditional_cases(n: int) -> list[helper_models.ConditionalCase]:
@@ -66,6 +66,44 @@ def _make_valid_if_node(n_cases=1, with_else=True):
     )
 
 
+def _make_runnable_if_node() -> if_recipe.IfRecipe:
+    """Return ``x`` if it is positive, else its negation -- i.e. ``abs``."""
+    return if_recipe.IfRecipe(
+        inputs=["x"],
+        outputs=["y"],
+        cases=[
+            helper_models.ConditionalCase(
+                condition=helper_models.LabeledRecipe(
+                    label="condition_0", recipe=library.is_positive.flowrep_recipe
+                ),
+                body=helper_models.LabeledRecipe(
+                    label="body_0", recipe=std.identity.flowrep_recipe
+                ),
+            )
+        ],
+        else_case=helper_models.LabeledRecipe(
+            label="else_body", recipe=std.neg.flowrep_recipe
+        ),
+        input_edges={
+            edge_models.TargetHandle(
+                node="condition_0", port="n"
+            ): edge_models.InputSource(port="x"),
+            edge_models.TargetHandle(node="body_0", port="x"): edge_models.InputSource(
+                port="x"
+            ),
+            edge_models.TargetHandle(
+                node="else_body", port="a"
+            ): edge_models.InputSource(port="x"),
+        },
+        prospective_output_edges={
+            edge_models.OutputTarget(port="y"): [
+                edge_models.SourceHandle(node="body_0", port="x"),
+                edge_models.SourceHandle(node="else_body", port="negative"),
+            ]
+        },
+    )
+
+
 class TestIfRecipeBasic(unittest.TestCase):
     def test_schema_generation(self):
         """model_json_schema() fails if forward refs aren't resolved."""
@@ -100,10 +138,18 @@ def test_type_field_immutable(self):
             node.type = base_models.RecipeElementType.WORKFLOW
         self.assertIn("frozen", str(ctx.exception).lower())
 
-    def test_call_raises(self):
-        recipe = _make_valid_if_node()
-        with self.assertRaises(NotImplementedError):
-            recipe(42)
+    def test_call(self):
+        """Calling an if-recipe runs the branch its condition selects."""
+        recipe = _make_runnable_if_node()
+        self.assertEqual(recipe(5), 5)
+
+    def test_call_with_keywords(self):
+        recipe = _make_runnable_if_node()
+        self.assertEqual(recipe(x=5), 5)
+
+    def test_call_else_branch(self):
+        recipe = _make_runnable_if_node()
+        self.assertEqual(recipe(-3), 3)
 
 
 class TestIfRecipeCasesValidation(unittest.TestCase):
diff --git a/tests/unit/prospective/test_try_recipe.py b/tests/unit/prospective/test_try_recipe.py
index 21cadb29..b3182f8e 100644
--- a/tests/unit/prospective/test_try_recipe.py
+++ b/tests/unit/prospective/test_try_recipe.py
@@ -3,7 +3,7 @@
 import pydantic
 from pyiron_snippets import versions
 
-from flowrep import base_models, edge_models, subgraph_validation
+from flowrep import base_models, edge_models, std, subgraph_validation
 from flowrep.prospective import (
     atomic_recipe,
     helper_models,
@@ -65,6 +65,42 @@ def _make_valid_try_node(n_exception_cases=1):
     )
 
 
+def _make_runnable_try_node() -> try_recipe.TryRecipe:
+    """Divide ``a`` by ``b``, falling back to ``a`` on a zero denominator."""
+    return try_recipe.TryRecipe(
+        inputs=["a", "b"],
+        outputs=["result"],
+        try_node=helper_models.LabeledRecipe(
+            label="try_body", recipe=std.truediv.flowrep_recipe
+        ),
+        exception_cases=[
+            helper_models.ExceptionCase(
+                exceptions=[versions.VersionInfo.of(ZeroDivisionError)],
+                body=helper_models.LabeledRecipe(
+                    label="except_body_0", recipe=std.identity.flowrep_recipe
+                ),
+            )
+        ],
+        input_edges={
+            edge_models.TargetHandle(
+                node="try_body", port="a"
+            ): edge_models.InputSource(port="a"),
+            edge_models.TargetHandle(
+                node="try_body", port="b"
+            ): edge_models.InputSource(port="b"),
+            edge_models.TargetHandle(
+                node="except_body_0", port="x"
+            ): edge_models.InputSource(port="a"),
+        },
+        prospective_output_edges={
+            edge_models.OutputTarget(port="result"): [
+                edge_models.SourceHandle(node="try_body", port="quotient"),
+                edge_models.SourceHandle(node="except_body_0", port="x"),
+            ]
+        },
+    )
+
+
 class TestTryRecipeBasic(unittest.TestCase):
     def test_schema_generation(self):
         """model_json_schema() fails if forward refs aren't resolved."""
@@ -86,10 +122,22 @@ def test_valid_multiple_exception_cases(self):
         node = _make_valid_try_node(n_exception_cases=3)
         self.assertEqual(len(node.exception_cases), 3)
 
-    def test_call_raises(self):
-        recipe = _make_valid_try_node()
-        with self.assertRaises(NotImplementedError):
-            recipe(42)
+    def test_call(self):
+        """Calling a try-recipe runs its try body when nothing goes wrong."""
+        recipe = _make_runnable_try_node()
+        self.assertEqual(recipe(10, 2), 5.0)
+
+    def test_call_with_keywords(self):
+        recipe = _make_runnable_try_node()
+        self.assertEqual(recipe(10, b=2), 5.0)
+
+    def test_call_exception_case(self):
+        recipe = _make_runnable_try_node()
+        self.assertEqual(
+            recipe(7, 0),
+            7,
+            msg="A ZeroDivisionError should be caught and handled by the except body",
+        )
 
 
 class TestTryRecipeExceptionCasesValidation(unittest.TestCase):
diff --git a/tests/unit/prospective/test_while_recipe.py b/tests/unit/prospective/test_while_recipe.py
index 0b9f8100..516e8a61 100644
--- a/tests/unit/prospective/test_while_recipe.py
+++ b/tests/unit/prospective/test_while_recipe.py
@@ -11,7 +11,36 @@
     workflow_recipe,
 )
 
-from flowrep_static import makers
+from flowrep_static import library, makers
+
+
+def _make_runnable_while_node() -> while_recipe.WhileRecipe:
+    """Decrement ``n`` for as long as it is positive."""
+    return while_recipe.WhileRecipe(
+        inputs=["n"],
+        outputs=["n"],
+        case=helper_models.ConditionalCase(
+            condition=helper_models.LabeledRecipe(
+                label="condition", recipe=library.is_positive.flowrep_recipe
+            ),
+            body=helper_models.LabeledRecipe(
+                label="body", recipe=library.decrement.flowrep_recipe
+            ),
+        ),
+        input_edges={
+            edge_models.TargetHandle(
+                node="condition", port="n"
+            ): edge_models.InputSource(port="n"),
+            edge_models.TargetHandle(node="body", port="x"): edge_models.InputSource(
+                port="n"
+            ),
+        },
+        output_edges={
+            edge_models.OutputTarget(port="n"): edge_models.SourceHandle(
+                node="body", port="output_0"
+            ),
+        },
+    )
 
 
 def make_valid_while_node(
@@ -101,10 +130,22 @@ def test_valid_fully_wired(self):
         self.assertEqual(len(wn.input_edges), 3)
         self.assertEqual(len(wn.output_edges), 1)
 
-    def test_call_raises(self):
-        recipe = make_valid_while_node()
-        with self.assertRaises(NotImplementedError):
-            recipe(42)
+    def test_call(self):
+        """Calling a while-recipe iterates its body until the condition fails."""
+        recipe = _make_runnable_while_node()
+        self.assertEqual(recipe(3), 0)
+
+    def test_call_with_keywords(self):
+        recipe = _make_runnable_while_node()
+        self.assertEqual(recipe(n=3), 0)
+
+    def test_call_condition_false_immediately(self):
+        recipe = _make_runnable_while_node()
+        self.assertEqual(
+            recipe(0),
+            0,
+            msg="With the body never running, the input should pass straight through",
+        )
 
 
 class TestWhileRecipeIOValidation(unittest.TestCase):
diff --git a/tests/unit/prospective/test_workflow_recipe.py b/tests/unit/prospective/test_workflow_recipe.py
index 5cf0c45f..7c6e857b 100644
--- a/tests/unit/prospective/test_workflow_recipe.py
+++ b/tests/unit/prospective/test_workflow_recipe.py
@@ -962,7 +962,18 @@ def test_not_subset_rejected(self):
 
 
 class TestWorkflowRecipeCall(unittest.TestCase):
-    def test_call_without_reference_raises(self):
+    def test_call_without_reference(self):
+        """Without an underlying python function to defer to, the recipe's own graph
+        gets executed."""
+        recipe = makers.make_simple_workflow_recipe()
+        self.assertIsNone(recipe.reference)
+        self.assertEqual(recipe(1, 2), 3)
+
+    def test_call_without_reference_with_keywords(self):
+        recipe = makers.make_simple_workflow_recipe()
+        self.assertEqual(recipe(1, b=2), 3)
+
+    def test_call_without_reference_or_nodes(self):
         recipe = workflow_recipe.WorkflowRecipe(
             inputs=[],
             outputs=[],
@@ -971,14 +982,12 @@ def test_call_without_reference_raises(self):
             edges={},
             output_edges={},
         )
-        with self.assertRaises(
-            ValueError,
-            msg="Calling a workflow recipe without a reference should alert us to "
-            "the reference's absence",
-        ) as ctx:
-            recipe()
-        self.assertIn("only callable when", str(ctx.exception))
-        self.assertIn("reference field", str(ctx.exception))
+        self.assertTupleEqual(
+            recipe(),
+            (),
+            msg="An empty graph has nothing to return, like a function with a bare "
+            "`return`",
+        )
 
     def test_call_with_reference(self):
         recipe = workflow_recipe.WorkflowRecipe(
diff --git a/tests/unit/test_base_models.py b/tests/unit/test_base_models.py
index 7783f7df..d66f0de8 100644
--- a/tests/unit/test_base_models.py
+++ b/tests/unit/test_base_models.py
@@ -17,6 +17,41 @@ class _ValidTestRecipe(base_models.NodeRecipe):
         default=base_models.RecipeElementType.ATOMIC, frozen=True
     )
 
+    def __call__(self, *args, **kwargs):
+        return {"args": args, "kwargs": kwargs}
+
+
+class TestNodeRecipeCall(unittest.TestCase):
+    """Tests for the abstract ``NodeRecipe.__call__``."""
+
+    def test_subclass_must_implement(self):
+        class _NoCall(base_models.NodeRecipe):
+            type: Literal[base_models.RecipeElementType.ATOMIC] = pydantic.Field(
+                default=base_models.RecipeElementType.ATOMIC, frozen=True
+            )
+
+        with self.assertRaises(TypeError) as ctx:
+            _NoCall(inputs=[], outputs=[])
+        self.assertIn("abstract", str(ctx.exception))
+        self.assertIn("__call__", str(ctx.exception))
+
+    def test_super_call_reports_uncallable(self):
+        """Subclasses that defer upwards get told the recipe type is not callable,
+        rather than something opaque about abstract methods."""
+
+        class _DefersUpwards(base_models.NodeRecipe):
+            type: Literal[base_models.RecipeElementType.ATOMIC] = pydantic.Field(
+                default=base_models.RecipeElementType.ATOMIC, frozen=True
+            )
+
+            def __call__(self, *args, **kwargs):
+                return super().__call__(*args, **kwargs)
+
+        with self.assertRaises(NotImplementedError) as ctx:
+            _DefersUpwards(inputs=[], outputs=[])()
+        self.assertIn("_DefersUpwards", str(ctx.exception))
+        self.assertIn("not a callable recipe type", str(ctx.exception))
+
 
 class TestLabelValidation(unittest.TestCase):
     """Tests for Label type alias and _validate_label."""
@@ -135,6 +170,9 @@ class _Valid(base_models.NodeRecipe):
                 default=base_models.RecipeElementType.WORKFLOW, frozen=True
             )
 
+            def __call__(self, *args, **kwargs):
+                return {"args": args, "kwargs": kwargs}
+
         node = _Valid(inputs=[], outputs=[])
         self.assertEqual(node.type, base_models.RecipeElementType.WORKFLOW)
 
@@ -294,11 +332,17 @@ class _TypeA(base_models.NodeRecipe):
                 default=base_models.RecipeElementType.ATOMIC, frozen=True
             )
 
+            def __call__(self, *args, **kwargs):
+                return {"args": args, "kwargs": kwargs}
+
         class _TypeB(base_models.NodeRecipe):
             type: Literal[base_models.RecipeElementType.WORKFLOW] = pydantic.Field(
                 default=base_models.RecipeElementType.WORKFLOW, frozen=True
             )
 
+            def __call__(self, *args, **kwargs):
+                return {"args": args, "kwargs": kwargs}
+
         a = _TypeA(inputs=[], outputs=[])
         b = _TypeB(inputs=[], outputs=[])
         self.assertEqual(a.type, base_models.RecipeElementType.ATOMIC)
diff --git a/tests/unit/test_retrospective_wfms.py b/tests/unit/test_retrospective_wfms.py
index f233c7b9..ee004e21 100644
--- a/tests/unit/test_retrospective_wfms.py
+++ b/tests/unit/test_retrospective_wfms.py
@@ -5,9 +5,10 @@
 import dataclasses
 import pickle
 import unittest
-from typing import TYPE_CHECKING, NamedTuple, get_origin
+from typing import TYPE_CHECKING, Literal, NamedTuple, get_origin
 from unittest import mock
 
+import pydantic
 from pyiron_snippets import versions
 
 from flowrep import base_models, edge_models, std, wfms
@@ -495,6 +496,17 @@ def _variadic_recipe(func, inputs, outputs=("result",)):
     )
 
 
+class _UnrunnableRecipe(base_models.NodeRecipe):
+    """A well-formed recipe of a type the WfMS has no runner for."""
+
+    type: Literal[base_models.RecipeElementType.ATOMIC] = pydantic.Field(
+        default=base_models.RecipeElementType.ATOMIC, frozen=True
+    )
+
+    def __call__(self, *args, **kwargs):
+        raise NotImplementedError()
+
+
 # ═══════════════════════════════════════════════════════════════════════════
 # datastructures.py tests
 # ═══════════════════════════════════════════════════════════════════════════
@@ -943,8 +955,36 @@ def test_multi_output(self):
         self.assertAlmostEqual(node.output_ports["remainder"].value, 2.0)
 
     def test_missing_input_raises(self):
-        with self.assertRaisesRegex(ValueError, "no value and no default"):
+        """``std.add`` has no defaults at all, so 'a' alone leaves 'b' unsourced."""
+        with self.assertRaises(TypeError) as ctx:
             wfms.run_recipe(std.add.flowrep_recipe, a=3)
+        self.assertEqual(
+            str(ctx.exception),
+            "One of your AtomicRecipe() calls is missing 1 required input: ['b']",
+        )
+
+    def test_not_data_input_raises(self):
+        """Binding insists an input be *given*, but a parent can legitimately hand
+        down a port it never set, so the runner still has to catch NOT_DATA."""
+        with self.assertRaises(ValueError) as ctx:
+            wfms.run_recipe(std.add.flowrep_recipe, a=1, b=NOT_DATA)
+        self.assertEqual(
+            str(ctx.exception), "Input port 'b' has no value and no default"
+        )
+
+    def test_defaulted_input_may_be_omitted(self):
+        """``increment(x, step=1)`` carries a reference, so 'step' need not be given
+        -- but 'x' still must be."""
+        recipe = library.increment.flowrep_recipe
+        self.assertListEqual(recipe.inputs_with_defaults, ["step"])
+        node = wfms.run_recipe(recipe, x=5)
+        self.assertEqual(node.output_ports["output_0"].value, 6)
+        with self.assertRaises(TypeError) as ctx:
+            wfms.run_recipe(recipe, step=2)
+        self.assertEqual(
+            str(ctx.exception),
+            "One of your AtomicRecipe() calls is missing 1 required input: ['x']",
+        )
 
     def test_input_ports_populated(self):
         node = wfms.run_recipe(std.add.flowrep_recipe, a=3, b=4)
@@ -961,10 +1001,10 @@ def test_not_unpacking(self):
         self.assertEqual(node.output_ports["result"].value, (0, 1))
 
     def test_unrecognized_input_raises(self):
-        with self.assertRaises(ValueError) as ctx:
-            wfms.run_recipe(std.add.flowrep_recipe, a=3, not_an_input=4)
+        with self.assertRaises(TypeError) as ctx:
+            wfms.run_recipe(std.add.flowrep_recipe, a=3, b=4, not_an_input=5)
         self.assertIn("not_an_input", str(ctx.exception))
-        self.assertIn("not found", str(ctx.exception))
+        self.assertIn("unexpected input", str(ctx.exception))
         self.assertIn(str(std.add.flowrep_recipe.inputs), str(ctx.exception))
 
     def test_positional_only_arguments(self):
@@ -984,6 +1024,27 @@ def test_linear(self):
         self.assertIsInstance(wf, datastructures.DagData)
         self.assertEqual(wf.output_ports["result"].value, (1 + 2) * 3)
 
+    def test_missing_input_raises(self):
+        """A reference-free workflow has no defaults to relax against, so every one
+        of its inputs must be supplied."""
+        recipe = _linear_workflow()
+        self.assertIsNone(recipe.reference)
+        self.assertListEqual(recipe.inputs_with_defaults, [])
+        with self.assertRaises(TypeError) as ctx:
+            wfms.run_recipe(recipe, x=1)
+        self.assertEqual(
+            str(ctx.exception),
+            "One of your WorkflowRecipe() calls is missing 2 required input: "
+            "['y', 'z']",
+        )
+
+    def test_referenced_workflow_defaults_may_be_omitted(self):
+        """``_passthrough_workflow(x=42)`` carries a reference, so 'x' is optional."""
+        recipe = _passthrough_workflow.flowrep_recipe
+        self.assertListEqual(recipe.inputs_with_defaults, ["x"])
+        wf = wfms.run_recipe(recipe)
+        self.assertEqual(wf.output_ports["x"].value, 42)
+
     def test_diamond(self):
         wf = wfms.run_recipe(_diamond_workflow.flowrep_recipe, a=3, b=7)
         self.assertEqual(wf.output_ports["result"].value, (3 + 7) * (-3))
@@ -1113,6 +1174,29 @@ def test_zipped_unequal_lengths_raises(self):
         with self.assertRaisesRegex(ValueError, "equal lengths"):
             wfms.run_recipe(_for_add_zipped(), xs=[1, 2], ys=[10, 20, 30])
 
+    def test_unfilled_nested_port_raises(self):
+        """There is no default to fall back on when there is nothing to iterate, so
+        say so rather than leaking a NotData into itertools.product.
+
+        Omitting the input outright is now caught up front by the argument binding, so
+        the surviving route here is a value that *arrives* as NOT_DATA -- which is what
+        a parent hands down for an input port nothing ever set.
+        """
+        with self.assertRaises(ValueError) as ctx:
+            wfms.run_recipe(_for_negate(), xs=NOT_DATA)
+        self.assertEqual(
+            str(ctx.exception),
+            "Iterated input 'xs' (body port 'a') has no value to iterate over",
+        )
+
+    def test_unfilled_zipped_port_raises(self):
+        with self.assertRaises(ValueError) as ctx:
+            wfms.run_recipe(_for_add_zipped(), xs=[1, 2], ys=NOT_DATA)
+        self.assertEqual(
+            str(ctx.exception),
+            "Iterated input 'ys' (body port 'b') has no value to iterate over",
+        )
+
 
 # ═══════════════════════════════════════════════════════════════════════════
 # wfms.py tests — while
@@ -1207,12 +1291,160 @@ def test_unhandled_exception_propagates(self):
 # ═══════════════════════════════════════════════════════════════════════════
 
 
+class TestPopulateInputPorts(unittest.TestCase):
+    def test_unknown_port_raises(self):
+        """No public path reaches this any more -- ``variadic_to_inputs`` rejects
+        unknown labels at the boundary first -- so it is pinned directly, as the
+        invariant every caller of this helper is required to have established.
+        """
+        node = datastructures.recipe2data(std.add.flowrep_recipe)
+        with self.assertRaises(ValueError) as ctx:
+            wfms._populate_input_ports(node, {"not_a_port": 1})
+        self.assertIn("Input port 'not_a_port' not found", str(ctx.exception))
+
+
 class TestUnrecognizedRecipe(unittest.TestCase):
     def test_unrecognized_input(self):
+        """Rejected by the guard that precedes argument binding, which would
+        otherwise trip over the missing ``inputs``."""
         not_a_recipe = "not at all"
         with self.assertRaises(TypeError) as ctx:
             wfms.run_recipe(not_a_recipe)
-        self.assertIn("Unsupported recipe type", str(ctx.exception))
+        self.assertIn("Unsupported recipe type: str", str(ctx.exception))
+
+    def test_unrecognized_recipe_subclass(self):
+        """A NodeRecipe the dispatch has no runner for still falls through."""
+        with self.assertRaises(TypeError) as ctx:
+            wfms.run_recipe(_UnrunnableRecipe(inputs=[], outputs=[]))
+        self.assertIn("Unsupported recipe type: _UnrunnableRecipe", str(ctx.exception))
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+# wfms.py tests — NodeRecipe.__call__ helpers
+# ═══════════════════════════════════════════════════════════════════════════
+
+
+class TestVariadicToInputs(unittest.TestCase):
+    """``std.add`` is a convenient stand-in: its inputs are ``a`` then ``b``."""
+
+    def setUp(self):
+        self.recipe = std.add.flowrep_recipe
+
+    def test_positional(self):
+        self.assertDictEqual(
+            wfms.variadic_to_inputs(self.recipe, 1, 2),
+            {"a": 1, "b": 2},
+            msg="Positional arguments should map onto inputs in declaration order",
+        )
+
+    def test_keyword(self):
+        self.assertDictEqual(
+            wfms.variadic_to_inputs(self.recipe, b=2, a=1),
+            {"a": 1, "b": 2},
+            msg="Keyword arguments should map by name, regardless of the order given",
+        )
+
+    def test_mixed(self):
+        self.assertDictEqual(
+            wfms.variadic_to_inputs(self.recipe, 1, b=2),
+            {"a": 1, "b": 2},
+            msg="Positional and keyword arguments should be combinable",
+        )
+
+    def test_no_inputs_to_fill(self):
+        recipe = workflow_recipe.WorkflowRecipe(
+            inputs=[], outputs=[], nodes={}, input_edges={}, edges={}, output_edges={}
+        )
+        self.assertDictEqual(wfms.variadic_to_inputs(recipe), {})
+
+    def test_recipe_is_positional_only(self):
+        """The recipe itself must not shadow an input that happens to be named
+        ``recipe``."""
+        recipe = _variadic_recipe(variadic_kwargs, ["recipe"])
+        self.assertDictEqual(wfms.variadic_to_inputs(recipe, recipe=42), {"recipe": 42})
+
+    def test_underfilled(self):
+        """Nothing downstream can supply a missing input: only reference-backed
+        recipes have defaults, and those never reach this helper."""
+        with self.assertRaises(TypeError) as ctx:
+            wfms.variadic_to_inputs(self.recipe, 1)
+        self.assertEqual(
+            str(ctx.exception),
+            "One of your AtomicRecipe() calls is missing 1 required input: ['b']",
+        )
+
+    def test_underfilled_multiple(self):
+        with self.assertRaises(TypeError) as ctx:
+            wfms.variadic_to_inputs(self.recipe)
+        self.assertEqual(
+            str(ctx.exception),
+            "One of your AtomicRecipe() calls is missing 2 required input: "
+            "['a', 'b']",
+        )
+
+    def test_too_many_positional(self):
+        with self.assertRaises(TypeError) as ctx:
+            wfms.variadic_to_inputs(self.recipe, 1, 2, 3)
+        self.assertEqual(
+            str(ctx.exception),
+            "One of your AtomicRecipe() calls takes 2 inputs but 3 positional "
+            "arguments were given -- its inputs are ['a', 'b']",
+        )
+
+    def test_duplicate(self):
+        with self.assertRaises(TypeError) as ctx:
+            wfms.variadic_to_inputs(self.recipe, 1, a=2)
+        self.assertIn("got multiple values for input 'a'", str(ctx.exception))
+
+    def test_unknown_keyword(self):
+        with self.assertRaises(TypeError) as ctx:
+            wfms.variadic_to_inputs(self.recipe, a=1, b=2, c=3)
+        self.assertIn("got an unexpected input 'c'", str(ctx.exception))
+
+    def test_binding_errors_are_not_value_errors(self):
+        """A try-recipe handling ValueError must not be able to swallow its caller's
+        mistake and quietly return partial data."""
+        for label, call in (
+            ("underfilled", lambda: wfms.variadic_to_inputs(self.recipe, 1)),
+            ("overfilled", lambda: wfms.variadic_to_inputs(self.recipe, 1, 2, 3)),
+            ("duplicate", lambda: wfms.variadic_to_inputs(self.recipe, 1, a=2)),
+            ("unknown", lambda: wfms.variadic_to_inputs(self.recipe, a=1, b=2, c=3)),
+        ):
+            with self.subTest(label):
+                with self.assertRaises(TypeError) as ctx:
+                    call()
+                self.assertNotIsInstance(ctx.exception, ValueError)
+
+
+class TestDataToReturn(unittest.TestCase):
+    def test_single_output(self):
+        data = wfms.run_recipe(std.add.flowrep_recipe, a=1, b=2)
+        self.assertEqual(
+            wfms.data_to_return(data),
+            3,
+            msg="A lone output should come back bare, like a single-return function",
+        )
+
+    def test_multiple_outputs(self):
+        data = wfms.run_recipe(library.divmod_func.flowrep_recipe, a=7, b=2)
+        self.assertTupleEqual(
+            wfms.data_to_return(data),
+            (3, 1),
+            msg="Multiple outputs should come back as a tuple in port order",
+        )
+
+    def test_no_outputs(self):
+        data = wfms.run_recipe(
+            workflow_recipe.WorkflowRecipe(
+                inputs=[],
+                outputs=[],
+                nodes={},
+                input_edges={},
+                edges={},
+                output_edges={},
+            )
+        )
+        self.assertTupleEqual(wfms.data_to_return(data), ())
 
 
 # ═══════════════════════════════════════════════════════════════════════════