fix(wac): report a task failure the child round's body catches (#10366)

* fix(wac): report a task failure the child round's body catches

* chore(wac): state the child-round failure invariant once

* test(wac): pin the catch-then-continue re-raise in the child round

python-client/wmill/tests/test_workflow.py

@@ -406,6 +406,24 @@ class TestChildMode:
         assert result["type"] == "complete"
         assert result["result"] == 14
 
+    def test_child_cannot_swallow_the_failure_of_the_step_it_executes(self):
+        # If `except Exception` could catch it, the child would report a success
+        # returning "swallowed" and the parent would record that as the step's value.
+        @task
+        async def boom():
+            raise ValueError("nope")
+
+        @workflow
+        async def wf():
+            try:
+                await boom()
+            except Exception:
+                return "swallowed"
+            return "unreachable"
+
+        with pytest.raises(ValueError, match="nope"):
+            _run_workflow(wf, {"_executing_key": "boom"}, {})
+
     def test_child_replays_cached_steps(self):
         checkpoint = {
             "completed_steps": {"extract_data": {"data": [1, 2, 3]}},

python-client/wmill/wmill/client.py

@@ -2673,6 +2673,19 @@ class _StepSuspend(BaseException):
         self.dispatch_info = dispatch_info
 
 
+class _StepFailure(BaseException):
+    """Carries the exception raised by the step a child round executes directly.
+
+    That exception *is* the round's result, so a broad ``except Exception`` in the
+    body must not be able to turn it into a successful complete — the parent would
+    then record the caught branch's value as the step result. BaseException for the
+    same reason ``_StepSuspend`` is; a bare ``except:`` still swallows both.
+    """
+
+    def __init__(self, exc: BaseException):
+        self.exc = exc
+
+
 class TaskError(Exception):
     """Raised when a WAC task step failed.
 
@@ -2799,9 +2812,12 @@ class WorkflowCtx:
         return value
 
     async def _execute_directly(self, func, **kwargs):
-        result = func(**kwargs)
-        if _asyncio.iscoroutine(result):
-            result = await result
+        try:
+            result = func(**kwargs)
+            if _asyncio.iscoroutine(result):
+                result = await result
+        except Exception as exc:
+            raise _StepFailure(exc) from exc
         raise _StepSuspend({"mode": "step_complete", "steps": [], "result": result})
 
     async def _never_resolve(self):
@@ -3266,6 +3282,9 @@ async def _run_workflow_async(func, checkpoint: dict, input_args: dict):
                 "steps": steps,
             }
         return {"type": "complete", "result": result}
+    except _StepFailure as e:
+        # Re-raise the step's own exception so the child job fails with it.
+        raise e.exc
     except _StepSuspend as e:
         info = e.dispatch_info
         mode = info.get("mode")