feat: add specify workflow definition for run-scoped definitions

docs/reference/workflows.md

@@ -109,6 +109,56 @@ 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 the gates it declares and their messages) 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; because `definition` is emitted as authored, an omitted `schema_version` defaults to `"1.0"` and the engine compares `str(schema_version)` (so unquoted `schema_version: 1.0` in YAML appears as the JSON number `1.0`). Keys may be added to the top-level response envelope (`run_id`, `definition`, `workflow_scopes`) in future versions, but existing envelope keys are never removed or renamed.
+- **Plain JSON.** YAML-native values that JSON cannot represent, such as an unquoted date, appear as strings. Binary values (`!!binary`) are not emitted: each appears as a `"<binary: N bytes>"` placeholder, and the value itself stays in the run's snapshot. A set (`!!set`) appears as a list sorted by its JSON form, so the output does not depend on hash randomization. Non-string mapping keys and non-finite numbers are stringified too. Two mapping keys that collapse to the same JSON property name (for example `1` and `"1"`) make the command fail rather than emit lossy JSON. JSON cannot share a reference, so a reused YAML alias (anchor) is duplicated; a recursive alias, or a non-recursive alias graph that expands past a fixed node or output-size budget, fails instead of consuming unbounded CPU, memory and stdout. The budget applies to each definition separately (the root and every entry in `workflow_scopes`), so a run with many bound workflow calls is not rejected for its total size.
+- **`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.
+- **Not included: dynamically generated steps.** The output contains the authored root definition and the frozen definitions of bound workflow calls. Steps that a custom step type generates at run time through `StepResult.next_steps` are persisted in the run's execution state but are not exported, so a gate created that way has no definition in this output. The built-in `if`, `switch`, `while`, `do-while` and `fan-out` steps take their steps from the authored definition and are covered.
+- **`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. Read `gate.scope_path` as `[]` when it is missing: `status --json` omits it for a gate directly in the root workflow. Then 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. 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. Step IDs are unique within a definition except across fan-out templates, which have their own ID namespace: a root step and a fan-out template step may share an ID. If `gate.scope_path` runs through a fan-out step, look only inside that step's `step` template; otherwise ignore fan-out templates. For example, a root shell `g` and a gate `g` in the template of fan-out `fan` both exist, and `fan:g:0` with `scope_path` `["fan", "0"]` is the template's gate, not the root shell.
+
+A gate generated at run time by a custom step has no definition to match; see the note on dynamically generated steps above.
+
+`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 or too deeply nested run state, a run with no persisted definition (for example one created before definitions were persisted), a corrupt definition or execution state, a definition whose mapping keys collapse to the same JSON property name, and a definition with recursive or too-large-to-expand YAML aliases. Usage errors with `--json` also use this JSON error contract and exit with code `2`.
+
 ## List Installed Workflows
 
 ```bash
@@ -496,7 +546,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",
@@ -165,6 +169,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.
 
