feat: surface gate detail in the workflow run/resume --json payload

src/specify_cli/__init__.py

@@ -2091,13 +2091,48 @@ def _parse_input_values(input_values: list[str] | None) -> dict[str, Any]:
 
 def _workflow_run_payload(state: Any) -> dict[str, Any]:
     """Machine-readable summary of a run/resume outcome."""
-    return {
+    payload = {
         "run_id": state.run_id,
         "workflow_id": state.workflow_id,
         "status": state.status.value,
         "current_step_id": state.current_step_id,
         "current_step_index": state.current_step_index,
     }
+    gate = _gate_outcome(state)
+    if gate is not None:
+        payload["gate"] = gate
+    return payload
+
+
+def _gate_outcome(state: Any) -> dict[str, Any] | None:
+    """Gate detail for the structured outcome, when the run rests at a gate.
+
+    A paused or gate-aborted run is otherwise indistinguishable from any
+    other pause/abort in the machine-readable payload; surfacing the gate's
+    prompt, options, and (after an interactive choice) the decision lets
+    orchestrators drive review gates without parsing the human-facing stream.
+    """
+    # Two run states rest *on* a gate: `paused` (awaiting a decision) and
+    # `aborted` (a gate rejected with `on_reject: abort` — the only path that
+    # sets ABORTED, leaving current_step_id on that gate). Any other status —
+    # notably `completed`/`failed` — must be suppressed: current_step_id is
+    # not cleared when a run whose last executed step was a gate moves on, so
+    # without this guard it would surface stale detail (run/resume/status).
+    if getattr(state.status, "value", state.status) not in ("paused", "aborted"):
+        return None
+    step = (getattr(state, "step_results", None) or {}).get(state.current_step_id)
+    if not isinstance(step, dict) or step.get("type") != "gate":
+        return None
+    output = step.get("output") or {}
+    # `message` may be a non-string YAML literal (GateStep coerces it only for
+    # expression interpolation), so normalise it here for a stable JSON schema.
+    message = output.get("message")
+    return {
+        "step_id": state.current_step_id,
+        "message": None if message is None else str(message),
+        "options": output.get("options"),
+        "choice": output.get("choice"),
+    }
 
 
 def _run_outcome_exit_code(status_value: str) -> int:

src/specify_cli/workflows/base.py

@@ -47,9 +47,10 @@ class StepContext:
     #: Resolved workflow inputs (from user prompts / defaults).
     inputs: dict[str, Any] = field(default_factory=dict)
 
-    #: Accumulated step results keyed by step ID.
-    #: Each entry is ``{"integration": ..., "model": ..., "options": ...,
-    #:   "input": ..., "output": ...}``.
+    #: Accumulated step results keyed by step ID. Each entry is the dict the
+    #: engine persists per step:
+    #: ``{"type": ..., "integration": ..., "model": ..., "options": ...,
+    #:   "input": ..., "output": ..., "status": ...}``.
     steps: dict[str, dict[str, Any]] = field(default_factory=dict)
 
     #: Current fan-out item (set only inside fan-out iterations).

src/specify_cli/workflows/engine.py

@@ -676,6 +676,7 @@ def _execute_steps(
 
             # Record step results — prefer resolved values from step output
             step_data = {
+                "type": step_type,
                 "integration": result.output.get("integration")
                 or step_config.get("integration")
                 or context.default_integration,

tests/test_workflows.py

@@ -5288,3 +5288,138 @@ def test_resume_failed_run_exits_nonzero(self, tmp_path, monkeypatch):
         assert resumed.exit_code == 1, resumed.stdout
         payload = _json.loads(resumed.stdout)
         assert payload["status"] == "failed"
+
+
+class TestWorkflowRunGateOutcomeJson:
+    """CLI-level tests: the --json payload surfaces gate pauses."""
+
+    _WF_GATE = """
+schema_version: "1.0"
+workflow:
+  id: "gate-json"
+  name: "Gate JSON"
+  version: "1.0.0"
+steps:
+  - id: review
+    type: gate
+    message: "Approve the thing?"
+    options: ["approve", "reject"]
+"""
+
+    _WF_PLAIN = """
+schema_version: "1.0"
+workflow:
+  id: "plain-json"
+  name: "Plain JSON"
+  version: "1.0.0"
+steps:
+  - id: fine
+    type: shell
+    run: "true"
+"""
+
+    def _run_json(self, tmp_path, monkeypatch, content, *, expected_exit=0):
+        import json as _json
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        path = tmp_path / "wf.yml"
+        path.write_text(content, encoding="utf-8")
+        monkeypatch.chdir(tmp_path)
+        result = CliRunner().invoke(app, ["workflow", "run", str(path), "--json"])
+        # Assert the expected exit code before parsing so a real failure
+        # surfaces the actual output instead of an opaque JSON decode error.
+        # A terminal run still emits its JSON payload, then exits non-zero on
+        # ``failed``/``aborted`` (see ``_run_outcome_exit_code``), so callers
+        # pass the expected code. Use ``result.output`` for the message:
+        # under ``--json`` step output is redirected off stdout, so the useful
+        # diagnostics live there.
+        assert result.exit_code == expected_exit, result.output
+        return _json.loads(result.stdout)
+
+    def test_gate_pause_carries_gate_block(self, tmp_path, monkeypatch):
+        # CliRunner stdin is not a TTY, so the gate pauses for resume.
+        payload = self._run_json(tmp_path, monkeypatch, self._WF_GATE)
+        assert payload["status"] == "paused"
+        assert payload["gate"] == {
+            "step_id": "review",
+            "message": "Approve the thing?",
+            "options": ["approve", "reject"],
+            "choice": None,
+        }
+
+    def test_completed_run_has_no_gate_block(self, tmp_path, monkeypatch):
+        payload = self._run_json(tmp_path, monkeypatch, self._WF_PLAIN)
+        assert payload["status"] == "completed"
+        assert "gate" not in payload
+
+    def test_gate_abort_carries_gate_block(self, tmp_path, monkeypatch):
+        # An interactive gate the operator rejects ends the run as `aborted`
+        # (on_reject defaults to abort), not `paused`. The JSON surface must
+        # still carry the gate block with the recorded choice so an
+        # orchestrator can see *why* the run stopped. A gate abort emits the
+        # payload and then exits non-zero (aborted → exit 1), so the helper
+        # is told to expect exit code 1.
+        from specify_cli.workflows.steps.gate import GateStep
+
+        _force_gate_stdin(monkeypatch, tty=True)
+        monkeypatch.setattr(
+            GateStep, "_prompt", staticmethod(lambda _msg, _opts: "reject")
+        )
+        payload = self._run_json(
+            tmp_path, monkeypatch, self._WF_GATE, expected_exit=1
+        )
+        assert payload["status"] == "aborted"
+        assert payload["gate"] == {
+            "step_id": "review",
+            "message": "Approve the thing?",
+            "options": ["approve", "reject"],
+            "choice": "reject",
+        }
+
+    def test_gate_block_emitted_only_when_run_rests_at_gate(self):
+        # A run rests *on* a gate only while `paused` (awaiting a decision) or
+        # `aborted` (gate rejected with on_reject: abort). current_step_id is
+        # not cleared afterwards, so a `completed`/`failed` run whose last
+        # executed step was a gate must NOT surface a stale gate block.
+        from types import SimpleNamespace
+        from specify_cli import _gate_outcome
+
+        gate_step = {
+            "type": "gate",
+            "output": {
+                "message": "m",
+                "options": ["approve", "reject"],
+                "choice": "reject",
+            },
+        }
+
+        def _state(status):
+            return SimpleNamespace(
+                status=SimpleNamespace(value=status),
+                current_step_id="review",
+                step_results={"review": gate_step},
+            )
+
+        assert _gate_outcome(_state("completed")) is None
+        assert _gate_outcome(_state("failed")) is None
+        assert _gate_outcome(_state("paused")) is not None
+        assert _gate_outcome(_state("aborted")) is not None
+
+    def test_gate_block_message_coerced_to_string(self):
+        # message may be a non-string YAML literal (e.g. a number); the JSON
+        # surface normalises it so the emitted schema stays stable.
+        from types import SimpleNamespace
+        from specify_cli import _gate_outcome
+
+        state = SimpleNamespace(
+            status=SimpleNamespace(value="paused"),
+            current_step_id="review",
+            step_results={
+                "review": {
+                    "type": "gate",
+                    "output": {"message": 12.5, "options": ["ok"], "choice": None},
+                }
+            },
+        )
+        assert _gate_outcome(state)["message"] == "12.5"