Control Flow

Control-flow blocks are composites: they embed sub-flows and are assembled directly by the flow builder rather than the leaf-block registry.

Each sub-flow slot (then, else, default, body, a fork branch, a switch case) has the same shape as a flow: an optional name and a process block list. Sub-flows must not declare source, workers, buffer, pool, or error — those are root-flow-only fields.

All condition and key fields are CEL expressions evaluated against the message (body, vars, eventID, correlationID, env, now). Expressions are compiled once at startup, so a malformed expression fails deployment rather than a message.

if

Runs one of two sub-flows depending on a boolean condition. When the condition is false and no else is configured, the message passes through unchanged.

- type: if
  name: any-orders
  condition: "size(body.orders) > 0"
  then:
    process:
      - type: log
        settings:
          message: '"processing " + string(size(body.orders)) + " orders"'
  else:
    process:
      - type: log
        settings:
          message: '"no orders to process"'

switch

Multi-case routing: runs the sub-flow of the first case whose when guard is true, in order. When no case matches, the default flow runs; with no default, the message passes through unchanged.

- type: switch
  name: classify-order
  cases:
    - when: "vars.order.amount >= vars.threshold"
      process:
        - type: log
          settings:
            message: '"HIGH order " + string(vars.order.id)'
  default:
    process:
      - type: log
        settings:
          message: '"low order " + string(vars.order.id)'

fork

Scatters the message across parallel branches, then joins. Each branch receives its own clone of the message and runs on the flow's shared worker pool (sized by the root flow's pool field), so branch effects on body and variables never leak back — on success the original input message passes through unchanged. Branch outputs are not aggregated.

Errors: the first branch error aborts the fork (cancelling the remaining branches) and propagates, labeled with the failing branch's name.

- type: fork
  name: notify-all
  branches:
    - name: audit
      process:
        - type: log
          settings: { message: '"audit " + correlationID' }
    - name: metrics
      process:
        - type: publish-event
          settings: { connector: bus, topic: orders }

foreach

Sequentially iterates an array, running the body sub-flow once per element with the element bound to a loop variable. mode picks what the loop is for: running the body for its side effects (iterate, the default), or transforming the array into a new one (map).

iterate mode (default)

• Iteration is sequential and runs on the shared message, so body edits (body, variables) carry into the next iteration and past the loop.
• The loop variable is restored to its pre-loop state (or removed) after the loop, so it does not leak.
• A body that drops the message stops the loop and drops it; a body error aborts the loop; a filter block inside the body that requests stop halts iteration and bubbles the stop to the root.

- type: foreach
  name: each-order
  items: "body.orders"
  as: order
  body:
    process:
      - type: log
        settings:
          message: '"order " + string(vars.order.id) + " amount=" + string(vars.order.amount)'

map mode

Each element's body runs on its own clone of the message, and the body it produces becomes one element of a new array, which replaces the message body. This is how you reshape a payload — call an API per element, project each element to a new shape — without hand-appending to a variable.

- type: foreach
  name: enrich-orders
  items: "body.orders"
  as: order
  mode: map
  body:
    process:
      - type: set-payload
        settings:
          value: '{"id": vars.order.id, "total": vars.order.amount * 1.2}'
# body is now the array of the objects each iteration produced

• As many elements out as in. An iteration whose body drops the message contributes null at that position rather than shortening the array or dropping the whole message — downstream blocks decide what a null element means.
• Iterations are independent. Each runs on a clone of the incoming message, so one element's variables cannot leak into the next, and the loop variable never escapes. (This is the opposite of iterate mode, where edits accumulate on the shared message.)
• A body error aborts the loop. A filter block inside the body that requests stop halts iteration, writes back what was collected so far, and bubbles the stop to the root.

enrich

Runs a body sub-flow on an isolated clone of the message, then merges back only what you name: setBody computes the new body and each setVars entry sets one variable. Everything else the scope did (scratch variables, intermediate bodies) stays isolated.

Errors: a body error aborts the block; a body that drops the message drops it here too.

- type: enrich
  name: derive-total
  # setBody omitted: the incoming order body is preserved.
  setVars:
    total: body.total   # pull just the total out of the scope's result
  body:
    process:
      - type: set-payload
        settings:
          value: '{"total": body.qty * body.price, "note": "scope-only body"}'

validate

A filter block: asserts a list of boolean CEL rules against the message. If all hold, the message passes through unchanged. If any fail, the block rejects — it shapes a terminal response and requests the flow stop, so the rest of the chain never runs.

On rejection:

• The failing rules' message texts are collected into vars.validationErrors (all rules are evaluated, so every failure is reported).
• Without onReject, the built-in response body is {"error": "validation_failed", "messages": [...]} and vars.httpStatus is set to rejectStatus. Over an HTTP source that becomes the response status.
• With onReject, the sub-flow's output is the terminal response; a sub-flow that drops the message drops it here too.

- type: validate
  name: check-order
  rejectStatus: 422
  rules:
    - expr: 'has(body.id)'
      message: "order id is required"
    - expr: 'body.amount > 0'
      message: "amount must be positive"

# Only reached when every rule held.
- type: set-payload
  settings:
    value: '{"orderId": body.id, "status": "accepted"}'

handle-errors

Inline error recovery: runs the process chain and, on error, exposes the failure as vars.error and runs the error chain. Both are bare block lists, so the block reads as a mini-flow embedded inline. When the error chain succeeds, its output is the block's result and the flow continues normally (recovery).

vars.error is a map: {message: <error text>, flow: <handle-errors block name>, block: <failing block label>}. If the error chain itself errors, the block fails with that error. A root flow's own error: chain provides the same recovery at flow level — see Error handling.

- type: handle-errors
  name: charge
  process:
    - type: rest
      name: call-charge
      settings:
        connector: payments
        method: POST
        path: /charges
        body: '{"amount": body.amount}'
  error:
    - type: set-payload
      settings:
        value: '{"status": "degraded", "reason": vars.error.message}'

cache-scope

Memoizes the body its wrapped flow produces in the runtime store, keyed by an evaluated expression. On a fresh hit it restores the cached body and skips the body flow entirely; on a miss (or an expired entry) it runs the flow and stores its body for next time.

Notes:

• Only the message body is cached; variables the wrapped flow sets are not restored on a hit.
• Storing is best-effort: a version conflict (another worker cached first) or an absent store leaves the result correct, just uncached. A body error aborts the block; a body that drops the message drops it (nothing is cached).
• Entries live in the same per-deployment store as object storage; see Runtime services.

- type: cache-scope
  name: cached-report
  key: '"report"'
  ttl: 1m
  body:
    process:
      - type: set-payload
        settings:
          value: '{"report": "generated", "event": eventID}'