feat(workflows): opt-in output_format: json exposes parsed shell stdout as output.data

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

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 import subprocess
 from typing import Any
 
@@ -49,6 +50,23 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
                     error=f"Shell command exited with code {proc.returncode}.",
                     output=output,
                 )
+            if config.get("output_format") == "json":
+                # Opt-in structured output: expose the parsed stdout under
+                # ``output.data`` so later steps can consume typed values
+                # (e.g. a fan-out's ``items:``). A parse failure fails the
+                # step — declaring ``output_format: json`` is a contract.
+                try:
+                    output["data"] = json.loads(proc.stdout)
+                except json.JSONDecodeError as exc:
+                    return StepResult(
+                        status=StepStatus.FAILED,
+                        error=(
+                            f"Shell step {config.get('id', '?')!r} declared "
+                            f"output_format: json but stdout is not valid "
+                            f"JSON: {exc}"
+                        ),
+                        output=output,
+                    )
             return StepResult(
                 status=StepStatus.COMPLETED,
                 output=output,
@@ -72,4 +90,10 @@ def validate(self, config: dict[str, Any]) -> list[str]:
             errors.append(
                 f"Shell step {config.get('id', '?')!r} is missing 'run' field."
             )
+        output_format = config.get("output_format")
+        if output_format is not None and output_format != "json":
+            errors.append(
+                f"Shell step {config.get('id', '?')!r}: 'output_format' must "
+                f"be 'json' when present, got {output_format!r}."
+            )
         return errors

tests/test_workflows.py

@@ -941,6 +941,55 @@ def test_validate_missing_run(self):
         assert any("missing 'run'" in e for e in errors)
 
 
+    def test_output_format_json_exposes_data(self, tmp_path):
+        from specify_cli.workflows.steps.shell import ShellStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        step = ShellStep()
+        ctx = StepContext(project_root=str(tmp_path))
+        config = {
+            "id": "emit",
+            "run": "echo '{\"items\": [1, 2]}'",
+            "output_format": "json",
+        }
+        result = step.execute(config, ctx)
+        assert result.status == StepStatus.COMPLETED
+        assert result.output["data"] == {"items": [1, 2]}
+        assert result.output["exit_code"] == 0  # raw keys still present
+
+    def test_output_format_json_invalid_stdout_fails(self, tmp_path):
+        from specify_cli.workflows.steps.shell import ShellStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        step = ShellStep()
+        ctx = StepContext(project_root=str(tmp_path))
+        config = {
+            "id": "emit",
+            "run": "echo not-json",
+            "output_format": "json",
+        }
+        result = step.execute(config, ctx)
+        assert result.status == StepStatus.FAILED
+        assert "output_format: json" in (result.error or "")
+
+    def test_no_output_format_keeps_raw_output_only(self, tmp_path):
+        from specify_cli.workflows.steps.shell import ShellStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        step = ShellStep()
+        ctx = StepContext(project_root=str(tmp_path))
+        config = {"id": "emit", "run": "echo '{\"items\": []}'"}
+        result = step.execute(config, ctx)
+        assert result.status == StepStatus.COMPLETED
+        assert "data" not in result.output
+
+    def test_validate_rejects_unknown_output_format(self):
+        from specify_cli.workflows.steps.shell import ShellStep
+
+        step = ShellStep()
+        errors = step.validate({"id": "emit", "run": "true", "output_format": "yaml"})
+        assert any("'output_format' must be 'json'" in e for e in errors)
+
 class _StubStdin:
     """Stdin stub exposing only a fixed ``isatty`` result.