feat: bind gate verdict to workflow input via verdict_input

docs/reference/workflows.md

@@ -528,6 +528,46 @@ 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.
+
+**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

@@ -883,6 +883,31 @@ def _require_specify_project(*args, **kwargs):
     return project_root
 
 
+def _failed_step_error(state: Any) -> str | None:
+    """First persisted step error for a failed/aborted run, if any.
+
+    Prefers the error on ``current_step_id`` (mirrors ``_gate_outcome``'s
+    approach); falls back to scanning for any step with ``status == "failed"``
+    carrying an error, so a stale ``current_step_id`` never hides the message.
+    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
+    results = getattr(state, "step_results", None) or {}
+    cur = results.get(getattr(state, "current_step_id", None))
+    if isinstance(cur, dict) and cur.get("error"):
+        return str(cur["error"])
+    for sd in reversed(results.values()):
+        if (
+            isinstance(sd, dict)
+            and sd.get("status") == "failed"
+            and sd.get("error")
+        ):
+            return str(sd["error"])
+    return None
+
+
 def _workflow_run_payload(state: Any) -> dict[str, Any]:
     """Machine-readable summary of a run/resume outcome."""
     payload = {
@@ -895,6 +920,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
 
 
@@ -1144,6 +1172,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]")
 
@@ -1243,6 +1275,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))
 
 

src/specify_cli/workflows/engine.py

@@ -297,7 +297,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
 
@@ -306,8 +314,15 @@ def _validate_steps(
     steps: list[dict[str, Any]],
     seen_ids: set[str],
     errors: list[str],
+    input_names: set[str] | None = None,
 ) -> 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). Threaded through recursion so
+    cross-reference checks (e.g. gate ``verdict_input``) can verify that
+    referenced inputs exist.
+    """
     from . import STEP_REGISTRY
 
     for step_config in steps:
@@ -400,30 +415,52 @@ def _validate_steps(
                             f"unknown or not-yet-declared step id {wid!r}."
                         )
 
+        # 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 (resume merges then re-resolves through the
+        # same path). So an undeclared ``verdict_input`` can never receive a
+        # value through any channel; the gate would pause forever. Surface
+        # this wiring error at validation time, mirroring the fan-in
+        # ``wait_for`` unknown-id check above. Only check when the value is
+        # a non-empty string — malformed shapes are already reported by
+        # ``GateStep.validate()``; never pile on a confusing duplicate.
+        if step_type == "gate" and input_names is not None:
+            verdict_input = step_config.get("verdict_input")
+            if (
+                isinstance(verdict_input, str)
+                and verdict_input
+                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)
 
         # 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)
 
         # 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)
 
         # 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)
             errors.extend(fan_errors)
 
 
@@ -1054,6 +1091,7 @@ def _execute_steps(
                 or step_config.get("input", {}),
                 "output": result.output,
                 "status": result.status.value,
+                "error": result.error,
             }
             self._record_result(context, state, step_id, step_data)
 

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 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
 
         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,17 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
                 },
             )
 
+        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."
+                ),
+            )
+
         show_file = config.get("show_file")
         if isinstance(show_file, str) and "{{" in show_file:
             show_file = evaluate_expression(show_file, context)
@@ -90,16 +104,48 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
             "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
+
+        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)
+            # 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 +165,8 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
                 )
             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 +282,13 @@ def validate(self, config: dict[str, Any]) -> list[str]:
                 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.