Merge branch 'main' into andystaples-add-purge-timeout

Author: andystaples

.github/copilot-instructions.md

@@ -224,3 +224,17 @@ works correctly:
 - If a new public API is added to the core SDK (e.g. a method on
   `OrchestrationContext`), confirm it is accessible through the
   azuremanaged package and add a test or example if appropriate.
+
+## Release Coordination
+
+- Package versions and inter-package minimum dependencies record the latest
+  released compatibility contract. They are not an instruction to bump versions
+  in a feature PR.
+- When a core `durabletask` API is consumed by `durabletask.azuremanaged` or
+  `azure-functions-durable`, do not update package versions or dependency
+  minimums in the feature PR. Document the affected package changelogs and
+  release coordination instead.
+- Create a dedicated release PR after the core package has been released. The
+  pipeline will publish `durabletask` first, so bump all dependency minimums,
+  package versions, and release notes together so resolvers cannot select a
+  provider release with an incompatible older core package.

CHANGELOG.md

@@ -12,6 +12,9 @@ ADDED
 - Added an optional timeout to filtered orchestration purges. Callers can now
 limit a purge operation's duration and inspect `PurgeInstancesResult.is_complete`
 for completed, partial, or backend-unknown completion status.
+- Added optional `reason` parameters to `TaskHubGrpcClient` and
+`AsyncTaskHubGrpcClient` suspend and resume operations. The reason is now sent
+to the backend with the lifecycle request.
 - Added the optional `new_version` argument to
 `OrchestrationContext.continue_as_new()` so continued orchestrations can
 switch to a new version.

azure-functions-durable/CHANGELOG.md

@@ -11,6 +11,16 @@ ADDED
 
 - Added an optional timeout to filtered orchestration purges through
 `DurableFunctionsClient` and `SyncDurableFunctionsClient`.
+> [!NOTE]
+> Release this change only in coordination with a new `durabletask` release
+> that contains the suspend/resume reason APIs. Publish `durabletask` first,
+> then update this package's minimum dependency in its dedicated release PR.
+
+FIXED
+
+- Fixed deprecated `DurableFunctionsClient.suspend()` and `resume()` methods
+discarding their `reason` arguments. Reasons are now forwarded to the Durable
+Task backend.
 
 ## v2.0.0b2
 

azure-functions-durable/azure/durable_functions/client.py

@@ -431,21 +431,13 @@ async def purge_instance_history(self, instance_id: str) -> PurgeHistoryResult:
 
     @deprecated("suspend is deprecated; use suspend_orchestration instead.")
     async def suspend(self, instance_id: str, reason: Optional[str] = None) -> None:
-        """Deprecated alias for :meth:`suspend_orchestration`.
-
-        The v1 ``reason`` argument has no equivalent in durabletask and is
-        ignored.
-        """
-        await self.suspend_orchestration(instance_id)
+        """Deprecated alias for :meth:`suspend_orchestration`."""
+        await self.suspend_orchestration(instance_id, reason=reason)
 
     @deprecated("resume is deprecated; use resume_orchestration instead.")
     async def resume(self, instance_id: str, reason: Optional[str] = None) -> None:
-        """Deprecated alias for :meth:`resume_orchestration`.
-
-        The v1 ``reason`` argument has no equivalent in durabletask and is
-        ignored.
-        """
-        await self.resume_orchestration(instance_id)
+        """Deprecated alias for :meth:`resume_orchestration`."""
+        await self.resume_orchestration(instance_id, reason=reason)
 
     @deprecated("restart is deprecated; use restart_orchestration instead.")
     async def restart(

durabletask-azuremanaged/CHANGELOG.md

@@ -11,6 +11,8 @@ ADDED
 
 - Added an optional timeout to filtered orchestration purges through
 `DurableTaskSchedulerClient` and `AsyncDurableTaskSchedulerClient`.
+- Added optional `reason` parameters to suspend and resume operations inherited
+from `TaskHubGrpcClient` and `AsyncTaskHubGrpcClient`.
 
 ## v1.9.0
 

durabletask/client.py

@@ -799,13 +799,21 @@ def terminate_orchestration(self, instance_id: str, *,
             )
         self._stub.TerminateInstance(req)
 
-    def suspend_orchestration(self, instance_id: str) -> None:
-        req = pb.SuspendRequest(instanceId=instance_id)
+    def suspend_orchestration(self, instance_id: str, *,
+                              reason: str | None = None) -> None:
+        req = pb.SuspendRequest(
+            instanceId=instance_id,
+            reason=helpers.get_string_value(reason),
+        )
         self._logger.info(f"Suspending instance '{instance_id}'.")
         self._stub.SuspendInstance(req)
 
-    def resume_orchestration(self, instance_id: str) -> None:
-        req = pb.ResumeRequest(instanceId=instance_id)
+    def resume_orchestration(self, instance_id: str, *,
+                             reason: str | None = None) -> None:
+        req = pb.ResumeRequest(
+            instanceId=instance_id,
+            reason=helpers.get_string_value(reason),
+        )
         self._logger.info(f"Resuming instance '{instance_id}'.")
         self._stub.ResumeInstance(req)
 
@@ -1335,13 +1343,21 @@ async def terminate_orchestration(self, instance_id: str, *,
             )
         await self._get_stub().TerminateInstance(req)
 
-    async def suspend_orchestration(self, instance_id: str) -> None:
-        req = pb.SuspendRequest(instanceId=instance_id)
+    async def suspend_orchestration(self, instance_id: str, *,
+                                    reason: str | None = None) -> None:
+        req = pb.SuspendRequest(
+            instanceId=instance_id,
+            reason=helpers.get_string_value(reason),
+        )
         self._logger.info(f"Suspending instance '{instance_id}'.")
         await self._get_stub().SuspendInstance(req)
 
-    async def resume_orchestration(self, instance_id: str) -> None:
-        req = pb.ResumeRequest(instanceId=instance_id)
+    async def resume_orchestration(self, instance_id: str, *,
+                                   reason: str | None = None) -> None:
+        req = pb.ResumeRequest(
+            instanceId=instance_id,
+            reason=helpers.get_string_value(reason),
+        )
         self._logger.info(f"Resuming instance '{instance_id}'.")
         await self._get_stub().ResumeInstance(req)
 

durabletask/internal/helpers.py

@@ -234,19 +234,23 @@ def new_event_raised_event(name: str, encoded_input: str | None = None) -> pb.Hi
     )
 
 
