Reopen a condition wait with its remaining timeout

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- A `wait_condition` timeout now counts from when the wait first opens. Before, a signal that left the predicate false reopened the wait with the full `timeout` again, so every such signal pushed the timeout later: a 300-second wait signalled after 2 minutes timed out after 7. The reopen now carries the seconds left before the first open's deadline, rounded up and never below 0, measured at the latest recorded history time. Histories without recorded times keep the authored timeout, and histories written by earlier versions still replay. (#245)
 - Replay now enforces the recorded search attributes (Workflow protocol 1.20 replay identity). A workflow whose `upsert_search_attributes` differs from its `SearchAttributesUpserted` history fails replay with a `NonDeterministicReplayError` whose detail starts `search_attribute_value_mismatch`, as memo replay already does for memos. Values are compared as Core's `UpsertSearchAttributesCall` normalizes them (trimmed strings and list items, a blank string as null) and with their types (`1`, `1.0`, `True` and `"1"` differ). An authored tuple is checked as the JSON list it is sent as. Values outside that portable domain, on either side, are refused rather than coerced: non-string or non-portable keys, objects, sets, bytes, maps, non-string list items, non-finite floats and out-of-range integers. A `SearchAttributesUpserted` event without `attributes` is refused rather than skipped, with a detail starting `search_attribute_update_missing`. One without a workflow sequence is refused with `durable_command_sequence_missing` (or `durable_command_sequence_invalid` for zero, negative, string, float or bool) rather than given one, and a second upsert at the same sequence with `duplicate_search_attribute_update`. An upsert that shares its workflow sequence with any other durable command record is refused with `durable_command_sequence_collision`. Every lifecycle row that carries a command's sequence claims it, including activity heartbeat, retry and cancellation-acknowledged rows, child cancellation rows, service operation scheduling, cancellation scope openings, continue-as-new, and cooperative and scope cancellation deliveries, which claim every sequence in their `sequence_span` (absent or null means 1, as in Core; any other value that is not a positive integer is refused with `durable_command_sequence_invalid`); events that only point at a step, such as a signal receipt, an update or a cancellation request, claim no sequence. When the history records `attribute_types`, it must type exactly the update's non-null keys, and each type must be a known type that can hold the value; anything else, including a malformed type value, fails with `search_attribute_type_mismatch`. The SDK's upsert declares no types and the Server records a key's namespace-registered type ahead of the inferred one, so a string may be recorded as `keyword`, `string` or `datetime` and an integer as `int` or `float`; search-attribute type identity is therefore not pinned for Python. History without `attribute_types` keeps value-only identity. Replay diagnostics name reasons, keys, JSON types, sequences and event types, never a recorded or authored value. (#235)
 
 ### Corrected

src/durable_workflow/workflow.py

@@ -2463,6 +2463,10 @@
         ``False`` (otherwise ``True``). The fractional ``timeout`` is rounded
         up to the next whole second to match the server's integer-second
         timer resolution.
