Sagas conformance: Python compensation cells stay waiting and lose typed errors (#218)

Author: durable-workflow-ops

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "0.4.93"
+version = "0.4.94"
 description = "Python SDK for the Durable Workflow server (language-neutral HTTP protocol)"
 readme = "README.md"
 requires-python = ">=3.10"

src/durable_workflow/client.py

@@ -3571,6 +3571,8 @@ async def fail_activity_task(
         lease_owner: str,
         message: str,
         failure_type: str | None = None,
+        failure_class: str | None = None,
+        failure_code: int | None = None,
         stack_trace: str | None = None,
         non_retryable: bool = False,
         details: Any | None = None,
@@ -3586,6 +3588,10 @@ async def fail_activity_task(
         failure: dict[str, Any] = {"message": message}
         if failure_type is not None:
             failure["type"] = failure_type
+        if failure_class is not None:
+            failure["class"] = failure_class
+        if failure_code is not None:
+            failure["code"] = failure_code
         if stack_trace is not None:
             failure["stack_trace"] = stack_trace
         if non_retryable:

src/durable_workflow/worker.py

@@ -123,6 +123,16 @@ def _string_or_none(value: Any) -> str | None:
     return value if isinstance(value, str) and value else None
 
 
+def _exception_class_name(error: BaseException | type[BaseException]) -> str:
+    cls = error if isinstance(error, type) else type(error)
+    return f"{cls.__module__}.{cls.__qualname__}"
+
+
+def _exception_code(error: BaseException) -> int | None:
+    code = getattr(error, "code", None)
+    return code if isinstance(code, int) else None
+
+
 def _is_missing_payload_placeholder(value: Any) -> bool:
     return value is None or (isinstance(value, str) and value == "")
 
@@ -1161,6 +1171,8 @@ async def _run_activity_task(self, task: dict[str, Any]) -> str:
                         f"Reinstall durable-workflow with its runtime dependencies."
                     ),
                     failure_type=type(e).__name__,
+                    failure_class=_exception_class_name(e),
+                    failure_code=_exception_code(e),
                     stack_trace=traceback.format_exc(),
                     non_retryable=True,
                 )
@@ -1179,6 +1191,8 @@ async def _run_activity_task(self, task: dict[str, Any]) -> str:
                         f"Verify the argument bytes match the declared codec and writer schema."
                     ),
                     failure_type=type(e).__name__,
+                    failure_class=_exception_class_name(e),
+                    failure_code=_exception_code(e),
                     stack_trace=traceback.format_exc(),
                     non_retryable=True,
                 )
@@ -1225,6 +1239,7 @@ async def _run_activity_task(self, task: dict[str, Any]) -> str:
                     lease_owner=self.worker_id,
                     message="activity cancelled",
                     failure_type="ActivityCancelled",
+                    failure_class=_exception_class_name(ActivityCancelled),
                     non_retryable=True,
                 )
             except Exception as fe:
@@ -1239,6 +1254,8 @@ async def _run_activity_task(self, task: dict[str, Any]) -> str:
                     lease_owner=self.worker_id,
                     message=str(e),
                     failure_type=type(e).__name__,
+                    failure_class=_exception_class_name(e),
+                    failure_code=_exception_code(e),
                     stack_trace=traceback.format_exc(),
                     non_retryable=True,
                 )
@@ -1254,6 +1271,8 @@ async def _run_activity_task(self, task: dict[str, Any]) -> str:
                     lease_owner=self.worker_id,
                     message=str(e),
                     failure_type=type(e).__name__,
+                    failure_class=_exception_class_name(e),
+                    failure_code=_exception_code(e),
                     stack_trace=traceback.format_exc(),
                 )
             except Exception as fe:

src/durable_workflow/workflow.py

@@ -577,6 +577,8 @@ class FailWorkflow:
 
     message: str
     exception_type: str | None = None
+    exception_class: str | None = None
+    exception: dict[str, Any] | None = None
     non_retryable: bool = False
 
     def to_server_command(
@@ -593,6 +595,10 @@ def to_server_command(
         }
         if self.exception_type is not None:
             cmd["exception_type"] = self.exception_type
+        if self.exception_class is not None:
+            cmd["exception_class"] = self.exception_class
+        if self.exception is not None:
+            cmd["exception"] = self.exception
         if self.non_retryable:
             cmd["non_retryable"] = True
         return cmd
@@ -1870,6 +1876,41 @@ def _fail_update_from_exception(update_id: str, prefix: str, exc: Exception) ->
     )
 
 
