fix(workflows): fail fan-in step on non-list wait_for instead of crashing (#3482)

`FanInStep.validate()` and the engine's fan-in checks both reject a
non-list `wait_for`, 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, `execute`
iterated the raw value with `for step_id in wait_for`, with two bad
outcomes:

  * a scalar (`wait_for: 5`, `wait_for: null`) raised `TypeError` and
    took down the whole run — the engine invokes `step_impl.execute()`
    with no surrounding try/except; and
  * a string (`wait_for: stepA`) silently iterated its characters and
    returned a join of empty results with a COMPLETED status — the exact
    "silent empty result + COMPLETED" wiring bug the engine's own fan-in
    validation comment warns against.

Guard `execute` to return a FAILED StepResult naming the type error
instead, mirroring the fan-out step's non-list `items` handling. A
missing `wait_for` key still defaults to an empty list (COMPLETED),
unchanged; the guard fires only on an explicit non-list value.

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

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

@@ -24,6 +24,24 @@ class FanInStep(StepBase):
         if not isinstance(output_config, dict):
             output_config = {}
 
+        # The engine does not auto-validate step config, so an unvalidated run
+        # with a non-list ``wait_for`` reaches here raw. Iterating it then
+        # either crashes the whole run (a scalar like an int or None raises
+        # TypeError) or, worse, silently iterates a string's characters and
+        # yields a bogus join of empty results with a COMPLETED status — the
+        # exact "silent empty result + COMPLETED" wiring bug the engine's
+        # fan-in validation guards against. Fail this step loudly instead,
+        # mirroring the fan-out step's non-list ``items`` handling.
+        if not isinstance(wait_for, list):
+            return StepResult(
+                status=StepStatus.FAILED,
+                error=(
+                    f"Fan-in step {config.get('id', '?')!r}: 'wait_for' must be "
+                    f"a list of step IDs, got {type(wait_for).__name__}."
+                ),
+                output={"results": []},
+            )
+
         # Collect results from referenced steps
         results = []
         for step_id in wait_for:

tests/test_workflows.py

@@ -2475,6 +2475,29 @@ class TestFanInStep:
         result = step.execute(config, ctx)
         assert result.output["results"] == [{}]
 
+    @pytest.mark.parametrize("bad_wait_for", ["stepA", 5, None, {"a": 1}])
+    def test_execute_non_list_wait_for_fails_loudly(self, bad_wait_for):
+        """A non-list ``wait_for`` must fail the step, not crash the run or
+        silently produce a bogus join.
+
+        ``validate`` rejects a non-list ``wait_for``, but the engine's
+        ``execute()`` does not auto-validate. Before the guard, ``execute``
+        iterated the raw value: a scalar (int/None) raised TypeError and took
+        down the whole run, while a string silently iterated its characters and
+        returned a join of empty results with a COMPLETED status — the exact
+        "silent empty result + COMPLETED" wiring bug the engine's fan-in
+        validation warns against. Mirrors the fan-out non-list ``items`` guard.
+        """
+        from specify_cli.workflows.steps.fan_in import FanInStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        step = FanInStep()
+        ctx = StepContext(steps={"a": {"output": {"x": 1}}})
+        result = step.execute({"id": "collect", "wait_for": bad_wait_for}, ctx)
+        assert result.status == StepStatus.FAILED
+        assert "'wait_for' must be a list" in (result.error or "")
+        assert result.output["results"] == []
+
     def test_validate_empty_wait_for(self):
         from specify_cli.workflows.steps.fan_in import FanInStep