feat(workflows): compose installed workflows via a scoped workflow step

docs/reference/overview.md

@@ -34,7 +34,7 @@ Artifacts are the commands, templates, and scripts a project exposes, whichever
 
 ## Workflows
 
-Workflows automate multi-step Spec-Driven Development processes into repeatable sequences. They chain commands, prompts, shell steps, and human checkpoints together, with support for conditional logic, loops, fan-out/fan-in, and the ability to pause and resume from the exact point of interruption.
+Workflows automate multi-step Spec-Driven Development processes into repeatable sequences. They chain commands, prompts, shell steps, and human checkpoints together, with support for conditional logic, loops, fan-out/fan-in, workflow composition (running another installed workflow as a scoped subtree via a `type: workflow` step), and the ability to pause and resume from the exact point of interruption.
 
 [Workflows reference →](workflows.md)
 

docs/reference/workflows.md

@@ -1,6 +1,6 @@
 # Workflows
 
-Workflows automate multi-step Spec-Driven Development processes — chaining commands, prompts, shell steps, and human checkpoints into repeatable sequences. They support conditional logic, loops, fan-out/fan-in, and can be paused and resumed from the exact point of interruption.
+Workflows automate multi-step Spec-Driven Development processes — chaining commands, prompts, shell steps, and human checkpoints into repeatable sequences. They support conditional logic, loops, fan-out/fan-in, composition (running another installed workflow as a scoped subtree), and can be paused and resumed from the exact point of interruption.
 
 ## Run a Workflow
 
@@ -539,9 +539,121 @@ specify workflow run speckit -i spec="Build a kanban board with drag-and-drop ta
 | `do-while`   | Execute at least once, then loop on condition    |
 | `fan-out`    | Dispatch a step for each item in a list          |
 | `fan-in`     | Aggregate results from a fan-out step            |
+| `workflow`   | Run an installed workflow as a scoped subtree    |
 
 > **Security note:** a `shell` step runs a local command with **your** privileges. There is no capability sandbox — `requires` is an advisory pre-condition block (spec-kit version, integrations), not a runtime gate, so it does **not** restrict what a step can do. In particular there is no `requires.permissions` capability gate: it is rejected by validation precisely because it would imply a sandbox that does not exist. Review any catalog or downloaded workflow before running it, and use a `gate` step to require explicit approval before sensitive or destructive shell commands.
 
+### Workflow composition (`type: workflow`)
+
+A `workflow` step runs an installed workflow as a **scoped subtree of the
+current run** — there is one run, one run directory, and one process. The
+included workflow behaves like a function call: values cross the boundary only
+through its declared `inputs` and `outputs`.
+
+```yaml
+inputs:
+  target:
+    type: string
+    required: true
+  report:
+    type: string
+    default: ""
+  slug:
+    type: string
+    default: ""
+
+steps:
+  - id: run-selected
+    type: workflow
+    workflow: "{{ inputs.target }}"
+    input:
+      report: "{{ inputs.report }}"
+      slug: "{{ inputs.slug }}"
+```
+
+| Field      | Required | Description |
+| ---------- | -------- | ----------- |
+| `workflow` | yes      | Installed workflow ID, or an expression evaluated in the caller's scope. The resolved value must be a valid ID of a registered, installed, and enabled workflow. Literal IDs are validated at definition time. |
+| `input`    | no       | Mapping of the target's declared input names to values evaluated in the caller's scope. An undeclared name is rejected. Defaults, required, type, and enum rules apply. |
+
+The `workflow` expression must resolve to a value the engine captures: a
+declared workflow input or a preceding step's captured output (for example a
+`shell` step's `stdout`) both work. A `prompt` step streams its output to the
+agent and returns an empty `stdout`, so it cannot drive `workflow` selection.
+The resolved value must exactly match the installed workflow ID; it is not
+trimmed, case-normalized, or otherwise transformed. A shell-driven selection
+must therefore emit only the ID, for example `printf child` rather than
+`echo child` (which includes a trailing newline).
+
+`type: workflow` is an engine facility (like `fan-out`), not a custom-step API.
+The engine owns the nested scope tree; custom steps still receive only a
+`StepContext`.
+
+#### Scope isolation
+
+The included workflow receives a separate expression scope:
+
+- `inputs` contains only the resolved, declared, and validated mapped inputs.
+- `steps` contains only the included workflow's own step results.
+- Caller inputs and caller step results are **not** visible unless explicitly
+  passed through the `input` mapping.
+- Project root, integration defaults, and the run ID remain available as
+  execution infrastructure.
+
+#### Declared outputs
+
+An included workflow exposes values back to its caller only through a top-level
+`outputs` block. Each entry requires a `value` expression evaluated in the
+included workflow's local scope once it completes:
+
+```yaml
+outputs:
+  result:
+    value: "{{ steps.fix.output.stdout }}"
+  tested:
+    value: "{{ steps.test.output.exit_code == 0 }}"
+```
+
+The caller reads them from the workflow step's output:
+
+```yaml
+"{{ steps.run-selected.output.result }}"
+```
+
+Output names must be safe lowercase identifiers and cannot use the reserved
+names `workflow`, `status`, `error`, `aborted`, `integration`, `model`,
+`options`, or `input`. A whole expression preserves its resolved type;
+interpolation mixed with text produces a string. Paused, failed, and aborted
+scopes do not evaluate outputs.
+
+#### Lifecycle and failure handling
+
+The workflow step reports the aggregate outcome of its subtree: all required
+steps complete → `completed`; an included step pauses → the run pauses; an
+included failure (unhandled) → the run fails; an included gate abort → the run
+aborts (`output.aborted: true`). `continue_on_error: true` on the workflow step
+lets the caller continue past an otherwise unhandled included failure; it never
+overrides an abort or bypasses a pause.
+
+#### Resume and composition limits
+
+The resolved target, its composed definition snapshot (including overlays), and
+the validated inputs are persisted with the run. The snapshot is written as an
+immutable YAML file under the run's `snapshots/` directory (so YAML-native
+values such as unquoted dates round-trip, unlike JSON), while `state.json`
+records only its reference. On resume the engine reuses the snapshot and
+resumes at the included scope's local step index; it does not re-resolve the
+target. Editing an installed workflow affects new invocations, not a scope
+already bound within a persisted run. `workflow resume --input` updates the
+**root** workflow's inputs; a composing workflow forwards them by mapping them
+into the child's declared inputs.
+
+Nested composition is allowed: `A -> B -> D` and `A -> C -> D` are legal
+diamonds. A workflow may not appear twice on the active path — `A -> B -> A` is
+a cycle and is rejected, so bounded recursion is not supported. Composition is
+limited to 16 included levels; the root is depth 0 and entering depth 17 is
+rejected.
+
 ### Per-Step Integration Configuration
 
 Command steps may pass structured runtime configuration to integrations that
@@ -651,6 +763,7 @@ Each workflow run persists its state at `.specify/workflows/runs/<run_id>/`:
 - `state.json` — current run state and step progress
 - `inputs.json` — resolved input values
 - `log.jsonl` — step-by-step execution log
+- `snapshots/*.yml` — immutable composed-child definition snapshots
 
 This enables `specify workflow resume` to continue from the exact step where a run was paused (e.g., at a gate) or failed.
 

src/specify_cli/workflows/__init__.py

@@ -56,6 +56,7 @@ def _register_builtin_steps() -> None:
     from .step.slot import SlotStep
     from .step.switch import SwitchStep
     from .step.while_loop import WhileStep
+    from .step.workflow import WorkflowStep
 
     _register_step(CommandStep())
     _register_step(DoWhileStep())
@@ -69,6 +70,7 @@ def _register_builtin_steps() -> None:
     _register_step(SlotStep())
     _register_step(SwitchStep())
     _register_step(WhileStep())
+    _register_step(WorkflowStep())
 
 
 _register_builtin_steps()

src/specify_cli/workflows/_command_run_ownership.py

