Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions src/specify_cli/workflows/_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,24 @@ def _fail_workflow_execution(
raise typer.Exit(1) from exc


def _fail_recorded_run(
exc: Exception, *, json_output: bool, state: Any, resume: bool = False,
) -> None:
"""Report *exc* as the run's outcome when the engine recorded it as one.
``WorkflowEngine.execute``/``resume`` persist an exception raised while
steps run (an expression error, say) as the run's ``failed`` status and
error, and mark that exception on the in-memory state, before re-raising
it. Errors raised before that point, such as input validation or a refused
resume, are not marked even when a loaded failed run carries the same
message, so this returns and the caller reports them as before.
"""
if state is not None and getattr(state, "_recorded_exception", None) is exc:
_fail_workflow_execution(
exc, json_output=json_output, state=state, resume=resume
)


def _open_workflow_registry(project_root: Path, out=None):
"""Construct a WorkflowRegistry, exiting cleanly on an unreadable file.
Expand Down
6 changes: 6 additions & 0 deletions src/specify_cli/workflows/command_resume.py
Original file line number Diff line number Diff line change
Expand Up @@ -110,9 +110,15 @@ def workflow_resume(
exc, json_output=json_output, state=engine._execution_state.get(), resume=True,
)
except ValueError as exc:
cli._fail_recorded_run(
exc, json_output=json_output, state=engine._execution_state.get(), resume=True,
)
err.print(f"[red]Error:[/red] {cli._escape_markup(str(exc))}")
raise cli.typer.Exit(1)
except Exception as exc:
cli._fail_recorded_run(
exc, json_output=json_output, state=engine._execution_state.get(), resume=True,
)
err.print(f"[red]Resume failed:[/red] {cli._escape_markup(str(exc))}")
raise cli.typer.Exit(1)

Expand Down
2 changes: 2 additions & 0 deletions src/specify_cli/workflows/command_run.py
Original file line number Diff line number Diff line change
Expand Up @@ -159,9 +159,11 @@ def workflow_run(
except OSError as exc:
cli._fail_workflow_execution(exc, json_output=json_output, state=engine._execution_state.get())
except ValueError as exc:
cli._fail_recorded_run(exc, json_output=json_output, state=engine._execution_state.get())
err.print(f"[red]Error:[/red] {cli._escape_markup(str(exc))}")
raise cli.typer.Exit(1)
except Exception as exc:
cli._fail_recorded_run(exc, json_output=json_output, state=engine._execution_state.get())
err.print(f"[red]Workflow failed:[/red] {cli._escape_markup(str(exc))}")
raise cli.typer.Exit(1)

Expand Down
5 changes: 5 additions & 0 deletions src/specify_cli/workflows/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -714,6 +714,9 @@ def __init__(
self.step_results: dict[str, dict[str, Any]] = {}
self.execution: dict[str, Any] | None = None
self._checkpoint_failed = False
# The exception this process just recorded as the run's failure (never
# persisted), so the CLI can tell it apart from an older failure.
self._recorded_exception: BaseException | None = None
# Guards step_results mutation and save() so a concurrent fan-out cannot
# mutate the dict while save() is serializing it (which would raise
# "dictionary changed size during iteration").
Expand Down Expand Up @@ -1148,6 +1151,7 @@ def execute(
raise
state.status = RunStatus.FAILED
state.error = str(exc)
state._recorded_exception = exc
state.append_log({"event": "workflow_failed", "error": str(exc)})
state.save()
raise
Expand Down Expand Up @@ -1262,6 +1266,7 @@ def resume(
raise
state.status = RunStatus.FAILED
state.error = str(exc)
state._recorded_exception = exc
state.append_log({"event": "resume_failed", "error": str(exc)})
state.save()
raise
Expand Down
112 changes: 112 additions & 0 deletions tests/specify_cli/workflows/test_command_resume.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,50 @@ class TestWorkflowJsonOutput:
run: "echo done"
"""

_WF_RAISE = """
schema_version: "1.0"
workflow:
id: "json-raise"
name: "JSON Raise"
version: "1.0.0"
inputs:
verdict:
type: string
default: ""
payload:
type: string
default: "[]"
steps:
- id: ask
type: gate
message: "Review"
options: [approve, reject]
verdict_input: verdict
- id: parse
type: if
condition: "{{ inputs.payload | from_json | length }}"
then:
- id: after
type: shell
run: "echo parsed"
"""

_WF_FAIL = """
schema_version: "1.0"
workflow:
id: "json-fail"
name: "JSON Fail"
version: "1.0.0"
inputs:
count:
type: number
default: 1
steps:
- id: boom
type: shell
run: "exit 3"
"""

def _write_wf(self, project_dir, text, name):
path = project_dir / f"{name}.yml"
path.write_text(text, encoding="utf-8")
Expand All @@ -56,6 +100,74 @@ def test_resume_json(self, project_dir):
assert resumed["run_id"] == rid
assert resumed["status"] == "paused"

def test_resume_json_step_exception_reports_failed_run(self, project_dir):
wf = self._write_wf(project_dir, self._WF_RAISE, "raise")
rid = json.loads(
self._invoke(project_dir, ["workflow", "run", str(wf), "--json"]).stdout
)["run_id"]
# The approved gate lets the run reach `parse`, whose `from_json`
# raises on the supplied payload and fails the resumed run.
result = self._invoke(
project_dir,
[
"workflow", "resume", rid, "--json",
"--input", "verdict=approve", "--input", "payload=not json",
],
)
assert result.exit_code == 1
payload = json.loads(result.stdout)
assert payload["run_id"] == rid
assert payload["status"] == "failed"
assert payload["current_step_id"] == "parse"
assert payload["error"].startswith("from_json: invalid JSON")
assert result.stderr == ""

def test_resume_json_refused_before_running_is_not_a_run_outcome(
self, project_dir
):
from specify_cli.workflows.engine import RunState

wf = self._write_wf(project_dir, self._WF_FAIL, "fail")
failed = json.loads(
self._invoke(project_dir, ["workflow", "run", str(wf), "--json"]).stdout
)
assert failed["status"] == "failed"
# Input validation rejects the resume before the run continues, so
# the run's recorded failure is not this error and is not reported.
result = self._invoke(
project_dir,
["workflow", "resume", failed["run_id"], "--json", "--input", "count=abc"],
)
assert result.exit_code == 1
assert result.stdout == ""
assert "Input 'count' expected a number" in result.stderr
state = RunState.load(failed["run_id"], project_dir)
assert state.error == failed["error"]

def test_resume_json_refused_with_matching_old_error_is_not_a_run_outcome(
self, project_dir
):
"""A refused resume is not reported as the run's outcome even when the
previously recorded failure has exactly the same message."""
from specify_cli.workflows.engine import RunState

wf = self._write_wf(project_dir, self._WF_FAIL, "fail")
failed = json.loads(
self._invoke(project_dir, ["workflow", "run", str(wf), "--json"]).stdout
)
message = "Input 'count' expected a number, got 'abc'."
state = RunState.load(failed["run_id"], project_dir)
state.error = message
state.save()

result = self._invoke(
project_dir,
["workflow", "resume", failed["run_id"], "--json", "--input", "count=abc"],
)
assert result.exit_code == 1
assert result.stdout == ""
assert message in result.stderr



class TestResumeWithInputs:
Expand Down
95 changes: 95 additions & 0 deletions tests/specify_cli/workflows/test_command_run.py
Original file line number Diff line number Diff line change
Expand Up @@ -288,6 +288,42 @@ class TestWorkflowJsonOutput:
run: "exit 3"
"""

_WF_RAISE = """
schema_version: "1.0"
workflow:
id: "json-raise"
name: "JSON Raise"
version: "1.0.0"
inputs:
payload:
type: string
default: "not json"
steps:
- id: parse
type: if
condition: "{{ inputs.payload | from_json | length }}"
then:
- id: after
type: shell
run: "echo parsed"
"""

_WF_REQUIRED = """
schema_version: "1.0"
workflow:
id: "json-required"
name: "JSON Required"
version: "1.0.0"
inputs:
spec:
type: string
required: true
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")
Expand Down Expand Up @@ -331,6 +367,65 @@ def test_run_json_failed_includes_error(self, project_dir):
assert payload["status"] == "failed"
assert payload.get("error")

def test_run_json_step_exception_reports_failed_run(self, project_dir):
# A step that raises (here `from_json` on invalid JSON) also fails the
# run: the engine persists `failed` and the error before re-raising, so
# stdout must carry that run's payload rather than nothing at all.
wf = self._write_wf(project_dir, self._WF_RAISE, "raise")
result = self._invoke(project_dir, ["workflow", "run", str(wf), "--json"])
assert result.exit_code == 1
payload = json.loads(result.stdout)
assert payload["workflow_id"] == "json-raise"
assert payload["status"] == "failed"
assert payload["current_step_id"] == "parse"
assert payload["error"].startswith("from_json: invalid JSON")
assert result.stderr == ""
state_file = (
project_dir / ".specify" / "workflows" / "runs"
/ payload["run_id"] / "state.json"
)
state = json.loads(state_file.read_text(encoding="utf-8"))
assert state["status"] == "failed"
assert state["error"] == payload["error"]

def test_run_json_non_value_error_reports_failed_run(
self, project_dir, monkeypatch
):
from specify_cli.workflows.step.shell import ShellStep

def crash(self, config, context):
raise RuntimeError("shell step crashed")

monkeypatch.setattr(ShellStep, "execute", crash)
wf = self._write_wf(project_dir, self._WF_DONE, "crash")
result = self._invoke(project_dir, ["workflow", "run", str(wf), "--json"])
assert result.exit_code == 1
payload = json.loads(result.stdout)
assert payload["status"] == "failed"
assert payload["current_step_id"] == "only"
assert payload["error"] == "shell step crashed"
assert result.stderr == ""

def test_run_step_exception_reports_failed_run_in_text(self, project_dir):
wf = self._write_wf(project_dir, self._WF_RAISE, "raise-text")
result = self._invoke(project_dir, ["workflow", "run", str(wf)])
assert result.exit_code == 1
assert "Workflow failed:" in result.stdout
assert "from_json: invalid JSON" in result.stdout

def test_run_json_missing_required_input_is_not_a_run_outcome(
self, project_dir
):
# Input resolution fails before any run state is saved, so there is no
# run to report: stdout stays empty and the error goes to stderr.
wf = self._write_wf(project_dir, self._WF_REQUIRED, "required")
result = self._invoke(project_dir, ["workflow", "run", str(wf), "--json"])
assert result.exit_code == 1
assert result.stdout == ""
assert "Required input 'spec' not provided." in result.stderr
runs_dir = project_dir / ".specify" / "workflows" / "runs"
assert list(runs_dir.glob("*/state.json")) == []

def test_run_json_completed_omits_error(self, project_dir):
# Successful runs must not carry an `error` key at all.
wf = self._write_wf(project_dir, self._WF_DONE, "noerr")
Expand Down
Loading