+def _exception_class_name(exc: BaseException) -> str:
+    return f"{type(exc).__module__}.{type(exc).__qualname__}"
+
+
+def _fail_workflow_from_exception(exc: BaseException, *, prefix: str | None = None) -> FailWorkflow:
+    message = str(exc) or type(exc).__name__
+    if prefix:
+        message = f"{prefix}: {message}"
+
+    exception_type = type(exc).__name__
+    exception_class = _exception_class_name(exc)
+    exception: dict[str, Any] | None = None
+    cause = exc.__cause__
+    activity_failure = exc if isinstance(exc, ActivityFailed) else cause if isinstance(cause, ActivityFailed) else None
+    if isinstance(activity_failure, ActivityFailed):
+        exception_type = activity_failure.exception_type or type(activity_failure).__name__
+        exception_class = activity_failure.exception_class or _exception_class_name(activity_failure)
+        exception = {
+            "type": exception_type,
+            "class": exception_class,
+            "message": message,
+        }
+        if activity_failure.activity_type is not None:
+            exception["activity_type"] = activity_failure.activity_type
+        if activity_failure.activity_attempt_id is not None:
+            exception["activity_attempt_id"] = activity_failure.activity_attempt_id
+
+    return FailWorkflow(
+        message=message,
+        exception_type=exception_type,
+        exception_class=exception_class,
+        exception=exception,
+    )
+
+
 def _history_event_type(event: Mapping[str, Any]) -> str | None:
     value = event.get("event_type") or event.get("type")
     return value if isinstance(value, str) and value else None
@@ -2781,9 +2822,9 @@ def _terminal_state(value: Any, *, include_pending: bool) -> _ReplayState:
                     try:
                         satisfied = bool(cmd.predicate())
                     except Exception as exc:
