Proposal: `for` step type for basic loop support

[tbrandenburg]

Motivation

Workflows often need to iterate over a list of items, such as processing multiple files, running agent tasks per item, or collecting results. Today users have to inline bash for or while loops inside a single bash step, which breaks the flowsh step model and makes it impossible to use per-item vars, agent steps, or structured logging.

Proposed Design: for wrapper step type

Add a new for step type that wraps a sequence of child steps and iterates over a variable from a vars step.

YAML shape

steps:
  - type: vars
    name: Collect items
    values:
      ITEMS: |
        printf "file-a\nfile-b\nfile-c"

  - type: for
    name: Process each item
    in: ITEMS
    item: ITEM
    steps:
      - type: bash
        name: Process
        run: |
          echo "Processing $ITEM"

      - type: agent
        name: Review
        expandPrompt: true
        prompt: |
          Review the file $ITEM and suggest improvements.

How it generates

Inner steps are rendered as top-level bash functions (once, outside the loop), then called via run_step inside a while read loop:

# Inner step -- defined once outside the loop
for_0_process() {
  bash -euo pipefail <<'BASH_EOF'
echo "Processing $ITEM"
BASH_EOF
}

# Loop step
step_process_for_item() {
  while IFS= read -r ITEM; do
    export ITEM
    run_step for_0_process
    run_step for_0_review
  done <<< "$ITEMS"
}
run_stateful_step step_process_for_item

This gives every inner step:

• Per-iteration logging via run_step
• Failure catching via catch() with the correct exit code
• Dry-run support
• State preservation via export ITEM

Models change

Add ForStep to models.py:

class ForStep(BaseStep):
    type: Literal["for"]
    in_: str = Field(..., validation_alias="in")
    item: str = Field(...)
    steps: list[Step] = Field(min_length=1)

Add to Step union.

Renderer change

In render.py:render_step, add a new elif isinstance(step, ForStep) branch:

1. Emit each inner step as its own function using the existing step_function_name and used_function_names for unique names.
2. Emit the loop function body with run_step <name> calls inside while IFS= read -r ....
3. Use run_stateful_step for the outer for function.

Accepted restrictions / downsides

• Items are newline-delimited
• Loop variable must match [A-Z_][A-Z0-9_]*
• agent inner steps need expandPrompt: true to see the loop variable
• No nested for loops in v1
• Flat function namespace

Decisions

• Inner steps support vars, bash, and agent.
• The loop variable is auto-exported.
• Parallel execution is out of scope here and stays in Support declarative parallel steps in workflow definitions #25.

[tbrandenburg]

Investigation: Proposal: for step type for basic loop support

Issue: #21 (#21)
Type: ENHANCEMENT
Investigated: 2026-06-03T00:00:00Z

Assessment

Metric	Value	Reasoning
Priority	MEDIUM	Valuable ergonomic improvement enabling per-item agent steps, but inline bash loops work today
Complexity	MEDIUM	3 files touched (models.py, render.py, tests), recursive type and inner-function naming logic
Confidence	HIGH	Issue is a fully-specified proposal; codebase is small and well-understood

Problem Statement

Workflows that need to iterate over a list of items (files, issue numbers, etc.) must today inline a bash for/while loop in a single bash step. This makes it impossible to use structured vars, agent, or bash steps per iteration, and loses per-step logging, dry-run support, and failure catching. The proposal adds a first-class for step type that wraps a list of child steps and iterates them over a newline-delimited variable.

Analysis

Change Rationale

The for step type does not fix a bug — it extends the step model to support iteration. The generating approach chosen (inner steps as top-level bash functions called inside a while IFS= read -r loop) preserves all existing runtime guarantees: run_step logging, catch() on failure, dry-run, and variable export.

Affected Files

File	Lines	Action	Description
src/flowsh_cli/models.py	157	UPDATE	Add ForStep model; update Step union; call model_rebuild()
src/flowsh_cli/render.py	6, 221-296	UPDATE	Import ForStep; refactor body to _render_step_body; add ForStep branch
tests/test_workflow_to_harness.py	NEW tests	UPDATE	Add parse, render, and E2E execution tests for for steps

Integration Points

