WIP Last completion result

Author: tconley1428

temporalio/bridge/sdk-core

@@ -563,6 +563,7 @@ 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,
         )
         if defn.sandboxed:
             return self._workflow_runner.create_instance(det)

temporalio/worker/_workflow.py

@@ -79,6 +79,7 @@
     WorkflowInboundInterceptor,
     WorkflowOutboundInterceptor,
 )
+from ..api.common.v1.message_pb2 import Payloads
 
 logger = logging.getLogger(__name__)
 
@@ -143,7 +144,7 @@ class WorkflowInstanceDetails:
     extern_functions: Mapping[str, Callable]
     disable_eager_activity_execution: bool
     worker_level_failure_exception_types: Sequence[Type[BaseException]]
-
+    last_completion_result: Payloads
 
 class WorkflowInstance(ABC):
     """Instance of a workflow that can handle activations."""
@@ -320,6 +321,8 @@ def __init__(self, det: WorkflowInstanceDetails) -> None:
         # metadata query
         self._current_details = ""
 
+        self._last_completion_result = det.last_completion_result
+
         # 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
@@ -1686,6 +1689,17 @@ def workflow_set_current_details(self, details: str):
         self._assert_not_read_only("set current details")
         self._current_details = details
 
+    def workflow_last_completion_result(self, type_hint: Optional[Type]) -> Optional[Any]:
+        print("workflow_last_completion_result: ", self._last_completion_result, type(self._last_completion_result), "payload length:", len(self._last_completion_result.payloads))
+        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
+
+        print("Payload:", self._last_completion_result.payloads[0])
+        return self._payload_converter.from_payload(self._last_completion_result.payloads[0], type_hint)
+
     #### Calls from outbound impl ####
     # These are in alphabetical order and all start with "_outbound_".
 
@@ -2766,6 +2780,14 @@ 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)
+<<<<<<< Updated upstream
+=======
+        print("Input summary:", self._input.summary)
+        if self._input.summary:
+            command.user_metadata.summary.CopyFrom(
+                self._instance._payload_converter.to_payload(self._input.summary)
+            )
+>>>>>>> Stashed changes
         v.cancellation_type = cast(
             temporalio.bridge.proto.workflow_commands.ActivityCancellationType.ValueType,
             int(self._input.cancellation_type),

temporalio/worker/_workflow_instance.py

@@ -27,6 +27,7 @@
 )
 from ._importer import Importer
 from ._restrictions import RestrictionContext, SandboxRestrictions
+from ...api.common.v1.message_pb2 import Payloads
 
 _fake_info = temporalio.workflow.Info(
     attempt=-1,
@@ -84,6 +85,7 @@ 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(),
             ),
         )
 

temporalio/worker/workflow_sandbox/_runner.py

@@ -897,6 +897,9 @@ def workflow_get_current_details(self) -> str: ...
     @abstractmethod
     def workflow_set_current_details(self, details: str): ...
 
+    @abstractmethod
+    def workflow_last_completion_result(self, type_hint: Optional[Type]) -> Optional[Any]: ...
+
 
 _current_update_info: contextvars.ContextVar[UpdateInfo] = contextvars.ContextVar(
     "__temporal_current_update_info"
@@ -1039,6 +1042,19 @@ def get_current_details() -> str:
     return _Runtime.current().workflow_get_current_details()
 
 
+@overload
+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.
+    """
+    return _Runtime.current().workflow_last_completion_result(type_hint)
+
+
 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

temporalio/workflow.py

@@ -1,3 +1,4 @@
+import asyncio
 import dataclasses
 import json
 import os
@@ -1501,3 +1502,43 @@ async def test_cloud_client_simple():
         GetNamespaceRequest(namespace=os.environ["TEMPORAL_CLIENT_CLOUD_NAMESPACE"])
     )
     assert os.environ["TEMPORAL_CLIENT_CLOUD_NAMESPACE"] == result.namespace.namespace
+
+
+@workflow.defn
+class LastCompletionResultWorkflow:
+    @workflow.run
+    async def run(self) -> str:
+        last_result = workflow.get_last_completion_result(type_hint=str)
+        if last_result is not None:
+            return "From last completion:" + last_result
+        else:
+            return "My First Result"
+
+
+async def test_schedule_last_completion_result(
+    client: Client, env: WorkflowEnvironment
+):
+    if env.supports_time_skipping:
+        pytest.skip("Java test server doesn't support schedules")
+
+    async with new_worker(client, LastCompletionResultWorkflow) as worker:
+        handle = await client.create_schedule(
+            f"schedule-{uuid.uuid4()}",
+            Schedule(
+                action=ScheduleActionStartWorkflow(
+                    "LastCompletionResultWorkflow",
+                    id=f"workflow-{uuid.uuid4()}",
+                    task_queue=worker.task_queue,
+                ),
+                spec=ScheduleSpec(),
+            ),
+        )
+        await handle.trigger()
+        await asyncio.sleep(1)
+        await handle.trigger()
+        await asyncio.sleep(1)
+        print(await handle.describe())
+
+        await handle.delete()
+        assert False
+