Support declarative parallel steps in workflow definitions

[tbrandenburg]

Problem

We want a simple way to express parallel execution in workflows without introducing a new execution engine. The generated artifacts must remain plain bash and keep the current vars / bash / agent model intact.

Proposed schema

Add one new step type: parallel. It is a wrapper step that contains ordinary child steps.

workflows:
  - id: wf_example
    name: Example
    steps:
      - type: parallel
        name: Fan out
        steps:
          - type: bash
            name: Build
            run: make build
          - type: bash
            name: Test
            run: make test
          - type: agent
            name: Review
            prompt: Review the build output.

Minimal constraints:

• parallel.steps should accept the same step types supported today: vars, bash, agent.
• Nested parallel can be deferred unless there is a strong need.
• The wrapper is declarative only; no new workflow runtime beyond generated bash.

Minimal renderer change

Keep the current per-step function generation. For a parallel step:

• Render each child step as a normal function, reusing the existing step renderer.
• Generate a wrapper function that starts each child in the background with "$fn" &.
• Capture PIDs with "$!", then wait for each PID.
• Return nonzero if any child fails.

Example bash shape:

parallel_step() {
  child_a &
  pid_a=$!
  child_b &
  pid_b=$!

  status=0
  wait "$pid_a" || status=$?
  wait "$pid_b" || status=$?
  return "$status"
}

Known tradeoffs

• Branches do not share mutable shell state.
• vars inside one branch do not affect sibling branches or the parent.
• Output from concurrent branches will interleave.
• Failure handling stays simple: collect exit codes, do not attempt cancellation.

Acceptance criteria

• Workflow YAML can declare a parallel wrapper step.
• Generated harnesses remain plain bash.
• Existing vars / bash / agent workflows continue to work unchanged.
• The implementation stays minimal and readable.

[github-actions]

Hello tbrandenburg!

Thanks for your feedback!

Here are the tools I am offering:

Commands:
/ghar-commit-push
/ghar-issue-fix
/ghar-issue-investigate
/ghar-maintainability-review
/ghar-resolve-ci-errors
/ghar-review
/ghar-security-check
/ghar-stale-check

Agents:
*review

Usage:

• Use commands directly: /command-name
• Use agents with commands: *agent-name /command-name PROMPT

Example: *review /security-review Please check this code for vulnerabilities

[github-actions]

Investigation: Support declarative parallel steps in workflow definitions

Issue: #25 (#25)
Type: ENHANCEMENT
Investigated: 2026-06-02T21:11:00Z

Assessment

Metric	Value	Reasoning
Priority	MEDIUM	Non-blocking feature that adds parallelism to workflows; existing sequential workflows remain fully functional and unaffected
Complexity	LOW	~3 files affected, isolated additions to model + renderer + tests, no architectural changes, reuses 100% of existing step render
Confidence	HIGH	Clear spec in issue, simple fork-join bash pattern, no integration with external systems, no risk to existing functionality

---

Problem Statement

Workflows today run steps strictly sequentially. Users need a way to express that independent steps (e.g., build and test) should run concurrently to reduce total wall-clock time, without introducing a new execution engine or changing the vars/bash/agent step model. The solution must generate plain bash that uses & + $! + wait for the fork-join pattern.

---

Analysis

Change Rationale

A parallel step type wraps an ordered list of child steps and renders them as bash functions that execute concurrently. This keeps the runtime pure bash, reuses the existing per-step rendering machinery, and adds zero new dependencies. The fork-join pattern (child &; pid=$!; wait "$pid") is POSIX-shell portable and well-understood.

Affected Files

File	Lines	Action	Description
src/flowsh_cli/models.py	157	UPDATE	Add ParallelStep model and include it in the Step union
src/flowsh_cli/render.py	221-296	UPDATE	Add ParallelStep branch in render_step() and default_step_title()
tests/test_workflow_to_harness.py	NEW	CREATE	Add tests for parallel step parse, render, and execution

Integration Points

• render_step() in render.py:221 — the main dispatch for all step types; parallel adds a new branch
• default_step_title() in render.py:289 — must handle ParallelStep to avoid the fallback f"step {index}"
• Step union type in models.py:157 — the discriminated union that drives Pydantic parsing and type narrowing
• render_harness() in render.py:208 — iterates workflow.steps and calls render_step() for each; no change needed because render_step() handles the new type

Git History

