Expose a typed orchestration ID reuse policy

Add public status-based deduplication policies while retaining protobuf callers and backend defaults. Document inherited provider support and protocol constraints.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: andystaples

CHANGELOG.md

@@ -7,6 +7,13 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+ADDED
+
+- Added public `OrchestrationIdReusePolicy` with typed deduplication statuses for
+synchronous and asynchronous orchestration scheduling, including legacy
+`OrchestrationStatus.CANCELED`. Omission preserves backend defaults; existing
+protobuf policies remain supported. See [ID reuse](docs/features.md#orchestration-id-reuse).
+
 FIXED
 
 - Activity and sub-orchestration `retry_timeout` decisions now use replay-safe

README.md

@@ -32,6 +32,14 @@ The Azure Functions Durable provider requires Python 3.13 or later.
 
 ## Optional Features
 
+### Orchestration ID Reuse
+
+Use `OrchestrationIdReusePolicy` from `durabletask.client` to select which
+existing-instance statuses reject duplicate IDs. Omission preserves backend
+defaults; an explicit empty deduplication list permits replacement in every
+stable status, subject to backend support. See the
+[policy documentation](./docs/features.md#orchestration-id-reuse).
+
 ### Task Hub Query Filters
 
 Set `OrchestrationQuery.task_hub_names` to filter orchestration queries by one or

azure-functions-durable/CHANGELOG.md

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ADDED
 
+- With the corresponding core SDK release, Functions-bound synchronous and
+asynchronous clients accept public `OrchestrationIdReusePolicy` values through
+`schedule_new_orchestration()`, preserving backend defaults when omitted.
+See [ID reuse](../docs/features.md#orchestration-id-reuse). Core release and
+dependency minimum updates are coordinated in the dedicated release PR.
 - Added opt-in `HttpRetryOptions` on `context.call_http()` for durable initial
 request retries with configurable timing and status selection. Empty or omitted
 status lists select all non-2xx responses; exhausted retries fail the call.

docs/features.md

@@ -11,6 +11,64 @@ fault-tolerant and durable, meaning that they can automatically recover from fai
 their local execution state. Orchestrator functions must be deterministic, meaning that they must
 always produce the same output given the same input.
 
+### Orchestration ID reuse
+
+Use the public `OrchestrationIdReusePolicy` to control what happens when scheduling
+an orchestration whose instance ID already exists:
+
+```python
+from durabletask.client import OrchestrationIdReusePolicy, OrchestrationStatus
+
+instance_id = client.schedule_new_orchestration(
+    "process_order",
+    instance_id="order-123",
+    reuse_id_policy=OrchestrationIdReusePolicy(
+        dedupe_statuses=[
+            OrchestrationStatus.RUNNING,
+            OrchestrationStatus.PENDING,
+            OrchestrationStatus.SUSPENDED,
+            OrchestrationStatus.TERMINATED,
+            OrchestrationStatus.CANCELED,
+        ],
+    ),
+)
+```
+
+In this example, only completed or failed instances are eligible for replacement.
+`dedupe_statuses` contains the statuses that **must not** be replaced, matching the
+.NET and Go SDKs' deduplication semantics.
+
+- Omitting `reuse_id_policy`, or using `OrchestrationIdReusePolicy()` with
+  `dedupe_statuses=None`, preserves the backend's default duplicate-ID behavior.
+- An explicit `dedupe_statuses=[]` permits replacement in every stable status,
+  including running instances. Include all stable statuses to reject every
+  duplicate.
+- Stable statuses are `RUNNING`, `PENDING`, `SUSPENDED`, `COMPLETED`, `FAILED`,
+  `TERMINATED`, and the legacy `CANCELED` status. `CONTINUED_AS_NEW` is an execution
+  transition, not a configurable instance status. Invalid status values raise
+  `ValueError` before the scheduling request is sent.
+
+The policy works with synchronous and asynchronous core, Azure Managed, and
+Functions-bound clients. Use `await client.schedule_new_orchestration(...)` with
+an asynchronous client. The deprecated Functions `start_new()` alias does not
+accept policy options.
+
+Existing callers may continue passing a generated protobuf policy through
+`reuse_id_policy`. Its `replaceableStatus` list has the **opposite polarity**:
+it lists statuses that may be replaced. An explicitly empty protobuf policy still
+permits no replacements; it is not equivalent to an empty public deduplication list.
+
+> [!WARNING]
+> Replacing an active instance can terminate its execution. Support, replacement
+> guarantees, and duplicate rejection are enforced by the backend, not by the SDK.
+> The SDK sends one scheduling request; it does not perform a separate
+> terminate-then-start sequence. The shared protocol does not provide an atomic
+> no-op/`IGNORE` action.
+
+Provider releases exposing this public policy require the corresponding core SDK
+release first. Package versions and dependency minimums are coordinated in the
+dedicated release PR.
+
 ### Orchestration versioning
 
 Orchestrations may be assigned a version when they are first created. If an orchestration is given a

durabletask-azuremanaged/CHANGELOG.md

@@ -7,6 +7,13 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+ADDED
+
+- With the corresponding core SDK release, synchronous and asynchronous clients
+accept public `OrchestrationIdReusePolicy` values to control duplicate instance
+IDs without changing backend defaults. See [ID reuse](../docs/features.md#orchestration-id-reuse).
+Core release and dependency minimum updates are coordinated in the dedicated release PR.
+
 FIXED
 
 - With the corresponding core SDK update, activity and sub-orchestration

durabletask/client.py

@@ -63,6 +63,7 @@ class OrchestrationStatus(Enum):
     RUNNING = pb.ORCHESTRATION_STATUS_RUNNING
     COMPLETED = pb.ORCHESTRATION_STATUS_COMPLETED
     FAILED = pb.ORCHESTRATION_STATUS_FAILED
+    CANCELED = pb.ORCHESTRATION_STATUS_CANCELED
     TERMINATED = pb.ORCHESTRATION_STATUS_TERMINATED
     CONTINUED_AS_NEW = pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW
     PENDING = pb.ORCHESTRATION_STATUS_PENDING
@@ -72,6 +73,23 @@ def __str__(self) -> str:
         return cast(str, helpers.get_orchestration_status_str(self.value))
 
 
+@dataclass
+class OrchestrationIdReusePolicy:
+    """Control duplicate instance IDs when scheduling an orchestration.
+
+    ``dedupe_statuses`` lists existing-instance statuses that must not be
+    replaced. Other stable statuses permit replacement if supported by the
+    backend. ``None`` preserves backend defaults; an empty sequence permits
+    replacement in every stable status. ``CONTINUED_AS_NEW`` is a transition,
+    not a configurable instance status.
+
+    This policy does not provide an atomic no-op/ignore action. Replacement
+    behavior and support are determined by the backend.
+    """
+
+    dedupe_statuses: Sequence[OrchestrationStatus] | None = None
+
+
 @dataclass
 class OrchestrationState:
     instance_id: str
@@ -631,13 +649,18 @@ def schedule_new_orchestration(self, orchestrator: task.Orchestrator[TInput, TOu
                                    input: TInput | None = None,
                                    instance_id: str | None = None,
                                    start_at: datetime | None = None,
-                                   reuse_id_policy: pb.OrchestrationIdReusePolicy | None = None,
+                                   reuse_id_policy: OrchestrationIdReusePolicy | pb.OrchestrationIdReusePolicy | None = None,
                                    tags: dict[str, str] | None = None,
                                    version: str | None = None) -> str:
         """Schedule a new orchestration instance for execution.
 
         If instance_id is not specified, a random UUID hex string is generated upfront
         and shared across both the OpenTelemetry producer span and the CreateInstanceRequest.
+
+        ``reuse_id_policy`` controls which existing-instance statuses reject
+        duplicate IDs. Omission preserves backend defaults. Existing protobuf
+        policies remain supported and retain their replaceable-status semantics.
+        Invalid public policy statuses raise ``ValueError`` before sending the request.
         """
         name = orchestrator if isinstance(orchestrator, str) else task.get_name(orchestrator)
         resolved_instance_id = instance_id if instance_id else uuid.uuid4().hex
@@ -1183,13 +1206,18 @@ async def schedule_new_orchestration(self, orchestrator: task.Orchestrator[TInpu
                                          input: TInput | None = None,
                                          instance_id: str | None = None,
                                          start_at: datetime | None = None,
-                                         reuse_id_policy: pb.OrchestrationIdReusePolicy | None = None,
+                                         reuse_id_policy: OrchestrationIdReusePolicy | pb.OrchestrationIdReusePolicy | None = None,
                                          tags: dict[str, str] | None = None,
                                          version: str | None = None) -> str:
         """Schedule a new orchestration instance for execution asynchronously.
 
         If instance_id is not specified, a random UUID hex string is generated upfront
         and shared across both the OpenTelemetry producer span and the CreateInstanceRequest.
+
+        ``reuse_id_policy`` controls which existing-instance statuses reject
+        duplicate IDs. Omission preserves backend defaults. Existing protobuf
+        policies remain supported and retain their replaceable-status semantics.
+        Invalid public policy statuses raise ``ValueError`` before sending the request.
         """
         name = orchestrator if isinstance(orchestrator, str) else task.get_name(orchestrator)
         resolved_instance_id = instance_id if instance_id else uuid.uuid4().hex

durabletask/internal/client_helpers.py

@@ -23,6 +23,7 @@
 if TYPE_CHECKING:
     from durabletask.client import (
         EntityQuery,
+        OrchestrationIdReusePolicy,
         OrchestrationQuery,
         OrchestrationState,
         OrchestrationStatus,
@@ -77,11 +78,30 @@ def build_schedule_new_orchestration_req(
         input: TInput | None,
         instance_id: str | None,
         start_at: datetime | None,
-        reuse_id_policy: pb.OrchestrationIdReusePolicy | None,
+        reuse_id_policy: OrchestrationIdReusePolicy | pb.OrchestrationIdReusePolicy | None,
         tags: dict[str, str] | None,
         version: str | None,
         data_converter: DataConverter) -> pb.CreateInstanceRequest:
     """Build a CreateInstanceRequest for scheduling a new orchestration."""
+    if reuse_id_policy is not None and not isinstance(reuse_id_policy, pb.OrchestrationIdReusePolicy):
+        from durabletask.client import OrchestrationStatus
+
+        dedupe_statuses = reuse_id_policy.dedupe_statuses
+        if dedupe_statuses is None:
+            reuse_id_policy = None
+        else:
+            stable_statuses = tuple(
+                status for status in OrchestrationStatus
+                if status is not OrchestrationStatus.CONTINUED_AS_NEW)
+            for status in dedupe_statuses:
+                if status not in stable_statuses:
+                    raise ValueError(f"Invalid orchestration dedupe status: {status!r}")
+            reuse_id_policy = pb.OrchestrationIdReusePolicy(
+                replaceableStatus=[
+                    status.value for status in stable_statuses
+                    if status not in dedupe_statuses
+                ])
+
     name = orchestrator if isinstance(orchestrator, str) else task.get_name(orchestrator)
     return pb.CreateInstanceRequest(
         name=name,

tests/azure-functions-durable/test_client_transport.py

@@ -11,7 +11,13 @@
 from google.protobuf.wrappers_pb2 import StringValue
 
 from azure.durable_functions import DurableFunctionsClient, SyncDurableFunctionsClient
-from durabletask.client import AsyncTaskHubGrpcClient, OrchestrationQuery, OrchestrationStatus, TaskHubGrpcClient
+from durabletask.client import (
+    AsyncTaskHubGrpcClient,
+    OrchestrationIdReusePolicy,
+    OrchestrationQuery,
+    OrchestrationStatus,
+    TaskHubGrpcClient,
+)
 from durabletask.internal import orchestrator_service_pb2 as pb
 from durabletask.internal import orchestrator_service_pb2_grpc as rpc
 
@@ -26,6 +32,35 @@ def _config(address: str) -> str:
     })
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+async def test_functions_client_accepts_public_reuse_policy(use_async):
+    stub = Mock()
+    response = pb.CreateInstanceResponse(instanceId="instance")
+    policy = OrchestrationIdReusePolicy(dedupe_statuses=[OrchestrationStatus.RUNNING])
+    if use_async:
+        stub.StartInstance = AsyncMock(return_value=response)
+        async with DurableFunctionsClient(_config("localhost:8080")) as client:
+            with patch.object(client, "_get_stub", return_value=stub):
+                result = await client.schedule_new_orchestration(
+                    "orchestrator", instance_id="instance", reuse_id_policy=policy)
+    else:
+        stub.StartInstance.return_value = response
+        with SyncDurableFunctionsClient(_config("localhost:8080")) as client:
+            with patch.object(client, "_stub", stub):
+                result = client.schedule_new_orchestration(
+                    "orchestrator", instance_id="instance", reuse_id_policy=policy)
+
+    assert result == "instance"
+    request = stub.StartInstance.call_args.args[0]
+    assert request.HasField("orchestrationIdReusePolicy")
+    assert list(request.orchestrationIdReusePolicy.replaceableStatus) == [
+        pb.ORCHESTRATION_STATUS_COMPLETED, pb.ORCHESTRATION_STATUS_FAILED,
+        pb.ORCHESTRATION_STATUS_CANCELED, pb.ORCHESTRATION_STATUS_TERMINATED,
+        pb.ORCHESTRATION_STATUS_PENDING, pb.ORCHESTRATION_STATUS_SUSPENDED,
+    ]
+
+
 @pytest.mark.parametrize("use_async", [False, True])
 @pytest.mark.parametrize("address, expected", [
     ("http://127.0.0.1:8080/durabletask/", "http://127.0.0.1:8080"),

tests/durabletask-azuremanaged/test_azuremanaged_grpc_resiliency.py

@@ -1,13 +1,14 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
-from unittest.mock import Mock, patch
+from unittest.mock import AsyncMock, Mock, patch
 from collections.abc import Mapping
 from typing import Any
 
 import pytest
 
 from durabletask import task
+from durabletask.client import OrchestrationIdReusePolicy, OrchestrationStatus
 from durabletask.internal import helpers
 from durabletask.internal import orchestrator_service_pb2 as pb
 from durabletask.azuremanaged.client import (
@@ -21,6 +22,35 @@
 )
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+async def test_dts_client_accepts_public_reuse_policy(use_async):
+    stub = Mock()
+    response = pb.CreateInstanceResponse(instanceId="instance")
+    stub.StartInstance = AsyncMock(return_value=response) if use_async else Mock(return_value=response)
+    policy = OrchestrationIdReusePolicy(dedupe_statuses=[OrchestrationStatus.RUNNING])
+    kwargs = dict(host_address="localhost:4001", taskhub="hub",
+                  token_credential=None, channel=Mock())
+
+    with patch("durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub):
+        if use_async:
+            async with AsyncDurableTaskSchedulerClient(**kwargs) as client:
+                assert await client.schedule_new_orchestration(
+                    "orchestrator", instance_id="instance", reuse_id_policy=policy) == "instance"
+        else:
+            with DurableTaskSchedulerClient(**kwargs) as client:
+                assert client.schedule_new_orchestration(
+                    "orchestrator", instance_id="instance", reuse_id_policy=policy) == "instance"
+
+    request = stub.StartInstance.call_args.args[0]
+    assert request.HasField("orchestrationIdReusePolicy")
+    assert list(request.orchestrationIdReusePolicy.replaceableStatus) == [
+        pb.ORCHESTRATION_STATUS_COMPLETED, pb.ORCHESTRATION_STATUS_FAILED,
+        pb.ORCHESTRATION_STATUS_CANCELED, pb.ORCHESTRATION_STATUS_TERMINATED,
+        pb.ORCHESTRATION_STATUS_PENDING, pb.ORCHESTRATION_STATUS_SUSPENDED,
+    ]
+
+
 def test_dts_client_passes_resiliency_options_to_base_client():
     resiliency = GrpcClientResiliencyOptions()
     with patch("durabletask.azuremanaged.client.TaskHubGrpcClient.__init__", return_value=None) as mock_init: