Add suspend and resume reasons

.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

@@ -9,6 +9,9 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ADDED
 
+- 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

@@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+> [!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
 
 ADDED

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

@@ -7,6 +7,11 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+ADDED
+
+- Added optional `reason` parameters to suspend and resume operations inherited
+from `TaskHubGrpcClient` and `AsyncTaskHubGrpcClient`.
+
 ## v1.9.0
 
 CHANGED

durabletask/client.py

@@ -786,13 +786,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)
 
@@ -1320,13 +1328,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')

tests/durabletask/test_orchestration_e2e.py

@@ -275,7 +275,7 @@ def orchestrator(ctx: task.OrchestrationContext, _):
  assert state.serialized_output == json.dumps("timed out")
 
 
-def test_suspend_and_resume():
+def test_suspend_and_resume_preserves_reasons_in_history():
  def orchestrator(ctx: task.OrchestrationContext, _):
  result = yield ctx.wait_for_external_event("my_event")
  return result
@@ -290,7 +290,7 @@ def orchestrator(ctx: task.OrchestrationContext, _):
  assert state is not None
 
  # Suspend the orchestration and wait for it to go into the SUSPENDED state
- task_hub_client.suspend_orchestration(id)
+ task_hub_client.suspend_orchestration(id, reason="maintenance")
  deadline = time.time() + 10
  while state.runtime_status == client.OrchestrationStatus.RUNNING:
  assert time.time() < deadline, "Timed out waiting for SUSPENDED status"
@@ -308,11 +308,18 @@ def orchestrator(ctx: task.OrchestrationContext, _):
  pass
 
  # Resume the orchestration and wait for it to complete
- task_hub_client.resume_orchestration(id)
+ task_hub_client.resume_orchestration(id, reason="maintenance complete")
  state = task_hub_client.wait_for_orchestration_completion(id, timeout=30)
+ events = task_hub_client.get_orchestration_history(id)
  assert state is not None
  assert state.runtime_status == client.OrchestrationStatus.COMPLETED
  assert state.serialized_output == json.dumps(42)
+ suspended_event = next(
+ event for event in events if isinstance(event, history.ExecutionSuspendedEvent))
+ resumed_event = next(
+ event for event in events if isinstance(event, history.ExecutionResumedEvent))
+ assert suspended_event.input == "maintenance"
+ assert resumed_event.input == "maintenance complete"
 
 
 def test_terminate():