feat(workflows): compose workflows with a unified execution tree

design/workflow-step.md

@@ -41,9 +41,10 @@ unless `continue_on_error: true` is set (an explicit abort always stops).
 The engine calls `validate()` during workflow validation but does not
 automatically validate a definition passed to `execute()`. Guard invalid
 configurations in `execute()` too, returning a failed result rather than a
-successful default or an unhandled exception. Resume restarts the current
-top-level step; a pause inside nested steps re-runs their parent and nested
-body. Design side effects accordingly.
+successful default or an unhandled exception. New runs persist an execution
+tree: resume retries unfinished occurrences and restores completed results and
+selected expansions. Legacy states enter through their old top-level index.
+Side effects performed before their completion checkpoint can still repeat.
 
 The registry holds one shared instance per type. Concurrent `fan-out` can
 invoke that instance from multiple threads: keep execution stateless and

docs/reference/workflows.md

@@ -540,6 +540,73 @@ 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`   | Call an installed workflow with private inputs and declared outputs |
+
+### Workflow composition
+
+A `workflow` step executes an installed, enabled workflow in the current project
+as a private scope of the same run. Targets can be literal IDs or expressions;
+the resolved string must match the ID exactly, including case and whitespace.
+
+```yaml
+inputs:
+  target: {type: string, required: true}
+  report: {type: string, required: true}
+steps:
+  - id: investigate
+    type: workflow
+    workflow: "{{ inputs.target }}"
+    input:
+      report: "{{ inputs.report }}"
+```
+
+The included workflow sees only declared inputs passed through `input` and its
+own step results. Unknown input names, missing required values, and invalid
+types/enums fail the call. Existing defaults and `integration: auto` resolution
+apply. Values cross back only through explicit declarations:
+
+```yaml
+outputs:
+  report:
+    value: "{{ steps.analyze.output.stdout }}"
+```
+
+The caller reads `{{ steps.investigate.output.report }}`. Output includes
+`workflow` and `status`; failures also include `error`, and an abort includes
+`aborted: true`. These names and `integration`, `model`, `options`, and `input`
+are reserved. Returned values must be JSON-safe. Private inputs, step records,
+and logs are not part of the return value. No separate child run is created.
+
+Failures may be handled with `continue_on_error: true` at either the included
+step or workflow-call boundary. Pauses and explicit aborts always stop execution.
+Nested composition is allowed, but repeated workflow IDs on the active call
+path are cycles. Diamonds are allowed. The maximum included depth is 16, with
+the root at depth zero.
+
+### Execution identity and resume
+
+Each step occurrence owns a record in a persisted execution tree. Authored step
+names are local expression aliases, not global execution IDs. Fan-out items
+have independent alias contexts; results remain ordered by item index.
+
+New runs persist selected branches, dynamic custom-step expansions, loop
+iterations, and fan-out items. Resume retries unfinished operations and retains
+completed work without reevaluating already selected branches. Workflow targets
+and overlay-resolved definitions remain bound even if installations change.
+Ordinary resume retains bound inputs; explicit `--input` updates rebind reached,
+incomplete calls through their original mappings. Completed calls retain their
+results. Failed output evaluation retries finalization without repeating child
+commands.
+
+Snapshots are stored as YAML strings inside the private JSON execution tree,
+preserving YAML scalar types. Inputs and results remain JSON values. Legacy runs
+without a tree enter through their saved top-level index, then use tree-backed
+resume. A tree-backed run left `running` by a crashed process can also be resumed;
+only one process may execute or resume a run at a time. A side effect completed
+before its checkpoint may execute again after a crash.
+
+Run/resume/status JSON includes `workflow_scopes` summaries when calls exist and
+reports the active nested gate with its `scope_path`.
 
 > **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.
 

src/specify_cli/workflows/__init__.py

@@ -49,6 +49,7 @@ def _register_builtin_steps() -> None:
     from .step.fan_in import FanInStep
     from .step.fan_out import FanOutStep
     from .step.gate import GateStep
+    from .step.workflow import WorkflowStep
     from .step.if_then import IfThenStep
     from .step.init import InitStep
     from .step.prompt import PromptStep
@@ -62,6 +63,7 @@ def _register_builtin_steps() -> None:
     _register_step(FanInStep())
     _register_step(FanOutStep())
     _register_step(GateStep())
+    _register_step(WorkflowStep())
     _register_step(IfThenStep())
     _register_step(InitStep())
     _register_step(PromptStep())

src/specify_cli/workflows/_commands.py

@@ -940,6 +940,12 @@ def _workflow_run_payload(state: Any) -> dict[str, Any]:
     error = _failed_step_error(state)
     if error is not None:
         payload["error"] = error
+    if getattr(state, "execution", None):
+        from ._execution import scope_summaries
+
+        scopes = scope_summaries(state.execution)
+        if scopes:
+            payload["workflow_scopes"] = scopes
     return payload
 
 
@@ -978,21 +984,36 @@ def _gate_outcome(state: Any) -> dict[str, Any] | None:
     if getattr(state.status, "value", state.status) not in ("paused", "aborted"):
         return None
     step = (getattr(state, "step_results", None) or {}).get(state.current_step_id)
+    step_id = state.current_step_id
+    scope_path = None
+    if getattr(state, "execution", None):
+        from ._execution import active_step
+
+        active = active_step(state.execution)
+        if active is None:
+            return None
+        path, node = active
+        step_id = path[-1]
+        scope_path = path[:-1]
+        step = node.get("result")
     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
+    # legacy or synthetic records, 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,
+    detail = {
+        "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),
     }
+    if scope_path:
+        detail["scope_path"] = scope_path
+    return detail
 
 
 def _normalize_gate_options(options: Any) -> list[str] | None: