diff --git a/src/specify_cli/__init__.py b/src/specify_cli/__init__.py index 595b38f1e9..0932c63ba1 100644 --- a/src/specify_cli/__init__.py +++ b/src/specify_cli/__init__.py @@ -541,6 +541,130 @@ def version( app.add_typer(_self_app, name="self") +# ===== Spec / Plan Commands (direct CLI access with dry-run) ===== + +specify_app = typer.Typer( + name="specify", + help="Create a feature specification (direct CLI alternative to /speckit.specify in coding agents)", + add_completion=False, +) +app.add_typer(specify_app, name="specify") + + +@specify_app.command("spec") +def specify_specify( + spec: str = typer.Option( + ..., "--spec", "-s", help="Feature description (what to build and why)" + ), +): + """Create a feature specification from a description. + + This is a direct CLI alternative to the /speckit.specify agent command. + Runs the spec workflow and generates spec.md in the feature directory. + + Examples: + specify spec --spec "Build a kanban board with drag-and-drop" + specify spec --spec "Photo album app" + """ + from .workflows.engine import WorkflowEngine + + project_root = _require_specify_project() + engine = WorkflowEngine(project_root) + engine.on_step_start = lambda sid, label: console.print(f" \u25b8 [{sid}] {label} \u2026") + + try: + definition = engine.load_workflow("speckit") + except FileNotFoundError: + console.print("[red]Error:[/red] speckit workflow not installed. Run 'specify init' first.") + raise typer.Exit(1) + except ValueError as exc: + console.print(f"[red]Error:[/red] Invalid workflow: {exc}") + raise typer.Exit(1) + + errors = engine.validate(definition) + if errors: + console.print("[red]Workflow validation failed:[/red]") + for err in errors: + console.print(f" \u2022 {err}") + raise typer.Exit(1) + + inputs = {"spec": spec, "integration": "auto", "scope": "full"} + + console.print(f"\n[bold cyan]Running:[/bold cyan] specify spec") + console.print(f"[dim]Spec: {spec[:60]}{'...' if len(spec) > 60 else ''}[/dim]\n") + + + try: + state = engine.execute(definition, inputs) + except ValueError as exc: + console.print(f"[red]Error:[/red] {exc}") + raise typer.Exit(1) + except Exception as exc: + console.print(f"[red]Workflow failed:[/red] {exc}") + raise typer.Exit(1) + + status_color = {"completed": "green", "paused": "yellow", "failed": "red", "aborted": "red"}.get( + state.status.value, "white" + ) + + +@specify_app.command("plan") +def specify_plan( + spec: str = typer.Option( + ..., "--spec", "-s", help="Feature description (what to build and why)" + ), +): + """Create an implementation plan from a feature description. + + This is a direct CLI alternative to the /speckit.plan agent command. + Runs the plan step of the speckit workflow. + + Examples: + specify plan --spec "Build a kanban board with drag-and-drop" + specify plan --spec "Photo album app" + """ + from .workflows.engine import WorkflowEngine + + project_root = _require_specify_project() + engine = WorkflowEngine(project_root) + engine.on_step_start = lambda sid, label: console.print(f" \u25b8 [{sid}] {label} \u2026") + + try: + definition = engine.load_workflow("speckit") + except FileNotFoundError: + console.print("[red]Error:[/red] speckit workflow not installed. Run 'specify init' first.") + raise typer.Exit(1) + except ValueError as exc: + console.print(f"[red]Error:[/red] Invalid workflow: {exc}") + raise typer.Exit(1) + + errors = engine.validate(definition) + if errors: + console.print("[red]Workflow validation failed:[/red]") + for err in errors: + console.print(f" \u2022 {err}") + raise typer.Exit(1) + + inputs = {"spec": spec, "integration": "auto", "scope": "full"} + + console.print(f"\n[bold cyan]Running:[/bold cyan] specify plan") + console.print(f"[dim]Spec: {spec[:60]}{'...' if len(spec) > 60 else ''}[/dim]\n") + + + try: + state = engine.execute(definition, inputs) + except ValueError as exc: + console.print(f"[red]Error:[/red] {exc}") + raise typer.Exit(1) + except Exception as exc: + console.print(f"[red]Workflow failed:[/red] {exc}") + raise typer.Exit(1) + + status_color = {"completed": "green", "paused": "yellow", "failed": "red", "aborted": "red"}.get( + state.status.value, "white" + ) + + # ===== Extension Commands ===== extension_app = typer.Typer( @@ -4049,6 +4173,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" + ), ): """Run a workflow from an installed ID or local YAML path.""" from .workflows.engine import WorkflowEngine @@ -4087,8 +4214,11 @@ 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: + console.print("[bold yellow]DRY RUN — no AI invocation will occur[/bold yellow]\n") + try: - 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) diff --git a/src/specify_cli/commands/workflow.py b/src/specify_cli/commands/workflow.py index 3fa1f48ddf..470a0183f7 100644 --- a/src/specify_cli/commands/workflow.py +++ b/src/specify_cli/commands/workflow.py @@ -1,2 +1,193 @@ -"""specify workflow * commands — placeholder for future extraction.""" +"""specify workflow * commands — run, list, resume workflow executions.""" + from __future__ import annotations + +import typer + +from .._console import console +from .._utils import _display_project_path +from ..workflows.engine import WorkflowEngine + +workflow_app = typer.Typer( + name="workflow", + help="Manage and execute workflow runs", + add_completion=False, +) + + +def _require_specify_project() -> str: + """Return project root path if within a spec-kit project.""" + from pathlib import Path + + project_root = Path.cwd() + if (project_root / ".specify").is_dir(): + return str(project_root) + console.print("[red]Error:[/red] Not a spec-kit project (no .specify/ directory)") + console.print("Run this command from a spec-kit project root") + raise typer.Exit(1) + + +@workflow_app.command("run") +def workflow_run( + workflow_id: str = typer.Argument(..., help="Workflow ID or path to workflow YAML"), + input_spec: str = typer.Option( + None, "--spec", "-s", help="Workflow input as key=value pairs (repeatable)" + ), + dry_run: bool = typer.Option( + False, "--dry-run", help="Preview step execution without invoking AI (no commands spawned, gates return COMPLETED)" + ), + run_id: str = typer.Option(None, "--run-id", help="Custom run ID (auto-generated if not provided)"), +): + """Execute a workflow by ID or YAML path. + + Dry-run mode previews the resolved inputs, prompts, and commands that + would be executed — without spawning any AI integration CLI or making + AI API calls. Gates return COMPLETED immediately. Use this to verify + workflow inputs before a live run. + + Examples: + specify workflow run speckit --spec "spec=Build a kanban board" + specify workflow run speckit --spec "spec=Build a kanban board" --dry-run + specify workflow run ./my-workflow.yml --spec "input1=value1" --spec "input2=value2" + """ + project_root = _require_specify_project() + + # Parse inputs + inputs: dict[str, str] = {} + if input_spec: + for item in input_spec: + if "=" in item: + key, value = item.split("=", 1) + inputs[key.strip()] = value.strip() + + engine = WorkflowEngine(project_root) + + try: + definition = engine.load_workflow(workflow_id) + except FileNotFoundError: + console.print(f"[red]Error:[/red] Workflow not found: {workflow_id!r}") + console.print("Run 'specify workflow list' to see installed workflows") + raise typer.Exit(1) + except ValueError as exc: + console.print(f"[red]Error:[/red] Invalid workflow: {exc}") + raise typer.Exit(1) + + errors = engine.validate(definition) + if errors: + console.print("[red]Workflow validation failed:[/red]") + for err in errors: + console.print(f" • {err}") + raise typer.Exit(1) + + console.print(f"\n[bold cyan]Running workflow:[/bold cyan] {definition.name or workflow_id}") + console.print(f"[dim]Version: {definition.version}[/dim]") + console.print(f"[dim]ID: {definition.id}[/dim]") + if inputs: + console.print(f"[dim]Inputs: {inputs}[/dim]") + console.print() + + if dry_run: + console.print("[bold yellow]DRY RUN — no AI invocation will occur[/bold yellow]\n") + + try: + state = engine.execute( + definition, + inputs if inputs else None, + run_id=run_id if run_id else None, + dry_run=dry_run, + ) + except ValueError as exc: + console.print(f"[red]Error:[/red] {exc}") + raise typer.Exit(1) + except Exception as exc: + console.print(f"[red]Workflow failed:[/red] {exc}") + raise typer.Exit(1) + + if dry_run: + status_color = "yellow" + console.print(f"\n[{status_color}]Status: {state.status.value}[/{status_color}] (dry-run)") + console.print("[dim]Run without --dry-run to execute for real.[/dim]") + else: + status_color_map = { + "completed": "green", + "paused": "yellow", + "failed": "red", + "aborted": "red", + } + status_color = status_color_map.get(state.status.value, "white") + console.print(f"\n[{status_color}]Status: {state.status.value}[/{status_color}]") + if state.status.value == "paused": + console.print(f"[dim]Run ID: {state.run_id}[/dim]") + console.print("[dim]Resume with: specify workflow resume {run_id}[/dim]") + + +@workflow_app.command("list") +def workflow_list(): + """List all workflow runs for this project.""" + project_root = _require_specify_project() + + engine = WorkflowEngine(project_root) + runs = engine.list_runs() + + if not runs: + console.print("[yellow]No workflow runs found.[/yellow]") + console.print("Run a workflow with: specify workflow run ") + return + + from rich.table import Table + + table = Table(title="Workflow Runs") + table.add_column("Run ID", style="cyan") + table.add_column("Workflow ID", style="cyan") + table.add_column("Status", style="bold") + table.add_column("Started", style="dim") + table.add_column("Last Updated", style="dim") + + for run in sorted(runs, key=lambda r: r.get("updated_at", ""), reverse=True): + status = run.get("status", "?") + status_color = {"completed": "green", "paused": "yellow", "failed": "red", "aborted": "red"}.get( + status, "white" + ) + table.add_row( + run.get("run_id", "?"), + run.get("workflow_id", "?"), + f"[{status_color}]{status}[/{status_color}]", + run.get("created_at", "?")[:19] if run.get("created_at") else "?", + run.get("updated_at", "?")[:19] if run.get("updated_at") else "?", + ) + + console.print(table) + console.print(f"\n[dim]{len(runs)} run(s) found[/dim]") + + +@workflow_app.command("resume") +def workflow_resume( + run_id: str = typer.Argument(..., help="Run ID to resume"), +): + """Resume a paused or failed workflow run.""" + project_root = _require_specify_project() + + engine = WorkflowEngine(project_root) + + try: + state = engine.resume(run_id) + except FileNotFoundError: + console.print(f"[red]Error:[/red] Run not found: {run_id!r}") + console.print("Run 'specify workflow list' to see available runs") + raise typer.Exit(1) + except ValueError as exc: + console.print(f"[red]Error:[/red] {exc}") + raise typer.Exit(1) + except Exception as exc: + console.print(f"[red]Resume failed:[/red] {exc}") + raise typer.Exit(1) + + status_color_map = { + "completed": "green", + "paused": "yellow", + "failed": "red", + "aborted": "red", + } + status_color = status_color_map.get(state.status.value, "white") + console.print(f"\n[{status_color}]Status: {state.status.value}[/{status_color}]") + console.print(f"[dim]Run ID: {state.run_id}[/dim]") \ No newline at end of file diff --git a/src/specify_cli/workflows/base.py b/src/specify_cli/workflows/base.py index b144ca903d..885055e5f0 100644 --- a/src/specify_cli/workflows/base.py +++ b/src/specify_cli/workflows/base.py @@ -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: diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py index 65d9b3a272..3d9b5b30cf 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, + dry_run: bool = False, ) -> RunState: """Execute a workflow definition. @@ -426,6 +427,14 @@ def execute( User-provided input values. run_id: Optional run ID (auto-generated if not provided). + 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. + Use this to preview the resolved inputs and prompts for a + workflow without making any AI API calls. Returns ------- @@ -462,6 +471,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 diff --git a/src/specify_cli/workflows/steps/command/__init__.py b/src/specify_cli/workflows/steps/command/__init__.py index 21fd4837d1..e4fee4188f 100644 --- a/src/specify_cli/workflows/steps/command/__init__.py +++ b/src/specify_cli/workflows/steps/command/__init__.py @@ -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, @@ -67,6 +61,34 @@ 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", "")) + # Reconstruct what the integration would build for the invocation + invoke_str = f"{command} {args_str}".strip() + output["dispatched"] = False + output["dry_run"] = True + output["exit_code"] = 0 + output["stdout"] = "" + output["stderr"] = "" + output["invoke_command"] = invoke_str + output["message"] = ( + f"[DRY RUN] Command: {invoke_str}\n" + f" Integration: {integration}\n" + f" Model: {model}\n" + f" (AI invocation skipped — use without --dry-run to execute)" + ) + 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"] diff --git a/src/specify_cli/workflows/steps/gate/__init__.py b/src/specify_cli/workflows/steps/gate/__init__.py index d4d32d763c..4529b58836 100644 --- a/src/specify_cli/workflows/steps/gate/__init__.py +++ b/src/specify_cli/workflows/steps/gate/__init__.py @@ -43,6 +43,13 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: "choice": None, } + # Dry-run: skip interactive gates + if context.dry_run: + 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 114e6b02cd..a4221305a2 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -641,6 +641,34 @@ def test_dispatch_failure_returns_failed_status(self, tmp_path): assert result.output["dispatched"] is True assert result.output["exit_code"] == 1 + def test_dry_run_returns_completed_without_dispatch(self): + """Dry-run mode: step returns COMPLETED and shows rendered prompt without invoking CLI.""" + from specify_cli.workflows.steps.command import CommandStep + from specify_cli.workflows.base import StepContext, StepStatus + + step = CommandStep() + ctx = StepContext( + inputs={"name": "login"}, + default_integration="claude", + project_root="/tmp", + dry_run=True, + ) + config = { + "id": "test", + "command": "speckit.specify", + "input": {"args": "{{ inputs.name }}"}, + } + result = step.execute(config, ctx) + + assert result.status == StepStatus.COMPLETED + assert result.output["dry_run"] is True + assert result.output["dispatched"] is False + assert result.output["command"] == "speckit.specify" + assert result.output["input"]["args"] == "login" + # No AI call was made (shutil.which would fail anyway, but dry_run short-circuits before that) + assert "invoke_command" in result.output + assert "DRY RUN" in result.output["message"] + class TestPromptStep: """Test the prompt step type.""" @@ -818,6 +846,23 @@ def test_validate_invalid_on_reject(self): }) assert any("on_reject" in e for e in errors) + def test_dry_run_skips_interactive_gate(self): + """Dry-run mode: gate step returns COMPLETED without pausing for user input.""" + from specify_cli.workflows.steps.gate import GateStep + from specify_cli.workflows.base import StepContext, StepStatus + + step = GateStep() + ctx = StepContext(dry_run=True) + config = { + "id": "review", + "message": "Review the spec.", + "options": ["approve", "reject"], + "on_reject": "abort", + } + result = step.execute(config, ctx) + assert result.status == StepStatus.COMPLETED + assert result.output["message"] == "Review the spec." + class TestIfThenStep: """Test the if/then/else step type.""" @@ -1459,6 +1504,49 @@ def test_execute_with_gate_pauses(self, project_dir): assert "gate" in state.step_results assert state.step_results["gate"]["status"] == "paused" + def test_execute_dry_run(self, project_dir): + """Dry-run: engine returns COMPLETED without invoking the AI for command steps.""" + from unittest.mock import patch + from specify_cli.workflows.engine import WorkflowEngine, WorkflowDefinition + from specify_cli.workflows.base import RunStatus + + yaml_str = """ +schema_version: "1.0" +workflow: + id: "dryrun-test" + name: "Dry Run Test" + version: "1.0.0" +inputs: + spec: + type: string + default: "test spec" +steps: + - id: specify + command: speckit.specify + input: + args: "{{ inputs.spec }}" + - id: plan + command: speckit.plan + input: + args: "{{ inputs.spec }}" +""" + definition = WorkflowDefinition.from_string(yaml_str) + engine = WorkflowEngine(project_dir) + + # In dry-run mode, even if the CLI tool is installed, no AI calls are made + with patch("specify_cli.workflows.steps.command.shutil.which", return_value="/usr/local/bin/claude"), \ + patch("subprocess.run") as mock_run: + state = engine.execute(definition, {"spec": "login feature"}, dry_run=True) + + assert state.status == RunStatus.COMPLETED + assert "specify" in state.step_results + assert state.step_results["specify"]["output"]["dry_run"] is True + assert state.step_results["specify"]["output"]["dispatched"] is False + assert "plan" in state.step_results + assert state.step_results["plan"]["output"]["dry_run"] is True + # subprocess.run should NOT have been called in dry-run mode + assert mock_run.call_count == 0 + def test_execute_with_shell_step(self, project_dir): from specify_cli.workflows.engine import WorkflowEngine, WorkflowDefinition from specify_cli.workflows.base import RunStatus