feat(workflows): add --dry-run flag to specify workflow run

src/specify_cli/__init__.py

@@ -810,6 +810,18 @@ def workflow_run(
         "--json",
         help="Emit the run outcome as a single JSON object instead of formatted text.",
     ),
+    dry_run: bool = typer.Option(
+        False,
+        "--dry-run",
+        help=(
+            "Preview the workflow without dispatching any AI or shell "
+            "commands for built-in command, prompt, and gate steps. "
+            "Those steps emit a synthetic preview message; the run is "
+            "persisted so it can be inspected but not resumed to a "
+            "real run. Other step types (e.g. init, shell) may still "
+            "perform their normal work during dry-run."
+        ),
+    ),
 ):
     """Run a workflow from an installed ID or local YAML path."""
     from .workflows import load_custom_steps
@@ -857,20 +869,40 @@ def workflow_run(
     # Parse inputs
     inputs = _parse_input_values(input_values)
 
+    if dry_run and not json_output:
+        console.print(
+            "\n[bold yellow]DRY RUN:[/bold yellow] previewing without "
+            "dispatching any AI or shell commands."
+        )
+
     if not json_output:
         console.print(f"\n[bold cyan]Running workflow:[/bold cyan] {definition.name} ({definition.id})")
         console.print(f"[dim]Version: {definition.version}[/dim]\n")
 
     try:
         with _stdout_to_stderr_when(json_output):
-            state = engine.execute(definition, inputs)
+            state = engine.execute(definition, inputs, dry_run=dry_run)
     except ValueError as exc:
         console.print(f"[red]Error:[/red] {exc}")
         raise typer.Exit(1)
     except Exception as exc:
+        if dry_run and not json_output:
+            _print_dry_run_previews(getattr(exc, "partial_state", None))
+        if json_output:
+            partial = getattr(exc, "partial_state", None)
+            if partial:
+                _emit_workflow_json(_workflow_run_payload(partial))
+            raise typer.Exit(
+                _run_outcome_exit_code(
+                    partial.status.value if partial else "failed"
+                )
+            )
         console.print(f"[red]Workflow failed:[/red] {exc}")
         raise typer.Exit(1)
 
+    if dry_run and not json_output:
+        _print_dry_run_previews(state)
+
     if json_output:
         _emit_workflow_json(_workflow_run_payload(state))
         raise typer.Exit(_run_outcome_exit_code(state.status.value))
@@ -891,6 +923,45 @@ def workflow_run(
     raise typer.Exit(_run_outcome_exit_code(state.status.value))
 
 
+def _print_dry_run_previews(state: Any) -> None:
+    """Print the dry-run preview message emitted by each step.
+
+    Called by ``workflow run`` after a successful dry-run and from
+    exception handlers so a mid-run failure still surfaces the
+    previews resolved by earlier steps. Skipped silently when
+    ``state`` is ``None`` (e.g. the engine raised before any step
+    ran) or when the run did not include a dry-run step.
+    """
+    if state is None:
+        return
+    step_results = getattr(state, "step_results", None) or {}
+    if not step_results:
+        return
+    console.print("\n[bold yellow]DRY RUN previews:[/bold yellow]")
+    for step_id, result in step_results.items():
+        if not isinstance(result, dict):
+            continue
+        output = result.get("output") or {}
+        if not output.get("dry_run"):
+            continue
+        step_id_display = _escape_markup(str(step_id))
+        preview = output.get("dry_run_message") or output.get("message") or ""
+        preview_escaped = _escape_markup(preview)
+        console.print(f"  [cyan][{step_id_display}][/cyan] {preview_escaped}")
+
+
+def _escape_markup(text: str) -> str:
+    """Escape Rich markup characters so a step ID can be printed safely.
+
+    Step IDs are user-controlled YAML; without escaping, an ID
+    containing ``[`` or ``]`` would raise ``MarkupError`` from Rich.
+    """
+    return (
+        text.replace("[", "\\[")
+        .replace("]", "\\]")
+    )
+
+
 @workflow_app.command("resume")
 def workflow_resume(
     run_id: str = typer.Argument(..., help="Run ID to resume"),

src/specify_cli/workflows/base.py

@@ -74,6 +74,21 @@ class StepContext:
     #: Current run ID.
     run_id: str | None = None
 
+    #: When ``True``, the built-in step implementations
+    #: (``command`` / ``prompt`` / ``gate``) short-circuit and return a
+    #: synthetic ``StepResult`` carrying a preview of what would have
+    #: been dispatched — no subprocess, no CLI call, no network I/O for
+    #: those step types. Custom steps and built-in steps that have not
+    #: been updated to honor ``dry_run`` may still perform their normal
+    #: side effects; the flag is opt-in per step. Step implementations
+    #: publish the preview on ``output["dry_run_message"]`` (consumed
+    #: by the CLI's preview loop). ``output["message"]`` is also set to
+    #: the preview string during dry-run (it may differ from the real
+    #: run's ``message``). Downstream templates that need the original
+    #: value should reference the step's pre-dispatch config fields
+    #: (e.g. ``command``, ``prompt``) instead of ``output.message``.
+    dry_run: bool = False
+
 
 @dataclass
 class StepResult:

src/specify_cli/workflows/engine.py

@@ -334,6 +334,11 @@ def __init__(
         self.created_at = datetime.now(timezone.utc).isoformat()
         self.updated_at = self.created_at
         self.log_entries: list[dict[str, Any]] = []
+        #: Whether the run was started in dry-run mode. Persisted via
+        #: :meth:`save` so :meth:`load` (and the resumed run's
+        #: ``StepContext``) can keep the run in preview mode across
+        #: process restarts.
+        self.dry_run: bool = False
 
     @property
     def runs_dir(self) -> Path:
@@ -352,6 +357,7 @@ def save(self) -> None:
             "current_step_index": self.current_step_index,
             "current_step_id": self.current_step_id,
             "step_results": self.step_results,
+            "dry_run": self.dry_run,
             "created_at": self.created_at,
             "updated_at": self.updated_at,
         }
@@ -398,6 +404,7 @@ def load(cls, run_id: str, project_root: Path) -> RunState:
         state.step_results = state_data.get("step_results", {})
         state.created_at = state_data.get("created_at", "")
         state.updated_at = state_data.get("updated_at", "")
+        state.dry_run = state_data.get("dry_run", False)
 
         inputs_path = runs_dir / "inputs.json"
         if inputs_path.exists():
@@ -478,6 +485,7 @@ def execute(
         definition: WorkflowDefinition,
         inputs: dict[str, Any] | None = None,
         run_id: str | None = None,
+        dry_run: bool = False,
     ) -> RunState:
         """Execute a workflow definition.
 
@@ -489,6 +497,21 @@ def execute(
             User-provided input values.
         run_id:
             Optional run ID (uses SPECKIT_WORKFLOW_RUN_ID when set, otherwise auto-generated).
+        dry_run:
+            Preview-only mode. When ``True``, the built-in ``command``,
+            ``prompt`` and ``gate`` step implementations skip
+            side-effecting work (AI invocations, interactive prompts,
+            subprocess dispatches) and emit a synthetic
+            ``dry_run_message`` instead. Other built-in steps (``init``,
+            ``shell``, custom user-registered steps) currently still
+            execute their normal logic during a dry run; the flag is
+            opt-in per step. ``dry_run`` propagates into each step's
+            ``StepContext`` and is persisted on the resulting
+            ``RunState`` so ``resume()`` keeps the run in preview mode
+            across restarts. Step ``output`` shape is unchanged;
+            downstream ``switch``/``do-while`` gates coerce any
+            dry-run-only fields (e.g. ``output.choice``) so the preview
+            branch is deterministic.
 
         Returns
         -------
@@ -507,6 +530,7 @@ def execute(
             workflow_id=definition.id,
             project_root=self.project_root,
         )
+        state.dry_run = dry_run
 
         # Persist a copy of the workflow definition so resume can
         # reload it even if the original source is no longer available
@@ -531,6 +555,7 @@ def execute(
             default_options=definition.default_options,
             project_root=str(self.project_root),
             run_id=state.run_id,
+            dry_run=dry_run,
         )
 
         # Execute steps
@@ -545,6 +570,10 @@ def execute(
             state.status = RunStatus.FAILED
             state.append_log({"event": "workflow_failed", "error": str(exc)})
             state.save()
+            # Attach the partially-populated state so the CLI can render
+            # any dry-run previews resolved by earlier steps when the
+            # engine raises mid-run (e.g. template resolution failure).
+            exc.partial_state = state  # type: ignore[attr-defined]
             raise
 
         if state.status == RunStatus.RUNNING:
@@ -587,7 +616,8 @@ def resume(
             merged = {**state.inputs, **inputs}
             state.inputs = self._resolve_inputs(definition, merged)
 
-        # Restore context
+        # Restore context — including the persisted ``dry_run`` flag so an
+        # interrupted dry-run stays a dry-run after a process restart.
         context = StepContext(
             inputs=state.inputs,
             steps=state.step_results,
@@ -596,6 +626,7 @@ def resume(
             default_options=definition.default_options,
             project_root=str(self.project_root),
             run_id=state.run_id,
+            dry_run=state.dry_run,
         )
 
         from . import STEP_REGISTRY
@@ -622,6 +653,10 @@ def resume(
             state.status = RunStatus.FAILED
             state.append_log({"event": "resume_failed", "error": str(exc)})
             state.save()
+            # Same preview surface as ``execute()`` — when the engine
+            # raises mid-resume the CLI wants the partially-resolved
+            # dry-run previews for debugging.
+            exc.partial_state = state  # type: ignore[attr-defined]
             raise
 
         if state.status == RunStatus.RUNNING:

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

@@ -55,9 +55,6 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
 
         # 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,
@@ -67,11 +64,64 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
             "input": resolved_input,
         }
 
+        # Dry-run short-circuit — surface a synthetic preview of what a
+        # real run would have dispatched, without invoking the CLI.
+        if context.dry_run:
+            preview_invocation: str | None = None
+            if integration:
+                try:
+                    from specify_cli.integrations import get_integration
+
+                    impl = get_integration(integration)
+                except (ImportError, AttributeError, TypeError):
+                    impl = None
+                if impl is not None:
+                    try:
+                        preview_invocation = impl.build_command_invocation(
+                            command, args_str
+                        )
+                    except (ImportError, AttributeError, TypeError):
+                        # ImportError: integrations module not importable in
+                        #   minimal environments or test sandboxes.
+                        # AttributeError: integration class is missing
+                        #   ``build_command_invocation`` (older integration
+                        #   API).
+                        # TypeError: integration returned a non-string value
+                        #   from ``build_command_invocation``.
+                        # Anything else (e.g. a real bug inside the
+                        # integration) bubbles up so it's not silently
+                        # masked by the dry-run preview path.
+                        preview_invocation = None
+            if preview_invocation:
+                preview = f"DRY RUN: would invoke {preview_invocation!r}"
+            else:
+                preview = (
+                    f"DRY RUN: would invoke command {command!r} "
+                    f"(integration {integration!r}, args {args_str!r})"
+                )
+            output["exit_code"] = 0
+            output["dispatched"] = False
+            output["executed"] = False
+            output["dry_run"] = True
+            output["dry_run_message"] = preview
+            # Preserve the original command/integration/input so
+            # ``{{ steps.<id>.output.message }}`` keeps resolving to the
+            # original command description for downstream templates.
+            output["message"] = preview
+            output["invoke_command"] = preview_invocation or command
+            return StepResult(status=StepStatus.COMPLETED, output=output)
+
+        dispatch_result = self._try_dispatch(
+            command, integration, model, args_str, context
+        )
+
         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
+            output["executed"] = True
+            output["dry_run"] = False
             if dispatch_result["exit_code"] != 0:
                 return StepResult(
                     status=StepStatus.FAILED,
@@ -85,6 +135,8 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
         else:
             output["exit_code"] = 1
             output["dispatched"] = False
+            output["executed"] = False
+            output["dry_run"] = False
             return StepResult(
                 status=StepStatus.FAILED,
                 output=output,