diff --git a/src/specify_cli/workflows/steps/fan_out/__init__.py b/src/specify_cli/workflows/steps/fan_out/__init__.py
index 22b9c37d43..81cf4f8d50 100644
--- a/src/specify_cli/workflows/steps/fan_out/__init__.py
+++ b/src/specify_cli/workflows/steps/fan_out/__init__.py
@@ -25,6 +25,33 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
         max_concurrency = config.get("max_concurrency", 1)
         step_template = config.get("step", {})
 
+        # The engine does not auto-validate step config (see
+        # ``WorkflowEngine.load_workflow``). On a COMPLETED fan-out it reads the
+        # ``step_template`` back out and, when it is truthy, calls
+        # ``template.get("id", ...)`` in ``_run_fan_out``. A truthy non-mapping
+        # ``step`` (a scalar or list authoring mistake) would crash the whole
+        # run with AttributeError there — the engine invokes ``execute`` and
+        # ``_run_fan_out`` with no surrounding try/except. ``validate`` already
+        # rejects a non-mapping ``step``; fail this step loudly on an
+        # unvalidated run instead, mirroring the ``items`` guard below. An empty
+        # or absent ``step`` defaults to ``{}`` (falsy) and the engine's
+        # ``if template and items`` skips fan-out, so it stays valid here.
+        if not isinstance(step_template, dict):
+            return StepResult(
+                status=StepStatus.FAILED,
+                error=(
+                    f"Fan-out step {config.get('id', '?')!r}: 'step' must be a "
+                    f"mapping (nested step template), got "
+                    f"{type(step_template).__name__}."
+                ),
+                output={
+                    "items": [],
+                    "max_concurrency": max_concurrency,
+                    "step_template": {},
+                    "item_count": 0,
+                },
+            )
+
         if not isinstance(items, list):
             # A non-list here is a wiring error (the expression did not
             # resolve to a collection); silently fanning out over zero
@@ -66,8 +93,13 @@ def validate(self, config: dict[str, Any]) -> list[str]:
                 f"Fan-out step {config.get('id', '?')!r} is missing "
                 f"'step' field (nested step template)."
             )
-        step = config.get("step")
-        if step is not None and not isinstance(step, dict):
+        elif not isinstance(config["step"], dict):
+            # A present-but-non-mapping ``step`` (including an explicit
+            # ``step: null``) is an authoring mistake. ``config.get("step", {})``
+            # in ``execute`` only substitutes the ``{}`` default for an *absent*
+            # key, so an explicit ``None`` reaches the runtime guard and FAILS
+            # the step. Reject it here too so a workflow cannot pass validation
+            # and then fail during execution.
             errors.append(
                 f"Fan-out step {config.get('id', '?')!r}: 'step' must be a mapping."
             )
diff --git a/tests/test_workflows.py b/tests/test_workflows.py
index 8fefad740e..e53366bf70 100644
--- a/tests/test_workflows.py
+++ b/tests/test_workflows.py
@@ -2426,6 +2426,39 @@ def test_execute_empty_list_items_is_valid(self):
         assert result.status == StepStatus.COMPLETED
         assert result.output["item_count"] == 0
 
+    def test_execute_non_dict_step_fails_loudly(self):
+        """A truthy non-mapping ``step`` must fail the step, not crash the run.
+
+        ``validate`` rejects a non-dict ``step``, but the engine's ``execute()``
+        does not auto-validate (see ``WorkflowEngine.load_workflow``). On a
+        COMPLETED fan-out the engine reads ``step_template`` back out and, when
+        it is truthy, calls ``template.get("id", ...)`` in ``_run_fan_out``. A
+        truthy non-mapping ``step`` (a scalar or list authoring mistake) raised
+        AttributeError there and took down the whole run. Mirrors the fan-out
+        non-list ``items`` guard and the switch non-dict ``cases`` guard.
+        """
+        from specify_cli.workflows.steps.fan_out import FanOutStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        step = FanOutStep()
+        ctx = StepContext(steps={"tasks": {"output": {"task_list": [1, 2]}}})
+        # ``None`` is an explicit ``step: null``: ``config.get("step", {})`` only
+        # substitutes the default for an *absent* key, so it reaches the guard
+        # and must fail here too — matching ``validate``.
+        for bad_step in (["impl"], "impl", 5, None):
+            result = step.execute(
+                {
+                    "id": "parallel",
+                    "items": "{{ steps.tasks.output.task_list }}",
+                    "step": bad_step,
+                },
+                ctx,
+            )
+            assert result.status == StepStatus.FAILED
+            assert "'step' must be a" in (result.error or "")
+            assert result.output["item_count"] == 0
+            assert result.output["step_template"] == {}
+
     def test_validate_missing_fields(self):
         from specify_cli.workflows.steps.fan_out import FanOutStep
 
@@ -2438,12 +2471,13 @@ def test_validate_step_not_mapping(self):
         from specify_cli.workflows.steps.fan_out import FanOutStep
 
         step = FanOutStep()
-        errors = step.validate({
-            "id": "test",
-            "items": "{{ x }}",
-            "step": "not-a-dict",
-        })
-        assert any("'step' must be a mapping" in e for e in errors)
+        for bad_step in ("not-a-dict", ["impl"], 5, None):
+            errors = step.validate({
+                "id": "test",
+                "items": "{{ x }}",
+                "step": bad_step,
+            })
+            assert any("'step' must be a mapping" in e for e in errors), bad_step
 
 
 class TestFanInStep: