Skip to content
55 changes: 54 additions & 1 deletion src/specify_cli/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -444,6 +444,16 @@ def _print_cli_warning(
from .commands import init as _init_cmd # noqa: E402
_init_cmd.register(app)

# Workflow commands are defined in-module below (see the
# ``workflow_app = typer.Typer(...)`` block near the end of this file).
# An earlier draft of #2661 also tried to register the
# ``src/specify_cli/commands/workflow.py`` module, which defined a second
# ``workflow`` Typer group with the same name. Typer raises on duplicate
# command names at startup, so the redundant registration has been
# removed here and ``commands/workflow.py`` deleted. The in-module
# commands (``specify workflow run``, ``... resume``, ``... status``,
# ``... list``, ``... add``, etc.) are the single source of truth.


@app.command()
def check():
Expand Down Expand Up @@ -568,6 +578,14 @@ def version(
app.add_typer(_self_app, name="self")


# NOTE: ``specify spec`` / ``specify plan`` were intentionally NOT added
# to this CLI. The ``specify`` CLI is scaffolding + workflow orchestration
# only; the per-stage surface (``/speckit.specify``, ``/speckit.plan``,
# \u2026) belongs to the agent, not the CLI. Adding a CLI shortcut would
# duplicate that surface with a weaker, second invocation path. See
# review #4624465842 from @mnriem on PR #2704.


# ===== Extension Commands =====

extension_app = typer.Typer(
Expand Down Expand Up @@ -2755,6 +2773,9 @@ def workflow_run(
input_values: list[str] | None = typer.Option(
None, "--input", "-i", help="Input values as key=value pairs"
),
dry_run: bool = typer.Option(
False, "--dry-run", help="Show the rendered prompt/inputs for each step without invoking the AI"
),
json_output: bool = typer.Option(
False,
"--json",
Expand Down Expand Up @@ -2809,9 +2830,17 @@ def workflow_run(
console.print(f"\n[bold cyan]Running workflow:[/bold cyan] {definition.name} ({definition.id})")
console.print(f"[dim]Version: {definition.version}[/dim]\n")

if dry_run and not json_output:
# When ``--json`` is set, the dry-run banner (and the per-step
# preview loop below) are suppressed entirely so stdout stays
# a single, well-formed JSON object. Redirecting to stderr
# would also work but is noisier than a script-friendly
# ``--json`` consumer actually wants.
console.print("[bold yellow]DRY RUN — no AI invocation will occur[/bold yellow]\n")

try:
with _stdout_to_stderr_when(json_output):
state = engine.execute(definition, inputs)
state = engine.execute(definition, inputs, dry_run=dry_run)
Comment thread
fuleinist marked this conversation as resolved.
Comment thread
mnriem marked this conversation as resolved.
except ValueError as exc:
console.print(f"[red]Error:[/red] {exc}")
raise typer.Exit(1)
Expand All @@ -2836,6 +2865,30 @@ def workflow_run(
if state.status.value == "paused":
console.print(f"\nResume with: [cyan]specify workflow resume {state.run_id}[/cyan]")

# Print dry-run step outputs so the user sees rendered command details.
# The dry-run preview loop runs regardless of the run's final status
# (completed / failed / paused) — when a step throws during template
# resolution, the previously-resolved dry-run previews are the
# most useful debug signal we can offer, and the user would
# otherwise lose them.
if dry_run:
for step_id, step_data in state.step_results.items():
output = step_data.get("output", {})
if output.get("dry_run"):
# Prefer the dedicated ``dry_run_message`` field (set
# by step implementations in dry-run mode) and fall
# back to ``message`` for compatibility with custom
# step types that have not adopted the new field.
msg = output.get("dry_run_message") or output.get("message", "")
if msg:
console.print(f"\n[bold cyan]Step:[/bold cyan] {step_id}")
# ``msg`` is plain text from the step implementation
# (e.g. ``[DRY RUN] Command: ...``). Disable Rich
# markup parsing so the literal ``[DRY RUN]`` bracket
# pair is shown verbatim and does not raise a
# ``MarkupError`` for an unknown tag.
console.print(msg, markup=False)
Comment thread
fuleinist marked this conversation as resolved.
Comment thread
mnriem marked this conversation as resolved.
Comment thread
mnriem marked this conversation as resolved.


@workflow_app.command("resume")
def workflow_resume(
Expand Down
3 changes: 3 additions & 0 deletions src/specify_cli/workflows/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,9 @@ class StepContext:
#: Current run ID.
run_id: str | None = None

#: Dry-run mode: preview rendered prompt/inputs without AI invocation.
dry_run: bool = False


@dataclass
class StepResult:
Expand Down
17 changes: 17 additions & 0 deletions src/specify_cli/workflows/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -331,6 +331,7 @@ def __init__(
self.current_step_id: str | None = None
self.step_results: dict[str, dict[str, Any]] = {}
self.inputs: dict[str, Any] = {}
self.dry_run: bool = False
self.created_at = datetime.now(timezone.utc).isoformat()
self.updated_at = self.created_at
self.log_entries: list[dict[str, Any]] = []
Expand All @@ -352,6 +353,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,
}
Expand Down Expand Up @@ -396,6 +398,7 @@ def load(cls, run_id: str, project_root: Path) -> RunState:
state.current_step_index = state_data.get("current_step_index", 0)
state.current_step_id = state_data.get("current_step_id")
state.step_results = state_data.get("step_results", {})
state.dry_run = state_data.get("dry_run", False)
state.created_at = state_data.get("created_at", "")
state.updated_at = state_data.get("updated_at", "")

Expand Down Expand Up @@ -478,6 +481,7 @@ def execute(
definition: WorkflowDefinition,
inputs: dict[str, Any] | None = None,
run_id: str | None = None,
dry_run: bool = False,
) -> RunState:
Comment thread
fuleinist marked this conversation as resolved.
"""Execute a workflow definition.
Comment thread
mnriem marked this conversation as resolved.

Expand All @@ -489,6 +493,16 @@ def execute(
User-provided input values.
run_id:
Optional run ID (uses SPECKIT_WORKFLOW_RUN_ID when set, otherwise auto-generated).
dry_run:
If ``True``, each step is executed normally but without
invoking the underlying AI integration (e.g. no CLI subprocess
is spawned for ``command`` steps, interactive gates return
``COMPLETED`` immediately, etc.). The workflow state is
still persisted to disk so ``specify workflow resume`` works,
and the dry-run flag is restored on resume so an interrupted
dry-run does not silently become a real run. Use this to
preview the resolved inputs and prompts for a workflow
without making any AI API calls.
Comment thread
mnriem marked this conversation as resolved.
Comment thread
mnriem marked this conversation as resolved.

Returns
-------
Expand Down Expand Up @@ -521,6 +535,7 @@ def execute(
# Resolve inputs
resolved_inputs = self._resolve_inputs(definition, inputs or {})
state.inputs = resolved_inputs
state.dry_run = dry_run
state.status = RunStatus.RUNNING
state.save()

Expand All @@ -531,6 +546,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
Expand Down Expand Up @@ -596,6 +612,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
Expand Down
78 changes: 72 additions & 6 deletions src/specify_cli/workflows/steps/command/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,12 +53,6 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
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,
Expand All @@ -67,11 +61,82 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
"input": resolved_input,
}

# Dry-run: show the rendered prompt without invoking the AI
if context.dry_run:
args_str = str(resolved_input.get("args", ""))
Comment thread
mnriem marked this conversation as resolved.
# Use the integration's own build_command_invocation() so the
# preview matches exactly what would be dispatched at runtime
invoke_str = f"{command} {args_str}".strip() if command else args_str
preview_note: str | None = None
if integration:
try:
from specify_cli.integrations import get_integration
impl = get_integration(integration)
if impl is None:
preview_note = (
f"(integration {integration!r} is not registered; using fallback invocation)"
)
else:
invoke_str = impl.build_command_invocation(command, args_str)
invoke_str = impl.build_command_invocation(command, args_str)
Comment thread
fuleinist marked this conversation as resolved.
Outdated
except (ImportError, AttributeError, KeyError, TypeError, ValueError) as exc:
Comment thread
mnriem marked this conversation as resolved.
Outdated
Comment thread
mnriem marked this conversation as resolved.
# ``build_command_invocation`` is optional in the
# integration protocol — fall back to ``<command> <args>``
# rather than swallowing the error silently. Record the
# reason so dry-run output makes the fallback explicit.
preview_note = (
f"(integration {integration!r} did not provide "
f"build_command_invocation: {type(exc).__name__}: {exc})"
)
output["dispatched"] = False
output["dry_run"] = True
# ``executed=False`` lets downstream branching/conditions
# distinguish a dry-run preview from a real successful run.
# ``exit_code`` is kept at 0 for backward compatibility
# (and because the step status is COMPLETED), but consumers
# that need to key on "did the integration actually run?"
# should check ``executed`` rather than ``exit_code``.
output["executed"] = False
output["exit_code"] = 0
output["stdout"] = ""
output["stderr"] = ""
output["invoke_command"] = invoke_str
Comment thread
fuleinist marked this conversation as resolved.
Comment thread
fuleinist marked this conversation as resolved.
message_body = (
f"[DRY RUN] Command: {invoke_str}\n"
f" Integration: {integration}\n"
f" Model: {model}\n"
Comment thread
fuleinist marked this conversation as resolved.
f" (AI invocation skipped — use without --dry-run to execute)"
)
if preview_note:
message_body += f"\n {preview_note}"
# Publish on both ``message`` (the original key the CLI's
# earlier dry-run loop read) and ``dry_run_message`` (the
# key GateStep adopted, which the CLI now prefers). Keeping
# the two in lockstep avoids special-casing per step type
# in the rendering loop.
output["message"] = message_body
Comment thread
fuleinist marked this conversation as resolved.
output["dry_run_message"] = message_body
return StepResult(
status=StepStatus.COMPLETED,
output=output,
)

# Attempt CLI dispatch
args_str = str(resolved_input.get("args", ""))
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
# Real run — executed=True so downstream templates can
# reliably key on `{{ steps.<id>.output.executed }}`.
# The dry-run branch sets executed=False; the no-dispatch
# branch below sets it False as well.
output["executed"] = True
if dispatch_result["exit_code"] != 0:
return StepResult(
status=StepStatus.FAILED,
Expand All @@ -85,6 +150,7 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
else:
output["exit_code"] = 1
output["dispatched"] = False
output["executed"] = False
return StepResult(
status=StepStatus.FAILED,
output=output,
Expand Down
64 changes: 63 additions & 1 deletion src/specify_cli/workflows/steps/gate/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

from __future__ import annotations

import collections.abc
import re
import sys
from pathlib import Path
Expand Down Expand Up @@ -40,7 +41,20 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
if isinstance(message, str) and "{{" in message:
message = evaluate_expression(message, context)

options = config.get("options", ["approve", "reject"])
# Normalize ``options`` defensively: workflows that bypass
# validation may set it to a non-sequence (string, dict, scalar).
# Without this guard, ``options[0]`` in the dry-run branch
# would index into a string (returning a single character) or
# raise on a dict. Accept any ``Sequence`` (list, tuple, etc.)
# other than ``str`` (which is itself a Sequence of chars but
# never a meaningful list of gate options).
raw_options = config.get("options", ["approve", "reject"])
if isinstance(raw_options, collections.abc.Sequence) and not isinstance(
raw_options, (str, bytes)
):
options: list[str] = [str(o) for o in raw_options if o is not None]
else:
options = []
Comment thread
fuleinist marked this conversation as resolved.
on_reject = config.get("on_reject", "abort")

show_file = config.get("show_file")
Expand All @@ -61,10 +75,58 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
"choice": None,
}

# Dry-run: skip interactive gates
if context.dry_run:
output["dry_run"] = True
# Pick a choice that won't unintentionally steer downstream
# branching. If the first option is a reject/abort sentinel
# (i.e. an option that would fail the gate when chosen for
# real), skip it; otherwise the first option is safe enough
# to preview. If no safe option exists, leave ``choice`` as
# ``None`` so downstream ``{{ steps.<id>.output.choice }}``
# expressions see a neutral value.
reject_sentinels = {"reject", "abort"}
safe_choice = next(
(opt for opt in options if opt.lower() not in reject_sentinels),
None,
)
output["choice"] = safe_choice
# Preserve the original ``message`` so downstream steps
# that reference ``{{ steps.<id>.output.message }}`` still
# see the prompt text. The DRY RUN preview is published
# on a separate ``dry_run_message`` field that the CLI
# rendering loop reads (with a fallback to ``message``
# for custom step types that have not adopted the new
# convention).
output["dry_run_message"] = (
f"[DRY RUN] Gate: {message}\n"
f" Options: {options}\n"
Comment thread
fuleinist marked this conversation as resolved.
f" (interactive prompt skipped — use without --dry-run to gate)"
)
Comment thread
fuleinist marked this conversation as resolved.
Comment thread
fuleinist marked this conversation as resolved.
return StepResult(
status=StepStatus.COMPLETED,
output=output,
)
Comment thread
fuleinist marked this conversation as resolved.
Comment on lines +78 to +109
Comment on lines +78 to +109
Comment thread
mnriem marked this conversation as resolved.
Comment thread
mnriem marked this conversation as resolved.

# Non-interactive: pause for later resume (the file is not read here)
if not sys.stdin.isatty():
return StepResult(status=StepStatus.PAUSED, output=output)

# Empty options would crash ``_prompt`` (it indexes the list to
# format ``Choose [1-N]`` and to pick the default on EOF). A
# workflow that bypassed validation and produced ``options=[]``
# is a clear authoring error — fail loudly here rather than
# masking it as an IndexError deep inside the prompt loop.
if not options:
return StepResult(
status=StepStatus.FAILED,
output=output,
error=(
f"Gate step {config.get('id', '?')!r} has no options; "
"interactive path cannot proceed."
),
)

# Interactive: prompt the user. ``show_file`` contents are folded
# into the displayed message so the operator can review the
# referenced material before choosing. Composing the prompt text
Expand Down
Loading
Loading