Send deterministic occurrence IDs for condition waits

CHANGELOG.md

@@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Condition wait opens carry a deterministic `python:condition-wait:{ordinal}` occurrence ID for each authored call, including selection members. False-wake reopens preserve the first open's ID and remaining timeout; legacy chains without an ID keep that absence even when later physical opens have an ID. Separate calls and grace waits receive new ordinals. These identities enable unresolved-open deduplication on supporting runtimes. (#259)
+
 - Opt-in typed workflow values (#184). `@workflow.defn(..., typed=True)` decodes the arguments of `run` (after `ctx`) and of every signal, query, update and update-validator handler to their annotations. `ctx.start_child_workflow(..., result_type=T)` decodes a child's result, and `WorkflowHandle.result(result_type=T)` / `Client.get_result(..., result_type=T)` decode a workflow's result on the client. Dataclass and `Enum` values in client workflow, signal, update, query and schedule arguments, child and continue-as-new arguments, and workflow and update results are sent as their neutral values. A typed `TypedValueError` names the path but never the value. Untyped workflows receive plain values as before, so in-flight histories replay unchanged.
 
 ### Fixed

src/durable_workflow/workflow.py

@@ -1645,6 +1645,7 @@ class WaitCondition:
     condition_key: str | None = None
     condition_definition_fingerprint: str | None = None
     timeout_seconds: int | None = None
+    condition_wait_occurrence_id: str | None = None
     _cancellation_scope_id: str = field(
         default_factory=_authored_cancellation_scope, init=False, repr=False, compare=False,
     )
@@ -1670,6 +1671,8 @@ def to_server_command(
             cmd["condition_definition_fingerprint"] = self.condition_definition_fingerprint
         if self.timeout_seconds is not None:
             cmd["timeout_seconds"] = self.timeout_seconds
+        if self.condition_wait_occurrence_id is not None:
+            cmd["condition_wait_occurrence_id"] = self.condition_wait_occurrence_id
         return cmd
 
 
@@ -2213,6 +2216,7 @@ def __init__(
         self._rng = random.Random(seed)
         self._uuid7_counter = 0
         self._nexus_call_counter = 0
+        self._condition_wait_counter = 0
         self._message_stream_messages: dict[str, list[MessageStreamMessage]] = {}
         self._message_stream_cursors: dict[str, int] = {}
         self._message_stream_waits: dict[str, int] = {}
@@ -2513,15 +2517,21 @@ def wait_condition(
         The deadline is fixed when the wait first opens. A signal that leaves
         the predicate false re-drives the workflow, and the wait reopens with
         only the time left before that deadline, so signals never extend it.
+
+        Each authored call gets a deterministic occurrence ID. Reopens keep
+        the first recorded ID, including its absence in legacy histories.
         """
         timeout_seconds: int | None = None
         if timeout is not None:
             timeout_seconds = max(0, math.ceil(timeout))
+        occurrence_id = f"python:condition-wait:{self._condition_wait_counter}"
+        self._condition_wait_counter += 1
         return WaitCondition(
             predicate=predicate,
             condition_key=key,
             condition_definition_fingerprint=_condition_predicate_fingerprint(predicate),
             timeout_seconds=timeout_seconds,
+            condition_wait_occurrence_id=occurrence_id,
         )
 
     def select(
@@ -7143,6 +7153,10 @@ def _reopened_with_remaining_timeout(command: WaitCondition, first_index: int) -
         if first_index >= len(wait_opened):
             return command
         reopened = copy(command)
+        # An upgraded worker must not add an ID to a legacy chain or inherit
+        # a later physical open's ID when duplicate chains merge (#259).
+        occurrence = wait_opened[first_index].get("condition_wait_occurrence_id")
+        reopened.condition_wait_occurrence_id = occurrence if isinstance(occurrence, str) and occurrence else None
         # A reopen sends the fingerprint its first open recorded, so a wait
         # first opened under v1 (2.14.0 and earlier) keeps v1 across the
         # upgrade and links v1 to v1, never v1 to v2 (#258).

tests/fixtures/replay_regressions/condition-occurrence-first-open-preserved.json

@@ -0,0 +1,91 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "protocol_version": "1.20",
+  "bindings": [
+    "python"
+  ],
+  "id": "condition-occurrence-first-open-preserved",
+  "workflow": {
+    "type": "tests.condition-occurrence.approval",
+    "input": [],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {
+      "event_type": "WorkflowStarted",
+      "recorded_at": "2026-11-04T00:00:00+00:00",
+      "payload": {
+        "workflow_type": "tests.condition-occurrence.approval"
+      },
+      "sequence": 1
+    },
+    {
+      "event_type": "ConditionWaitOpened",
+      "recorded_at": "2026-11-04T00:00:00+00:00",
+      "payload": {
+        "condition_wait_id": "w1",
+        "sequence": 1,
+        "condition_key": "approval",
+        "timeout_seconds": 60,
+        "condition_wait_occurrence_id": "original-occurrence"
+      },
+      "sequence": 2
+    },
+    {
+      "event_type": "SignalReceived",
+      "recorded_at": "2026-11-04T00:00:10+00:00",
+      "payload": {
+        "signal_name": "note",
+        "workflow_sequence": 1,
+        "value": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNwwA"
+        },
+        "payload_codec": "avro"
+      },
+      "sequence": 3
+    },
+    {
+      "event_type": "ConditionWaitSatisfied",
+      "recorded_at": "2026-11-04T00:00:10+00:00",
+      "payload": {
+        "condition_wait_id": "w1",
+        "sequence": 1
+      },
+      "sequence": 4
+    },
+    {
+      "event_type": "ConditionWaitOpened",
+      "recorded_at": "2026-11-04T00:00:11+00:00",
+      "payload": {
+        "condition_wait_id": "w2",
+        "sequence": 2,
+        "condition_key": "approval",
+        "timeout_seconds": 50,
+        "condition_wait_occurrence_id": "later-occurrence"
+      },
+      "sequence": 5
+    },
+    {
+      "event_type": "SignalReceived",
+      "recorded_at": "2026-11-04T00:00:20+00:00",
+      "payload": {
+        "signal_name": "note",
+        "workflow_sequence": 2,
+        "value": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNwwA"
+        },
+        "payload_codec": "avro"
+      },
+      "sequence": 6
+    }
+  ],
+  "expected": {
+    "type": "open_condition_wait",
+    "condition_key": "approval",
+    "condition_wait_occurrence_id": "original-occurrence",
+    "timeout_seconds": 40
+  }
+}

tests/fixtures/replay_regressions/condition-occurrence-selection-authoring-order.json

@@ -0,0 +1,67 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "protocol_version": "1.20",
+  "bindings": [
+    "python"
+  ],
+  "id": "condition-occurrence-selection-authoring-order",
+  "workflow": {
+    "type": "tests.condition-occurrence.selection",
+    "input": [],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {
+      "event_type": "WorkflowStarted",
+      "payload": {
+        "workflow_type": "tests.condition-occurrence.selection"
+      },
+      "sequence": 1
+    }
+  ],
+  "command_sequence": [
+    {
+      "type": "open_condition_wait",
+      "condition_key": "first",
+      "condition_wait_occurrence_id": "python:condition-wait:0"
+    },
+    {
+      "type": "open_condition_wait",
+      "condition_key": "nested",
+      "condition_wait_occurrence_id": "python:condition-wait:1"
+    },
+    {
+      "type": "start_timer",
+      "delay_seconds": 5
+    },
+    {
+      "type": "open_condition_wait",
+      "condition_key": "last",
+      "condition_wait_occurrence_id": "python:condition-wait:2"
+    }
+  ],
+  "expected": {
+    "command_sequence": [
+      {
+        "type": "open_condition_wait",
+        "condition_key": "first",
+        "condition_wait_occurrence_id": "python:condition-wait:0"
+      },
+      {
+        "type": "open_condition_wait",
+        "condition_key": "nested",
+        "condition_wait_occurrence_id": "python:condition-wait:1"
+      },
+      {
+        "type": "start_timer",
+        "delay_seconds": 5
+      },
+      {
+        "type": "open_condition_wait",
+        "condition_key": "last",
+        "condition_wait_occurrence_id": "python:condition-wait:2"
+      }
+    ]
+  }
+}