diff --git a/src/specify_cli/workflows/__init__.py b/src/specify_cli/workflows/__init__.py index 13782f620b..5847ed00c5 100644 --- a/src/specify_cli/workflows/__init__.py +++ b/src/specify_cli/workflows/__init__.py @@ -48,6 +48,7 @@ def _register_builtin_steps() -> None: from .steps.fan_out import FanOutStep from .steps.gate import GateStep from .steps.if_then import IfThenStep + from .steps.init import InitStep from .steps.prompt import PromptStep from .steps.shell import ShellStep from .steps.switch import SwitchStep @@ -59,6 +60,7 @@ def _register_builtin_steps() -> None: _register_step(FanOutStep()) _register_step(GateStep()) _register_step(IfThenStep()) + _register_step(InitStep()) _register_step(PromptStep()) _register_step(ShellStep()) _register_step(SwitchStep()) diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py index d24bc29501..0d56f7df70 100644 --- a/src/specify_cli/workflows/engine.py +++ b/src/specify_cli/workflows/engine.py @@ -94,7 +94,7 @@ def _get_valid_step_types() -> set[str]: if STEP_REGISTRY: return set(STEP_REGISTRY.keys()) return { - "command", "shell", "prompt", "gate", "if", + "command", "shell", "prompt", "gate", "if", "init", "switch", "while", "do-while", "fan-out", "fan-in", } diff --git a/src/specify_cli/workflows/steps/init/__init__.py b/src/specify_cli/workflows/steps/init/__init__.py new file mode 100644 index 0000000000..ce326976c3 --- /dev/null +++ b/src/specify_cli/workflows/steps/init/__init__.py @@ -0,0 +1,265 @@ +"""Init step — bootstrap a Spec Kit project from within a workflow. + +Runs the same scaffolding as ``specify init`` so a workflow can create +(or merge into) a project before driving the rest of the spec-driven +process. The step invokes the ``init`` command in-process and captures +its exit code and output. +""" + +from __future__ import annotations + +import os +from typing import Any + +from specify_cli._agent_config import DEFAULT_INIT_INTEGRATION +from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus +from specify_cli.workflows.expressions import evaluate_expression + +#: Valid ``script`` values, mirroring ``specify init --script``. +VALID_SCRIPT_TYPES = ("sh", "ps") + +#: Directories the workflow engine may create before steps run. +#: These are excluded from the "non-empty directory" fast-fail check so +#: that ``here: true`` works without requiring ``force: true`` when the +#: only pre-existing content is engine run-state. +_ENGINE_OWNED_DIRS = {".specify"} + + +class InitStep(StepBase): + """Bootstrap a project, equivalent to running ``specify init``. + + The step runs the bundled ``specify init`` command non-interactively, + scaffolding templates, scripts, shared infrastructure, and the + selected coding agent integration into the target directory. + + Because workflows run unattended, the step defaults to + ``--ignore-agent-tools`` (skip checks for an installed agent CLI) and + resolves the integration from the step config, falling back to the + workflow-level default integration. + + Example YAML:: + + - id: bootstrap + type: init + here: true + integration: copilot + script: sh + + Supported config fields (all optional): + + ``project`` + Project name or path to create. Use ``"."`` for the current + directory. Ignored when ``here`` is truthy. + ``here`` + Initialize in the target directory instead of creating a new one. + ``integration`` + Integration key (e.g. ``copilot``). Defaults to the workflow's + default integration, then to ``DEFAULT_INIT_INTEGRATION``. + ``integration_options`` + Extra options for the integration (e.g. ``"--skills"`` or + ``"--commands-dir .myagent/cmds"``). + ``script`` + Script type, ``sh`` or ``ps``. + ``force`` + Merge/overwrite without confirmation when the directory is not + empty. + ``ignore_agent_tools`` + Skip checks for the coding agent CLI (defaults to ``true``). + ``preset`` + Preset ID to install during initialization. + """ + + type_key = "init" + + def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: + project = self._resolve(config.get("project"), context) + here = self._resolve_bool(config.get("here"), context) + + integration = config.get("integration") or context.default_integration + integration = self._resolve(integration, context) + # Apply the same default that specify init uses in non-interactive mode + # so that output.integration reflects the actual integration used. + if not integration: + integration = DEFAULT_INIT_INTEGRATION + + integration_options = self._resolve( + config.get("integration_options"), context + ) + script = self._resolve(config.get("script"), context) + preset = self._resolve(config.get("preset"), context) + + force = self._resolve_bool(config.get("force"), context) + # Workflows run unattended; skip the agent CLI presence check by default. + ignore_agent_tools = self._resolve_bool( + config.get("ignore_agent_tools", True), context + ) + + argv: list[str] = ["init"] + if here: + argv.append("--here") + elif project: + argv.append(str(project)) + else: + # No explicit target → initialize the current directory. + argv.append(".") + + # When the target is the current directory and ``force`` is not set, + # ``specify init`` prompts for confirmation if the directory is not + # empty. Workflows run unattended (no stdin), so the prompt would + # abort with a confusing error. Fail fast with an actionable message. + # Exception: if the only pre-existing content is engine-owned (e.g. + # .specify/workflows/runs/), treat it as implicitly empty and auto-add + # --force so init can proceed unattended. + targets_current_dir = here or not project or str(project) == "." + if targets_current_dir and not force: + base = context.project_root or os.getcwd() + try: + with os.scandir(base) as it: + non_engine_entries = [ + entry for entry in it + if entry.name not in _ENGINE_OWNED_DIRS + ] + except OSError: + non_engine_entries = [] + if non_engine_entries: + error_message = ( + f"Target directory {base!r} is not empty. Set " + "'force: true' to merge into a non-empty directory." + ) + return StepResult( + status=StepStatus.FAILED, + output={ + "argv": argv, + "project": project, + "here": here, + "integration": integration, + "integration_options": integration_options, + "script": script, + "exit_code": 1, + "stdout": "", + "stderr": error_message, + }, + error=error_message, + ) + else: + # Only engine-owned dirs exist — implicitly force so specify + # init doesn't prompt about the non-empty directory. + force = True + + if integration: + argv.extend(["--integration", str(integration)]) + if integration_options: + argv.extend(["--integration-options", str(integration_options)]) + if script: + argv.extend(["--script", str(script)]) + if preset: + argv.extend(["--preset", str(preset)]) + if force: + argv.append("--force") + if ignore_agent_tools: + argv.append("--ignore-agent-tools") + + exit_code, stdout, stderr = self._run_init(argv, context) + + output: dict[str, Any] = { + "argv": argv, + "project": project, + "here": here, + "integration": integration, + "integration_options": integration_options, + "script": script, + "exit_code": exit_code, + "stdout": stdout, + "stderr": stderr, + } + + if exit_code != 0: + return StepResult( + status=StepStatus.FAILED, + output=output, + error=( + stderr.strip() + or stdout.strip() + or f"specify init exited with code {exit_code}." + ), + ) + return StepResult(status=StepStatus.COMPLETED, output=output) + + @staticmethod + def _resolve(value: Any, context: StepContext) -> Any: + """Resolve ``{{ ... }}`` expressions in string config values.""" + if isinstance(value, str) and "{{" in value: + return evaluate_expression(value, context) + return value + + @classmethod + def _resolve_bool(cls, value: Any, context: StepContext) -> bool: + """Coerce a config value (possibly an expression) to a boolean.""" + resolved = cls._resolve(value, context) + if isinstance(resolved, str): + return resolved.strip().lower() in ("true", "1", "yes") + return bool(resolved) + + @staticmethod + def _run_init( + argv: list[str], context: StepContext + ) -> tuple[int, str, str]: + """Invoke ``specify init`` in-process and capture exit code/output. + + Runs with the working directory set to ``context.project_root`` so + that ``--here`` and relative project paths target the right place. + """ + from typer.testing import CliRunner + + from specify_cli import app + + runner = CliRunner() + + prev_cwd = os.getcwd() + if context.project_root: + try: + os.chdir(context.project_root) + except OSError as exc: + return (1, "", f"Cannot enter project root: {exc}") + try: + result = runner.invoke(app, argv, catch_exceptions=True) + finally: + try: + os.chdir(prev_cwd) + except OSError: + # Best-effort cleanup: avoid masking the init command result + # if restoring the previous working directory fails. + pass + + stdout = result.output or "" + # click >= 8.2 captures stderr separately; older versions mix it into + # stdout and raise when ``result.stderr`` is accessed. + try: + stderr = result.stderr or "" + except (ValueError, AttributeError): + stderr = "" + + if result.exit_code != 0 and result.exception is not None: + detail = f"{type(result.exception).__name__}: {result.exception}" + stderr = f"{stderr}\n{detail}".strip() if stderr else detail + + return (result.exit_code, stdout, stderr) + + def validate(self, config: dict[str, Any]) -> list[str]: + errors = super().validate(config) + script = config.get("script") + if script is not None and not isinstance(script, str): + errors.append( + f"Init step {config.get('id', '?')!r}: 'script' must be a string " + f"({' or '.join(repr(s) for s in VALID_SCRIPT_TYPES)})." + ) + elif ( + isinstance(script, str) + and "{{" not in script + and script not in VALID_SCRIPT_TYPES + ): + errors.append( + f"Init step {config.get('id', '?')!r}: 'script' must be " + f"{' or '.join(repr(s) for s in VALID_SCRIPT_TYPES)}." + ) + return errors diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 51da5cc86b..f0ff557b1e 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -101,7 +101,7 @@ def test_all_step_types_registered(self): expected = { "command", "shell", "prompt", "gate", "if", "switch", - "while", "do-while", "fan-out", "fan-in", + "while", "do-while", "fan-out", "fan-in", "init", } assert expected.issubset(set(STEP_REGISTRY.keys())) @@ -979,6 +979,171 @@ def _force_gate_stdin(monkeypatch, *, tty: bool): monkeypatch.setattr(gate_module, "sys", _FakeSys(tty=tty)) +class TestInitStep: + """Test the init step type.""" + + def test_builds_here_argv_and_bootstraps(self, tmp_path): + from specify_cli.workflows.steps.init import InitStep + from specify_cli.workflows.base import StepContext, StepStatus + + step = InitStep() + ctx = StepContext( + project_root=str(tmp_path), default_integration="copilot" + ) + config = {"id": "bootstrap", "here": True, "script": "sh"} + result = step.execute(config, ctx) + + assert result.status == StepStatus.COMPLETED + assert result.output["exit_code"] == 0 + argv = result.output["argv"] + assert argv[0] == "init" + assert "--here" in argv + assert "--integration" in argv and "copilot" in argv + assert "--ignore-agent-tools" in argv + assert (tmp_path / ".specify").is_dir() + + def test_default_integration_falls_back_to_workflow_default(self, tmp_path): + from specify_cli.workflows.steps.init import InitStep + from specify_cli.workflows.base import StepContext, StepStatus + + step = InitStep() + ctx = StepContext( + project_root=str(tmp_path), default_integration="copilot" + ) + result = step.execute( + {"id": "bootstrap", "here": True, "script": "sh"}, ctx + ) + assert result.status == StepStatus.COMPLETED + assert result.output["integration"] == "copilot" + + def test_project_name_creates_subdirectory(self, tmp_path): + from specify_cli.workflows.steps.init import InitStep + from specify_cli.workflows.base import StepContext, StepStatus + + step = InitStep() + ctx = StepContext( + project_root=str(tmp_path), default_integration="copilot" + ) + result = step.execute( + { + "id": "bootstrap", + "project": "demo", + "script": "sh", + }, + ctx, + ) + assert result.status == StepStatus.COMPLETED + assert (tmp_path / "demo" / ".specify").is_dir() + + def test_invalid_integration_fails(self, tmp_path): + from specify_cli.workflows.steps.init import InitStep + from specify_cli.workflows.base import StepContext, StepStatus + + step = InitStep() + ctx = StepContext(project_root=str(tmp_path)) + result = step.execute( + { + "id": "bootstrap", + "here": True, + "integration": "no-such-agent", + "script": "sh", + }, + ctx, + ) + assert result.status == StepStatus.FAILED + assert result.output["exit_code"] != 0 + assert result.error is not None + + def test_non_empty_current_dir_without_force_fails_fast(self, tmp_path): + from specify_cli.workflows.steps.init import InitStep + from specify_cli.workflows.base import StepContext, StepStatus + + (tmp_path / "existing.txt").write_text("data") + + step = InitStep() + ctx = StepContext( + project_root=str(tmp_path), default_integration="copilot" + ) + result = step.execute( + {"id": "bootstrap", "here": True, "script": "sh"}, + ctx, + ) + assert result.status == StepStatus.FAILED + assert "force: true" in (result.error or "") + assert not (tmp_path / ".specify").exists() + + def test_engine_owned_dirs_do_not_trigger_non_empty_check(self, tmp_path): + from specify_cli.workflows.steps.init import InitStep + from specify_cli.workflows.base import StepContext, StepStatus + + # Simulate the engine creating its run-state directory before steps run + (tmp_path / ".specify" / "workflows" / "runs" / "abc123").mkdir( + parents=True + ) + + step = InitStep() + ctx = StepContext( + project_root=str(tmp_path), default_integration="copilot" + ) + result = step.execute( + {"id": "bootstrap", "here": True, "script": "sh"}, + ctx, + ) + assert result.status == StepStatus.COMPLETED + # Verify --force was implicitly added + assert "--force" in result.output["argv"] + + def test_default_integration_when_none_provided(self, tmp_path): + from specify_cli.workflows.steps.init import InitStep + from specify_cli.workflows.base import StepContext, StepStatus + + step = InitStep() + # No default_integration on context either + ctx = StepContext(project_root=str(tmp_path)) + result = step.execute( + {"id": "bootstrap", "here": True, "script": "sh"}, + ctx, + ) + assert result.status == StepStatus.COMPLETED + assert result.output["integration"] == "copilot" + + def test_integration_options_passed_through(self, tmp_path): + from specify_cli.workflows.steps.init import InitStep + from specify_cli.workflows.base import StepContext, StepStatus + + step = InitStep() + ctx = StepContext( + project_root=str(tmp_path), default_integration="copilot" + ) + result = step.execute( + { + "id": "bootstrap", + "here": True, + "script": "sh", + "integration": "copilot", + "integration_options": "--skills", + }, + ctx, + ) + assert result.status == StepStatus.COMPLETED + assert "--integration-options" in result.output["argv"] + assert "--skills" in result.output["argv"] + assert result.output["integration_options"] == "--skills" + + def test_validate_rejects_bad_script(self): + from specify_cli.workflows.steps.init import InitStep + + step = InitStep() + errors = step.validate({"id": "bootstrap", "script": "bogus"}) + assert any("'script' must be 'sh' or 'ps'" in e for e in errors) + + def test_validate_accepts_valid(self): + from specify_cli.workflows.steps.init import InitStep + + step = InitStep() + assert step.validate({"id": "bootstrap", "script": "sh"}) == [] + + class TestGateStep: """Test the gate step type.""" diff --git a/workflows/ARCHITECTURE.md b/workflows/ARCHITECTURE.md index 892333473c..113545f334 100644 --- a/workflows/ARCHITECTURE.md +++ b/workflows/ARCHITECTURE.md @@ -77,13 +77,14 @@ When a `gate` step pauses execution, the engine persists `current_step_index` an ## Step Types -The engine ships with 10 built-in step types, each in its own subpackage under `src/specify_cli/workflows/steps/`: +The engine ships with 11 built-in step types, each in its own subpackage under `src/specify_cli/workflows/steps/`: | Type Key | Class | Purpose | Returns `next_steps`? | |----------|-------|---------|-----------------------| | `command` | `CommandStep` | Invoke an installed Spec Kit command via integration CLI | No | | `prompt` | `PromptStep` | Send an arbitrary inline prompt to integration CLI | No | | `shell` | `ShellStep` | Run a shell command, capture output | No | +| `init` | `InitStep` | Bootstrap a project (equivalent to `specify init`) | No | | `gate` | `GateStep` | Interactive human review/approval | No (pauses in CI) | | `if` | `IfThenStep` | Conditional branching (then/else) | Yes | | `switch` | `SwitchStep` | Multi-branch dispatch on expression | Yes | @@ -197,6 +198,7 @@ src/specify_cli/ │ └── steps/ │ ├── command/ # Dispatch command to AI integration │ ├── shell/ # Run shell command +│ ├── init/ # Bootstrap a project (specify init) │ ├── gate/ # Human review checkpoint │ ├── if_then/ # Conditional branching │ ├── prompt/ # Arbitrary inline prompts diff --git a/workflows/README.md b/workflows/README.md index 0e3e74a924..d77a7cddd1 100644 --- a/workflows/README.md +++ b/workflows/README.md @@ -78,7 +78,7 @@ specify workflow run speckit \ ## Step Types -Workflows support 10 built-in step types: +Workflows support 11 built-in step types: ### Command Steps (default) @@ -114,6 +114,24 @@ Run a shell command and capture output: run: "cd {{ inputs.project_dir }} && npm test" ``` +### Init Steps + +Bootstrap a project the same way `specify init` does — scaffolding +templates, scripts, shared infrastructure, and the selected coding agent +integration. Runs non-interactively (defaults to `--ignore-agent-tools`) +and resolves the integration from the step config or the workflow default: + +```yaml +- id: bootstrap + type: init + here: true # or: project: my-project + integration: copilot # Optional: defaults to workflow integration + integration_options: "--skills" # Optional: extra options for the integration + script: sh # Optional: sh or ps + force: true # Optional: required to merge into a non-empty directory + preset: healthcare-compliance # Optional preset ID +``` + ### Gate Steps Pause for human review. The workflow resumes when `specify workflow resume` is called: