fix(workflows): fail a fan-in step whose output is not a mapping

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

@@ -20,9 +20,31 @@ class FanInStep(StepBase):
 
     def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
         wait_for = config.get("wait_for", [])
-        output_config = config.get("output") or {}
-        if not isinstance(output_config, dict):
+        output_config = config.get("output")
+        if output_config is None:
             output_config = {}
+        elif not isinstance(output_config, dict):
+            # ``validate`` rejects a non-mapping ``output`` and its comment says
+            # why: "execute() silently coerces a non-mapping output to {}, so the
+            # author's declared aggregation keys would vanish with no error."
+            # The engine does not auto-validate before ``execute``, so on an
+            # unvalidated run that is exactly what happened -- and ``x or {}``
+            # masked the falsy shapes ([], false, 0, '') before the isinstance
+            # check even ran. Every declared key vanished while the step still
+            # reported COMPLETED, so downstream ``steps.<id>.output.<key>``
+            # resolved to None and interpolated as "": the same "silent empty
+            # result + COMPLETED" wiring bug the ``wait_for`` guard below
+            # rejects. Fail loudly with validate()'s own message instead. An
+            # explicit ``output:`` (YAML null) stays valid, matching validate.
+            return StepResult(
+                status=StepStatus.FAILED,
+                error=(
+                    f"Fan-in step {config.get('id', '?')!r}: 'output' must be a "
+                    f"mapping of key -> expression, got "
+                    f"{type(output_config).__name__}."
+                ),
+                output={"results": []},
+            )
 
         # The engine does not auto-validate step config, so an unvalidated run
         # with a non-list ``wait_for`` reaches here raw. Iterating it then

tests/test_workflows.py

@@ -3626,6 +3626,44 @@ def test_execute_non_list_wait_for_fails_loudly(self, bad_wait_for):
         assert "'wait_for' must be a list" in (result.error or "")
         assert result.output["results"] == []
 
+    @pytest.mark.parametrize(
+        "bad_output", [[], False, 0, "", ["a"], "oops", 5]
+    )
+    def test_execute_non_mapping_output_fails_loudly(self, bad_output):
+        """A non-mapping ``output`` must fail the step, not drop every key.
+
+        ``validate`` rejects it and says why: "execute() silently coerces a
+        non-mapping output to {}, so the author's declared aggregation keys would
+        vanish with no error." The engine does not auto-validate before
+        ``execute``, so that is exactly what happened — and ``x or {}`` masked
+        the falsy shapes (``[]``, ``false``, ``0``, ``''``) before the isinstance
+        check even ran. The step still returned COMPLETED, so downstream
+        ``steps.<id>.output.<key>`` resolved to None and interpolated as "".
+        """
+        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": ["a"], "output": bad_output}, ctx
+        )
+        assert result.status == StepStatus.FAILED
+        assert "'output' must be a mapping" in (result.error or "")
+        assert result.output["results"] == []
+
+    def test_execute_explicit_null_output_stays_valid(self):
+        """An explicit ``output:`` (YAML null) is valid, matching ``validate``."""
+        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": ["a"], "output": None}, ctx
+        )
+        assert result.status == StepStatus.COMPLETED
+
     @pytest.mark.parametrize("bad_entry", [["a", "b"], {"a": 1}, 123, None])
     def test_execute_non_string_wait_for_entry_fails_loudly(self, bad_entry):
         """A ``wait_for`` list with a non-string entry must fail the step, not