fix(workflows): fail while/do-while steps on non-list steps instead of crashing (#3519)

`WhileStep.validate()` and `DoWhileStep.validate()` already reject a
non-list `steps` body, but the engine's `execute()` path does not
auto-validate (see `WorkflowEngine.load_workflow`, whose docstring notes
the definition is "not yet validated"). On an unvalidated run the body
is returned as `next_steps`, and the engine feeds it straight into
`_execute_steps`, which iterates it as step mappings. A non-list `steps`
— a single mapping or scalar authoring mistake — was iterated
element-wise (a dict yields its string keys, a str its characters) and
raised `AttributeError` on `.get()`, taking down the whole run; the
engine invokes `step_impl.execute()` with no surrounding try/except.

Guard both `execute` paths to return a FAILED StepResult naming the type
error instead, mirroring the if/switch non-list-branch and fan-out
non-list `items` handling. The do-while body always dispatches on the
first call, so its guard is unconditional; the while body only
dispatches when the condition is truthy, so its guard fires only then —
a false condition leaves a non-list `steps` benign and the step
completes, unchanged. The condition/expression is still evaluated first,
so its result is surfaced in the step output for downstream context.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/specify_cli/workflows/steps/do_while/__init__.py

@@ -27,6 +27,30 @@ class DoWhileStep(StepBase):
         nested_steps = config.get("steps", [])
         condition = config.get("condition", "false")
 
+        # The engine does not auto-validate step config (see
+        # ``WorkflowEngine.load_workflow``) and feeds ``next_steps`` straight
+        # into ``_execute_steps``, which iterates them as step mappings. A
+        # non-list ``steps`` (a single mapping or scalar authoring mistake)
+        # would otherwise be iterated element-wise — a dict yields its string
+        # keys, a str its characters — and crash the whole run with
+        # AttributeError on ``.get()``. ``validate`` already rejects a non-list
+        # ``steps``; fail this step loudly on an unvalidated run instead,
+        # mirroring the if/switch/fan-out steps. The body always runs on the
+        # first call, so unlike the while step this guard is unconditional.
+        if not isinstance(nested_steps, list):
+            return StepResult(
+                status=StepStatus.FAILED,
+                output={
+                    "condition": condition,
+                    "max_iterations": max_iterations,
+                    "loop_type": "do-while",
+                },
+                error=(
+                    f"Do-while step {config.get('id', '?')!r}: 'steps' must be "
+                    f"a list of steps, got {type(nested_steps).__name__}."
+                ),
+            )
+
         # Always execute body at least once; the engine layer evaluates
         # `condition` after each iteration to decide whether to loop.
         return StepResult(

src/specify_cli/workflows/steps/while_loop/__init__.py

@@ -26,6 +26,32 @@ class WhileStep(StepBase):
         nested_steps = config.get("steps", [])
 
         result = evaluate_condition(condition, context)
+
+        # The engine does not auto-validate step config (see
+        # ``WorkflowEngine.load_workflow``) and feeds ``next_steps`` straight
+        # into ``_execute_steps``, which iterates them as step mappings. A
+        # non-list ``steps`` (a single mapping or scalar authoring mistake)
+        # would otherwise be iterated element-wise — a dict yields its string
+        # keys, a str its characters — and crash the whole run with
+        # AttributeError on ``.get()``. ``validate`` already rejects a non-list
+        # ``steps``; fail this step loudly on an unvalidated run instead,
+        # mirroring the if/switch/fan-out steps. The guard fires only when the
+        # body would actually be dispatched (condition truthy). The condition is
+        # still evaluated first, so its result is surfaced for downstream context.
+        if result and not isinstance(nested_steps, list):
+            return StepResult(
+                status=StepStatus.FAILED,
+                output={
+                    "condition_result": True,
+                    "max_iterations": max_iterations,
+                    "loop_type": "while",
+                },
+                error=(
+                    f"While step {config.get('id', '?')!r}: 'steps' must be a "
+                    f"list of steps, got {type(nested_steps).__name__}."
+                ),
+            )
+
         if result:
             return StepResult(
                 status=StepStatus.COMPLETED,

tests/test_workflows.py

@@ -2260,6 +2260,47 @@ class TestWhileStep:
         assert result.output["condition_result"] is False
         assert result.next_steps == []
 
+    @pytest.mark.parametrize("bad_steps", [{"id": "x"}, "oops", 5])
+    def test_execute_non_list_steps_fails_loudly(self, bad_steps):
+        """A non-list ``steps`` reached at runtime must fail the step, not crash.
+
+        ``validate`` rejects a non-list ``steps``, but the engine does not
+        auto-validate (see ``WorkflowEngine.load_workflow``) and feeds
+        ``next_steps`` straight into ``_execute_steps``, which iterates them as
+        step mappings. The while body only dispatches when the condition is
+        truthy, so a non-list ``steps`` reaches ``next_steps`` and would crash
+        the engine's step iteration on an unvalidated run. Mirrors the
+        if/switch/fan-out non-list handling.
+        """
+        from specify_cli.workflows.steps.while_loop import WhileStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        step = WhileStep()
+        ctx = StepContext(inputs={})
+        result = step.execute(
+            {"id": "retry", "condition": "true", "steps": bad_steps}, ctx
+        )
+        assert result.status == StepStatus.FAILED
+        assert "'steps' must be a list of steps" in (result.error or "")
+        assert result.next_steps == []
+
+    @pytest.mark.parametrize("bad_steps", [{"id": "x"}, "oops", 5])
+    def test_execute_non_list_steps_ok_when_condition_false(self, bad_steps):
+        """A false condition never dispatches the body, so a non-list ``steps``
+        stays benign — the step completes without touching ``next_steps``.
+        """
+        from specify_cli.workflows.steps.while_loop import WhileStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        step = WhileStep()
+        ctx = StepContext(inputs={})
+        result = step.execute(
+            {"id": "retry", "condition": "false", "steps": bad_steps}, ctx
+        )
+        assert result.status == StepStatus.COMPLETED
+        assert result.output["condition_result"] is False
+        assert result.next_steps == []
+
     def test_validate_missing_fields(self):
         from specify_cli.workflows.steps.while_loop import WhileStep
 
@@ -2350,6 +2391,30 @@ class TestDoWhileStep:
         assert result.next_steps == []
         assert result.status.value == "completed"
 
+    @pytest.mark.parametrize("bad_steps", [{"id": "x"}, "oops", 5])
+    def test_execute_non_list_steps_fails_loudly(self, bad_steps):
+        """A non-list ``steps`` must fail the step, not crash the run.
+
+        ``validate`` rejects a non-list ``steps``, but the engine does not
+        auto-validate (see ``WorkflowEngine.load_workflow``) and feeds
+        ``next_steps`` straight into ``_execute_steps``, which iterates them as
+        step mappings. The do-while body always dispatches on the first call
+        regardless of condition, so a non-list ``steps`` always reaches
+        ``next_steps`` and would crash the engine's step iteration on an
+        unvalidated run. Mirrors the if/switch/fan-out non-list handling.
+        """
+        from specify_cli.workflows.steps.do_while import DoWhileStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        step = DoWhileStep()
+        ctx = StepContext(inputs={})
+        result = step.execute(
+            {"id": "cycle", "condition": "false", "steps": bad_steps}, ctx
+        )
+        assert result.status == StepStatus.FAILED
+        assert "'steps' must be a list of steps" in (result.error or "")
+        assert result.next_steps == []
+
     def test_validate_missing_fields(self):
         from specify_cli.workflows.steps.do_while import DoWhileStep