-                        return _state([FailWorkflow(
-                            message=f"wait_condition predicate raised: {exc}",
-                            exception_type=type(exc).__name__,
+                        return _state([_fail_workflow_from_exception(
+                            exc,
+                            prefix="wait_condition predicate raised",
                         )])
                     if resolution == "satisfied":
                         if has_reopened_same_wait:
@@ -2848,14 +2889,8 @@ def _terminal_state(value: Any, *, include_pending: bool) -> _ReplayState:
         except Exception as exc:
             if isinstance(exc, NonDeterministicReplayError):
                 raise
-            return _state([FailWorkflow(
-                message=str(exc),
-                exception_type=type(exc).__name__,
-            )])
+            return _state([_fail_workflow_from_exception(exc)])
     except NonDeterministicReplayError:
         raise
     except Exception as exc:
-        return _state([FailWorkflow(
-            message=str(exc),
-            exception_type=type(exc).__name__,
-        )])
+        return _state([_fail_workflow_from_exception(exc)])

tests/test_client.py

@@ -2155,7 +2155,7 @@ async def test_get_schedule_history_matches_polyglot_fixture(self, client: Clien
 
         wire_events = fixture["response_body"]["events"]
         assert len(page.events) == len(wire_events)
-        for parsed, wire in zip(page.events, wire_events):
+        for parsed, wire in zip(page.events, wire_events, strict=True):
             assert parsed.id == wire["id"]
             assert parsed.sequence == wire["sequence"]
             assert parsed.event_type == wire["event_type"]
@@ -2511,11 +2511,17 @@ async def test_body_shape(self, client: Client) -> None:
                 activity_attempt_id="a1",
                 lease_owner="w1",
                 message="activity error",
+                failure_type="TypedActivityError",
+                failure_class="app.activities.TypedActivityError",
+                failure_code=409,
                 non_retryable=True,
             )
             body = mock.call_args.kwargs.get("json") or mock.call_args[1].get("json")
             assert "failure" in body
             assert body["failure"]["message"] == "activity error"
+            assert body["failure"]["type"] == "TypedActivityError"
+            assert body["failure"]["class"] == "app.activities.TypedActivityError"
+            assert body["failure"]["code"] == 409
             assert body["failure"]["non_retryable"] is True
 
     @pytest.mark.asyncio

tests/test_replay.py

@@ -152,6 +152,20 @@ def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
         raise ValueError("something went wrong")
 
 
+@workflow.defn(name="typed-compensation-failure-workflow")
+class TypedCompensationFailureWorkflow:
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        try:
+            yield ctx.schedule_activity("charge_card", [])
+        except ActivityFailed:
+            try:
+                yield ctx.schedule_activity("cancel_flight", [])
+            except ActivityFailed as exc:
+                failure_type = exc.exception_type or type(exc).__name__
+                raise RuntimeError(f"compensation failed for cancel_flight: {failure_type}: {exc}") from exc
+        return "completed"
+
+
 def _activity_completed_event(sequence: int, activity_type: str) -> dict:
     return {
         "event_type": "ActivityCompleted",
@@ -668,13 +682,59 @@ def test_exception_produces_fail_command(self) -> None:
         assert isinstance(cmd, FailWorkflow)
         assert "something went wrong" in cmd.message
         assert cmd.exception_type == "ValueError"
+        assert cmd.exception_class == "builtins.ValueError"
 
     def test_fail_server_command_shape(self) -> None:
         history = [{"event_type": "ActivityCompleted", "payload": {"result": '"ok"'}}]
         outcome = replay(FailingWorkflow, history, [])
         server_cmd = outcome.commands[0].to_server_command("q")
         assert server_cmd["type"] == "fail_workflow"
         assert "something went wrong" in server_cmd["message"]
+        assert server_cmd["exception_class"] == "builtins.ValueError"
+
+    def test_typed_compensation_activity_failure_reaches_terminal_command(self) -> None:
+        history = [
+            {
+                "event_type": "ActivityFailed",
+                "payload": {
+                    "sequence": 1,
+                    "activity_type": "charge_card",
+                    "exception_type": "PlannedSagaFailure",
+                    "exception_class": "sagas.PlannedSagaFailure",
+                    "message": "planned saga failure",
+                },
+            },
+            {
+                "event_type": "ActivityFailed",
+                "payload": {
+                    "sequence": 2,
+                    "activity_type": "cancel_flight",
+                    "activity_attempt_id": "attempt-cancel-flight",
+                    "exception_type": "TypedCancelFlightError",
+                    "exception_class": "__main__.TypedCancelFlightError",
+                    "message": "cancel_flight typed compensation failure",
+                },
+            },
+        ]
+
+        outcome = replay(TypedCompensationFailureWorkflow, history, [])
+
+        assert len(outcome.commands) == 1
+        cmd = outcome.commands[0]
+        assert isinstance(cmd, FailWorkflow)
+        assert cmd.exception_type == "TypedCancelFlightError"
+        assert cmd.exception_class == "__main__.TypedCancelFlightError"
+        assert cmd.exception is not None
+        assert cmd.exception["type"] == "TypedCancelFlightError"
+        assert cmd.exception["message"] == cmd.message
+        assert cmd.exception["activity_type"] == "cancel_flight"
+        assert cmd.exception["activity_attempt_id"] == "attempt-cancel-flight"
+        assert "cancel_flight typed compensation failure" in cmd.message
+
+        server_cmd = cmd.to_server_command("q")
+        assert server_cmd["exception_type"] == "TypedCancelFlightError"
+        assert server_cmd["exception_class"] == "__main__.TypedCancelFlightError"
+        assert server_cmd["exception"]["message"] == cmd.message
 
 
 class TestCompleteWorkflowCommand:

tests/test_worker.py

@@ -39,6 +39,10 @@
 )
 
 
+class TypedCancelFlightError(Exception):
+    code = 712
+
+
 @workflow.defn(name="test-wf")
 class TestWorkflow:
     def run(self, ctx, *args):  # type: ignore[no-untyped-def]
@@ -2000,6 +2004,31 @@ def failing_act() -> None:
         call_kwargs = mock_client.fail_activity_task.call_args.kwargs
         assert "boom" in call_kwargs["message"]
         assert call_kwargs["failure_type"] == "RuntimeError"
+        assert call_kwargs["failure_class"] == "builtins.RuntimeError"
+
+    @pytest.mark.asyncio
+    async def test_activity_exception_reports_typed_failure_metadata(self, mock_client: AsyncMock) -> None:
+        @activity.defn(name="cancel-flight")
+        def cancel_flight() -> None:
+            raise TypedCancelFlightError("cancel_flight typed compensation failure")
+
+        worker = Worker(mock_client, task_queue="q1", workflows=[], activities=[cancel_flight])
+        task = {
+            "task_id": "at-typed",
+            "activity_attempt_id": "aa-typed",
+            "activity_type": "cancel-flight",
+            "arguments": "[]",
+            "payload_codec": "json",
+        }
+
+        await worker._run_activity_task(task)
+
+        mock_client.fail_activity_task.assert_called_once()
+        call_kwargs = mock_client.fail_activity_task.call_args.kwargs
+        assert call_kwargs["failure_type"] == "TypedCancelFlightError"
+        assert call_kwargs["failure_class"] == f"{__name__}.TypedCancelFlightError"
+        assert call_kwargs["failure_code"] == 712
+        assert "cancel_flight typed compensation failure" in call_kwargs["message"]
 
 
 class TestWorkerInterceptors: