# Spec: Parallel Fan-Out and Foreach Iteration ## Status Implemented. ## Scope This spec defines two item-expansion step fields: - `parallel`, which expands items into represented child DAG runs or enqueue requests. - `foreach`, which runs an inline step body once for each item in the same DAG document. This spec covers: - accepted field shapes - item source parsing - item-scoped value references - concurrency limits - parent step status - aggregate outputs - validation and runtime errors This spec does not define: - graph DAG scheduling of independent top-level steps - nested-array step shorthand for parallel branches - the `call` and `params` aliases as normative syntax - scheduler, queue worker, coordinator, storage, API, or UI internals - how a child DAG or body step executes after the item-specific inputs have been resolved - action-specific behavior other than the `dag.run` and `dag.enqueue` fan-out surfaces named here ## Related Specs - YAML schema: [Spec 002: YAML Schema](002-yaml-schema.md) - Value resolution: [Spec 003: Value Resolution and Field Evaluation](003-value-resolution.md) - Environment values: [Spec 006: Value Resolution Env](006-value-resolution-env.md) - Step output references: [Spec 007: Value Resolution Steps](007-value-resolution-steps.md) - Step identity: [Spec 009: Step Reference](009-step-reference.md) - Step outputs: [Spec 012: Step Outputs](012-step-outputs.md) - Step run: [Spec 013: Step Run](013-step-run.md) ## Terms An item source is a workflow value that expands to zero or more items. An item is one expanded value from an item source. An item slot is the zero-based position of an item in the expanded item list. An item key is the stable string identity for one item slot inside one `foreach` step. An item body is the inline `steps` list under `foreach`. An item run is one represented child DAG run or enqueue request created by `parallel` after item-specific child DAG target and params are resolved. ## Behavior ### Common Behavior #### Item Ordering - Item sources are expanded after the owning step's dependencies have reached the status required for the owning step to start. - Static YAML arrays are expanded in YAML array order. - JSON arrays are expanded in JSON array order. - A spec section below must explicitly say when consumers may rely on aggregate output array order. - Starting and completion order are not guaranteed unless an owning action spec explicitly says otherwise. #### Concurrency - `max_concurrent` limits the number of represented child runs, enqueue requests, or item bodies that may be active at the same time for the owning expansion construct. - `max_concurrent` must be an integer from `1` through `1000`. - When `max_concurrent` is omitted, Dagu uses the inherited `defaults.max_concurrent`, or `10` when no default is configured. - `defaults.max_concurrent` follows the same integer rules. It applies to every `parallel` and `foreach` step, including steps inside a `foreach` body, in every item source form, and never adds `parallel` or `foreach` to a step. - The nearest `defaults.max_concurrent` wins: the DAG, then the workspace base configuration, then the global base configuration. Setting other `defaults` fields does not reset an inherited `max_concurrent`. - `max_concurrent` does not limit normal top-level graph scheduling outside the owning expansion step. #### Value Resolution - Item source strings are value-resolved before expansion. - Item-scoped references are available only in the fields explicitly named by the `parallel` or `foreach` sections below. - Item-scoped references do not create dependencies. - References to prior top-level step outputs are ordered under Spec 007, through `depends` or through an inferred dependency on the owning step. - Dynamic evaluation is not run for item-scoped references unless another spec explicitly opts in. ### Parallel #### Field Shape `parallel` is an optional step field. Rules: - A step with `parallel` must use `action: dag.run` or `action: dag.enqueue`. - A step with `parallel` must not define `foreach`. - `parallel` accepts a string item source: ```yaml steps: - id: fanout action: dag.run with: dag: process-item parallel: ${params.items} ``` - `parallel` accepts a direct array item source: ```yaml steps: - id: fanout action: dag.run with: dag: process-item parallel: - item-a - item-b ``` - `parallel` accepts an object form with required `items` and optional `max_concurrent` and `separator`: ```yaml steps: - id: fanout action: dag.run with: dag: process-item parallel: items: - item-a - item-b max_concurrent: 2 ``` - Object-form `parallel` must not contain fields other than `items`, `max_concurrent`, and `separator`. - `parallel.items` may be a string item source or a direct array item source. - Direct array items may be strings, numbers, or flat mappings. - A flat mapping item represents item fields. - Flat mapping item values may be strings, numbers, or booleans. - Nested mapping items and nested array items are invalid for static `parallel` arrays. - A `parallel` expansion must produce at least one item. - A `parallel` expansion must not produce more than `1000` items. #### String Item Sources For `parallel`, a string item source supports these forms. Rules: - If the resolved string is a valid JSON array, each JSON array entry is one item. - If the resolved string is a valid JSON object, the whole object is one item. - If the resolved string is not JSON, non-empty tokens become items. - Token separators are line breaks, commas, semicolons, pipes, tabs, or runs of unquoted whitespace. - Double-quoted token groups preserve whitespace inside the group. - Empty tokens are ignored. - An empty or whitespace-only resolved string produces zero items and therefore fails the `parallel` step. Separator detection cannot distinguish a single item that contains a delimiter character from several delimited items on one line. `parallel.separator` names the separator explicitly: ```yaml steps: - id: fanout action: dag.run with: dag: process-item parallel: items: ${steps.list.outputs.rows} separator: newline ``` Rules: - `parallel.separator` is optional and must be exactly one of `newline`, `comma`, `semicolon`, `pipe`, `tab`, or `space`. - `parallel.separator` requires a string item source. It is invalid with a direct array item source. - When `parallel.separator` is set, the resolved string is split only on that separator. JSON detection, quoted-token handling, and separator detection are not applied, so a token may contain any other delimiter character. - `space` splits on runs of whitespace, including tabs and line breaks. - Tokens are trimmed of surrounding whitespace and empty tokens are ignored. #### Item Scope For each expanded item, Dagu evaluates the child DAG target and child DAG params with an item scope. Rules: - `${ITEM}` resolves to the current item. - For a scalar item, `${ITEM}` resolves to that scalar as text. - For an object item, `${ITEM}` resolves to a compact JSON object. - `${ITEM.field}` resolves to a top-level field from an object item. - `${ITEM.field}` is not defined for scalar items. - Nested item paths such as `${ITEM.a.b}` are outside this spec. - Item scope is available to `with.dag`. - Item scope is available to `with.params`. - Item scope is not available to `with.queue`. - Item scope is not available to unrelated step fields unless another spec explicitly opts in. - If `with.params` is omitted, the child DAG receives the current item as its whole parameter payload. - If `with.params` is present, the resolved `with.params` value is the child DAG parameter payload. Child DAG parameter payloads include item data only when `with.params` contains `${ITEM}` or `${ITEM.field}`. #### Represented Child Runs `parallel` first expands item slots, then resolves the child DAG target and child DAG params for each item slot. Rules: - A represented child run is identified by the resolved child DAG target and resolved child DAG params. - If more than one item slot resolves to the same child DAG target and the same child DAG params, Dagu represents those item slots as one child run. - Parent step child-run lists and `parallel` aggregate counts count represented child runs, not raw expanded item slots. - One child run per item slot requires distinct resolved child DAG params for each item slot. #### `dag.run` Semantics When `parallel` is used with `action: dag.run`, the parent step starts child DAG runs and waits for their terminal status. Rules: - If every represented child DAG run is `succeeded`, the parent step is `succeeded`. - If one or more represented child DAG runs fail, the parent step fails after the fan-out reaches a terminal state. - If no represented child DAG run fails, aborts, is rejected, or ends in another non-success terminal status, and at least one represented child DAG run is `partially_succeeded`, the parent step is `partially_succeeded`. - If one or more represented child DAG runs wait for external approval or input, the parent step waits. - Aborting the parent step stops launching pending item runs and asks active child DAG runs to abort. - A parent step timeout is handled as a parent step failure or abort according to the step timeout contract. - Parent step retry retries the fan-out step. This spec does not define per-item retry of only failed items. - The step-level `pass_env` field applies to every represented child run. Each child run receives the same requested values, evaluated once when the child run parameters are resolved. See Spec 006, "Sub-DAG Passed Environment". #### `dag.enqueue` Semantics When `parallel` is used with `action: dag.enqueue`, the parent step creates or finds queued child DAG runs and does not wait for those child runs to execute. Rules: - The parent step succeeds when every represented child enqueue request is accepted or resolves to an existing child run. - The parent step fails when any represented child enqueue request fails. - `with.queue` selects the queue for all item runs in that step. - `max_concurrent` does not control later queue processing. Queue processing is owned by the queue configuration. - `pass_env` is not supported for `dag.enqueue`; queued child runs cannot carry transient parent environment values through queue persistence. #### Aggregate Outputs When a `parallel` step writes an aggregate output, the payload is JSON. For `action: dag.run`, the payload object has these required fields: | Field | Meaning | | --- | --- | | `summary.total` | Number of represented child DAG runs in the aggregate. | | `summary.succeeded` | Number of represented child DAG runs counted as successful. | | `summary.failed` | Number of represented child DAG runs not counted as successful. | | `results` | Child run result objects. | | `outputs` | Output variable maps from successful child DAG runs. | Rules: - `summary.succeeded` includes child DAG runs that are `succeeded` or `partially_succeeded`. - `outputs` contains only child DAG runs counted as successful. - A failed child DAG run contributes to `summary.failed` and must not contribute an output map to `outputs`. - `results` follows `parallel.items` order. A child DAG run that more than one item produces takes the position of the first of those items, so duplicate coalescing drops entries without reordering the ones that remain. - `outputs` follows the `results` order with the entries for child DAG runs not counted as successful removed, so `outputs[N]` addresses the Nth successful child DAG run rather than the Nth item. - A finished `parallel` step also publishes its collected child outputs on the step outputs channel as a JSON array of per-child output maps, so `${step.outputs}` resolves after the step finishes. The array follows the `outputs` ordering rule above, and a successful child DAG run that published nothing contributes an empty object. When no child DAG run is counted as successful, the step publishes nothing on the channel. - An entry in the published array merges the child run's output variables with its declared outputs, so it carries more than the matching `outputs` entry in the aggregate payload, which carries output variables alone. - The published array is readable through `${step.outputs}` and `${step.outputs[N].NAME}` within the DAG that declares the step. It does not merge into the run's collected outputs or into a parent run's `${step.outputs}` map, because those carry name and value pairs and the array has no names to merge under. For a `parallel` step using `action: dag.enqueue`, the payload object has these required fields, including when expansion or duplicate coalescing leaves exactly one represented enqueue request: | Field | Meaning | | --- | --- | | `summary.total` | Number of represented enqueue requests. | | `summary.queued` | Number of child runs newly queued by this step. | | `runs` | Per-child enqueue result objects. | Each `runs` entry must include: | Field | Meaning | | --- | --- | | `name` | Child DAG name. | | `dagRunId` | Child DAG-run id. | | `queue` | Queue selected for that child run. | | `status` | Status observed after the enqueue request. | Each `runs` entry may include: | Field | Meaning | | --- | --- | | `params` | Child parameter payload. | | `alreadyExists` | True when the child run already existed. | ### Foreach #### Field Shape `foreach` is an optional step field. Rules: - A step with `foreach` must not define another execution selector, such as `run`, `action`, `command`, `script`, `call`, or `parallel`. - `foreach` must be an object. - `foreach.items` is required. - `foreach.steps` is required. - `foreach.steps` must be a non-empty sequence of step objects. - `foreach.as` is optional and defaults to `item`. - `foreach.as` must match `^[A-Za-z][A-Za-z0-9_]*$`. - `foreach.as` must not be `index` or `key`. - `foreach.key` is optional. - `foreach.max_concurrent` is optional and follows the common concurrency rules above. - `foreach.collect` is optional. - `foreach` must not contain fields other than `items`, `as`, `key`, `max_concurrent`, `steps`, and `collect`. Example shape: ```yaml steps: - id: summarize_episodes foreach: items: ${steps.list_episodes.outputs.episodes} as: episode key: ${foreach.episode.slug} max_concurrent: 3 steps: - id: summarize run: ./summarize.sh ${foreach.episode.url} outputs: - name: markdown collect: markdown: ${steps.summarize.outputs.markdown} output: EPISODE_SUMMARIES ``` #### Item Sources Rules: - `foreach.items` may be a YAML array. - `foreach.items` may be a string that resolves to a JSON array. - A non-array JSON value is invalid for `foreach.items`. - A non-JSON resolved string is invalid for `foreach.items`. - An empty array is valid and produces zero item body executions. - A `foreach` expansion must not produce more than `1000` items. - Static YAML array items may be any JSON-compatible value. #### Item Keys Rules: - If `foreach.key` is omitted, the item key is the decimal item slot index. - If `foreach.key` is present, Dagu resolves it once per item with item scope available. - The resolved item key must be a non-empty string. - Item keys must be unique inside one `foreach` step. - Duplicate item keys fail the `foreach` step before any item body starts. #### Item Scope For each item body execution, Dagu exposes structured item references. Rules: - `${foreach.index}` resolves to the zero-based item slot index as decimal text. - `${foreach.key}` resolves to the item key. - `${foreach.}` resolves to the current item. - When `as` is omitted, `${foreach.item}` resolves to the current item. - For a scalar item, `${foreach.}` resolves to that scalar as text. - For an object or array item, `${foreach.}` resolves to compact JSON. - `${foreach..field}` resolves to a top-level object field. - A field name that starts with a letter or underscore and continues with letters, marks, digits, or underscores, in any script, may be written unquoted. Any other field name is written as `${foreach.."name"}` or `${foreach.["name"]}`. - Nested item field paths such as `${foreach.item.a.b}` are outside this spec. - Item scope is available to `foreach.key`. - Item scope is available to every value-resolved string field inside `foreach.steps`. - Item scope is available to every `foreach.collect` value. - The uppercase `${ITEM}` item reference is owned by `parallel`. It is not a `foreach` item reference. - `${.field}` without the `foreach.` namespace is not an item reference. When a field with item scope is evaluated, an unescaped `${.field}` whose field may be written unquoted fails that item unless another source resolves it: an environment value named `` holding a JSON object with that field, from any source the field reads (root `env`, dotenv, params, step `env`, an output variable, or the process environment where the field reads it), or a step property such as `${.stdout}` of a step whose id is ``. Validation cannot decide this, because run-time params, dotenv files, and the process environment may define ``. `${}` and `$.field` are not affected, so shell variables keep their meaning. #### Body Step Scope `foreach.steps` defines steps scoped to one item body. Rules: - Body steps inherit the DAG's step `defaults` as top-level steps do. A field set on a body step overrides the inherited default, except `env` and `preconditions`, which add to the inherited entries. - Body step ids must be unique inside the body. - Body step names must be unique inside the body after loading. - Body step ids and names must not collide with top-level step ids or names visible to the `foreach` step. - Body step dependencies stay inside the body. - Body steps must not depend on top-level steps directly. - Top-level steps must not depend on body steps directly. - A body step can reference top-level step outputs. The reference adds an inferred dependency from the producer to the owning `foreach` step under Spec 007, so the producer is ordered before the body runs. - A body step can reference an earlier body step output from the same item body using normal step-output reference syntax. Such a reference creates no inferred dependency; body ordering comes from body `depends`. - Body step output references never cross item bodies. - Top-level steps after the `foreach` step cannot reference body step outputs directly. They must consume the `foreach` aggregate output. #### Collect `foreach.collect` defines output values copied from each successful item body into the parent aggregate output. Rules: - `foreach.collect` must be a mapping from output names to string expressions. - Collect output names must match `^[A-Za-z][A-Za-z0-9_]*$`. - Collect output names must be unique inside one `foreach` step. - Collect expressions are value-resolved after an item body succeeds. - Collect expressions are resolved with both item scope and that item body's step-output scope available. - A collect expression that still contains a supported unresolved reference after item body success fails that item. - Unsupported braced text follows Spec 003 preservation behavior. #### Execution Semantics Rules: - A `foreach` step with zero items succeeds without running body steps. - A `foreach` step with one or more items starts at most `max_concurrent` item bodies at a time. - Each item body runs the same `foreach.steps` graph with that item's scope. - Item body execution order is not guaranteed. - The parent `foreach` step does not finish until every item body has reached a terminal status, unless the parent step is aborted or times out. - If every item body succeeds, the parent `foreach` step succeeds. - If one or more item bodies fail and at least one succeeds, the parent `foreach` step is `partially_succeeded` once all item bodies that can run have reached a terminal status: its dependents run, and the DAG run ends partially succeeded. This differs from `parallel`, where a failed child DAG run fails the parent step, because a foreach body is the work of one item and the other items' results remain usable. - If every item body fails, the parent `foreach` step fails after all item bodies that can run have reached a terminal status. - A failed item body does not prevent later item bodies from starting unless the parent step is aborted or times out. - Aborting the parent `foreach` step stops launching pending item bodies and asks active item bodies to abort. - Parent step retry retries the entire `foreach` step. This spec does not define per-item retry of only failed item bodies. #### Aggregate Outputs When a `foreach` step defines `output`, the captured value is a JSON object with these required fields: | Field | Meaning | | --- | --- | | `summary.total` | Number of expanded items. | | `summary.succeeded` | Number of item bodies that succeeded. | | `summary.failed` | Number of item bodies that failed. | | `items` | Per-item status records. | | `outputs` | Collect maps from successful item bodies. | Each `items` entry must include: | Field | Meaning | | --- | --- | | `index` | Zero-based item slot index. | | `key` | Item key. | | `status` | Item body status. | Each successful `items` entry must include `outputs`. Each failed `items` entry must include `error` when an error message is available. Rules: - `items` is ordered by item slot index. - `outputs` is ordered by item slot index and includes only successful item bodies. - The aggregate is written whether or not every item body succeeded: failed item bodies appear in `items` with `error`, and only successful ones contribute to `outputs`. A dependent step reads it through the string-form `output` variable (Spec 012), which is set even when the parent step fails. - If `foreach.collect` is omitted, successful item output maps are empty. - Dagu must not include the raw item value in the aggregate output. ## Errors The errors below define conformance behavior. ### Parallel Errors Validation must fail when: - `parallel` appears on a step whose action is not `dag.run` or `dag.enqueue`. - `parallel` and `foreach` appear on the same step. - `parallel` has an invalid shape. - object-form `parallel` omits `items`. - object-form `parallel` contains an unknown field. - `max_concurrent` is not an integer from `1` through `1000`. - a static array item is a nested array. - a static array item is a nested mapping. - a static mapping item contains a value that is not a string, number, or boolean. - `pass_env` appears on a `dag.enqueue` step (with or without `parallel`). Runtime execution must fail when: - value resolution for an item source fails. - value resolution for a child DAG target or child params fails. - item expansion produces zero items. - item expansion produces more than `1000` items. - an expanded string item source cannot be parsed according to the `parallel` string item source rules. - a child DAG target cannot be loaded. - a `dag.run` child DAG run fails. - a `dag.enqueue` child enqueue request fails. ### Foreach Errors Validation must fail when: - `foreach` appears with another execution selector. - `foreach` has an invalid shape. - `foreach.items` is missing. - `foreach.steps` is missing or empty. - `foreach.as` is invalid. - `foreach.max_concurrent` is not an integer from `1` through `1000`. - `foreach.collect` is not a mapping from valid output names to strings. - a body step identity collides with another body step identity. - a body step identity collides with a visible top-level step identity. - a body step dependency names a top-level step. - a top-level step dependency names a body step. Runtime execution must fail when: - value resolution for `foreach.items` fails. - a string `foreach.items` value does not resolve to a JSON array. - item expansion produces more than `1000` items. - `foreach.key` resolves to an empty string. - two item slots resolve to the same item key. - a field with item scope keeps `${.field}` without the `foreach.` namespace, as the Item Scope section defines; that item fails. - every item body fails; a collect expression that fails to resolve after its item body succeeded counts as that item's failure. When some item bodies fail and others succeed, the step is `partially_succeeded`, as the Behavior section states. ### Timeout and Abort Rules: - Timeout and abort must not start new pending item runs or item bodies. - Active child DAG runs or item bodies must receive the same stop signal as a normal running step in the same lifecycle state. - A timed-out or aborted expansion step must not be reported as succeeded. - Partial aggregate output on timeout or abort is allowed only when the output clearly reports that not all items succeeded. ## Examples ### Parallel Child DAG Run ```yaml steps: - id: fanout_accounts action: dag.run with: dag: workflows/process-account params: account_id: ${ITEM.account_id} region: ${ITEM.region} parallel: max_concurrent: 2 items: - account_id: acct_1 region: us-east-1 - account_id: acct_2 region: eu-west-1 output: ACCOUNT_RESULTS ``` Expected behavior: - Two child DAG runs are represented. - Each child receives item-specific params. - The parent waits for both child DAG runs. - `ACCOUNT_RESULTS` is JSON with `summary`, `results`, and `outputs`. ### Parallel Child Enqueue ```yaml steps: - id: enqueue_accounts action: dag.enqueue with: dag: workflows/process-account queue: background params: account_id: ${ITEM.account_id} parallel: items: - account_id: acct_1 - account_id: acct_2 output: ENQUEUE_RESULTS ``` Expected behavior: - Two child enqueue requests are represented. - The parent succeeds after both enqueue requests are accepted or found. - The parent does not wait for queued child DAG execution. - `ENQUEUE_RESULTS` is JSON with `summary` and `runs`. ### Foreach Inline Body ```yaml steps: - id: list_episodes run: ./list-episodes.sh outputs: - name: episodes type: json - id: summarize_episodes depends: list_episodes foreach: items: ${steps.list_episodes.outputs.episodes} as: episode key: ${foreach.episode.slug} max_concurrent: 3 steps: - id: fetch_transcript run: ./fetch-transcript.sh ${foreach.episode.url} outputs: - name: transcript_path - id: summarize depends: fetch_transcript run: ./summarize-ja.sh ${steps.fetch_transcript.outputs.transcript_path} outputs: - name: markdown collect: markdown: ${steps.summarize.outputs.markdown} output: EPISODE_SUMMARIES ``` Expected behavior: - The `episodes` JSON array controls the number of item bodies. - Each item body sees `${foreach.episode.*}` for its own item only. - Each item body can pass data between body steps through normal step-output references. - The top-level workflow consumes body results through `EPISODE_SUMMARIES`. ### Foreach Empty Input ```yaml steps: - id: summarize_episodes foreach: items: [] steps: - id: summarize run: ./summarize.sh ${foreach.item} output: EPISODE_SUMMARIES ``` Expected behavior: - No body step runs. - The `summarize_episodes` step succeeds. - `EPISODE_SUMMARIES.summary.total` is `0`. ### Invalid Parallel Action ```yaml steps: - id: invalid action: http.request with: url: https://example.com parallel: items: [a, b] ``` Expected behavior: - Validation fails because `parallel` is only valid with `dag.run` or `dag.enqueue`.