fix: include legacy sequence claims in replay clock ambiguity checks

src/durable_workflow/replay_clock.py

@@ -142,7 +142,10 @@ def _declares_redrive(event: Mapping[str, Any]) -> bool:
 class BusinessReplayClock:
     """Index identities once, then read time only at actual scalar consumption."""
 
-    def __init__(self, events: Sequence[Mapping[str, Any]], workflow_id: str, run_id: str) -> None:
+    def __init__(
+        self, events: Sequence[Mapping[str, Any]], workflow_id: str, run_id: str,
+        *, sequence_claim: Callable[[Mapping[str, Any]], int | None],
+    ) -> None:
         self.events = events
         self._by_sequence: dict[int, list[int]] = {}
         self._child_calls: dict[str, list[int]] = {}
@@ -161,13 +164,17 @@ def __init__(self, events: Sequence[Mapping[str, Any]], workflow_id: str, run_id
         self._pending = self.initial
         self._current = self.initial
         for index, event in enumerate(events):
-            self._index_event(index, event)
+            self._index_event(index, event, sequence_claim)
 
-    def _index_event(self, index: int, event: Mapping[str, Any]) -> None:
+    def _index_event(
+        self, index: int, event: Mapping[str, Any], sequence_claim: Callable[[Mapping[str, Any]], int | None],
+    ) -> None:
         payload = event.get("payload")
         if not isinstance(payload, Mapping):
             return
-        if (sequence := _sequence(payload)) is not None:
+        # Visibility follows the resolver's actual claims, including legacy
+        # counters. Authority still requires strict integers on every row.
+        if (sequence := sequence_claim(payload)) is not None:
             self._by_sequence.setdefault(sequence, []).append(index)
         call = payload.get("child_call_id")
         if _kind(event) in _CHILD_EVENTS and _identity(call):
@@ -229,6 +236,9 @@ def observe(self, index: int | None, family: str, sequence: int) -> None:
         self._pending = (None, None)
         if index is None or not 0 <= index < len(self.events) or family not in {"activity", "timer", _CHILD_FAMILY}:
             return
+        if any(_sequence(self.events[position]["payload"]) != sequence
+               for position in self._by_sequence.get(sequence, [])):
+            return
         boundary = (self._child_boundary(index, sequence) if family == _CHILD_FAMILY
                     else self._scalar_boundary(index, family, sequence))
         if not boundary:

src/durable_workflow/workflow.py

@@ -2202,7 +2202,9 @@ def __init__(
         if self._current_time is None and _allow_wall_clock:
             self._current_time = datetime.now(timezone.utc)
         self._replay_clock_error: str | None = None
-        self._business_clock = replay_clock.BusinessReplayClock([], workflow_id, run_id)
+        self._business_clock = replay_clock.BusinessReplayClock(
+            [], workflow_id, run_id, sequence_claim=_workflow_sequence,
+        )
         self._workflow_command_id = workflow_command_id or run_id or workflow_id
         self._cancel_requested = bool(cancel_requested)
         self._cancellation_request_id: str | None = None
@@ -5139,7 +5141,9 @@ def _replay_state(
         _allow_wall_clock=not events,
     )
     ctx._replay_clock_error = replay_clock_error if workflow_start_time is None else None
-    ctx._business_clock = replay_clock.BusinessReplayClock(events, workflow_id or "", run_id)
+    ctx._business_clock = replay_clock.BusinessReplayClock(
+        events, workflow_id or "", run_id, sequence_claim=_workflow_sequence,
+    )
     lifetime = _ReplayLifetime(ctx.logger)
 
     def _set_replaying(replaying: bool) -> None:

tests/fixtures/replay_regressions/business-clock-legacy-sequence-claim-ambiguity.json

@@ -0,0 +1,107 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "business-clock-legacy-sequence-claim-ambiguity",
+  "protocol_version": "1.20",
+  "bindings": [
+    "python"
+  ],
+  "workflow": {
+    "type": "tests.replay.business-clock",
+    "input": [],
+    "payload_codec": "avro",
+    "workflow_id": "business-clock-instance",
+    "run_id": "business-clock-run"
+  },
+  "history": [
+    {
+      "event_type": "WorkflowStarted",
+      "recorded_at": "2026-11-01T00:00:00.123456Z",
+      "payload": {
+        "workflow_type": "tests.replay.business-clock",
+        "workflow_instance_id": "business-clock-instance",
+        "workflow_run_id": "business-clock-run"
+      },
+      "sequence": 1
+    },
+    {
+      "event_type": "ActivityScheduled",
+      "payload": {
+        "sequence": "1",
+        "activity_execution_id": "clock-activity",
+        "activity_type": "clock.first"
+      },
+      "sequence": 2
+    },
+    {
+      "event_type": "ActivityScheduled",
+      "payload": {
+        "sequence": 1,
+        "activity_execution_id": "clock-activity",
+        "activity_type": "clock.first"
+      },
+      "sequence": 3
+    },
+    {
+      "event_type": "ActivityCompleted",
+      "recorded_at": "2026-11-01T00:00:03.123456Z",
+      "payload": {
+        "sequence": 1,
+        "activity_execution_id": "clock-activity",
+        "activity_type": "clock.first",
+        "result": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNwoKZmlyc3Q="
+        }
+      },
+      "sequence": 4
+    },
+    {
+      "event_type": "TimerScheduled",
+      "payload": {
+        "sequence": 2,
+        "timer_id": "clock-timer",
+        "timer_kind": "durable_timer",
+        "delay_seconds": 2
+      },
+      "sequence": 5
+    },
+    {
+      "event_type": "TimerFired",
+      "recorded_at": "2026-11-01T00:00:05.654321Z",
+      "payload": {
+        "sequence": 2,
+        "timer_id": "clock-timer",
+        "timer_kind": "durable_timer",
+        "delay_seconds": 2
+      },
+      "sequence": 6
+    },
+    {
+      "event_type": "WorkflowTaskCompleted",
+      "recorded_at": "2026-12-01T00:00:00Z",
+      "payload": {},
+      "sequence": 7
+    }
+  ],
+  "expected": {
+    "command_sequence": [
+      {
+        "type": "complete_workflow",
+        "result": {
+          "clock": [
+            null,
+            null,
+            "2026-11-01T00:00:05.654321+00:00"
+          ],
+          "legacy": [
+            "2026-11-01T00:00:00.123456+00:00",
+            "2026-11-01T00:00:00.123456+00:00",
+            "2026-11-01T00:00:00.123456+00:00"
+          ],
+          "elapsed_seconds": null
+        }
+      }
+    ]
+  }
+}

tests/test_business_replay_clock_sequence_claims.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+from copy import deepcopy
+from typing import Any
+
+import pytest
+
+from durable_workflow.errors import NonDeterministicReplayError
+from tests.test_business_replay_clock import history, result
+from tests.test_business_replay_clock_child import execute
+from tests.test_business_replay_clock_child import history as child_history
+
+
+def malformed_claim(payload: dict[str, Any], shape: str) -> None:
+    sequence = payload["sequence"]
+    if shape == "digits":
+        payload["sequence"] = str(sequence)
+    elif shape == "leading-zero":
+        payload["sequence"] = f"0{sequence}"
+    elif shape == "unicode-digits":
+        payload["sequence"] = "١" if sequence == 1 else "٢"
+    elif shape == "alias-only":
+        payload["workflow_sequence"] = payload.pop("sequence")
+    elif shape == "alias-conflict":
+        payload["workflow_sequence"] = sequence + 1
+    elif shape == "alias-string":
+        payload["workflow_sequence"] = str(sequence)
+    elif shape == "alias-fallback":
+        payload.update(sequence=0, workflow_sequence=sequence)
+    else:
+        raise AssertionError(shape)
+
+
+CLAIMS = ["digits", "leading-zero", "unicode-digits", "alias-only", "alias-conflict", "alias-string", "alias-fallback"]
+
+
+@pytest.mark.parametrize("family,position,observation", [("activity", 1, 1), ("timer", 3, 2)])
+@pytest.mark.parametrize("shape", CLAIMS)
+def test_legacy_claimed_duplicate_opening_is_visible_but_cannot_prove_time(
+    family: str, position: int, observation: int, shape: str,
+) -> None:
+    events = history()
+    expected = result(events)
+    duplicate = deepcopy(events[position])
+    malformed_claim(duplicate["payload"], shape)
+    events.insert(position, duplicate)
+    expected["clock"][observation] = None
+    expected["elapsed_seconds"] = None
+    original = deepcopy(events)
+    assert result(events) == expected, family
+    assert result(events) == expected
+    assert events == original
+
+
+@pytest.mark.parametrize("shape", CLAIMS)
+@pytest.mark.parametrize("foreign_call", [False, True])
+def test_child_duplicate_claim_cannot_hide_behind_an_unindexed_counter_or_another_call(
+    shape: str, foreign_call: bool,
+) -> None:
+    events = child_history()
+    expected = execute(events)
+    duplicate = deepcopy(events[1])
+    malformed_claim(duplicate["payload"], shape)
+    if foreign_call:
+        duplicate["payload"]["child_call_id"] = "foreign-call"
+    events.insert(1, duplicate)
+    expected["clock"][1] = None
+    assert execute(events) == expected
+    assert execute(events) == expected
+
+
+@pytest.mark.parametrize("value", [True, "1", "01", "١"])
+def test_boolean_and_string_claims_do_not_weaken_the_required_actual_integer(value: Any) -> None:
+    events = history()
+    duplicate = deepcopy(events[1])
+    duplicate["payload"]["sequence"] = value
+    events.insert(1, duplicate)
+    assert result(events)["clock"] == [None, None, "2026-11-01T00:00:05.654321+00:00"]
+
+
+@pytest.mark.parametrize("position", [2, 4])
+def test_duplicate_terminal_preserves_the_original_replay_error(position: int) -> None:
+    events = history()
+    duplicate = deepcopy(events[position])
+    duplicate["payload"]["sequence"] = str(duplicate["payload"]["sequence"])
+    events.insert(position, duplicate)
+    with pytest.raises(NonDeterministicReplayError, match="Keep yielded workflow steps stable"):
+        result(events)
+
+
+def test_nonparseable_digit_counter_preserves_the_original_parser_error() -> None:
+    events = history()
+    events[1]["payload"]["sequence"] = "²"
+    with pytest.raises(ValueError, match="invalid literal for int"):
+        result(events)
+
+
+def test_unclaimed_foreign_opening_does_not_poison_a_proved_sequence() -> None:
+    events = history()
+    expected = result(events)
+    duplicate = deepcopy(events[1])
+    duplicate["payload"]["sequence"] = "not-a-counter"
+    duplicate["payload"]["activity_execution_id"] = "unrelated-activity"
+    events.insert(1, duplicate)
+    assert result(events) == expected