Consume an already-open duplicate when its orphan is satisfied

CHANGELOG.md

@@ -16,6 +16,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Replay consumes an already-open duplicate of a condition wait when the older physical wait is satisfied with a true predicate. A wait opened after that satisfaction remains a separate authored call with its own timeout. (#256)
+
 - `wait_condition` predicate fingerprints no longer change across Python versions or worker processes. Every earlier release hashed the predicate's bytecode and the `repr` of its constants. So upgrading Python (for example 3.12 to 3.13) failed replay of every in-flight run at its first wait, and a predicate containing a nested function (a generator expression, a nested lambda, or a comprehension on 3.10 and 3.11) got a new fingerprint in every process, so a restart or cache loss failed replay with `condition_fingerprint_changed`. New waits record a v2 fingerprint (`dwpy2:sha256:...`) computed from the predicate's module, qualified name and a canonical form of its source AST, with no bytecode, addresses, paths or line numbers. Without source (for example `eval` or a REPL) no fingerprint is sent, and the wait is identified by its key and call order. A v1 value recorded by an earlier release still matches when this process computes the same v1; otherwise it is accepted as unverifiable instead of failing replay, so drift in those predicates is no longer detected. A reopen sends the fingerprint its first open recorded, so a wait opened before the upgrade keeps its v1 value, and a keyless wait links a v1 value only within its own chain. A keyless wait without source links its false-wake reopens within its chain, as before. **Rollback and mixed fleets:** a run that opened a wait under this release records a v2 fingerprint, which 2.14.0 and earlier refuse on replay. Don't roll a task queue back below this release once such waits exist, and upgrade every worker on a queue together. (#258)
 
 - Same-line `wait_condition` lambdas whose bodies contain list, dict or set comprehensions share one combined fingerprint on every Python version. Put those predicates on separate lines to detect a change from one candidate to another. (#258)

src/durable_workflow/workflow.py

@@ -7855,10 +7855,13 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                             prefix="wait_condition predicate raised",
                         )])
                     if resolution == "satisfied":
-                        if has_reopened_same_wait and not satisfied:
-                            # Only an open consumed as a reopen is bounded (#248);
-                            # a later same-key wait after a true predicate is a
-                            # separate call with its own timeout.
+                        if has_reopened_same_wait and (
+                            not satisfied or _opened_before_terminal(next_wait_index, opened)
+                        ):
+                            # An already-open duplicate belongs to this wait
+                            # even when the older open satisfied it (#256).
+                            # An open after satisfaction can be a separate
+                            # authored call, with its own timeout (#248).
                             _assert_reopen_within_first_timeout(first_open_index, next_wait_index)
                             consumed_reopen_for_current_wait = True
                             wait_yield_count = next_wait_index

tests/fixtures/replay_regressions/condition-wait-orphaned-duplicate-satisfied.json

@@ -0,0 +1,109 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "condition-wait-orphaned-duplicate-satisfied",
+  "protocol_version": "1.20",
+  "bindings": [
+    "python"
+  ],
+  "source": {
+    "runtime": "synthetic-python-replay",
+    "version": "unreleased",
+    "issue": "https://github.com/durable-workflow/sdk-python/issues/256",
+    "note": "Synthetic row-order regression; no published Server producer or runtime observation is claimed."
+  },
+  "workflow": {
+    "type": "tests.replay.condition-orphaned-duplicate",
+    "input": [],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {
+      "event_type": "WorkflowStarted",
+      "recorded_at": "2026-11-04T00:00:00+00:00",
+      "payload": {
+        "workflow_type": "tests.replay.condition-orphaned-duplicate"
+      },
+      "sequence": 1
+    },
+    {
+      "event_type": "ConditionWaitOpened",
+      "recorded_at": "2026-11-04T00:00:00+00:00",
+      "payload": {
+        "condition_wait_id": "older",
+        "condition_key": "decision-1",
+        "sequence": 1,
+        "timeout_seconds": 300
+      },
+      "sequence": 2
+    },
+    {
+      "event_type": "TimerScheduled",
+      "recorded_at": "2026-11-04T00:00:00+00:00",
+      "payload": {
+        "timer_kind": "condition_timeout",
+        "condition_wait_id": "older",
+        "condition_key": "decision-1",
+        "sequence": 1,
+        "delay_seconds": 300
+      },
+      "sequence": 3
+    },
+    {
+      "event_type": "ConditionWaitOpened",
+      "recorded_at": "2026-11-04T00:00:10+00:00",
+      "payload": {
+        "condition_wait_id": "duplicate",
+        "condition_key": "decision-1",
+        "sequence": 2,
+        "timeout_seconds": 180
+      },
+      "sequence": 4
+    },
+    {
+      "event_type": "TimerScheduled",
+      "recorded_at": "2026-11-04T00:00:10+00:00",
+      "payload": {
+        "timer_kind": "condition_timeout",
+        "condition_wait_id": "duplicate",
+        "condition_key": "decision-1",
+        "sequence": 2,
+        "delay_seconds": 180
+      },
+      "sequence": 5
+    },
+    {
+      "event_type": "SignalReceived",
+      "payload": {
+        "signal_name": "decision",
+        "value": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNwwCAgEA"
+        },
+        "payload_codec": "avro",
+        "workflow_sequence": 1
+      },
+      "recorded_at": "2026-11-04T00:00:10+00:00",
+      "sequence": 6
+    },
+    {
+      "event_type": "ConditionWaitSatisfied",
+      "recorded_at": "2026-11-04T00:00:10+00:00",
+      "payload": {
+        "condition_wait_id": "older",
+        "condition_key": "decision-1",
+        "sequence": 1
+      },
+      "sequence": 7
+    }
+  ],
+  "expected": {
+    "command_sequence": [
+      {
+        "type": "complete_workflow",
+        "command_type": "CompleteWorkflow",
+        "result": "decided"
+      }
+    ]
+  }
+}

