Fail non-generator workflow construction exceptions

Author: durable-workflow-ops

src/durable_workflow/workflow.py

@@ -2825,7 +2825,12 @@ def _same_logical_condition_wait(opened: Mapping[str, Any], cmd: WaitCondition)
     result_cursor = 0
     pending_step_cursor = 0
     wait_yield_count = 0
-    gen = instance.run(ctx, *start_input)
+    try:
+        gen = instance.run(ctx, *start_input)
+    except NonDeterministicReplayError:
+        raise
+    except Exception as exc:
+        return _state([_fail_workflow_from_exception(exc)])
     if not hasattr(gen, "__next__"):
         if isinstance(gen, ContinueAsNew):
             _assert_no_unconsumed_history("continue as new")

tests/test_replay.py

@@ -155,6 +155,18 @@ def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
         raise ValueError("something went wrong")
 
 
+@workflow.defn(name="construction-failing-workflow")
+class ConstructionFailingWorkflow:
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        raise RuntimeError("failed before first yield")
+
+
+@workflow.defn(name="construction-stop-iteration-workflow")
+class ConstructionStopIterationWorkflow:
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        raise StopIteration("stopped before first yield")
+
+
 @workflow.defn(name="typed-compensation-failure-workflow")
 class TypedCompensationFailureWorkflow:
     def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
@@ -848,6 +860,24 @@ def test_exception_produces_fail_command(self) -> None:
         assert cmd.exception_type == "ValueError"
         assert cmd.exception_class == "builtins.ValueError"
 
+    def test_non_generator_exception_produces_fail_command(self) -> None:
+        outcome = replay(ConstructionFailingWorkflow, [], [])
+        assert len(outcome.commands) == 1
+        cmd = outcome.commands[0]
+        assert isinstance(cmd, FailWorkflow)
+        assert "failed before first yield" in cmd.message
+        assert cmd.exception_type == "RuntimeError"
+        assert cmd.exception_class == "builtins.RuntimeError"
+
+    def test_non_generator_stop_iteration_is_not_completion(self) -> None:
+        outcome = replay(ConstructionStopIterationWorkflow, [], [])
+        assert len(outcome.commands) == 1
+        cmd = outcome.commands[0]
+        assert isinstance(cmd, FailWorkflow)
+        assert "stopped before first yield" in cmd.message
+        assert cmd.exception_type == "StopIteration"
+        assert cmd.exception_class == "builtins.StopIteration"
+
     def test_fail_server_command_shape(self) -> None:
         history = [{"event_type": "ActivityCompleted", "payload": {"result": '"ok"'}}]
         outcome = replay(FailingWorkflow, history, [])