• models.py:157 — Step union is the discriminated type used by Workflow.steps and now also by ForStep.steps (recursive)
• render.py:208-209 — render_harness iterates workflow.steps and calls render_step; no change needed there
• render.py:284 — runner selection logic (run_stateful_step vs run_step) must include ForStep
• render.py:6 — import line must add ForStep
• render.py:289-296 — default_step_title must handle ForStep

Git History

• Last relevant change: ee97fb5 - feat: support named positional parameters (recent, codebase stable)
• Implication: No in-progress refactor; clean baseline for adding the new type

Implementation Plan

Step 1: Add ForStep to models.py

File: src/flowsh_cli/models.py
Lines: 157 (insert after VarsStep), end of file for model_rebuild()
Action: UPDATE

Current code at line 157:

Step = Annotated[VarsStep | BashStep | AgentStep, Field(discriminator="type")]

Required change — insert ForStep class before line 157, then update Step union, then call model_rebuild():

# Insert BEFORE the Step type alias (after VarsStep at line 155):

class ForStep(BaseStep):
    type: Literal["for"]
    in_: str = Field(..., validation_alias="in")
    item: str = Field(...)
    steps: list["Step"] = Field(min_length=1)

    @field_validator("in_", "item")
    @classmethod
    def validate_var_name(cls, value: str) -> str:
        if not re.fullmatch(r"[A-Z_][A-Z0-9_]*", value):
            raise ValueError("must match ^[A-Z_][A-Z0-9_]*$")
        return value


# Update the Step union to include ForStep:
Step = Annotated[VarsStep | BashStep | AgentStep | ForStep, Field(discriminator="type")]

Then at the end of models.py (after the UniqueKeySafeLoader constructor setup and all helpers) add:

# Resolve the forward reference in ForStep.steps
ForStep.model_rebuild()

Why: Pydantic v2 requires model_rebuild() when a model contains a forward reference to a type that was not fully defined at class definition time. ForStep.steps: list["Step"] uses the string "Step" because Step is defined after ForStep. Calling model_rebuild() after Step is redefined resolves it.

---

Step 2: Refactor render.py — extract _render_step_body helper

File: src/flowsh_cli/render.py
Lines: 6, 221-286
Action: UPDATE

Current import line 6:

from flowsh_cli.models import AgentStep, BashStep, Step, VarsStep, Workflow, WorkflowParam

Change to:

from flowsh_cli.models import AgentStep, BashStep, ForStep, Step, VarsStep, Workflow, WorkflowParam

Current render_step body (lines 221-286) — extract a _render_step_body helper:

Replace the body of render_step with a call to a new helper, and add the helper:

def render_step(index: int, step: Step, used_function_names: set[str] | None = None) -> list[str]:
    function_name = step_function_name(index, step.name, used_function_names)
    title = step.name or default_step_title(index, step)

    prefix_lines: list[str] = []  # inner function definitions (for ForStep)

    if isinstance(step, ForStep):
        inner_fns: list[str] = []
        for i, inner_step in enumerate(step.steps, start=1):
            inner_fn = _for_inner_function_name(index, i, inner_step, used_function_names)
            inner_fns.append(inner_fn)
            inner_title = inner_step.name or default_step_title(i, inner_step)
            prefix_lines.append(section(f"For-inner step ({inner_step.type}): {inner_title}"))
            prefix_lines.append(f"{inner_fn}() {{")
            prefix_lines.extend(_render_step_body(inner_step))
            prefix_lines.append("}")
            prefix_lines.append("")

        body_lines = [
            f"  while IFS= read -r {step.item}; do",
            f"    export {step.item}",
            *[f"    run_step {fn}" for fn in inner_fns],
            f"  done <<< \"${{{step.in_}}}\"",
        ]
    else:
        body_lines = _render_step_body(step)

    runner = "run_stateful_step" if isinstance(step, (VarsStep, ForStep)) else "run_step"
    outer_lines = [
        section(f"Step {index} ({step.type}): {title}"),
        f"{function_name}() {{",
        *body_lines,
        "}",
        f"{runner} {function_name}",
        "",
    ]
    return prefix_lines + outer_lines