@@ -6,6 +6,7 @@
 from pathlib import Path
 
 from . import _commands as cli
+from .engine import is_valid_workflow_id
 
 
 def _same_existing_path(left: Path, right: Path) -> bool:
@@ -125,7 +126,7 @@ def ownership_for(candidate: Path) -> tuple[Path, str] | None:
             if (
                 not isinstance(workflow_id, str)
                 or workflow_id in cli._RESERVED_WORKFLOW_IDS
-                or not cli._WORKFLOW_ID_PATTERN.fullmatch(workflow_id)
+                or not is_valid_workflow_id(workflow_id)
             ):
                 continue
             try:

src/specify_cli/workflows/_commands.py

@@ -30,6 +30,7 @@
 )
 from .._project import _resolve_init_dir_override as _resolve_init_dir_override
 from ..shared_infra import verify_archive_sha256
+from .engine import is_valid_workflow_id
 
 workflow_app = typer.Typer(
     name="workflow",
@@ -154,7 +155,6 @@ def _resolve_installed_workflow_ownership(
     return resolver(source_path, err)
 
 
-_WORKFLOW_ID_PATTERN = re.compile(r"^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$")
 _RESERVED_WORKFLOW_IDS: frozenset[str] = frozenset({"overlays", "runs", "steps"})
 
 
@@ -263,7 +263,7 @@ def _validate_workflow_id_or_exit(workflow_id: str) -> None:
     """Validate that ``workflow_id`` is a safe installed-workflow directory name."""
     if (
         workflow_id in _RESERVED_WORKFLOW_IDS
-        or not _WORKFLOW_ID_PATTERN.fullmatch(workflow_id)
+        or not is_valid_workflow_id(workflow_id)
     ):
         console.print(
             f"[red]Error:[/red] Invalid workflow ID: {_escape_markup(repr(workflow_id))}"
@@ -925,6 +925,54 @@ def _failed_step_error(state: Any) -> str | None:
     return getattr(state, "error", None)
 
 
+def _scope_current_step_id(record: dict[str, Any]) -> str | None:
+    """Step id a serialized scope rests on.
+
+    Prefers the persisted ``current_step_id``: it is the only accurate source
+    when a scope paused inside a nested control-flow body (``if``/``switch``/
+    loop), where the snapshot index still points at the enclosing step. Falls
+    back to deriving the id from that index for states written before the field
+    was persisted.
+    """
+    persisted = record.get("current_step_id")
+    if isinstance(persisted, str) and persisted:
+        return persisted
+    index = record.get("current_step_index")
+    definition = record.get("definition")
+    steps = definition.get("steps") if isinstance(definition, dict) else None
+    if (
+        isinstance(index, int)
+        and not isinstance(index, bool)
+        and isinstance(steps, list)
+        and 0 <= index < len(steps)
+        and isinstance(steps[index], dict)
+    ):
+        step_id = steps[index].get("id")
+        if isinstance(step_id, str):
+            return step_id
+    return None
+
+
+def _scope_summary(scopes: Any) -> list[dict[str, Any]]:
+    """Compact nested-scope summary for the machine-readable payload."""
+    summary: list[dict[str, Any]] = []
+    if not isinstance(scopes, dict):
+        return summary
+    for key, record in scopes.items():
+        if not isinstance(record, dict):
+            continue
+        summary.append(
+            {
+                "invocation_id": key,
+                "workflow_id": record.get("workflow_id"),
+                "status": record.get("status"),
+                "current_step_id": _scope_current_step_id(record),
+                "scopes": _scope_summary(record.get("workflow_scopes")),
+            }
+        )
+    return summary
+
+
 def _workflow_run_payload(state: Any) -> dict[str, Any]:
     """Machine-readable summary of a run/resume outcome."""
     payload = {
@@ -940,6 +988,11 @@ def _workflow_run_payload(state: Any) -> dict[str, Any]:
     error = _failed_step_error(state)
     if error is not None:
         payload["error"] = error
+    # Only present when composition is in play, so existing payloads stay
+    # byte-for-byte stable for runs without nested scopes.
+    scopes = _scope_summary(getattr(state, "workflow_scopes", None))
+    if scopes:
+        payload["scopes"] = scopes
     return payload
 
 
@@ -961,6 +1014,61 @@ def _is_gate_step(step: dict[str, Any]) -> bool:
     return isinstance(output, dict) and "on_reject" in output
 
 
+def _gate_details(step_id: str, output: Any) -> dict[str, Any]:
+    """Normalise a gate step's output into the stable JSON gate schema.
+
+    Unvalidated or legacy gate records may contain non-string values, so all
+    fields are normalised: message → str, options → list[str] | None, choice →
+    str | None (None means no decision yet).
+    """
+    output = output if isinstance(output, dict) else {}
+    message = output.get("message")
+    choice = output.get("choice")
+    return {
+        "step_id": step_id,
+        "message": None if message is None else str(message),
+        "options": _normalize_gate_options(output.get("options")),
+        "choice": None if choice is None else str(choice),
+    }
+
+
+def _scope_gate(
+    scopes: Any, path: list[str], active_status: str
+) -> dict[str, Any] | None:
+    """Find the active gate inside a serialized scope tree.
+
+    A pause/abort inside a composed workflow leaves the *root* resting on the
+    ``workflow`` call, so the gate itself lives in a nested scope. Return its
+    detail augmented with the ``scope_path`` (invocation ids root → leaf) so an
+    orchestrator can drive a nested gate exactly as a top-level one.
+    """
+    if not isinstance(scopes, dict):
+        return None
+    for key, record in scopes.items():
+        if not isinstance(record, dict):
+            continue
+        child_path = [*path, key]
+        if str(record.get("status")) != active_status:
+            continue
+        # A deeper active scope is more specific; check descendants first.
+        nested = _scope_gate(
+            record.get("workflow_scopes"), child_path, active_status
+        )
+        if nested is not None:
+            return nested
+        step_id = _scope_current_step_id(record)
+        step_results = record.get("step_results")
+        if step_id is None or not isinstance(step_results, dict):
+            continue
+        step = step_results.get(step_id)
+        if isinstance(step, dict) and _is_gate_step(step):
+            return {
+                **_gate_details(step_id, step.get("output")),
+                "scope_path": child_path,
+            }
+    return None
+
+
 def _gate_outcome(state: Any) -> dict[str, Any] | None:
     """Gate detail for the structured outcome, when the run rests at a gate.
 
@@ -975,24 +1083,14 @@ def _gate_outcome(state: Any) -> dict[str, Any] | None:
     # notably `completed`/`failed` — must be suppressed: current_step_id is
     # not cleared when a run whose last executed step was a gate moves on, so
     # without this guard it would surface stale detail (run/resume/status).
-    if getattr(state.status, "value", state.status) not in ("paused", "aborted"):
+    status = getattr(state.status, "value", state.status)
+    if status not in ("paused", "aborted"):
         return None
     step = (getattr(state, "step_results", None) or {}).get(state.current_step_id)
-    if not isinstance(step, dict) or not _is_gate_step(step):
-        return None
-    output = step.get("output") or {}
-    # `message`, `options`, and `choice` may be non-string YAML literals in an
-    # unvalidated workflow (GateStep coerces none of them for the payload), so
-    # normalise all three for a stable JSON schema: message → str, options →
-    # list[str] | None, choice → str | None (None means no decision yet).
-    message = output.get("message")
-    choice = output.get("choice")
-    return {
-        "step_id": state.current_step_id,
-        "message": None if message is None else str(message),
-        "options": _normalize_gate_options(output.get("options")),
-        "choice": None if choice is None else str(choice),
-    }
+    if isinstance(step, dict) and _is_gate_step(step):
+        return _gate_details(state.current_step_id, step.get("output"))
+    # Not a top-level gate: a composed workflow may be paused on a nested gate.
+    return _scope_gate(getattr(state, "workflow_scopes", None), [], status)
 
 
 def _normalize_gate_options(options: Any) -> list[str] | None: