`parallel_subworkflows`

Execute multiple subworkflows concurrently and collect their results.

Parameters

flowsarrayconditional: Static array of flow configurations to execute in parallel. Use either flows or dynamic flow + source_key.
flowstringconditional: Dynamic fan-out child flow path. Required when flows is omitted.
source_keystringconditional: Context key containing the runtime array to fan out over. Required when flows is omitted.
inputobject: In dynamic mode, base input mapping applied to every child run.
item_keystringdefault "item": Dynamic mode child context key that receives the current source item.
index_keystringdefault "index": Dynamic mode child context key that receives the 1-based source item index.
child_output_keystring: Dynamic mode namespace for each child context inside its result entry.
output_keystringdefault "parallel_results": Key for the results array in context
on_errorstringdefault "fail_fast": Error handling: "fail_fast" (fail on any error) or "ignore" (collect all results)
max_concurrentnumberdefault CPU count: Maximum child workflows executing at the same time. Hard-capped at 1024.

Flow Entry Parameters

Each entry in the flows array:

Parameter	Type	Required	Description
`flow`	string	yes	Path to the `.lua` flow file (relative to `_flow_dir`)
`input`	object	no	Context mapping — keys are child context keys, values are parent context keys or literals. String values are treated as a parent context key when present, otherwise kept as string literals.
`output_key`	string	no	Namespace the child's output under this key in the result entry

Context Output

Key	Type	Description
`{output_key}`	array	Array of result objects, one per flow (in original order)
`{output_key}_count`	number	Total number of flows
`{output_key}_errors`	number	Number of flows that failed
`{output_key}_all_succeeded`	boolean	`true` if all flows succeeded

Each result entry contains:

Key	Type	Description
`success`	boolean	Whether the subworkflow succeeded
`flow`	string	The flow name
`error`	string	Error message (only present on failure)
context keys	any	Child flow output (merged directly or under per-flow `output_key`)

Example

local flow = Flow.new("parallel_workers")

-- Run three flows concurrently
flow:step("run_all", nodes.parallel_subworkflows({
    flows = {
        { flow = "fetch_users.lua", output_key = "users" },
        { flow = "fetch_orders.lua", output_key = "orders" },
        { flow = "fetch_metrics.lua", output_key = "metrics" }
    },
    max_concurrent = 3
}))

-- Use results from all three
flow:step("summarize", function(ctx)
    local results = ctx.parallel_results
    return {
        all_ok = ctx.parallel_results_all_succeeded,
        count = ctx.parallel_results_count
    }
end):depends_on("run_all")

return flow

With input mapping

flow:step("process", nodes.parallel_subworkflows({
    flows = {
        { flow = "worker.lua", input = { job_id = "job_1" } },
        { flow = "worker.lua", input = { job_id = "job_2" } },
        { flow = "worker.lua", input = { job_id = "job_3" } }
    },
    on_error = "ignore"
}))

Dynamic fan-out from context

Use flow + source_key when the parent workflow discovers work items at runtime. The child flow receives the current item under item and its 1-based position under index by default.

flow:step("run_jobs", nodes.parallel_subworkflows({
    flow = "worker.lua",
    source_key = "jobs",
    input = {
        run_id = "parent_run_id",
    },
    item_key = "job",
    index_key = "job_index",
    child_output_key = "result",
    max_concurrent = 5,
    output_key = "job_results",
}))

If ctx.jobs is:

[
  { "id": "a", "path": "/tmp/a.txt" },
  { "id": "b", "path": "/tmp/b.txt" }
]

then worker.lua runs twice. Each child context receives job, job_index, plus any mapped input fields. Dynamic mode allows an empty source array and returns an empty results array with {output_key}_all_succeeded = true.

Error handling

-- fail_fast (default): step fails if any subworkflow fails
flow:step("strict", nodes.parallel_subworkflows({
    flows = { ... }
}))

-- ignore: collect all results, check errors yourself
flow:step("tolerant", nodes.parallel_subworkflows({
    flows = { ... },
    on_error = "ignore"
}))