Parse raw recipe steps when parsing workflow functions

[liamhuber]

Already if there is a function with a recipe attached to it, we directly leverage the recipe. I tried naively duplicating the accompanying logic in the case that the call is a recipe, but a few of the workflow parser tests failed. I think it should be doable though, so I'll take a crack at it.

This would make a few edge cases easier, e.g. directly leveraging a the recipe wrapper around an awkward function like np.linspace.

For linter completeness a __call__ method might need to be added to the node model, but this could be a doc-like function explaining that it's there for using the recipe as a step in a parsed workflow.

[liamhuber]

To give a concrete example, where we currently accept

def add(a, b):
    return a + b

@flowrep.models.workflow
def my_wf(a, b, c):
    x = add(a, b)
    y = add(x, c)
    return y

it would be nice to also accept

add_recipe = flowrep.models.parse_atomic(add)

@flowrep.models.workflow
def my_wf(a, b, c):
    x = add_recipe(a, b)
    y = add_recipe(x, c)
    return y

After sleeping on this, I guess it is more complex that I hoped, because it entangles with the promise that workflow-decorated functions are still just executable functions.

Now, in terms of parsing and recipe validation, I still think this is pretty straightforward. Even when you think about the recipes being control flow recipes, since their IO is still concretely known, we ought to still be able to successfully do the symbol resolution for IO edges just fine.

However, where I start to struggle is thinking about our promise that the decorated functions ought to still be runnable as plain python. My thought is that this requires the recipe objects to implement a __call__ method compliant with the recipe. For atomic nodes, this is dead simple -- we know all the inputs and what their signature kind is from the PythonReference​, so we can just import the function from the fully qualified name, pass it the input, and return the result in __call__​. For workflows this only works when the reference is populated -- otherwise, we need to use the recipe to construct a graph, analyze the topology to compute execution dependency order, and then recursively execute the child recipes. For the remaining flow controls, it's similar to workflows where we need to explicitly construct an execution pattern using the (prospective) node recipes.

I.e., if we want to parse recipes in workflow definitions and stick by our promise that decorated functions are still just executable, we need to implement a (simple) WfMS inside the __call__ dunders of our recipe objects.

@samwaseda already offers such a proof-of-concept WfMS in the current flowrep.workflow module, and maybe this is something we ought to do anyhow. I'm nonetheless a little on the fence about this. @samwaseda?

[liamhuber]

I chatted with Claude about this, and I think it has a decent take:

The __call__ question is where it gets interesting. A few tiers of ambition:

Tier 1 — reference-backed only (low effort, high value):
Every recipe produced by the parsers has a PythonReference with a fully qualified name. AtomicNode.__call__ imports and calls the function. WorkflowNode.__call__ with a reference does the same. This covers your motivating example (add_recipe = parse_atomic(add)) with almost no work and zero WfMS logic. For recipes without a reference, raise TypeError("recipe is not directly callable — no backing function").

Tier 2 — topo-sort execution for flat workflows (moderate effort):
For WorkflowNode without a reference (e.g. from pwd2flowrep or manual construction), do a topological sort on the edges, recursively __call__ each child node, and thread the data through. This is ~30-50 lines and conceptually clean because workflows are static DAGs. Still no WfMS for dynamic nodes.

Tier 3 — full mini-WfMS on all node types (high effort, questionable value):
Implement execution semantics for ForNode (scatter/gather), WhileNode (loop-until-false), IfNode (condition evaluation + branch selection), TryNode (exception routing). This is a real execution engine.

My take on the tradeoffs:

The case for tier 3 is conceptual elegance — every recipe is callable, full composability, round-trip fidelity between "recipe as code" and "code as recipe." It also serves as a reference implementation that documents the intended semantics.

The case against is scope creep. flowrep is a representation layer, and baking execution semantics into the models couples them tightly. Every model change requires a corresponding __call__ change. The control-flow execution logic (scatter/gather for ForNode, iteration for WhileNode) is exactly what downstream WfMS implementations need to own — duplicating it here means two places to maintain and two places where subtle semantic divergence can hide. You'd also need to decide things like: does ForNode.__call__ actually parallelize, or just loop sequentially? Either choice is a semantic commitment that a representation library arguably shouldn't make.

My recommendation: start with tier 1, optionally tier 2, explicitly skip tier 3.

Tier 1 is the 90% case — recipes produced by the parsers always have references, and those are the ones you'd use in @workflow decorated functions. Tier 2 is a natural extension for pwd2flowrep round-tripping. For control-flow nodes without references, a clear error saying "this recipe requires a WfMS to execute" is honest and keeps the boundary clean.

One nuance: if you go with tier 1, you might want __call__ on NodeModel base class as a raise NotImplementedError, with overrides on AtomicNode and WorkflowNode. That way the contract is explicit and discoverable.

I had jumped from thinking about Tier 1 straight into how to handle Tier 3. I think the point is well made though, that Tier 1 is going to cover 90% of the cases. The only time an @workflow function wouldn't stay executable is if you're invoking a workflow recipe with no python reference, or a flow control as a recipe. E.g., consider taking a simple workflow with a flow control

from flowrep.models import atomic, workflow

@atomic
def my_identity(x):
    return x

@atomic
def my_range(n):
    return list(range(n))

@workflow
def my_runnable_workflow(n):
    domain = my_range(n)
    vecs = []
    for n in domain:
        x = my_identity(n)
        vecs.append(x)
    return vecs

And re-writing it with the flow control captured as a recipe

my_for = ForNode.model_validate(
    {
        "type": "for",
        "inputs": ["pass_through"],
        "outputs": ["vecs"],
        "body_node": {"label": "body", "node": my_identity.flowrep_recipe},
        "input_edges": {"body.x": "pass_through"},
        "output_edges": {"vecs": "body.x"},
        "nested_ports": ["x"],
        "zipped_ports": [],
    }
)

@workflow
def my_unrunnable_workflow(n):
    domain = my_range(n)
    vecs = my_for(pass_through=domain)
    return vecs

In this case both would be valid (identical) recipes, but the latter no longer executes as a python function.

That is annoying, but the upside to allowing recipes directly is it lets us define recipes around awkward functions (e.g. np.linspace) and then invoke those recipes directly in the workflow definition. In this simple case where it's a function recipe, the Tier 1 approach would still leave the workflow python function as callable.

So I rather like Tier 1. Unlike Claude, I think Tier 2 is a horrible middle ground, but I agree with the concerns on Tier 3. @samwaseda?