Enforce search-attribute replay identity and qualify redriven memo and search-attribute reuse

CHANGELOG.md

@@ -6,6 +6,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- 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)
+
+### Added
+
+- Replay of Workflow 2.19 redrive reuse is qualified for `MemoUpserted` and for `SearchAttributesUpserted` value identity. Neither event is a replay clock source: `ctx.now()` stays the persisted start time, which on a redriven run is the failed run's start, and the cancellation budget does not move on an upsert. Search-attribute type identity is not qualified, because the Server records namespace-registered types the SDK cannot see. Signal waits (`SignalWaitOpened`, `SignalReceived`, `SignalApplied`) are not qualified for Python: the SDK has no signal-wait primitive. Replaying such reused events through a signal handler applies the signal once and takes no time from them, but that does not qualify signal waits. (#235)
+
 ## [2.13.1] - 2026-11-03
 
 ### Fixed

src/durable_workflow/_cancellation_scope_history.py

@@ -122,10 +122,15 @@ def _cleanup_operation(event: dict[str, Any], prefix: Sequence[dict[str, Any]],
         "authority_deadline_at": _canonical_time(authority),
     }
     sequence = payload.get("sequence")
+    # Only an explicit null reads as one call, as in Core; any other value
+    # that is not a positive integer is refused, never narrowed.
+    span = 1 if recorded.get("sequence_span") is None else recorded["sequence_span"]
+    if not _positive(span) or not _positive(recorded.get("sequence")):
+        raise _invalid(operation + " follows a delivery without a canonical sequence span")
     deadline = scope_timestamp(expected["authority_deadline_at"])
     timestamp = scope_timestamp(event.get("timestamp", event.get("recorded_at")))
     if (snapshot != expected or scope_id != context.scope_id
-        or not _positive(sequence) or sequence < recorded["sequence"] + recorded["sequence_span"]
+        or not _positive(sequence) or sequence < recorded["sequence"] + span
         or not _positive(event.get("sequence")) or delivery["sequence"] >= event["sequence"]
         or timestamp < scope_timestamp(delivery.get("timestamp", delivery.get("recorded_at")))
         or timestamp >= deadline or not local and (

src/durable_workflow/_cooperative_cancellation.py

@@ -112,7 +112,9 @@ def from_payload(cls, payload: Mapping[str, Any]) -> CancellationDelivery:
         request_id = _text(payload.get("workflow_command_id"), "workflow_command_id")
         sequence = _positive(payload.get("sequence"), "sequence")
         kind = _text(payload.get("call_kind"), "call_kind")
-        span = _positive(payload.get("sequence_span", 1), "sequence_span", maximum=1000)
+        # Absent and null both mean one call, as Core's WorkflowStepHistory reads it.
+        recorded_span = payload.get("sequence_span")
+        span = _positive(1 if recorded_span is None else recorded_span, "sequence_span", maximum=1000)
         operation = payload.get("operation_sequence")
         operation_span = _positive(payload.get("operation_sequence_span", 1), "operation_sequence_span", maximum=1000)
         if kind not in _CALL_KINDS or (kind != "parallel" and span != 1):

src/durable_workflow/testing.py

@@ -400,8 +400,19 @@ def _apply_commands(
                     }
                 )
             elif isinstance(cmd, UpsertSearchAttributes):
+                # Record the JSON wire form, as the Server does; replay
+                # checks the workflow's update against it.
                 history.append(
-                    {"event_type": "SearchAttributesUpserted", "payload": {}}
+                    {
+                        "event_type": "SearchAttributesUpserted",
+                        "payload": {
+                            # Every recorded event but a signal is a step.
+                            "sequence": 1 + sum(
+                                event["event_type"] != "SignalReceived" for event in history
+                            ),
+                            "attributes": json.loads(json.dumps(cmd.attributes)),
+                        },
+                    }
                 )
             elif isinstance(cmd, RecordVersionMarker):
                 history.append(