Sagas conformance: Python reverse compensation stays waiting after first refund (#213)

Author: durable-workflow-ops

src/durable_workflow/workflow.py

@@ -1884,10 +1884,34 @@ def _workflow_sequence(payload: Mapping[str, Any]) -> int | None:
     return None
 
 
+def _activity_type_from_payload(payload: Mapping[str, Any]) -> str | None:
+    activity_type = _optional_str(payload.get("activity_type"))
+    if activity_type is not None:
+        return activity_type
+
+    activity_name = _optional_str(payload.get("activity_name"))
+    if activity_name is not None:
+        return activity_name
+
+    activity_payload = payload.get("activity")
+    if isinstance(activity_payload, Mapping):
+        return (
+            _optional_str(activity_payload.get("activity_type"))
+            or _optional_str(activity_payload.get("activity_name"))
+            or _optional_str(activity_payload.get("type"))
+            or _optional_str(activity_payload.get("name"))
+        )
+
+    return None
+
+
 def _recorded_step_details(payload: Mapping[str, Any]) -> dict[str, Any]:
     details: dict[str, Any] = {}
+    activity_type = _activity_type_from_payload(payload)
+    if activity_type is not None:
+        details["activity_type"] = activity_type
+
     for key in (
-        "activity_type",
         "workflow_type",
         "child_workflow_type",
         "timer_kind",
@@ -2024,7 +2048,7 @@ def _activity_failed_from_payload(payload: Mapping[str, Any]) -> ActivityFailed:
 
     return ActivityFailed(
         message or "activity failed",
-        activity_type=_optional_str(payload.get("activity_type")),
+        activity_type=_activity_type_from_payload(payload),
         activity_execution_id=_optional_str(payload.get("activity_execution_id")),
         activity_attempt_id=_optional_str(payload.get("activity_attempt_id")),
         failure_id=_optional_str(payload.get("failure_id")),

tests/test_replay.py

@@ -75,6 +75,43 @@ def run(self, ctx: WorkflowContext, order_id: str):  # type: ignore[no-untyped-d
         return {"charged": True}
 
 
+@workflow.defn(name="reverse-compensation-saga")
+class ReverseCompensationSaga:
+    def run(self, ctx: WorkflowContext, payload: dict):  # type: ignore[type-arg]
+        completed: list[str] = []
+        compensations: list[str] = []
+        steps = [
+            ("reserve_flight", "cancel_flight"),
+            ("reserve_hotel", "cancel_hotel"),
+            ("charge_card", "refund_card"),
+            ("send_confirmation", ""),
+        ]
+
+        try:
+            for action, compensation in steps:
+                yield ctx.schedule_activity(action, [payload])
+                completed.append(action)
+                if compensation:
+                    compensations.append(compensation)
+        except ActivityFailed:
+            if not compensations:
+                raise
+            for compensation in reversed(compensations):
+                yield ctx.schedule_activity(compensation, [payload])
+                completed.append(compensation)
+            return {
+                "status": "compensated",
+                "activity_log": completed,
+                "compensations": compensations,
+            }
+
+        return {
+            "status": "completed",
+            "activity_log": completed,
+            "compensations": compensations,
+        }
+
+
 @workflow.defn(name="activity-failure-payload-inspector")
 class ActivityFailurePayloadInspector:
     def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
@@ -115,6 +152,34 @@ def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
         raise ValueError("something went wrong")
 
 
+def _activity_completed_event(sequence: int, activity_type: str) -> dict:
+    return {
+        "event_type": "ActivityCompleted",
+        "payload": {
+            "sequence": sequence,
+            "activity_type": activity_type,
+            "payload_codec": "json",
+            "result": serializer.encode({"activity": activity_type}, codec="json"),
+        },
+    }
+
+
+def _activity_failed_event(sequence: int, activity_type: str) -> dict:
+    return {
+        "event_type": "ActivityFailed",
+        "payload": {
+            "sequence": sequence,
+            "activity_type": activity_type,
+            "failure_id": f"failure-{sequence}",
+            "failure_category": "activity",
+            "exception_type": "RuntimeError",
+            "exception_class": "builtins.RuntimeError",
+            "message": f"{activity_type} planned saga failure before forward effect",
+            "non_retryable": True,
+        },
+    }
+
+
 class TestSimpleReturn:
     def test_non_generator_completes(self) -> None:
         outcome = replay(SimpleReturn, [], [])
@@ -215,7 +280,7 @@ def test_failed_activity_is_thrown_into_workflow_for_compensation(self) -> None:
             {
                 "event_type": "ActivityFailed",
                 "payload": {
-                    "activity_type": "charge-card",
+                    "activity_name": "charge-card",
                     "activity_execution_id": "act-1",
                     "activity_attempt_id": "attempt-1",
                     "failure_id": "failure-1",
@@ -486,6 +551,56 @@ def test_schedule_activity_rejects_incoherent_timeout_envelopes(
             cmd.to_server_command("default-queue")
 
 
+class TestReverseCompensationSaga:
+    def test_resumes_reverse_compensation_after_first_refund(self) -> None:
+        payload = {"scenario_id": "reverse-compensation"}
+        history = [
+            _activity_completed_event(1, "reserve_flight"),
+            _activity_completed_event(2, "reserve_hotel"),
+            _activity_completed_event(3, "charge_card"),
+            _activity_failed_event(4, "send_confirmation"),
+            _activity_completed_event(5, "refund_card"),
+        ]
+
+        outcome = replay(ReverseCompensationSaga, history, [payload], payload_codec="json")
+
+        assert len(outcome.commands) == 1
+        cmd = outcome.commands[0]
+        assert isinstance(cmd, ScheduleActivity)
+        assert cmd.activity_type == "cancel_hotel"
+        assert cmd.arguments == [payload]
+
+    def test_completes_reverse_compensation_without_duplicates(self) -> None:
+        payload = {"scenario_id": "reverse-compensation"}
+        history = [
+            _activity_completed_event(1, "reserve_flight"),
+            _activity_completed_event(2, "reserve_hotel"),
+            _activity_completed_event(3, "charge_card"),
+            _activity_failed_event(4, "send_confirmation"),
+            _activity_completed_event(5, "refund_card"),
+            _activity_completed_event(6, "cancel_hotel"),
+            _activity_completed_event(7, "cancel_flight"),
+        ]
+
+        outcome = replay(ReverseCompensationSaga, history, [payload], payload_codec="json")
+
+        assert len(outcome.commands) == 1
+        cmd = outcome.commands[0]
+        assert isinstance(cmd, CompleteWorkflow)
+        assert cmd.result == {
+            "status": "compensated",
+            "activity_log": [
+                "reserve_flight",
+                "reserve_hotel",
+                "charge_card",
+                "refund_card",
+                "cancel_hotel",
+                "cancel_flight",
+            ],
+            "compensations": ["cancel_flight", "cancel_hotel", "refund_card"],
+        }
+
+
 class TestTwoActivities:
     def test_first_schedules(self) -> None:
         outcome = replay(TwoActivities, [], [])