fix(workflows): route run/resume errors to stderr under --json (#3352)

workflow run/resume --json is contracted to emit a single JSON object on
stdout, but every error path (workflow-not-found, invalid workflow,
validation failure, execute/resume failure, and the shared
_parse_input_values invalid-input error) used console.print, landing the
human error text on stdout and corrupting the machine-readable stream.
Route those messages through err_console when --json is set (a no-op for
normal text mode), mirroring the stderr-only error routing already used
by 'specify bundle' (_fail) and err_console elsewhere in this module.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

src/specify_cli/workflows/_commands.py

@@ -50,7 +50,17 @@ workflow_step_catalog_app = typer.Typer(
 workflow_step_app.add_typer(workflow_step_catalog_app, name="catalog")
 
 
-def _parse_input_values(input_values: list[str] | None) -> dict[str, Any]:
+def _error_console(json_output: bool):
+    """Console for error text: stderr under ``--json`` so the JSON stdout
+    stream stays parseable, the normal console otherwise. Mirrors the
+    stderr-only error routing already used by ``specify bundle``.
+    """
+    return err_console if json_output else console
+
+
+def _parse_input_values(
+    input_values: list[str] | None, *, json_output: bool = False
+) -> dict[str, Any]:
     """Parse repeated ``key=value`` CLI inputs into a dict.
 
     Shared by ``workflow run`` and ``workflow resume``. Exits with an error
@@ -59,7 +69,9 @@ def _parse_input_values(input_values: list[str] | None) -> dict[str, Any]:
     inputs: dict[str, Any] = {}
     for kv in input_values or []:
         if "=" not in kv:
-            console.print(f"[red]Error:[/red] Invalid input format: {kv!r} (expected key=value)")
+            _error_console(json_output).print(
+                f"[red]Error:[/red] Invalid input format: {kv!r} (expected key=value)"
+            )
             raise typer.Exit(1)
         key, _, value = kv.partition("=")
         inputs[key.strip()] = value.strip()
@@ -335,25 +347,26 @@ def workflow_run(
     if not json_output:
         engine.on_step_start = lambda sid, label: console.print(f"  \u25b8 [{sid}] {label} \u2026")
 
+    err = _error_console(json_output)
     try:
         definition = engine.load_workflow(source_path if is_file_source else source)
     except FileNotFoundError:
-        console.print(f"[red]Error:[/red] Workflow not found: {source}")
+        err.print(f"[red]Error:[/red] Workflow not found: {source}")
         raise typer.Exit(1)
     except ValueError as exc:
-        console.print(f"[red]Error:[/red] Invalid workflow: {exc}")
+        err.print(f"[red]Error:[/red] Invalid workflow: {exc}")
         raise typer.Exit(1)
 
     # Validate
     errors = engine.validate(definition)
     if errors:
-        console.print("[red]Workflow validation failed:[/red]")
-        for err in errors:
-            console.print(f"  • {err}")
+        err.print("[red]Workflow validation failed:[/red]")
+        for verr in errors:
+            err.print(f"  • {verr}")
         raise typer.Exit(1)
 
     # Parse inputs
-    inputs = _parse_input_values(input_values)
+    inputs = _parse_input_values(input_values, json_output=json_output)
 
     if not json_output:
         console.print(f"\n[bold cyan]Running workflow:[/bold cyan] {definition.name} ({definition.id})")
@@ -363,10 +376,10 @@ def workflow_run(
         with _stdout_to_stderr_when(json_output):
             state = engine.execute(definition, inputs)
     except ValueError as exc:
-        console.print(f"[red]Error:[/red] {exc}")
+        err.print(f"[red]Error:[/red] {exc}")
         raise typer.Exit(1)
     except Exception as exc:
-        console.print(f"[red]Workflow failed:[/red] {exc}")
+        err.print(f"[red]Workflow failed:[/red] {exc}")
         raise typer.Exit(1)
 
     if json_output:
@@ -411,19 +424,20 @@ def workflow_resume(
     if not json_output:
         engine.on_step_start = lambda sid, label: console.print(f"  \u25b8 [{sid}] {label} \u2026")
 
-    inputs = _parse_input_values(input_values)
+    inputs = _parse_input_values(input_values, json_output=json_output)
+    err = _error_console(json_output)
 
     try:
         with _stdout_to_stderr_when(json_output):
             state = engine.resume(run_id, inputs or None)
     except FileNotFoundError:
-        console.print(f"[red]Error:[/red] Run not found: {run_id}")
+        err.print(f"[red]Error:[/red] Run not found: {run_id}")
         raise typer.Exit(1)
     except ValueError as exc:
-        console.print(f"[red]Error:[/red] {exc}")
+        err.print(f"[red]Error:[/red] {exc}")
         raise typer.Exit(1)
     except Exception as exc:
-        console.print(f"[red]Resume failed:[/red] {exc}")
+        err.print(f"[red]Resume failed:[/red] {exc}")
         raise typer.Exit(1)
 
     if json_output:

tests/test_workflow_run_without_project.py

@@ -322,3 +322,87 @@ class TestWorkflowRunWithoutProject:
 
         assert result.exit_code != 0
         assert ".specify path exists but is not a directory" in result.output
+
+
+class TestWorkflowRunJsonErrorStream:
+    """Under --json, error text must go to stderr so stdout stays parseable."""
+
+    def _bad_workflow(self, tmp_path):
+        wf = tmp_path / "bad.yml"
+        wf.write_text(
+            yaml.dump(
+                {
+                    "schema_version": "1.0",
+                    "workflow": {
+                        "id": "bad-wf",
+                        "name": "Bad",
+                        "version": "1.0.0",
+                        "description": "fails validation",
+                    },
+                    # shell step missing required 'run' -> validation error
+                    "steps": [{"id": "s", "type": "shell"}],
+                }
+            ),
+            encoding="utf-8",
+        )
+        return wf
+
+    def test_run_json_validation_error_not_on_stdout(self, tmp_path):
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        wf = self._bad_workflow(tmp_path)
+        runner = CliRunner()
+        old = os.getcwd()
+        try:
+            os.chdir(tmp_path)
+            result = runner.invoke(
+                app, ["workflow", "run", str(wf), "--json"], catch_exceptions=False
+            )
+        finally:
+            os.chdir(old)
+
+        assert result.exit_code == 1
+        # stdout must carry only JSON (here: nothing) — never human error text.
+        assert "validation failed" not in result.stdout
+        assert "Error" not in result.stdout
+        # The message is routed to stderr instead.
+        assert "validation failed" in result.stderr
+
+    def test_run_json_invalid_input_not_on_stdout(self, tmp_path):
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        # A valid single-shell workflow so we get past load/validate to
+        # _parse_input_values, which rejects the malformed --input.
+        wf = tmp_path / "ok.yml"
+        wf.write_text(
+            yaml.dump(
+                {
+                    "schema_version": "1.0",
+                    "workflow": {
+                        "id": "ok-wf",
+                        "name": "OK",
+                        "version": "1.0.0",
+                        "description": "x",
+                    },
+                    "steps": [{"id": "s", "type": "shell", "run": "echo hi"}],
+                }
+            ),
+            encoding="utf-8",
+        )
+        runner = CliRunner()
+        old = os.getcwd()
+        try:
+            os.chdir(tmp_path)
+            result = runner.invoke(
+                app,
+                ["workflow", "run", str(wf), "--json", "--input", "no-equals"],
+                catch_exceptions=False,
+            )
+        finally:
+            os.chdir(old)
+
+        assert result.exit_code == 1
+        assert "Invalid input format" not in result.stdout
+        assert "Invalid input format" in result.stderr