feat(workflows): add --json output to workflow run, resume, and status

Adds an opt-in `--json` flag to `workflow run`, `workflow resume`, and
`workflow status` that emits a single machine-readable object (run_id,
workflow_id, status, current step; status also reports per-step states
and a runs list) for automation and external orchestrators.

JSON is written via a small `_emit_workflow_json` helper using plain
stdout, so Rich markup, highlighting, and line-wrapping can never alter
the emitted object. Default human-readable output and exit codes are
unchanged when `--json` is omitted. Reference docs updated.

Closes #2811.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: doquanghuy

docs/reference/workflows.md

@@ -11,6 +11,7 @@ specify workflow run <source>
 | Option              | Description                                              |
 | ------------------- | -------------------------------------------------------- |
 | `-i` / `--input`    | Pass input values as `key=value` (repeatable)            |
+| `--json`            | Emit the run outcome as a single JSON object             |
 
 Runs a workflow from a catalog ID, URL, or local file path. Inputs declared by the workflow can be provided via `--input` or will be prompted interactively.
 
@@ -20,6 +21,14 @@ Example:
 specify workflow run speckit -i spec="Build a kanban board with drag-and-drop task management" -i scope=full
 ```
 
+With `--json`, a single machine-readable object is printed instead of formatted text (the default output is unchanged when the flag is omitted):
+
+```bash
+specify workflow run wf.yml --json
+# {"run_id": "662bf791", "workflow_id": "wf", "status": "paused",
+#  "current_step_id": "review", "current_step_index": 0}
+```
+
 > **Note:** All workflow commands require a project already initialized with `specify init`.
 
 ## Resume a Workflow
@@ -28,6 +37,10 @@ specify workflow run speckit -i spec="Build a kanban board with drag-and-drop ta
 specify workflow resume <run_id>
 ```
 
+| Option              | Description                                              |
+| ------------------- | -------------------------------------------------------- |
+| `--json`            | Emit the resume outcome as a single JSON object          |
+
 Resumes a paused or failed workflow run from the exact step where it stopped. Useful after responding to a gate step or fixing an issue that caused a failure.
 
 ## Workflow Status
