Merge remote-tracking branch 'origin/main' into otel/benign_exceptions

Author: tconley1428

temporalio/bridge/Cargo.lock

@@ -313,7 +313,7 @@ def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig:
         config["data_converter"] = DataConverter(
             payload_converter_class=_OpenAIPayloadConverter
         )
-        return config
+        return self.next_worker_plugin.configure_replayer(config)
 
     @asynccontextmanager
     async def run_replayer(

temporalio/bridge/sdk-core

@@ -352,6 +352,8 @@ class StartLocalActivityInput:
     local_retry_threshold: Optional[timedelta]
     cancellation_type: temporalio.workflow.ActivityCancellationType
     headers: Mapping[str, temporalio.api.common.v1.Payload]
+    summary: Optional[str]
+
     # The types may be absent
     arg_types: Optional[List[Type]]
     ret_type: Optional[Type]

temporalio/contrib/openai_agents/_temporal_openai_agents.py

@@ -552,6 +552,10 @@ def _create_workflow_instance(
             priority=temporalio.common.Priority._from_proto(init.priority),
         )
 
+        last_failure = (
+            init.continued_failure if init.HasField("continued_failure") else None
+        )
+
         # Create instance from details
         det = WorkflowInstanceDetails(
             payload_converter_class=self._data_converter.payload_converter_class,
@@ -563,6 +567,8 @@ def _create_workflow_instance(
             extern_functions=self._extern_functions,
             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,
+            last_failure=last_failure,
         )
         if defn.sandboxed:
             return self._workflow_runner.create_instance(det)

temporalio/worker/_interceptor.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,
@@ -143,6 +144,8 @@ class WorkflowInstanceDetails:
     extern_functions: Mapping[str, Callable]
     disable_eager_activity_execution: bool
     worker_level_failure_exception_types: Sequence[Type[BaseException]]
+    last_completion_result: temporalio.api.common.v1.Payloads
+    last_failure: Optional[Failure]
 
 
 class WorkflowInstance(ABC):
@@ -320,6 +323,9 @@ def __init__(self, det: WorkflowInstanceDetails) -> None:
         # metadata query
         self._current_details = ""
 
+        self._last_completion_result = det.last_completion_result
+        self._last_failure = det.last_failure
+
         # The versioning behavior of this workflow, as established by annotation or by the dynamic
         # config function. Is only set once upon initialization.
         self._versioning_behavior: Optional[temporalio.common.VersioningBehavior] = None
@@ -1464,6 +1470,7 @@ def workflow_start_local_activity(
         local_retry_threshold: Optional[timedelta],
         cancellation_type: temporalio.workflow.ActivityCancellationType,
         activity_id: Optional[str],
+        summary: Optional[str],
     ) -> temporalio.workflow.ActivityHandle[Any]:
         # Get activity definition if it's callable
         name: str
@@ -1496,6 +1503,7 @@ def workflow_start_local_activity(
                 retry_policy=retry_policy,
                 local_retry_threshold=local_retry_threshold,
                 cancellation_type=cancellation_type,
+                summary=summary,
                 headers={},
                 arg_types=arg_types,
                 ret_type=ret_type,
@@ -1703,6 +1711,37 @@ def workflow_is_failure_exception(self, err: BaseException) -> bool:
             )
         )
 
+    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]:
+        if len(self._last_completion_result.payloads) == 0:
+            return None
+        elif len(self._last_completion_result.payloads) > 1:
+            warnings.warn(
+                f"Expected single last completion result, got {len(self._last_completion_result.payloads)}"
+            )
+            return None
+
+        if type_hint is None:
+            return self._payload_converter.from_payload(
+                self._last_completion_result.payloads[0]
+            )
+        else:
+            return self._payload_converter.from_payload(
+                self._last_completion_result.payloads[0], type_hint
+            )
+
+    def workflow_last_failure(self) -> Optional[BaseException]:
+        if self._last_failure:
+            return self._failure_converter.from_failure(
+                self._last_failure, self._payload_converter
+            )
+
+        return None
+
     #### Calls from outbound impl ####
     # These are in alphabetical order and all start with "_outbound_".
 
@@ -2766,6 +2805,10 @@ def _apply_schedule_command(
             v.start_to_close_timeout.FromTimedelta(self._input.start_to_close_timeout)
         if self._input.retry_policy:
             self._input.retry_policy.apply_to_proto(v.retry_policy)
+        if self._input.summary:
+            command.user_metadata.summary.CopyFrom(
+                self._instance._payload_converter.to_payload(self._input.summary)
+            )
         v.cancellation_type = cast(
             temporalio.bridge.proto.workflow_commands.ActivityCancellationType.ValueType,
             int(self._input.cancellation_type),
@@ -2787,11 +2830,6 @@ def _apply_schedule_command(
                 command.schedule_activity.versioning_intent = (
                     self._input.versioning_intent._to_proto()
                 )
-            if self._input.summary:
-                command.user_metadata.summary.CopyFrom(
-                    self._instance._payload_converter.to_payload(self._input.summary)
-                )
-            print("Activity summary: ", command.user_metadata.summary)
             if self._input.priority:
                 command.schedule_activity.priority.CopyFrom(
                     self._input.priority._to_proto()

temporalio/worker/_workflow.py

@@ -18,6 +18,9 @@
 import temporalio.worker._workflow_instance
 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 (
     UnsandboxedWorkflowRunner,
@@ -84,6 +87,8 @@ def prepare_workflow(self, defn: temporalio.workflow._Definition) -> None:
                 extern_functions={},
                 disable_eager_activity_execution=False,
                 worker_level_failure_exception_types=self._worker_level_failure_exception_types,
+                last_completion_result=Payloads(),
+                last_failure=Failure(),
             ),
         )