Add cancel reasons on caller side

Author: Sushisource

temporalio/client/_impl.py

@@ -352,6 +352,7 @@ async def cancel_workflow(self, input: CancelWorkflowInput) -> None:
                 identity=self._client.identity,
                 request_id=str(uuid.uuid4()),
                 first_execution_run_id=input.first_execution_run_id or "",
+                reason=input.reason,
             ),
             retry=True,
             metadata=input.rpc_metadata,

temporalio/client/_interceptor.py

@@ -111,6 +111,7 @@ class CancelWorkflowInput:
     id: str
     run_id: str | None
     first_execution_run_id: str | None
+    reason: str
     rpc_metadata: Mapping[str, str | bytes]
     rpc_timeout: timedelta | None
 

temporalio/client/_workflow.py

@@ -318,6 +318,7 @@ async def result(
     async def cancel(
         self,
         *,
+        reason: str = "",
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> None:
@@ -334,6 +335,8 @@ async def cancel(
             workflow ID even if it is unrelated to the started workflow.
 
         Args:
+            reason: Reason recorded with the cancellation request. Available
+                inside the workflow via :py:func:`temporalio.workflow.cancellation_reason`.
             rpc_metadata: Headers used on the RPC call. Keys here override
                 client-level RPC metadata keys.
             rpc_timeout: Optional RPC deadline to set for the RPC call.
@@ -346,6 +349,7 @@ async def cancel(
                 id=self._id,
                 run_id=self._run_id,
                 first_execution_run_id=self._first_execution_run_id,
+                reason=reason,
                 rpc_metadata=rpc_metadata,
                 rpc_timeout=rpc_timeout,
             )

temporalio/worker/_workflow_instance.py

@@ -1988,10 +1988,12 @@ async def _outbound_start_child_workflow(
         handle: _ChildWorkflowHandle
 
         # Common code for handling cancel for start and run
-        def apply_child_cancel_error() -> None:
-            # Send a cancel request to the child
+        def apply_child_cancel_error(err: asyncio.CancelledError) -> None:
+            # Send a cancel request to the child, forwarding the msg passed to
+            # Task.cancel(msg) (if any) as the cancellation reason.
+            reason = err.args[0] if err.args and isinstance(err.args[0], str) else ""
             cancel_command = self._add_command()
-            handle._apply_cancel_command(cancel_command)
+            handle._apply_cancel_command(cancel_command, reason=reason)
             # If the cancel command is for external workflow, we
             # have to add a seq and mark it pending
             if cancel_command.HasField("request_cancel_external_workflow_execution"):
@@ -2014,8 +2016,8 @@ async def run_child() -> Any:
                     # We have to shield because we don't want the future itself
                     # to be cancelled
                     return await asyncio.shield(handle._result_fut)
-                except asyncio.CancelledError:
-                    apply_child_cancel_error()
+                except asyncio.CancelledError as err:
+                    apply_child_cancel_error(err)
                     # Clear the cancellation counter on Python 3.11+ so the
                     # next await does not immediately re-raise CancelledError
                     if (
@@ -2038,8 +2040,8 @@ async def run_child() -> Any:
                 # to be cancelled
                 await asyncio.shield(handle._start_fut)
                 return handle
-            except asyncio.CancelledError:
-                apply_child_cancel_error()
+            except asyncio.CancelledError as err:
+                apply_child_cancel_error(err)
                 # Clear the cancellation counter on Python 3.11+ so the
                 # next await does not immediately re-raise CancelledError
                 if (
@@ -3383,8 +3385,12 @@ def _apply_start_command(self) -> None:
     def _apply_cancel_command(
         self,
         command: temporalio.bridge.proto.workflow_commands.WorkflowCommand,
+        *,
+        reason: str = "",
     ) -> None:
-        command.cancel_child_workflow_execution.child_workflow_seq = self._seq
+        v = command.cancel_child_workflow_execution
+        v.child_workflow_seq = self._seq
+        v.reason = reason
 
 
 class _ExternalWorkflowHandle(temporalio.workflow.ExternalWorkflowHandle[Any]):
@@ -3428,14 +3434,15 @@ async def signal(
             )
         )
 
-    async def cancel(self) -> None:
+    async def cancel(self, *, reason: str = "") -> None:
         self._instance._assert_not_read_only("cancel external handle")
         command = self._instance._add_command()
         v = command.request_cancel_external_workflow_execution
         v.workflow_execution.namespace = self._instance._info.namespace
         v.workflow_execution.workflow_id = self._id
         if self._run_id:
             v.workflow_execution.run_id = self._run_id
+        v.reason = reason
         await self._instance._cancel_external_workflow(command)
 
 

temporalio/workflow/_workflow_ops.py

@@ -604,11 +604,15 @@ async def signal(
         """
         raise NotImplementedError
 
-    async def cancel(self) -> None:
+    async def cancel(self, *, reason: str = "") -> None:
         """Send a cancellation request to this external workflow.
 
         This will fail if the workflow cannot accept the request (e.g. if the
         workflow is not found).
+
+        Args:
+            reason: Reason recorded with the cancellation request. Available in
+                the target workflow via :py:func:`cancellation_reason`.
         """
         raise NotImplementedError
 

tests/worker/test_workflow.py

@@ -1149,16 +1149,81 @@ async def started() -> bool:
         # cancel has already been observed.
         assert await handle.query(CancelReasonWorkflow.reason_inner) is None
 
-        await handle.cancel()
+        await handle.cancel(reason="user-supplied reason")
         with pytest.raises(WorkflowFailureError) as err:
             await handle.result()
         assert isinstance(err.value.cause, CancelledError)
 
-        # After external cancel, reason is a string (empty since the Python
-        # client does not send one) — non-None is the load-bearing distinction.
         outer = await handle.query(CancelReasonWorkflow.reason_outer)
-        assert outer is not None
-        assert isinstance(outer, str)
+        assert outer == "user-supplied reason"
+
+
+@workflow.defn
+class CancelReasonReporter:
+    """Workflow that swallows a cancel and returns the observed reason."""
+
+    @workflow.run
+    async def run(self) -> str:
+        try:
+            await asyncio.sleep(1000)
+        except asyncio.CancelledError:
+            return workflow.cancellation_reason() or ""
+        raise RuntimeError("unreachable")
+
+
+@workflow.defn
+class ChildCancelReasonWorkflow:
+    @workflow.run
+    async def run(self, msg: str) -> str:
+        child = await workflow.start_child_workflow(
+            CancelReasonReporter.run,
+            id=f"{workflow.info().workflow_id}_child",
+        )
+        child.cancel(msg)
+        return await child
+
+
+async def test_workflow_child_cancel_reason(client: Client):
+    async with new_worker(
+        client, ChildCancelReasonWorkflow, CancelReasonReporter
+    ) as worker:
+        result = await client.execute_workflow(
+            ChildCancelReasonWorkflow.run,
+            "from-parent",
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        assert result == "from-parent"
+
+
+@workflow.defn
+class ExternalCancelReasonWorkflow:
+    @workflow.run
+    async def run(self, target_id: str) -> None:
+        await workflow.get_external_workflow_handle(target_id).cancel(
+            reason="from-external-caller"
+        )
+
+
+async def test_workflow_external_cancel_reason(client: Client):
+    async with new_worker(
+        client, ExternalCancelReasonWorkflow, CancelReasonReporter
+    ) as worker:
+        target_id = f"workflow-{uuid.uuid4()}"
+        target = await client.start_workflow(
+            CancelReasonReporter.run,
+            id=target_id,
+            task_queue=worker.task_queue,
+        )
+        await client.execute_workflow(
+            ExternalCancelReasonWorkflow.run,
+            target_id,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        # Server wraps the user-supplied reason with metadata about the caller
+        # when one workflow cancels another, so check for substring.
+        assert "from-external-caller" in await target.result()
 
 
 @workflow.defn