-def new_suspend_event() -> pb.HistoryEvent:
+def new_suspend_event(*, encoded_input: str | None = None) -> pb.HistoryEvent:
     return pb.HistoryEvent(
         eventId=-1,
         timestamp=timestamp_pb2.Timestamp(),
-        executionSuspended=pb.ExecutionSuspendedEvent()
+        executionSuspended=pb.ExecutionSuspendedEvent(
+            input=get_string_value(encoded_input)
+        )
     )
 
 
-def new_resume_event() -> pb.HistoryEvent:
+def new_resume_event(*, encoded_input: str | None = None) -> pb.HistoryEvent:
     return pb.HistoryEvent(
         eventId=-1,
         timestamp=timestamp_pb2.Timestamp(),
-        executionResumed=pb.ExecutionResumedEvent()
+        executionResumed=pb.ExecutionResumedEvent(
+            input=get_string_value(encoded_input)
+        )
     )
 
 

durabletask/testing/in_memory_backend.py

@@ -386,7 +386,9 @@ def SuspendInstance(self, request: pb.SuspendRequest, context: grpc.ServicerCont
             if instance.status == pb.ORCHESTRATION_STATUS_SUSPENDED:
                 return pb.SuspendResponse()
 
-            event = helpers.new_suspend_event()
+            event = helpers.new_suspend_event(
+                encoded_input=request.reason.value if request.HasField("reason") else None
+            )
             instance.pending_events.append(event)
             instance.last_updated_at = datetime.now(timezone.utc)
             self._enqueue_orchestration(instance.instance_id)
@@ -403,7 +405,9 @@ def ResumeInstance(self, request: pb.ResumeRequest, context: grpc.ServicerContex
                               f"Orchestration instance '{request.instanceId}' not found")
                 return pb.ResumeResponse()
 
-            event = helpers.new_resume_event()
+            event = helpers.new_resume_event(
+                encoded_input=request.reason.value if request.HasField("reason") else None
+            )
             instance.pending_events.append(event)
             instance.last_updated_at = datetime.now(timezone.utc)
             self._enqueue_orchestration(instance.instance_id)

tests/azure-functions-durable/test_client_compat.py

@@ -472,13 +472,13 @@ async def test_suspend_resume_delegate():
                           new=AsyncMock()) as suspend_mock:
             with pytest.warns(DeprecationWarning):
                 await client.suspend("abc", "reason")
-        suspend_mock.assert_awaited_once_with("abc")
+        suspend_mock.assert_awaited_once_with("abc", reason="reason")
 
         with patch.object(client, "resume_orchestration",
                           new=AsyncMock()) as resume_mock:
             with pytest.warns(DeprecationWarning):
                 await client.resume("abc", "reason")
-        resume_mock.assert_awaited_once_with("abc")
+        resume_mock.assert_awaited_once_with("abc", reason="reason")
     finally:
         await client.close()
 

tests/durabletask/test_client.py

@@ -1258,6 +1258,43 @@ def test_worker_stores_resiliency_options():
     assert worker._resiliency_options is resiliency
 
 
+def test_suspend_resume_orchestration_includes_optional_reason():
+    stub = MagicMock()
+
+    with patch('durabletask.client.shared.get_grpc_channel', return_value=MagicMock()), patch(
+            'durabletask.client.stubs.TaskHubSidecarServiceStub', return_value=stub):
+        client = TaskHubGrpcClient()
+        client.suspend_orchestration('suspended', reason='maintenance')
+        client.resume_orchestration('resumed')
+
+    suspend_request = stub.SuspendInstance.call_args.args[0]
+    assert suspend_request.instanceId == 'suspended'
+    assert suspend_request.reason.value == 'maintenance'
+    resume_request = stub.ResumeInstance.call_args.args[0]
+    assert resume_request.instanceId == 'resumed'
+    assert not resume_request.HasField('reason')
+
+
+@pytest.mark.asyncio
+async def test_async_suspend_resume_orchestration_includes_optional_reason():
+    stub = MagicMock()
+    stub.SuspendInstance = AsyncMock()
+    stub.ResumeInstance = AsyncMock()
+
+    with patch('durabletask.client.shared.get_async_grpc_channel', return_value=MagicMock()), patch(
+            'durabletask.client.stubs.TaskHubSidecarServiceStub', return_value=stub):
+        client = AsyncTaskHubGrpcClient()
+        await client.suspend_orchestration('suspended')
+        await client.resume_orchestration('resumed', reason='maintenance complete')
+
+    suspend_request = stub.SuspendInstance.call_args.args[0]
+    assert suspend_request.instanceId == 'suspended'
+    assert not suspend_request.HasField('reason')
+    resume_request = stub.ResumeInstance.call_args.args[0]
+    assert resume_request.instanceId == 'resumed'
+    assert resume_request.reason.value == 'maintenance complete'
+
+
 def test_get_orchestration_history_aggregates_chunks_and_deexternalizes_payloads():
     store = FakePayloadStore()
     token = store.upload(b'history payload')