Control Flow

Master conditional branching, loops, error handling, and other control flow patterns in FlowMason pipelines.

Control flow components modify pipeline execution - they enable branching, looping, error handling, and early returns. Unlike nodes and operators that process data, control flow components return directives that tell the executor how to proceed.

Control Flow Types

Conditional (If/Else)

Branch execution based on a boolean condition:

{
  "id": "check-length",
  "component": "conditional",
  "config": {
    "condition": "{{len(input.text) > 1000}}",
    "true_branch": "summarize-long",
    "false_branch": "summarize-short"
  }
}

The condition can be any expression that evaluates to true/false:

"condition": "{{input.score > 0.8}}"
"condition": "{{stages.validate.output.is_valid}}"
"condition": "{{input.type == 'premium'}}"

ForEach (Loop)

Iterate over a collection, executing stages for each item:

{
  "id": "process-items",
  "component": "foreach",
  "config": {
    "items": "{{input.documents}}",
    "loop_stages": ["extract", "summarize"],
    "item_variable": "current_doc",
    "index_variable": "doc_index",
    "parallel": true,
    "max_parallel": 3
  }
}

Accessing Loop Variables

Inside loop stages, access the current item via {{context.variable_name}}:

{
  "id": "extract",
  "component": "json-transform",
  "depends_on": ["process-items"],
  "config": {
    "data": {
      "item": "{{context.current_doc}}",
      "index": "{{context.doc_index}}"
    },
    "expression": "{ content: item.text, position: index }"
  }
}

Parallel Execution

Enable parallel processing for better performance:

{
  "parallel": true,
  "max_parallel": 5,
  "break_on_error": true
}

TryCatch (Error Handling)

Handle errors with recovery paths:

{
  "id": "safe-api-call",
  "component": "trycatch",
  "config": {
    "try_stages": ["fetch-data"],
    "catch_stages": ["use-fallback"],
    "finally_stages": ["cleanup"],
    "error_scope": "continue"
  }
}

Error Scopes

• propagate - Re-raise error after catch (like on-error-propagate)
• continue - Swallow error, continue pipeline (like on-error-continue)

Accessing Results

Downstream stages can access results from nested stages:

{
  "id": "process-result",
  "component": "json-transform",
  "depends_on": ["safe-api-call"],
  "config": {
    "data": {
      "api_result": "{{upstream.fetch-data}}",
      "error_info": "{{upstream.safe-api-call}}"
    }
  }
}

Router (Switch/Case)

Route to different branches based on a value:

{
  "id": "route-by-type",
  "component": "router",
  "config": {
    "value": "{{input.document_type}}",
    "routes": {
      "pdf": "process-pdf",
      "doc": "process-doc",
      "txt": "process-text"
    },
    "default": "process-generic"
  }
}

SubPipeline (Composition)

Call another pipeline as a sub-routine:

{
  "id": "run-preprocessing",
  "component": "subpipeline",
  "config": {
    "pipeline_id": "preprocessing-pipeline",
    "input_data": {
      "documents": "{{input.raw_documents}}"
    },
    "timeout_ms": 60000
  }
}

This enables pipeline composition and reuse across your organization.

Return (Early Exit)

Exit the pipeline early with a return value:

{
  "id": "check-input",
  "component": "return",
  "config": {
    "condition": "{{not input.text}}",
    "return_value": {"error": "No input provided"},
    "message": "Empty input"
  }
}

Use for guard clauses to validate inputs early:

{
  "stages": [
    {
      "id": "validate-input",
      "component": "return",
      "config": {
        "condition": "{{len(input.items) == 0}}",
        "return_value": {"items": [], "count": 0},
        "message": "No items to process"
      }
    },
    {
      "id": "process",
      "component": "generator",
      "depends_on": ["validate-input"]
    }
  ]
}

Best Practices

1. Keep control flow simple - Complex logic belongs in operators
2. Use meaningful stage IDs - Makes pipelines readable
3. Handle edge cases - Empty lists in foreach, missing routes
4. Consider error handling - Wrap risky operations in trycatch
5. Use guard clauses - Return early for invalid inputs
6. Avoid deep nesting - Flatten when possible