[cross-repo from server#387] Conformance blocker: complete remaining child-workflows parity scenarios (#149)

Author: durable-workflow-ops

src/durable_workflow/__init__.py

@@ -54,7 +54,9 @@
 from .errors import (
     ActivityCancelled,
     ActivityFailed,
+    ChildWorkflowCancelled,
     ChildWorkflowFailed,
+    ChildWorkflowTerminated,
     DurableWorkflowError,
     InvalidArgument,
     NamespaceNotFound,
@@ -188,7 +190,9 @@
     "ActivityRetryPolicy",
     "BridgeAdapterOutcome",
     "ChildWorkflowRetryPolicy",
+    "ChildWorkflowCancelled",
     "ChildWorkflowFailed",
+    "ChildWorkflowTerminated",
     "Client",
     "ContinueAsNew",
     "NamespaceDescription",

src/durable_workflow/errors.py

@@ -236,9 +236,66 @@ class ChildWorkflowFailed(DurableWorkflowError):
     :attr:`exception_class` mirrors the child's recorded exception class.
     """
 
-    def __init__(self, message: str, exception_class: str | None = None) -> None:
+    def __init__(
+        self,
+        message: str,
+        exception_class: str | None = None,
+        *,
+        failure_kind: str | None = None,
+        child_workflow_run_id: str | None = None,
+        child_workflow_type: str | None = None,
+    ) -> None:
         super().__init__(message)
         self.exception_class = exception_class
+        self.failure_kind = failure_kind
+        self.failure_category = failure_kind
+        self.child_workflow_run_id = child_workflow_run_id
+        self.child_workflow_type = child_workflow_type
+
+
+class ChildWorkflowCancelled(ChildWorkflowFailed):
+    """A child workflow finished in the ``cancelled`` state.
+
+    Raised inside the parent workflow when it awaits a child that was
+    cancelled directly or by parent-close policy. It is a child outcome, so it
+    remains catchable as :class:`ChildWorkflowFailed`.
+    """
+
+    def __init__(
+        self,
+        message: str = "child workflow was cancelled",
+        exception_class: str | None = None,
+        *,
+        child_workflow_run_id: str | None = None,
+        child_workflow_type: str | None = None,
+    ) -> None:
+        super().__init__(
+            message,
+            exception_class,
+            failure_kind="cancelled",
+            child_workflow_run_id=child_workflow_run_id,
+            child_workflow_type=child_workflow_type,
+        )
+
+
+class ChildWorkflowTerminated(ChildWorkflowFailed):
+    """A child workflow finished in the ``terminated`` state."""
+
+    def __init__(
+        self,
+        message: str = "child workflow was terminated",
+        exception_class: str | None = None,
+        *,
+        child_workflow_run_id: str | None = None,
+        child_workflow_type: str | None = None,
+    ) -> None:
+        super().__init__(
+            message,
+            exception_class,
+            failure_kind="terminated",
+            child_workflow_run_id=child_workflow_run_id,
+            child_workflow_type=child_workflow_type,
+        )
 
 
 class ActivityFailed(DurableWorkflowError):

src/durable_workflow/workflow.py

@@ -33,7 +33,9 @@
 from .external_storage import ExternalPayloadCache, ExternalStorageDriver
 from .errors import (
     ActivityFailed,
+    ChildWorkflowCancelled,
     ChildWorkflowFailed,
+    ChildWorkflowTerminated,
     NonDeterministicReplayError,
     QueryFailed,
     WorkflowPayloadDecodeError,
@@ -2009,6 +2011,8 @@ def _activity_failed_from_payload(payload: Mapping[str, Any]) -> ActivityFailed:
     message = _optional_str(payload.get("message"))
     if message is None and exception is not None:
         message = _optional_str(exception.get("message"))
+    if message is None:
+        message = _optional_str(payload.get("closed_reason"))
     exception_type = _optional_str(payload.get("exception_type"))
     if exception_type is None and exception is not None:
         exception_type = _optional_str(exception.get("type"))
@@ -2032,6 +2036,51 @@ def _activity_failed_from_payload(payload: Mapping[str, Any]) -> ActivityFailed:
     )
 
 
+def _child_workflow_failed_from_payload(
+    payload: Mapping[str, Any],
+    event_type: str,
+) -> ChildWorkflowFailed:
+    exception_payload = payload.get("exception")
+    exception = dict(exception_payload) if isinstance(exception_payload, Mapping) else None
+
+    message = _optional_str(payload.get("message"))
+    if message is None and exception is not None:
+        message = _optional_str(exception.get("message"))
+
+    exception_class = _optional_str(payload.get("exception_class"))
+    if exception_class is None:
+        exception_class = _optional_str(payload.get("exception_type"))
+    if exception_class is None and exception is not None:
+        exception_class = _optional_str(exception.get("class")) or _optional_str(exception.get("type"))
+
+    child_workflow_run_id = _optional_str(payload.get("child_workflow_run_id"))
+    child_workflow_type = _optional_str(payload.get("child_workflow_type"))
+
+    if event_type == "ChildRunCancelled":
+        return ChildWorkflowCancelled(
+            message or "child workflow was cancelled",
+            exception_class,
+            child_workflow_run_id=child_workflow_run_id,
+            child_workflow_type=child_workflow_type,
+        )
+
+    if event_type == "ChildRunTerminated":
+        return ChildWorkflowTerminated(
+            message or "child workflow was terminated",
+            exception_class,
+            child_workflow_run_id=child_workflow_run_id,
+            child_workflow_type=child_workflow_type,
+        )
+
+    return ChildWorkflowFailed(
+        message or "child workflow failed",
+        exception_class,
+        failure_kind=_optional_str(payload.get("failure_category")) or "child_workflow",
+        child_workflow_run_id=child_workflow_run_id,
+        child_workflow_type=child_workflow_type,
+    )
+
+
 def _first_yield_failure(values: Iterable[Any]) -> ActivityFailed | ChildWorkflowFailed | None:
     for value in values:
         if isinstance(value, ActivityFailed):
@@ -2392,9 +2441,9 @@ def _receiver_condition_wait_bindings() -> dict[int, str | None]:
                 shape,
                 ev,
             )
-        elif etype == "ChildRunFailed":
+        elif etype in ("ChildRunFailed", "ChildRunCancelled", "ChildRunTerminated"):
             _append_resolved_result(
-                ChildWorkflowFailed(payload.get("message", "child workflow failed")),
+                _child_workflow_failed_from_payload(payload, etype),
                 "child workflow",
                 ev,
             )

tests/test_replay.py

@@ -6,7 +6,13 @@
 import pytest
 
 from durable_workflow import Replayer, ReplayOutcome, serializer, workflow
-from durable_workflow.errors import ActivityFailed, ChildWorkflowFailed, NonDeterministicReplayError
+from durable_workflow.errors import (
+    ActivityFailed,
+    ChildWorkflowCancelled,
+    ChildWorkflowFailed,
+    ChildWorkflowTerminated,
+    NonDeterministicReplayError,
+)
 from durable_workflow.workflow import (
     ActivityRetryPolicy,
     ChildWorkflowRetryPolicy,
@@ -837,6 +843,48 @@ def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
             return "handled"
 
 
+@workflow.defn(name="child-wf-failure-details")
+class ChildWorkflowFailureDetailsWf:
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        try:
+            yield ctx.start_child_workflow("sub-workflow", [])
+        except ChildWorkflowFailed as exc:
+            return {
+                "message": str(exc),
+                "exception_class": exc.exception_class,
+                "failure_kind": exc.failure_kind,
+                "child_workflow_run_id": exc.child_workflow_run_id,
+                "child_workflow_type": exc.child_workflow_type,
+            }
+
+
+@workflow.defn(name="child-wf-cancelled")
+class ChildWorkflowCancelledWf:
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        try:
+            yield ctx.start_child_workflow("sub-workflow", [])
+        except ChildWorkflowCancelled as exc:
+            return {
+                "message": str(exc),
+                "failure_kind": exc.failure_kind,
+                "child_workflow_run_id": exc.child_workflow_run_id,
+                "child_workflow_type": exc.child_workflow_type,
+            }
+
+
+@workflow.defn(name="child-wf-terminated")
+class ChildWorkflowTerminatedWf:
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        try:
+            yield ctx.start_child_workflow("sub-workflow", [])
+        except ChildWorkflowTerminated as exc:
+            return {
+                "message": str(exc),
+                "failure_kind": exc.failure_kind,
+                "child_workflow_run_id": exc.child_workflow_run_id,
+            }
+
+
 @workflow.defn(name="child-wf-failed-fallback")
 class ChildWorkflowFailedFallbackWf:
     def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
@@ -949,6 +997,79 @@ def test_child_failed_caught(self) -> None:
         assert isinstance(cmd, CompleteWorkflow)
         assert cmd.result == "handled"
 
+    def test_child_failed_preserves_typed_details(self) -> None:
+        history = [
+            {
+                "event_type": "ChildRunFailed",
+                "payload": {
+                    "message": "child failed",
+                    "exception_class": "ExampleChildError",
+                    "failure_category": "child_workflow",
+                    "child_workflow_run_id": "child-run-1",
+                    "child_workflow_type": "sub-workflow",
+                },
+            },
+        ]
+
+        outcome = replay(ChildWorkflowFailureDetailsWf, history, [])
+
+        assert len(outcome.commands) == 1
+        cmd = outcome.commands[0]
+        assert isinstance(cmd, CompleteWorkflow)
+        assert cmd.result == {
+            "message": "child failed",
+            "exception_class": "ExampleChildError",
+            "failure_kind": "child_workflow",
+            "child_workflow_run_id": "child-run-1",
+            "child_workflow_type": "sub-workflow",
+        }
+
+    def test_child_cancelled_is_observed_by_parent_as_typed_cancellation(self) -> None:
+        history = [
+            {
+                "event_type": "ChildRunCancelled",
+                "payload": {
+                    "message": "cancelled by operator",
+                    "child_workflow_run_id": "child-run-1",
+                    "child_workflow_type": "sub-workflow",
+                },
+            },
+        ]
+
+        outcome = replay(ChildWorkflowCancelledWf, history, [])
+
+        assert len(outcome.commands) == 1
+        cmd = outcome.commands[0]
+        assert isinstance(cmd, CompleteWorkflow)
+        assert cmd.result == {
+            "message": "cancelled by operator",
+            "failure_kind": "cancelled",
+            "child_workflow_run_id": "child-run-1",
+            "child_workflow_type": "sub-workflow",
+        }
+
+    def test_child_terminated_is_observed_by_parent_as_typed_child_failure(self) -> None:
+        history = [
+            {
+                "event_type": "ChildRunTerminated",
+                "payload": {
+                    "message": "terminated by operator",
+                    "child_workflow_run_id": "child-run-1",
+                },
+            },
+        ]
+
+        outcome = replay(ChildWorkflowTerminatedWf, history, [])
+
+        assert len(outcome.commands) == 1
+        cmd = outcome.commands[0]
+        assert isinstance(cmd, CompleteWorkflow)
+        assert cmd.result == {
+            "message": "terminated by operator",
+            "failure_kind": "terminated",
+            "child_workflow_run_id": "child-run-1",
+        }
+
     def test_child_failed_then_fallback_yields_command(self) -> None:
         history = [{"event_type": "ChildRunFailed", "payload": {"message": "child failed"}}]
         outcome = replay(ChildWorkflowFailedFallbackWf, history, [])