def _render_step_body(step: Step) -> list[str]:
    """Return indented body lines for a step function (no wrapper, no run_step call)."""
    lines: list[str] = []
    if isinstance(step, VarsStep):
        lines.append("  local status=0")
        for name, command in step.values.items():
            delimiter = heredoc_delimiter(f"VARS_{name}", command)
            lines.append(f"  {name}=$(bash -euo pipefail <<'{delimiter}'")
            lines.extend(script_lines(command))
            lines.extend(
                [
                    delimiter,
                    "  )",
                    "  status=$?",
                    "  if [[ $status -ne 0 ]]; then",
                    '    return "$status"',
                    "  fi",
                    f"  export {name}",
                ]
            )
    elif isinstance(step, BashStep):
        delimiter = heredoc_delimiter("BASH", step.run)
        lines.extend(
            [
                f"  bash -euo pipefail <<'{delimiter}'",
                *step.run.strip().splitlines(),
                delimiter,
            ]
        )
    elif isinstance(step, AgentStep):
        delimiter = heredoc_delimiter("PROMPT", step.prompt)
        lines.extend(
            [
                "  local prompt",
                f"  prompt=$(cat <<'{delimiter}'",
                *step.prompt.splitlines(),
                delimiter,
                "  )",
            ]
        )
        if step.expandPrompt:
            braced = re.findall(r"\$\{([A-Z_][A-Z0-9_]*)\}", step.prompt)
            bare = re.findall(r"\$([A-Z_][A-Z0-9_]*)(?!\w)", step.prompt)
            seen: dict[str, None] = {}
            for var in braced + bare:
                seen[var] = None
            for var in seen:
                lines.append(f'  _p=\'${{{var}}}\'; prompt="${{prompt//"$_p"/"${var}"}}"')
                lines.append(f'  _p=\'${var}\'; prompt="${{prompt//"$_p"/"${var}"}}"')
        lines.append(f"  local agent={bash_quote(step.agent or '')}")
        lines.append(f"  local model={bash_quote(step.model or '')}")
        lines.append(f"  local command={bash_quote(step.command or '')}")
        dangerous_skip_permissions = "true" if step.dangerouslySkipPermissions else "false"
        lines.append(f"  local dangerously_skip_permissions={dangerous_skip_permissions}")
        lines.append(
            '  run_agent "$prompt" "$agent" "$model" "$command" "$dangerously_skip_permissions"'
        )
    else:
        raise AssertionError(f"Unsupported step type: {step}")
    return lines


def _for_inner_function_name(
    outer_index: int,
    inner_index: int,
    step: Step,
    used_function_names: set[str] | None,
) -> str:
    source = step.name or f"step_{inner_index}"
    slug = re.sub(r"[^A-Za-z0-9_]+", "_", source).strip("_").lower()
    if not slug or slug[0].isdigit():
        slug = f"step_{slug}" if slug else f"step_{inner_index}"
    base = f"for_{outer_index}_{slug}"
    fn = base
    suffix = 2
    while used_function_names is not None and fn in used_function_names:
        fn = f"{base}_{suffix}"
        suffix += 1
    if used_function_names is not None:
        used_function_names.add(fn)
    return fn

Also update default_step_title (lines 289-296) to handle ForStep:

def default_step_title(index: int, step: Step) -> str:
    if isinstance(step, VarsStep):
        return ", ".join(step.values.keys())
    if isinstance(step, BashStep):
        return truncate_one_line(step.run)
    if isinstance(step, AgentStep):
        return truncate_one_line(step.prompt)
    if isinstance(step, ForStep):
        return f"for {step.item} in {step.in_}"
    return f"step {index}"

Why: The existing render_step body logic is replicated verbatim by extracting it into _render_step_body. This avoids duplicating the three step-type branches while keeping the inner step rendering consistent with top-level steps. The ForStep branch generates inner function names with a for_{outer_index}_{slug} pattern to avoid collision with top-level step names and to remain human-readable.

---

Step 3: Add/Update Tests

File: tests/test_workflow_to_harness.py
Action: UPDATE

Test cases to add:

# --- ForStep model parsing ---

def test_parse_workflows_accepts_for_step(tmp_path: Path) -> None:
    workflow_file = tmp_path / "workflows.yml"
    workflow_file.write_text(
        """
workflows:
  - id: wf_for_basic
    name: For Basic
    steps:
      - type: vars
        name: Collect items
        values:
          ITEMS: printf 'file-a\\nfile-b'
      - type: for
        name: Process each item
        in: ITEMS
        item: ITEM
        steps:
          - type: bash
            name: Process
            run: |
              echo "Processing $ITEM"
""".lstrip(),
        encoding="utf-8",
    )
    from flowsh_cli.models import ForStep

    workflows = parse_workflows(workflow_file)
    for_step = workflows[0].steps[1]

    assert isinstance(for_step, ForStep)
    assert for_step.in_ == "ITEMS"
    assert for_step.item == "ITEM"
    assert len(for_step.steps) == 1
    assert isinstance(for_step.steps[0], BashStep)


def test_parse_workflows_rejects_for_step_with_invalid_item_name(tmp_path: Path) -> None:
    workflow_file = tmp_path / "workflows.yml"
    workflow_file.write_text(
        """
workflows:
  - id: wf_for_bad_item
    name: For Bad Item
    steps:
      - type: for
        in: ITEMS
        item: bad-item
        steps:
          - type: bash
            run: echo "$bad_item"
""".lstrip(),
        encoding="utf-8",
    )

    with pytest.raises(WorkflowParseError, match="item"):
        parse_workflows(workflow_file)


def test_parse_workflows_rejects_for_step_with_empty_steps(tmp_path: Path) -> None:
    workflow_file = tmp_path / "workflows.yml"
    workflow_file.write_text(
        """
workflows:
  - id: wf_for_empty
    name: For Empty
    steps:
      - type: for
        in: ITEMS
        item: ITEM
        steps: []
""".lstrip(),
        encoding="utf-8",
    )

    with pytest.raises(WorkflowParseError):
        parse_workflows(workflow_file)


# --- ForStep render ---

def test_render_harness_for_step_emits_inner_functions_and_loop() -> None:
    from flowsh_cli.models import ForStep

    workflow = Workflow(
        id="wf_for_render",
        name="For Render",
        steps=[
            ForStep(
                type="for",
                name="Process items",
                **{"in": "ITEMS"},
                item="ITEM",
                steps=[BashStep(type="bash", name="Process", run='echo "$ITEM"')],
            )
        ],
    )

    script = render_harness(workflow)

    assert "for_1_process() {" in script
    assert "while IFS= read -r ITEM; do" in script
    assert "export ITEM" in script
    assert "run_step for_1_process" in script
    assert 'done <<< "$ITEMS"' in script
    assert "run_stateful_step step_process_items" in script


# --- ForStep end-to-end execution ---

def test_generated_harness_runs_for_step_over_vars_items(tmp_path: Path) -> None:
    workflow_file = tmp_path / "workflows.yml"
    workflow_file.write_text(
        """
workflows:
  - id: wf_for_e2e
    name: For E2E
    steps:
      - type: vars
        name: Collect items
        values:
          ITEMS: printf 'alpha\\nbeta\\ngamma'
      - type: for
        name: Process each item
        in: ITEMS
        item: ITEM
        steps:
          - type: bash
            name: Emit item
            run: |
              printf 'item:%s\\n' "$ITEM"
""".lstrip(),
        encoding="utf-8",
    )
    generated = subprocess.run(
        [sys.executable, "-m", "flowsh_cli", str(workflow_file)],
        check=False,
        capture_output=True,
        text=True,
        cwd=tmp_path,
    )

    assert generated.returncode == 0, generated.stderr
    harness = tmp_path / ".harness" / "for_e2e.sh"
    assert harness.exists()

    executed = subprocess.run(
        ["bash", str(harness)],
        check=False,
        capture_output=True,
        text=True,
        cwd=tmp_path,
        env={**os.environ, "FLOWSH_LOG_DIR": "logs"},
    )

    assert executed.returncode == 0, executed.stderr
    assert "item:alpha" in executed.stdout
    assert "item:beta" in executed.stdout
    assert "item:gamma" in executed.stdout


def test_generated_harness_for_step_dry_run_skips_inner_steps(tmp_path: Path) -> None:
    workflow_file = tmp_path / "workflows.yml"
    workflow_file.write_text(
        """
workflows:
  - id: wf_for_dry
    name: For Dry
    steps:
      - type: vars
        name: Collect items
        values:
          ITEMS: printf 'x\\ny'
      - type: for
        name: Process each item
        in: ITEMS
        item: ITEM
        steps:
          - type: bash
            name: Emit
            run: printf 'should-not-appear\\n'
""".lstrip(),
        encoding="utf-8",
    )
    generated = subprocess.run(
        [sys.executable, "-m", "flowsh_cli", str(workflow_file)],
        check=False,
        capture_output=True,
        text=True,
        cwd=tmp_path,
    )

    assert generated.returncode == 0, generated.stderr
    executed = subprocess.run(
        ["bash", str(tmp_path / ".harness" / "for_dry.sh"), "--dry-run"],
        check=False,
        capture_output=True,
        text=True,
        cwd=tmp_path,
    )

    assert executed.returncode == 0, executed.stderr
    assert "should-not-appear" not in executed.stdout
    assert "[DRY-RUN]" in executed.stderr

Patterns to Follow

From codebase — Step type union with discriminator:

# SOURCE: src/flowsh_cli/models.py:157
# Pattern for Pydantic discriminated union step types
Step = Annotated[VarsStep | BashStep | AgentStep, Field(discriminator="type")]

From codebase — VarsStep uses run_stateful_step (state must survive subshell boundary):

# SOURCE: src/flowsh_cli/render.py:284
runner = "run_stateful_step" if isinstance(step, VarsStep) else "run_step"

ForStep must also use run_stateful_step because the while loop must run in the main shell to export ITEM into scope.

From codebase — step_function_name deduplication pattern:

# SOURCE: src/flowsh_cli/render.py:306-330
# Unique name generation: base + suffix counter, tracked in used_function_names set
function_name = base
suffix = 2
while function_name in used_function_names:
    function_name = f"{base}_{suffix}"
    suffix += 1
used_function_names.add(function_name)

Mirror this for _for_inner_function_name.

Edge Cases & Risks

Risk/Edge Case	Mitigation
Recursive ForStep (nested for)	Pydantic will parse it (Step union includes ForStep), but render will AssertionError — add explicit guard in _render_step_body with raise AssertionError("nested for not supported")
Inner step name collides with outer step	used_function_names is shared; for_{outer_index}_ prefix avoids collision in practice
in_ var not defined at runtime	Bash while read on empty string produces zero iterations — silent, correct behaviour
item name shadows an exported env var	Documented restriction — user responsibility; same as any vars step
ITEMS contains trailing newline	while IFS= read -r will produce one empty-string iteration; acceptable bash behaviour
ForStep with VarsStep inner steps	VarsStep normally uses run_stateful_step; inside the loop it must use run_step (the loop function already runs in main shell via run_stateful_step) — _render_step_body does not emit the runner call, only the loop calls run_step for all inner steps

Validation

Automated Checks

make qa
# which runs:
uv run ruff check src tests
uv run ruff format --check src tests
uv run pytest tests/

Manual Verification

1. Write a workflows.yml with a for step using vars, bash, and agent inner steps
2. Run flowsh-cli workflows.yml and confirm .harness/*.sh is generated without errors
3. Run bash .harness/<name>.sh --dry-run and confirm inner steps appear as [DRY-RUN] in stderr
4. Run bash .harness/<name>.sh (with a fake opencode if needed) and confirm each item is processed
5. Confirm bash -n .harness/<name>.sh exits 0 (syntax check)

Scope Boundaries

IN SCOPE:

• ForStep model with in_, item, steps fields
• Validation: in_ and item must match [A-Z_][A-Z0-9_]*; steps must be non-empty
• Inner steps support vars, bash, and agent (the existing three types)
• Rendered inner functions use for_{outer_index}_{slug} naming
• Loop uses while IFS= read -r; outer function uses run_stateful_step
• Tests: parse accept/reject, render structure, E2E execution, dry-run

OUT OF SCOPE (do not touch):

• Nested for loops (issue explicitly defers this)
• Parallel iteration (issue references Support declarative parallel steps in workflow definitions #25 for that)
• in accepting a command expression directly (only variable names)
• Changes to cli.py, __init__.py, or __main__.py

Metadata

• Investigated by: issue-resolution-workflow
• Timestamp: 2026-06-03T00:00:00Z
• Artifact: .agents/issues/issue-21.md