From a15756835ca8e01d4597711ab1a0a7eef97f6497 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Tue, 7 Jul 2026 08:30:43 +0800 Subject: [PATCH] fix(workflow-orchestration): correct resume and failure-branch routing in WorkflowRunner - Resume no longer infinite-loops when a crash occurs between append_step_result() and update_run() for a BranchStep: the recorded branch target is now read from the persisted result instead of recomputed as None. - Resume no longer silently discards a recorded step failure that happened right before the crash; the run correctly ends failed. - A failed step's own next_step_id (e.g. routing to a BranchStep that evaluates step_result_success) is now honored instead of unconditionally forcing the run to failed, per spec. openspec: workflow-orchestration capability, archived change workflow-orchestration-runtime --- tests/test_workflow_runner.py | 140 ++++++++++++++++++++++++++++++++++ workflow/runner.py | 48 +++++++++--- 2 files changed, 176 insertions(+), 12 deletions(-) diff --git a/tests/test_workflow_runner.py b/tests/test_workflow_runner.py index b341dd9..da2fccd 100644 --- a/tests/test_workflow_runner.py +++ b/tests/test_workflow_runner.py @@ -14,6 +14,7 @@ from workflow.models import ( SkillInvocationStep, WaitForConditionStep, WorkflowDefinition, + WorkflowStepResult, ) from workflow.runner import WorkflowRunner from workflow.store import WorkflowStore @@ -266,6 +267,145 @@ def test_workflow_runner_resume_skips_completed_steps_after_restart(tmp_path) -> assert calls == ["one", "two", "three"] +def test_workflow_runner_resume_after_branch_crash_advances_to_recorded_target( + tmp_path, +) -> None: + """Simulates a crash between append_step_result() and update_run() for a + BranchStep: the step's result is persisted but the checkpoint never + happened, so current_step_id still points at the branch step itself. + Resuming must recompute the next step from the *recorded* branch + target instead of losing it (and must not loop forever re-processing + the branch step). + """ + calls: list[str] = [] + store = _store(tmp_path) + definition = WorkflowDefinition( + name="branch-resume", + entry_step_id="branch", + steps=[ + BranchStep( + "branch", + ConditionSpec("world_variable_equals", {"name": "ready", "value": True}), + on_true="true-step", + on_false="false-step", + ), + PlannedGoalStep("false-step", "false path"), + PlannedGoalStep("true-step", "true path"), + ], + ) + store.save_definition(definition) + run = store.create_run(definition.id, {"ready": True}, device_id="phone") + store.append_step_result( + run.id, + WorkflowStepResult( + step_id="branch", + kind="branch", + success=True, + detail={"condition_result": True, "next_step_id": "true-step"}, + ), + ) + + runner = WorkflowRunner( + store, + task_runner_factory=lambda: FakeTaskRunner(calls), + ) + resumed = runner.resume(run.id) + + assert resumed.status == "completed" + assert calls == ["true path"] + assert [result.step_id for result in resumed.step_results] == [ + "branch", + "true-step", + ] + + +def test_workflow_runner_resume_after_failed_step_crash_ends_failed(tmp_path) -> None: + """Simulates a crash between append_step_result(success=False) and + update_run(): the failed result is persisted but current_step_id still + points at the failed step. Resuming must recognize the recorded + failure and end the run as failed rather than silently continuing (or + looping forever re-processing the same step). + """ + store = _store(tmp_path) + definition = WorkflowDefinition( + name="fail-resume", + entry_step_id="only", + steps=[PlannedGoalStep("only", "risky")], + ) + store.save_definition(definition) + run = store.create_run(definition.id, {}, device_id="phone") + store.append_step_result( + run.id, + WorkflowStepResult( + step_id="only", + kind="planned_goal", + success=False, + detail={"task_status": "failed", "failure_reason": "boom"}, + ), + ) + + runner = WorkflowRunner(store) + resumed = runner.resume(run.id) + + assert resumed.status == "failed" + assert resumed.current_step_id is None + + +def test_workflow_runner_failed_step_routes_to_recovery_branch(tmp_path) -> None: + """A planned-goal step that fails still routes to its declared + next_step_id (a branch step evaluating step_result_success). Live + execution must reach and execute the on_false recovery path rather + than immediately terminating the run as failed. + """ + calls: list[str] = [] + + class ScriptedTaskRunner: + def __init__(self, calls: list[str], failing_goals: set[str]) -> None: + self.calls = calls + self.failing_goals = failing_goals + + def run(self, task: Task) -> Task: + self.calls.append(task.goal) + if task.goal in self.failing_goals: + task.status = "failed" # type: ignore[assignment] + task.failure_reason = "boom" + else: + task.status = "completed" # type: ignore[assignment] + task.failure_reason = None + return task + + definition = WorkflowDefinition( + name="failure-recovery", + entry_step_id="goal", + steps=[ + PlannedGoalStep("goal", "risky goal", next_step_id="check"), + BranchStep( + "check", + ConditionSpec("step_result_success", {"step_id": "goal"}), + on_true="happy", + on_false="recover", + ), + PlannedGoalStep("happy", "happy path"), + PlannedGoalStep("recover", "recover path"), + ], + ) + runner = WorkflowRunner( + _store(tmp_path), + task_runner_factory=lambda: ScriptedTaskRunner(calls, {"risky goal"}), + ) + + run = runner.run(definition, "phone") + + assert run.status == "completed" + assert calls == ["risky goal", "recover path"] + assert [result.step_id for result in run.step_results] == [ + "goal", + "check", + "recover", + ] + assert run.step_results[0].success is False + + def test_workflow_run_persists_variables_and_ordered_results(tmp_path) -> None: store = _store(tmp_path) definition = WorkflowDefinition( diff --git a/workflow/runner.py b/workflow/runner.py index 4eaf163..146a085 100644 --- a/workflow/runner.py +++ b/workflow/runner.py @@ -98,19 +98,21 @@ class WorkflowRunner: if self.step_limit is not None and executed >= self.step_limit: return run step = definition.step_by_id(run.current_step_id) - if _already_recorded(run, step.step_id): - next_step_id = self._next_step_id(definition, step, None) - run = self._checkpoint(run, next_step_id, "running") + recorded = _recorded_result(run, step.step_id) + if recorded is not None: + branch_next_step_id = None + if isinstance(step, BranchStep): + branch_next_step_id = recorded.detail.get("next_step_id") + next_status, next_step_id = self._resolve_outcome( + definition, step, recorded.success, branch_next_step_id + ) + run = self._checkpoint(run, next_step_id, next_status) continue result, branch_next_step_id = self._execute_step(definition, run, step) - next_status = "running" - next_step_id = self._next_step_id(definition, step, branch_next_step_id) - if not result.success: - next_status = "failed" - next_step_id = None - elif next_step_id is None: - next_status = "completed" + next_status, next_step_id = self._resolve_outcome( + definition, step, result.success, branch_next_step_id + ) self.store.append_step_result(run.id, result) run = self._checkpoint( @@ -121,6 +123,25 @@ class WorkflowRunner: executed += 1 return run + def _resolve_outcome( + self, + definition: WorkflowDefinition, + step: WorkflowStep, + success: bool, + branch_next_step_id: str | None, + ) -> tuple[str, str | None]: + """Compute the (status, next_step_id) for a step's outcome. + + Shared by live execution and resume so a failed step's routing + (e.g. to a branch step evaluating the failure) can never drift + between the two code paths. The run only becomes terminal when + there is truly no next step to route to. + """ + next_step_id = self._next_step_id(definition, step, branch_next_step_id) + if next_step_id is None: + return ("completed" if success else "failed"), None + return "running", next_step_id + def _checkpoint( self, run: WorkflowRun, @@ -317,5 +338,8 @@ class WorkflowRunner: return definition.next_step_id_after(step.step_id) -def _already_recorded(run: WorkflowRun, step_id: str) -> bool: - return any(result.step_id == step_id for result in run.step_results) +def _recorded_result(run: WorkflowRun, step_id: str) -> WorkflowStepResult | None: + for result in reversed(run.step_results): + if result.step_id == step_id: + return result + return None