tests/test_wait_condition.py

@@ -1561,6 +1561,69 @@ def test_the_reopened_waits_own_timeout_expires_the_decision(self) -> None:
         assert isinstance(command, ScheduleActivity) and command.activity_type == "expire-proposal"
 
 
+def _satisfied_orphan_history(duplicate_timeout: int = 180, *, equal_times: bool = False) -> list[dict]:
+    """Synthetic #256 history: the older open is satisfied after a duplicate."""
+    first_at = "2026-11-04T00:00:00+00:00"
+    end_at = "2026-11-04T00:00:10+00:00"
+    duplicate_at = end_at if equal_times else "2026-11-04T00:00:09+00:00"
+    history = [{"event_type": "WorkflowStarted", "recorded_at": first_at, "payload": {}}]
+    for wait_id, sequence, at, timeout in (("older", 1, first_at, 300),
+                                          ("duplicate", 2, duplicate_at, duplicate_timeout)):
+        history.extend([
+            {"event_type": "ConditionWaitOpened", "recorded_at": at, "payload": {
+                "condition_wait_id": wait_id, "condition_key": "decision-1", "sequence": sequence,
+                "timeout_seconds": timeout}},
+            {"event_type": "TimerScheduled", "recorded_at": at, "payload": {
+                "timer_kind": "condition_timeout", "condition_wait_id": wait_id,
+                "condition_key": "decision-1", "sequence": sequence, "delay_seconds": timeout}},
+        ])
+    history.extend([
+        {**_signal_received_event("decision", [True], workflow_sequence=1), "recorded_at": end_at},
+        {"event_type": "ConditionWaitSatisfied", "recorded_at": end_at, "payload": {
+            "condition_wait_id": "older", "condition_key": "decision-1", "sequence": 1}},
+    ])
+    return history
+
+
+@pytest.mark.parametrize("duplicate_timeout", [0, 180, 300])
+@pytest.mark.parametrize("equal_times", [False, True], ids=["distinct-times", "equal-times"])
+def test_a_satisfied_orphan_consumes_its_already_open_duplicate(
+    duplicate_timeout: int, equal_times: bool,
+) -> None:
+    outcome = replay(OrphanedDuplicateDecision, _satisfied_orphan_history(
+        duplicate_timeout, equal_times=equal_times,
+    ), [])
+    assert outcome.commands == [CompleteWorkflow("decided")]
+
+
+@pytest.mark.parametrize("duplicate_timeout", [600, None], ids=["longer", "dropped"])
+def test_a_satisfied_orphans_duplicate_keeps_the_reopen_timeout_bound(duplicate_timeout: int | None) -> None:
+    history = _satisfied_orphan_history()
+    duplicate = history[3]["payload"]
+    if duplicate_timeout is None:
+        del duplicate["timeout_seconds"]
+    else:
+        duplicate["timeout_seconds"] = duplicate_timeout
+    with pytest.raises(NonDeterministicReplayError, match="condition_wait_reopen_timeout_exceeded"):
+        replay(OrphanedDuplicateDecision, history, [])
+
+
+def test_a_call_after_the_satisfied_orphan_chain_keeps_its_own_timeout() -> None:
+    history = _satisfied_orphan_history()
+    for event in history:
+        payload = event["payload"]
+        if "condition_key" in payload:
+            payload["condition_key"] = "approval"
+        if event["event_type"] == "SignalReceived":
+            payload.update(signal_name="approve", value=serializer.envelope([]))
+    history.append({"event_type": "ConditionWaitOpened", "recorded_at": "2026-11-04T00:00:11+00:00", "payload": {
+        "condition_wait_id": "next-call", "condition_key": "approval", "sequence": 3, "timeout_seconds": 400,
+    }})
+    (command,) = replay(TwoApprovals, history, []).commands
+    assert isinstance(command, WaitCondition)
+    assert command.timeout_seconds == 400
+
+
 @pytest.mark.parametrize("duplicate_timeout", [600, None], ids=["longer", "dropped"])
 @pytest.mark.parametrize("orphan_fired", [False, True], ids=["before-orphan-fires", "after-orphan-fires"])
 def test_an_orphans_duplicate_open_is_bounded_like_any_reopen(