Add previous run failure

Author: tconley1428

temporalio/worker/_workflow.py

@@ -552,6 +552,10 @@ def _create_workflow_instance(
             priority=temporalio.common.Priority._from_proto(init.priority),
         )
 
+        previous_run_failure = None
+        if init.HasField("continued_failure"):
+            previous_run_failure = init.continued_failure
+
         # Create instance from details
         det = WorkflowInstanceDetails(
             payload_converter_class=self._data_converter.payload_converter_class,
@@ -564,6 +568,7 @@ def _create_workflow_instance(
             disable_eager_activity_execution=self._disable_eager_activity_execution,
             worker_level_failure_exception_types=self._workflow_failure_exception_types,
             last_completion_result=init.last_completion_result,
+            previous_run_failure=previous_run_failure,
         )
         if defn.sandboxed:
             return self._workflow_runner.create_instance(det)

temporalio/worker/_workflow_instance.py

@@ -64,6 +64,7 @@
 import temporalio.workflow
 from temporalio.service import __version__
 
+from ..api.failure.v1.message_pb2 import Failure
 from ._interceptor import (
     ContinueAsNewInput,
     ExecuteWorkflowInput,
@@ -144,6 +145,7 @@ class WorkflowInstanceDetails:
     disable_eager_activity_execution: bool
     worker_level_failure_exception_types: Sequence[Type[BaseException]]
     last_completion_result: temporalio.api.common.v1.Payloads
+    previous_run_failure: Optional[Failure]
 
 
 class WorkflowInstance(ABC):
@@ -322,6 +324,7 @@ def __init__(self, det: WorkflowInstanceDetails) -> None:
         self._current_details = ""
 
         self._last_completion_result = det.last_completion_result
+        self._previous_run_failure = det.previous_run_failure
 
         # The versioning behavior of this workflow, as established by annotation or by the dynamic
         # config function. Is only set once upon initialization.
@@ -1689,6 +1692,9 @@ def workflow_set_current_details(self, details: str):
         self._assert_not_read_only("set current details")
         self._current_details = details
 
+    def workflow_has_last_completion_result(self) -> bool:
+        return len(self._last_completion_result.payloads) > 0
+
     def workflow_last_completion_result(
         self, type_hint: Optional[Type]
     ) -> Optional[Any]:
@@ -1709,6 +1715,14 @@ def workflow_last_completion_result(
                 self._last_completion_result.payloads[0], type_hint
             )
 
+    def workflow_previous_run_failure(self) -> Optional[BaseException]:
+        if self._previous_run_failure:
+            return self._failure_converter.from_failure(
+                self._previous_run_failure, self._payload_converter
+            )
+
+        return None
+
     #### Calls from outbound impl ####
     # These are in alphabetical order and all start with "_outbound_".
 

temporalio/worker/workflow_sandbox/_runner.py

@@ -19,6 +19,7 @@
 import temporalio.workflow
 
 from ...api.common.v1.message_pb2 import Payloads
+from ...api.failure.v1.message_pb2 import Failure
 
 # Workflow instance has to be relative import
 from .._workflow_instance import (
@@ -87,6 +88,7 @@ def prepare_workflow(self, defn: temporalio.workflow._Definition) -> None:
                 disable_eager_activity_execution=False,
                 worker_level_failure_exception_types=self._worker_level_failure_exception_types,
                 last_completion_result=Payloads(),
+                previous_run_failure=Failure(),
             ),
         )
 

temporalio/workflow.py

@@ -61,6 +61,7 @@
 import temporalio.workflow
 from temporalio.nexus._util import ServiceHandlerT
 
+from .api.failure.v1.message_pb2 import Failure
 from .types import (
     AnyType,
     CallableAsyncNoParam,
@@ -897,11 +898,17 @@ def workflow_get_current_details(self) -> str: ...
     @abstractmethod
     def workflow_set_current_details(self, details: str): ...
 
+    @abstractmethod
+    def workflow_has_last_completion_result(self) -> bool: ...
+
     @abstractmethod
     def workflow_last_completion_result(
         self, type_hint: Optional[Type]
     ) -> Optional[Any]: ...
 
+    @abstractmethod
+    def workflow_previous_run_failure(self) -> Optional[BaseException]: ...
+
 
 _current_update_info: contextvars.ContextVar[UpdateInfo] = contextvars.ContextVar(
     "__temporal_current_update_info"
@@ -1044,6 +1051,13 @@ def get_current_details() -> str:
     return _Runtime.current().workflow_get_current_details()
 
 
+def has_last_completion_result() -> bool:
+    """Get the last completion result of the workflow. This be None if there was
+    no previous completion or the result was None
+    """
+    return _Runtime.current().workflow_has_last_completion_result()
+
+
 @overload
 def get_last_completion_result() -> Optional[Any]: ...
 
@@ -1053,14 +1067,18 @@ def get_last_completion_result(type_hint: Type[ParamType]) -> Optional[ParamType
 
 
 def get_last_completion_result(type_hint: Optional[Type] = None) -> Optional[Any]:
-    """Get the current details of the workflow which may appear in the UI/CLI.
-    Unlike static details set at start, this value can be updated throughout
-    the life of the workflow and is independent of the static details.
-    This can be in Temporal markdown format and can span multiple lines.
+    """Get the last completion result of the workflow. This be None if there was
+    no previous completion or the result was None. has_last_completion_result()
+    can be used to differentiate.
     """
     return _Runtime.current().workflow_last_completion_result(type_hint)
 
 
+def get_previous_run_failure() -> Optional[BaseException]:
+    """Get the last failure of the workflow."""
+    return _Runtime.current().workflow_previous_run_failure()
+
+
 def set_current_details(description: str) -> None:
     """Set the current details of the workflow which may appear in the UI/CLI.
     Unlike static details set at start, this value can be updated throughout

tests/worker/test_workflow.py

@@ -8327,3 +8327,29 @@ async def test_workflow_headers_with_codec(
             assert headers["foo"].data == b"bar"
         else:
             assert headers["foo"].data != b"bar"
+
+
+@workflow.defn
+class PreviousRunFailureWorkflow:
+    @workflow.run
+    async def run(self) -> str:
+        if workflow.info().attempt != 1:
+            previous_failure = workflow.get_previous_run_failure()
+            assert isinstance(previous_failure, ApplicationError)
+            assert previous_failure.message == "Intentional Failure"
+            return "Done"
+        raise ApplicationError("Intentional Failure")
+
+
+async def test_previous_run_failure(client: Client):
+    async with new_worker(client, PreviousRunFailureWorkflow) as worker:
+        handle = await client.start_workflow(
+            PreviousRunFailureWorkflow.run,
+            id=f"previous-run-failure-workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+            retry_policy=RetryPolicy(
+                initial_interval=timedelta(milliseconds=10),
+            ),
+        )
+        result = await handle.result()
+        assert result == "Done"