fix(workflows): validate command step input/options are mappings

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

@@ -31,6 +31,20 @@ class CommandStep(StepBase):
     def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
         command = config.get("command", "")
         input_data = config.get("input", {})
+        # validate() rejects a non-mapping input, but the engine does not
+        # auto-validate before execute(); a workflow that skipped validation can
+        # still reach here. Fail the step with the same contract error rather
+        # than silently coercing to {} and dispatching with empty args — that
+        # would change the command's meaning, hide the config error, and report
+        # COMPLETED, defeating the per-step FAILED / continue_on_error behavior.
+        if not isinstance(input_data, dict):
+            return StepResult(
+                status=StepStatus.FAILED,
+                error=(
+                    f"Command step {config.get('id', '?')!r}: 'input' must be a "
+                    f"mapping, got {type(input_data).__name__}."
+                ),
+            )
 
         # Resolve expressions in input
         resolved_input: dict[str, Any] = {}
@@ -50,8 +64,18 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
         # Merge options (workflow defaults ← step overrides)
         options = dict(context.default_options)
         step_options = config.get("options", {})
-        if step_options:
-            options.update(step_options)
+        # Same rationale as 'input': a malformed options fails the step rather
+        # than being silently ignored (which would let an invalid step run and
+        # apparently complete).
+        if not isinstance(step_options, dict):
+            return StepResult(
+                status=StepStatus.FAILED,
+                error=(
+                    f"Command step {config.get('id', '?')!r}: 'options' must be a "
+                    f"mapping, got {type(step_options).__name__}."
+                ),
+            )
+        options.update(step_options)
 
         # Attempt CLI dispatch
         args_str = str(resolved_input.get("args", ""))
@@ -155,4 +179,16 @@ def validate(self, config: dict[str, Any]) -> list[str]:
             errors.append(
                 f"Command step {config.get('id', '?')!r} is missing 'command' field."
             )
+        # execute() iterates input.items() and options.update(step_options); a
+        # non-mapping here would raise at run time. Validate the shape like the
+        # sibling steps (switch 'cases', fan-out 'step') so it is reported, not
+        # crashed on.
+        if "input" in config and not isinstance(config["input"], dict):
+            errors.append(
+                f"Command step {config.get('id', '?')!r}: 'input' must be a mapping."
+            )
+        if "options" in config and not isinstance(config["options"], dict):
+            errors.append(
+                f"Command step {config.get('id', '?')!r}: 'options' must be a mapping."
+            )
         return errors

tests/test_workflows.py

@@ -925,6 +925,33 @@ def test_validate_missing_command(self):
         errors = step.validate({"id": "test"})
         assert any("missing 'command'" in e for e in errors)
 
+    def test_validate_rejects_non_mapping_input_and_options(self):
+        from specify_cli.workflows.steps.command import CommandStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        step = CommandStep()
+        # execute() does input.items() / options.update(); a non-mapping must be
+        # reported by validate(), not crash at run time (like switch 'cases').
+        for bad in (None, "args", ["a", "b"], 5):
+            errs = step.validate({"id": "c", "command": "/x", "input": bad})
+            assert any("'input' must be a mapping" in e for e in errs), bad
+        errs = step.validate({"id": "c", "command": "/x", "options": 42})
+        assert any("'options' must be a mapping" in e for e in errs)
+        # a valid mapping config is still accepted
+        assert step.validate({"id": "c", "command": "/x", "input": {"args": "y"}, "options": {"k": 1}}) == []
+        # execute() has no auto-validation guarantee (the engine may skip
+        # validate), so a non-mapping input/options FAILS the step with the same
+        # contract error — it does not silently coerce to empty and report
+        # COMPLETED (which would defeat continue_on_error).
+        res_in = step.execute({"id": "c", "command": "echo", "input": None}, StepContext())
+        assert res_in.status is StepStatus.FAILED
+        assert "'input' must be a mapping" in (res_in.error or "")
+        res_opt = step.execute(
+            {"id": "c", "command": "echo", "input": {}, "options": 42}, StepContext()
+        )
+        assert res_opt.status is StepStatus.FAILED
+        assert "'options' must be a mapping" in (res_opt.error or "")
+
     def test_step_override_integration(self):
         from unittest.mock import patch
         from specify_cli.workflows.steps.command import CommandStep