Chapter 22 — Flow Control as Data

Sequencing a workflow is a property of the workflow file, not of the engine. The engine reads gates and jumps from data; the data lives in the YAML the author wrote.

The previous chapters introduced the model (chapter 18), the shipped tasks (chapter 19), the YAML grammar (chapter 20), and the four communication channels (chapter 21). This chapter walks the engine that runs the workflow — specifically, the small set of mechanisms by which flow control becomes data in the YAML and behaviour at runtime.

The FlowEngine class shipped in codecraftai-taskplus is small — under a hundred lines. It implements three flow-control mechanisms:

The if_expr gate — conditional inclusion of a node.
The onfailure / onsuccess jump — label-based jumps driven by task return codes.
The recursion into TaskGroups — sequencing of nested structure.

Each mechanism is data-driven: the workflow author declares the behaviour in the YAML, the engine reads the declaration and acts on it. No imperative orchestration code; the workflow file is the source of truth for how the workflow behaves.

The engine's outer loop

The shipped FlowEngine.execute method is the outermost layer. It runs once per workflow invocation:

def execute(self, workflow: Workflow) -> int:

    result_code = 0

    environment = workflow.environment
    parameters = workflow.parameters

    flowitems = workflow.flow

    flowstate = {}
    errors = []
    jump_context = JumpContext()

    self.process_flowitems(environment, parameters, flowstate, errors, flowitems, jump_context)

    if len(errors) > 0:
        result_code = 1

    return result_code

Six lines of visible behaviour:

Pull environment and parameters from the workflow. These are the chapter-21 channels 1 and 2. The engine threads them through every task as the first two arguments to perform_task.
Initialise an empty flowstate dict. This is channel 3 — the per-run mutable inter-task communication channel.
Initialise an empty errors list. Tasks (or the engine itself) can append error records during the run; the final non-empty errors list translates to a non-zero workflow result.
Initialise a JumpContext. The label-jump state for the whole workflow; the engine threads this through every level of recursion.
Call process_flowitems on the top-level flow list. This is the recursive walk; every flow node visited or skipped flows through this method.
Translate errors into a final result code. Empty errors list → success; non-empty → failure.

The engine is single-pass. No retries at the engine level (the chapter-18 design choice — retries belong inside tasks, not in the engine), no parallelism, no backtracking. One walk through the flow tree, in order, with skip-and-jump primitives applied.

`process_flowitems` — the recursive walk

The recursion is where every flow control decision is made. In its entirety:

def process_flowitems(self, environment, parameters, flowstate, errors,
                     flowitems: List[FlowNode], jump_context: JumpContext):

    for fitem in flowitems:

        node_label = fitem.label

        # If we are seeking a specific node, skip any nodes that are not that node
        if jump_context.seeking_node is not None:
            if jump_context.seeking_node != node_label:
                continue
            else:
                jump_context.seeking_node = None

        include_node = True
        if fitem.if_expr is not None:
            include_node = self.evaluate_expression(fitem.if_expr, environment, parameters)

        if include_node:
            if isinstance(fitem, Task):

                task_result = fitem.execute(environment, parameters, flowstate)
                if task_result != 0:
                    if fitem.onfailure is not None:
                        jump_context.seeking_node = fitem.onfailure
                elif fitem.onsuccess is not None:
                    jump_context.seeking_node = fitem.onsuccess

            elif isinstance(fitem, TaskGroup):

                group_items = fitem.items

                self.process_flowitems(environment, parameters, flowstate, errors, group_items,
                                       jump_context)

    return

Twenty-five lines of method body. Walk what each block does.

The jump check (lines 67–72)

The first thing the loop does for each node is check whether a jump is in progress. When jump_context.seeking_node is set, the engine is in seek mode — looking for a specific labelled node, skipping everything until it finds it.

if jump_context.seeking_node is not None:
    if jump_context.seeking_node != node_label:
        continue
    else:
        jump_context.seeking_node = None

Three behaviours:

Not seeking (seeking_node is None): proceed normally.
Seeking and current node doesn't match (seeking_node != node_label): continue — skip this node entirely. Do not evaluate its gate; do not execute it; do not recurse into it if it's a taskgroup.
Seeking and current node matches (seeking_node == node_label): clear the seeking state and fall through to process this node normally.

The seek-and-skip pattern is the engine's only flow-control primitive beyond sequential traversal. All cross-node jumps reduce to setting seeking_node and letting the loop's next iterations skip until they reach the named target.

The gate evaluation (lines 74–76)

When the node is not being skipped by a jump, the engine checks the node's if_expr:

include_node = True
if fitem.if_expr is not None:
    include_node = self.evaluate_expression(fitem.if_expr, environment, parameters)

Two behaviours:

No if_expr (fitem.if_expr is None): include_node stays True; the node will run.
if_expr set: evaluate the expression with the workflow's environment and parameters as context. When the expression returns truthy, include_node becomes True; when it returns falsy, include_node becomes False and the node is skipped.

The gate runs after the seek check. This means: a node can be skipped by either being on the wrong side of a jump or by having its gate evaluate false. Both produce the same outcome (the node is bypassed) through different mechanisms.

The execution branch (lines 78–93)

When the node is included, the engine dispatches by node type:

Task path

if isinstance(fitem, Task):

    task_result = fitem.execute(environment, parameters, flowstate)
    if task_result != 0:
        if fitem.onfailure is not None:
            jump_context.seeking_node = fitem.onfailure
    elif fitem.onsuccess is not None:
        jump_context.seeking_node = fitem.onsuccess

The task's execute method runs (which delegates to perform_task per chapter 18's Template Method pattern) and returns an integer. Then:

Task failed (task_result != 0) and onfailure is declared: set seeking_node to the onfailure label. The loop's next iteration will start skipping until the named task is found.
Task failed and onfailure is not declared: do nothing — the task failure is silently absorbed and execution continues to the next node. (Whether this is the right semantic is discussed below.)
Task succeeded (task_result == 0) and onsuccess is declared: set seeking_node to the onsuccess label.
Task succeeded and onsuccess is not declared: fall through to the next node normally.

TaskGroup path

elif isinstance(fitem, TaskGroup):

    group_items = fitem.items

    self.process_flowitems(environment, parameters, flowstate, errors, group_items,
                           jump_context)

The engine recurses into the taskgroup's items, threading the same jump_context into the recursion. This is what allows cross-group jumps: when a task inside group A sets seeking_node to a label inside group B, the recursion bubbles back up to the outer loop, which keeps skipping until the recursion descends into group B and finds the labelled node.

The recursion is what makes the engine's flow control compose. The same process_flowitems method handles both top-level sequences and nested taskgroup items; every node is at some depth, but the depth doesn't change how flow control works.

`evaluate_expression` — the gate evaluator

The gate evaluator is two non-trivial lines:

def evaluate_expression(self, if_expr: str, environment: Dict[str, str],
                        parameters: Dict[str, Any]) -> bool:

    flowstate = self._flowstate

    eval_result = eval(if_expr)

    return eval_result

The implementation uses Python's built-in eval against a string expression, with three names available in the evaluation scope: environment, parameters, and flowstate.

Three observations.

The expression is Python

The if_expr field in the YAML is a Python expression string. A workflow's gate looks like this:

- task: Run benchmark
  tasktype: my.module@Benchmark
  if: "parameters['build_id'].startswith('nightly')"

- task: Tear down staging
  tasktype: my.module@Teardown
  if: "parameters['cleanup_after_run'] == True and not flowstate.get('teardown_skipped', False)"

The YAML author writes Python. This is more powerful than (say) a custom expression DSL — comparison operators, string methods, boolean logic, dict subscript, and method calls are all available.

It is also more dangerous than a custom DSL. eval on attacker-controlled input is well-known as a security vulnerability; the discipline relies on workflow YAML being authored by trusted developers and being checked into the repository. When a workflow's if_expr would consume external input, the workflow author should prefer to compute the condition in a task and write a boolean to flowstate, then gate on the simpler flowstate['xyz'] expression.

`flowstate` is the engine's, not the workflow's

The evaluator pulls flowstate from self._flowstate — specifically, the engine's instance variable. The workflow's flowstate dict (the per-run dict initialised in execute and threaded through process_flowitems) is different from self._flowstate. This is a subtle implementation detail of the shipped engine; gates that try to read flowstate may not see what tasks have written.

This is a known limitation. Workflow authors who want gates to read task-produced state should use parameters (read-only, visible everywhere) for stable values, or work around the limitation by writing a custom task that combines computation and conditional behaviour into one step.

Errors propagate

When eval raises (a syntax error in the expression, a KeyError from missing parameters, a NameError from referring to an undefined name), the exception propagates through evaluate_expression and through process_flowitems up to the caller. The engine does not catch the exception; the workflow fails.

This is consistent with the volume's "errors should never pass silently" discipline (chapter 0, foundation tier). A malformed gate is a workflow bug; failing loudly is better than treating the gate as falsy and silently skipping the node.

A worked example of jump propagation

To make the jump mechanics concrete, walk through what happens when a task fails and triggers a cross-group jump.

The workflow:

name: cross-group-jump-example
flow:
    - task: Setup
      tasktype: my.module@Setup

    - taskgroup: Test phase
      items:
          - task: Smoke
            tasktype: my.module@Smoke
          - task: Integration
            tasktype: my.module@Integration
            onfailure: Run cleanup
          - task: Benchmark
            tasktype: my.module@Benchmark

    - taskgroup: Cleanup phase
      items:
          - task: Stop services
            tasktype: my.module@StopServices
          - task: Run cleanup
            tasktype: my.module@RunCleanup

