diff --git a/src/specify_cli/__init__.py b/src/specify_cli/__init__.py index c0bdbaabe3..89790e4183 100644 --- a/src/specify_cli/__init__.py +++ b/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) diff --git a/src/specify_cli/workflows/base.py b/src/specify_cli/workflows/base.py index b144ca903d..2625e25953 100644 --- a/src/specify_cli/workflows/base.py +++ b/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: diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py index 934cfbe5ee..b8c0b22ce5 100644 --- a/src/specify_cli/workflows/engine.py +++ b/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 `` 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 diff --git a/src/specify_cli/workflows/gate_script.py b/src/specify_cli/workflows/gate_script.py new file mode 100644 index 0000000000..4e16370b5a --- /dev/null +++ b/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 ` +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 diff --git a/src/specify_cli/workflows/steps/gate/__init__.py b/src/specify_cli/workflows/steps/gate/__init__.py index d4d32d763c..5e3ad816ae 100644 --- a/src/specify_cli/workflows/steps/gate/__init__.py +++ b/src/specify_cli/workflows/steps/gate/__init__.py @@ -7,6 +7,10 @@ from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus from specify_cli.workflows.expressions import evaluate_expression +from specify_cli.workflows.gate_script import ( + extract_base_gate_id, + lookup_scripted_verdict, +) class GateStep(StepBase): @@ -19,6 +23,14 @@ class GateStep(StepBase): The user's choice is stored in ``output.choice``. ``on_reject`` controls abort / skip behaviour. + + When the engine receives a ``--gate-script`` (or a caller passes + one through ``WorkflowEngine.execute(gate_script=...)``), the + gate consults the script before doing anything else: a matching + ``(gate_id, iteration)`` entry's verdict is used directly and + ``output.scripted`` is set to ``True``. This is the + non-interactive harness used by CI tests. When no script entry + matches, the gate falls back to its normal behaviour. """ type_key = "gate" @@ -35,14 +47,46 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: if show_file and isinstance(show_file, str) and "{{" in show_file: show_file = evaluate_expression(show_file, context) - output = { + output: dict[str, Any] = { "message": message, "options": options, "on_reject": on_reject, "show_file": show_file, "choice": None, + "scripted": False, } + # Scripted verdict path: consult the gate-script first. + # The firing counter increments for every gate execution + # regardless of whether a script entry matches, so reordering + # the YAML doesn't shift scripted verdicts onto unrelated + # firings. + base_gate_id = extract_base_gate_id(str(config.get("id", ""))) + iteration = context.gate_firing_counts.get(base_gate_id, 0) + context.gate_firing_counts[base_gate_id] = iteration + 1 + + scripted_verdict = lookup_scripted_verdict( + context.gate_script, base_gate_id, iteration + ) + if scripted_verdict is not None: + output["choice"] = scripted_verdict + output["scripted"] = True + if scripted_verdict in ("reject", "abort"): + if on_reject == "abort": + output["aborted"] = True + return StepResult( + status=StepStatus.FAILED, + output=output, + error=( + f"Gate scripted-rejected at step " + f"{config.get('id', '?')!r} (gate-script verdict)" + ), + ) + if on_reject == "retry": + return StepResult(status=StepStatus.PAUSED, output=output) + # on_reject == "skip" → completed, downstream decides + return StepResult(status=StepStatus.COMPLETED, output=output) + # Non-interactive: pause for later resume if not sys.stdin.isatty(): return StepResult(status=StepStatus.PAUSED, output=output) diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 3fa71f3404..9213dc385f 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -779,6 +779,485 @@ def test_validate_invalid_on_reject(self): assert any("on_reject" in e for e in errors) +class TestGateScriptLoader: + """Test the gate-script loader/validator helpers.""" + + def test_parse_valid_script_returns_verdicts(self): + from specify_cli.workflows.gate_script import parse_gate_script + + data = { + "schema": "speckit.gate-script/v1", + "verdicts": [ + {"gate_id": "g1", "iteration": 0, "verdict": "approve"}, + {"gate_id": "g1", "iteration": 1, "verdict": "reject"}, + ], + } + verdicts = parse_gate_script(data) + assert len(verdicts) == 2 + assert verdicts[0]["verdict"] == "approve" + assert verdicts[1]["verdict"] == "reject" + + def test_parse_rejects_wrong_schema(self): + from specify_cli.workflows.gate_script import parse_gate_script + + with pytest.raises(ValueError, match="unsupported schema"): + parse_gate_script( + { + "schema": "speckit.gate-script/v999", + "verdicts": [], + } + ) + + def test_parse_rejects_missing_verdicts(self): + from specify_cli.workflows.gate_script import parse_gate_script + + with pytest.raises(ValueError, match="'verdicts'"): + parse_gate_script({"schema": "speckit.gate-script/v1"}) + + def test_parse_rejects_non_mapping_verdict(self): + from specify_cli.workflows.gate_script import parse_gate_script + + with pytest.raises(ValueError, match="must be a mapping"): + parse_gate_script( + { + "schema": "speckit.gate-script/v1", + "verdicts": ["not-a-mapping"], + } + ) + + def test_parse_rejects_missing_required_fields(self): + from specify_cli.workflows.gate_script import parse_gate_script + + with pytest.raises(ValueError, match="'iteration'"): + parse_gate_script( + { + "schema": "speckit.gate-script/v1", + "verdicts": [{"gate_id": "g", "verdict": "approve"}], + } + ) + + def test_parse_rejects_non_int_iteration(self): + from specify_cli.workflows.gate_script import parse_gate_script + + with pytest.raises(ValueError, match="'iteration' must be an integer"): + parse_gate_script( + { + "schema": "speckit.gate-script/v1", + "verdicts": [ + {"gate_id": "g", "iteration": "0", "verdict": "x"} + ], + } + ) + + def test_parse_rejects_bool_iteration(self): + """``bool`` is a subclass of ``int`` so a naive ``isinstance`` + check accepts ``True``. Validator rejects it explicitly so a + YAML authoring mistake (``iteration: true``) doesn't silently + coerce to 1. + """ + from specify_cli.workflows.gate_script import parse_gate_script + + with pytest.raises(ValueError, match="'iteration' must be an integer"): + parse_gate_script( + { + "schema": "speckit.gate-script/v1", + "verdicts": [ + {"gate_id": "g", "iteration": True, "verdict": "x"} + ], + } + ) + + def test_lookup_finds_matching_entry(self): + from specify_cli.workflows.gate_script import lookup_scripted_verdict + + script = [ + {"gate_id": "g1", "iteration": 0, "verdict": "improve"}, + {"gate_id": "g1", "iteration": 1, "verdict": "approve"}, + {"gate_id": "g2", "iteration": 0, "verdict": "reject"}, + ] + assert lookup_scripted_verdict(script, "g1", 0) == "improve" + assert lookup_scripted_verdict(script, "g1", 1) == "approve" + assert lookup_scripted_verdict(script, "g2", 0) == "reject" + + def test_lookup_returns_none_for_no_match(self): + from specify_cli.workflows.gate_script import lookup_scripted_verdict + + script = [{"gate_id": "g1", "iteration": 0, "verdict": "approve"}] + assert lookup_scripted_verdict(script, "g1", 1) is None + assert lookup_scripted_verdict(script, "g2", 0) is None + + def test_lookup_empty_script_returns_none(self): + from specify_cli.workflows.gate_script import lookup_scripted_verdict + + assert lookup_scripted_verdict([], "g1", 0) is None + + def test_extract_base_gate_id_unwrapped(self): + from specify_cli.workflows.gate_script import extract_base_gate_id + + assert extract_base_gate_id("my-gate") == "my-gate" + + def test_extract_base_gate_id_unwraps_loop_namespacing(self): + """The engine namespaces nested loop steps as + ``parent:child:iter``. Gate scripts match on the + author-visible base id, so this helper inverts that. + """ + from specify_cli.workflows.gate_script import extract_base_gate_id + + assert extract_base_gate_id("my-loop:my-gate:1") == "my-gate" + assert extract_base_gate_id("outer:inner:my-gate:42") == "my-gate" + + def test_load_from_file(self, tmp_path): + from specify_cli.workflows.gate_script import load_gate_script + + script_path = tmp_path / "script.yaml" + script_path.write_text( + "schema: speckit.gate-script/v1\n" + "verdicts:\n" + " - {gate_id: review, iteration: 0, verdict: improve}\n", + encoding="utf-8", + ) + verdicts = load_gate_script(script_path) + assert verdicts == [ + {"gate_id": "review", "iteration": 0, "verdict": "improve"} + ] + + def test_load_missing_file_raises(self, tmp_path): + from specify_cli.workflows.gate_script import load_gate_script + + with pytest.raises(FileNotFoundError): + load_gate_script(tmp_path / "does-not-exist.yaml") + + def test_parse_rejects_duplicate_gate_id_iteration(self): + """Two entries sharing the same ``(gate_id, iteration)`` would + silently shadow each other (``lookup_scripted_verdict`` + returns the first match). The validator rejects them at parse + time so copy-paste authoring mistakes surface immediately. + """ + from specify_cli.workflows.gate_script import parse_gate_script + + with pytest.raises(ValueError, match="duplicate"): + parse_gate_script( + { + "schema": "speckit.gate-script/v1", + "verdicts": [ + {"gate_id": "g", "iteration": 0, "verdict": "approve"}, + {"gate_id": "g", "iteration": 0, "verdict": "reject"}, + ], + } + ) + + def test_parse_allows_same_gate_id_different_iterations(self): + """Distinct iterations under the same gate_id are valid — that's + how the script drives multi-firing gates (e.g. improve → approve). + """ + from specify_cli.workflows.gate_script import parse_gate_script + + verdicts = parse_gate_script( + { + "schema": "speckit.gate-script/v1", + "verdicts": [ + {"gate_id": "g", "iteration": 0, "verdict": "improve"}, + {"gate_id": "g", "iteration": 1, "verdict": "approve"}, + ], + } + ) + assert len(verdicts) == 2 + + +class TestGateScriptIntegration: + """End-to-end tests for gate-script with GateStep + engine.""" + + def test_gate_uses_scripted_verdict_instead_of_prompting( + self, project_dir, monkeypatch + ): + """A gate with a matching script entry returns the scripted + verdict directly. ``output.scripted == True`` and the gate + never prompts. + + Locks the core contract from issue #2594: workflows can be + driven non-interactively in CI by supplying a verdict script. + """ + from specify_cli.workflows.engine import ( + WorkflowDefinition, + WorkflowEngine, + ) + from specify_cli.workflows.base import RunStatus + from specify_cli.workflows.steps.gate import GateStep + + # Sentinel: if the gate ever prompts, fail the test loudly. + def _should_not_prompt(*_args, **_kwargs): + raise AssertionError("Gate prompted despite scripted verdict.") + + monkeypatch.setattr( + GateStep, "_prompt", staticmethod(_should_not_prompt) + ) + + definition = WorkflowDefinition.from_string(""" +schema_version: "1.0" +workflow: + id: "scripted-gate" + name: "Scripted Gate" + version: "1.0.0" +steps: + - id: review + type: gate + message: "Approve?" + options: [approve, reject] + on_reject: abort +""") + engine = WorkflowEngine(project_dir) + state = engine.execute( + definition, + gate_script=[ + {"gate_id": "review", "iteration": 0, "verdict": "approve"}, + ], + ) + + assert state.status == RunStatus.COMPLETED + review = state.step_results["review"] + assert review["output"]["choice"] == "approve" + assert review["output"]["scripted"] is True + + def test_improve_then_approve_cycle_via_switch_unroll( + self, project_dir, monkeypatch + ): + """End-to-end: the canonical CI-test scenario from issue + #2594 — improve → re-review → approve, with zero operator + input. Uses the manually-unrolled gate+switch pattern + (rather than a while/do-while loop) so this test stays + decoupled from the separate loop-iteration namespacing + change tracked in issue #2592. + """ + from specify_cli.workflows.engine import ( + WorkflowDefinition, + WorkflowEngine, + ) + from specify_cli.workflows.base import RunStatus + from specify_cli.workflows.steps.gate import GateStep + + def _should_not_prompt(*_args, **_kwargs): + raise AssertionError("Gate prompted despite scripted verdict.") + + monkeypatch.setattr( + GateStep, "_prompt", staticmethod(_should_not_prompt) + ) + + definition = WorkflowDefinition.from_string(""" +schema_version: "1.0" +workflow: + id: "improve-approve-cycle" + name: "Improve Approve Cycle" + version: "1.0.0" +steps: + - id: review-first + type: gate + message: "Approve?" + options: [approve, improve, reject] + on_reject: abort + - id: maybe-improve + type: switch + expression: "{{ steps.review-first.output.choice }}" + cases: + improve: + - id: review-second + type: gate + message: "Approve after improve?" + options: [approve, reject] + on_reject: abort +""") + engine = WorkflowEngine(project_dir) + state = engine.execute( + definition, + gate_script=[ + {"gate_id": "review-first", "iteration": 0, "verdict": "improve"}, + {"gate_id": "review-second", "iteration": 0, "verdict": "approve"}, + ], + ) + + assert state.status == RunStatus.COMPLETED + assert state.step_results["review-first"]["output"]["choice"] == "improve" + assert state.step_results["review-first"]["output"]["scripted"] is True + assert state.step_results["review-second"]["output"]["choice"] == "approve" + assert state.step_results["review-second"]["output"]["scripted"] is True + + def test_iteration_counter_increments_on_repeated_gate_id( + self, project_dir, monkeypatch + ): + """A workflow with two separate ``gate`` steps that share + the same base ``id`` (allowed by the engine's step-id + uniqueness rules only across non-sibling positions) would + normally conflict — but the more practical scenario is two + DIFFERENT gates, each firing once at iteration 0. + + This test locks the per-gate iteration counter: each gate's + counter is independent, so two consecutive single-firing + gates both look up at iteration 0 against their own + ``gate_id``. + """ + from specify_cli.workflows.engine import ( + WorkflowDefinition, + WorkflowEngine, + ) + from specify_cli.workflows.base import RunStatus + from specify_cli.workflows.steps.gate import GateStep + + monkeypatch.setattr( + GateStep, + "_prompt", + staticmethod( + lambda *_a, **_kw: (_ for _ in ()).throw( + AssertionError("Gate prompted despite scripted verdict.") + ) + ), + ) + + definition = WorkflowDefinition.from_string(""" +schema_version: "1.0" +workflow: + id: "two-gates" + name: "Two Gates" + version: "1.0.0" +steps: + - id: gate-a + type: gate + message: "A?" + options: [approve, reject] + - id: gate-b + type: gate + message: "B?" + options: [approve, reject] +""") + engine = WorkflowEngine(project_dir) + state = engine.execute( + definition, + gate_script=[ + {"gate_id": "gate-a", "iteration": 0, "verdict": "approve"}, + {"gate_id": "gate-b", "iteration": 0, "verdict": "approve"}, + ], + ) + + assert state.status == RunStatus.COMPLETED + assert state.step_results["gate-a"]["output"]["choice"] == "approve" + assert state.step_results["gate-b"]["output"]["choice"] == "approve" + + def test_default_behaviour_preserved_without_script( + self, project_dir, monkeypatch + ): + """When no script is provided, the gate falls back to its + normal behaviour: ``PAUSED`` in non-TTY environments. + + Locks the byte-equivalent default required by the issue's + acceptance criteria. + """ + from specify_cli.workflows.engine import ( + WorkflowDefinition, + WorkflowEngine, + ) + from specify_cli.workflows.base import RunStatus + from specify_cli.workflows.steps import gate as gate_module + + monkeypatch.setattr(gate_module.sys.stdin, "isatty", lambda: False) + + definition = WorkflowDefinition.from_string(""" +schema_version: "1.0" +workflow: + id: "unscripted" + name: "Unscripted" + version: "1.0.0" +steps: + - id: review + type: gate + message: "Approve?" + options: [approve, reject] +""") + engine = WorkflowEngine(project_dir) + state = engine.execute(definition) + + assert state.status == RunStatus.PAUSED + review = state.step_results["review"] + assert review["output"]["choice"] is None + assert review["output"]["scripted"] is False + + def test_non_matching_script_entry_falls_back_to_prompt( + self, project_dir, monkeypatch + ): + """When the script has entries but none matches the current + gate firing, the gate falls back to its normal behaviour. + Locks the partial-script contract: workflow authors can + script only the gates they care about. + """ + from specify_cli.workflows.engine import ( + WorkflowDefinition, + WorkflowEngine, + ) + from specify_cli.workflows.base import RunStatus + from specify_cli.workflows.steps import gate as gate_module + + monkeypatch.setattr(gate_module.sys.stdin, "isatty", lambda: False) + + definition = WorkflowDefinition.from_string(""" +schema_version: "1.0" +workflow: + id: "partial-script" + name: "Partial Script" + version: "1.0.0" +steps: + - id: unrelated-gate + type: gate + message: "Approve?" + options: [approve, reject] +""") + engine = WorkflowEngine(project_dir) + state = engine.execute( + definition, + gate_script=[ + {"gate_id": "other-gate", "iteration": 0, "verdict": "approve"}, + ], + ) + + assert state.status == RunStatus.PAUSED + assert state.step_results["unrelated-gate"]["output"]["scripted"] is False + + def test_scripted_reject_with_abort_halts_run(self, project_dir): + """A scripted ``reject`` verdict on a gate with + ``on_reject: abort`` halts the run with ``ABORTED`` status, + same as a human-driven reject would. + """ + 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: "scripted-reject" + name: "Scripted Reject" + version: "1.0.0" +steps: + - id: review + type: gate + message: "Approve?" + options: [approve, reject] + on_reject: abort + - id: after + type: shell + run: "echo should-not-run" +""") + engine = WorkflowEngine(project_dir) + state = engine.execute( + definition, + gate_script=[ + {"gate_id": "review", "iteration": 0, "verdict": "reject"}, + ], + ) + + assert state.status == RunStatus.ABORTED + assert state.step_results["review"]["output"]["aborted"] is True + assert "after" not in state.step_results + + class TestIfThenStep: """Test the if/then/else step type.""" diff --git a/workflows/README.md b/workflows/README.md index 31f736ff76..b63360f6ae 100644 --- a/workflows/README.md +++ b/workflows/README.md @@ -287,6 +287,51 @@ specify workflow resume Run states: `created` → `running` → `completed` | `paused` | `failed` | `aborted` +## Non-Interactive Gate Testing + +`gate` steps prompt interactively by default. For CI / non-interactive +test runs, supply a `--gate-script` YAML that pre-records verdicts. +Both `workflow run` and `workflow resume` accept the flag, so a +scripted CI run can also drive gates that fire only after a prior +pause: + +```bash +specify workflow run my-pipeline --gate-script verdicts.yaml +specify workflow resume --gate-script verdicts.yaml +``` + +```yaml +# verdicts.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: review-final + iteration: 0 + verdict: approve +``` + +Behaviour: + +- `gate_id` matches the workflow YAML step `id` (the author-visible + base id — engine-internal loop namespacing like `parent:child:N` is + unwrapped automatically). +- `iteration` selects which firing the verdict applies to (0-indexed, + counting from the first time the gate runs within the run). +- `verdict` is the option string the gate would otherwise produce — + typically `approve` / `reject` / `edit` / a custom route name. +- When the engine finds no matching entry, the gate falls back to its + normal behaviour (interactive prompt on TTY, `PAUSED` otherwise). + This lets workflows partially-script only the gates they care about. +- `output.scripted` records `True` for scripted verdicts and `False` + for interactive ones, so workflows can distinguish them downstream. +- Scripted `reject` verdicts honour the gate's `on_reject` setting + identically to operator-driven rejects. + ## Catalog Management Workflows are discovered through catalogs. By default, Spec Kit uses the official and community catalogs: