diff --git a/docs/reference/workflows.md b/docs/reference/workflows.md
index 93377f4bfb..4e3a43066e 100644
--- a/docs/reference/workflows.md
+++ b/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:
 
diff --git a/src/specify_cli/_installed_list_json.py b/src/specify_cli/_installed_list_json.py
index da671b293a..646a97e35e 100644
--- a/src/specify_cli/_installed_list_json.py
+++ b/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)
diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py
index 2941f2c5c0..991d8b594c 100644
--- a/src/specify_cli/workflows/_commands.py
+++ b/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
diff --git a/src/specify_cli/workflows/_execution.py b/src/specify_cli/workflows/_execution.py
index 6d2a2e89df..a87fcd5f2d 100644
--- a/src/specify_cli/workflows/_execution.py
+++ b/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
 
 
diff --git a/src/specify_cli/workflows/command_definition.py b/src/specify_cli/workflows/command_definition.py
new file mode 100644
index 0000000000..6d3d9683c9
--- /dev/null
+++ b/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)
diff --git a/src/specify_cli/workflows/command_status.py b/src/specify_cli/workflows/command_status.py
index 97471eb10c..a6ba15a7bc 100644
--- a/src/specify_cli/workflows/command_status.py
+++ b/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
diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py
index 5e85353834..4ce5e77135 100644
--- a/src/specify_cli/workflows/engine.py
+++ b/src/specify_cli/workflows/engine.py
@@ -736,6 +736,52 @@ def __init__(
     def runs_dir(self) -> Path:
         return self.project_root / ".specify" / "workflows" / "runs" / self.run_id
 
+    @property
+    def definition_path(self) -> Path:
+        """Snapshot of the workflow definition the run was started with.
+
+        Storage contract: ``resume`` and ``specify workflow definition`` both
+        read this file, so its location and persisted shape must not change.
+        """
+        return self.runs_dir / "workflow.yml"
+
+    def load_definition(self) -> WorkflowDefinition:
+        """Load the definition persisted when this run started.
+
+        Raises ``FileNotFoundError`` if no snapshot exists and ``ValueError``
+        for invalid YAML or a non-mapping document. Never consults the
+        installed workflow.
+        """
+        return WorkflowDefinition.from_yaml(self.definition_path)
+
+    def load_workflow_scopes(
+        self,
+    ) -> list[tuple[list[str], str, WorkflowDefinition]]:
+        """Bound workflow calls as ``(scope_path, workflow_id, definition)``.
+
+        Returned in execution order, with the same ``scope_path`` and
+        ``workflow_id`` that ``status`` reports. Empty when the run has no
+        execution tree. Read-only: the tree is structurally validated but never
+        cross-checked against the root workflow snapshot.
+        Raises ``ValueError`` if the tree or a frozen definition is unusable.
+        """
+        tree = self.execution
+        if tree is None:
+            return []
+        from ._execution import bound_scopes, validate_execution
+
+        validate_execution(tree)
+        scopes = []
+        for path, binding in bound_scopes(tree):
+            scopes.append(
+                (
+                    path,
+                    binding["workflow"],
+                    WorkflowDefinition.from_string(binding["definition"]),
+                )
+            )
+        return scopes
+
     def save(self) -> None:
         """Persist current state to disk.
 
@@ -1107,14 +1153,11 @@ def execute(
         )
         self._execution_state.set(state)
 
-        # Persist a copy of the workflow definition so resume can
-        # reload it even if the original source is no longer available
-        # (e.g. a local YAML path that was moved or deleted).
-        run_dir = self.project_root / ".specify" / "workflows" / "runs" / state.run_id
-        run_dir.mkdir(parents=True, exist_ok=True)
-        workflow_copy = run_dir / "workflow.yml"
-        import yaml
-        with open(workflow_copy, "w", encoding="utf-8") as f:
+        # Persist the definition the run was started with. Storage contract
+        # (see RunState.definition_path): read by resume and by
+        # `specify workflow definition`; do not change location or content.
+        state.runs_dir.mkdir(parents=True, exist_ok=True)
+        with open(state.definition_path, "w", encoding="utf-8") as f:
             yaml.safe_dump(definition.data, f, sort_keys=False, allow_unicode=True)
 
         # Resolve inputs
@@ -1182,11 +1225,9 @@ def resume(
         # Load the workflow definition — try the persisted copy in the
         # run directory first so resume works even if the original
         # source (e.g. a local YAML path) is no longer available.
-        run_dir = self.project_root / ".specify" / "workflows" / "runs" / run_id
-        run_copy = run_dir / "workflow.yml"
-        if run_copy.exists():
-            definition = WorkflowDefinition.from_yaml(run_copy)
-        else:
+        try:
+            definition = state.load_definition()
+        except FileNotFoundError:
             definition = self.load_workflow(state.workflow_id)
 
         # RunState.load() rejects a non-int/negative current_step_index but
diff --git a/tests/specify_cli/workflows/test_command_definition.py b/tests/specify_cli/workflows/test_command_definition.py
new file mode 100644
index 0000000000..6c3ad81429
--- /dev/null
+++ b/tests/specify_cli/workflows/test_command_definition.py
@@ -0,0 +1,602 @@
+"""Tests for ``specify workflow definition``."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+import yaml
+
+from specify_cli.workflows.catalog import WorkflowRegistry
+from tests.specify_cli.workflows import helpers
+
+
+def _invoke(project_dir, args):
+    from unittest.mock import patch
+
+    from typer.testing import CliRunner
+
+    from specify_cli import app
+
+    with patch.object(Path, "cwd", return_value=project_dir):
+        return CliRunner().invoke(app, args, catch_exceptions=False)
+
+
+def _run(project_dir, text, name="root"):
+    path = project_dir / f"{name}.yml"
+    path.write_text(text, encoding="utf-8")
+    result = _invoke(project_dir, ["workflow", "run", str(path), "--json"])
+    return json.loads(result.stdout)
+
+
+def _definition(project_dir, run_id, *extra):
+    return _invoke(project_dir, ["workflow", "definition", run_id, "--json", *extra])
+
+
+def _status(project_dir, run_id):
+    return json.loads(
+        _invoke(project_dir, ["workflow", "status", run_id, "--json"]).stdout
+    )
+
+
+def _install(project_dir, workflow_id, body):
+    wf_dir = project_dir / ".specify" / "workflows" / workflow_id
+    wf_dir.mkdir(parents=True, exist_ok=True)
+    (wf_dir / "workflow.yml").write_text(body, encoding="utf-8")
+    WorkflowRegistry(project_dir).add(workflow_id, {"enabled": True})
+
+
+def _snapshot(project_dir, run_id) -> Path:
+    return project_dir / ".specify" / "workflows" / "runs" / run_id / "workflow.yml"
+
+
+def _assert_json_error(result, fragment):
+    assert result.exit_code == 1, result.output
+    assert result.stdout == ""
+    error = json.loads(result.stderr)
+    assert list(error) == ["error"]
+    assert fragment in error["error"]
+
+
+GATE_WF = """
+schema_version: "1.0"
+workflow:
+  id: "gate-wf"
+  name: "Gate WF"
+  version: "1.0.0"
+steps:
+  - id: approve
+    type: gate
+    message: "Review"
+    options: [approve, reject]
+    on_reject: abort
+  - id: after
+    type: shell
+    run: "echo done"
+"""
+
+CHILD = (
+    "workflow: {id: child, name: Child}\n"
+    "inputs:\n  verdict: {type: string, default: ''}\n"
+    "steps:\n  - {id: review, type: gate, message: Review, verdict_input: verdict}\n"
+)
+
+PARENT = (
+    "workflow: {id: parent, name: Parent}\n"
+    "inputs:\n  verdict: {type: string, default: ''}\n"
+    "steps:\n  - id: call\n    type: workflow\n    workflow: child\n"
+    "    input: {verdict: '{{ inputs.verdict }}'}\n"
+)
+
+
+def _gate_scope(definition, scopes, gate):
+    """Join rule from the contract: longest-prefix scope, then step_id."""
+    path = gate.get("scope_path", [])
+    best = None
+    for scope in scopes:
+        n = len(scope["scope_path"])
+        if path[:n] == scope["scope_path"] and (best is None or n > len(best["scope_path"])):
+            best = scope
+    root = best["definition"] if best else definition
+    step_id = gate["step_id"]
+    if step_id.count(":") >= 2:
+        step_id = step_id.split(":")[-2]
+
+    def find(steps):
+        for step in steps:
+            if step.get("id") == step_id:
+                return step
+            for key in ("then", "else", "steps", "default"):
+                if isinstance(step.get(key), list) and (found := find(step[key])):
+                    return found
+            for case in (step.get("cases") or {}).values():
+                if found := find(case):
+                    return found
+            if isinstance(step.get("step"), dict) and (found := find([step["step"]])):
+                return found
+        return None
+
+    return find(root["steps"])
+
+
+class TestPositive:
+    def test_metadata_and_gate_fields(self, project_dir):
+        run = _run(project_dir, GATE_WF)
+        result = _definition(project_dir, run["run_id"])
+        assert result.exit_code == 0, result.output
+        payload = json.loads(result.stdout)
+        # 14. exactly the contracted envelope
+        assert list(payload) == ["run_id", "definition", "workflow_scopes"]
+        assert payload["run_id"] == run["run_id"]
+        definition = payload["definition"]
+        assert definition["schema_version"] == "1.0"
+        assert definition["workflow"] == {
+            "id": "gate-wf", "name": "Gate WF", "version": "1.0.0",
+        }
+        gate = definition["steps"][0]
+        assert gate["message"] == "Review"
+        assert gate["options"] == ["approve", "reject"]
+        assert gate["on_reject"] == "abort"
+        assert payload["workflow_scopes"] == []
+
+    def test_nested_steps_preserved(self, project_dir):
+        text = """
+workflow: {id: nested, name: Nested}
+steps:
+  - {id: g, type: gate, message: first}
+  - id: branch
+    type: if
+    condition: "{{ true }}"
+    then: [{id: t1, type: shell, run: "echo t"}]
+    else: [{id: e1, type: shell, run: "echo e"}]
+  - id: pick
+    type: switch
+    expression: "{{ 'a' }}"
+    cases:
+      a: [{id: ca, type: shell, run: "echo a"}]
+    default: [{id: dflt, type: shell, run: "echo d"}]
+  - id: w
+    type: while
+    condition: "{{ false }}"
+    max_iterations: 2
+    steps: [{id: wbody, type: shell, run: "echo w"}]
+  - id: dw
+    type: do-while
+    condition: "{{ false }}"
+    max_iterations: 2
+    steps: [{id: dwbody, type: shell, run: "echo dw"}]
+  - id: fan
+    type: fan-out
+    items: "{{ ['a'] }}"
+    step: {id: item, type: shell, run: "echo item"}
+"""
+        run = _run(project_dir, text)
+        definition = json.loads(_definition(project_dir, run["run_id"]).stdout)["definition"]
+        steps = {s["id"]: s for s in definition["steps"]}
+        assert steps["branch"]["then"][0]["id"] == "t1"
+        assert steps["branch"]["else"][0]["id"] == "e1"
+        assert steps["pick"]["cases"]["a"][0]["id"] == "ca"
+        assert steps["pick"]["default"][0]["id"] == "dflt"
+        assert steps["w"]["steps"][0]["id"] == "wbody"
+        assert steps["dw"]["steps"][0]["id"] == "dwbody"
+        assert steps["fan"]["step"]["id"] == "item"
+
+    def test_verbatim_no_defaults_and_unknown_keys_kept(self, project_dir):
+        text = (
+            "workflow: {id: bare, name: Bare}\n"
+            "custom_key: {a: 1}\n"
+            "requires: {speckit_version: '>=0.1.0'}\n"
+            "steps:\n  - {id: g, type: gate, message: hi}\n"
+        )
+        run = _run(project_dir, text)
+        definition = json.loads(_definition(project_dir, run["run_id"]).stdout)["definition"]
+        assert definition["steps"] == [{"id": "g", "type": "gate", "message": "hi"}]
+        assert definition["custom_key"] == {"a": 1}
+        assert definition["requires"] == {"speckit_version": ">=0.1.0"}
+        assert "schema_version" not in definition
+
+    def test_plain_json_dates_are_strings(self, project_dir):
+        # An unknown top-level key is passed through verbatim (C3), so an
+        # unquoted date reaches the JSON boundary as a datetime.date.
+        text = (
+            "workflow: {id: dated, name: Dated}\n"
+            "released: 2026-01-01\n"
+            "steps:\n  - {id: g, type: gate, message: hi}\n"
+        )
+        run = _run(project_dir, text)
+        snapshot = yaml.safe_load(_snapshot(project_dir, run["run_id"]).read_text())
+        assert str(snapshot["released"]) == "2026-01-01"
+        assert not isinstance(snapshot["released"], str)  # snapshot keeps YAML type
+        result = _definition(project_dir, run["run_id"])
+        assert result.exit_code == 0, result.output
+        assert json.loads(result.stdout)["definition"]["released"] == "2026-01-01"
+
+    def test_plain_json_date_mapping_keys_are_strings(self, project_dir):
+        text = (
+            "workflow: {id: dated-key, name: Dated Key}\n"
+            "steps:\n"
+            "  - id: route\n"
+            "    type: switch\n"
+            "    expression: '{{ 1 }}'\n"
+            "    cases:\n"
+            "      2026-01-01: [{id: g, type: gate, message: hi}]\n"
+        )
+        run = _run(project_dir, text)
+        result = _definition(project_dir, run["run_id"])
+        assert result.exit_code == 0, result.output
+        assert json.loads(result.stdout)["definition"]["steps"][0]["cases"] == {
+            "2026-01-01": [{"id": "g", "type": "gate", "message": "hi"}],
+        }
+
+    @pytest.mark.parametrize("value", [".nan", ".inf"])
+    def test_plain_json_non_finite_floats_are_strings(self, project_dir, value):
+        text = (
+            "workflow: {id: non-finite, name: Non Finite}\n"
+            f"x: {value}\n"
+            "steps:\n  - {id: g, type: gate, message: hi}\n"
+        )
+        run = _run(project_dir, text)
+        result = _definition(project_dir, run["run_id"])
+        assert result.exit_code == 0, result.output
+
+        def reject_constant(constant):
+            raise ValueError(constant)
+
+        assert json.loads(result.stdout, parse_constant=reject_constant)["definition"]["x"] == value[1:]
+
+    def test_installed_workflow_modified_does_not_change_output(self, project_dir):
+        _install(project_dir, "gate-wf", GATE_WF)
+        run = json.loads(
+            _invoke(project_dir, ["workflow", "run", "gate-wf", "--json"]).stdout
+        )
+        before = _definition(project_dir, run["run_id"]).stdout
+        wf = project_dir / ".specify" / "workflows" / "gate-wf" / "workflow.yml"
+        wf.write_text(
+            GATE_WF.replace("approve\n    type", "renamed\n    type").replace(
+                'message: "Review"', 'message: "Changed"'
+            ),
+            encoding="utf-8",
+        )
+        after = _definition(project_dir, run["run_id"])
+        assert after.exit_code == 0
+        assert after.stdout == before
+        assert "Changed" not in after.stdout
+
+    def test_installed_workflow_removed_still_returns_definition(self, project_dir):
+        _install(project_dir, "gate-wf", GATE_WF)
+        run = json.loads(
+            _invoke(project_dir, ["workflow", "run", "gate-wf", "--json"]).stdout
+        )
+        removed = _invoke(project_dir, ["workflow", "remove", "gate-wf"])
+        assert removed.exit_code == 0, removed.output
+        result = _definition(project_dir, run["run_id"])
+        assert result.exit_code == 0, result.output
+        assert json.loads(result.stdout)["definition"]["workflow"]["id"] == "gate-wf"
+
+    def test_overlay_applied(self, project_dir):
+        _install(project_dir, "gate-wf", GATE_WF)
+        helpers.write_overlay(
+            project_dir, "gate-wf", "ov1", {
+                "id": "ov1",
+                "extends": "gate-wf",
+                "priority": 10,
+                "edits": [{
+                    "operation": "insert_after",
+                    "anchor": "approve",
+                    "step": {"id": "added", "type": "shell", "run": "echo added"},
+                }],
+            },
+        )
+        run = json.loads(
+            _invoke(project_dir, ["workflow", "run", "gate-wf", "--json"]).stdout
+        )
+        steps = json.loads(_definition(project_dir, run["run_id"]).stdout)["definition"]["steps"]
+        assert [s["id"] for s in steps] == ["approve", "added", "after"]
+
+
+class TestContract:
+    def test_json_flag_required(self, project_dir):
+        run = _run(project_dir, GATE_WF)
+        for rid in (run["run_id"], "does-not-exist"):
+            result = _invoke(project_dir, ["workflow", "definition", rid])
+            assert result.exit_code == 2
+            assert result.stdout == ""
+            assert "--json" in result.stderr
+
+    def test_json_usage_error_uses_json_contract(self, project_dir):
+        result = _invoke(project_dir, ["workflow", "definition", "--json"])
+        assert result.exit_code == 2
+        assert result.stdout == ""
+        assert list(json.loads(result.stderr)) == ["error"]
+
+
+class TestComposition:
+    def _call_run(self, project_dir, parent=PARENT):
+        _install(project_dir, "child", CHILD)
+        return _run(project_dir, parent, "parent")
+
+    def test_bound_call_matches_status(self, project_dir):
+        run = self._call_run(project_dir)
+        assert run["status"] == "paused"
+        payload = json.loads(_definition(project_dir, run["run_id"]).stdout)
+        status = _status(project_dir, run["run_id"])
+        # the root keeps the call step as authored; nothing is injected
+        assert payload["definition"]["steps"] == [{
+            "id": "call", "type": "workflow", "workflow": "child",
+            "input": {"verdict": "{{ inputs.verdict }}"},
+        }]
+        [scope] = payload["workflow_scopes"]
+        assert (scope["scope_path"], scope["workflow_id"]) == (
+            status["workflow_scopes"][0]["scope_path"],
+            status["workflow_scopes"][0]["workflow_id"],
+        )
+        assert scope["scope_path"] == ["call"]
+        assert scope["definition"]["steps"][0]["message"] == "Review"
+        gate = _gate_scope(payload["definition"], payload["workflow_scopes"], status["gate"])
+        assert gate["id"] == "review"
+
+    def test_nested_gate_join_uses_prefix_not_exact_match(self, project_dir):
+        _install(
+            project_dir, "child",
+            "workflow: {id: child, name: Child}\n"
+            "steps:\n"
+            "  - id: wrap\n    type: if\n    condition: '{{ true }}'\n"
+            "    then: [{id: review, type: gate, message: Review}]\n",
+        )
+        parent = (
+            "workflow: {id: parent, name: Parent}\n"
+            "steps:\n"
+            "  - id: route\n    type: if\n    condition: '{{ true }}'\n"
+            "    then: [{id: call, type: workflow, workflow: child}]\n"
+        )
+        run = _run(project_dir, parent, "parent")
+        payload = json.loads(_definition(project_dir, run["run_id"]).stdout)
+        gate = _status(project_dir, run["run_id"])["gate"]
+        assert gate["scope_path"] == ["route", "call", "wrap"]
+        assert [s["scope_path"] for s in payload["workflow_scopes"]] == [["route", "call"]]
+        found = _gate_scope(payload["definition"], payload["workflow_scopes"], gate)
+        assert found["message"] == "Review"
+
+    def test_root_gate_inside_if_has_no_scope(self, project_dir):
+        text = (
+            "workflow: {id: r, name: R}\n"
+            "steps:\n  - id: route\n    type: if\n    condition: '{{ true }}'\n"
+            "    then: [{id: g, type: gate, message: Hello}]\n"
+        )
+        run = _run(project_dir, text)
+        payload = json.loads(_definition(project_dir, run["run_id"]).stdout)
+        gate = _status(project_dir, run["run_id"])["gate"]
+        assert gate["scope_path"] == ["route"]
+        assert payload["workflow_scopes"] == []
+        assert _gate_scope(payload["definition"], [], gate)["message"] == "Hello"
+
+    def test_nested_calls_longest_prefix_wins(self, project_dir):
+        _install(project_dir, "inner", CHILD.replace("id: child", "id: inner"))
+        _install(
+            project_dir, "outer",
+            "workflow: {id: outer, name: Outer}\n"
+            "inputs:\n  verdict: {type: string, default: ''}\n"
+            "steps:\n  - id: inner\n    type: workflow\n    workflow: inner\n"
+            "    input: {verdict: '{{ inputs.verdict }}'}\n",
+        )
+        parent = (
+            "workflow: {id: parent, name: Parent}\n"
+            "inputs:\n  verdict: {type: string, default: ''}\n"
+            "steps:\n  - id: outer\n    type: workflow\n    workflow: outer\n"
+            "    input: {verdict: '{{ inputs.verdict }}'}\n"
+        )
+        run = _run(project_dir, parent, "parent")
+        payload = json.loads(_definition(project_dir, run["run_id"]).stdout)
+        assert [(s["scope_path"], s["workflow_id"]) for s in payload["workflow_scopes"]] == [
+            (["outer"], "outer"), (["outer", "inner"], "inner"),
+        ]
+        gate = _status(project_dir, run["run_id"])["gate"]
+        assert _gate_scope(
+            payload["definition"], payload["workflow_scopes"], gate
+        )["id"] == "review"
+
+    def test_call_inside_loop_scope_path(self, project_dir):
+        _install(project_dir, "child", CHILD)
+        parent = (
+            "workflow: {id: parent, name: Parent}\n"
+            "inputs:\n  verdict: {type: string, default: ''}\n"
+            "steps:\n  - id: loop\n    type: do-while\n    condition: '{{ false }}'\n"
+            "    max_iterations: 1\n"
+            "    steps:\n      - id: call\n        type: workflow\n        workflow: child\n"
+            "        input: {verdict: '{{ inputs.verdict }}'}\n"
+        )
+        run = _run(project_dir, parent, "parent")
+        payload = json.loads(_definition(project_dir, run["run_id"]).stdout)
+        assert [s["scope_path"] for s in payload["workflow_scopes"]] == [["loop", "0", "call"]]
+
+    def test_callee_drift_frozen(self, project_dir):
+        run = self._call_run(project_dir)
+        before = _definition(project_dir, run["run_id"]).stdout
+        child = project_dir / ".specify" / "workflows" / "child" / "workflow.yml"
+        child.write_text(CHILD.replace("message: Review", "message: Changed"), encoding="utf-8")
+        assert _definition(project_dir, run["run_id"]).stdout == before
+        removed = _invoke(project_dir, ["workflow", "remove", "child"])
+        assert removed.exit_code == 0, removed.output
+        after = _definition(project_dir, run["run_id"])
+        assert after.exit_code == 0, after.output
+        assert after.stdout == before
+        assert "Changed" not in after.stdout
+
+    def test_unbound_call_is_absent(self, project_dir):
+        _install(project_dir, "child", CHILD)
+        parent = (
+            "workflow: {id: parent, name: Parent}\n"
+            "steps:\n"
+            "  - {id: first, type: gate, message: Wait}\n"
+            "  - {id: call, type: workflow, workflow: child}\n"
+        )
+        run = _run(project_dir, parent, "parent")
+        payload = json.loads(_definition(project_dir, run["run_id"]).stdout)
+        assert payload["workflow_scopes"] == []
+        assert payload["definition"]["steps"][1] == {
+            "id": "call", "type": "workflow", "workflow": "child",
+        }
+
+    def test_failed_binding_is_absent(self, project_dir):
+        parent = (
+            "workflow: {id: parent, name: Parent}\n"
+            "steps:\n  - {id: call, type: workflow, workflow: missing}\n"
+        )
+        run = _run(project_dir, parent, "parent")
+        assert run["status"] == "failed"
+        payload = json.loads(_definition(project_dir, run["run_id"]).stdout)
+        assert payload["workflow_scopes"] == []
+        assert payload["definition"]["steps"][0]["workflow"] == "missing"
+
+    def test_same_call_bound_per_fan_out_item(self, project_dir):
+        _install(
+            project_dir,
+            "child",
+            "workflow: {id: child, name: Child}\n"
+            "steps: [{id: noop, type: shell, run: 'true'}]\n",
+        )
+        parent = (
+            "workflow: {id: parent, name: Parent}\n"
+            "steps:\n  - id: fan\n    type: fan-out\n    items: \"{{ ['a', 'b'] }}\"\n"
+            "    max_concurrency: 1\n"
+            "    step: {id: call, type: workflow, workflow: child}\n"
+        )
+        run = _run(project_dir, parent, "parent")
+        payload = json.loads(_definition(project_dir, run["run_id"]).stdout)
+        paths = [s["scope_path"] for s in payload["workflow_scopes"]]
+        assert paths == [["fan", "0", "call"], ["fan", "1", "call"]]
+        assert paths == [s["scope_path"] for s in _status(project_dir, run["run_id"])["workflow_scopes"]]
+
+    def test_same_call_bound_per_loop_iteration(self, project_dir):
+        _install(
+            project_dir,
+            "child",
+            "workflow: {id: child, name: Child}\n"
+            "steps: [{id: noop, type: shell, run: 'true'}]\n",
+        )
+        parent = (
+            "workflow: {id: parent, name: Parent}\n"
+            "steps:\n  - id: loop\n    type: do-while\n"
+            "    condition: '{{ true }}'\n    max_iterations: 2\n"
+            "    steps:\n      - {id: call, type: workflow, workflow: child}\n"
+        )
+        run = _run(project_dir, parent, "parent")
+        payload = json.loads(_definition(project_dir, run["run_id"]).stdout)
+        paths = [s["scope_path"] for s in payload["workflow_scopes"]]
+        assert paths == [["loop", "0", "call"], ["loop", "1", "call"]]
+        assert paths == [s["scope_path"] for s in _status(project_dir, run["run_id"])["workflow_scopes"]]
+
+    def test_overlaid_callee(self, project_dir):
+        _install(project_dir, "child", CHILD)
+        helpers.write_overlay(
+            project_dir, "child", "ov1", {
+                "id": "ov1", "extends": "child", "priority": 10,
+                "edits": [{
+                    "operation": "insert_before", "anchor": "review",
+                    "step": {"id": "prep", "type": "shell", "run": "echo prep"},
+                }],
+            },
+        )
+        run = _run(project_dir, PARENT, "parent")
+        [scope] = json.loads(_definition(project_dir, run["run_id"]).stdout)["workflow_scopes"]
+        assert [s["id"] for s in scope["definition"]["steps"]] == ["prep", "review"]
+
+    def test_legacy_run_without_execution_tree(self, project_dir):
+        run = self._call_run(project_dir)
+        state_path = project_dir / ".specify" / "workflows" / "runs" / run["run_id"] / "state.json"
+        state = json.loads(state_path.read_text())
+        state.pop("execution", None)
+        state_path.write_text(json.dumps(state), encoding="utf-8")
+        payload = json.loads(_definition(project_dir, run["run_id"]).stdout)
+        assert payload["workflow_scopes"] == []
+        assert payload["definition"]["steps"][0]["id"] == "call"
+
+
+class TestNegative:
+    def test_unknown_run(self, project_dir):
+        _assert_json_error(_definition(project_dir, "nope"), "Run not found: nope")
+
+    def test_invalid_run_id(self, project_dir):
+        _assert_json_error(_definition(project_dir, "../escape"), "Invalid run_id")
+
+    def test_not_a_project(self, temp_dir):
+        _assert_json_error(_definition(temp_dir, "x"), "Not a Spec Kit project")
+
+    def test_symlinked_workflow_runs_storage(self, project_dir, tmp_path):
+        runs_dir = project_dir / ".specify" / "workflows" / "runs"
+        runs_dir.mkdir(parents=True)
+        runs_dir.rmdir()
+        runs_dir.symlink_to(tmp_path, target_is_directory=True)
+        result = _definition(project_dir, "run")
+        _assert_json_error(result, "Refusing to use symlinked .specify/workflows/runs path")
+
+    def test_invalid_specify_init_dir(self, project_dir, tmp_path, monkeypatch):
+        invalid_root = tmp_path / "not-a-project"
+        invalid_root.mkdir()
+        monkeypatch.setenv("SPECIFY_INIT_DIR", str(invalid_root))
+        result = _definition(project_dir, "run")
+        _assert_json_error(
+            result,
+            f"SPECIFY_INIT_DIR is not a Spec Kit project (no .specify/ directory): {invalid_root}",
+        )
+
+    def test_missing_snapshot_does_not_fall_back(self, project_dir):
+        _install(project_dir, "gate-wf", GATE_WF)
+        run = json.loads(
+            _invoke(project_dir, ["workflow", "run", "gate-wf", "--json"]).stdout
+        )
+        _snapshot(project_dir, run["run_id"]).unlink()
+        result = _definition(project_dir, run["run_id"])
+        _assert_json_error(result, "has no persisted workflow definition")
+        assert "approve" not in result.stderr
+
+    def test_recursive_yaml_alias_returns_json_error(self, project_dir):
+        recursive_workflow = GATE_WF + "\nmetadata: &loop\n  self: *loop\n"
+        run = _run(project_dir, recursive_workflow)
+
+        result = _definition(project_dir, run["run_id"])
+
+        _assert_json_error(result, "cannot represent recursive YAML aliases")
+
+    @pytest.mark.parametrize("content", [": : :\n  - [", "- a\n- b\n"])
+    def test_corrupt_snapshot(self, project_dir, content):
+        run = _run(project_dir, GATE_WF)
+        _snapshot(project_dir, run["run_id"]).write_text(content, encoding="utf-8")
+        _assert_json_error(
+            _definition(project_dir, run["run_id"]), "Invalid workflow definition"
+        )
+
+    def _mutate_state(self, project_dir, run_id, mutate):
+        path = project_dir / ".specify" / "workflows" / "runs" / run_id / "state.json"
+        state = json.loads(path.read_text())
+        mutate(state["execution"])
+        path.write_text(json.dumps(state), encoding="utf-8")
+
+    def test_unsupported_execution_version(self, project_dir):
+        _install(project_dir, "child", CHILD)
+        run = _run(project_dir, PARENT, "parent")
+        self._mutate_state(project_dir, run["run_id"], lambda e: e.update(version=999))
+        _assert_json_error(
+            _definition(project_dir, run["run_id"]), "Unsupported execution version"
+        )
+
+    def test_malformed_execution_tree(self, project_dir):
+        _install(project_dir, "child", CHILD)
+        run = _run(project_dir, PARENT, "parent")
+        self._mutate_state(project_dir, run["run_id"], lambda e: e.pop("sequence"))
+        _assert_json_error(
+            _definition(project_dir, run["run_id"]), "Invalid execution state"
+        )
+
+    def test_invalid_bound_callee_definition(self, project_dir):
+        _install(project_dir, "child", CHILD)
+        run = _run(project_dir, PARENT, "parent")
+
+        def corrupt(execution):
+            execution["sequence"]["nodes"][0]["binding"]["definition"] = "- a\n- b\n"
+
+        self._mutate_state(project_dir, run["run_id"], corrupt)
+        _assert_json_error(
+            _definition(project_dir, run["run_id"]), "Invalid bound workflow definition"
+        )
diff --git a/tests/test_workflows.py b/tests/test_workflows.py
index a4ea5d1786..1fb2a05217 100644
--- a/tests/test_workflows.py
+++ b/tests/test_workflows.py
@@ -8593,6 +8593,126 @@ def test_workflow_snapshot_keeps_non_ascii_text_readable(self, project_dir):
         assert "\\u" not in raw and "\\x" not in raw
         assert yaml.safe_load(raw)["workflow"]["name"] == name
 
+    _GATED = """
+schema_version: "1.0"
+workflow:
+  id: "gated"
+  name: "Gated"
+  version: "1.0.0"
+inputs:
+  verdict: {type: string, default: ""}
+steps:
+  - {id: ask, type: gate, message: "Review", verdict_input: verdict}
+"""
+
+    def _gated_run(self, project_dir):
+        from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine
+
+        wf_dir = project_dir / ".specify" / "workflows" / "gated"
+        wf_dir.mkdir(parents=True)
+        wf_file = wf_dir / "workflow.yml"
+        wf_file.write_text(self._GATED, encoding="utf-8")
+        engine = WorkflowEngine(project_dir)
+        return engine, engine.execute(WorkflowDefinition.from_yaml(wf_file))
+
+    def test_definition_path_and_load_definition(self, project_dir):
+        from specify_cli.workflows.engine import RunState
+
+        _, state = self._gated_run(project_dir)
+        loaded = RunState.load(state.run_id, project_dir)
+        assert loaded.definition_path == loaded.runs_dir / "workflow.yml"
+        definition = loaded.load_definition()
+        assert definition.data == yaml.safe_load(self._GATED)
+
+    def test_load_definition_missing_snapshot_raises_file_not_found(self, project_dir):
+        _, state = self._gated_run(project_dir)
+        state.definition_path.unlink()
+        with pytest.raises(FileNotFoundError):
+            state.load_definition()
+
+    @pytest.mark.parametrize("content", [": : :\n  - [", "- a\n- b\n"])
+    def test_load_definition_rejects_corrupt_snapshot(self, project_dir, content):
+        _, state = self._gated_run(project_dir)
+        state.definition_path.write_text(content, encoding="utf-8")
+        with pytest.raises(ValueError):
+            state.load_definition()
+
+    def test_snapshot_storage_contract(self, project_dir):
+        """``runs/<id>/workflow.yml`` is read by ``resume`` and ``workflow
+        definition``: pin its location and persisted shape."""
+        _, state = self._gated_run(project_dir)
+        path = project_dir / ".specify" / "workflows" / "runs" / state.run_id / "workflow.yml"
+        assert state.definition_path == path
+        data = yaml.safe_load(path.read_text(encoding="utf-8"))
+        assert list(data) == ["schema_version", "workflow", "inputs", "steps"]
+        assert data == yaml.safe_load(self._GATED)
+        assert yaml.safe_load(yaml.safe_dump(data, sort_keys=False)) == data
+
+    def test_resume_uses_snapshot_when_installed_workflow_is_gone(self, project_dir):
+        from specify_cli.workflows.base import RunStatus
+
+        engine, state = self._gated_run(project_dir)
+        (project_dir / ".specify" / "workflows" / "gated" / "workflow.yml").unlink()
+        resumed = engine.resume(state.run_id, {"verdict": "approve"})
+        assert resumed.status == RunStatus.COMPLETED
+
+    def test_resume_falls_back_to_installed_workflow_without_snapshot(self, project_dir):
+        from specify_cli.workflows.base import RunStatus
+
+        engine, state = self._gated_run(project_dir)
+        state.definition_path.unlink()
+        resumed = engine.resume(state.run_id, {"verdict": "approve"})
+        assert resumed.status == RunStatus.COMPLETED
+
+    def test_load_workflow_scopes_without_tree_is_empty(self, project_dir):
+        from specify_cli.workflows.engine import RunState
+
+        state = RunState(run_id="plain", workflow_id="w", project_root=project_dir)
+        state.save()
+        assert RunState.load("plain", project_dir).load_workflow_scopes() == []
+
+    def test_load_workflow_scopes_matches_scope_summaries(self, project_dir):
+        from specify_cli.workflows._execution import scope_summaries
+        from specify_cli.workflows.catalog import WorkflowRegistry
+        from specify_cli.workflows.engine import (
+            RunState,
+            WorkflowDefinition,
+            WorkflowEngine,
+        )
+
+        child_dir = project_dir / ".specify" / "workflows" / "child"
+        child_dir.mkdir(parents=True)
+        (child_dir / "workflow.yml").write_text(
+            "workflow: {id: child, name: Child}\n"
+            "steps:\n  - {id: review, type: gate, message: Review}\n",
+            encoding="utf-8",
+        )
+        WorkflowRegistry(project_dir).add("child", {"enabled": True})
+        root = WorkflowDefinition({
+            "workflow": {"id": "parent", "name": "Parent"},
+            "steps": [{"id": "call", "type": "workflow", "workflow": "child"}],
+        })
+        state = WorkflowEngine(project_dir).execute(root)
+        loaded = RunState.load(state.run_id, project_dir)
+        scopes = loaded.load_workflow_scopes()
+        summaries = scope_summaries(loaded.execution, loaded.status.value)
+        assert [(p, w) for p, w, _ in scopes] == [
+            (s["scope_path"], s["workflow_id"]) for s in summaries
+        ] == [(["call"], "child")]
+        assert scopes[0][2].data["steps"][0]["id"] == "review"
+
+    @pytest.mark.parametrize(
+        "tree",
+        [{"version": 999, "sequence": {}}, {"version": 2}, "not-a-dict"],
+    )
+    def test_load_workflow_scopes_wraps_bad_tree_in_value_error(self, project_dir, tree):
+        from specify_cli.workflows.engine import RunState
+
+        state = RunState(run_id="bad", workflow_id="w", project_root=project_dir)
+        state.execution = tree
+        with pytest.raises(ValueError):
+            state.load_workflow_scopes()
+
     @pytest.mark.parametrize("invalid_step_results", [None, [], "invalid", 1, True])
     def test_load_rejects_non_object_step_results(
         self, project_dir, invalid_step_results