Refuse a condition-wait reopen longer than its first open

CHANGELOG.md

@@ -10,6 +10,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Opt-in typed activity values (#184). `@activity.defn(..., typed=True)` decodes an activity's positional arguments to its parameter annotations, and `ctx.schedule_activity(..., result_type=T)` decodes the completed result in single calls, parallel groups and selections. Supported types are dataclasses, `Enum` (including `IntEnum` and `StrEnum`), `Optional[T]`, `list[T]`, `dict[str, T]` and scalars; other annotations raise `TypeError` up front. Undeclared keys are ignored and missing fields take their defaults. A mismatch raises the new `TypedValueError`, which names the field path but never the value. It is non-retryable in a typed activity and is raised at the `yield` for `result_type`. Dataclass and `Enum` activity arguments and results are now sent as their neutral values, the same bytes as `dataclasses.asdict()` with each enum's `.value`; these values were refused before. Untyped activities and calls receive plain values as before, so in-flight histories replay unchanged.
 
+### Fixed
+
+- Replay now refuses a recorded condition-wait reopen that records a longer timeout than the wait's first open, or that adds or drops the timeout, with a `NonDeterministicReplayError` whose detail starts `condition_wait_reopen_timeout_exceeded`. Earlier versions linked such a reopen silently. Reopens with the full timeout (written before 2.13.2) or with the time remaining (2.13.2 and later) still replay. This matches sdk-php#271. (#248)
+
 ## [2.13.2] - 2026-11-04
 
 ### Fixed

src/durable_workflow/workflow.py

@@ -7020,6 +7020,30 @@ def _consume_cancellation(command: Any, boundary: CancellationDelivery) -> _Repl
         except StopIteration as stop:
             return _terminal_state(stop.value, include_pending=True)
 
+    def _assert_reopen_within_first_timeout(first_index: int, reopen_index: int) -> None:
+        """A physical reopen keeps the first open's timeout or undercuts it.
+
+        Earlier SDKs reopened with the full timeout and later ones with the time
+        remaining, so a reopen may equal or undercut the first open. Anything
+        else, including a reopen that adds or drops the timeout, is not a
+        history this workflow code writes. This is the sdk-php#271 rule.
+        """
+        first = wait_opened[first_index].get("timeout_seconds")
+        reopen = wait_opened[reopen_index].get("timeout_seconds")
+        if first is None and reopen is None:
+            return
+        if (isinstance(first, int) and not isinstance(first, bool) and isinstance(reopen, int)
+                and not isinstance(reopen, bool) and reopen <= first):
+            return
+        sequence = wait_opened[reopen_index].get("sequence")
+        raise NonDeterministicReplayError(
+            sequence if isinstance(sequence, int) else current_call_sequence,
+            "condition wait reopen within its first open's timeout",
+            ["ConditionWaitOpened"],
+            detail="condition_wait_reopen_timeout_exceeded: a reopen records a longer timeout than the wait's "
+            "first open, or adds or drops it",
+        )
+
     def _reopened_with_remaining_timeout(command: WaitCondition, first_index: int) -> WaitCondition:
         """Reopen a logical wait with the time left before its first deadline.
 
@@ -7699,7 +7723,6 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                         and next_wait_index < len(wait_opened)
                         and _same_logical_condition_wait(wait_opened[next_wait_index], cmd)
                     )
-
                     if resolution == "timed_out":
                         terminal_condition_reopen_cmd = None
                         next_value = False
@@ -7714,6 +7737,10 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                         )])
                     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.
+                            _assert_reopen_within_first_timeout(first_open_index, next_wait_index)
                             consumed_reopen_for_current_wait = True
                             wait_yield_count = next_wait_index
                             continue
@@ -7741,6 +7768,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                     # with the same key/fingerprint before declaring the
                     # logical wait still pending.
                     if has_reopened_same_wait:
+                        _assert_reopen_within_first_timeout(first_open_index, next_wait_index)
                         consumed_reopen_for_current_wait = True
                         wait_yield_count = next_wait_index
                         continue

tests/fixtures/replay_regressions/condition-wait-reopen-longer-timeout-refused.json

@@ -0,0 +1,106 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "condition-wait-reopen-longer-timeout-refused",
+  "protocol_version": "1.20",
+  "bindings": [
+    "python"
+  ],
+  "workflow": {
+    "type": "tests.replay.condition-remaining-timeout",
+    "input": [],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {
+      "event_type": "WorkflowStarted",
+      "recorded_at": "2026-11-03T00:00:00+00:00",
+      "payload": {
+        "workflow_type": "tests.replay.condition-remaining-timeout"
+      },
+      "sequence": 1
+    },
+    {
+      "event_type": "ConditionWaitOpened",
+      "recorded_at": "2026-11-03T00:00:00+00:00",
+      "payload": {
+        "condition_wait_id": "w1",
+        "condition_key": "approval",
+        "sequence": 1,
+        "timeout_seconds": 300
+      },
+      "sequence": 2
+    },
+    {
+      "event_type": "TimerScheduled",
+      "recorded_at": "2026-11-03T00:00:00+00:00",
+      "payload": {
+        "timer_kind": "condition_timeout",
+        "condition_wait_id": "w1",
+        "condition_key": "approval",
+        "sequence": 1,
+        "delay_seconds": 300
+      },
+      "sequence": 3
+    },
+    {
+      "event_type": "SignalReceived",
+      "payload": {
+        "signal_name": "note",
+        "value": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNwwA"
+        },
+        "payload_codec": "avro",
+        "workflow_sequence": 1
+      },
+      "recorded_at": "2026-11-03T00:02:00+00:00",
+      "sequence": 4
+    },
+    {
+      "event_type": "ConditionWaitOpened",
+      "recorded_at": "2026-11-03T00:02:01+00:00",
+      "payload": {
+        "condition_wait_id": "w2",
+        "condition_key": "approval",
+        "sequence": 2,
+        "timeout_seconds": 400
+      },
+      "sequence": 5
+    },
+    {
+      "event_type": "TimerScheduled",
+      "recorded_at": "2026-11-03T00:02:01+00:00",
+      "payload": {
+        "timer_kind": "condition_timeout",
+        "condition_wait_id": "w2",
+        "condition_key": "approval",
+        "sequence": 2,
+        "delay_seconds": 400
+      },
+      "sequence": 6
+    },
+    {
+      "event_type": "SignalReceived",
+      "payload": {
+        "signal_name": "note",
+        "value": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNwwA"
+        },
+        "payload_codec": "avro",
+        "workflow_sequence": 2
+      },
+      "recorded_at": "2026-11-03T00:03:00+00:00",
+      "sequence": 7
+    }
+  ],
+  "expected": {
+    "command_sequence": []
+  },
+  "expected_replay_error": {
+    "type": "NonDeterministicReplayError",
+    "workflow_sequence": 2,
+    "message_contains": "condition_wait_reopen_timeout_exceeded"
+  }
+}

tests/test_wait_condition.py

@@ -625,17 +625,6 @@
                     "condition_wait_id": "wait-finish",
                     "condition_key": "done",
                     "sequence": 22,
-                    "timeout_seconds": 30,
-                },
-            },
-            {
-                "event_type": "TimerScheduled",
-                "payload": {
-                    "timer_kind": "condition_timeout",
-                    "condition_wait_id": "wait-finish",
-                    "condition_key": "done",
-                    "sequence": 22,
-                    "delay_seconds": 30,
                 },
             },
         ]
@@ -674,17 +663,6 @@
                     "condition_wait_id": "wait-finish",
                     "condition_key": "done",
                     "sequence": 22,
-                    "timeout_seconds": 30,
-                },
-            },
-            {
-                "event_type": "TimerScheduled",
-                "payload": {
-                    "timer_kind": "condition_timeout",
-                    "condition_wait_id": "wait-finish",
-                    "condition_key": "done",
-                    "sequence": 22,
-                    "delay_seconds": 30,
                 },
             },
         ]
@@ -1377,6 +1355,18 @@
         assert isinstance(command, WaitCondition) and command.timeout_seconds == 271
 
 
+@workflow.defn(name="tests.replay.condition-longer-grace-after-timeout")
+class LongerGraceAfterTimeout(GraceAfterTimeout):
+    grace_seconds = 400
+
+
+def test_a_longer_grace_wait_after_a_timeout_is_a_separate_call() -> None:
+    # The 300 s wait timed out, so a 400 s wait with the same key is not a
+    # reopen and the reopen bound (#248) does not apply to it.
+    (command,) = replay(LongerGraceAfterTimeout, _timed_out_then_grace(400, "note"), []).commands
+    assert isinstance(command, WaitCondition) and command.timeout_seconds == 371
+
+
 def test_a_selection_condition_member_is_not_reopened_by_a_wrong_signal() -> None:
     # Ordinary parallel lists refuse WaitCondition. A condition inside
     # ctx.select is never re-sent while open, so its original timer keeps its
@@ -1394,3 +1384,103 @@
         "payload_codec": serializer.AVRO_CODEC}}
     outcome = replay(DurableSelectionConditionSignalReplayWorkflow, [START, *opens, wrong], [])
     assert list(outcome.commands) == []
+
+
+# #248: a physical reopen never records more time than the wait's first open
+# (parity with sdk-php#271). Earlier SDKs reopened with the full timeout, later
+# ones with the remainder, so equal and shorter reopens replay.
+def _reopen_history(first: int | None, reopen: int | None) -> list[dict]:
+    def opened(wait_id: str, sequence: int, at: str, timeout: int | None) -> list[dict]:
+        events = _timed_open(wait_id, sequence, at, timeout=timeout or 0)
+        if timeout is None:
+            del events[0]["payload"]["timeout_seconds"]
+            events = events[:1]
+        return events
+
+    return [START, *opened("w1", 1, "2026-11-03T00:00:00+00:00", first),
+            _timed_signal("note", 1, "2026-11-03T00:02:00+00:00"),
+            *opened("w2", 2, "2026-11-03T00:02:01+00:00", reopen),
+            _timed_signal("note", 2, "2026-11-03T00:03:00+00:00")]
+
+
+@workflow.defn(name="tests.replay.condition-untimed-reopen")
+class UntimedApproval(RemainingTimeoutApproval):
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        return (yield ctx.wait_condition(lambda: self.approved, key="approval"))
+
+
+class TestReopenTimeoutBound:
+    @pytest.mark.parametrize("reopen", [300, 179, 0])
+    def test_an_equal_or_shorter_reopen_replays(self, reopen: int) -> None:
+        assert _reopened_timeout(_reopen_history(300, reopen)) == 120
+
+    @pytest.mark.parametrize(("first", "reopen"), [(300, 301), (300, 600)])
+    def test_a_longer_reopen_is_refused(self, first: int, reopen: int) -> None:
+        with pytest.raises(NonDeterministicReplayError, match="condition_wait_reopen_timeout_exceeded") as raised:
+            replay(RemainingTimeoutApproval, _reopen_history(first, reopen), [])
+        assert raised.value.workflow_sequence == 2
+        assert str(reopen) not in str(raised.value)
+
+    def test_untimed_reopens_of_an_untimed_wait_replay(self) -> None:
+        (command,) = replay(UntimedApproval, _reopen_history(None, None), []).commands
+        assert isinstance(command, WaitCondition) and command.timeout_seconds is None
+
+    @pytest.mark.parametrize(("first", "reopen"), [(None, 60), (300, None)])
+    def test_a_reopen_that_adds_or_drops_the_timeout_is_refused(self, first: int | None, reopen: int | None) -> None:
+        # An untimed wait never records a timeout and a timed one never drops
+        # it, so either history comes from changed code (sdk-php#271).
+        workflow_cls = UntimedApproval if first is None else RemainingTimeoutApproval
+        with pytest.raises(NonDeterministicReplayError, match="condition_wait_reopen_timeout_exceeded"):
+            replay(workflow_cls, _reopen_history(first, reopen), [])
+
+
+# A same-key wait that follows a satisfied one is a separate call, so the
+# reopen bound (#248) does not apply to it, whatever its timeout.
+@workflow.defn(name="tests.replay.condition-two-approvals")
+class TwoApprovals:
+    first_timeout: int | None = 300
+    second_timeout: int | None = 400
+
+    def __init__(self) -> None:
+        self.approvals = 0
+
+    @workflow.signal("approve")
+    def approve(self) -> None:
+        self.approvals += 1
+
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        cls = type(self)
+        yield ctx.wait_condition(lambda: self.approvals >= 1, key="approval", timeout=cls.first_timeout)
+        yield ctx.wait_condition(lambda: self.approvals >= 2, key="approval", timeout=cls.second_timeout)
+        return self.approvals
+
+
+def _two_approvals_history(first: int | None, second: int | None, satisfied_recorded: bool) -> list[dict]:
+    def open_wait(wait_id: str, sequence: int, at: str, timeout: int | None) -> dict:
+        payload: dict = {"condition_wait_id": wait_id, "condition_key": "approval", "sequence": sequence}
+        if timeout is not None:
+            payload["timeout_seconds"] = timeout
+        return {"event_type": "ConditionWaitOpened", "recorded_at": at, "payload": payload}
+
+    history = [START, open_wait("w1", 1, "2026-11-03T00:00:00+00:00", first),
+               _timed_signal("approve", 1, "2026-11-03T00:00:10+00:00")]
+    if satisfied_recorded:
+        history.append({"event_type": "ConditionWaitSatisfied", "recorded_at": "2026-11-03T00:00:10+00:00",
+                        "payload": {"condition_wait_id": "w1", "condition_key": "approval", "sequence": 1}})
+    history.append(open_wait("w2", 2, "2026-11-03T00:00:11+00:00", second))
+    return history
+
+
+@pytest.mark.parametrize("satisfied_recorded", [True, False])
+@pytest.mark.parametrize(("first", "second"), [(300, 400), (None, 60), (300, None)],
+                         ids=["longer", "added", "dropped"])
+def test_a_separate_wait_after_a_satisfied_one_keeps_its_own_timeout(
+    first: int | None, second: int | None, satisfied_recorded: bool,
+) -> None:
+    TwoApprovals.first_timeout, TwoApprovals.second_timeout = first, second
+    try:
+        (command,) = replay(TwoApprovals, _two_approvals_history(first, second, satisfied_recorded), []).commands
+    finally:
+        TwoApprovals.first_timeout, TwoApprovals.second_timeout = 300, 400
+    assert isinstance(command, WaitCondition) and command.condition_key == "approval"
+    assert command.timeout_seconds == second