feat: bind gate verdict to workflow input via verdict_input (#3725)

* feat: bind gate verdict to workflow input via verdict_input

Add an optional `verdict_input` field to gate steps that lets an
external system supply a verdict through a declared workflow input
instead of an interactive TTY prompt.

When the referenced input carries a non-empty string value that
matches one of the gate's `options` (case-insensitive), the gate
auto-decides, records the matched spelling in `output.choice`, and
applies the existing `on_reject` / abort / skip / retry semantics.
If the value is present but does not match an option, or is a
non-string, the gate fails immediately with a clear error message.
When the input is absent, null, or empty, the gate falls back to
today's TTY-prompt-or-pause behaviour unchanged.

The engine now persists `result.error` alongside each step's status
and output so that failed-step error messages survive across runs.
The CLI (`workflow run` and `workflow resume`) surfaces these
persisted errors after a failed or aborted run. `validate_workflow`
cross-references `verdict_input` against the workflow's declared
inputs block and reports an error for undeclared names, consistent
with the existing `wait_for` id cross-check.

Closes discussion: https://github.com/github/spec-kit/discussions/3717

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Assisted-by: GitHub Copilot (model: claude-sonnet-4.6, autonomous)

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Fix workflow JSON error payloads

Include persisted step errors in _workflow_run_payload so workflow run/resume/status --json all surface failure reasons consistently. Add JSON-path tests for failed and successful runs.

Assisted-by: GitHub Copilot (model: gpt-5.3-codex, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* fix(workflows): reject verdict inputs in fan-out

Fan-out items share workflow inputs and cannot safely consume a bound gate verdict. Reject verdict_input bindings during validation and at runtime while preserving unbound gates.

Assisted-by: GitHub Copilot (model: GPT-5.6 Sol, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix: update workflow command handling

Assisted-by: GitHub Copilot (model: gpt-5.3-codex, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

---------

Co-authored-by: Markus <markus@example.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

docs/reference/workflows.md

@@ -39,6 +39,21 @@ specify workflow run my-pipeline.yml --json
 
 `workflow_id` is the `workflow.id` declared inside the YAML, not the file name. The object is printed exactly as shown — pretty-printed with two-space indentation, on plain stdout with no Rich markup — so it always parses. While the workflow runs under `--json`, any progress a step would print (for example a gate prompt, or output from a prompt step's CLI subprocess) is redirected to stderr, so stdout carries only the JSON object. Read the object from stdout; leave stderr attached to the terminal or capture it separately.
 
+For `failed` and `aborted` runs, the payload includes an `error` field carrying the terminal step's error message:
+
+```json
+{
+  "run_id": "662bf791",
+  "workflow_id": "build-and-review",
+  "status": "failed",
+  "current_step_id": "boom",
+  "current_step_index": 0,
+  "error": "Command exited with code 3"
+}
+```
+
+`completed` and `paused` runs omit the `error` field. The error is persisted in the run's `state.json`, so `specify workflow status <run_id> --json` surfaces the same message after the fact.
+
 > **Note:** Most workflow commands require a project already initialized with `specify init`. The exception is `specify workflow run <local-file.{yml,yaml}>`, which can run outside a project; in that case, run state is stored under the current directory's `.specify/workflows/runs/<run_id>/`.
 
 ## Resume a Workflow
@@ -554,6 +569,51 @@ Each workflow run persists its state at `.specify/workflows/runs/<run_id>/`:
 
 This enables `specify workflow resume` to continue from the exact step where a run was paused (e.g., at a gate) or failed.
 
+### Gate Verdict Inputs
+
+`verdict_input` binds a gate's verdict to a named workflow input. The input must be declared in the workflow's `inputs` block; `specify workflow validate` reports an undeclared reference.
+
+`verdict_input` is not supported inside a `fan-out` template. Fan-out items
+share workflow inputs, while workflow state can represent only one paused
+gate. Place a gate before the fan-out to approve the whole batch, or after a
+fan-in to review the aggregated results.
+
+**Input value semantics:**
+
+| Value | Behavior |
+|---|---|
+| Non-empty string, matches an option (case-insensitive) | Gate auto-decides; `output.choice` is set to the configured option spelling |
+| Non-empty string, no match | Gate fails immediately |
+| Non-string | Gate fails immediately |
+| Missing or empty | Gate prompts on a TTY; pauses otherwise |
+
+**Default value semantics:** A non-empty `default` is consumed as a verdict on the first run — matching an option auto-decides the gate, not matching fails it immediately.
+
+```yaml
+inputs:
+  spec_verdict:
+    type: string
+    default: ""
+steps:
+  - id: review-spec
+    type: gate
+    message: "Approve the specification?"
+    options: [approve, reject]
+    on_reject: retry
+    verdict_input: spec_verdict
+```
+
+Supply a verdict when resuming:
+
+```bash
+specify workflow resume <run_id> --input spec_verdict=approve
+```
+
+For `on_reject: retry`, a bound reject verdict is consumed before the gate
+pauses: the named stored input is reset to `""`. A later resume therefore
+prompts or pauses again until another verdict is supplied. Approve, abort, and
+skip outcomes leave the input unchanged.
+
 ## FAQ
 
 ### What happens when a workflow hits a gate step?

src/specify_cli/workflows/_commands.py

@@ -889,6 +889,18 @@ def _require_specify_project(*args, **kwargs):
     return project_root
 
 
+def _failed_step_error(state: Any) -> str | None:
+    """Terminal error for a failed/aborted run, if any.
+
+    Returns the run-level error persisted by the engine at the moment
+    the run terminated. Returns ``None`` for non-terminal statuses so
+    the caller can print unconditionally.
+    """
+    if getattr(state.status, "value", state.status) not in ("failed", "aborted"):
+        return None
+    return getattr(state, "error", None)
+
+
 def _workflow_run_payload(state: Any) -> dict[str, Any]:
     """Machine-readable summary of a run/resume outcome."""
     payload = {
@@ -901,6 +913,9 @@ def _workflow_run_payload(state: Any) -> dict[str, Any]:
     gate = _gate_outcome(state)
     if gate is not None:
         payload["gate"] = gate
+    error = _failed_step_error(state)
+    if error is not None:
+        payload["error"] = error
     return payload
 
 
@@ -1161,6 +1176,10 @@ def workflow_run(
     console.print(f"\n[{color}]Status: {state.status.value}[/{color}]")
     console.print(f"[dim]Run ID: {state.run_id}[/dim]")
 
+    err_msg = _failed_step_error(state)
+    if err_msg:
+        console.print(f"[red]Error:[/red] {_escape_markup(err_msg)}")
+
     if state.status.value == "paused":
         console.print(f"\nResume with: [cyan]specify workflow resume {state.run_id}[/cyan]")
 
@@ -1271,6 +1290,10 @@ def workflow_resume(
     color = status_colors.get(state.status.value, "white")
     console.print(f"\n[{color}]Status: {state.status.value}[/{color}]")
 
+    err_msg = _failed_step_error(state)
+    if err_msg:
+        console.print(f"[red]Error:[/red] {_escape_markup(err_msg)}")
+
     raise typer.Exit(_run_outcome_exit_code(state.status.value))
 
 
@@ -1338,6 +1361,10 @@ def workflow_status(
         if state.current_step_id:
             console.print(f"  Current:  {state.current_step_id}")
 
+        err_msg = _failed_step_error(state)
+        if err_msg:
+            console.print(f"  [red]Error:    {_escape_markup(err_msg)}[/red]")
+
         if state.step_results:
             console.print(f"\n  [bold]Steps ({len(state.step_results)}):[/bold]")
             for step_id, step_data in state.step_results.items():

src/specify_cli/workflows/base.py

@@ -56,6 +56,9 @@ class StepContext:
     #: Current fan-out item (set only inside fan-out iterations).
     item: Any = None
 
+    #: Whether the current step is executing inside a fan-out template.
+    inside_fan_out: bool = False
+
     #: Fan-in aggregated results (set only for fan-in steps).
     fan_in: dict[str, Any] = field(default_factory=dict)
 

src/specify_cli/workflows/engine.py

@@ -308,7 +308,15 @@ def validate_workflow(definition: WorkflowDefinition) -> list[str]:
         errors.append("Workflow has no steps defined.")
 
     seen_ids: set[str] = set()
-    _validate_steps(definition.steps, seen_ids, errors)
+    # ``input_names`` is the set of declared workflow input names — used by
+    # ``_validate_steps`` to cross-reference gate ``verdict_input`` bindings.
+    # ``None`` means the inputs block itself is malformed (already reported
+    # above); the cross-check is then disabled so one authoring mistake does
+    # not cascade into N spurious "undeclared" errors.
+    input_names: set[str] | None = (
+        set(definition.inputs) if isinstance(definition.inputs, dict) else None
+    )
+    _validate_steps(definition.steps, seen_ids, errors, input_names)
 
     return errors
 
@@ -317,8 +325,16 @@ def _validate_steps(
     steps: list[dict[str, Any]],
     seen_ids: set[str],
     errors: list[str],
+    input_names: set[str] | None = None,
+    inside_fan_out: bool = False,
 ) -> None:
-    """Recursively validate a list of steps."""
+    """Recursively validate a list of steps.
+
+    ``input_names`` is the set of declared workflow input names (or ``None``
+    when the inputs block is malformed). ``inside_fan_out`` is threaded
+    through nested control-flow steps so gate verdict bindings can be rejected
+    anywhere inside a fan-out template.
+    """
     from . import STEP_REGISTRY
 
     for step_config in steps:
@@ -411,30 +427,73 @@ def _validate_steps(
                             f"unknown or not-yet-declared step id {wid!r}."
                         )
 
+        # Gate verdict_input: fan-out items cannot bind shared workflow inputs
+        # as per-item verdicts. Outside fan-out, the binding must reference a
+        # declared workflow input because ``_resolve_inputs`` drops undeclared
+        # names at both initial run and resume. Only check a non-empty string;
+        # malformed shapes are already reported by ``GateStep.validate()``.
+        if step_type == "gate":
+            verdict_input = step_config.get("verdict_input")
+            if isinstance(verdict_input, str) and verdict_input:
+                if inside_fan_out:
+                    errors.append(
+                        f"Gate step {step_id!r}: 'verdict_input' is not "
+                        "supported inside fan-out templates."
+                    )
+                elif input_names is not None and verdict_input not in input_names:
+                    errors.append(
+                        f"Gate step {step_id!r}: 'verdict_input' references "
+                        f"undeclared input {verdict_input!r}."
+                    )
+
         # Recursively validate nested steps
         for nested_key in ("then", "else", "steps"):
             nested = step_config.get(nested_key)
             if isinstance(nested, list):
-                _validate_steps(nested, seen_ids, errors)
+                _validate_steps(
+                    nested,
+                    seen_ids,
+                    errors,
+                    input_names,
+                    inside_fan_out=inside_fan_out,
+                )
 
         # Validate switch cases
         cases = step_config.get("cases")
         if isinstance(cases, dict):
             for _case_key, case_steps in cases.items():
                 if isinstance(case_steps, list):
-                    _validate_steps(case_steps, seen_ids, errors)
+                    _validate_steps(
+                        case_steps,
+                        seen_ids,
+                        errors,
+                        input_names,
+                        inside_fan_out=inside_fan_out,
+                    )
 
         # Validate switch default
         default = step_config.get("default")
         if isinstance(default, list):
-            _validate_steps(default, seen_ids, errors)
+            _validate_steps(
+                default,
+                seen_ids,
+                errors,
+                input_names,
+                inside_fan_out=inside_fan_out,
+            )
 
         # Validate fan-out nested step (template — not added to seen_ids
         # since the engine generates parentId:templateId:index at runtime)
         fan_step = step_config.get("step")
         if isinstance(fan_step, dict):
             fan_errors: list[str] = []
-            _validate_steps([fan_step], set(), fan_errors)
+            _validate_steps(
+                [fan_step],
+                set(),
+                fan_errors,
+                input_names,
+                inside_fan_out=True,
+            )
             errors.extend(fan_errors)
 
 
@@ -560,6 +619,7 @@ class RunState:
         self.created_at = datetime.now(timezone.utc).isoformat()
         self.updated_at = self.created_at
         self.log_entries: list[dict[str, Any]] = []
+        self.error: str | None = None
 
     @property
     def runs_dir(self) -> Path:
@@ -614,6 +674,7 @@ class RunState:
                 "workflow_dir": self.workflow_dir,
                 "created_at": self.created_at,
                 "updated_at": self.updated_at,
+                "error": self.error,
             }
             self._atomic_write_json(runs_dir / "state.json", state_data)
             self._atomic_write_json(runs_dir / "inputs.json", {"inputs": self.inputs})
@@ -707,6 +768,7 @@ class RunState:
         state.workflow_dir = state_data.get("workflow_dir")
         state.created_at = state_data.get("created_at", "")
         state.updated_at = state_data.get("updated_at", "")
+        state.error = state_data.get("error")
 
         inputs_path = runs_dir / "inputs.json"
         if inputs_path.exists():
@@ -901,6 +963,7 @@ class WorkflowEngine:
             return state
         except Exception as exc:
             state.status = RunStatus.FAILED
+            state.error = str(exc)
             state.append_log({"event": "workflow_failed", "error": str(exc)})
             state.save()
             raise
@@ -959,6 +1022,7 @@ class WorkflowEngine:
 
         from . import STEP_REGISTRY
 
+        state.error = None
         state.status = RunStatus.RUNNING
         state.save()
 
@@ -979,6 +1043,7 @@ class WorkflowEngine:
             return state
         except Exception as exc:
             state.status = RunStatus.FAILED
+            state.error = str(exc)
             state.append_log({"event": "resume_failed", "error": str(exc)})
             state.save()
             raise
@@ -1038,6 +1103,7 @@ class WorkflowEngine:
             step_impl = registry.get(step_type)
             if not step_impl:
                 state.status = RunStatus.FAILED
+                state.error = f"Unknown step type: {step_type!r}"
                 state.append_log(
                     {
                         "event": "step_failed",
@@ -1065,6 +1131,7 @@ class WorkflowEngine:
                 or step_config.get("input", {}),
                 "output": result.output,
                 "status": result.status.value,
+                "error": result.error,
             }
             self._record_result(context, state, step_id, step_data)
 
@@ -1090,6 +1157,7 @@ class WorkflowEngine:
                 # is for transient/expected step failures only.
                 if result.output.get("aborted"):
                     state.status = RunStatus.ABORTED
+                    state.error = result.error
                     state.append_log(
                         {
                             "event": "workflow_aborted",
@@ -1132,6 +1200,7 @@ class WorkflowEngine:
                     continue
 
                 state.status = RunStatus.FAILED
+                state.error = result.error
                 state.append_log(
                     {
                         "event": "step_failed",
@@ -1305,11 +1374,18 @@ class WorkflowEngine:
         # Sequential path — identical to the historical behavior.
         if workers <= 1:
             results: list[Any] = []
-            for item_idx, item_val in enumerate(items):
-                context.item = item_val
-                results.append(run_item(item_idx, context))
-                if state.status in halting:
-                    break
+            previous_item = context.item
+            previous_inside_fan_out = context.inside_fan_out
+            context.inside_fan_out = True
+            try:
+                for item_idx, item_val in enumerate(items):
+                    context.item = item_val
+                    results.append(run_item(item_idx, context))
+                    if state.status in halting:
+                        break
+            finally:
+                context.item = previous_item
+                context.inside_fan_out = previous_inside_fan_out
             return results
 
         # Concurrent path — bounded sliding window; results assembled in item order.
@@ -1320,7 +1396,14 @@ class WorkflowEngine:
             # Each item runs against its own context copy so context.item is not
             # clobbered across threads; the shared steps dict is written only on the
             # disjoint parentId:templateId:index key (GIL-safe on distinct keys).
-            return run_item(idx, dataclasses.replace(context, item=items[idx]))
+            return run_item(
+                idx,
+                dataclasses.replace(
+                    context,
+                    item=items[idx],
+                    inside_fan_out=True,
+                ),
+            )
 
         def item_halt_status(idx: int) -> RunStatus | None:
             # If THIS item's own execution halted the run, return the resulting run
@@ -1401,6 +1484,16 @@ class WorkflowEngine:
             # pool joined; restore the halting item's own outcome so the final run
             # status matches the sequential semantics.
             state.status = halted_status
+            # Restore the halting item's error so it matches the terminal
+            # status — a concurrent item may have overwritten state.error
+            # before the pool joined. Assign unconditionally when a record
+            # exists (even when the halting item's own error is falsy) so a
+            # third-party step returning FAILED with no message never inherits
+            # an unrelated concurrent item's error; this mirrors the sequential
+            # path, which sets state.error = result.error verbatim.
+            halt_rec = context.steps.get(item_id(halted_at))
+            if isinstance(halt_rec, dict):
+                state.error = halt_rec.get("error")
             return slots[: halted_at + 1]
         return slots[:collected]
 

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

@@ -26,7 +26,8 @@ class GateStep(StepBase):
     later with ``specify workflow resume``.
 
     The user's choice is stored in ``output.choice``.  ``on_reject``
-    controls abort / skip / retry behaviour.
+    controls abort / skip / retry behaviour.  ``verdict_input`` can name a
+    workflow input to use as the choice when resuming non-interactively.
     """
 
     type_key = "gate"
@@ -42,6 +43,8 @@ class GateStep(StepBase):
 
         options = config.get("options", ["approve", "reject"])
         on_reject = config.get("on_reject", "abort")
+        has_verdict_input = "verdict_input" in config
+        verdict_input = config.get("verdict_input")
 
         # ``validate`` rejects a non-list (or empty) ``options``, and requires
         # every option to be a string, but the engine does not auto-validate
@@ -72,6 +75,26 @@ class GateStep(StepBase):
                 },
             )
 
+        if has_verdict_input and (
+            not isinstance(verdict_input, str) or not verdict_input
+        ):
+            return StepResult(
+                status=StepStatus.FAILED,
+                error=(
+                    f"Gate step {config.get('id', '?')!r}: 'verdict_input' must be "
+                    "a non-empty string."
+                ),
+            )
+
+        if has_verdict_input and context.inside_fan_out:
+            return StepResult(
+                status=StepStatus.FAILED,
+                error=(
+                    f"Gate step {config.get('id', '?')!r}: 'verdict_input' is "
+                    "not supported inside fan-out templates."
+                ),
+            )
+
         show_file = config.get("show_file")
         if isinstance(show_file, str) and "{{" in show_file:
             show_file = evaluate_expression(show_file, context)
@@ -90,16 +113,48 @@ class GateStep(StepBase):
             "choice": None,
         }
 
-        # Non-interactive: pause for later resume (the file is not read here)
-        if not sys.stdin.isatty():
-            return StepResult(status=StepStatus.PAUSED, output=output)
+        choice: str | None = None
+        bound_verdict_input: str | None = None
+        if verdict_input is not None:
+            value = context.inputs.get(verdict_input)
+            if value is not None and value != "":
+                if not isinstance(value, str):
+                    return StepResult(
+                        status=StepStatus.FAILED,
+                        output=output,
+                        error=(
+                            f"Gate step {config.get('id', '?')!r}: verdict input "
+                            f"{verdict_input!r} must be a string, got "
+                            f"{type(value).__name__}."
+                        ),
+                    )
+                choice = next(
+                    (option for option in options if option.lower() == value.lower()),
+                    None,
+                )
+                if choice is None:
+                    return StepResult(
+                        status=StepStatus.FAILED,
+                        output=output,
+                        error=(
+                            f"Gate step {config.get('id', '?')!r}: verdict input "
+                            f"{verdict_input!r} value {value!r} does not match any "
+                            "configured option."
+                        ),
+                    )
+                bound_verdict_input = verdict_input
 
-        # Interactive: prompt the user. ``show_file`` contents are folded
-        # into the displayed message so the operator can review the
-        # referenced material before choosing. Composing the prompt text
-        # here keeps ``_prompt`` to its ``(message, options)`` contract, so
-        # adding review material never widens the interactive seam.
-        choice = self._prompt(self._compose_prompt(message, show_file), options)
+        if choice is None:
+            # Non-interactive: pause for later resume (the file is not read here)
+            if not sys.stdin.isatty():
+                return StepResult(status=StepStatus.PAUSED, output=output)
+
+            # Interactive: prompt the user. ``show_file`` contents are folded
+            # into the displayed message so the operator can review the
+            # referenced material before choosing. Composing the prompt text
+            # here keeps ``_prompt`` to its ``(message, options)`` contract, so
+            # adding review material never widens the interactive seam.
+            choice = self._prompt(self._compose_prompt(message, show_file), options)
         output["choice"] = choice
 
         # Match rejection case-insensitively. ``_prompt`` echoes the option's
@@ -119,6 +174,8 @@ class GateStep(StepBase):
                 )
             if on_reject == "retry":
                 # Pause so the next resume re-executes this gate
+                if bound_verdict_input is not None:
+                    context.inputs[bound_verdict_input] = ""
                 return StepResult(status=StepStatus.PAUSED, output=output)
             # on_reject == "skip" → completed, downstream steps decide
             return StepResult(status=StepStatus.COMPLETED, output=output)
@@ -234,6 +291,13 @@ class GateStep(StepBase):
                 f"Gate step {config.get('id', '?')!r}: 'on_reject' must be "
                 f"'abort', 'skip', or 'retry'."
             )
+        if "verdict_input" in config and (
+            not isinstance(config["verdict_input"], str) or not config["verdict_input"]
+        ):
+            errors.append(
+                f"Gate step {config.get('id', '?')!r}: 'verdict_input' must be "
+                "a non-empty string."
+            )
         # Only inspect option text when every option is a string; otherwise the
         # `o.lower()` below would raise AttributeError on a non-string option
         # (already reported above) and break validate_workflow's never-raise contract.

tests/test_workflows.py

@@ -2352,6 +2352,185 @@ class TestGateStep:
         assert result.output["message"] == "Review the spec."
         assert result.output["options"] == ["approve", "reject"]
 
+    @pytest.mark.parametrize(
+        "inputs", [{}, {"spec_verdict": None}, {"spec_verdict": ""}]
+    )
+    def test_missing_or_empty_verdict_input_uses_existing_pause_behavior(self, inputs):
+        from specify_cli.workflows.steps.gate import GateStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        result = GateStep().execute(
+            {
+                "id": "review",
+                "message": "Review the spec.",
+                "options": ["approve", "reject"],
+                "verdict_input": "spec_verdict",
+            },
+            StepContext(inputs=inputs),
+        )
+        assert result.status == StepStatus.PAUSED
+        assert result.output["choice"] is None
+
+    @pytest.mark.parametrize("inputs", [{}, {"spec_verdict": ""}])
+    def test_missing_or_empty_verdict_input_prompts_on_tty(self, monkeypatch, inputs):
+        from specify_cli.workflows.steps.gate import GateStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        _force_gate_stdin(monkeypatch, tty=True)
+        monkeypatch.setattr(
+            GateStep, "_prompt", staticmethod(lambda _message, _options: "approve")
+        )
+        result = GateStep().execute(
+            {
+                "id": "review",
+                "message": "Review the spec.",
+                "options": ["approve", "reject"],
+                "verdict_input": "spec_verdict",
+            },
+            StepContext(inputs=inputs),
+        )
+        assert result.status == StepStatus.COMPLETED
+        assert result.output["choice"] == "approve"
+
+    def test_verdict_input_uses_canonical_option_spelling(self):
+        from specify_cli.workflows.steps.gate import GateStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        result = GateStep().execute(
+            {
+                "id": "review",
+                "message": "Review the spec.",
+                "options": ["Approve", "Reject"],
+                "verdict_input": "spec_verdict",
+            },
+            StepContext(inputs={"spec_verdict": "aPpRoVe"}),
+        )
+        assert result.status == StepStatus.COMPLETED
+        assert result.output["choice"] == "Approve"
+
+    @pytest.mark.parametrize(
+        ("value", "error_fragment"),
+        [(42, "must be a string"), ("maybe", "does not match")],
+    )
+    def test_invalid_verdict_input_value_fails(self, value, error_fragment):
+        from specify_cli.workflows.steps.gate import GateStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        result = GateStep().execute(
+            {
+                "id": "review",
+                "message": "Review the spec.",
+                "options": ["approve", "reject"],
+                "verdict_input": "spec_verdict",
+            },
+            StepContext(inputs={"spec_verdict": value}),
+        )
+        assert result.status == StepStatus.FAILED
+        assert error_fragment in (result.error or "")
+
+    def test_verdict_input_fails_inside_fan_out_context(self):
+        from specify_cli.workflows.steps.gate import GateStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        result = GateStep().execute(
+            {
+                "id": "review",
+                "message": "Review the item.",
+                "options": ["approve", "reject"],
+                "verdict_input": "spec_verdict",
+            },
+            StepContext(
+                inputs={"spec_verdict": "approve"},
+                inside_fan_out=True,
+            ),
+        )
+
+        assert result.status == StepStatus.FAILED
+        assert "'verdict_input' is not supported inside fan-out" in (
+            result.error or ""
+        )
+
+    @pytest.mark.parametrize(
+        ("value", "error_fragment"),
+        [(42, "must be a string"), ("maybe", "does not match")],
+    )
+    def test_failed_gate_persists_error_in_step_results(
+        self, tmp_path, value, error_fragment
+    ):
+        """Engine persists result.error into step_results for failed gates."""
+        from specify_cli.workflows.engine import WorkflowEngine
+
+        wf_yaml = f"""
+schema_version: "1.0"
+workflow:
+  id: "gate-error-persist"
+  name: "Gate Error Persist"
+  version: "1.0.0"
+inputs:
+  spec_verdict:
+    type: {"number" if isinstance(value, int) else "string"}
+    default: {value}
+steps:
+  - id: review
+    type: gate
+    message: "Review the spec."
+    options: [approve, reject]
+    on_reject: abort
+    verdict_input: spec_verdict
+"""
+        wf_path = tmp_path / "wf.yml"
+        wf_path.write_text(wf_yaml, encoding="utf-8")
+        (tmp_path / ".specify").mkdir()
+        engine = WorkflowEngine(tmp_path)
+        definition = engine.load_workflow(str(wf_path))
+        state = engine.execute(definition, {})
+        assert state.status.value == "failed"
+        step_data = state.step_results["review"]
+        assert step_data["status"] == "failed"
+        assert error_fragment in (step_data.get("error") or "")
+
+    @pytest.mark.parametrize("invalid_value", ["", 42, None])
+    def test_validate_invalid_verdict_input(self, invalid_value):
+        from specify_cli.workflows.steps.gate import GateStep
+
+        errors = GateStep().validate({
+            "id": "review",
+            "message": "Review the spec.",
+            "verdict_input": invalid_value,
+        })
+        assert any("verdict_input" in error for error in errors)
+
+    @pytest.mark.parametrize(
+        ("on_reject", "status", "aborted"),
+        [
+            ("abort", "failed", True),
+            ("skip", "completed", False),
+            ("retry", "paused", False),
+        ],
+    )
+    def test_reject_verdict_input_preserves_reject_behavior(
+        self, on_reject, status, aborted
+    ):
+        from specify_cli.workflows.steps.gate import GateStep
+        from specify_cli.workflows.base import StepContext
+
+        context = StepContext(inputs={"spec_verdict": "reject"})
+        result = GateStep().execute(
+            {
+                "id": "review",
+                "message": "Review the spec.",
+                "options": ["approve", "reject"],
+                "on_reject": on_reject,
+                "verdict_input": "spec_verdict",
+            },
+            context,
+        )
+        assert result.status.value == status
+        assert result.output.get("aborted", False) is aborted
+        assert context.inputs["spec_verdict"] == (
+            "" if on_reject == "retry" else "reject"
+        )
+
     def test_validate_missing_message(self):
         from specify_cli.workflows.steps.gate import GateStep
 
@@ -3630,6 +3809,36 @@ class TestFanOutConcurrency:
         results, _ = self._run(tmp_path, items, 6)
         assert [r["seen"]["id"] for r in results] == [f"x{i}" for i in range(6)]
 
+    @pytest.mark.parametrize("max_concurrency", [1, 2])
+    def test_marks_item_context_as_inside_fan_out(self, tmp_path, max_concurrency):
+        from specify_cli.workflows.base import StepBase, StepResult, StepStatus
+
+        class _ContextProbeStep(StepBase):
+            type_key = "context-probe"
+
+            def execute(self, config, context):
+                return StepResult(
+                    status=StepStatus.COMPLETED,
+                    output={"inside_fan_out": context.inside_fan_out},
+                )
+
+        engine, context, state, _registry, _template = self._build(tmp_path)
+        results = engine._run_fan_out(
+            ["a", "b"],
+            {"id": "probe", "type": "context-probe"},
+            "fan",
+            context,
+            state,
+            {"context-probe": _ContextProbeStep()},
+            max_concurrency,
+        )
+
+        assert results == [
+            {"inside_fan_out": True},
+            {"inside_fan_out": True},
+        ]
+        assert context.inside_fan_out is False
+
     def test_empty_items(self, tmp_path):
         results, _ = self._run(tmp_path, [], 4)
         assert results == []
@@ -3675,6 +3884,56 @@ class TestFanOutConcurrency:
         assert results == [{"seen": 0}, {"seen": 1}, {"seen": 2}]
         assert state.status == RunStatus.FAILED
 
+    def test_concurrent_restores_halting_item_error(self, tmp_path):
+        # After a concurrent fan-out halts, the run-level error must be the first
+        # halting item's own error (parity with the sequential path), even when a
+        # later concurrent item failed with a different error AND the halting
+        # item's error is falsy. Covers the pool-join restore branch, which must
+        # assign unconditionally rather than skip a falsy value.
+        from specify_cli.workflows.base import (
+            RunStatus,
+            StepBase,
+            StepResult,
+            StepStatus,
+        )
+
+        class _ErrorProbe(StepBase):
+            type_key = "err-probe"
+
+            def execute(self, config, context):
+                item = context.item
+                if item == "halt":
+                    # First failing item in item order; empty (falsy) error.
+                    return StepResult(
+                        status=StepStatus.FAILED, error="", output={}
+                    )
+                if item == "leak":
+                    return StepResult(
+                        status=StepStatus.FAILED,
+                        error="leaked-error",
+                        output={},
+                    )
+                return StepResult(
+                    status=StepStatus.COMPLETED, output={"seen": item}
+                )
+
+        engine, context, state, _registry, _template = self._build(tmp_path)
+        engine._run_fan_out(
+            ["ok0", "halt", "ok2", "leak"],
+            {"id": "impl", "type": "err-probe"},
+            "fan",
+            context,
+            state,
+            {"err-probe": _ErrorProbe()},
+            4,
+        )
+
+        assert state.status == RunStatus.FAILED
+        # Halt is attributed to "halt" (index 1). Its empty error must win over
+        # the later "leak" item's error — the restore assigns the halting item's
+        # error verbatim, even when falsy.
+        assert state.error == ""
+
     def test_continue_on_error_item_does_not_halt_concurrent(self, tmp_path):
         # A failing item whose template sets continue_on_error must NOT truncate
         # the fan-out: every item still runs and is returned in order.
@@ -4314,6 +4573,252 @@ steps:
         assert not any("requires" in e for e in errors)
 
 
+class TestGateVerdictInputValidation:
+    """Gate verdict_input must reference a declared workflow input.
+
+    ``_resolve_inputs`` iterates only over ``definition.inputs`` — a provided
+    value for an undeclared name is silently dropped at both initial run and
+    resume. So an undeclared ``verdict_input`` can never receive a value; the
+    gate would pause forever. Surface this wiring error at validation time.
+    """
+
+    @staticmethod
+    def _errors(yaml_text):
+        from specify_cli.workflows.engine import (
+            WorkflowDefinition,
+            validate_workflow,
+        )
+
+        return validate_workflow(WorkflowDefinition.from_string(yaml_text))
+
+    def test_undeclared_verdict_input_is_rejected(self):
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+inputs:
+  spec_verdict:
+    type: string
+    default: ""
+steps:
+  - id: review
+    type: gate
+    message: "Review?"
+    options: [approve, reject]
+    verdict_input: spec_verdit
+""")
+        assert any(
+            "'verdict_input' references undeclared input 'spec_verdit'" in e
+            for e in errors
+        )
+
+    def test_declared_verdict_input_passes(self):
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+inputs:
+  spec_verdict:
+    type: string
+    default: ""
+steps:
+  - id: review
+    type: gate
+    message: "Review?"
+    options: [approve, reject]
+    verdict_input: spec_verdict
+""")
+        assert not any("verdict_input" in e for e in errors)
+
+    def test_gate_without_verdict_input_passes(self):
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+steps:
+  - id: review
+    type: gate
+    message: "Review?"
+    options: [approve, reject]
+""")
+        assert not any("verdict_input" in e for e in errors)
+
+    def test_malformed_verdict_input_no_duplicate_error(self):
+        # Non-string verdict_input is already reported by GateStep.validate();
+        # the cross-reference check must not pile on a confusing duplicate.
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+inputs:
+  spec_verdict:
+    type: string
+    default: ""
+steps:
+  - id: review
+    type: gate
+    message: "Review?"
+    options: [approve, reject]
+    verdict_input: 123
+""")
+        # Shape error from GateStep.validate()
+        assert any("verdict_input" in e and "non-empty string" in e for e in errors)
+        # No undeclared-input error (123 is not a string, so cross-check skips)
+        assert not any("undeclared input" in e for e in errors)
+
+    def test_verdict_input_in_switch_case(self):
+        # Recursion coverage: bad reference inside a switch case must surface.
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+steps:
+  - id: branch
+    type: switch
+    expression: "{{ inputs.flag }}"
+    cases:
+      yes:
+        - id: review
+          type: gate
+          message: "Review?"
+          options: [approve, reject]
+          verdict_input: ghost_input
+""")
+        assert any(
+            "'verdict_input' references undeclared input 'ghost_input'" in e
+            for e in errors
+        )
+
+    def test_verdict_input_in_if_branch(self):
+        # Recursion coverage: bad reference inside an if-then branch.
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+steps:
+  - id: maybe
+    type: if
+    condition: "{{ inputs.flag }}"
+    then:
+      - id: review
+        type: gate
+        message: "Review?"
+        options: [approve, reject]
+        verdict_input: ghost_input
+""")
+        assert any(
+            "'verdict_input' references undeclared input 'ghost_input'" in e
+            for e in errors
+        )
+
+    def test_verdict_input_in_fan_out_template(self):
+        # Fan-out items share workflow inputs, so a bound verdict would be
+        # consumed by multiple item gates with undefined pause/resume semantics.
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+inputs:
+  spec_verdict:
+    type: string
+    default: ""
+steps:
+  - id: fan
+    type: fan-out
+    items: [a, b]
+    step:
+      id: review
+      type: gate
+      message: "Review?"
+      options: [approve, reject]
+      verdict_input: spec_verdict
+""")
+        assert any(
+            "'verdict_input' is not supported inside fan-out templates" in e
+            for e in errors
+        )
+
+    def test_verdict_input_nested_inside_fan_out_template(self):
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+inputs:
+  spec_verdict:
+    type: string
+    default: ""
+steps:
+  - id: fan
+    type: fan-out
+    items: [a, b]
+    step:
+      id: maybe-review
+      type: if
+      condition: "{{ item }}"
+      then:
+        - id: review
+          type: gate
+          message: "Review?"
+          options: [approve, reject]
+          verdict_input: spec_verdict
+""")
+        assert any(
+            "'verdict_input' is not supported inside fan-out templates" in e
+            for e in errors
+        )
+
+    def test_gate_without_verdict_input_in_fan_out_template_passes(self):
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+steps:
+  - id: fan
+    type: fan-out
+    items: [a, b]
+    step:
+      id: review
+      type: gate
+      message: "Review?"
+      options: [approve, reject]
+""")
+        assert not any(
+            "not supported inside fan-out templates" in e for e in errors
+        )
+
+    def test_malformed_inputs_block_no_cascade(self):
+        # When the inputs block itself is malformed (already reported), the
+        # cross-check is disabled so one authoring mistake does not cascade
+        # into N spurious "undeclared" errors.
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+inputs:
+  - not_a_mapping
+steps:
+  - id: review
+    type: gate
+    message: "Review?"
+    options: [approve, reject]
+    verdict_input: spec_verdict
+""")
+        # Inputs-shape error is reported
+        assert any("'inputs' must be a mapping" in e for e in errors)
+        # No cascade of undeclared-input errors
+        assert not any("undeclared input" in e for e in errors)
+
+
 # ===== Workflow Engine Tests =====
 
 class TestWorkflowEngine:
@@ -6157,6 +6662,74 @@ steps:
         assert state.status == RunStatus.FAILED
         assert "should-not-run" not in state.step_results
 
+    def test_continue_on_error_failure_not_surfaced_as_terminal_error(
+        self, project_dir
+    ):
+        """A continue_on_error step's error must not be reported as the
+        terminal run error when a later step fails for a different reason.
+
+        Regression test: the engine sets state.error only at terminal
+        branches, not in the continue_on_error branch. A handled failure
+        must not leak into the run-level error.
+        """
+        from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine
+        from specify_cli.workflows.base import RunStatus
+
+        definition = WorkflowDefinition.from_string("""
+schema_version: "1.0"
+workflow:
+  id: "coe-leak"
+  name: "COE Leak"
+  version: "1.0.0"
+steps:
+  - id: handled-failure
+    type: shell
+    run: "exit 42"
+    continue_on_error: true
+  - id: terminal-failure
+    type: shell
+    run: "exit 7"
+""")
+        engine = WorkflowEngine(project_dir)
+        state = engine.execute(definition)
+
+        assert state.status == RunStatus.FAILED
+        # The terminal error must be from the terminal-failure step, not
+        # the handled-failure step.
+        assert state.error is not None
+        assert "42" not in (state.error or "")
+        # The handled step's per-step error is still preserved.
+        assert state.step_results["handled-failure"]["status"] == "failed"
+        assert state.step_results["handled-failure"].get("error") is not None
+
+    def test_unknown_step_type_sets_run_error(self, project_dir):
+        """An unregistered step type fails the run with a descriptive
+        run-level error persisted on state.error.
+
+        The engine sets state.error at the unknown-step-type terminal
+        branch, mirroring the other terminal failure paths.
+        """
+        from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine
+        from specify_cli.workflows.base import RunStatus
+
+        # execute() bypasses validate_workflow(), which is what would
+        # otherwise reject the unknown type up front.
+        definition = WorkflowDefinition.from_string("""
+schema_version: "1.0"
+workflow:
+  id: "unknown-type"
+  name: "Unknown Type"
+  version: "1.0.0"
+steps:
+  - id: mystery
+    type: definitely-not-a-real-step
+""")
+        engine = WorkflowEngine(project_dir)
+        state = engine.execute(definition)
+
+        assert state.status == RunStatus.FAILED
+        assert state.error == "Unknown step type: 'definitely-not-a-real-step'"
+
 
 # ===== State Persistence Tests =====
 
@@ -6352,6 +6925,67 @@ class TestRunState:
         assert entry["event"] == "test_event"
         assert "timestamp" in entry
 
+    def test_error_persists_across_save_and_load(self, project_dir):
+        """Run-level error survives a save/load round trip."""
+        from specify_cli.workflows.engine import RunState
+        from specify_cli.workflows.base import RunStatus
+
+        state = RunState(
+            run_id="err-run",
+            workflow_id="test-wf",
+            project_root=project_dir,
+        )
+        state.status = RunStatus.FAILED
+        state.error = "Something went wrong"
+        state.save()
+
+        loaded = RunState.load("err-run", project_dir)
+        assert loaded.error == "Something went wrong"
+
+    def test_error_defaults_none_for_legacy_state(self, project_dir):
+        """Old state.json files without an error field load with error=None."""
+        from specify_cli.workflows.engine import RunState
+
+        state = RunState(
+            run_id="legacy-run",
+            workflow_id="test-wf",
+            project_root=project_dir,
+        )
+        state.save()
+
+        # Manually strip the error field to simulate a legacy state file.
+        state_path = state.runs_dir / "state.json"
+        data = json.loads(state_path.read_text())
+        data.pop("error", None)
+        state_path.write_text(json.dumps(data), encoding="utf-8")
+
+        loaded = RunState.load("legacy-run", project_dir)
+        assert loaded.error is None
+
+    def test_resume_clears_stale_error(self, project_dir):
+        """A resumed run starts with state.error = None."""
+        from specify_cli.workflows.engine import RunState
+        from specify_cli.workflows.base import RunStatus
+
+        state = RunState(
+            run_id="resume-err",
+            workflow_id="test-wf",
+            project_root=project_dir,
+        )
+        state.status = RunStatus.FAILED
+        state.error = "Previous failure"
+        state.save()
+
+        loaded = RunState.load("resume-err", project_dir)
+        assert loaded.error == "Previous failure"
+
+        loaded.error = None
+        loaded.status = RunStatus.RUNNING
+        loaded.save()
+
+        reloaded = RunState.load("resume-err", project_dir)
+        assert reloaded.error is None
+
 
 class TestListRuns:
     """Test listing workflow runs."""
@@ -9338,6 +9972,18 @@ steps:
     run: "echo done"
 """
 
+    _WF_FAIL = """
+schema_version: "1.0"
+workflow:
+  id: "json-fail"
+  name: "JSON Fail"
+  version: "1.0.0"
+steps:
+  - id: boom
+    type: shell
+    run: "exit 3"
+"""
+
     def _write_wf(self, project_dir, text, name):
         path = project_dir / f"{name}.yml"
         path.write_text(text, encoding="utf-8")
@@ -9370,6 +10016,45 @@ steps:
         assert payload["current_step_id"] == "ask"
         assert payload["current_step_index"] == 0
 
+    def test_run_json_failed_includes_error(self, project_dir):
+        # A run that ends in `failed` (a step failing, not an exception) must
+        # carry the persisted step error in the JSON payload so external
+        # callers get a reason, not a bare {"status": "failed"}.
+        wf = self._write_wf(project_dir, self._WF_FAIL, "boom")
+        result = self._invoke(project_dir, ["workflow", "run", str(wf), "--json"])
+        assert result.exit_code != 0
+        payload = json.loads(result.stdout)
+        assert payload["status"] == "failed"
+        assert payload.get("error")
+
+    def test_status_json_failed_includes_error(self, project_dir):
+        # `status --json` reuses the shared payload, so a failed run inspected
+        # after the fact surfaces the same error text as `run`/`resume`.
+        wf = self._write_wf(project_dir, self._WF_FAIL, "boom2")
+        rid = json.loads(
+            self._invoke(
+                project_dir, ["workflow", "run", str(wf), "--json"]
+            ).stdout
+        )["run_id"]
+        status = json.loads(
+            self._invoke(
+                project_dir, ["workflow", "status", rid, "--json"]
+            ).stdout
+        )
+        assert status["status"] == "failed"
+        assert status.get("error")
+
+    def test_run_json_completed_omits_error(self, project_dir):
+        # Successful runs must not carry an `error` key at all.
+        wf = self._write_wf(project_dir, self._WF_DONE, "noerr")
+        payload = json.loads(
+            self._invoke(
+                project_dir, ["workflow", "run", str(wf), "--json"]
+            ).stdout
+        )
+        assert payload["status"] == "completed"
+        assert "error" not in payload
+
     def test_run_json_output_has_no_markup_or_ansi(self, project_dir):
         wf = self._write_wf(project_dir, self._WF_DONE, "clean")
         out = self._invoke(
@@ -9495,6 +10180,25 @@ steps:
     options: [approve, reject]
 """
 
+    _WF_GATE_VERDICT = """
+schema_version: "1.0"
+workflow:
+  id: "resume-gate-verdict-wf"
+  name: "Resume Gate Verdict WF"
+  version: "1.0.0"
+inputs:
+  spec_verdict:
+    type: string
+    default: ""
+steps:
+  - id: gate
+    type: gate
+    message: "Review"
+    options: [approve, reject]
+    on_reject: retry
+    verdict_input: spec_verdict
+"""
+
     def _engine(self, project_dir):
         from specify_cli.workflows.engine import WorkflowEngine
         return WorkflowEngine(project_dir)
@@ -9556,6 +10260,31 @@ steps:
         with pytest.raises(ValueError):
             engine.resume(state.run_id, {"count": "not-a-number"})
 
+    def test_retry_verdict_input_is_consumed_and_can_be_replaced(self, project_dir):
+        import json as _json
+        from specify_cli.workflows.engine import WorkflowDefinition
+        from specify_cli.workflows.base import RunStatus
+
+        definition = WorkflowDefinition.from_string(self._WF_GATE_VERDICT)
+        engine = self._engine(project_dir)
+
+        state = engine.execute(definition, {"spec_verdict": "reject"})
+        assert state.status == RunStatus.PAUSED
+        assert state.inputs["spec_verdict"] == ""
+
+        inputs_file = (
+            project_dir / ".specify" / "workflows" / "runs" / state.run_id / "inputs.json"
+        )
+        assert _json.loads(inputs_file.read_text())["inputs"]["spec_verdict"] == ""
+
+        paused_again = engine.resume(state.run_id)
+        assert paused_again.status == RunStatus.PAUSED
+        assert paused_again.inputs["spec_verdict"] == ""
+
+        completed = engine.resume(state.run_id, {"spec_verdict": "approve"})
+        assert completed.status == RunStatus.COMPLETED
+        assert completed.step_results["gate"]["output"]["choice"] == "approve"
+
     def test_cli_resume_input_invalid_format_errors(self, project_dir):
         from typer.testing import CliRunner
         from unittest.mock import patch
@@ -10109,6 +10838,149 @@ steps:
         payload = _json.loads(resumed.stdout)
         assert payload["status"] == "failed"
 
+    _WF_GATE_INVALID_VERDICT = """
+schema_version: "1.0"
+workflow:
+  id: "gate-invalid-verdict"
+  name: "Gate Invalid Verdict"
+  version: "1.0.0"
+inputs:
+  review_verdict:
+    type: string
+    default: ""
+steps:
+  - id: review
+    type: gate
+    message: "Approve the review?"
+    options: [approve, reject]
+    on_reject: abort
+    verdict_input: review_verdict
+"""
+
+    _WF_GATE_INVALID_TYPE = """
+schema_version: "1.0"
+workflow:
+  id: "gate-invalid-type"
+  name: "Gate Invalid Type"
+  version: "1.0.0"
+inputs:
+  review_verdict:
+    type: number
+    default: 1
+steps:
+  - id: review
+    type: gate
+    message: "Approve the review?"
+    options: [approve, reject]
+    on_reject: abort
+    verdict_input: review_verdict
+"""
+
+    _WF_GATE_ABORT = """
+schema_version: "1.0"
+workflow:
+  id: "gate-abort"
+  name: "Gate Abort"
+  version: "1.0.0"
+inputs:
+  review_verdict:
+    type: string
+    default: ""
+steps:
+  - id: review
+    type: gate
+    message: "Approve the review?"
+    options: [approve, reject]
+    on_reject: abort
+    verdict_input: review_verdict
+"""
+
+    def test_run_invalid_verdict_prints_error(self, tmp_path, monkeypatch):
+        """Invalid verdict value prints explanatory error in human output."""
+        import re
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        monkeypatch.chdir(tmp_path)
+        runner = CliRunner()
+        result = runner.invoke(
+            app,
+            [
+                "workflow",
+                "run",
+                str(self._write(tmp_path, self._WF_GATE_INVALID_VERDICT)),
+                "--input",
+                "review_verdict=maybe",
+            ],
+        )
+        assert result.exit_code == 1
+        assert "Status: failed" in result.stdout
+        # Normalize whitespace to handle Rich console line wrapping
+        normalized = re.sub(r"\s+", " ", result.stdout)
+        assert "does not match any configured option" in normalized
+
+    def test_run_invalid_verdict_type_prints_error(self, tmp_path, monkeypatch):
+        """Non-string verdict value prints explanatory error in human output."""
+        import re
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        monkeypatch.chdir(tmp_path)
+        runner = CliRunner()
+        result = runner.invoke(
+            app,
+            ["workflow", "run", str(self._write(tmp_path, self._WF_GATE_INVALID_TYPE))],
+        )
+        assert result.exit_code == 1
+        assert "Status: failed" in result.stdout
+        # Normalize whitespace to handle Rich console line wrapping
+        normalized = re.sub(r"\s+", " ", result.stdout)
+        assert "must be a string" in normalized
+
+    def test_run_gate_abort_prints_status_and_error(self, tmp_path, monkeypatch):
+        """Gate abort prints Status: aborted and the rejection message."""
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        monkeypatch.chdir(tmp_path)
+        runner = CliRunner()
+        result = runner.invoke(
+            app,
+            [
+                "workflow",
+                "run",
+                str(self._write(tmp_path, self._WF_GATE_ABORT)),
+                "--input",
+                "review_verdict=reject",
+            ],
+        )
+        assert result.exit_code == 1
+        assert "Status: aborted" in result.stdout
+        assert "Gate rejected by user" in result.stdout
+
+    def test_run_gate_abort_json_includes_error(self, tmp_path, monkeypatch):
+        """Gate abort --json includes the rejection message in the error field."""
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        monkeypatch.chdir(tmp_path)
+        runner = CliRunner()
+        result = runner.invoke(
+            app,
+            [
+                "workflow",
+                "run",
+                str(self._write(tmp_path, self._WF_GATE_ABORT)),
+                "--input",
+                "review_verdict=reject",
+                "--json",
+            ],
+        )
+        assert result.exit_code == 1
+        payload = json.loads(result.stdout)
+        assert payload["status"] == "aborted"
+        assert "Gate rejected by user" in (payload.get("error") or "")
+
 
 class TestWorkflowRunGateOutcomeJson:
     """CLI-level tests: the --json payload surfaces gate pauses."""