fix(workflows): reject a retry gate whose verdict enum forbids the reset value (#3912)

A gate with `on_reject: retry` consumes a bound reject verdict before
pausing by resetting the named input to `""` (documented behaviour, so a
later resume prompts again). Every `resume()` re-resolves the persisted
inputs through `_coerce_input`.

Those two rules collide when the bound input declares an `enum` that does
not list `""`. The reset writes a value the input's own enum forbids, and
the run wedges:

    inputs:
      spec_verdict:
        type: string
        enum: [approve, reject]
    steps:
      - id: review
        type: gate
        options: [approve, reject]
        on_reject: retry
        verdict_input: spec_verdict

    $ specify workflow run wf --input spec_verdict=reject
    Status: paused
    $ specify workflow resume <run_id> --input note=b
    Error: Input 'spec_verdict' value '' not in allowed values:
           ['approve', 'reject'].

The workflow validates clean and the first run looks fine, so the failure
only appears at the second resume. It is also unrecoverable in practice:
`_resolve_inputs` re-coerces the whole persisted map, so *any* resume that
supplies an input dies on the stored `""`. Only a resume with no inputs at
all still works -- and that is precisely the call that cannot deliver a new
verdict, which is the one thing the retry cycle exists to allow.

Extend the existing `verdict_input` cross-check (which already confirms the
name is declared) to also require that a retry-bound input's `enum` admits
the reset sentinel, and report it with a fix hint. To do that, thread the
input *definitions* through `_validate_steps` instead of just their names.

Rejected the alternative of popping the key instead of writing `""`: that
lets the input's `default` flow back in on the next resume, so a gate the
user just rejected would silently auto-approve.

Docs: note the `enum` requirement next to the reset behaviour it follows
from.

Adds 4 validation tests for the new guard plus a characterization test that
drives the engine directly to pin the wedge it prevents.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Assisted-by: Claude Opus 5 (1M context)

Author: Noor-ul-ain001

docs/reference/workflows.md

@@ -623,6 +623,19 @@ 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.
 
+Because of that reset, a verdict input used with `on_reject: retry` must accept
+`""`. If it declares an `enum`, include the empty string — otherwise the reset
+value violates the input's own `enum` and the run can no longer be resumed with
+any input. `specify workflow add` reports this as a validation error.
+
+```yaml
+inputs:
+  spec_verdict:
+    type: string
+    enum: ["", approve, reject]
+    default: ""
+```
+
 ## FAQ
 
 ### What happens when a workflow hits a gate step?

src/specify_cli/workflows/engine.py

@@ -308,15 +308,16 @@ def validate_workflow(definition: WorkflowDefinition) -> list[str]:
         errors.append("Workflow has no steps defined.")
 
     seen_ids: set[str] = set()
-    # ``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
+    # ``input_defs`` maps declared workflow input names to their definitions —
+    # used by ``_validate_steps`` to cross-reference gate ``verdict_input``
+    # bindings (both that the name exists and that its ``enum`` permits the
+    # reset sentinel). ``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_defs: dict[str, Any] | None = (
+        dict(definition.inputs) if isinstance(definition.inputs, dict) else None
     )
-    _validate_steps(definition.steps, seen_ids, errors, input_names)
+    _validate_steps(definition.steps, seen_ids, errors, input_defs)
 
     return errors
 