@@ -36,6 +49,10 @@ Resumes a paused or failed workflow run from the exact step where it stopped. Us
 specify workflow status [<run_id>]
 ```
 
+| Option              | Description                                              |
+| ------------------- | -------------------------------------------------------- |
+| `--json`            | Emit run status (or the runs list) as a JSON object      |
+
 Shows the status of a specific run, or lists all runs if no ID is given. Run states: `created`, `running`, `completed`, `paused`, `failed`, `aborted`.
 
 ## List Installed Workflows

src/specify_cli/__init__.py

@@ -4174,19 +4174,46 @@ def extension_set_priority(
 workflow_app.add_typer(workflow_catalog_app, name="catalog")
 
 
+def _workflow_run_payload(state: Any) -> dict[str, Any]:
+    """Machine-readable summary of a run/resume outcome."""
+    return {
+        "run_id": state.run_id,
+        "workflow_id": state.workflow_id,
+        "status": state.status.value,
+        "current_step_id": state.current_step_id,
+        "current_step_index": state.current_step_index,
+    }
+
+
+def _emit_workflow_json(payload: dict[str, Any]) -> 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.
+    """
+    print(json.dumps(payload, indent=2))
+
+
 @workflow_app.command("run")
 def workflow_run(
     source: str = typer.Argument(..., help="Workflow ID or YAML file path"),
     input_values: list[str] | None = typer.Option(
         None, "--input", "-i", help="Input values as key=value pairs"
     ),
+    json_output: bool = typer.Option(
+        False,
+        "--json",
+        help="Emit the run outcome as a single JSON object instead of formatted text.",
+    ),
 ):
     """Run a workflow from an installed ID or local YAML path."""
     from .workflows.engine import WorkflowEngine
 
     project_root = _require_specify_project()
     engine = WorkflowEngine(project_root)
-    engine.on_step_start = lambda sid, label: console.print(f"  \u25b8 [{sid}] {label} \u2026")
+    if not json_output:
+        engine.on_step_start = lambda sid, label: console.print(f"  \u25b8 [{sid}] {label} \u2026")
 
     try:
         definition = engine.load_workflow(source)
@@ -4215,8 +4242,9 @@ def workflow_run(
             key, _, value = kv.partition("=")
             inputs[key.strip()] = value.strip()
 
-    console.print(f"\n[bold cyan]Running workflow:[/bold cyan] {definition.name} ({definition.id})")
-    console.print(f"[dim]Version: {definition.version}[/dim]\n")
+    if not json_output:
+        console.print(f"\n[bold cyan]Running workflow:[/bold cyan] {definition.name} ({definition.id})")
+        console.print(f"[dim]Version: {definition.version}[/dim]\n")
 
     try:
         state = engine.execute(definition, inputs)
@@ -4227,6 +4255,10 @@ def workflow_run(
         console.print(f"[red]Workflow failed:[/red] {exc}")
         raise typer.Exit(1)
 
+    if json_output:
+        _emit_workflow_json(_workflow_run_payload(state))
+        return
+
     status_colors = {
         "completed": "green",
         "paused": "yellow",
@@ -4244,13 +4276,19 @@ def workflow_run(
 @workflow_app.command("resume")
 def workflow_resume(
     run_id: str = typer.Argument(..., help="Run ID to resume"),
+    json_output: bool = typer.Option(
+        False,
+        "--json",
+        help="Emit the resume outcome as a single JSON object instead of formatted text.",
+    ),
 ):
     """Resume a paused or failed workflow run."""
     from .workflows.engine import WorkflowEngine
 
     project_root = _require_specify_project()
     engine = WorkflowEngine(project_root)
-    engine.on_step_start = lambda sid, label: console.print(f"  \u25b8 [{sid}] {label} \u2026")
+    if not json_output:
+        engine.on_step_start = lambda sid, label: console.print(f"  \u25b8 [{sid}] {label} \u2026")
 
     try:
         state = engine.resume(run_id)
@@ -4264,6 +4302,10 @@ def workflow_resume(
         console.print(f"[red]Resume failed:[/red] {exc}")
         raise typer.Exit(1)
 
+    if json_output:
+        _emit_workflow_json(_workflow_run_payload(state))
+        return
+
     status_colors = {
         "completed": "green",
         "paused": "yellow",
@@ -4277,6 +4319,11 @@ def workflow_resume(
 @workflow_app.command("status")
 def workflow_status(
     run_id: str | None = typer.Argument(None, help="Run ID to inspect (shows all if omitted)"),
+    json_output: bool = typer.Option(
+        False,
+        "--json",
+        help="Emit run status as a single JSON object instead of formatted text.",
+    ),
 ):
     """Show workflow run status."""
     from .workflows.engine import WorkflowEngine
@@ -4292,6 +4339,22 @@ def workflow_status(
             console.print(f"[red]Error:[/red] Run not found: {run_id}")
             raise typer.Exit(1)
 
+        if json_output:
+            payload = {
+                "run_id": state.run_id,
+                "workflow_id": state.workflow_id,
+                "status": state.status.value,
+                "created_at": state.created_at,
+                "updated_at": state.updated_at,
+                "current_step_id": state.current_step_id,
+                "steps": {
+                    sid: sd.get("status", "unknown")
+                    for sid, sd in state.step_results.items()
+                },
+            }
+            _emit_workflow_json(payload)
+            return
+
         status_colors = {
             "completed": "green",
             "paused": "yellow",
@@ -4319,6 +4382,22 @@ def workflow_status(
                 console.print(f"    [{sc}]●[/{sc}] {step_id}: {s}")
     else:
         runs = engine.list_runs()
+
+        if json_output:
+            payload = {
+                "runs": [
+                    {
+                        "run_id": r["run_id"],
+                        "workflow_id": r.get("workflow_id"),
+                        "status": r.get("status", "unknown"),
+                        "updated_at": r.get("updated_at"),
+                    }
+                    for r in runs
+                ]
+            }
+            _emit_workflow_json(payload)
+            return
+
         if not runs:
             console.print("[yellow]No workflow runs found.[/yellow]")
             return

tests/test_workflows.py

@@ -2726,3 +2726,117 @@ def test_switch_workflow(self, project_dir):
         assert state.status == RunStatus.COMPLETED
         assert "do-plan" in state.step_results
         assert "do-specify" not in state.step_results
+
+
+class TestWorkflowJsonOutput:
+    """Test the --json machine-readable output for run/resume/status."""
+
+    _WF = """
+schema_version: "1.0"
+workflow:
+  id: "json-wf"
+  name: "JSON WF"
+  version: "1.0.0"
+steps:
+  - id: ask
+    type: gate
+    message: "Review"
+    options: [approve, reject]
+  - id: after
+    type: shell
+    run: "echo done"
+"""
+
+    _WF_DONE = """
+schema_version: "1.0"
+workflow:
+  id: "json-done"
+  name: "JSON Done"
+  version: "1.0.0"
+steps:
+  - id: only
+    type: shell
+    run: "echo done"
+"""
+
+    def _write_wf(self, project_dir, text, name):
+        path = project_dir / f"{name}.yml"
+        path.write_text(text, encoding="utf-8")
+        return path
+
+    def _invoke(self, project_dir, args):
+        from typer.testing import CliRunner
+        from unittest.mock import patch
+        from specify_cli import app
+
+        runner = CliRunner()
+        with patch.object(Path, "cwd", return_value=project_dir):
+            return runner.invoke(app, args, catch_exceptions=False)
+
+    def test_run_json_completed(self, project_dir):
+        wf = self._write_wf(project_dir, self._WF_DONE, "done")
+        result = self._invoke(project_dir, ["workflow", "run", str(wf), "--json"])
+        assert result.exit_code == 0
+        payload = json.loads(result.stdout)
+        assert payload["workflow_id"] == "json-done"
+        assert payload["status"] == "completed"
+        assert "run_id" in payload
+
+    def test_run_json_paused(self, project_dir):
+        wf = self._write_wf(project_dir, self._WF, "gated")
+        result = self._invoke(project_dir, ["workflow", "run", str(wf), "--json"])
+        assert result.exit_code == 0
+        payload = json.loads(result.stdout)
+        assert payload["status"] == "paused"
+        assert payload["current_step_id"] == "ask"
+        assert payload["current_step_index"] == 0
+
+    def test_run_json_output_has_no_markup_or_ansi(self, project_dir):
+        wf = self._write_wf(project_dir, self._WF_DONE, "clean")
+        out = self._invoke(
+            project_dir, ["workflow", "run", str(wf), "--json"]
+        ).stdout
+        # Machine output must be exactly the JSON object: no Rich markup
+        # tags and no ANSI escape sequences leaking in.
+        assert "\x1b[" not in out
+        assert "[/" not in out
+        assert out.strip() == json.dumps(json.loads(out), indent=2)
+
+    def test_run_default_output_is_human_not_json(self, project_dir):
+        wf = self._write_wf(project_dir, self._WF_DONE, "done2")
+        result = self._invoke(project_dir, ["workflow", "run", str(wf)])
+        assert result.exit_code == 0
+        assert "Running workflow" in result.stdout
+        with pytest.raises(json.JSONDecodeError):
+            json.loads(result.stdout)
+
+    def test_status_json_single_and_list(self, project_dir):
+        wf = self._write_wf(project_dir, self._WF, "gated2")
+        run = json.loads(
+            self._invoke(project_dir, ["workflow", "run", str(wf), "--json"]).stdout
+        )
+        rid = run["run_id"]
+
+        single = json.loads(
+            self._invoke(project_dir, ["workflow", "status", rid, "--json"]).stdout
+        )
+        assert single["run_id"] == rid
+        assert single["status"] == "paused"
+        assert single["steps"]["ask"] == "paused"
+
+        listing = json.loads(
+            self._invoke(project_dir, ["workflow", "status", "--json"]).stdout
+        )
+        assert any(r["run_id"] == rid for r in listing["runs"])
+
+    def test_resume_json(self, project_dir):
+        wf = self._write_wf(project_dir, self._WF, "gated3")
+        rid = json.loads(
+            self._invoke(project_dir, ["workflow", "run", str(wf), "--json"]).stdout
+        )["run_id"]
+        # Non-interactive resume re-runs the gate, which pauses again.
+        resumed = json.loads(
+            self._invoke(project_dir, ["workflow", "resume", rid, "--json"]).stdout
+        )
+        assert resumed["run_id"] == rid
+        assert resumed["status"] == "paused"