Correct duplicate-start contract and publish Python 2.6.1

CHANGELOG.md

@@ -14,6 +14,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `deprecate_patch` for the same change ID, while validating each call's range
   and preserving retained duplicate-marker sequences.
 
+## [2.6.1] - 2026-10-19
+
+### Fixed
+
+- Raise `WorkflowAlreadyStarted` for the current Server's
+  `instance_already_started` duplicate-start reason, retaining support for the
+  historical `duplicate_not_allowed` reason.
+- Correct start-policy documentation to `fail` and `use-existing`, including
+  active versus closed instances and preservation of original input/deadlines.
+  Start admission and workflow lifecycle behavior are unchanged.
+
 ## [2.6.0] - 2026-10-18
 
 ### Added

docs/sdk-reference.md

@@ -84,14 +84,42 @@ own `/api` routes. For example, use `http://server:8080`, not
 `http://server:8080/api`. Managed-runtime paths such as
 `https://cloud.example/api/runtime/v1/namespaces/acme` are valid as written.
 
-A workflow ID is the durable identity of an instance, not a per-attempt request
-ID. Starting the same ID again with the default `reject` duplicate policy raises
-the typed `WorkflowAlreadyStarted` exception. Generate a unique ID for each new
-instance, as the quickstart does. For an intentionally idempotent start, catch
-`WorkflowAlreadyStarted` and reconnect with
-`client.get_workflow_handle(workflow_id)`; choose `allow` or
-`terminate_existing` only when creating another run or replacing the current
-instance is the intended behavior.
+### Duplicate workflow IDs
+
+A workflow ID is the durable identity of an instance. The PHP Server start
+contract accepts these policies:
+
+| `duplicate_policy` | ID not yet used | Existing pending, running or waiting run | Closed instance |
+| --- | --- | --- | --- |
+| Omitted, `None`, or `"fail"` | Start a new instance | Raise `WorkflowAlreadyStarted` | Raise `WorkflowAlreadyStarted` |
+| `"use-existing"` | Start a new instance | Return a handle to the original run | Raise `WorkflowAlreadyStarted` |
+
+An active duplicate keeps its original input, run and deadlines. It does not
+apply the new request's arguments. Duplicate requests can append start-accepted
+or start-rejected audit events without creating another business execution.
+
+```python
+handle = await client.start_workflow(
+    workflow_type="greeter",
+    workflow_id="greet-customer-42",
+    task_queue="python-workers",
+    input=["world"],
+    duplicate_policy="use-existing",
+)
+```
+
+Use a unique workflow ID for each independent execution, as the quickstart does.
+To inspect an existing instance, including a closed one, use
+`client.get_workflow_handle(workflow_id)` and its describe/result methods instead
+of starting it again. `"use-existing"` does not restart or replace a closed run.
+The SDK forwards the policy unchanged. `"reject"`, `"allow"` and
+`"terminate_existing"` are invalid values and the PHP Server rejects them with
+HTTP 422 (`InvalidArgument`).
+
+A resource-keyed ID can coordinate admission to one durable instance. It is not
+a physical global mutex: cancellation or termination cannot revoke a request
+already sent to an external system. Protect those effects with downstream
+idempotency and, where required, resource-enforced fencing.
 
 For a fuller deployable example, see
 [`examples/order_processing`](https://github.com/durable-workflow/sdk-python/tree/main/examples/order_processing), which runs a

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "2.6.0"
+version = "2.6.1"
 description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -71,8 +71,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main"
 durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main"
 
 [tool.durable-workflow]
-product-train = "2.6.0"
-registry-version = "2.6.0"
+product-train = "2.6.1"
+registry-version = "2.6.1"
 supported-server-versions = "2.8.0"
 worker-protocol-version = "1.20"
 control-plane-version = "2"

src/durable_workflow/client.py

@@ -3527,9 +3527,18 @@ async def start_workflow(
         ``workflow_type`` is the language-neutral type key registered via
         :func:`durable_workflow.workflow.defn`. ``task_queue`` selects which
         worker pool picks up the workflow. ``workflow_id`` is the
-        caller-supplied instance id — if it collides with an existing
-        workflow, behavior depends on ``duplicate_policy``
-        (``reject`` | ``allow`` | ``terminate_existing``).
+        caller-supplied instance ID. The PHP Server start contract accepts
+        ``duplicate_policy="fail"`` (also the default when omitted) or
+        ``"use-existing"``. ``fail`` raises :class:`WorkflowAlreadyStarted`
+        if the ID already exists. ``use-existing`` returns the original run
+        only while it is pending, running or waiting. Both policies reject
+        a closed instance. Neither replaces a run or changes its input or
+        deadlines. Use a new workflow ID for a new independent execution.
+
+        The SDK forwards the policy unchanged. ``reject``, ``allow`` and
+        ``terminate_existing`` are not supported wire values. A resource-keyed
+        workflow ID coordinates start admission; it does not revoke work
+        already running in an external system.
 
         ``input`` is a list of positional arguments passed to the workflow's
         ``run`` method; the SDK encodes the list with the default payload

src/durable_workflow/errors.py

@@ -312,8 +312,10 @@ def __init__(self, workflow_id: str) -> None:
 class WorkflowAlreadyStarted(DurableWorkflowError):
     """A start request collided with an existing instance id.
 
-    Raised when duplicate-start policy is ``reject`` (the default) and the
-    caller-supplied ``workflow_id`` is already in use.
+    The PHP Server raises this for an existing ``workflow_id`` when the
+    duplicate policy is ``fail`` (the default), or when ``use-existing``
+    addresses a closed instance. ``use-existing`` returns a handle only
+    while the original run is pending, running or waiting.
     """
 
     def __init__(self, workflow_id: str) -> None:
@@ -874,7 +876,7 @@ def update_validation_failed(default: str) -> UpdateValidationFailed:
     if status == 409:
         if reason == "schedule_already_exists":
             raise ScheduleAlreadyExists(context)
-        if reason == "duplicate_not_allowed":
+        if reason in ("duplicate_not_allowed", "instance_already_started"):
             raise WorkflowAlreadyStarted(context)
         if reason == "run_not_active":
             if operation == "signal":

tests/integration/test_duplicate_start_policy.py

@@ -0,0 +1,124 @@
+"""Qualify start policies against real admission, audit history and completion."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+from uuid import uuid4
+
+import httpx
+import pytest
+
+from durable_workflow import Client, Worker, workflow
+from durable_workflow.errors import InvalidArgument, WorkflowAlreadyStarted, WorkflowNotFound
+
+
+@workflow.defn(name="tests.duplicate-policy.echo")
+class PolicyWorkflow:
+    def run(self, ctx: Any, value: Any) -> Any:
+        yield from ()
+        return value
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("initial_policy", [None, "fail", "use-existing"])
+async def test_duplicate_policies_preserve_active_and_completed_original(
+    server_url: str, server_token: str, initial_policy: str | None,
+) -> None:
+    suffix = uuid4().hex
+    queue = f"policy-{suffix}"
+    workflow_id = f"policy-{suffix}"
+    value = {"integer": 9223372036854775807, "null": None, "text": "original 雪"}
+    async with Client(server_url, token=server_token, namespace="default") as client:
+        starts: list[tuple[dict[str, Any], int]] = []
+
+        async def observe(response: httpx.Response) -> None:
+            request = response.request
+            if request.method == "POST" and request.url.path.endswith("/workflows"):
+                starts.append((json.loads(request.content), response.status_code))
+
+        client._http.event_hooks["response"].append(observe)
+
+        async def start(policy: str | None, *, original: bool = False) -> Any:
+            return await client.start_workflow(
+                workflow_type="tests.duplicate-policy.echo", task_queue=queue,
+                workflow_id=workflow_id, input=[value if original else "replacement"],
+                execution_timeout_seconds=120 if original else 600,
+                run_timeout_seconds=60 if original else 300, duplicate_policy=policy,
+            )
+
+        handle = await start(initial_policy, original=True)
+        assert starts[-1][1] == 201
+        assert starts[-1][0].get("duplicate_policy") == initial_policy
+        assert ("duplicate_policy" in starts[-1][0]) is (initial_policy is not None)
+
+        for completed in (False, True):
+            if completed:
+                worker = Worker(
+                    client, task_queue=queue, workflows=[PolicyWorkflow], activities=[],
+                    poll_timeout=0.5, shutdown_timeout=10,
+                )
+                await worker.run_until(workflow_id=workflow_id, timeout=30, poll_interval=0.1)
+                assert await handle.result(timeout=30) == value
+
+            before = await handle.describe_run()
+            assert (before.status.lower() == "completed") is completed
+            assert before.input is not None
+            assert before.execution_deadline_at
+            assert before.run_deadline_at
+            history = (await handle.get_history())["events"]
+
+            for policy in (None, "fail", "use-existing"):
+                returned = policy == "use-existing" and not completed
+                if returned:
+                    existing = await start(policy)
+                    assert (existing.workflow_id, existing.run_id) == (handle.workflow_id, handle.run_id)
+                else:
+                    with pytest.raises(WorkflowAlreadyStarted) as duplicate:
+                        await start(policy)
+                    assert duplicate.value.workflow_id == workflow_id
+                request, status = starts[-1]
+                assert request.get("duplicate_policy") == policy
+                assert ("duplicate_policy" in request) is (policy is not None)
+                assert status == (200 if returned else 409)
+                after_history = (await handle.get_history())["events"]
+                assert after_history[:len(history)] == history
+                # Canonical duplicate starts append an admission audit event,
+                # while retaining the original business execution.
+                assert [event["event_type"] for event in after_history[len(history):]] == [
+                    "StartAccepted" if returned else "StartRejected",
+                ]
+                history = after_history
+
+            for obsolete in ("reject", "allow", "terminate_existing"):
+                with pytest.raises(InvalidArgument):
+                    await start(obsolete)
+                assert starts[-1][0]["duplicate_policy"] == obsolete
+                assert starts[-1][1] == 422
+                assert (await handle.get_history())["events"] == history
+
+            after = await handle.describe_run()
+            for field in ("run_id", "input", "status", "execution_deadline_at", "run_deadline_at", "output"):
+                assert getattr(after, field) == getattr(before, field), field
+            runs = await handle.list_runs()
+            assert len(runs.runs) == 1
+            assert runs.runs[0].run_id == handle.run_id
+            assert sum(event["event_type"] == "WorkflowCompleted" for event in history) == int(completed)
+        print(json.dumps({"initial_policy": initial_policy, "workflow_id": workflow_id,
+                          "run_id": handle.run_id, "start_statuses": [status for _, status in starts]}))
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("obsolete", ["reject", "allow", "terminate_existing"])
+async def test_invalid_policy_does_not_admit_a_fresh_workflow(
+    server_url: str, server_token: str, obsolete: str,
+) -> None:
+    workflow_id = f"invalid-policy-{uuid4().hex}"
+    async with Client(server_url, token=server_token, namespace="default") as client:
+        with pytest.raises(InvalidArgument):
+            await client.start_workflow(
+                workflow_type="tests.duplicate-policy.echo", task_queue="invalid-policy",
+                workflow_id=workflow_id, input=[], duplicate_policy=obsolete,
+            )
+        with pytest.raises(WorkflowNotFound):
+            await client.describe_workflow(workflow_id)

tests/test_errors.py

@@ -57,9 +57,11 @@ def test_404_generic(self) -> None:
         with pytest.raises(ServerError):
             _raise_for_status(404, {"reason": "other"})
 
-    def test_409_duplicate(self) -> None:
-        with pytest.raises(WorkflowAlreadyStarted):
-            _raise_for_status(409, {"reason": "duplicate_not_allowed"}, context="wf-1")
+    @pytest.mark.parametrize("reason", ["duplicate_not_allowed", "instance_already_started"])
+    def test_409_duplicate(self, reason: str) -> None:
+        with pytest.raises(WorkflowAlreadyStarted) as duplicate:
+            _raise_for_status(409, {"reason": reason}, context="resource-017")
+        assert duplicate.value.workflow_id == "resource-017"
 
     def test_409_query_rejected(self) -> None:
         with pytest.raises(QueryFailed):