Add workflow engine with catalog system (#2158)

* Initial plan

* Add workflow engine with step registry, expression engine, catalog system, and CLI commands

Agent-Logs-Url: https://github.com/github/spec-kit/sessions/72a7bb5d-071f-4d67-a507-7e1abae2384d

Co-authored-by: mnriem <15701806+mnriem@users.noreply.github.com>

* Add comprehensive tests for workflow engine (94 tests)

Agent-Logs-Url: https://github.com/github/spec-kit/sessions/72a7bb5d-071f-4d67-a507-7e1abae2384d

Co-authored-by: mnriem <15701806+mnriem@users.noreply.github.com>

* Address review feedback: do-while condition preservation and URL scheme validation

Agent-Logs-Url: https://github.com/github/spec-kit/sessions/72a7bb5d-071f-4d67-a507-7e1abae2384d

Co-authored-by: mnriem <15701806+mnriem@users.noreply.github.com>

* Address review feedback, add CLI dispatch, interactive gates, and docs

Review comments (7/7):
- Add explanatory comment to empty except block
- Implement workflow catalog download with cleanup on failure
- Add input type coercion for number/boolean/enum
- Fix example workflow to remove non-existent output references
- Fix while_loop and if_then condition defaults (string 'false' → bool False)
- Fix resume step index tracking with step_offset parameter

CLI dispatch:
- Add build_exec_args() and dispatch_command() to IntegrationBase
- Override for Claude (skills: /speckit-specify), Gemini (-m flag),
  Codex (codex exec), Copilot (--agent speckit.specify)
- CommandStep invokes installed commands by name via integration CLI
- Add PromptStep for arbitrary inline prompts (10th step type)
- Stream CLI output live to terminal (no silent blocking)
- Remove timeout when streaming (user can Ctrl+C)
- Ctrl+C saves state as PAUSED for clean resume

Interactive gates:
- Gate steps prompt [1] approve [2] reject in TTY
- Fall back to PAUSED in non-interactive environments
- Resume re-executes the gate for interactive prompting

Documentation:
- workflows/README.md — user guide
- workflows/ARCHITECTURE.md — internals with Mermaid diagrams
- workflows/PUBLISHING.md — catalog submission guide

Tests: 94 → 122 workflow tests, 1362 total (all passing)

* Fix ruff lint errors: unused imports, f-string placeholders, undefined name

* Address second review: registry-backed validation, shell failures, loop/fan-out execution, URL validation

- VALID_STEP_TYPES now queries STEP_REGISTRY dynamically
- Shell step returns FAILED on non-zero exit code
- Persist workflow YAML in run directory for reliable resume
- Resume loads from run copy, falls back to installed workflow
- Engine iterates while/do-while loops up to max_iterations
- Engine expands fan-out per item with context.item
- HTTPS URL validation for catalog workflow installs (HTTP allowed for localhost)
- Fix catalog merge priority docstring (lower number wins)
- Fix dispatch_command docstring (no build_exec_args_for_command)
- Gate on_reject=retry pauses for re-prompt on resume
- Update docs to 10 step types, add prompt step to tables and README

* Potential fix for pull request finding 'Empty except'

Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com>

* Address third review: fan-out IDs, catalog guards, shell coercion, docs

- Fan-out generates unique per-item step IDs and collects results
- Catalog merge skips non-dict workflow entries (malformed data guard)
- Shell step coerces run_cmd to str after expression evaluation
- urlopen timeout=30 for catalog workflow installs
- yaml.dump with sort_keys=False, allow_unicode=True for catalog configs
- Document streaming timeout as intentionally unbounded (user Ctrl+C)
- Document --allow-all-tools as required for non-interactive + future enhancement
- Update test docstring and PUBLISHING.md to 10 step types with prompt

* Validate final URL after redirects in catalog fetch

urlopen follows redirects, so validate the response URL against the
same HTTPS/localhost rules to prevent redirect-based downgrade attacks.

* Address fourth review: filter arg eval, tags normalization, install redirect check

- Filter arguments now evaluated via _evaluate_simple_expression() so
  default(42) returns int not string
- Tags normalized: non-list/non-string values handled gracefully
- Install URL redirect validation (same as catalog fetch)
- Remove unused 'skipped' variable in catalog config parsing
- Author 'github' → 'GitHub' in example workflow
- Document nested step resume limitation (re-runs parent step)

* Add explanatory comment to empty except ValueError block

* Address fifth review: expression parsing, fan-out output, URL install, gate options

- Move string literal parsing before operator detection in expressions
  so quoted strings with operators (e.g. 'a in b') are not mis-parsed
- Fan-out: remove max_concurrency from persisted output, fix docstring
  to reflect sequential execution
- workflow add: support URL sources with HTTPS/redirect validation,
  validate workflow ID is non-empty before writing files
- Deduplicate local install logic via _validate_and_install_local()
- Remove 'edit' gate option from speckit workflow (not implemented)

* Add comments to empty except ValueError blocks in URL install

* Address sixth review: operator precedence, fan_in cleanup, registry resilience, docs

- Fix or/and operator precedence (or parsed first = lower precedence)
- Restore context.fan_in after fan-in step completes
- Catch JSONDecodeError in registry load for corrupted files
- Replace print() with on_step_start callback (library-safe)
- Gate validation warns when on_reject set but no reject option
- Shell step: document shell=True security tradeoff
- README: sdd-pipeline → speckit, parallel → sequential for fan-out
- ARCHITECTURE.md: parallel → fan-out/fan-in in diagram

* Address seventh review: string literal before pipe, type annotations, validate on install

- Move string literal check above pipe filter parsing so 'a | b' works
- Fix type annotations: input_values list[str] | None, run_id str | None
- Run validate_workflow() before installing from local path/URL
- Remove duplicate string literal check from expression parser

* Address eighth review: fan-out namespaced IDs, early return, catalog validation

- Fan-out per-item step IDs use _fanout_{step_id}_{base}_{idx} namespace
  to avoid collisions with user-defined step IDs
- Early return after fan-out loop when state is paused/failed/aborted
- Catalog installs parse + validate downloaded YAML before registering,
  using definition metadata instead of catalog entry for registry

* Address ninth review: populate catalog, fix indentation, priority, README

- Add speckit workflow entry to catalog.json so it's discoverable
- Fix shell step output dict indentation
- Catalog add_catalog priority derived from max existing + 1
- README Quick Start clarified with install + local file examples

* Address tenth review: max_iterations validation, catalog config guard, version alignment

- Validate max_iterations is int >= 1 in while and do-while steps
- Guard add_catalog against corrupted config (non-dict/non-list)
- Align speckit_version requirement to >=0.6.1 (current package version)
- Fan-out template validation uses separate seen_ids set to avoid
  false duplication errors with user-defined step IDs

* Address eleventh review: command step fails without CLI, ID mismatch warning, state persistence

- Command step returns FAILED when CLI not installed (was silent COMPLETED)
- Catalog install warns on workflow ID vs catalog key mismatch
- Engine persists state.save() before returning on unknown step type
- Update tests to expect FAILED for command steps without CLI
- Integration tests use shell steps for CLI-independent execution

* Address twelfth review: type annotations, version examples, streaming docs, requires

- Fix workflow_search type annotations (str | None)
- PUBLISHING.md: speckit_version >=0.15.0 → >=0.6.1
- Document that exit_code is captured and referenceable by later steps
- Mark requires as declared-but-not-enforced (planned enhancement)
- Note full stdout/stderr capture as planned enhancement

* Enforce catalog key matches workflow ID (fail instead of warn)

* Bundle speckit workflow: auto-install during specify init

- Add workflows/speckit to pyproject.toml force-include for wheel builds
- Add _locate_bundled_workflow() helper (mirrors _locate_bundled_extension)
- Auto-install speckit workflow during specify init (after git extension)
- Update all integration file inventory tests to expect workflow files

* Address fourteenth review: prompt fails without CLI, resolved step data, fan-out normalization

- PromptStep returns FAILED when CLI not installed (was silent COMPLETED)
- Engine step_data prefers resolved values from step output
- Fan-out normalizes output.results=[] for empty item lists
- subprocess.run inherits stdout/stderr (no explicit sys.stdout)
- Registry tests use issubset for extensibility

* Address fifteenth review: fan_in docstring, gate defaults, validation guards, reserved prefix

- FanInStep docstring: aggregate-only, no blocking semantics
- FanInStep: guard output_config as dict, handle None
- Gate validate: use same default options as execute
- Validate inputs is dict and steps is list before iterating
- Reserve _fanout_ prefix in step ID validation
- PUBLISHING.md: remove unenforced checklist items, add _fanout_ note

* Address sixteenth review: docs regex, fan_in try/finally, hyphenated dot-path keys

- PUBLISHING.md: update ID regex docs to match implementation (single-char OK)
- FanInStep: wrap expression evaluation in try/finally for context.fan_in
- Expression dot-path: allow hyphens in keys before list index (e.g. run-tests[0])

* Make speckit workflow integration-agnostic, document Copilot CLI requirement

- Workflow integration selectable via input (default: claude)
- Each command step uses {{ inputs.integration }} instead of hardcoded copilot
- Copilot docstring documents CLI requirement for workflow dispatch
- Added install_url for Copilot CLI docs

* Address seventeenth review: project checks, catalog robustness

- Add .specify/ project check to workflow run/resume/status/search/info
- remove_catalog validates config shape (dict + list) before indexing
- _fetch_single_catalog validates response is a dict
- _get_merged_workflows raises when all catalogs fail to fetch
- add_catalog guards against non-dict catalog entries in config

* Address eighteenth review: condition coercion, gate abort result, while default, cache guard, resume log

- evaluate_condition treats plain 'false'/'true' strings as booleans
- Gate abort returns StepResult(FAILED) instead of raising exception
  so step output is persisted in state for inspection
- while_loop max_iterations optional (default 10), validation aligned
- Catalog cache fallback catches invalid JSON gracefully
- resume() appends workflow_finished log entry like execute()

* Address nineteenth review: allow-all-tools opt-in, empty catalogs, abort dead code, while docstring

- --allow-all-tools controlled by SPECKIT_ALLOW_ALL_TOOLS env var (default: 1)
  Set to 0 to disable automatic tool approval for Copilot CLI
- Empty catalogs list falls back to built-in defaults (not an error)
- Remove unreachable WorkflowAbortError catches from execute/resume
  (gate abort now returns StepResult(FAILED) instead of raising)
- while_loop docstring updated: max_iterations is optional (default 10)

* Address twentieth review: gate abort maps to ABORTED status, do-while max_iterations optional

- Engine detects output.aborted from gate step and sets RunStatus.ABORTED
  (was unreachable — gate abort returned FAILED but status was always FAILED)
- do-while max_iterations now optional (default 10), aligned with while_loop
- do-while docstring and validation updated accordingly

* Coerce default_options to dict, align bundled workflow ID regex with validator

* Gate validates string options, prompt uses resolved integration, loop normalizes max_iterations

* Use parentId:childId convention for nested step IDs

- Fan-out per-item IDs use parentId:templateId:index (e.g. parallel:impl:0)
- Reserve ':' in user step IDs (validation rejects)
- Replaces _fanout_ prefix with cleaner namespacing
- Expressions like {{ steps.parallel:impl:0.output.file }} work naturally

* Validate workflow version is semantic versioning (X.Y.Z)

* Schema version validation, strict semver, load_workflow docstring, preserve max_concurrency

- Validate schema_version is '1.0' (reject unknown future schemas)
- Strict semver regex: ^\d+\.\d+\.\d+$ (rejects 1.0.0beta etc.)
- load_workflow docstring: 'parsed' not 'validated'
- Keep max_concurrency in fan-out output (was dropped)
- do_while docstring: engine re-evaluates step_config condition
- ARCHITECTURE.md: document nested resume limitation

* Path traversal prevention, loop step ID namespacing

- RunState validates run_id is alphanumeric+hyphens (no path separators)
- workflow_add validates catalog source doesn't escape workflows_dir
- Loop iterations namespace nested step IDs as parentId:childId:iteration
  so multiple iterations don't overwrite each other in context/state

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: mnriem <15701806+mnriem@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com>

src/specify_cli/workflows/steps/__init__.py

@@ -0,0 +1 @@
+"""Auto-discovery for built-in step types."""

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

@@ -0,0 +1,155 @@
+"""Command step — dispatches a Spec Kit command to an integration CLI."""
+
+from __future__ import annotations
+
+import shutil
+from pathlib import Path
+from typing import Any
+
+from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus
+from specify_cli.workflows.expressions import evaluate_expression
+
+
+class CommandStep(StepBase):
+    """Default step type — invokes a Spec Kit command via the integration CLI.
+
+    The command files (skills, markdown, TOML) are already installed in
+    the integration's directory on disk.  This step tells the CLI to
+    execute the command by name (e.g. ``/speckit.specify`` or
+    ``/speckit-specify``) rather than reading the file contents.
+
+    .. note::
+
+        CLI output is streamed to the terminal for live progress.
+        ``output.exit_code`` is always captured and can be referenced
+        by later steps (e.g. ``{{ steps.specify.output.exit_code }}``).
+        Full ``stdout``/``stderr`` capture is a planned enhancement.
+    """
+
+    type_key = "command"
+
+    def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
+        command = config.get("command", "")
+        input_data = config.get("input", {})
+
+        # Resolve expressions in input
+        resolved_input: dict[str, Any] = {}
+        for key, value in input_data.items():
+            resolved_input[key] = evaluate_expression(value, context)
+
+        # Resolve integration (step → workflow default → project default)
+        integration = config.get("integration") or context.default_integration
+        if integration and isinstance(integration, str) and "{{" in integration:
+            integration = evaluate_expression(integration, context)
+
+        # Resolve model
+        model = config.get("model") or context.default_model
+        if model and isinstance(model, str) and "{{" in model:
+            model = evaluate_expression(model, context)
+
+        # Merge options (workflow defaults ← step overrides)
+        options = dict(context.default_options)
+        step_options = config.get("options", {})
+        if step_options:
+            options.update(step_options)
+
+        # Attempt CLI dispatch
+        args_str = str(resolved_input.get("args", ""))
+        dispatch_result = self._try_dispatch(
+            command, integration, model, args_str, context
+        )
+
+        output: dict[str, Any] = {
+            "command": command,
+            "integration": integration,
+            "model": model,
+            "options": options,
+            "input": resolved_input,
+        }
+
+        if dispatch_result is not None:
+            output["exit_code"] = dispatch_result["exit_code"]
+            output["stdout"] = dispatch_result["stdout"]
+            output["stderr"] = dispatch_result["stderr"]
+            output["dispatched"] = True
+            if dispatch_result["exit_code"] != 0:
+                return StepResult(
+                    status=StepStatus.FAILED,
+                    output=output,
+                    error=dispatch_result["stderr"] or f"Command exited with code {dispatch_result['exit_code']}",
+                )
+            return StepResult(
+                status=StepStatus.COMPLETED,
+                output=output,
+            )
+        else:
+            output["exit_code"] = 1
+            output["dispatched"] = False
+            return StepResult(
+                status=StepStatus.FAILED,
+                output=output,
+                error=(
+                    f"Cannot dispatch command {command!r}: "
+                    f"integration {integration!r} CLI not found or not installed. "
+                    f"Install the CLI tool or check 'specify integration list'."
+                ),
+            )
+
+    @staticmethod
+    def _try_dispatch(
+        command: str,
+        integration_key: str | None,
+        model: str | None,
+        args: str,
+        context: StepContext,
+    ) -> dict[str, Any] | None:
+        """Invoke *command* by name through the integration CLI.
+
+        The integration's ``dispatch_command`` builds the native
+        slash-command invocation (e.g. ``/speckit.specify`` for
+        markdown agents, ``/speckit-specify`` for skills agents),
+        then executes the CLI non-interactively.
+
+        Returns the dispatch result dict, or ``None`` if dispatch is
+        not possible (integration not found, CLI not installed, or
+        dispatch not supported).
+        """
+        if not integration_key:
+            return None
+
+        try:
+            from specify_cli.integrations import get_integration
+        except ImportError:
+            return None
+
+        impl = get_integration(integration_key)
+        if impl is None:
+            return None
+
+        # Check if the integration supports CLI dispatch
+        if impl.build_exec_args("test") is None:
+            return None
+
+        # Check if the CLI tool is actually installed
+        if not shutil.which(impl.key):
+            return None
+
+        project_root = Path(context.project_root) if context.project_root else None
+
+        try:
+            return impl.dispatch_command(
+                command,
+                args=args,
+                project_root=project_root,
+                model=model,
+            )
+        except (NotImplementedError, OSError):
+            return None
+
+    def validate(self, config: dict[str, Any]) -> list[str]:
+        errors = super().validate(config)
+        if "command" not in config:
+            errors.append(
+                f"Command step {config.get('id', '?')!r} is missing 'command' field."
+            )
+        return errors

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

@@ -0,0 +1,61 @@
+"""Do-While loop step — execute at least once, then repeat while condition is truthy."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus
+
+
+class DoWhileStep(StepBase):
+    """Execute body at least once, then check condition.
+
+    Continues while condition is truthy.  ``max_iterations`` is an
+    optional safety cap (defaults to 10 if omitted).
+
+    The first invocation always returns the nested steps for execution.
+    The engine re-evaluates ``step_config['condition']`` after each
+    iteration to decide whether to loop again.
+    """
+
+    type_key = "do-while"
+
+    def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
+        max_iterations = config.get("max_iterations")
+        if max_iterations is None:
+            max_iterations = 10
+        nested_steps = config.get("steps", [])
+        condition = config.get("condition", "false")
+
+        # Always execute body at least once; the engine layer evaluates
+        # `condition` after each iteration to decide whether to loop.
+        return StepResult(
+            status=StepStatus.COMPLETED,
+            output={
+                "condition": condition,
+                "max_iterations": max_iterations,
+                "loop_type": "do-while",
+            },
+            next_steps=nested_steps,
+        )
+
+    def validate(self, config: dict[str, Any]) -> list[str]:
+        errors = super().validate(config)
+        if "condition" not in config:
+            errors.append(
+                f"Do-while step {config.get('id', '?')!r} is missing "
+                f"'condition' field."
+            )
+        max_iter = config.get("max_iterations")
+        if max_iter is not None:
+            if not isinstance(max_iter, int) or max_iter < 1:
+                errors.append(
+                    f"Do-while step {config.get('id', '?')!r}: "
+                    f"'max_iterations' must be an integer >= 1."
+                )
+        nested = config.get("steps", [])
+        if not isinstance(nested, list):
+            errors.append(
+                f"Do-while step {config.get('id', '?')!r}: 'steps' must be a list."
+            )
+        return errors

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

@@ -0,0 +1,61 @@
+"""Fan-in step — join point for parallel steps."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus
+from specify_cli.workflows.expressions import evaluate_expression
+
+
+class FanInStep(StepBase):
+    """Join point that aggregates results from ``wait_for:`` steps.
+
+    Reads completed step outputs from ``context.steps`` and collects
+    them into ``output.results``.  Does not block; relies on the
+    engine executing steps sequentially.
+    """
+
+    type_key = "fan-in"
+
+    def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
+        wait_for = config.get("wait_for", [])
+        output_config = config.get("output") or {}
+        if not isinstance(output_config, dict):
+            output_config = {}
+
+        # Collect results from referenced steps
+        results = []
+        for step_id in wait_for:
+            step_data = context.steps.get(step_id, {})
+            results.append(step_data.get("output", {}))
+
+        # Resolve output expressions with fan_in in context
+        prev_fan_in = getattr(context, "fan_in", None)
+        context.fan_in = {"results": results}
+        resolved_output: dict[str, Any] = {"results": results}
+
+        try:
+            for key, expr in output_config.items():
+                if isinstance(expr, str) and "{{" in expr:
+                    resolved_output[key] = evaluate_expression(expr, context)
+                else:
+                    resolved_output[key] = expr
+        finally:
+            # Restore previous fan_in state even if evaluation fails
+            context.fan_in = prev_fan_in
+
+        return StepResult(
+            status=StepStatus.COMPLETED,
+            output=resolved_output,
+        )
+
+    def validate(self, config: dict[str, Any]) -> list[str]:
+        errors = super().validate(config)
+        wait_for = config.get("wait_for", [])
+        if not isinstance(wait_for, list) or not wait_for:
+            errors.append(
+                f"Fan-in step {config.get('id', '?')!r}: "
+                f"'wait_for' must be a non-empty list of step IDs."
+            )
+        return errors

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

@@ -0,0 +1,58 @@
+"""Fan-out step — dispatch a step template over a collection."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus
+from specify_cli.workflows.expressions import evaluate_expression
+
+
+class FanOutStep(StepBase):
+    """Dispatch a step template for each item in a collection.
+
+    The engine executes the nested ``step:`` template once per item,
+    setting ``context.item`` for each iteration.  Execution is
+    currently sequential; ``max_concurrency`` is accepted but not
+    enforced.
+    """
+
+    type_key = "fan-out"
+
+    def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
+        items_expr = config.get("items", "[]")
+        items = evaluate_expression(items_expr, context)
+        if not isinstance(items, list):
+            items = []
+
+        max_concurrency = config.get("max_concurrency", 1)
+        step_template = config.get("step", {})
+
+        return StepResult(
+            status=StepStatus.COMPLETED,
+            output={
+                "items": items,
+                "max_concurrency": max_concurrency,
+                "step_template": step_template,
+                "item_count": len(items),
+            },
+        )
+
+    def validate(self, config: dict[str, Any]) -> list[str]:
+        errors = super().validate(config)
+        if "items" not in config:
+            errors.append(
+                f"Fan-out step {config.get('id', '?')!r} is missing "
+                f"'items' field."
+            )
+        if "step" not in config:
+            errors.append(
+                f"Fan-out step {config.get('id', '?')!r} is missing "
+                f"'step' field (nested step template)."
+            )
+        step = config.get("step")
+        if step is not None and not isinstance(step, dict):
+            errors.append(
+                f"Fan-out step {config.get('id', '?')!r}: 'step' must be a mapping."
+            )
+        return errors

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

@@ -0,0 +1,121 @@
+"""Gate step — human review gate."""
+
+from __future__ import annotations
+
+import sys
+from typing import Any
+
+from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus
+from specify_cli.workflows.expressions import evaluate_expression
+
+
+class GateStep(StepBase):
+    """Interactive review gate.
+
+    When running in an interactive terminal, prompts the user to choose
+    an option (e.g. approve / reject).  Falls back to ``PAUSED`` when
+    stdin is not a TTY (CI, piped input) so the run can be resumed
+    later with ``specify workflow resume``.
+
+    The user's choice is stored in ``output.choice``.  ``on_reject``
+    controls abort / skip behaviour.
+    """
+
+    type_key = "gate"
+
+    def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
+        message = config.get("message", "Review required.")
+        if isinstance(message, str) and "{{" in message:
+            message = evaluate_expression(message, context)
+
+        options = config.get("options", ["approve", "reject"])
+        on_reject = config.get("on_reject", "abort")
+
+        show_file = config.get("show_file")
+        if show_file and isinstance(show_file, str) and "{{" in show_file:
+            show_file = evaluate_expression(show_file, context)
+
+        output = {
+            "message": message,
+            "options": options,
+            "on_reject": on_reject,
+            "show_file": show_file,
+            "choice": None,
+        }
+
+        # Non-interactive: pause for later resume
+        if not sys.stdin.isatty():
+            return StepResult(status=StepStatus.PAUSED, output=output)
+
+        # Interactive: prompt the user
+        choice = self._prompt(message, options)
+        output["choice"] = choice
+
+        if choice in ("reject", "abort"):
+            if on_reject == "abort":
+                output["aborted"] = True
+                return StepResult(
+                    status=StepStatus.FAILED,
+                    output=output,
+                    error=f"Gate rejected by user at step {config.get('id', '?')!r}",
+                )
+            if on_reject == "retry":
+                # Pause so the next resume re-executes this gate
+                return StepResult(status=StepStatus.PAUSED, output=output)
+            # on_reject == "skip" → completed, downstream steps decide
+            return StepResult(status=StepStatus.COMPLETED, output=output)
+
+        return StepResult(status=StepStatus.COMPLETED, output=output)
+
+    @staticmethod
+    def _prompt(message: str, options: list[str]) -> str:
+        """Display gate message and prompt for a choice."""
+        print("\n  ┌─ Gate ─────────────────────────────────────")
+        print(f"  │ {message}")
+        print("  │")
+        for i, opt in enumerate(options, 1):
+            print(f"  │  [{i}] {opt}")
+        print("  └────────────────────────────────────────────")
+
+        while True:
+            try:
+                raw = input(f"  Choose [1-{len(options)}]: ").strip()
+            except (EOFError, KeyboardInterrupt):
+                print()
+                return options[-1]  # default to last (usually reject)
+            if raw.isdigit() and 1 <= int(raw) <= len(options):
+                return options[int(raw) - 1]
+            # Also accept the option name directly
+            if raw.lower() in [o.lower() for o in options]:
+                return next(o for o in options if o.lower() == raw.lower())
+            print(f"  Invalid choice. Enter 1-{len(options)} or an option name.")
+
+    def validate(self, config: dict[str, Any]) -> list[str]:
+        errors = super().validate(config)
+        if "message" not in config:
+            errors.append(
+                f"Gate step {config.get('id', '?')!r} is missing 'message' field."
+            )
+        options = config.get("options", ["approve", "reject"])
+        if not isinstance(options, list) or not options:
+            errors.append(
+                f"Gate step {config.get('id', '?')!r}: 'options' must be a non-empty list."
+            )
+        elif not all(isinstance(o, str) for o in options):
+            errors.append(
+                f"Gate step {config.get('id', '?')!r}: all options must be strings."
+            )
+        on_reject = config.get("on_reject", "abort")
+        if on_reject not in ("abort", "skip", "retry"):
+            errors.append(
+                f"Gate step {config.get('id', '?')!r}: 'on_reject' must be "
+                f"'abort', 'skip', or 'retry'."
+            )
+        if on_reject in ("abort", "retry") and isinstance(options, list):
+            reject_choices = {"reject", "abort"}
+            if not any(o.lower() in reject_choices for o in options):
+                errors.append(
+                    f"Gate step {config.get('id', '?')!r}: on_reject={on_reject!r} "
+                    f"but options has no 'reject' or 'abort' choice."
+                )
+        return errors

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

@@ -0,0 +1,55 @@
+"""If/Then/Else step — conditional branching."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus
+from specify_cli.workflows.expressions import evaluate_condition
+
+
+class IfThenStep(StepBase):
+    """Branch based on a boolean condition expression.
+
+    Both ``then:`` and ``else:`` contain inline step arrays — full step
+    definitions, not ID references.
+    """
+
+    type_key = "if"
+
+    def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
+        condition = config.get("condition", False)
+        result = evaluate_condition(condition, context)
+
+        if result:
+            branch = config.get("then", [])
+        else:
+            branch = config.get("else", [])
+
+        return StepResult(
+            status=StepStatus.COMPLETED,
+            output={"condition_result": result},
+            next_steps=branch,
+        )
+
+    def validate(self, config: dict[str, Any]) -> list[str]:
+        errors = super().validate(config)
+        if "condition" not in config:
+            errors.append(
+                f"If step {config.get('id', '?')!r} is missing 'condition' field."
+            )
+        if "then" not in config:
+            errors.append(
+                f"If step {config.get('id', '?')!r} is missing 'then' field."
+            )
+        then_branch = config.get("then", [])
+        if not isinstance(then_branch, list):
+            errors.append(
+                f"If step {config.get('id', '?')!r}: 'then' must be a list of steps."
+            )
+        else_branch = config.get("else", [])
+        if else_branch and not isinstance(else_branch, list):
+            errors.append(
+                f"If step {config.get('id', '?')!r}: 'else' must be a list of steps."
+            )
+        return errors

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

@@ -0,0 +1,156 @@
+"""Prompt step — sends an arbitrary prompt to an integration CLI."""
+
+from __future__ import annotations
+
+import shutil
+from pathlib import Path
+from typing import Any
+
+from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus
+from specify_cli.workflows.expressions import evaluate_expression
+
+
+class PromptStep(StepBase):
+    """Send a free-form prompt to an integration CLI.
+
+    Unlike ``CommandStep`` which invokes an installed Spec Kit command
+    by name (e.g. ``/speckit.specify`` or ``/speckit-specify``),
+    ``PromptStep`` sends an arbitrary inline ``prompt:`` string
+    directly to the CLI.  This is useful for ad-hoc instructions
+    that don't map to a registered command.
+
+    .. note::
+
+        CLI output is streamed to the terminal for live progress.
+        ``output.exit_code`` is always captured and can be referenced
+        by later steps.  Full response text capture is a planned
+        enhancement.
+
+    Example YAML::
+
+        - id: review-security
+          type: prompt
+          prompt: "Review {{ inputs.file }} for security vulnerabilities"
+          integration: claude
+    """
+
+    type_key = "prompt"
+
+    def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
+        prompt_template = config.get("prompt", "")
+        prompt = evaluate_expression(prompt_template, context)
+        if not isinstance(prompt, str):
+            prompt = str(prompt)
+
+        # Resolve integration (step → workflow default)
+        integration = config.get("integration") or context.default_integration
+        if integration and isinstance(integration, str) and "{{" in integration:
+            integration = evaluate_expression(integration, context)
+
+        # Resolve model
+        model = config.get("model") or context.default_model
+        if model and isinstance(model, str) and "{{" in model:
+            model = evaluate_expression(model, context)
+
+        # Attempt CLI dispatch
+        dispatch_result = self._try_dispatch(
+            prompt, integration, model, context
+        )
+
+        output: dict[str, Any] = {
+            "prompt": prompt,
+            "integration": integration,
+            "model": model,
+        }
+
+        if dispatch_result is not None:
+            output["exit_code"] = dispatch_result["exit_code"]
+            output["stdout"] = dispatch_result["stdout"]
+            output["stderr"] = dispatch_result["stderr"]
+            output["dispatched"] = True
+            if dispatch_result["exit_code"] != 0:
+                return StepResult(
+                    status=StepStatus.FAILED,
+                    output=output,
+                    error=(
+                        dispatch_result["stderr"]
+                        or f"Prompt exited with code {dispatch_result['exit_code']}"
+                    ),
+                )
+            return StepResult(
+                status=StepStatus.COMPLETED,
+                output=output,
+            )
+        else:
+            output["exit_code"] = 1
+            output["dispatched"] = False
+            return StepResult(
+                status=StepStatus.FAILED,
+                output=output,
+                error=(
+                    f"Cannot dispatch prompt: "
+                    f"integration {integration!r} "
+                    f"CLI not found or not installed."
+                ),
+            )
+
+    @staticmethod
+    def _try_dispatch(
+        prompt: str,
+        integration_key: str | None,
+        model: str | None,
+        context: StepContext,
+    ) -> dict[str, Any] | None:
+        """Dispatch *prompt* directly through the integration CLI."""
+        if not integration_key or not prompt:
+            return None
+
+        try:
+            from specify_cli.integrations import get_integration
+        except ImportError:
+            return None
+
+        impl = get_integration(integration_key)
+        if impl is None:
+            return None
+
+        exec_args = impl.build_exec_args(prompt, model=model, output_json=False)
+        if exec_args is None:
+            return None
+
+        if not shutil.which(impl.key):
+            return None
+
+        import subprocess
+
+        project_root = (
+            Path(context.project_root) if context.project_root else Path.cwd()
+        )
+
+        try:
+            result = subprocess.run(
+                exec_args,
+                text=True,
+                cwd=str(project_root),
+            )
+            return {
+                "exit_code": result.returncode,
+                "stdout": "",
+                "stderr": "",
+            }
+        except KeyboardInterrupt:
+            return {
+                "exit_code": 130,
+                "stdout": "",
+                "stderr": "Interrupted by user",
+            }
+        except OSError:
+            return None
+
+    def validate(self, config: dict[str, Any]) -> list[str]:
+        errors = super().validate(config)
+        if "prompt" not in config:
+            errors.append(
+                f"Prompt step {config.get('id', '?')!r} is missing 'prompt' field."
+            )
+        return errors

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

@@ -0,0 +1,75 @@
+"""Shell step — run a local shell command."""
+
+from __future__ import annotations
+
+import subprocess
+from typing import Any
+
+from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus
+from specify_cli.workflows.expressions import evaluate_expression
+
+
+class ShellStep(StepBase):
+    """Run a local shell command (non-agent).
+
+    Captures exit code and stdout/stderr.
+    """
+
+    type_key = "shell"
+
+    def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
+        run_cmd = config.get("run", "")
+        if isinstance(run_cmd, str) and "{{" in run_cmd:
+            run_cmd = evaluate_expression(run_cmd, context)
+        run_cmd = str(run_cmd)
+
+        cwd = context.project_root or "."
+
+        # NOTE: shell=True is required to support pipes, redirects, and
+        # multi-command expressions in workflow YAML.  Workflow authors
+        # control commands; catalog-installed workflows should be reviewed
+        # before use (see PUBLISHING.md for security guidance).
+        try:
+            proc = subprocess.run(
+                run_cmd,
+                shell=True,
+                capture_output=True,
+                text=True,
+                cwd=cwd,
+                timeout=300,
+            )
+            output = {
+                "exit_code": proc.returncode,
+                "stdout": proc.stdout,
+                "stderr": proc.stderr,
+            }
+            if proc.returncode != 0:
+                return StepResult(
+                    status=StepStatus.FAILED,
+                    error=f"Shell command exited with code {proc.returncode}.",
+                    output=output,
+                )
+            return StepResult(
+                status=StepStatus.COMPLETED,
+                output=output,
+            )
+        except subprocess.TimeoutExpired:
+            return StepResult(
+                status=StepStatus.FAILED,
+                error="Shell command timed out after 300 seconds.",
+                output={"exit_code": -1, "stdout": "", "stderr": "timeout"},
+            )
+        except OSError as exc:
+            return StepResult(
+                status=StepStatus.FAILED,
+                error=f"Shell command failed: {exc}",
+                output={"exit_code": -1, "stdout": "", "stderr": str(exc)},
+            )
+
+    def validate(self, config: dict[str, Any]) -> list[str]:
+        errors = super().validate(config)
+        if "run" not in config:
+            errors.append(
+                f"Shell step {config.get('id', '?')!r} is missing 'run' field."
+            )
+        return errors

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

@@ -0,0 +1,70 @@
+"""Switch step — multi-branch dispatch."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus
+from specify_cli.workflows.expressions import evaluate_expression
+
+
+class SwitchStep(StepBase):
+    """Multi-branch dispatch on an expression.
+
+    Evaluates ``expression:`` once, matches against ``cases:`` keys
+    (exact match, string-coerced).  Falls through to ``default:`` if
+    no case matches.
+    """
+
+    type_key = "switch"
+
+    def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
+        expression = config.get("expression", "")
+        value = evaluate_expression(expression, context)
+
+        # String-coerce for matching
+        str_value = str(value) if value is not None else ""
+
+        cases = config.get("cases", {})
+        for case_key, case_steps in cases.items():
+            if str(case_key) == str_value:
+                return StepResult(
+                    status=StepStatus.COMPLETED,
+                    output={"matched_case": str(case_key), "expression_value": value},
+                    next_steps=case_steps,
+                )
+
+        # Default fallback
+        default_steps = config.get("default", [])
+        return StepResult(
+            status=StepStatus.COMPLETED,
+            output={"matched_case": "__default__", "expression_value": value},
+            next_steps=default_steps,
+        )
+
+    def validate(self, config: dict[str, Any]) -> list[str]:
+        errors = super().validate(config)
+        if "expression" not in config:
+            errors.append(
+                f"Switch step {config.get('id', '?')!r} is missing "
+                f"'expression' field."
+            )
+        cases = config.get("cases", {})
+        if not isinstance(cases, dict):
+            errors.append(
+                f"Switch step {config.get('id', '?')!r}: 'cases' must be a mapping."
+            )
+        else:
+            for key, val in cases.items():
+                if not isinstance(val, list):
+                    errors.append(
+                        f"Switch step {config.get('id', '?')!r}: "
+                        f"case {key!r} must be a list of steps."
+                    )
+        default = config.get("default")
+        if default is not None and not isinstance(default, list):
+            errors.append(
+                f"Switch step {config.get('id', '?')!r}: "
+                f"'default' must be a list of steps."
+            )
+        return errors

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

@@ -0,0 +1,68 @@
+"""While loop step — repeat while condition is truthy."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus
+from specify_cli.workflows.expressions import evaluate_condition
+
+
+class WhileStep(StepBase):
+    """Repeat nested steps while condition is truthy.
+
+    Evaluates condition *before* each iteration.  If falsy on first
+    check, the body never runs.  ``max_iterations`` is an optional
+    safety cap (defaults to 10 if omitted).
+    """
+
+    type_key = "while"
+
+    def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
+        condition = config.get("condition", False)
+        max_iterations = config.get("max_iterations")
+        if max_iterations is None:
+            max_iterations = 10
+        nested_steps = config.get("steps", [])
+
+        result = evaluate_condition(condition, context)
+        if result:
+            return StepResult(
+                status=StepStatus.COMPLETED,
+                output={
+                    "condition_result": True,
+                    "max_iterations": max_iterations,
+                    "loop_type": "while",
+                },
+                next_steps=nested_steps,
+            )
+
+        return StepResult(
+            status=StepStatus.COMPLETED,
+            output={
+                "condition_result": False,
+                "max_iterations": max_iterations,
+                "loop_type": "while",
+            },
+        )
+
+    def validate(self, config: dict[str, Any]) -> list[str]:
+        errors = super().validate(config)
+        if "condition" not in config:
+            errors.append(
+                f"While step {config.get('id', '?')!r} is missing "
+                f"'condition' field."
+            )
+        max_iter = config.get("max_iterations")
+        if max_iter is not None:
+            if not isinstance(max_iter, int) or max_iter < 1:
+                errors.append(
+                    f"While step {config.get('id', '?')!r}: "
+                    f"'max_iterations' must be an integer >= 1."
+                )
+        nested = config.get("steps", [])
+        if not isinstance(nested, list):
+            errors.append(
+                f"While step {config.get('id', '?')!r}: 'steps' must be a list."
+            )
+        return errors