Deprecate reuse-instance-id

dapr/ext/workflow/AGENTS.md

@@ -131,7 +131,7 @@ Workflow (orchestrator) functions must remain generators (`def` with `yield`). T
 
 Client for workflow lifecycle management:
 
-- `schedule_new_workflow(workflow, *, input, instance_id, start_at, reuse_id_policy)` → returns `instance_id`
+- `schedule_new_workflow(workflow, *, input, instance_id, start_at, reuse_id_policy)` → returns `instance_id`. `reuse_id_policy` is deprecated (no effect, emits `DeprecationWarning`): an instance ID can always be reused once the existing instance is in a terminal state.
 - `get_workflow_state(instance_id, *, fetch_payloads=True)` → `Optional[WorkflowState]`
 - `wait_for_workflow_start(instance_id, *, fetch_payloads, timeout_in_seconds)`
 - `wait_for_workflow_completion(instance_id, *, fetch_payloads, timeout_in_seconds)`

dapr/ext/workflow/_durabletask/client.py

@@ -62,8 +62,15 @@ def __str__(self):
         return helpers.get_orchestration_status_str(self.value)
 
 
+# TODO(v1.21): remove OrchestrationIdReuseAction, WorkflowIdReusePolicy, and the reuse_id_policy
+# parameter on schedule_new_orchestration (sync and aio) — deprecated in 1.18, kept for the
+# 3-version policy.
 class OrchestrationIdReuseAction(Enum):
-    """Action to take when scheduling an orchestration whose ID already exists."""
+    """Action to take when scheduling an orchestration whose ID already exists.
+
+    Deprecated: has no effect and will be removed in a future release. An orchestration ID
+    can always be reused once the existing instance has reached a terminal state.
+    """
 
     ERROR = 0
     IGNORE = 1
@@ -72,7 +79,11 @@ class OrchestrationIdReuseAction(Enum):
 
 @dataclass
 class WorkflowIdReusePolicy:
-    """Policy controlling what happens when a new orchestration is scheduled with an ID that already exists."""
+    """Policy controlling what happens when a new orchestration is scheduled with an ID that already exists.
+
+    Deprecated: has no effect and will be removed in a future release. An orchestration ID
+    can always be reused once the existing instance has reached a terminal state.
+    """
 
     action: OrchestrationIdReuseAction
     operation_status: list[OrchestrationStatus]

dapr/ext/workflow/aio/dapr_workflow_client.py

@@ -17,6 +17,7 @@
 
 from datetime import datetime
 from typing import Any, Optional, TypeVar, Union
+from warnings import warn
 
 from grpc.aio import AioRpcError
 
@@ -109,12 +110,22 @@ async def schedule_new_workflow(
             start_at: The time when the workflow instance should start executing.
             If not specified or if a date-time in the past is specified, the workflow instance will
             be scheduled immediately.
-            reuse_id_policy: Optional policy to reuse the workflow id when there is a conflict with
-            an existing workflow instance.
+            reuse_id_policy: Deprecated and has no effect; it will be removed in a future
+            release. A workflow instance ID can always be reused once the existing instance
+            with that ID has reached a terminal state (e.g. COMPLETED, FAILED, or TERMINATED).
 
         Returns:
             The ID of the scheduled workflow instance.
         """
+        # TODO(v1.21): remove reuse_id_policy — deprecated in 1.18, kept for the 3-version policy.
+        if reuse_id_policy is not None:
+            warn(
+                'reuse_id_policy is deprecated and has no effect; it will be removed in a '
+                'future release. A workflow instance ID can always be reused once the existing '
+                'instance with that ID has reached a terminal state.',
+                DeprecationWarning,
+                stacklevel=2,
+            )
         if isinstance(workflow, str):
             workflow_name = workflow
         elif hasattr(workflow, '_dapr_alternate_name'):

dapr/ext/workflow/dapr_workflow_client.py

@@ -17,6 +17,7 @@
 
 from datetime import datetime
 from typing import Any, Optional, TypeVar, Union
+from warnings import warn
 
 from grpc import RpcError
 
@@ -111,12 +112,22 @@ def schedule_new_workflow(
             start_at: The time when the workflow instance should start executing.
             If not specified or if a date-time in the past is specified, the workflow instance will
             be scheduled immediately.
-            reuse_id_policy: Optional policy to reuse the workflow id when there is a conflict with
-            an existing workflow instance.
+            reuse_id_policy: Deprecated and has no effect; it will be removed in a future
+            release. A workflow instance ID can always be reused once the existing instance
+            with that ID has reached a terminal state (e.g. COMPLETED, FAILED, or TERMINATED).
 
         Returns:
             The ID of the scheduled workflow instance.
         """
+        # TODO(v1.21): remove reuse_id_policy — deprecated in 1.18, kept for the 3-version policy.
+        if reuse_id_policy is not None:
+            warn(
+                'reuse_id_policy is deprecated and has no effect; it will be removed in a '
+                'future release. A workflow instance ID can always be reused once the existing '
+                'instance with that ID has reached a terminal state.',
+                DeprecationWarning,
+                stacklevel=2,
+            )
         if isinstance(workflow, str):
             workflow_name = workflow
         elif hasattr(workflow, '_dapr_alternate_name'):

examples/workflow/monitor.py

@@ -71,7 +71,8 @@ def send_alert(ctx, message: str):
         pass
 
     if not status or status.runtime_status.name != 'RUNNING':
-        # TODO update to use reuse_id_policy
+        # Scheduling over a RUNNING instance fails; reuse of the ID is only allowed once the
+        # existing instance has reached a terminal state, hence the pre-check above.
         instance_id = wf_client.schedule_new_workflow(
             workflow=status_monitor_workflow,
             input=JobStatus(job_id=job_id, is_healthy=True),

tests/ext/workflow/test_workflow_client.py

@@ -14,6 +14,7 @@
 """
 
 import unittest
+import warnings
 from datetime import datetime
 from typing import Any, Union
 from unittest import mock
@@ -198,6 +199,33 @@ def test_schedule_workflow_by_name_string(self):
             assert result == mock_schedule_result
             assert fake_client.last_scheduled_workflow_name == 'my_registered_workflow'
 
+    def test_schedule_workflow_reuse_id_policy_is_deprecated(self):
+        fake_client = FakeTaskHubGrpcClient()
+        with mock.patch(
+            'dapr.ext.workflow._durabletask.client.TaskHubGrpcClient', return_value=fake_client
+        ):
+            wfClient = DaprWorkflowClient()
+            reuse_id_policy = client.WorkflowIdReusePolicy(
+                action=client.OrchestrationIdReuseAction.TERMINATE,
+                operation_status=[client.OrchestrationStatus.RUNNING],
+            )
+            with self.assertWarns(DeprecationWarning):
+                result = wfClient.schedule_new_workflow(
+                    workflow='my_registered_workflow', reuse_id_policy=reuse_id_policy
+                )
+            assert result == mock_schedule_result
+
+    def test_schedule_workflow_without_reuse_id_policy_does_not_warn(self):
+        fake_client = FakeTaskHubGrpcClient()
+        with mock.patch(
+            'dapr.ext.workflow._durabletask.client.TaskHubGrpcClient', return_value=fake_client
+        ):
+            wfClient = DaprWorkflowClient()
+            with warnings.catch_warnings():
+                warnings.simplefilter('error', DeprecationWarning)
+                result = wfClient.schedule_new_workflow(workflow='my_registered_workflow')
+            assert result == mock_schedule_result
+
     def test_client_functions(self):
         with mock.patch(
             'dapr.ext.workflow._durabletask.client.TaskHubGrpcClient',

tests/ext/workflow/test_workflow_client_aio.py

@@ -14,6 +14,7 @@
 """
 
 import unittest
+import warnings
 from datetime import datetime
 from typing import Any, Union
 from unittest import mock
@@ -202,6 +203,35 @@ async def test_schedule_workflow_by_name_string(self):
             assert result == mock_schedule_result
             assert fake_client.last_scheduled_workflow_name == 'my_registered_workflow'
 
+    async def test_schedule_workflow_reuse_id_policy_is_deprecated(self):
+        fake_client = FakeAsyncTaskHubGrpcClient()
+        with mock.patch(
+            'dapr.ext.workflow._durabletask.aio.client.AsyncTaskHubGrpcClient',
+            return_value=fake_client,
+        ):
+            wfClient = DaprWorkflowClient()
+            reuse_id_policy = client.WorkflowIdReusePolicy(
+                action=client.OrchestrationIdReuseAction.TERMINATE,
+                operation_status=[client.OrchestrationStatus.RUNNING],
+            )
+            with self.assertWarns(DeprecationWarning):
+                result = await wfClient.schedule_new_workflow(
+                    workflow='my_registered_workflow', reuse_id_policy=reuse_id_policy
+                )
+            assert result == mock_schedule_result
+
+    async def test_schedule_workflow_without_reuse_id_policy_does_not_warn(self):
+        fake_client = FakeAsyncTaskHubGrpcClient()
+        with mock.patch(
+            'dapr.ext.workflow._durabletask.aio.client.AsyncTaskHubGrpcClient',
+            return_value=fake_client,
+        ):
+            wfClient = DaprWorkflowClient()
+            with warnings.catch_warnings():
+                warnings.simplefilter('error', DeprecationWarning)
+                result = await wfClient.schedule_new_workflow(workflow='my_registered_workflow')
+            assert result == mock_schedule_result
+
     async def test_client_functions(self):
         with mock.patch(
             'dapr.ext.workflow._durabletask.aio.client.AsyncTaskHubGrpcClient',