Replay a redriven run from the failed run's start time

CHANGELOG.md

@@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- A run created by redrive now replays from the failed run's start time. `ctx.now()` and `ctx.uuid7()` read `WorkflowStarted.replayed_started_at` when `recovery_kind` is `redrive`, as Core does, so workflow code before the failed step sees the same clock as the failed run. An invalid value is refused rather than ignored. The cancellation clock prefers a reused step's `reused_recorded_at`. (#232)
+
 ## [2.13.0] - 2026-11-03
 
 ### Added

src/durable_workflow/workflow.py

@@ -2205,9 +2205,15 @@ def cancellation_context(self) -> CancellationContext | ScopedCancellationContex
         return self._cancellation_context
 
     def _observe_cancellation_replay_time(self, event: Mapping[str, Any] | None) -> None:
-        timestamp = event.get("timestamp") if event is not None else None
-        if timestamp is None and event is not None:
-            timestamp = event.get("recorded_at")
+        payload = event.get("payload") if event is not None else None
+        # A step reused by redrive keeps the time it was first recorded, as in
+        # Core's RedriveReplayClock::recordedAt; it overrides the event's own time.
+        if isinstance(payload, Mapping) and isinstance(payload.get("reused_recorded_at"), str):
+            timestamp: Any = payload["reused_recorded_at"]
+        else:
+            timestamp = event.get("timestamp") if event is not None else None
+            if timestamp is None and event is not None:
+                timestamp = event.get("recorded_at")
         self._cancellation_replay_time_available = False
         if not isinstance(timestamp, str):
             return
@@ -4680,6 +4686,10 @@ def _replay_state(
             explicit_time = "timestamp" in start_payload
             ts = start_payload.get("timestamp") if explicit_time else ev.get("recorded_at")
             clock_source = "payload.timestamp" if explicit_time else "recorded_at"
+            # A redriven run replays from the failed run's start time, as Core does.
+            if start_payload.get("recovery_kind") == "redrive" and "replayed_started_at" in start_payload:
+                ts = start_payload.get("replayed_started_at")
+                clock_source = "payload.replayed_started_at"
             replay_clock_error = f"Workflow replay clock unavailable: invalid or missing WorkflowStarted.{clock_source}"
             if isinstance(ts, str):
                 with contextlib.suppress(ValueError):

tests/fixtures/replay_regressions/redrive-start-replayed-clock.json

@@ -0,0 +1,10 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "redrive-start-replayed-clock",
+  "protocol_version": "1.20",
+  "bindings": ["python"],
+  "workflow": {"type": "tests.replay.persisted-clock", "input": [], "payload_codec": "avro"},
+  "history": [{"event_type": "WorkflowStarted", "recorded_at": "2026-06-01T10:05:00.000000Z", "payload": {"recovery_kind": "redrive", "replayed_started_at": "2026-06-01T10:00:00.000000Z"}}],
+  "expected": {"type": "complete_workflow", "result": "2026-06-01T10:00:00.000000+00:00"}
+}

tests/test_cancellation_remaining_time.py

@@ -263,3 +263,22 @@ def execute(command: RecordLocalActivity) -> str:
     ]
     assert observations == [22.123456, 22.123456]
     assert calls == ["once"]
+
+
+def test_a_reused_step_keeps_its_original_time_for_the_remaining_budget() -> None:
+    class Cleanup:
+        def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+            try:
+                yield ctx.start_timer(60)
+            except WorkflowCancelled as error:
+                assert error.context is not None
+                with ctx.cancellation_shield():
+                    yield ctx.schedule_activity("cleanup", [])
+                return error.context.remaining()
+
+    reused = completed_activity(2, "cleanup", "cleaned")
+    reused["payload"] = {**reused["payload"], "reused_from_run_id": "failed-run",
+                         "reused_recorded_at": "2026-10-01T00:00:12Z"}
+    events = history() + [{**reused, "timestamp": "2026-10-01T00:00:29Z"}]
+    # Redrive recorded the step again at 00:29; its workflow time stays 00:12.
+    assert replay(Cleanup, events, [], run_id="run-1").commands == [CompleteWorkflow(18.123456)]

tests/test_history_event_contract.py

@@ -674,6 +674,22 @@
 }
 
 
+# Keys Core adds only to runs created by redrive (workflow#864). The SDK reads
+# replayed_started_at for the start clock and reused_recorded_at for step time.
+REDRIVE_HISTORY_EVENT_PAYLOAD_KEYS = {
+    "WorkflowStarted": ("resume_step_sequence", "recovery_kind", "replayed_started_at"),
+    "ActivityCompleted": ("reused_from_run_id", "reused_activity_execution_id", "reused_recorded_at"),
+    "TimerFired": ("reused_from_run_id", "reused_recorded_at"),
+    "SideEffectRecorded": ("reused_from_run_id", "reused_recorded_at"),
+    "VersionMarkerRecorded": ("reused_from_run_id", "reused_recorded_at"),
+}
+
+for _event_type, _keys in REDRIVE_HISTORY_EVENT_PAYLOAD_KEYS.items():
+    DOCUMENTED_HISTORY_EVENT_PAYLOAD_KEYS[_event_type] = (
+        *DOCUMENTED_HISTORY_EVENT_PAYLOAD_KEYS[_event_type],
+        *(key for key in _keys if key not in DOCUMENTED_HISTORY_EVENT_PAYLOAD_KEYS[_event_type]),
+    )
+
 SDK_INTERPRETED_HISTORY_EVENTS = frozenset(
     {
         "WorkflowStarted",
@@ -732,9 +748,18 @@ def test_schedule_history_events_are_preserved_only() -> None:
 
 def test_php_history_fixture_key_expectations_match_documented_contract() -> None:
     for event_type, fixture_keys in PHP_HISTORY_EVENT_EXPECTED_KEYS.items():
-        documented_keys = set(DOCUMENTED_HISTORY_EVENT_PAYLOAD_KEYS[event_type])
+        # Published PHP fixtures come from ordinary runs, which never carry redrive keys.
+        documented_keys = set(DOCUMENTED_HISTORY_EVENT_PAYLOAD_KEYS[event_type]) - set(
+            REDRIVE_HISTORY_EVENT_PAYLOAD_KEYS.get(event_type, ())
+        )
         expected_keys = set(fixture_keys)
         tolerated_extra_keys = set(SDK_TOLERATED_FIXTURE_EXTRA_KEYS.get(event_type, frozenset()))
 
         assert expected_keys - documented_keys == tolerated_extra_keys, event_type
         assert documented_keys - expected_keys == set(), event_type
+
+
+def test_redrive_payload_keys_are_documented_on_the_interpreted_events() -> None:
+    for event_type, keys in REDRIVE_HISTORY_EVENT_PAYLOAD_KEYS.items():
+        assert event_type in SDK_INTERPRETED_HISTORY_EVENTS, event_type
+        assert set(keys) <= set(DOCUMENTED_HISTORY_EVENT_PAYLOAD_KEYS[event_type]), event_type

tests/test_replay_clock.py

@@ -7,7 +7,7 @@
 
 import pytest
 
-from durable_workflow import workflow
+from durable_workflow import serializer, workflow
 from durable_workflow.errors import NonDeterministicReplayError
 from durable_workflow.workflow import CompleteWorkflow, WorkflowContext, replay
 
@@ -91,3 +91,43 @@ def test_missing_replay_time_is_explicit(history: list[dict[str, Any]]) -> None:
 def test_empty_local_authoring_remains_available() -> None:
     assert datetime.fromisoformat(result(PersistedClockWorkflow, [])).tzinfo is not None
     assert UUID(result(PersistedUUID7Workflow, [])).version == 7
+
+
+@workflow.defn(name="tests.replay.redrive-clock")
+class RedriveClockWorkflow:
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        before = ctx.now().isoformat(timespec="microseconds")
+        token = str(ctx.uuid7())
+        recorded = yield ctx.schedule_activity("tests.first", [before])
+        return [before, token, recorded]
+
+
+def redriven_history(**start_payload: Any) -> list[dict[str, Any]]:
+    return [
+        {"event_type": "WorkflowStarted", "recorded_at": "2026-06-01T10:05:00.000000Z", "payload": start_payload},
+        {"event_type": "ActivityCompleted", "recorded_at": "2026-06-01T10:05:00.500000Z",
+         "payload": {"sequence": 1, "activity_type": "tests.first",
+                     "result": serializer.encode("recorded", codec="avro"), "payload_codec": "avro",
+                     "reused_from_run_id": "failed-run",
+                     "reused_recorded_at": "2026-06-01T10:00:01.000000Z"}},
+    ]
+
+
+def test_redriven_run_replays_from_the_failed_run_start_time() -> None:
+    history = redriven_history(recovery_kind="redrive", replayed_started_at="2026-06-01T10:00:00.000000Z")
+    before, token, recorded = result(RedriveClockWorkflow, history)
+    assert before == "2026-06-01T10:00:00.000000+00:00"
+    assert UUID(token).int >> 80 == int(datetime.fromisoformat(before).timestamp() * 1000)
+    assert recorded == "recorded"
+
+
+def test_replayed_start_time_requires_a_redrive_start() -> None:
+    history = redriven_history(replayed_started_at="2026-06-01T10:00:00.000000Z")
+    assert result(RedriveClockWorkflow, history)[0] == "2026-06-01T10:05:00.000000+00:00"
+
+
+@pytest.mark.parametrize("invalid", [None, 123, "", "not-a-time", "2026-06-01T10:00:00"])
+def test_invalid_replayed_start_time_is_refused_without_fallback(invalid: Any) -> None:
+    history = redriven_history(recovery_kind="redrive", replayed_started_at=invalid)
+    with pytest.raises(NonDeterministicReplayError, match="replayed_started_at"):
+        replay(RedriveClockWorkflow, history, [], run_id="redriven-run", payload_codec="avro")