feat(workflows): add --gate-script for non-interactive gate testing

src/specify_cli/__init__.py

@@ -4736,9 +4736,22 @@ def workflow_run(
     input_values: list[str] | None = typer.Option(
         None, "--input", "-i", help="Input values as key=value pairs"
     ),
+    gate_script: Path | None = typer.Option(
+        None,
+        "--gate-script",
+        help=(
+            "Path to a gate-script YAML (schema speckit.gate-script/v1) "
+            "that supplies verdicts for gate steps. Gates with a "
+            "matching (gate_id, iteration) entry use the scripted "
+            "verdict instead of prompting; non-matching gates fall "
+            "back to their normal behaviour. Intended for CI / "
+            "non-interactive testing of gated workflows."
+        ),
+    ),
 ):
     """Run a workflow from an installed ID or local YAML path."""
     from .workflows.engine import WorkflowEngine
+    from .workflows.gate_script import load_gate_script
 
     project_root = _require_specify_project()
     engine = WorkflowEngine(project_root)
@@ -4771,11 +4784,22 @@ def workflow_run(
             key, _, value = kv.partition("=")
             inputs[key.strip()] = value.strip()
 
+    # Load gate script if supplied. Failures are fatal — the operator
+    # asked for non-interactive run, so silently falling back to
+    # prompts on a broken script would be worse than failing fast.
+    parsed_script: list[dict[str, Any]] | None = None
+    if gate_script is not None:
+        try:
+            parsed_script = load_gate_script(gate_script)
+        except (FileNotFoundError, ValueError) as exc:
+            console.print(f"[red]Error:[/red] {exc}")
+            raise typer.Exit(1)
+
     console.print(f"\n[bold cyan]Running workflow:[/bold cyan] {definition.name} ({definition.id})")
     console.print(f"[dim]Version: {definition.version}[/dim]\n")
 
     try:
-        state = engine.execute(definition, inputs)
+        state = engine.execute(definition, inputs, gate_script=parsed_script)
     except ValueError as exc:
         console.print(f"[red]Error:[/red] {exc}")
         raise typer.Exit(1)
@@ -4800,16 +4824,36 @@ def workflow_run(
 @workflow_app.command("resume")
 def workflow_resume(
     run_id: str = typer.Argument(..., help="Run ID to resume"),
+    gate_script: Path | None = typer.Option(
+        None,
+        "--gate-script",
+        help=(
+            "Path to a gate-script YAML (schema speckit.gate-script/v1) "
+            "consulted by gate steps that fire after resume. Same "
+            "format as `specify workflow run --gate-script`."
+        ),
+    ),
 ):
     """Resume a paused or failed workflow run."""
     from .workflows.engine import WorkflowEngine
+    from .workflows.gate_script import load_gate_script
 
     project_root = _require_specify_project()
     engine = WorkflowEngine(project_root)
     engine.on_step_start = lambda sid, label: console.print(f"  \u25b8 [{sid}] {label} \u2026")
 
+    # Load gate script if supplied \u2014 same fail-fast semantics as
+    # `workflow run --gate-script`.
+    parsed_script: list[dict[str, Any]] | None = None
+    if gate_script is not None:
+        try:
+            parsed_script = load_gate_script(gate_script)
+        except (FileNotFoundError, ValueError) as exc:
+            console.print(f"[red]Error:[/red] {exc}")
+            raise typer.Exit(1)
+
     try:
-        state = engine.resume(run_id)
+        state = engine.resume(run_id, gate_script=parsed_script)
     except FileNotFoundError:
         console.print(f"[red]Error:[/red] Run not found: {run_id}")
         raise typer.Exit(1)

src/specify_cli/workflows/base.py

@@ -73,6 +73,18 @@ class StepContext:
     #: Current run ID.
     run_id: str | None = None
 
+    #: Optional scripted gate verdicts for CI / non-interactive testing.
+    #: When set, ``GateStep`` consults this list before prompting the
+    #: operator. Loaded from ``--gate-script`` on ``specify workflow
+    #: run``. Each entry is a mapping with ``gate_id``, ``iteration``,
+    #: and ``verdict`` keys (see ``speckit.gate-script/v1`` schema).
+    gate_script: list[dict[str, Any]] = field(default_factory=list)
+
+    #: Per-gate firing counter. Increments each time a gate step with
+    #: the same base ID fires within the current run. Used to match
+    #: ``gate_script`` entries by ``(gate_id, iteration)``.
+    gate_firing_counts: dict[str, int] = field(default_factory=dict)
+
 
 @dataclass
 class StepResult:

src/specify_cli/workflows/engine.py

@@ -415,6 +415,7 @@ def execute(
         definition: WorkflowDefinition,
         inputs: dict[str, Any] | None = None,
         run_id: str | None = None,
+        gate_script: list[dict[str, Any]] | None = None,
     ) -> RunState:
         """Execute a workflow definition.
 
@@ -426,6 +427,12 @@ def execute(
             User-provided input values.
         run_id:
             Optional run ID (auto-generated if not provided).
+        gate_script:
+            Optional list of pre-parsed gate-script verdicts (see
+            ``workflows.gate_script``). When provided, ``GateStep``
+            consults this list before prompting the operator, so the
+            workflow runs non-interactively. Used by CI tests and the
+            ``specify workflow run --gate-script <path>`` flag.
 
         Returns
         -------
@@ -462,6 +469,7 @@ def execute(
             default_options=definition.default_options,
             project_root=str(self.project_root),
             run_id=state.run_id,
+            gate_script=gate_script or [],
         )
 
         # Execute steps
@@ -484,8 +492,18 @@ def execute(
         state.save()
         return state
 
-    def resume(self, run_id: str) -> RunState:
-        """Resume a paused or failed workflow run."""
+    def resume(
+        self,
+        run_id: str,
+        gate_script: list[dict[str, Any]] | None = None,
+    ) -> RunState:
+        """Resume a paused or failed workflow run.
+
+        ``gate_script`` is consulted by ``GateStep`` for any gate that
+        fires after resume — useful for scripted CI tests that pause
+        on a gate, capture intermediate state, and then resume with a
+        scripted verdict.
+        """
         state = RunState.load(run_id, self.project_root)
         if state.status not in (RunStatus.PAUSED, RunStatus.FAILED):
             msg = f"Cannot resume run {run_id!r} with status {state.status.value!r}."
@@ -510,6 +528,7 @@ def resume(self, run_id: str) -> RunState:
             default_options=definition.default_options,
             project_root=str(self.project_root),
             run_id=state.run_id,
+            gate_script=gate_script or [],
         )
 
         from . import STEP_REGISTRY

src/specify_cli/workflows/gate_script.py

@@ -0,0 +1,192 @@
+"""Gate-script loader and validator for non-interactive gate testing.
+
+A gate script lets CI or test runners drive gate verdicts without
+operator interaction. Used by `specify workflow run --gate-script <path>`
+and by the workflow engine to consult scripted verdicts before
+prompting.
+
+The schema (``speckit.gate-script/v1``) is intentionally minimal:
+
+.. code-block:: yaml
+
+    schema: speckit.gate-script/v1
+    verdicts:
+      - gate_id: review-overview
+        iteration: 0
+        verdict: improve
+      - gate_id: review-overview
+        iteration: 1
+        verdict: approve
+
+``gate_id`` matches the YAML step ``id``. For gates that fire multiple
+times (e.g. inside a ``while`` loop), ``iteration`` selects which
+firing the verdict applies to (0-indexed, counting from the first
+time the gate runs within the current workflow run). ``verdict`` is
+the option string the gate would otherwise produce — typically one
+of ``approve`` / ``reject`` / ``edit`` / a custom route name.
+
+When the engine consults the script and finds no matching entry for a
+gate firing, the gate falls back to its normal behaviour (interactive
+prompt on TTY, ``PAUSED`` otherwise).
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+
+GATE_SCRIPT_SCHEMA = "speckit.gate-script/v1"
+
+
+def load_gate_script(path: Path) -> list[dict[str, Any]]:
+    """Load and validate a gate-script YAML file.
+
+    Returns the parsed ``verdicts`` list. Raises ``ValueError`` with a
+    clear message on any structural problem so CLI callers can surface
+    the error directly to the operator.
+    """
+    if not path.exists():
+        msg = f"Gate script not found: {path}"
+        raise FileNotFoundError(msg)
+
+    with open(path, encoding="utf-8") as f:
+        data = yaml.safe_load(f)
+
+    return parse_gate_script(data, source=str(path))
+
+
+def parse_gate_script(
+    data: Any, *, source: str | None = None
+) -> list[dict[str, Any]]:
+    """Validate a parsed gate-script mapping and return its verdicts.
+
+    Separated from ``load_gate_script`` so tests and callers with
+    pre-parsed YAML (e.g. from a string) can reuse the validator
+    without writing to disk.
+    """
+    where = f" in {source}" if source else ""
+    if not isinstance(data, dict):
+        msg = f"Gate script{where} must be a mapping at the top level."
+        raise ValueError(msg)
+
+    schema = data.get("schema")
+    if schema != GATE_SCRIPT_SCHEMA:
+        msg = (
+            f"Gate script{where} has unsupported schema {schema!r}. "
+            f"Expected {GATE_SCRIPT_SCHEMA!r}."
+        )
+        raise ValueError(msg)
+
+    verdicts = data.get("verdicts")
+    if not isinstance(verdicts, list):
+        msg = (
+            f"Gate script{where}: 'verdicts' must be a list, "
+            f"got {type(verdicts).__name__}."
+        )
+        raise ValueError(msg)
+
+    seen_keys: set[tuple[str, int]] = set()
+    for index, entry in enumerate(verdicts):
+        if not isinstance(entry, dict):
+            msg = (
+                f"Gate script{where}: verdict at index {index} must be a "
+                f"mapping with 'gate_id', 'iteration', and 'verdict'."
+            )
+            raise ValueError(msg)
+        for required in ("gate_id", "iteration", "verdict"):
+            if required not in entry:
+                msg = (
+                    f"Gate script{where}: verdict at index {index} is "
+                    f"missing required field {required!r}."
+                )
+                raise ValueError(msg)
+        if not isinstance(entry["gate_id"], str):
+            msg = (
+                f"Gate script{where}: verdict at index {index}: "
+                f"'gate_id' must be a string."
+            )
+            raise ValueError(msg)
+        if not isinstance(entry["iteration"], int) or isinstance(
+            entry["iteration"], bool
+        ):
+            msg = (
+                f"Gate script{where}: verdict at index {index}: "
+                f"'iteration' must be an integer (got "
+                f"{type(entry['iteration']).__name__})."
+            )
+            raise ValueError(msg)
+        if entry["iteration"] < 0:
+            msg = (
+                f"Gate script{where}: verdict at index {index}: "
+                f"'iteration' must be >= 0."
+            )
+            raise ValueError(msg)
+        if not isinstance(entry["verdict"], str):
+            msg = (
+                f"Gate script{where}: verdict at index {index}: "
+                f"'verdict' must be a string."
+            )
+            raise ValueError(msg)
+
+        # Reject duplicate (gate_id, iteration) pairs at parse time.
+        # `lookup_scripted_verdict` returns the first match, so a
+        # duplicate would silently shadow the later entry — almost
+        # always a copy-paste authoring mistake. Failing fast surfaces
+        # the conflict immediately instead of leaving it for runtime
+        # detective work.
+        key = (entry["gate_id"], entry["iteration"])
+        if key in seen_keys:
+            msg = (
+                f"Gate script{where}: verdict at index {index}: "
+                f"duplicate (gate_id={entry['gate_id']!r}, "
+                f"iteration={entry['iteration']}) — each pair must "
+                f"appear at most once."
+            )
+            raise ValueError(msg)
+        seen_keys.add(key)
+
+    return verdicts
+
+
+def lookup_scripted_verdict(
+    script: list[dict[str, Any]],
+    base_gate_id: str,
+    iteration: int,
+) -> str | None:
+    """Return the scripted verdict for ``(base_gate_id, iteration)``,
+    or ``None`` if no entry matches.
+
+    The engine maintains per-gate firing counters and consults this
+    helper before prompting interactively.
+    """
+    if not script:
+        return None
+    for entry in script:
+        if entry.get("gate_id") == base_gate_id and entry.get("iteration") == iteration:
+            verdict = entry.get("verdict")
+            if isinstance(verdict, str):
+                return verdict
+    return None
+
+
+def extract_base_gate_id(step_id: str) -> str:
+    """Strip loop-iteration suffix from a step id to recover its base.
+
+    The engine namespaces nested loop steps as
+    ``parent_id:child_id:iter_num``. Gate scripts match on the
+    workflow-author-visible ``child_id`` — not the engine-internal
+    namespaced form — so this helper inverts that namespacing.
+
+    Examples::
+
+        "my-gate"                  → "my-gate"
+        "my-loop:my-gate:1"        → "my-gate"
+        "outer:inner:my-gate:2"    → "my-gate"
+    """
+    parts = step_id.split(":")
+    if len(parts) >= 3 and parts[-1].isdigit():
+        return parts[-2]
+    return step_id