@@ -325,15 +326,15 @@ def _validate_steps(
     steps: list[dict[str, Any]],
     seen_ids: set[str],
     errors: list[str],
-    input_names: set[str] | None = None,
+    input_defs: dict[str, Any] | None = None,
     inside_fan_out: bool = False,
 ) -> None:
     """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.
+    ``input_defs`` maps declared workflow input names to their definitions (or
+    is ``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
 
@@ -440,11 +441,39 @@ def _validate_steps(
                         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:
+                elif input_defs is not None and verdict_input not in input_defs:
                     errors.append(
                         f"Gate step {step_id!r}: 'verdict_input' references "
                         f"undeclared input {verdict_input!r}."
                     )
+                elif input_defs is not None:
+                    # ``on_reject: retry`` resets the bound input to "" before
+                    # pausing, and every later resume re-resolves the persisted
+                    # inputs through ``_coerce_input``. If the input declares an
+                    # ``enum`` that omits "", that reset value is instantly
+                    # illegal: the run pauses fine, but the next resume that
+                    # supplies any input raises "value '' not in allowed
+                    # values", and no verdict can be routed through the gate
+                    # again. Require the enum to admit the sentinel so the
+                    # retry cycle the field advertises is actually reachable.
+                    verdict_def = input_defs.get(verdict_input)
+                    enum_values = (
+                        verdict_def.get("enum")
+                        if isinstance(verdict_def, dict)
+                        else None
+                    )
+                    if (
+                        step_config.get("on_reject") == "retry"
+                        and isinstance(enum_values, list)
+                        and "" not in enum_values
+                    ):
+                        errors.append(
+                            f"Gate step {step_id!r}: on_reject='retry' resets "
+                            f"verdict input {verdict_input!r} to '' when the "
+                            f"gate is rejected, but that input's 'enum' does "
+                            f"not allow ''. Add '' to the enum or use "
+                            f"on_reject='abort'/'skip'."
+                        )
 
         # Recursively validate nested steps
         for nested_key in ("then", "else", "steps"):
@@ -454,7 +483,7 @@ def _validate_steps(
                     nested,
                     seen_ids,
                     errors,
-                    input_names,
+                    input_defs,
                     inside_fan_out=inside_fan_out,
                 )
 
@@ -467,7 +496,7 @@ def _validate_steps(
                         case_steps,
                         seen_ids,
                         errors,
-                        input_names,
+                        input_defs,
                         inside_fan_out=inside_fan_out,
                     )
 
@@ -478,7 +507,7 @@ def _validate_steps(
                 default,
                 seen_ids,
                 errors,
-                input_names,
+                input_defs,
                 inside_fan_out=inside_fan_out,
             )
 
@@ -491,7 +520,7 @@ def _validate_steps(
                 [fan_step],
                 set(),
                 fan_errors,
-                input_names,
+                input_defs,
                 inside_fan_out=True,
             )
             errors.extend(fan_errors)

tests/test_workflows.py

@@ -4673,6 +4673,149 @@ def test_malformed_verdict_input_no_duplicate_error(self):
         # 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_retry_verdict_enum_must_allow_reset_sentinel(self):
+        # on_reject: retry resets the bound input to "" before pausing, and
+        # every resume re-resolves persisted inputs through _coerce_input. An
+        # enum that omits "" makes that reset value instantly illegal, so the
+        # next resume supplying any input dies with "value '' not in allowed
+        # values" and no verdict can reach the gate again.
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+inputs:
+  spec_verdict:
+    type: string
+    enum: [approve, reject]
+steps:
+  - id: review
+    type: gate
+    message: "Review?"
+    options: [approve, reject]
+    on_reject: retry
+    verdict_input: spec_verdict
+""")
+        assert any(
+            "on_reject='retry' resets verdict input 'spec_verdict'" in e
+            for e in errors
+        ), errors
+
+    def test_retry_verdict_enum_including_sentinel_passes(self):
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+inputs:
+  spec_verdict:
+    type: string
+    enum: ["", approve, reject]
+    default: ""
+steps:
+  - id: review
+    type: gate
+    message: "Review?"
+    options: [approve, reject]
+    on_reject: retry
+    verdict_input: spec_verdict
+""")
+        assert not any("on_reject='retry'" in e for e in errors), errors
+
+    def test_verdict_enum_without_sentinel_passes_when_not_retry(self):
+        # abort/skip never reset the input, so the enum need not admit "".
+        for on_reject in ("abort", "skip"):
+            errors = self._errors(f"""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+inputs:
+  spec_verdict:
+    type: string
+    enum: [approve, reject]
+steps:
+  - id: review
+    type: gate
+    message: "Review?"
+    options: [approve, reject]
+    on_reject: {on_reject}
+    verdict_input: spec_verdict
+""")
+            assert not any("on_reject='retry'" in e for e in errors), (
+                on_reject,
+                errors,
+            )
+
+    def test_retry_verdict_without_enum_passes(self):
+        # No enum means _coerce_input accepts "" — the documented shape.
+        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]
+    on_reject: retry
+    verdict_input: spec_verdict
+""")
+        assert not any("on_reject='retry'" in e for e in errors), errors
+
+    def test_retry_verdict_enum_wedge_is_reachable_end_to_end(self, tmp_path):
+        """The validation error above guards a real, unrecoverable run state.
+
+        Without the guard this workflow installs and runs fine, then wedges:
+        the retry reset writes "" into the persisted inputs, and the next
+        resume that supplies *any* input re-resolves them and dies on the
+        enum. Only a resume with no inputs at all still works, so the bound
+        verdict can never be delivered.
+        """
+        import pytest
+        import yaml as _yaml
+
+        from specify_cli.workflows.engine import WorkflowEngine
+
+        definition_data = {
+            "schema_version": "1.0",
+            "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"},
+            "inputs": {
+                "spec_verdict": {"type": "string", "enum": ["approve", "reject"]},
+                "note": {"type": "string", "default": "a"},
+            },
+            "steps": [
+                {
+                    "id": "review",
+                    "type": "gate",
+                    "message": "Review?",
+                    "options": ["approve", "reject"],
+                    "on_reject": "retry",
+                    "verdict_input": "spec_verdict",
+                }
+            ],
+        }
+        wf_dir = tmp_path / ".specify" / "workflows" / "wf"
+        wf_dir.mkdir(parents=True)
+        (wf_dir / "workflow.yml").write_text(
+            _yaml.safe_dump(definition_data), encoding="utf-8"
+        )
+
+        engine = WorkflowEngine(tmp_path)
+        definition = engine.load_workflow("wf")
+        state = engine.execute(definition, inputs={"spec_verdict": "reject"})
+        assert state.status.value == "paused"
+        # The retry reset persisted a value the input's own enum forbids.
+        assert state.inputs["spec_verdict"] == ""
+
+        with pytest.raises(ValueError, match="not in allowed values"):
+            engine.resume(state.run_id, inputs={"note": "b"})
+
     def test_verdict_input_in_switch_case(self):
         # Recursion coverage: bad reference inside a switch case must surface.
         errors = self._errors("""