@@ -173,26 +186,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(
@@ -1060,14 +1076,165 @@ def _run_outcome_exit_code(status_value: str) -> int:
     return 1 if status_value in ("failed", "aborted") else 0
 
 
-def _emit_workflow_json(payload: dict[str, Any]) -> None:
+# Real workflow definitions are tiny (largest shipped: ~67 nodes); this leaves
+# >1000x headroom while capping the expansion of alias graphs. Callers that
+# emit user-authored definitions pass it as ``max_nodes``.
+_MAX_JSON_NODES = 100000
+
+# Byte cap on each emitted definition. Nodes do not bound size: one alias can
+# repeat a large scalar or deeply indented container many times, so the cap is
+# enforced on the encoded output itself rather than estimated per value type.
+_MAX_JSON_BYTES = 5 * 1024 * 1024
+
+
+def _check_json_byte_budget(plain: Any) -> None:
+    """Raise ``ValueError`` once *plain* encodes to more than the byte cap.
+
+    The encoding is streamed and discarded, so memory stays bounded even when
+    aliases expand a small input into a huge document, and no value type can
+    escape the budget.
+    """
+    total = 0
+    for chunk in json.JSONEncoder(indent=2).iterencode(plain):
+        total += len(chunk)
+        if total > _MAX_JSON_BYTES:
+            raise ValueError(
+                "Workflow JSON output is too large to represent: it "
+                f"exceeds {_MAX_JSON_BYTES} bytes"
+            )
+
+
+def _plain_workflow_json(payload: Any, *, max_nodes: int | None = None) -> Any:
+    """Convert *payload* to plain JSON-representable values.
+
+    YAML-native scalars that JSON cannot represent (e.g. dates), mapping
+    keys, and non-finite floats are stringified, so output is always plain
+    JSON. Binary values are replaced by a ``<binary: N bytes>`` placeholder
+    rather than emitted, and sets become lists sorted so that output does
+    not depend on hash randomization.
+
+    JSON has no way to share a reference, so a YAML alias graph must be
+    duplicated into a tree; a non-recursive alias DAG can therefore expand
+    exponentially even though it is linear on disk. Cycle detection alone
+    lets that terminate, so a caller emitting user-authored definitions passes
+    ``max_nodes`` to bound the expansion instead of rejecting every reused
+    alias. The byte budget is measured by streaming the encoded output, so it
+    covers every value type. Each call has its own node and byte budget, so
+    independent definitions can be budgeted separately. The budget is opt-in:
+    payloads such as ``workflow status --json`` grow with the number of runs
+    and must not be capped. Excessive nesting depth is always reported as
+    ``ValueError``.
+    """
+    remaining_nodes = max_nodes
+
+    def json_key(key: Any) -> str:
+        """Canonicalize a mapping key to the property name json.dumps emits."""
+        if isinstance(key, str):
+            return key
+        if isinstance(key, (int, float, bool)) or key is None:
+            return json.dumps(key)
+        return str(key)
+
+    def plain_json(value: Any, active_containers: set[int]) -> Any:
+        nonlocal remaining_nodes
+        if remaining_nodes is not None:
+            remaining_nodes -= 1
+            if remaining_nodes < 0:
+                raise ValueError(
+                    "Workflow JSON output is too large to represent: it "
+                    f"exceeds {max_nodes} nodes"
+                )
+        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:
+                result: dict[str, Any] = {}
+                for key, item in value.items():
+                    canonical = json_key(key)
+                    if canonical in result:
+                        raise ValueError(
+                            "Workflow JSON output cannot represent duplicate "
+                            f"mapping key {canonical!r}"
+                        )
+                    result[canonical] = plain_json(item, active_containers)
+                return result
+            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)
+        if isinstance(value, (bytes, bytearray)):
+            return f"<binary: {len(value)} bytes>"
+        if isinstance(value, (set, frozenset)):
+            items = [plain_json(item, active_containers) for item in value]
+            return sorted(items, key=json.dumps)
+        if isinstance(value, (str, int, float)) or value is None:
+            return value
+        return str(value)
+
+    try:
+        plain = plain_json(payload, set())
+        if max_nodes is not None:
+            _check_json_byte_budget(plain)
+        return plain
+    except RecursionError:
+        raise ValueError(
+            "Workflow JSON output is nested too deeply to represent"
+        ) from None
+
+
+def _emit_workflow_json(
+    payload: dict[str, Any], *, max_nodes: int | None = None
+) -> None:
     """Write a workflow payload as machine-readable JSON to stdout.
 
     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. See ``_plain_workflow_json`` for the
+    value conversion and the opt-in ``max_nodes`` budget.
     """
-    print(json.dumps(payload, indent=2))
+    plain = _plain_workflow_json(payload, max_nodes=max_nodes)
+    try:
+        text = json.dumps(plain, indent=2)
+    except RecursionError:
+        raise ValueError(
+            "Workflow JSON output is nested too deeply to represent"
+        ) from None
+    print(text)
+
+
+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
+    except json.JSONDecodeError as exc:
+        # Newer interpreters report pathological nesting as a decode error
+        # rather than RecursionError.
+        raise ValueError(f"Invalid run state for {run_id}: {exc}") from None
+    except RecursionError:
+        raise ValueError(
+            f"Invalid run state for {run_id}: nested too deeply"
+        ) from None
 
 
 @contextlib.contextmanager
@@ -1602,6 +1769,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