• Introduced: N/A (greenfield feature)
• Last modified: N/A
• Implication: No regression risk; all files to touch were last changed in commit 05692fe ("feat: Added ghar")

---

Implementation Plan

Step 1: Add ParallelStep model

File: src/flowsh_cli/models.py
Lines: after line 155 (before Step union at line 157)
Action: INSERT

Current code:

# line 155-157
        return value


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

Required change:

        return value


class ParallelStep(BaseStep):
    type: Literal["parallel"]
    steps: list[Step]

    @field_validator("steps")
    @classmethod
    def validate_steps(cls, value: list[Step]) -> list[Step]:
        if not value:
            raise ValueError("must contain at least one step")
        return value


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

Why: Adds the new model before the union so ParallelStep is defined when referenced. The union update makes parallel a valid discriminator value in all step positions.

---

Step 2: Add ParallelStep rendering in render_step()

File: src/flowsh_cli/render.py
Lines: 221-286 (the render_step function)
Action: INSERT before the else: raise AssertionError clause

Current code:

# lines 281-282
    else:
        raise AssertionError(f"Unsupported step type: {step}")

Required change — insert before the else:

    elif isinstance(step, ParallelStep):
        child_function_names: list[str] = []
        for child_idx, child in enumerate(step.steps, start=1):
            child_lines = render_step(child_idx, child, used_function_names)
            child_function_names.append(
                step_function_name(child_idx, child.name, None)
            )
            lines.extend(child_lines)
        lines.append("  local status=0")
        for child_fn in child_function_names:
            lines.append(f'  "{child_fn}" &')
            lines.append(f"  local pid_{child_fn}=$!")
        for child_fn in child_function_names:
            lines.append(f'  wait "$pid_{child_fn}" || status=$?')
        lines.append('  return "$status"')
        runner = "run_step"

And update the runner assignment at line 284 (currently runner = "run_stateful_step" if isinstance(step, VarsStep) else "run_step") — no change needed since ParallelStep uses run_step.

Why: Renders each child step into its own function using the existing render_step() (recursive reuse), then generates the fork-join bash wrapper. The function name for each child is computed via step_function_name() with used_function_names to ensure uniqueness across the full workflow. The child_idx is passed relative to the parallel block (starting at 1 within its siblings) — this is fine since step_function_name produces deterministic slugs and used_function_names deduplicates globally.

Important: The import in render.py line 6 must be updated to include ParallelStep:

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

---

Step 3: Add ParallelStep default title in default_step_title()

File: src/flowsh_cli/render.py
Lines: 289-296
Action: INSERT a new branch

Current code:

# lines 289-296
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)
    return f"step {index}"

Required change — insert before the fallback return:

    if isinstance(step, ParallelStep):
        return f"parallel ({len(step.steps)} steps)"

Why: Provides a descriptive default title instead of the generic step N.

---

Step 4: Update __init__.py exports (optional)

File: src/flowsh_cli/__init__.py
Action: REVIEW (likely no change needed — ParallelStep is not imported at the package level currently, and Step is not exported either. But if needed for test convenience, add it.)

Actually, looking at test imports (line 13-18 of test file), they import individual step types for constructing Workflow objects directly. The new test will need ParallelStep too. Add to __init__.py:

from flowsh_cli.models import Workflow, WorkflowParseError, ParallelStep, parse_workflows

And the test file import should include ParallelStep:

from flowsh_cli.models import (
    MAX_WORKFLOW_YAML_BYTES,
    AgentStep,
    BashStep,
    ParallelStep,
    Workflow,
    WorkflowFile,
    WorkflowParam,
)

---

Step 5: Add tests

File: tests/test_workflow_to_harness.py
Action: CREATE (add tests)

Test cases to add:

# ---------------------------------------------------------------------------
# Parallel step tests (issue #25)
# ---------------------------------------------------------------------------


def test_parse_workflows_accepts_parallel_step(tmp_path: Path) -> None:
    workflow_file = tmp_path / "workflows.yml"
    workflow_file.write_text(
        """
workflows:
  - id: wf_parallel
    name: Parallel Workflow
    steps:
      - type: parallel
        name: Fan out
        steps:
          - type: bash
            name: Build
            run: printf 'build\\n'
          - type: bash
            name: Test
            run: printf 'test\\n'
""".lstrip(),
        encoding="utf-8",
    )

    workflows = parse_workflows(workflow_file)

    assert len(workflows) == 1
    assert len(workflows[0].steps) == 1
    parallel_step = workflows[0].steps[0]
    assert isinstance(parallel_step, ParallelStep)
    assert parallel_step.name == "Fan out"
    assert len(parallel_step.steps) == 2
    assert isinstance(parallel_step.steps[0], BashStep)
    assert parallel_step.steps[0].run == "printf 'build\\n'"
    assert isinstance(parallel_step.steps[1], BashStep)
    assert parallel_step.steps[1].run == "printf 'test\\n'"


def test_parse_workflows_rejects_empty_parallel_steps(tmp_path: Path) -> None:
    workflow_file = tmp_path / "workflows.yml"
    workflow_file.write_text(
        """
workflows:
  - id: wf_empty_parallel
    name: Empty Parallel
    steps:
      - type: parallel
        name: Empty
        steps: []
""".lstrip(),
        encoding="utf-8",
    )

    with pytest.raises(WorkflowParseError, match="at least one step"):
        parse_workflows(workflow_file)


def test_parse_workflows_accepts_parallel_step_with_vars_and_agent(tmp_path: Path) -> None:
    workflow_file = tmp_path / "workflows.yml"
    workflow_file.write_text(
        """
workflows:
  - id: wf_parallel_mixed
    name: Parallel Mixed
    steps:
      - type: parallel
        name: Mixed
        steps:
          - type: vars
            name: Capture value
            values:
              VALUE: printf 'hello'
          - type: agent
            name: Ask
            prompt: Say hello.
""".lstrip(),
        encoding="utf-8",
    )

    workflows = parse_workflows(workflow_file)
    parallel_step = workflows[0].steps[0]
    assert isinstance(parallel_step, ParallelStep)
    assert isinstance(parallel_step.steps[0], VarsStep)
    assert isinstance(parallel_step.steps[1], AgentStep)


def test_render_harness_parallel_step_generates_fork_join_bash() -> None:
    workflow = Workflow(
        id="wf_parallel_render",
        name="Parallel Render",
        steps=[
            ParallelStep(
                type="parallel",
                name="Fan out",
                steps=[
                    BashStep(type="bash", name="Build", run="printf 'build\\n'"),
                    BashStep(type="bash", name="Test", run="printf 'test\\n'"),
                ],
            )
        ],
    )

    script = render_harness(workflow)

    # Each child gets its own function
    assert "step_build()" in script
    assert "step_test()" in script
    # Fork-join pattern
    assert "&" in script
    assert '"step_build" &' in script or "'step_build' &" in script
    assert "$!" in script
    assert "wait" in script
    # Wrapper function
    assert "step_fan_out()" in script
    # Runner calls the parallel wrapper
    assert "run_step step_fan_out" in script


def test_generated_harness_runs_parallel_steps_concurrently(tmp_path: Path) -> None:
    workflow_file = tmp_path / "workflows.yml"
    workflow_file.write_text(
        """
workflows:
  - id: wf_parallel_exec
    name: Parallel Exec
    steps:
      - type: parallel
        name: Fan out
        steps:
          - type: bash
            name: Write A
            run: printf 'a\\n'
          - type: bash
            name: Write B
            run: printf 'b\\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
    harness = tmp_path / ".harness" / "parallel_exec.sh"

    # Check bash syntax is valid
    syntax = subprocess.run(
        ["bash", "-n", str(harness)], check=False, capture_output=True, text=True
    )
    assert syntax.returncode == 0, syntax.stderr

    # Run the harness (no special env needed for bash-only steps)
    executed = subprocess.run(
        ["bash", str(harness)],
        check=False,
        capture_output=True,
        text=True,
        cwd=tmp_path,
    )
    assert executed.returncode == 0, executed.stderr
    # Both outputs must appear (order is non-deterministic due to concurrency)
    assert "a" in executed.stdout
    assert "b" in executed.stdout


def test_generated_harness_parallel_step_fails_when_child_fails(tmp_path: Path) -> None:
    workflow_file = tmp_path / "workflows.yml"
    workflow_file.write_text(
        """
workflows:
  - id: wf_parallel_fail
    name: Parallel Fail
    steps:
      - type: parallel
        name: Fan out
        steps:
          - type: bash
            name: Success
            run: printf 'ok\\n'
          - type: bash
            name: Failure
            run: false
""".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" / "parallel_fail.sh"

    executed = subprocess.run(
        ["bash", str(harness)],
        check=False,
        capture_output=True,
        text=True,
        cwd=tmp_path,
    )
    assert executed.returncode != 0
    assert "Step failed: step_fan_out" in executed.stderr


def test_render_harness_parallel_step_with_existing_sequential_steps() -> None:
    """Parallel step coexists with regular sequential steps."""
    workflow = Workflow(
        id="wf_mixed",
        name="Mixed",
        steps=[
            BashStep(type="bash", name="Setup", run="printf 'setup\\n'"),
            ParallelStep(
                type="parallel",
                steps=[
                    BashStep(type="bash", name="Build", run="printf 'build\\n'"),
                    BashStep(type="bash", name="Test", run="printf 'test\\n'"),
                ],
            ),
            BashStep(type="bash", name="Teardown", run="printf 'done\\n'"),
        ],
    )

    script = render_harness(workflow)

    # All step functions present
    assert "step_setup()" in script
    assert "step_build()" in script
    assert "step_test()" in script
    # The parallel wrapper replaces the inline children with fork-join
    assert "&" in script
    assert "wait" in script
    # Sequential order preserved
    assert "run_step step_setup" in script
    assert "run_step step_build" not in script  # children are part of parallel
    assert "step_teardown()" in script
    assert "run_step step_teardown" in script

Note on test_render_harness_parallel_step_generates_fork_join_bash: The exact content of "$fn" & depends on how render_step formats the lines. The test above uses a more flexible assertion (checks for &, $!, wait separately) to avoid brittleness. Adjust the exact match if needed after implementation.

---

Patterns to Follow

From codebase — mirror these exactly:

# SOURCE: src/flowsh_cli/models.py:60-71
# Pattern for step model definition with type discriminator and field validation
class BashStep(BaseStep):
    type: Literal["bash"]
    run: str

    @field_validator("run")
    @classmethod
    def validate_run(cls, value: str) -> str:
        ...

# SOURCE: src/flowsh_cli/render.py:243-251
# Pattern for rendering a step type — function definition + runner call
    elif isinstance(step, BashStep):
        delimiter = heredoc_delimiter("BASH", step.run)
        lines.extend([
            f"  bash -euo pipefail <<'{delimiter}'",
            *step.run.strip().splitlines(),
            delimiter,
        ])

# SOURCE: tests/test_workflow_to_harness.py:540-551
# Pattern for testing rendering output
def test_render_harness_uses_only_opencode_agent_invocation(tmp_path: Path) -> None:
    ...
    script = render_harness(workflow)
    assert "opencode run --format json" in script

---

Edge Cases & Risks

Risk/Edge Case	Mitigation
Nested parallel (parallel inside parallel) is not needed	Not supported; nested parallel will fail validation since Step union does not recurse (it's the same flat union — actually it would work recursively since ParallelStep.steps is list[Step] and Step includes ParallelStep. To explicitly defer, add a model validator to reject ParallelStep inside child steps.)
Very large fan-out (100+ children)	Pure bash wait on 100+ PIDs is fine; no special handling needed
Child vars step state leaks to siblings or parent	Each child is a separate function — run_stateful_step for VarsStep sources vars into the current shell, but each child runs as a backgrounded function within the parallel wrapper, so exported vars are visible to that child's function only. No cross-branch leak.
Child agent step fails to find opencode CLI	Returns exit 127; caught by `wait "$pid"
run_step vs run_stateful_step for parallel children	Children can use run_step (which wraps in a subshell ()) since VarsStep children already use run_stateful_step internally; the parallel wrapper just needs run_step for itself. No change needed.

Optional: Add model validator to explicitly defer nested parallel:

@model_validator(mode="after")
def reject_nested_parallel(cls, value: object) -> object:
    for step in value.steps:
        if isinstance(step, ParallelStep):
            raise ValueError("nested parallel steps are not supported")
    return value

If you want to keep the door open for future nested parallel and the code is simpler without the validator, simply omit it. The issue says "Nested parallel can be deferred unless there is a strong need."

---

Validation

Automated Checks

make qa   # runs tests + lint + typecheck

Manual Verification

1. Create a workflow YAML with a parallel step containing 2-3 bash children, generate the harness, and run it — verify all children execute and outputs appear
2. Create a mixed workflow (sequential + parallel + sequential) and verify correct ordering
3. Verify existing workflows (no parallel steps) still generate identical output

---

Scope Boundaries

IN SCOPE:

• ParallelStep model with type: "parallel" and steps: list[Step]
• Rendering of ParallelStep using fork-join bash pattern
• Default title for parallel steps
• Validation: at least one child step required
• Tests: parse, render, execution, failure propagation

OUT OF SCOPE (do not touch):

• Nested parallel (parallel inside parallel) — explicitly deferred
• Cancellation of in-flight children on first failure
• Shared state or communication between parallel branches
• Timeout or resource limits per branch
• UI or CLI changes for parallel step visualization
• Changes to existing step types (vars, bash, agent) or their rendering

---

Metadata

• Investigated by: GHAR
• Timestamp: 2026-06-02T21:11:00Z
• Artifact: .ghar/issues/issue-25.md

[tbrandenburg]

Investigation: Support declarative parallel steps in workflow definitions

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

Assessment

Metric	Value	Reasoning
Priority	MEDIUM	Adds valuable concurrency without blocking any existing functionality; sequential workflows remain fully functional
Complexity	LOW	3 files affected (models.py, render.py, test file), isolated additions that mirror the existing ForStep pattern exactly
Confidence	HIGH	Issue contains full bash shape, the codebase has a clear ForStep precedent to mirror, no external integration required

Problem Statement

Workflows today execute steps strictly sequentially. Users need a declarative parallel step type that wraps child steps (vars, bash, agent) and runs them concurrently via a fork-join bash pattern (& + $! + wait). The generated artifact must remain plain bash with no new runtime dependencies, and all existing workflows must continue to work unchanged.

Analysis

Change Rationale

A ParallelStep model wraps an ordered list of child steps. During rendering it produces individual bash functions for each child (reusing the existing render_step recursion, exactly as ForStep does for its inner steps), then emits a wrapper function that backgrounds each child, captures PIDs, and waits for all of them — propagating the worst non-zero exit code.

Evidence Chain

WHY: Users cannot express concurrent step execution in workflow YAML
↓ BECAUSE: The Step union in models.py:171 only includes VarsStep | BashStep | AgentStep | ForStep
Evidence: src/flowsh_cli/models.py:171 — Step = Annotated[VarsStep | BashStep | AgentStep | ForStep, Field(discriminator="type")]

↓ BECAUSE: render_step() in render.py:221 has no branch for a parallel wrapper
Evidence: src/flowsh_cli/render.py:227-248 — only ForStep gets prefix-function treatment; everything else delegates to _render_step_body

↓ ROOT CAUSE: No ParallelStep model exists and no renderer branch exists for it
Evidence: src/flowsh_cli/render.py:317-320 — _render_step_body raises AssertionError for any unrecognised step type including a hypothetical parallel step

Affected Files

File	Lines	Action	Description
src/flowsh_cli/models.py	157-174	UPDATE	Add ParallelStep class + model_rebuild() call + union entry
src/flowsh_cli/render.py	6, 227	UPDATE	Import ParallelStep; add branch in render_step(); update default_step_title()
tests/test_workflow_to_harness.py	EOF	UPDATE	Add parse, render, execution, and failure-propagation tests

Integration Points

• models.py:171 — Step union drives Pydantic discriminated parsing everywhere YAML is loaded
• models.py:174 — ForStep.model_rebuild() is required because ForStep.steps references Step (a forward ref); ParallelStep needs the same treatment
• render.py:221 — render_step() is the top-level dispatch called by render_harness() for every top-level step; it also recurses for ForStep inner steps and will recurse for ParallelStep children
• render.py:248 — runner selection "run_stateful_step" if isinstance(step, (VarsStep, ForStep)) else "run_step" — ParallelStep should use run_step (not stateful; no shell state to export)
• render.py:345 — default_step_title() needs a ParallelStep branch to avoid the f"step {index}" fallback
• tests/test_workflow_to_harness.py:14-16 — existing import block must add ForStep, VarsStep, and ParallelStep

Git History

• Introduced: 1bf676f — feat: add 'for' step type for basic loop support (#21) (#26) — most recent shape-defining commit
• Implication: ForStep (added in 1bf676f) is the direct precedent for ParallelStep; the prefix_lines pattern in render_step and the model_rebuild() call are both patterns established there and must be mirrored exactly

Implementation Plan

Step 1: Add ParallelStep model to models.py

File: src/flowsh_cli/models.py
Lines: 157-174
Action: UPDATE — insert ParallelStep class before the Step union and add model_rebuild()

Current code:

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


ForStep.model_rebuild()

Required change:

class ParallelStep(BaseStep):
    type: Literal["parallel"]
    steps: list[Step] = Field(min_length=1)


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


ForStep.model_rebuild()
ParallelStep.model_rebuild()

Why: ParallelStep.steps is list[Step] which references Step — a forward ref at definition time. model_rebuild() resolves the forward reference after Step is defined, exactly mirroring the ForStep pattern at models.py:174. Field(min_length=1) enforces the "at least one child" constraint using the same approach as ForStep.steps.

---

Step 2: Import ParallelStep in render.py

File: src/flowsh_cli/render.py
Lines: 6
Action: UPDATE

Current code:

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

Required change:

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

---

Step 3: Add ParallelStep branch in render_step()

File: src/flowsh_cli/render.py
Lines: 227-248
Action: UPDATE — add elif isinstance(step, ParallelStep) branch alongside the existing ForStep branch

Current code:

    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"

Required change:

    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_}}}"',
        ]
    elif isinstance(step, ParallelStep):
        child_fns: list[str] = []
        for i, child in enumerate(step.steps, start=1):
            child_title = child.name or default_step_title(i, child)
            child_fn = step_function_name(i, child.name, used_function_names)
            child_fns.append(child_fn)
            prefix_lines.append(section(f"Parallel child {i} ({child.type}): {child_title}"))
            prefix_lines.append(f"{child_fn}() {{")
            prefix_lines.extend(_render_step_body(child))
            prefix_lines.append("}")
            prefix_lines.append("")

        body_lines = ["  local status=0"]
        for child_fn in child_fns:
            body_lines.append(f'  "{child_fn}" &')
            body_lines.append(f"  local pid_{child_fn}=$!")
        for child_fn in child_fns:
            body_lines.append(f'  wait "$pid_{child_fn}" || status=$?')
        body_lines.append('  return "$status"')
    else:
        body_lines = _render_step_body(step)

    runner = "run_stateful_step" if isinstance(step, (VarsStep, ForStep)) else "run_step"

Why: Mirrors the ForStep prefix-function pattern exactly. Each child gets its own bash function via _render_step_body (same as ForStep inner steps). The wrapper body backgrounds each child, captures PIDs via $!, then waits for all — collecting the worst non-zero exit code. run_step (not run_stateful_step) is correct because the parallel wrapper does not export shell variables.

Note on _render_step_body and ParallelStep: _render_step_body will hit AssertionError if called with a ParallelStep child. This is intentional — nested parallel is not supported. No changes to _render_step_body are needed; it will naturally raise for ParallelStep children, which is the right guard.

---

Step 4: Add ParallelStep title to default_step_title()

File: src/flowsh_cli/render.py
Lines: 345-354
Action: UPDATE — insert branch before the fallback

Current code:

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}"

Required change:

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_}"
    if isinstance(step, ParallelStep):
        return f"parallel ({len(step.steps)} steps)"
    return f"step {index}"

---

Step 5: Update test imports and add tests

File: tests/test_workflow_to_harness.py
Lines: 14-16 (import block)
Action: UPDATE — add ForStep, VarsStep, ParallelStep to the model import

Current import:

from flowsh_cli.models import (
    MAX_WORKFLOW_YAML_BYTES,
    AgentStep,
    BashStep,
    Workflow,
    WorkflowFile,
    WorkflowParam,
)

Required change:

from flowsh_cli.models import (
    MAX_WORKFLOW_YAML_BYTES,
    AgentStep,
    BashStep,
    ForStep,
    ParallelStep,
    VarsStep,
    Workflow,
    WorkflowFile,
    WorkflowParam,
)

Test cases to add (append to end of file):

# ---------------------------------------------------------------------------
# ParallelStep tests (issue #25)
# ---------------------------------------------------------------------------


def test_parse_workflows_accepts_parallel_step(tmp_path: Path) -> None:
    workflow_file = tmp_path / "workflows.yml"
    workflow_file.write_text(
        """\
workflows:
  - id: wf_parallel
    name: Parallel Workflow
    steps:
      - type: parallel
        name: Fan out
        steps:
          - type: bash
            name: Build
            run: printf 'build\\n'
          - type: bash
            name: Test
            run: printf 'test\\n'
""",
        encoding="utf-8",
    )

    workflows = parse_workflows(workflow_file)

    assert len(workflows) == 1
    assert len(workflows[0].steps) == 1
    step = workflows[0].steps[0]
    assert isinstance(step, ParallelStep)
    assert step.name == "Fan out"
    assert len(step.steps) == 2
    assert isinstance(step.steps[0], BashStep)
    assert isinstance(step.steps[1], BashStep)


def test_parse_workflows_rejects_empty_parallel_steps(tmp_path: Path) -> None:
    workflow_file = tmp_path / "workflows.yml"
    workflow_file.write_text(
        """\
workflows:
  - id: wf_empty_parallel
    name: Empty Parallel
    steps:
      - type: parallel
        name: Empty
        steps: []
""",
        encoding="utf-8",
    )

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


def test_parse_workflows_accepts_parallel_step_with_vars_and_agent(tmp_path: Path) -> None:
    workflow_file = tmp_path / "workflows.yml"
    workflow_file.write_text(
        """\
workflows:
  - id: wf_parallel_mixed
    name: Parallel Mixed
    steps:
      - type: parallel
        name: Mixed
        steps:
          - type: vars
            name: Capture value
            values:
              VALUE: printf 'hello'
          - type: agent
            name: Ask
            prompt: Say hello.
""",
        encoding="utf-8",
    )

    workflows = parse_workflows(workflow_file)
    step = workflows[0].steps[0]
    assert isinstance(step, ParallelStep)
    assert isinstance(step.steps[0], VarsStep)
    assert isinstance(step.steps[1], AgentStep)


def test_render_harness_parallel_step_generates_fork_join_bash() -> None:
    workflow = Workflow(
        id="wf_parallel_render",
        name="Parallel Render",
        steps=[
            ParallelStep(
                type="parallel",
                name="Fan out",
                steps=[
                    BashStep(type="bash", name="Build", run="printf 'build\\n'"),
                    BashStep(type="bash", name="Test", run="printf 'test\\n'"),
                ],
            )
        ],
    )

    script = render_harness(workflow)

    # Child functions are emitted
    assert "step_build()" in script
    assert "step_test()" in script
    # Fork-join pattern present
    assert '"step_build" &' in script
    assert '"step_test" &' in script
    assert "pid_step_build=$!" in script
    assert "pid_step_test=$!" in script
    assert 'wait "$pid_step_build"' in script
    assert 'wait "$pid_step_test"' in script
    # Wrapper function emitted
    assert "step_fan_out()" in script
    # Wrapper is called via run_step
    assert "run_step step_fan_out" in script
    # Children are NOT directly called via run_step at top level
    assert "run_step step_build" not in script
    assert "run_step step_test" not in script


def test_render_harness_parallel_step_section_comment() -> None:
    workflow = Workflow(
        id="wf_par_comment",
        name="Par Comment",
        steps=[
            ParallelStep(
                type="parallel",
                steps=[
                    BashStep(type="bash", run="echo a"),
                    BashStep(type="bash", run="echo b"),
                ],
            )
        ],
    )

    script = render_harness(workflow)

    assert "Parallel child 1" in script
    assert "Parallel child 2" in script
    assert "parallel (2 steps)" in script


def test_generated_harness_runs_parallel_steps(tmp_path: Path) -> None:
    workflow_file = tmp_path / "workflows.yml"
    workflow_file.write_text(
        """\
workflows:
  - id: wf_parallel_exec
    name: Parallel Exec
    steps:
      - type: parallel
        name: Fan out
        steps:
          - type: bash
            name: Write A
            run: printf 'output_a\\n'
          - type: bash
            name: Write B
            run: printf 'output_b\\n'
""",
        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" / "parallel_exec.sh"
    assert harness.exists()

    syntax = subprocess.run(
        ["bash", "-n", str(harness)], check=False, capture_output=True, text=True
    )
    assert syntax.returncode == 0, syntax.stderr

    executed = subprocess.run(
        ["bash", str(harness)],
        check=False,
        capture_output=True,
        text=True,
        cwd=tmp_path,
        env={**os.environ, "FLOWSH_LOG_DIR": str(tmp_path / "logs")},
    )
    assert executed.returncode == 0, executed.stderr
    assert "output_a" in executed.stdout
    assert "output_b" in executed.stdout


def test_generated_harness_parallel_step_propagates_child_failure(tmp_path: Path) -> None:
    workflow_file = tmp_path / "workflows.yml"
    workflow_file.write_text(
        """\
workflows:
  - id: wf_parallel_fail
    name: Parallel Fail
    steps:
      - type: parallel
        name: Fan out
        steps:
          - type: bash
            name: Success
            run: printf 'ok\\n'
          - type: bash
            name: Failure
            run: "false"
""",
        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" / "parallel_fail.sh"

    executed = subprocess.run(
        ["bash", str(harness)],
        check=False,
        capture_output=True,
        text=True,
        cwd=tmp_path,
        env={**os.environ, "FLOWSH_LOG_DIR": str(tmp_path / "logs")},
    )
    assert executed.returncode != 0
    assert "Step failed: step_fan_out" in executed.stderr


def test_render_harness_parallel_coexists_with_sequential_steps() -> None:
    workflow = Workflow(
        id="wf_mixed_seq_par",
        name="Mixed Seq Par",
        steps=[
            BashStep(type="bash", name="Setup", run="echo setup"),
            ParallelStep(
                type="parallel",
                steps=[
                    BashStep(type="bash", name="Build", run="echo build"),
                    BashStep(type="bash", name="Test", run="echo test"),
                ],
            ),
            BashStep(type="bash", name="Teardown", run="echo done"),
        ],
    )

    script = render_harness(workflow)

    # All three top-level functions called sequentially via run_step
    assert "run_step step_setup" in script
    assert "run_step step_teardown" in script
    # The parallel wrapper (not children) is the top-level call
    assert "run_step step_2" in script or "run_step step_parallel" in script or "run_step step_" in script
    # Children backgrounded inside wrapper, not at top-level
    assert '"step_build" &' in script
    assert '"step_test" &' in script
    # Children not individually called at top level
    assert "run_step step_build" not in script
    assert "run_step step_test" not in script

---

Patterns to Follow

From codebase — mirror these exactly:

# SOURCE: src/flowsh_cli/models.py:157-174
# Pattern: self-referential step model requires model_rebuild() after Step union is defined
class ForStep(BaseStep):
    type: Literal["for"]
    in_: str = Field(..., validation_alias="in")
    item: str = Field(...)
    steps: list[Step] = Field(min_length=1)

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

# SOURCE: src/flowsh_cli/render.py:227-244
# Pattern: prefix_lines for child function emission, body_lines for wrapper content
    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("")

---

Edge Cases & Risks

Risk/Edge Case	Mitigation
Nested parallel (parallel inside parallel)	_render_step_body has no branch for ParallelStep and raises AssertionError — naturally rejected. No explicit guard needed unless a clearer error message is desired.
VarsStep child: exported var not visible to siblings	Documented known tradeoff in issue; each child function runs in its own scope. No mitigation — by design.
VarsStep child uses run_stateful_step internally	_render_step_body for VarsStep does not call run_stateful_step; it inlines the $() capture. So a VarsStep inside a parallel child will work correctly — vars are exported within that child's function scope.
Child function name collision with outer steps	step_function_name(..., used_function_names) deduplicates globally via the shared used_function_names set passed through from render_harness.
run_step wraps the call in a subshell ()	run_step at render.py:119 calls "$step_name" inside ( : >> "$LOG_FILE" ) — the outer parallel wrapper is still called correctly since run_step step_fan_out invokes the wrapper synchronously, and the backgrounding happens inside the wrapper function itself.
Interleaved stdout from concurrent children	Documented known tradeoff; no mitigation — by design.

Validation

Automated Checks

make qa

Manual Verification

1. Write a workflows.yml with a parallel step containing 2 bash children that each sleep 1 && echo X. Run the harness; total time should be ~1 s not ~2 s — proving concurrency.
2. Write a mixed workflow (bash → parallel → bash) and verify sequential ordering: setup runs first, teardown runs last.
3. Run make qa against a checkout with no changes to verify all existing tests still pass (no regression).

Scope Boundaries

IN SCOPE:

• ParallelStep model with type: "parallel" and steps: list[Step] (min 1 child)
• Rendering of ParallelStep using fork-join bash: "$child" &; pid=$!; wait "$pid" || status=$?
• default_step_title() branch for ParallelStep
• model_rebuild() call after Step union
• Tests: YAML parse, validation rejection, render structure assertions, execution, failure propagation, coexistence with sequential steps

OUT OF SCOPE (do not touch):

• Nested parallel inside parallel — deferred per issue; naturally rejected by _render_step_body anyway
• Cancellation of in-flight children on first failure — deferred
• Shared mutable state across parallel branches — explicitly a non-goal per issue
• Timeout/resource limits per branch
• Changes to existing step types or their rendering
• CLI or schema output changes beyond what Pydantic auto-generates

Metadata

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