The Integration task at ordinal "2.2" has onfailure: Run cleanup, naming the task at ordinal "3.2". Trace what the engine does when Integration fails.

Time T₀: Engine begins

seeking_node = None
flowitems = [Setup, TestPhase, CleanupPhase]

The outer loop starts walking the top-level flow.

Time T₁: Setup runs

Setup is processed normally. seeking_node was None; if_expr was None; the engine executes the task. Suppose it succeeds; no onsuccess declared; fall through.

seeking_node = None
just-processed = Setup (ordinal 1)

Time T₂: Engine recurses into Test phase

TestPhase is a TaskGroup; the engine calls process_flowitems recursively with flowitems = [Smoke, Integration, Benchmark].

seeking_node = None
recursion-depth = 1
inner-flowitems = [Smoke, Integration, Benchmark]

Time T₃: Smoke runs

The inner loop processes Smoke. Succeeds; no onsuccess; fall through.

seeking_node = None
just-processed = Smoke (ordinal 2.1)

Time T₄: Integration runs and fails

The inner loop processes Integration. The task's perform_task returns non-zero. The engine sees the failure and the onfailure label:

if task_result != 0:
    if fitem.onfailure is not None:
        jump_context.seeking_node = fitem.onfailure

seeking_node becomes "Run cleanup". The inner loop's next iteration begins.

seeking_node = "Run cleanup"
just-processed = Integration (ordinal 2.2, FAILED)

Time T₅: Benchmark is skipped

The inner loop reaches Benchmark. seeking_node is set to "Run cleanup"; Benchmark's label is "Benchmark"; mismatch. The engine continues. No gate evaluation, no execution.

seeking_node = "Run cleanup"
just-skipped = Benchmark (ordinal 2.3)

Time T₆: Inner loop ends; recursion returns

The inner loop's iteration is complete. process_flowitems returns to its caller (the outer loop). Critically, the seeking_node = "Run cleanup" state survives the return — both the outer and inner loops share the same jump_context reference.

seeking_node = "Run cleanup"
recursion-depth = 0 (returned)

Time T₇: Outer loop resumes; CleanupPhase reached

The outer loop's next iteration is CleanupPhase. The engine checks: seeking_node = "Run cleanup"; CleanupPhase's label is "Cleanup phase"; mismatch.

But this is a TaskGroup, not a Task. Does the seek-and-skip treatment of TaskGroups skip the whole group or just the group node itself?

Looking at the loop structure:

for fitem in flowitems:
    node_label = fitem.label

    if jump_context.seeking_node is not None:
        if jump_context.seeking_node != node_label:
            continue       # <-- skips this iteration entirely

The continue skips the entire iteration — including the recursion into the TaskGroup's items. So when the outer loop encounters CleanupPhase while seeking "Run cleanup" and the group's own label doesn't match, the whole group is skipped without descending into its items.

This is a subtle behaviour worth naming: the engine's seek-and-skip applies at every level of the tree, including taskgroup boundaries. When a jump's target is inside a taskgroup, the engine has to first enter the taskgroup — which means matching the taskgroup's own label is the only way for seek mode to descend into the items.

Look back at the workflow: CleanupPhase is labelled "Cleanup phase". The jump target is "Run cleanup". Mismatch. The group is skipped. The cross-group jump doesn't reach its target.

This is a real subtlety in the shipped engine. Workflow authors who want cross-group jumps to work as intuited (jumping from inside one group to a label inside another) should declare the outer group's label to match the seek target — or restructure so that the jump targets a top-level task rather than one inside a different taskgroup.

The volume's foundation-document outline notes this kind of subtlety as a place where the discipline of expressing flow control as data exposes its limits: jumps are powerful in theory, brittle in nested practice. Chapter 26's case study revisits the pattern and shows the workaround.

What flow control as data buys

Despite the subtlety above, the discipline of flow control as data has clear payoffs over imperative orchestration.

Reviewability

A workflow's behaviour is visible in the YAML. When a workflow fails in production, the engineer reading the YAML can see which gates and jumps applied to that run. An imperative orchestration with branches, loops, and exception handlers distributed across a Python script does not have this property — understanding what would have happened requires running the code in your head.

Captured-profile compatibility

Chapter 16's capture-and-replay primitive serialises the resolved configuration but not the engine's runtime behaviour. This works because the engine's behaviour is derivable from the workflow YAML plus the captured configurations. An imperative orchestration's behaviour would also depend on the engine's internal state, the order of exception handlers, and other runtime details that wouldn't serialise cleanly.

Audit story