+
+        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.
         """
         timeout_seconds: int | None = None
         if timeout is not None:
@@ -3082,6 +3086,20 @@
     return codec
 
 
+def _history_event_time(event: Mapping[str, Any]) -> datetime | None:
+    """The recorded time of a history event, or None if it has none."""
+    value = event.get("timestamp")
+    if value is None:
+        value = event.get("recorded_at")
+    if not isinstance(value, str):
+        return None
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    return parsed if parsed.tzinfo is not None and parsed.utcoffset() is not None else None
+
+
 def _decode_history_result(
     payload: dict[str, Any],
     fallback_codec: str | None,
@@ -4979,6 +4997,10 @@
         ):
             resolved_sequences.add(sequence)
 
+    # The latest recorded time in this history: when a re-driven wait reopens.
+    history_time = max(
+        (time for time in (_history_event_time(ev) for ev in events) if time is not None), default=None,
+    )
     workflow_start_time: datetime | None = None
     replay_clock_error = "Workflow replay clock unavailable: missing WorkflowStarted persisted start time"
     for ev in events:
@@ -5101,6 +5123,9 @@
     # to match against their corresponding opened wait in history
     # (Nth yield ↔ Nth opened).
     wait_opened: list[dict[str, Any]] = []
+    # The recorded time of each ``wait_opened`` row, or None when history
+    # carries no parseable time. A reopened wait keeps its first open's deadline.
+    wait_opened_times: list[datetime | None] = []
     # Map condition_wait_id → resolution: 'satisfied' (from ConditionWaitSatisfied
     # in history, future server-recorded) or 'timed_out' (from a matching
     # condition_timeout TimerFired event).
@@ -5648,6 +5673,7 @@
                 else:
                     recorded_wait_steps.append(_recorded_step("condition wait", ev))
                     wait_opened.append(dict(payload))
+                    wait_opened_times.append(_history_event_time(ev))
         elif etype == "ConditionWaitSatisfied":
             wait_id = payload.get("condition_wait_id")
             if isinstance(wait_id, str) and wait_id:
@@ -6959,6 +6985,27 @@
         except StopIteration as stop:
             return _terminal_state(stop.value, include_pending=True)
 
+    def _reopened_with_remaining_timeout(command: WaitCondition, first_index: int) -> WaitCondition:
+        """Reopen a logical wait with the time left before its first deadline.
+
+        The deadline is fixed when the wait first opens, as in embedded
+        Workflow: the first ConditionWaitOpened time plus the authored
+        timeout. A non-satisfying signal re-drives the wait, and the reopen
+        carries the remaining whole seconds measured at the latest recorded
+        history time, rounded up and clamped to 0..authored (the sdk-php#269
+        rule). Without recorded times the authored timeout is kept.
+        """
+        authored = command.timeout_seconds
+        if authored is None or first_index >= len(wait_opened):
+            return command
+        opened_at = wait_opened_times[first_index]
+        if opened_at is None or history_time is None:
+            return command
+        remaining = math.ceil(authored - (history_time - opened_at).total_seconds())
+        reopened = copy(command)
+        reopened.timeout_seconds = min(authored, max(0, remaining))
+        return reopened
+
     def _advance_cancellation_clock(indexes: Iterable[int | None]) -> None:
         if ctx._cancellation_context is None and not consumed_scope_deliveries:
             return
@@ -7541,6 +7588,7 @@
                 # Terminal cleanup may only skip a later open after this
                 # yielded wait has already replayed at least one false reopen.
                 consumed_reopen_for_current_wait = False
+                first_open_index = wait_yield_count
                 while True:
                     resolution: str | None = None
                     opened: dict[str, Any] | None = None
@@ -7643,7 +7691,10 @@
                         continue
 
                     _set_replaying(False)
-                    pending.append(cmd)
+                    pending.append(
+                        _reopened_with_remaining_timeout(cmd, first_open_index)
+                        if opened is not None else cmd
+                    )
                     wait_yield_count = next_wait_index
                     return _state(pending)
                 continue

tests/fixtures/replay_regressions/condition-wait-reopen-remaining-timeout.json

@@ -0,0 +1,71 @@
+{
+  "$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-remaining-timeout",
+  "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": "approval-wait-1",
+        "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": "approval-wait-1",
+        "condition_key": "approval",
+        "sequence": 1,
+        "delay_seconds": 300
+      },
+      "sequence": 3
+    },
+    {
+      "sequence": 4,
+      "event_type": "SignalReceived",
+      "recorded_at": "2026-11-03T00:02:00+00:00",
+      "payload": {
+        "signal_name": "note",
+        "value": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNwwA"
+        },
+        "payload_codec": "avro",
+        "workflow_sequence": 1
+      }
+    }
+  ],
+  "expected": {
+    "command_sequence": [
+      {
+        "type": "open_condition_wait",
+        "command_type": "WaitCondition",
+        "condition_key": "approval",
+        "timeout_seconds": 180
+      }
+    ]
+  }
+}

tests/test_replay_regression_corpus.py

@@ -59,7 +59,7 @@
     PostConditionReceiversWorkflow,
     UpdateSignalConditionTimerWorkflow,
 )
-from tests.test_wait_condition import SignalCounterUntilFinished
+from tests.test_wait_condition import RemainingTimeoutApproval, SignalCounterUntilFinished
 from tests.test_worker import compatible_cluster_info
 from tests.test_worker_status import RecoveryWorkflow as StatusRecoveryWorkflow
 from tests.test_workflow_identity import IdentityAcrossReplay, IdentityContinueAsNew
@@ -414,6 +414,7 @@ def run(self, ctx: WorkflowContext, message: str) -> None:
     WorkflowMemoAuthorWorkflow,
     YieldedContinueAfterMetadataWorkflow,
     StatusRecoveryWorkflow,
+    RemainingTimeoutApproval,
 ]
 WORKFLOW_TYPES = {str(getattr(workflow, "__workflow_name__", workflow.__name__)): workflow for workflow in WORKFLOWS}
 WORKFLOW_TYPES["tests.replay.condition-identity"] = ConditionIdentityWorkflow

tests/test_wait_condition.py

@@ -1222,3 +1222,102 @@
         assert len(outcome.commands) == 1
         assert isinstance(outcome.commands[0], CompleteWorkflow)
         assert outcome.commands[0].result == "done"
+
+
+# #245: a reopened wait keeps the deadline fixed at its first open.
+@workflow.defn(name="tests.replay.condition-remaining-timeout")
+class RemainingTimeoutApproval:
+    def __init__(self) -> None:
+        self.approved = False
+
+    @workflow.signal("note")
+    def note(self) -> None:
+        return None
+
+    @workflow.signal("approve")
+    def approve(self) -> None:
+        self.approved = True
+
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        return (yield ctx.wait_condition(lambda: self.approved, key="approval", timeout=300))
+
+
+def _timed_open(wait_id: str, sequence: int, at: str, timeout: int = 300) -> list[dict]:
+    return [
+        {"event_type": "ConditionWaitOpened", "recorded_at": at, "payload": {
+            "condition_wait_id": wait_id, "condition_key": "approval", "sequence": sequence,
+            "timeout_seconds": timeout}},
+        {"event_type": "TimerScheduled", "recorded_at": at, "payload": {
+            "timer_kind": "condition_timeout", "condition_wait_id": wait_id, "condition_key": "approval",
+            "sequence": sequence, "delay_seconds": timeout}},
+    ]
+
+
+def _timed_signal(name: str, sequence: int, at: str) -> dict:
+    return {**_signal_received_event(name, [], workflow_sequence=sequence), "recorded_at": at}
+
+
+START = {"event_type": "WorkflowStarted", "recorded_at": "2026-11-03T00:00:00+00:00", "payload": {}}
+
+
+def _reopened_timeout(history: list[dict]) -> int | None:
+    (command,) = replay(RemainingTimeoutApproval, history, []).commands
+    assert isinstance(command, WaitCondition)
+    return command.timeout_seconds
+
+
+class TestReopenedWaitRemainingTimeout:
+    def test_a_non_satisfying_signal_reopens_with_the_remaining_time(self) -> None:
+        history = [START, *_timed_open("w1", 1, "2026-11-03T00:00:00+00:00"),
+                   _timed_signal("note", 1, "2026-11-03T00:02:00.250000+00:00")]
+        # Deadline 00:05:00 less 00:02:00.25, rounded up.
+        assert _reopened_timeout(history) == 180
+
+    def test_a_second_reopen_keeps_the_first_deadline(self) -> None:
+        history = [START, *_timed_open("w1", 1, "2026-11-03T00:00:00+00:00"),
+                   _timed_signal("note", 1, "2026-11-03T00:02:00+00:00"),
+                   *_timed_open("w2", 2, "2026-11-03T00:02:01+00:00", timeout=180),
+                   _timed_signal("note", 2, "2026-11-03T00:04:30+00:00")]
+        assert _reopened_timeout(history) == 30
+
+    def test_a_reopen_after_the_deadline_carries_zero(self) -> None:
+        history = [START, *_timed_open("w1", 1, "2026-11-03T00:00:00+00:00"),
+                   _timed_signal("note", 1, "2026-11-03T00:06:00+00:00")]
+        assert _reopened_timeout(history) == 0
+
+    def test_history_recorded_with_full_reopen_timeouts_still_replays(self) -> None:
+        # Earlier SDKs reopened with the full timeout; the open is not compared.
+        history = [START, *_timed_open("w1", 1, "2026-11-03T00:00:00+00:00"),
+                   _timed_signal("note", 1, "2026-11-03T00:02:00+00:00"),
+                   *_timed_open("w2", 2, "2026-11-03T00:02:01+00:00", timeout=300),
+                   _timed_signal("note", 2, "2026-11-03T00:03:00+00:00")]
+        assert _reopened_timeout(history) == 120
+
+    def test_a_satisfying_signal_completes_without_a_reopen(self) -> None:
+        history = [START, *_timed_open("w1", 1, "2026-11-03T00:00:00+00:00"),
+                   _timed_signal("approve", 1, "2026-11-03T00:02:00+00:00")]
+        (command,) = replay(RemainingTimeoutApproval, history, []).commands
+        assert command == CompleteWorkflow(True)
+
+    def test_a_first_open_keeps_the_authored_timeout(self) -> None:
+        (command,) = replay(RemainingTimeoutApproval, [START], []).commands
+        assert isinstance(command, WaitCondition) and command.timeout_seconds == 300
+
+    def test_history_without_recorded_times_keeps_the_authored_timeout(self) -> None:
+        history = [{"event_type": "WorkflowStarted", "payload": {"timestamp": "2026-11-03T00:00:00+00:00"}},
+                   *[{key: value for key, value in event.items() if key != "recorded_at"}
+                     for event in _timed_open("w1", 1, "2026-11-03T00:00:00+00:00")],
+                   _signal_received_event("note", [], workflow_sequence=1)]
+        assert _reopened_timeout(history) == 300
+
+    @pytest.mark.parametrize(("signalled_at", "expected"), [
+        ("2026-11-04T02:38:05.345000+00:00", 180),
+        ("2026-11-04T02:40:05+00:00", 61),
+        ("2026-11-04T02:41:06+00:00", 0),
+    ])
+    def test_sdk_php_false_wake_deadline_vectors(self, signalled_at: str, expected: int) -> None:
+        # The same vectors as sdk-php ReplayerTest::falseWakeDeadlines (#269).
+        start = {**START, "recorded_at": "2026-11-04T02:36:05+00:00"}
+        history = [start, *_timed_open("w1", 1, "2026-11-04T02:36:05.295000+00:00"),
+                   _timed_signal("note", 1, signalled_at)]
+        assert _reopened_timeout(history) == expected