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). JSON is written with plain stdout so it is never
altered by Rich markup, highlighting, or wrapping.

Default human-readable output is unchanged when `--json` is omitted,
and exit codes are unchanged. Mirrors the existing `specify version
--features --json` precedent.

Closes #2811.

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

Author: doquanghuy

src/specify_cli/__init__.py

@@ -4174,19 +4174,36 @@ 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,
+    }
+
+
 @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 +4232,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 +4245,10 @@ def workflow_run(
         console.print(f"[red]Workflow failed:[/red] {exc}")
         raise typer.Exit(1)
 
+    if json_output:
+        print(json.dumps(_workflow_run_payload(state), indent=2))
+        return
+
     status_colors = {
         "completed": "green",
         "paused": "yellow",
@@ -4244,13 +4266,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 +4292,10 @@ def workflow_resume(
         console.print(f"[red]Resume failed:[/red] {exc}")
         raise typer.Exit(1)
 
+    if json_output:
+        print(json.dumps(_workflow_run_payload(state), indent=2))
+        return
+
     status_colors = {
         "completed": "green",
         "paused": "yellow",
@@ -4277,6 +4309,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 +4329,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()
+                },
+            }
+            print(json.dumps(payload, indent=2))
+            return
+
         status_colors = {
             "completed": "green",
             "paused": "yellow",
@@ -4319,6 +4372,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
+                ]
+            }
+            print(json.dumps(payload, indent=2))
+            return
+
         if not runs:
             console.print("[yellow]No workflow runs found.[/yellow]")
             return

tests/test_workflows.py

@@ -2726,3 +2726,105 @@ 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"
+
+    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"