A configuration change to the workflow's flow is a git diff, visible at code review time. An imperative-orchestration change spread across multiple Python files is harder to review as a unit. The discipline keeps the changes small and reviewable, the same way every other chapter of the volume has argued for.

Predictability

The engine's behaviour is bounded. The shipped process_flowitems is twenty-five lines; the worst-case behaviour for a given input is computable. The engine does not retry, does not back off, does not introduce non-determinism beyond what's in the workflow's if_expr strings. When a workflow misbehaves, the cause is in the YAML or in a task's implementation, not in the engine's hidden state.

What flow control as data does not buy

Three honest scoping notes.

Not arbitrary control flow

The engine has gates and jumps. It does not have loops, parallelism, try/except, or coroutines. When the orchestration needs richer control flow — running three tasks concurrently, retrying with exponential backoff, branching on runtime conditions richer than a single boolean — the right answer is either:

A custom Task implementation that does the richer behaviour internally (chapter 23 develops the extension recipe).
A higher-level orchestration tool entirely (Airflow, Argo, Step Functions) that the workflow author chooses because it has the primitives this engine doesn't.

The chapter-27 ("where this approach is not the right fit") honest-scoping discussion includes "heavy parallel fan-out" and "dynamic workflow generation" precisely because the engine's small set of primitives doesn't fit those use cases.

Not transactional

The engine has no notion of atomic multi-task operations. When task A succeeds and task B fails halfway through, there's no built-in mechanism to "undo" task A. Tasks that need transactional semantics implement them themselves (most commonly through chapter-19's onfailure jumps to a cleanup task).

The limitation is acceptable for most orchestration use cases — pipeline steps are typically idempotent or have explicit cleanup paths — but workflow authors should be aware that the engine is not a transaction manager.

Not parallelism

A TaskGroup does not run its items in parallel. Items run sequentially in declaration order. When parallelism is needed, the workflow author either:

Spawns parallel work from inside a custom Task implementation.
Uses the CI vendor's native parallelism (matrix builds, parallel jobs).
Restructures the workflow into multiple workflows that the CI vendor runs in parallel.

The shipped engine is a sequential one-pass walker. This is deliberate (chapter 18's "what's not in the model" discussion); parallelism at the engine level would require a scheduling abstraction the simple model doesn't need.

Audit findings on `flowengine.py`

The pre-book audit pass on this file found and fixed several bugs. All are now covered by the 115-test suite that came out of the audit:

evaluate_expression had a topology parameter in its signature that was never used and never passed. Calls into the function from process_flowitems only passed (if_expr, environment, parameters); the extra parameter was a vestigial leftover from an earlier API. Fixed by removing the parameter.
process_flowitems had topology in its signature with the same vestigial-parameter problem. execute was passing it; the inner uses didn't reference it. Fixed by removing.
execute had result_code = 0 before process_flowitems was called, then never updated unless errors were appended. Tasks failing without onfailure jumps were silently absorbed; the workflow result code reflected only errors that someone had explicitly added to the errors list. This is still the shipped behaviour — failing tasks without onfailure declarations don't increment the errors list — and the audit chose to preserve the behaviour as documented rather than change it; chapter-27's scoping discussion notes this as a known limitation.

The errors-list-as-result-channel pattern is the engine's weakest design point. A future iteration could replace it with a strict-by-default mode where any unhandled task failure marks the workflow as failed; the volume documents the current behaviour rather than papering over it.

Summary

The FlowEngine is a small (~100 LOC) sequential engine implementing three flow-control mechanisms: the if_expr gate, the onfailure/onsuccess label jump, and the recursive TaskGroup walk.
The engine's outer loop (execute) sets up environment, parameters, flowstate, errors, and a JumpContext, then delegates to the recursive process_flowitems.
Flow control is seek-and-skip: when a jump fires, jump_context.seeking_node is set; subsequent loop iterations skip every node whose label doesn't match, including taskgroup boundaries.
The if_expr gate is a Python expression evaluated with eval against an environment containing environment, parameters, and flowstate (the engine's _flowstate, with the limitation that this is not the same as the workflow's per-run flowstate dict).
Cross-group jumps work if the engine descends into the right taskgroup before encountering the target. Jumps targeting labels inside a different taskgroup will skip the wrong group unless the workflow author structures the labels carefully — a real subtlety chapter 26 will revisit.
The discipline of "flow control as data" buys reviewability, captured-profile compatibility, an audit story, and bounded predictability — at the cost of giving up arbitrary control flow, transactionality, and engine-level parallelism.
The audit pass on flowengine.py removed two vestigial parameters (topology) and noted the errors-list-as-result- channel pattern as the engine's weakest design point.

The next chapter is the last in Part V: extending the task model. How to add a new task type when the five shipped types don't cover the use case, with the chapter-19 GitCheckout stub as the worked target.