feat: add specify workflow definition for run-scoped definitions

docs/reference/workflows.md

@@ -103,6 +103,53 @@ specify workflow status [<run_id>]
 
 Shows the status of a specific run, or lists all runs if no ID is given. Run states: `created`, `running`, `completed`, `paused`, `failed`, `aborted`.
 
+## Workflow Definition
+
+```bash
+specify workflow definition <run_id> --json
+```
+
+| Option   | Description                                                      |
+| -------- | ---------------------------------------------------------------- |
+| `--json` | Required. Emit the run's workflow definition as a JSON object    |
+
+Returns the workflow definition a run was **started with**, so an integration can render a run (for example every gate and its message) without drifting from what the engine actually executes. It reads only the run's own persisted data and never the installed workflow: editing or removing the installed workflow does not change the output. Runtime progress and the active gate come from [`specify workflow status --json`](#workflow-status).
+
+Omitting `--json` exits with code `2` and prints a usage message on stderr; no stdout is produced. A text renderer is deferred, so the JSON shape below is the only contract.
+
+```json
+{
+  "run_id": "662bf791",
+  "definition": {
+    "schema_version": "1.0",
+    "workflow": { "id": "spec-review", "name": "Spec review", "version": "1.0.0" },
+    "steps": [
+      { "id": "draft", "type": "command", "command": "speckit.specify" },
+      { "id": "approve", "type": "gate", "message": "Review the specification." }
+    ]
+  },
+  "workflow_scopes": []
+}
+```
+
+- **`definition`** is the workflow as persisted for the run, verbatim and with overlays already applied, so it reflects the steps as executed. Every key is passed through (including `requires`, `outputs` and unknown keys). Nothing is added, renamed or filled in: an omitted field, such as a gate's `options`, means the engine default applies at run time. Expressions (`{{ ... }}`) are not evaluated. The structure follows the [workflow YAML schema](#workflow-yaml-schema), so `schema_version` is the version of this interface and a schema change is a change to it. Keys may be added to the top-level object but are never removed or renamed.
+- **Plain JSON.** YAML-native values that JSON cannot represent, such as an unquoted date, appear as strings. Non-string mapping keys and non-finite numbers are stringified too.
+- **`workflow_scopes`** (always present) holds the frozen definition of every [workflow call](#workflow-composition) that has been reached, in execution order. Each entry is `{ "scope_path", "workflow_id", "definition" }`, with `scope_path` and `workflow_id` identical to the values in `status --json`. A call step inside `definition` stays as authored; the callee is never expanded in place. A call that has not been reached yet, or whose target could not be resolved, has no entry, and its definition is never read from the installed workflow. The list grows as calls are reached, and a call inside a loop or fan-out has one entry per iteration or item.
+- **`scope_path`** is a list of strings from the root: the authored `id` of every enclosing step (`step-<index>` for a step without an `id`). Containers such as `if`, `switch`, loops and fan-out are elements; each loop iteration and fan-out item adds its index as a string, including the first (`["loop", "0", "call"]`). `workflow_scopes` is always present here, while `status --json` includes it only when the run has workflow calls.
+
+### Matching a gate to its definition
+
+`status --json` reports an active gate as `gate.step_id` and `gate.scope_path`. To find the gate's definition:
+
+1. Take the `workflow_scopes` entry with the longest `scope_path` that is a prefix of `gate.scope_path`. If there is none, use the root `definition`. Do not require an exact match: `gate.scope_path` lists every enclosing step, including `if`, `switch`, loops and fan-out containers, while an entry's `scope_path` ends at the call step. A gate in an `if` named `wrap` inside a workflow called by `call`, itself in an `if` named `route`, reports `["route", "call", "wrap"]`, and its entry has `["route", "call"]`.
+2. In that definition, find the step whose `id` equals the authored step ID. Step IDs are unique within one workflow definition. In later loop iterations and in fan-out items `step_id` is qualified as `<container>:<id>:<n>` (for example `fan:g:0`); the authored ID is the second-to-last `:`-separated segment, because `:` is not allowed in authored step IDs.
+
+`status --json` does not report per-step status for steps inside called workflows.
+
+### Errors
+
+Failures write a single `{"error": "<message>"}` object to stderr, leave stdout empty and exit with code `1`: not a Spec Kit project, an invalid `SPECIFY_INIT_DIR`, unsafe (including symlinked) workflow storage, an invalid or unknown run ID, an unreadable run state, a run with no persisted definition (for example one created before definitions were persisted), and a corrupt definition or execution state. Usage errors with `--json` also use this JSON error contract and exit with code `2`.
+
 ## List Installed Workflows
 
 ```bash
@@ -482,7 +529,7 @@ Catalogs are resolved in this order (first match wins):
 3. **User config** — `~/.specify/workflow-catalogs.yml`
 4. **Built-in defaults** — official catalog + community catalog
 
-## Workflow Definition
+## Workflow YAML Schema
 
 Workflows are defined in YAML files. Here is the built-in **Full SDD Cycle** workflow that ships with Spec Kit:
 

src/specify_cli/_installed_list_json.py

@@ -1,8 +1,8 @@
-"""Private JSON output helpers for installed preset and extension lists.
+"""Private strict-JSON output helpers for installed-resource commands.
 
-This module intentionally serves only the two installed-list commands.  Their
-human-facing renderers retain the legacy manager records, while this adapter
-defines the public machine-readable wire contract.
+Installed preset and extension lists use the item helpers below. The shared
+error renderer and Typer command class also keep `info` and workflow-definition
+commands on the strict JSON contract.
 """
 from __future__ import annotations
 
@@ -21,7 +21,7 @@
 
 
 class InstalledListJSONCommand(TyperCommand):
-    """Keep JSON list parse failures on the JSON error contract."""
+    """Keep `--json` parse failures on the JSON error contract."""
 
     def make_context(self, info_name, args, parent=None, **extra):
         json_output = "--json" in args
@@ -78,7 +78,7 @@ def emit_json(value: Any) -> None:
 
 
 def emit_json_error(error: Exception, exit_code: int = 1) -> NoReturn:
-    """Write the list-command error contract and terminate unsuccessfully."""
+    """Write the strict JSON error contract and terminate unsuccessfully."""
     message = str(error).strip() or error.__class__.__name__
     typer.echo(json.dumps({"error": message}, ensure_ascii=False), err=True)
     raise typer.Exit(code=exit_code)

src/specify_cli/workflows/_commands.py

@@ -7,11 +7,12 @@
 
 import contextlib
 import json
+import math
 import os
 import re
 import sys
 from pathlib import Path, PurePosixPath as PurePosixPath
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import typer
 import yaml
@@ -31,6 +32,9 @@
 from .._project import _resolve_init_dir_override as _resolve_init_dir_override
 from ..shared_infra import verify_archive_sha256
 
+if TYPE_CHECKING:
+    from .engine import RunState
+
 workflow_app = typer.Typer(
     name="workflow",
     help="Manage and run automation workflows",
@@ -147,6 +151,15 @@ def _parse_input_values(
     return inputs
 
 
+def _unsafe_dir_error(path: Path, label: str) -> str | None:
+    """Return the unsafe-storage error for *path*, if any."""
+    if path.is_symlink():
+        return f"Refusing to use symlinked {label} path"
+    if path.exists() and not path.is_dir():
+        return f"{label} path exists but is not a directory"
+    return None
+
+
 def _reject_unsafe_dir(path: Path, label: str) -> None:
     """Refuse to proceed when *path* is a symlink or an existing non-directory.
 
@@ -155,26 +168,29 @@ def _reject_unsafe_dir(path: Path, label: str) -> None:
     writes files beneath it must bail first. Absence is tolerated — the caller
     creates the directory — only an existing-but-wrong target is rejected.
     """
-    if path.is_symlink():
-        err_console.print(f"[red]Error:[/red] Refusing to use symlinked {label} path")
-        raise typer.Exit(1)
-    if path.exists() and not path.is_dir():
-        err_console.print(f"[red]Error:[/red] {label} path exists but is not a directory")
+    if error := _unsafe_dir_error(path, label):
+        err_console.print(f"[red]Error:[/red] {error}")
         raise typer.Exit(1)
 
 
+def _workflow_storage_error(project_root: Path) -> str | None:
+    """Return the first unsafe workflow-storage error, if any."""
+    for path, label in (
+        (project_root / ".specify", ".specify"),
+        (project_root / ".specify" / "workflows", ".specify/workflows"),
+        (project_root / ".specify" / "workflows" / "runs", ".specify/workflows/runs"),
+        (project_root / ".specify" / "workflows" / "overlays", ".specify/workflows/overlays"),
+    ):
+        if error := _unsafe_dir_error(path, label):
+            return error
+    return None
+
+
 def _reject_unsafe_workflow_storage(project_root: Path) -> None:
-    """Refuse symlinked workflow storage directories before workflow commands run."""
-    _reject_unsafe_dir(project_root / ".specify", ".specify")
-    _reject_unsafe_dir(project_root / ".specify" / "workflows", ".specify/workflows")
-    _reject_unsafe_dir(
-        project_root / ".specify" / "workflows" / "runs",
-        ".specify/workflows/runs",
-    )
-    _reject_unsafe_dir(
-        project_root / ".specify" / "workflows" / "overlays",
-        ".specify/workflows/overlays",
-    )
+    """Refuse unsafe workflow storage directories before workflow commands run."""
+    if error := _workflow_storage_error(project_root):
+        err_console.print(f"[red]Error:[/red] {error}")
+        raise typer.Exit(1)
 
 
 def _resolve_installed_workflow_ownership(
@@ -1047,9 +1063,53 @@ def _emit_workflow_json(payload: dict[str, Any]) -> None:
 
     Uses the builtin ``print`` rather than ``console.print`` so Rich
     markup interpretation, syntax highlighting, and line-wrapping can
-    never alter the emitted JSON.
+    never alter the emitted JSON. YAML-native scalars that JSON cannot
+    represent (e.g. dates), mapping keys, and non-finite floats are stringified,
+    so output is always plain JSON.
     """
-    print(json.dumps(payload, indent=2))
+    def plain_json(value: Any, active_containers: set[int]) -> Any:
+        if isinstance(value, dict):
+            container_id = id(value)
+            if container_id in active_containers:
+                raise ValueError(
+                    "Workflow JSON output cannot represent recursive YAML aliases"
+                )
+            active_containers.add(container_id)
+            try:
+                return {
+                    key if isinstance(key, (str, int, float, bool)) or key is None else str(key): plain_json(item, active_containers)
+                    for key, item in value.items()
+                }
+            finally:
+                active_containers.remove(container_id)
+        if isinstance(value, float) and not math.isfinite(value):
+            return str(value)
+        if isinstance(value, (list, tuple)):
+            container_id = id(value)
+            if container_id in active_containers:
+                raise ValueError(
+                    "Workflow JSON output cannot represent recursive YAML aliases"
+                )
+            active_containers.add(container_id)
+            try:
+                return [plain_json(item, active_containers) for item in value]
+            finally:
+                active_containers.remove(container_id)
+        return value
+
+    print(json.dumps(plain_json(payload, set()), indent=2, default=str))
+
+
+def _load_run_state(run_id: str, project_root: Path) -> RunState:
+    """Load a run's state; raise ValueError with a user-facing message on failure."""
+    from .engine import RunState
+
+    try:
+        return RunState.load(run_id, project_root)
+    except FileNotFoundError:
+        raise ValueError(f"Run not found: {run_id}") from None
+    except OSError as exc:
+        raise ValueError(str(exc)) from exc
 
 
 @contextlib.contextmanager
@@ -1584,6 +1644,7 @@ def register(app: typer.Typer) -> None:
     from . import command_run  # noqa: F401 -- registers handler
     from . import command_resume  # noqa: F401 -- registers handler
     from . import command_status  # noqa: F401 -- registers handler
+    from . import command_definition  # noqa: F401 -- registers handler
     from . import command_list  # noqa: F401 -- registers handler
     from . import command_add  # noqa: F401 -- registers handler
     from . import command_remove  # noqa: F401 -- registers handler

src/specify_cli/workflows/_execution.py

@@ -331,6 +331,13 @@ def active_step(tree):
     return None
 
 
+def bound_scopes(tree):
+    """Yield (path, binding) for every bound workflow call, in execution order."""
+    for _, node, path, _ in walk_execution(tree["sequence"]):
+        if binding := node.get("binding"):
+            yield path, binding
+
+
 def scope_summaries(tree, run_status="running"):
     """Report workflow boundaries without exposing private inputs or results.
 
@@ -340,24 +347,25 @@ def scope_summaries(tree, run_status="running"):
     summaries = []
     active = active_step(tree)
     active_path = active[0] if active is not None else ()
-    for _, node, path, _ in walk_execution(tree["sequence"]):
-        binding = node.get("binding")
-        if binding:
-            output = node.get("result", {}).get("output", {})
-            status = output.get("status")
-            if status is None:
-                status = (
-                    run_status
-                    if run_status in HALTING and active_path[:len(path)] == path
-                    else "running"
-                )
-            summaries.append(
-                {
-                    "scope_path": path,
-                    "workflow_id": binding["workflow"],
-                    "status": status,
-                }
+    nodes_by_path = {
+        tuple(path): node for _, node, path, _ in walk_execution(tree["sequence"])
+    }
+    for path, binding in bound_scopes(tree):
+        output = nodes_by_path[tuple(path)].get("result", {}).get("output", {})
+        status = output.get("status")
+        if status is None:
+            status = (
+                run_status
+                if run_status in HALTING and active_path[:len(path)] == path
+                else "running"
             )
+        summaries.append(
+            {
+                "scope_path": path,
+                "workflow_id": binding["workflow"],
+                "status": status,
+            }
+        )
     return summaries
 
 

src/specify_cli/workflows/command_definition.py

@@ -0,0 +1,64 @@
+"""Command handler for ``specify workflow definition``."""
+
+from __future__ import annotations
+
+import sys
+
+from .._installed_list_json import InstalledListJSONCommand, emit_json_error
+from .._project import ProjectResolutionError, resolve_specify_project_root
+from . import _commands as cli
+
+
+def _require_json_flag(json_output: bool) -> None:
+    if not json_output:
+        print(
+            "specify workflow definition requires --json for now; "
+            "text output is not yet implemented.",
+            file=sys.stderr,
+        )
+        raise cli.typer.Exit(2)
+
+
+@cli.workflow_app.command("definition", cls=InstalledListJSONCommand)
+def workflow_definition(
+    run_id: str = cli.typer.Argument(..., help="Run ID to inspect"),
+    json_output: bool = cli.typer.Option(
+        False,
+        "--json",
+        help="Emit the run's workflow definition as a JSON object on stdout.",
+    ),
+):
+    """Show the workflow definition a run was started with."""
+    _require_json_flag(json_output)
+    try:
+        project_root = resolve_specify_project_root()
+    except ProjectResolutionError as exc:
+        emit_json_error(exc)
+    if error := cli._workflow_storage_error(project_root):
+        emit_json_error(ValueError(error))
+    try:
+        state = cli._load_run_state(run_id, project_root)
+    except ValueError as exc:
+        emit_json_error(exc)
+
+    try:
+        definition = state.load_definition()
+        scopes = state.load_workflow_scopes()
+    except FileNotFoundError:
+        emit_json_error(ValueError(f"Run {state.run_id} has no persisted workflow definition"))
+    except (OSError, ValueError) as exc:
+        emit_json_error(ValueError(f"Invalid workflow definition for run {state.run_id}: {exc}"))
+
+    try:
+        cli._emit_workflow_json(
+            {
+                "run_id": state.run_id,
+                "definition": definition.data,
+                "workflow_scopes": [
+                    {"scope_path": path, "workflow_id": workflow_id, "definition": d.data}
+                    for path, workflow_id, d in scopes
+                ],
+            }
+        )
+    except ValueError as exc:
+        emit_json_error(exc)

src/specify_cli/workflows/command_status.py

@@ -27,22 +27,12 @@ def workflow_status(
         # parseable (mirrors `workflow run`/`workflow resume`); both handlers
         # fire before the json_output branch below.
         err = cli._error_console(json_output)
-        try:
-            from .engine import RunState
 
-            state = RunState.load(run_id, project_root)
-        except FileNotFoundError:
-            err.print(f"[red]Error:[/red] Run not found: {run_id}")
-            raise cli.typer.Exit(1)
+        try:
+            state = cli._load_run_state(run_id, project_root)
         except ValueError as exc:
             err.print(f"[red]Error:[/red] {cli._escape_markup(str(exc))}")
             raise cli.typer.Exit(1)
-        except OSError as exc:
-            # An unreadable state.json (bad permissions, a directory in its
-            # place, I/O error) must fail as cleanly as the malformed-JSON
-            # case above -- `workflow resume` already handles OSError here.
-            err.print(f"[red]Error:[/red] {cli._escape_markup(str(exc))}")
-            raise cli.typer.Exit(1)
 
         if json_output:
             # Build on the shared run/resume payload so the common fields